Use local cache for pip+pex builds (faster), use methods to change config settings instead of directly changing vars in Main.py

Author: timvaillancourt

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:

mongodb_consistent_backup/Archive/Archive.py

@@ -8,29 +8,39 @@ 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 not archive_method or archive_method.lower() is "none":
+        if not archive_method or archive_method.lower() == "none":
             logging.info("Archiving disabled, skipping")
         else:
-            config_vars = ""
-            method = archive_method.lower()
-            for key in self.config.archive:
-                config_vars += "%s=%s," % (key, self.config.archive[key])
-            logging.info("Using archiving method: %s (options: %s)" % (method, str(config_vars[:-1])))
+            self.method = archive_method.lower()
+            logging.info("Using archiving method: %s" % self.method)
             try:
-                self._archiver = globals()[method.capitalize()](
+                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):

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.tar.compression
-        self.thread_count    = self.config.archive.tar.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")

mongodb_consistent_backup/Backup/Backup.py

@@ -15,7 +15,7 @@ def __init__(self, config, backup_dir, secondaries, config_server=None):
 
     def init(self):
         backup_method = self.config.backup.method
-        if not backup_method or backup_method.lower() is "none":
+        if not backup_method or backup_method.lower() == "none":
             raise Exception, 'Must specify a backup method!', None
         method = backup_method.lower()
         logging.info("Using backup method: %s" % method)

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)

mongodb_consistent_backup/Notify/Notify.py

@@ -9,7 +9,7 @@ def __init__(self, config):
 
     def init(self):
         notify_method = self.config.notify.method
-        if not notify_method or notify_method.lower() is "none":
+        if not notify_method or notify_method.lower() == "none":
             logging.info("Notifying disabled, skipping")
         else:
             method = notify_method.lower()

mongodb_consistent_backup/Oplog/Tailer/Tailer.py

@@ -16,14 +16,20 @@ def __init__(self, config, secondaries, base_dir):
         self.password    = self.config.password
         self.authdb      = self.config.authdb
 
-        self.dump_gzip = False
-        if self.config.oplog.compression == 'gzip':
-            self.dump_gzip = True
-
         self.response_queue = Queue()
         self.threads        = []
         self._summary       = {}
 
+    def compression(self, method=None):
+        if method:
+	    self.config.oplog.compression = method.lower()
+	return self.config.oplog.compression
+
+    def do_gzip(self):
+	if self.compression() == 'gzip':
+	    return True
+        return False
+
     def summary(self):
         return self._summary
 
@@ -38,7 +44,7 @@ def run(self):
                 self.base_dir,
                 host,
                 port,
-                self.dump_gzip,
+                self.do_gzip(),
                 self.user,
                 self.password,
                 self.authdb

mongodb_consistent_backup/Upload/Upload.py

@@ -14,7 +14,7 @@ def __init__(self, config, base_dir, backup_dir):
 
     def init(self):
         upload_method = self.config.upload.method
-        if not upload_method or upload_method.lower() is "none":
+        if not upload_method or upload_method.lower() == "none":
             logging.info("Uploading disabled, skipping")
         else:
             #TODO Remove this line and move to  S3 Lib for checking

scripts/build.sh

@@ -96,8 +96,12 @@ if [ -d ${srcdir} ]; then
 		exit 1
 	fi
 
+	if [ ! -d ${pexdir} ]; then
+		mkdir -p ${pexdir}
+	else
+		rm -f ${pexdir}/build/mongodb_consistent_backup-* 
+	fi
 	[ ! -d ${bindir} ] && mkdir -p ${bindir}
-	[ ! -d ${pexdir} ] && mkdir -p ${pexdir}
 	${venvdir}/bin/python2.7 ${venvdir}/bin/pex -o ${output_file} -m ${mod_name} -r ${require_file} --pex-root=${pexdir}  ${builddir}
 	if [ $? -lt 1 ] && [ -x ${output_file} ]; then
 		echo "pex executable written to '$output_file'"