Merge pull request #84 from timvaillancourt/MCB_1.0-cleanup

timvaillancourt · web-flow · commit 1fef902cc3b9 · 2017-01-24T17:32:05.000+01:00
Mcb 1.0 cleanup: changes to get backups running again
diff --git a/.gitignore b/.gitignore
@@ -1,5 +1,6 @@
 bin
 build
 rpmbuild
+tmp
 *.pyc
 .idea
diff --git a/Makefile b/Makefile
@@ -34,4 +34,4 @@ rpm: clean
 	rpmbuild -D "_topdir $(PWD)/rpmbuild" -D "version $(VERSION)" -bb rpmbuild/SPECS/$(NAME).spec
 
 clean:
-	rm -rf bin build rpmbuild $(NAME).egg-info 2>/dev/null
+	rm -rf bin build rpmbuild $(NAME).egg-info tmp 2>/dev/null
diff --git a/VERSION b/VERSION
@@ -1 +1 @@
-1.0.0
+1.0.0-MCB_1.0
diff --git a/conf/example.yml b/conf/example.yml
@@ -10,7 +10,7 @@ production:
     location: /opt/mongodb/backup
     mongodump:
       binary: /usr/bin/mongodump
-      compression: gzip
+    #  compression: gzip
   replication:
     max_lag_secs: 5
     min_priority: 0
@@ -21,12 +21,13 @@ production:
       wait_secs: 300
       ping_secs: 3
   oplog:
-    # compression: gzip
-    # resolver_threads: 4
+    #compression: gzip
+    #resolver_threads: 4
   archive:
     method: tar
-    compression: gzip
-    # threads: 2
+    #tar:
+    #  compression: gzip
+    #  threads: 2
   notify:
     method: none
     #nsca:
diff --git a/mongodb_consistent_backup/Archive/Archive.py b/mongodb_consistent_backup/Archive/Archive.py
@@ -1,33 +1,46 @@
 import logging
 
+from Tar import Tar
+
 
 class Archive:
     def __init__(self, config, backup_dir):
         self.config     = config
         self.backup_dir = backup_dir
 
+        self.method    = None
         self._archiver = None
         self.init()
 
     def init(self):
         archive_method = self.config.archive.method
-        if archive_method is None:
+        if not archive_method or archive_method.lower() == "none":
             logging.info("Archiving disabled, skipping")
         else:
-            config_vars = ""
-            for key in self.config.archive:
-                config_vars += "%s=%s," % (key, self.config.archive[key])
-            logging.info("Using archiving method: %s (options: %s)" % archive_method, config_vars[:-1])
+            self.method = archive_method.lower()
+            logging.info("Using archiving method: %s" % self.method)
             try:
-                self._archiver = globals()[archive_method](
+                self._archiver = globals()[self.method.capitalize()](
                     self.config,
                     self.backup_dir
                 )
             except Exception, e:
                 raise e
 
+    def compression(self, method=None):
+        if self._archiver:
+            return self._archiver.compression(method)
+
+    def threads(self, threads=None):
+        if self._archiver:
+            return self._archiver.threads(threads)
+
     def archive(self):
         if self._archiver:
+            config_vars = ""
+            for key in self.config.archive:
+                config_vars += "%s=%s," % (key, self.config.archive[key])
+            logging.info("Archiving with method: %s (options: %s)" % (self.method, str(config_vars[:-1])))
             return self._archiver.run()
 
     def close(self):
diff --git a/mongodb_consistent_backup/Archive/Tar/Tar.py b/mongodb_consistent_backup/Archive/Tar/Tar.py
@@ -22,34 +22,44 @@ class Tar:
     def __init__(self, config, backup_base_dir):
         self.config          = config
         self.backup_base_dir = backup_base_dir
-        self.compression     = self.config.archive.compression
-        self.thread_count    = self.config.archive.threads
         self.verbose         = self.config.verbose
         self.binary          = "tar"
+        self.do_gzip         = False
+        self._pool           = None
 
-        if self.thread_count is None or self.thread_count < 1:
-            self.thread_count = cpu_count()
+    def compression(self, method=None):
+        if method:
+            self.config.archive.tar.compression = method.lower()
+        return self.config.archive.tar.compression
 
+    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)
+
+    def run(self):
         try:
-            self._pool = Pool(processes=self.thread_count)
+            thread_count = self.threads()
+            self._pool   = Pool(processes=thread_count)
+            logging.info("Archiving backup directories with pool of %i thread(s)" % thread_count)
         except Exception, e:
             logging.fatal("Could not start pool! Error: %s" % e)
             raise e
 
-    def run(self):
-        logging.info("Archiving backup directories with pool of %i thread(s)" % self.thread_count)
         if os.path.isdir(self.backup_base_dir):
             try:
                 for backup_dir in os.listdir(self.backup_base_dir):
                     subdir_name = "%s/%s" % (self.backup_base_dir, backup_dir)
                     output_file = "%s.tar" % subdir_name
 
-                    do_gzip = False
-                    if self.compression == "gzip":
-                        output_file = "%s.tgz" % subdir_name
-                        do_gzip = True
+                    if self.compression() == 'gzip':
+                        output_file  = "%s.tgz" % subdir_name
+                        self.do_gzip = True
 
-                    self._pool.apply_async(TarThread(subdir_name, output_file, do_gzip, self.verbose, self.binary).run)
+                    self._pool.apply_async(TarThread(subdir_name, output_file, self.do_gzip, self.verbose, self.binary).run)
             except Exception, e:
                 self._pool.terminate()
                 logging.fatal("Could not create archiving thread! Error: %s" % e)
@@ -59,8 +69,8 @@ def run(self):
         logging.info("Archiver threads completed")
 
     def close(self):
-        logging.info("Killing all Archiver threads...")
+        logging.debug("Stopping Archiver threads")
         if self._pool is not None:
             self._pool.terminate()
             self._pool.join()
-        logging.info("Killed all Archiver threads")
+        logging.info("Stopped all Archiver threads")
diff --git a/mongodb_consistent_backup/Archive/Tar/__init__.py b/mongodb_consistent_backup/Archive/Tar/__init__.py
@@ -2,6 +2,8 @@
 
 
 def config(parser):
-    parser.add_argument("--archive.compression", dest="archive.compression",
+    parser.add_argument("--archive.tar.compression", dest="archive.tar.compression",
                         help="Archiver compression method (default: gzip)", default='gzip', choices=['gzip', 'none'])
+    parser.add_argument("--archive.tar.threads", dest="archive.tar.threads", 
+                        help="Number of threads to use in archive phase (default: 1-per-CPU)", default=0, type=int)
     return parser
diff --git a/mongodb_consistent_backup/Archive/__init__.py b/mongodb_consistent_backup/Archive/__init__.py
@@ -3,5 +3,4 @@
 
 def config(parser):
     parser.add_argument("--archive.method", dest="archive.method", help="Archiver method (default: tar)", default='tar', choices=['tar','none'])
-    parser.add_argument("--archive.threads", dest="archive.threads", help="Number of threads to use in archive phase (default: 1-per-CPU)", default=0, type=int)
     return parser
diff --git a/mongodb_consistent_backup/Backup/Backup.py b/mongodb_consistent_backup/Backup/Backup.py
@@ -1,5 +1,7 @@
 import logging
 
+from Mongodump import Mongodump
+
 
 class Backup:
     def __init__(self, config, backup_dir, secondaries, config_server=None):
@@ -13,18 +15,19 @@ def __init__(self, config, backup_dir, secondaries, config_server=None):
 
     def init(self):
         backup_method = self.config.backup.method
-        if backup_method is None:
+        if not backup_method or backup_method.lower() == "none":
             raise Exception, 'Must specify a backup method!', None
-        logging.info("Using backup method: %s" % backup_method)
+        method = backup_method.lower()
+        logging.info("Using backup method: %s" % method)
         try:
-            self._method = globals()[backup_method](
+            self._method = globals()[method.capitalize()](
                 self.config,
                 self.backup_dir,
                 self.secondaries,
                 self.config_server
             )
         except Exception, e:
-            raise Exception, "Problem performing %s! Error: %s" % (backup_method, e), None
+            raise Exception, "Problem performing %s! Error: %s" % (method, e), None
 
     def is_compressed(self):
         if self._method:
diff --git a/mongodb_consistent_backup/Backup/Mongodump/Mongodump.py b/mongodb_consistent_backup/Backup/Mongodump/Mongodump.py
@@ -124,16 +124,16 @@ def run(self):
 
         # start all threads and wait
         logging.info(
-            "Starting backups using mongodump %s (inline gzip: %s, threads per dump: %i)" % (self.version, str(self.dump_gzip), self.threads_per_dump))
+            "Starting backups using mongodump %s (inline gzip: %s, threads per dump: %i)" % (self.version, str(self.do_gzip), self.threads_per_dump))
         for thread in self.threads:
             thread.start()
         self.wait()
 
         # backup a single non-replset config server, if exists:
         if not self.config_replset and isinstance(self.config_server, dict):
             logging.info("Using non-replset backup method for config server mongodump")
-	    config_server = self.config_server['host']
-	    if not ":" in config_server:
+            config_server = self.config_server['host']
+            if not ":" in config_server:
                 config_server = config_server+":27019"
             self.threads = [MongodumpThread(
                 self.response_queue,
diff --git a/mongodb_consistent_backup/Common/Config.py b/mongodb_consistent_backup/Common/Config.py
@@ -71,18 +71,18 @@ def _get(self, keys, data=None):
             return data[keys]
 
     def parse_submodules(self):
-	for _, modname, ispkg in walk_packages(path="."):
-	    if ispkg:
-	        try:
-		    components = modname.split('.')
-		    mod = __import__(components[0])
-		    for comp in components[1:]:
-		        mod = getattr(mod, comp)
-		    mod.config(self._config.parser)
-		except AttributeError, e:
-		    continue
-	        except Exception, e:
-		    raise e
+        for _, modname, ispkg in walk_packages(path="."):
+            if ispkg:
+                try:
+                    components = modname.split('.')
+                    mod = __import__(components[0])
+                    for comp in components[1:]:
+                        mod = getattr(mod, comp)
+                    mod.config(self._config.parser)
+                except AttributeError, e:
+                    continue
+                except Exception, e:
+                    raise e
 
     def check_required(self):
         required = [
diff --git a/mongodb_consistent_backup/Main.py b/mongodb_consistent_backup/Main.py
@@ -152,11 +152,18 @@ def run(self):
         except Exception, e:
             raise e
 
+        # Setup the archiver
+        try:
+            self.archive = Archive(
+                self.config,
+                self.backup_root_directory, 
+            )
+        except Exception, e:
+            raise e
+
         if not self.is_sharded:
             logging.info("Running backup of %s:%s in replset mode" % (self.config.host, self.config.port))
 
-            self.config.archive.threads = 1
-
             # get shard secondary
             try:
                 self.replset = Replset(
@@ -179,12 +186,13 @@ def run(self):
                 )
                 self.backup.backup()
                 if self.backup.is_compressed():
-                    logging.info("Backup method supports gzip compression, setting config overrides: { archive.compression: 'none' }")
-                    self.config.archive.compression = 'none'
-                    self.config.oplog.compression = 'gzip'
+                    logging.info("Backup method supports gzip compression, disabling compression in archive step")
+                    self.archive.compression('none')
             except Exception, e:
                 self.exception("Problem performing replset mongodump! Error: %s" % e)
 
+            # use 1 archive thread for single replset
+            self.archive.threads(1)
         else:
             logging.info("Running backup of %s:%s in sharded mode" % (self.config.host, self.config.port))
 
@@ -215,6 +223,16 @@ def run(self):
             except Exception, e:
                 self.exception("Problem stopping the balancer! Error: %s" % e)
 
+            # init the oplogtailers
+            try:
+                self.oplogtailer = Tailer(
+                    self.config,
+                    self.secondaries,
+                    self.backup_root_directory
+                )
+            except Exception, e:
+                self.exception("Problem initializing oplog tailer! Error: %s" % e)
+
             # init the backup
             try:
                 self.backup = Backup(
@@ -224,19 +242,14 @@ def run(self):
                     self.sharding.get_config_server()
                 )
                 if self.backup.is_compressed():
-                    logging.info("Backup method supports gzip compression, setting config overrides: { archive.compression: 'none', oplog.compression: 'gzip' }")
-                    self.config.archive.compression = 'none'
-                    self.config.oplog.compression = 'gzip'
+                    logging.info("Backup method supports gzip compression, disabling compression in archive step and enabling oplog compression")
+                    self.archive.compression('none')
+                    self.oplogtailer.compression('gzip')
             except Exception, e:
                 self.exception("Problem initializing backup! Error: %s" % e)
 
-            # start the oplog tailer(s)
+            # start the oplog tailers, before the backups start
             try:
-                self.oplogtailer = Tailer(
-                    self.config,
-                    self.secondaries,
-                    self.backup_root_directory
-                )
                 self.oplogtailer.run()
             except Exception, e:
                 self.exception("Failed to start oplog tailing threads! Error: %s" % e)
@@ -264,10 +277,6 @@ def run(self):
 
         # archive backup directories
         try:
-            self.archive = Archive(
-                self.config,
-                self.backup_root_directory, 
-            )
             self.archive.archive()
         except Exception, e:
             self.exception("Problem performing archiving! Error: %s" % e)
diff --git a/mongodb_consistent_backup/Notify/Notify.py b/mongodb_consistent_backup/Notify/Notify.py
@@ -2,21 +2,22 @@
 
 class Notify:
     def __init__(self, config):
-        self.config = config
+        self.config    = config
+
         self._notifier = None
         self.init()
 
     def init(self):
         notify_method = self.config.notify.method
-        if notify_method is None:
+        if not notify_method or notify_method.lower() == "none":
             logging.info("Notifying disabled, skipping")
-
-        logging.info("Using notify method: %s" % notify_method)
-        try:
-            self._notifier = globals()[notify_method](self.config)
-        except Exception, e:
-            raise e
-
+        else:
+            method = notify_method.lower()
+            logging.info("Using notify method: %s" % method)
+            try:
+                self._notifier = globals()[method.capitalize()](self.config)
+            except Exception, e:
+                raise e
 
     def notify(self, message, success=False):
         if self._notifier:
diff --git a/mongodb_consistent_backup/Oplog/Tailer/Tailer.py b/mongodb_consistent_backup/Oplog/Tailer/Tailer.py
diff --git a/mongodb_consistent_backup/Upload/Upload.py b/mongodb_consistent_backup/Upload/Upload.py
diff --git a/scripts/build.sh b/scripts/build.sh

-Original file line number
+Diff line change
@@ @@ -1,5 +1,6 @@ @@
 bin
 build
 rpmbuild
 +tmp
 *.pyc
 .idea