geo-rep: minimize xsync crawl usage and set upper limit to xsync crawl

For effective handling of deletes and renames use history crawl
as much as possible. History crawl will run in loop till it syncs
all data before live changelog time.

When it uses xsync crawl(fallback when changelog not available, or
very first crawl) it sets upper limit to crawl.

After completing History crawl, it checks actual end time returned
by history api to compare with register time, if actual end is less
than register time then run history crawl one more time.

If first turn history processing time is less than the CHANGELOG
ROLLOVER TIME then sleep for the difference, After sleep if it is
guaranteed that rollover will happen and switches to live changelog
consumption without switching to xsync. This sleep is only when
history processing completed < CHANGELOG_ROLLOVER_TIME and sleep
only after the first turn, So will not affect the performance.

BUG: 1112238
Change-Id: I644ef24b07e42e81cec96a025ebd21244a555ec0
Signed-off-by: Aravinda VK <avishwan@redhat.com>
Reviewed-on: http://review.gluster.org/8151
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Venky Shankar <vshankar@redhat.com>
Tested-by: Venky Shankar <vshankar@redhat.com>

1 files changed, 82 insertions, 19 deletions

diff --git a/geo-replication/syncdaemon/master.py b/geo-replication/syncdaemon/master.py
index 2decc5de930..e3904736ba2 100644
--- a/geo-replication/syncdaemon/master.py
+++ b/geo-replication/syncdaemon/master.py
@@ -29,6 +29,14 @@ from syncdutils import NoPurgeTimeAvailable, PartialHistoryAvailable
 
 URXTIME = (-1, 0)
 
+# Default rollover time set in changelog translator
+# changelog rollover time is hardcoded here to avoid the
+# xsync usage when crawling switch happens from history
+# to changelog. If rollover time increased in translator
+# then geo-rep can enter into xsync crawl after history
+# crawl before starting live changelog crawl.
+CHANGELOG_ROLLOVER_TIME = 15
+
 # Utility functions to help us to get to closer proximity
 # of the DRY principle (no, don't look for elevated or
 # perspectivistic things here)
@@ -160,7 +168,10 @@ class NormalMixin(object):
             raise GsyncdError("timestamp corruption for " + path)
 
     def need_sync(self, e, xte, xtrd):
-        return xte > xtrd
+        if self.xsync_upper_limit is None:
+            return xte > xtrd
+        else:
+            return xte > xtrd and xte < self.xsync_upper_limit
 
     def set_slave_xtime(self, path, mark):
         self.slave.server.set_stime(path, self.uuid, mark)
@@ -431,7 +442,7 @@ class GMasterCommon(object):
     def register(self):
         self.register()
 
-    def crawlwrap(self, oneshot=False):
+    def crawlwrap(self, oneshot=False, no_stime_update=False):
         if oneshot:
             # it's important to do this during the oneshot crawl as
             # for a passive gsyncd (ie. in a replicate scenario)
@@ -499,7 +510,7 @@ class GMasterCommon(object):
                 time.sleep(5)
                 continue
             self.update_worker_health("Active")
-            self.crawl()
+            self.crawl(no_stime_update=no_stime_update)
             if oneshot:
                 return
             time.sleep(self.sleep_interval)
@@ -1119,7 +1130,7 @@ class GMasterChangelogMixin(GMasterCommon):
             except:
                 raise
 
-    def crawl(self):
+    def crawl(self, no_stime_update=False):
         self.update_worker_crawl_status("Changelog Crawl")
         changes = []
         # get stime (from the brick) and purge changelogs
@@ -1147,20 +1158,25 @@ class GMasterChangelogMixin(GMasterCommon):
                 logging.debug('processing changes %s' % repr(changes))
                 self.process(changes)
 
-    def register(self, changelog_agent):
+    def register(self, register_time, changelog_agent):
         self.changelog_agent = changelog_agent
         self.sleep_interval = int(gconf.change_interval)
         self.changelog_done_func = self.changelog_agent.done
 
 
 class GMasterChangeloghistoryMixin(GMasterChangelogMixin):
-    def register(self, changelog_agent):
+    def register(self, register_time, changelog_agent):
         self.changelog_agent = changelog_agent
-        self.changelog_register_time = int(time.time())
+        self.changelog_register_time = register_time
+        self.history_crawl_start_time = register_time
         self.changelog_done_func = self.changelog_agent.history_done
+        self.history_turns = 0
 
-    def crawl(self):
+    def crawl(self, no_stime_update=False):
+        self.history_turns += 1
         self.update_worker_crawl_status("History Crawl")
+        logging.info('starting history crawl... turns: %s' %
+                     self.history_turns)
 
         # get stime (from the brick) and purge changelogs
         # that are _historical_ to that time.
@@ -1169,11 +1185,9 @@ class GMasterChangeloghistoryMixin(GMasterChangelogMixin):
             purge_time = None
 
         if not purge_time or purge_time == URXTIME:
+            logging.info("stime not available, abandoning history crawl")
             raise NoPurgeTimeAvailable()
 
-        logging.debug("Get changelog history between %s and %s" %
-                      (purge_time[0], self.changelog_register_time))
-
         # Changelogs backend path is hardcoded as
         # <BRICK_PATH>/.glusterfs/changelogs, if user configured to different
         # location then consuming history will not work(Known issue as of now)
@@ -1210,11 +1224,26 @@ class GMasterChangeloghistoryMixin(GMasterChangelogMixin):
                 logging.debug('processing changes %s' % repr(changes))
                 self.process(changes)
 
+        history_turn_time = int(time.time()) - self.history_crawl_start_time
+
+        logging.info('finished history crawl syncing between %s - %s.' %
+                     (purge_time[0], actual_end))
+
         # If TS returned from history_changelog is < register_time
         # then FS crawl may be required, since history is only available
         # till TS returned from history_changelog
         if actual_end < self.changelog_register_time:
-            raise PartialHistoryAvailable(str(actual_end))
+            if self.history_turns < 2:
+                sleep_time = 1
+                if history_turn_time < CHANGELOG_ROLLOVER_TIME:
+                    sleep_time = CHANGELOG_ROLLOVER_TIME - history_turn_time
+                time.sleep(sleep_time)
+                self.history_crawl_start_time = int(time.time())
+                self.crawl()
+            else:
+                # This exeption will be catched in resource.py and
+                # fallback to xsync for the small gap.
+                raise PartialHistoryAvailable(str(actual_end))
 
 
 class GMasterXsyncMixin(GMasterChangelogMixin):
@@ -1231,7 +1260,7 @@ class GMasterXsyncMixin(GMasterChangelogMixin):
 
     XSYNC_MAX_ENTRIES = 1 << 13
 
-    def register(self, changelog_agent=None):
+    def register(self, register_time, changelog_agent=None):
         self.counter = 0
         self.comlist = []
         self.stimes = []
@@ -1248,7 +1277,18 @@ class GMasterXsyncMixin(GMasterChangelogMixin):
             else:
                 raise
 
-    def crawl(self):
+        # After changelogs history processing completes, it switches
+        # to xsync/hibrid crawl if history actual end time is less than
+        # live changelog register time. Xsync should only run for that
+        # small gap, ie.. changelog_register_time - history_actual_end_time
+        # If we don't have upper limit to limit the XSync change detection
+        # It will keep on detecting the files even though changelogs are
+        # available for the same. Set upper limit during register
+        # and reset at the end of each crawl. Reseting at the end of
+        # crawl is required if change_detector is set to xsync.
+        self.xsync_upper_limit = (register_time, 0)
+
+    def crawl(self, no_stime_update=False):
         """
         event dispatcher thread
 
@@ -1265,19 +1305,36 @@ class GMasterXsyncMixin(GMasterChangelogMixin):
             try:
                 item = self.comlist.pop(0)
                 if item[0] == 'finale':
-                    logging.info('finished hybrid crawl syncing')
+                    if self.xsync_upper_limit is not None:
+                        logging.info('finished hybrid crawl syncing, endtime: '
+                                     '%s' % self.xsync_upper_limit[0])
+                    else:
+                        logging.info('finished hybrid crawl syncing')
+
                     break
                 elif item[0] == 'xsync':
                     logging.info('processing xsync changelog %s' % (item[1]))
                     self.process([item[1]], 0)
                 elif item[0] == 'stime':
-                    logging.debug('setting slave time: %s' % repr(item[1]))
-                    self.upd_stime(item[1][1], item[1][0])
+                    if not no_stime_update:
+                        # xsync is started after running history but if
+                        # history actual end time is less than register time
+                        # then if we update stime, live changelog processing
+                        # will skip the changelogs for which TS is less than
+                        # stime. During this deletes and renames are not
+                        # propogated. By not setting stime live changelog will
+                        # start processing from the register time. Since we
+                        # have xsync_upper_limit their will not be much
+                        # overlap/redo of changelogs.
+                        logging.debug('setting slave time: %s' % repr(item[1]))
+                        self.upd_stime(item[1][1], item[1][0])
                 else:
                     logging.warn('unknown tuple in comlist (%s)' % repr(item))
             except IndexError:
                 time.sleep(1)
 
+        self.xsync_upper_limit = None
+
     def write_entry_change(self, prefix, data=[]):
         self.fh.write("%s %s\n" % (prefix, ' '.join(data)))
 
@@ -1402,7 +1459,10 @@ class GMasterXsyncMixin(GMasterChangelogMixin):
                 self.write_entry_change("M", [gfid, "SETATTR", str(st.st_uid),
                                               str(st.st_gid), str(st.st_mode)])
                 self.Xcrawl(e, xtr_root)
-                self.stimes.append((e, xte))
+                stime_to_update = xte
+                if self.xsync_upper_limit is not None:
+                    stime_to_update = min(self.xsync_upper_limit, xte)
+                self.stimes.append((e, stime_to_update))
             elif stat.S_ISLNK(mo):
                 self.write_entry_change(
                     "E", [gfid, 'SYMLINK', escape(os.path.join(pargfid,
@@ -1426,7 +1486,10 @@ class GMasterXsyncMixin(GMasterChangelogMixin):
                                                                 bname))])
                 self.write_entry_change("D", [gfid])
         if path == '.':
-            self.stimes.append((path, xtl))
+            stime_to_update = xtl
+            if self.xsync_upper_limit is not None:
+                stime_to_update = min(self.xsync_upper_limit, xtl)
+            self.stimes.append((path, stime_to_update))
             self.sync_done(self.stimes, True)