adding subclasses for "tasks" and "stages" for uniform code execution

Author: timvaillancourt

mongodb_consistent_backup/Archive/Archive.py

@@ -5,5 +5,5 @@
 class Archive(Stage):
     def __init__(self, manager, config, timer, base_dir, backup_dir):
         super(Archive, self).__init__(self.__class__.__name__, manager, config, timer, base_dir, backup_dir)
-        self.method = self.config.archive.method
+        self.task = self.config.archive.method
         self.init()

mongodb_consistent_backup/Archive/Tar/Tar.py

@@ -2,12 +2,13 @@
 import logging
 
 from copy_reg import pickle
-from multiprocessing import Pool, cpu_count
+from multiprocessing import Pool
 from types import MethodType
 
 from TarThread import TarThread
 from mongodb_consistent_backup.Common import parse_method
-from mongodb_consistent_backup.Errors import Error, OperationError
+from mongodb_consistent_backup.Errors import Error
+from mongodb_consistent_backup.Pipeline import Task
 
 
 # Allows pooled .apply_async()s to work on Class-methods:
@@ -20,35 +21,13 @@ def _reduce_method(m):
 pickle(MethodType, _reduce_method)
 
 
-class Tar:
+class Tar(Task):
     def __init__(self, manager, config, timer, base_dir, backup_dir, **kwargs):
-        self.config          = config
-        self.backup_base_dir = backup_dir
-        self.verbose         = self.config.verbose
-        self.binary          = "tar"
+        super(Tar, self).__init__(self.__class__.__name__, manager, config, timer, base_dir, backup_dir, **kwargs)
+	self.compression_method = self.config.archive.tar.compression
+        self.binary             = "tar"
 
-        self._pool     = None
-        self.stopped   = False
-        self.completed = False
-
-    def compression(self, method=None):
-        if method:
-            self.config.archive.tar.compression = parse_method(method)
-            logging.info("Setting tar compression method to: %s" % self.config.archive.tar.compression)
-        return parse_method(self.config.archive.tar.compression)
-
-    def do_gzip(self):
-        if self.compression() == 'gzip':
-            return True
-        return False
-
-    def threads(self, thread_count=None):
-        if thread_count:
-            self.config.archive.tar.threads = int(thread_count)
-            logging.info("Setting tar thread count to: %i" % self.config.archive.tar.threads)
-        if self.config.archive.tar.threads is None or self.config.archive.tar.threads < 1:
-            self.config.archive.tar.threads = cpu_count()
-        return int(self.config.archive.tar.threads)
+        self._pool  = None
 
     def run(self):
         try:
@@ -59,10 +38,11 @@ def run(self):
             logging.fatal("Could not start pool! Error: %s" % e)
             raise Error(e)
 
-        if os.path.isdir(self.backup_base_dir):
+        if os.path.isdir(self.backup_dir):
             try:
-                for backup_dir in os.listdir(self.backup_base_dir):
-                    subdir_name = os.path.join(self.backup_base_dir, backup_dir)
+                self.running = True
+                for backup_dir in os.listdir(self.backup_dir):
+                    subdir_name = os.path.join(self.backup_dir, backup_dir)
                     if not os.path.isdir(os.path.join(subdir_name, "dump")):
                         continue
                     output_file = "%s.tar" % subdir_name
@@ -73,8 +53,10 @@ def run(self):
                 self._pool.terminate()
                 logging.fatal("Could not create tar archiving thread! Error: %s" % e)
                 raise Error(e)
-            self._pool.close()
-            self._pool.join()
+            finally:
+                self._pool.close()
+                self._pool.join()
+                self.stopped = True
             self.completed = True
 
     def close(self):

mongodb_consistent_backup/Backup/Backup.py

@@ -5,5 +5,5 @@
 class Backup(Stage):
     def __init__(self, manager, config, timer, base_dir, backup_dir, replsets, sharding=None):
         super(Backup, self).__init__(self.__class__.__name__, manager, config, timer, base_dir, backup_dir, replsets=replsets, sharding=sharding)
-        self.method = self.config.backup.method
+        self.task = self.config.backup.method
         self.init()

mongodb_consistent_backup/Backup/Mongodump/Mongodump.py

@@ -5,73 +5,52 @@
 from fabric.api import hide, settings, local
 from math import floor
 from multiprocessing import cpu_count
-from signal import signal, SIGINT, SIG_IGN
 from time import sleep
 
 from mongodb_consistent_backup.Common import MongoUri
 from mongodb_consistent_backup.Errors import Error, OperationError
 from mongodb_consistent_backup.Oplog import OplogState
+from mongodb_consistent_backup.Pipeline import Task
 
 from MongodumpThread import MongodumpThread
 
 
-class Mongodump:
-    def __init__(self, manager, config, timer, base_dir, backup_dir, **kwargs):
-        self.manager    = manager
-        self.config     = config
-        self.timer      = timer
-        self.base_dir   = base_dir
-        self.backup_dir = backup_dir
-        self.binary     = self.config.backup.mongodump.binary
-        self.user       = self.config.user
-        self.password   = self.config.password
-        self.authdb     = self.config.authdb
-        self.verbose    = self.config.verbose
+class Mongodump(Task):
+    def __init__(self, manager, config, timer, base_dir, backup_dir, replsets, sharding=None):
+        super(Mongodump, self).__init__(self.__class__.__name__, manager, config, timer, base_dir, backup_dir)
+        self.compression_method = self.config.backup.mongodump.compression
+        self.binary             = self.config.backup.mongodump.binary
+        self.user               = self.config.user
+        self.password           = self.config.password
+        self.authdb             = self.config.authdb
+        self.replsets           = replsets
+        self.sharding           = sharding
 
-        try:
-            self.replsets = kwargs['replsets']
-        except KeyError:
-            raise Error("'replsets' kwargs required!")
-
-        self.sharding = None
-        if 'sharding' in kwargs:
-            self.sharding = kwargs['sharding']
-
-        signal(SIGINT, SIG_IGN)
-        signal(SIGINT, self.close)
-
-        self.completed            = False
-        self.timer_name           = self.__class__.__name__
-        self.threads_per_dump_max = 16
-        self.config_replset       = False
-        self.cpu_count            = cpu_count()
-        self.threads              = []
-        self.states               = {}
-        self._summary             = {}
-        self._threads_per_dump    = None
+        self.threads_max    = 16
+        self.config_replset = False
+        self.dump_threads   = []
+        self.states         = {}
+        self._summary       = {}
 
         with hide('running', 'warnings'), settings(warn_only=True):
             self.version = local("%s --version|awk 'NR >1 {exit}; /version/{print $NF}'" % self.binary, capture=True)
-        self.do_gzip = self.can_gzip()
 
-        if not self.do_gzip and self.config.backup.mongodump.compression == 'gzip':
+        if self.can_gzip():
+            if self.compression() == 'none':
+                self.compression('gzip')
+        elif self.compression() == 'gzip':
             logging.warning("mongodump gzip compression requested on binary that does not support gzip!")
 
-        if not isinstance(self.replsets, dict):
-            raise Error("Field 'replsets' must be a dictionary of mongodb_consistent_backup.Replication.Replset classes!")
-
     def can_gzip(self):
         if os.path.isfile(self.binary) and os.access(self.binary, os.X_OK):
+            logging.debug("Mongodump binary supports gzip")
             if tuple("3.2.0".split(".")) <= tuple(self.version.split(".")):
                 return True
             return False
         else:
             logging.fatal("Cannot find or execute the mongodump binary file %s!" % self.binary)
             sys.exit(1)
 
-    def is_compressed(self):
-        return self.can_gzip()
-
     def summary(self):
         return self._summary
 
@@ -85,14 +64,14 @@ def get_summaries(self):
 
     def wait(self):
         completed = 0
-        start_threads = len(self.threads)
+        start_threads = len(self.dump_threads)
         # wait for all threads to finish
-        while len(self.threads) > 0:
-            for thread in self.threads:
+        while len(self.dump_threads) > 0:
+            for thread in self.dump_threads:
                 if not thread.is_alive():
                     if thread.exitcode == 0:
                         completed += 1
-                    self.threads.remove(thread)
+                    self.dump_threads.remove(thread)
             sleep(0.5)
 
         # sleep for 3 sec to fix logging order before gathering summaries
@@ -106,19 +85,19 @@ def wait(self):
         else:
             raise OperationError("Not all mongodump threads completed successfully!")
 
-    def threads_per_dump(self, threads=None):
+    def threads(self, threads=None):
         if threads:
-            self._threads_per_dump = int(threads)
-        elif not self._threads_per_dump:
+            self.thread_count = int(threads)
+        elif not self.thread_count:
             if tuple(self.version.split(".")) >= tuple("3.2.0".split(".")):
-                self._threads_per_dump = 1
+                self.thread_count = 1
                 if self.cpu_count > len(self.replsets):
-                    self._threads_per_dump = int(floor(self.cpu_count / len(self.replsets)))
-                    if self._threads_per_dump > self.threads_per_dump_max:
-                        self._threads_per_dump = self.threads_per_dump_max
+                    self.thread_count = int(floor(self.cpu_count / len(self.replsets)))
+                    if self.thread_count > self.threads_max:
+                        self.thread_count = self.threads_max
             else:
                 logging.warn("Threading unsupported by mongodump version %s. Use mongodump 3.2.0 or greater to enable per-dump threading." % self.version)
-        return self._threads_per_dump
+        return self.thread_count
 
     def run(self):
         self.timer.start(self.timer_name)
@@ -137,18 +116,18 @@ def run(self):
                 self.authdb,
                 self.backup_dir,
                 self.binary,
-                self.threads_per_dump(),
+                self.threads(),
                 self.do_gzip,
                 self.verbose
             )
-            self.threads.append(thread)
+            self.dump_threads.append(thread)
 
-        if not len(self.threads) > 0:
+        if not len(self.dump_threads) > 0:
             raise OperationError('No backup threads started!')
 
         logging.info(
-            "Starting backups using mongodump %s (options: gzip=%s, threads_per_dump=%i)" % (self.version, str(self.do_gzip), self.threads_per_dump()))
-        for thread in self.threads:
+            "Starting backups using mongodump %s (options: compression=%s, threads_per_dump=%i)" % (self.version, self.compression(), self.threads()))
+        for thread in self.dump_threads:
             thread.start()
         self.wait()
 
@@ -159,7 +138,7 @@ def run(self):
                 logging.info("Using non-replset backup method for config server mongodump")
                 mongo_uri = MongoUri(config_server['host'], 27019, 'configsvr')
                 self.states['configsvr'] = OplogState(self.manager, mongo_uri)
-                self.threads = [MongodumpThread(
+                self.dump_threads = [MongodumpThread(
                     self.states['configsvr'],
                     mongo_uri,
                     self.timer,
@@ -168,20 +147,20 @@ def run(self):
                     self.authdb,
                     self.backup_dir,
                     self.binary,
-                    self.threads_per_dump(),
+                    self.threads(),
                     self.do_gzip,
                     self.verbose
                 )]
-                self.threads[0].start()
+                self.dump_threads[0].start()
                 self.wait()
 
         self.completed = True
         return self._summary
 
     def close(self):
         logging.info("Stopping all mongodump threads")
-        if len(self.threads) > 0:
-            for thread in self.threads:
+        if len(self.dump_threads) > 0:
+            for thread in self.dump_threads:
                 thread.terminate()
         try:
             self.timer.stop(self.timer_name)

mongodb_consistent_backup/Main.py

@@ -271,7 +271,7 @@ def run(self):
                     self.replsets
                 )
                 if self.backup.is_compressed():
-                    logging.info("Backup method supports gzip compression, disabling compression in archive step")
+                    logging.info("Backup method supports compression, disabling compression in archive step")
                     self.archive.compression('none')
                 self.backup_summary = self.backup.run()
                 self.state.set('backup_oplog', self.backup_summary)
@@ -321,8 +321,9 @@ def run(self):
                     self.manager,
                     self.config,
                     self.timer,
-                    self.replsets,
-                    self.backup_directory
+                    self.backup_root_subdirectory,
+                    self.backup_directory,
+                    self.replsets
                 )
             except Exception, e:
                 self.exception("Problem initializing oplog tailer! Error: %s" % e, e)
@@ -339,9 +340,9 @@ def run(self):
                     self.sharding
                 )
                 if self.backup.is_compressed():
-                    logging.info("Backup method supports gzip compression, disabling compression in archive step and enabling oplog compression")
+                    logging.info("Backup method supports compression, disabling compression in archive step and enabling oplog compression")
                     self.archive.compression('none')
-                    self.oplogtailer.compression('gzip')
+                    self.oplogtailer.compression(self.backup.compression())
             except Exception, e:
                 self.exception("Problem initializing backup! Error: %s" % e, e)
 
@@ -387,8 +388,16 @@ def run(self):
                 self.db.close()
 
             # resolve/merge tailed oplog into mongodump oplog.bson to a consistent point for all shards
-            if self.backup.method == "mongodump" and self.oplogtailer:
-                self.resolver = Resolver(self.manager, self.config, self.timer, self.oplog_summary, self.backup_summary)
+            if self.backup.task.lower() == "mongodump" and self.oplogtailer:
+                self.resolver = Resolver(
+                    self.manager,
+                    self.config,
+                    self.timer,
+                    self.backup_root_subdirectory,
+                    self.backup_directory,
+		    self.oplog_summary,
+		    self.backup_summary
+		)
                 self.resolver.compression(self.oplogtailer.compression())
                 resolver_summary = self.resolver.run()
                 for shard in resolver_summary:

mongodb_consistent_backup/Notify/Notify.py

@@ -7,7 +7,7 @@
 class Notify(Stage):
     def __init__(self, manager, config, timer, base_dir, backup_dir):
         super(Notify, self).__init__(self.__class__.__name__, manager, config, timer, base_dir, backup_dir)
-        self.method = self.config.notify.method
+        self.task = self.config.notify.method
 
         self.completed = False
         self.notifications = []
@@ -18,21 +18,21 @@ def notify(self, message, success=False):
         self.notifications.append(notification)
 
     def run(self, *args):
-        if self._method and len(self.notifications) > 0:
-            logging.info("Sending %i notification(s) to: %s" % (len(self.notifications), self._method.server))
+        if self._task and len(self.notifications) > 0:
+            logging.info("Sending %i notification(s) to: %s" % (len(self.notifications), self._task.server))
             self.timers.start(self.stage)
             while len(self.notifications) > 0:
                 try:
                     (success, message) = self.notifications.pop()
-                    state = self._method.failed
+                    state = self._task.failed
                     if success:
-                        state = self._method.success
-                    self._method.notify(success, message)
+                        state = self._task.success
+                    self._task.run(success, message)
                 except:
                     continue
             self.timers.stop(self.stage)
         self.completed = True
 
     def close(self):
-        if self._method:
-            return self._method.close()
+        if self._task:
+            return self._task.close()