Merge branch 'master' into dockerfile_v1

Author: timvaillancourt

.pullapprove.yml

@@ -12,7 +12,7 @@
 # noinspection PyStringFormat
 class Dump(Process):
     def __init__(self, response_queue, backup_name, host_port, user, password, authdb, base_dir, binary,
-                 dump_gzip=False, verbose=False):
+                 threads=0, dump_gzip=False, verbose=False):
         Process.__init__(self)
         self.host, port     = host_port.split(":")
         self.host_port      = host_port
@@ -24,6 +24,7 @@ def __init__(self, response_queue, backup_name, host_port, user, password, authd
         self.authdb         = authdb
         self.base_dir       = base_dir
         self.binary         = binary
+        self.threads        = threads
         self.dump_gzip      = dump_gzip
         self.verbose        = verbose
 
@@ -52,6 +53,8 @@ def run(self):
         ))
 
         mongodump_flags = ["-h", self.host_port, "--oplog", "-o", "%s/dump" % self.backup_dir]
+        if self.threads > 0:
+            mongodump_flags.extend(["--numParallelCollections="+str(self.threads)])
         if self.dump_gzip:
             mongodump_flags.extend(["--gzip"])
         if self.authdb and self.authdb != "admin":

MongoBackup/Methods/Dump.py

@@ -1,7 +1,8 @@
 import logging
 
 from fabric.api import hide, settings, local
-from multiprocessing import Queue
+from math import floor
+from multiprocessing import Queue, cpu_count
 from time import sleep
 
 
@@ -22,6 +23,7 @@ def __init__(self, secondaries, base_dir, binary, dump_gzip=False, user=None, pa
         self.verbose       = verbose
 
         self.config_replset = False
+        self.cpu_count      = cpu_count()
         self.response_queue = Queue()
         self.threads        = []
         self._summary       = {}
@@ -65,6 +67,18 @@ def wait(self):
             raise Exception, "Not all mongodump threads completed successfully!", None
 
     def run(self):
+        # decide how many parallel dump workers to use based on cpu count vs # of shards (if 3.2+), 8 max workers max to protect the db
+        self.threads_per_dump     = 0
+        self.threads_per_dump_max = 8
+        if tuple(self.version.split(".")) >= tuple("3.2.0".split(".")): 
+            self.threads_per_dump = 1
+            if self.cpu_count > len(self.secondaries):
+                self.threads_per_dump = int(floor(self.cpu_count / len(self.secondaries)))
+                if self.threads_per_dump > self.threads_per_dump_max:
+                    self.threads_per_dump = self.threads_per_dump_max
+        else:
+            logging.warn("Threading unsupported by mongodump version %s. Use mongodump 3.2.0 or greater to enable per-dump threading." % self.version)
+
         # backup a secondary from each shard:
         for shard in self.secondaries:
             secondary = self.secondaries[shard]
@@ -77,6 +91,7 @@ def run(self):
                 self.authdb,
                 self.base_dir,
                 self.binary,
+                self.threads_per_dump,
                 self.dump_gzip,
                 self.verbose
             )
@@ -87,23 +102,27 @@ def run(self):
 
         # start all threads and wait
         logging.info(
-            "Starting backups in threads using mongodump %s (inline gzip: %s)" % (self.version, str(self.dump_gzip)))
+                "Starting backups using mongodump %s (inline gzip: %s, threads per dump: %i)" % (self.version, str(self.dump_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 = config_server+":27019"
             self.threads = [Dump(
                 self.response_queue,
                 'configsvr',
-                self.config_server['host'],
+                config_server,
                 self.user,
                 self.password,
                 self.authdb,
                 self.base_dir,
                 self.binary,
+                self.threads_per_dump,
                 self.dump_gzip,
                 self.verbose
             )]

MongoBackup/Methods/Dumper.py

@@ -67,7 +67,7 @@ def run(self):
                         logging.info("No oplog changes to resolve for %s:%s" % (host, port))
                     elif backup_oplog['last_ts'] > tailed_oplog['last_ts']:
                         logging.fatal(
-                            "Backup oplog is newer than the tailed oplog! This situation is unsupported. Please retry backup")
+                            "Backup oplog is newer than the tailed oplog, this situation is unsupported! Please retry backup.\nBackup oplog: %s\nTailed oplog: %s" % (backup_oplog, tailed_oplog))
                         raise Exception, "Backup oplog is newer than the tailed oplog!", None
                     else:
                         try:

MongoBackup/Oplog/Resolver.py

@@ -127,7 +127,10 @@ def get_config_server(self, force=False):
         if force or not self.config_server:
             configdb_hosts = self.get_configdb_hosts()
             try:
-                config_host, config_port = configdb_hosts[0].split(":")
+                config_host = configdb_hosts[0]
+                config_port = 27019
+                if ":" in configdb_hosts[0]:
+                    config_host, config_port = configdb_hosts[0].split(":")
                 validate_hostname(config_host)
                 logging.info("Found sharding config server: %s:%s" % (config_host, config_port))
 

MongoBackup/Sharding.py

@@ -1 +1 @@
-0.4.0
+0.4.0

VERSION

@@ -85,14 +85,14 @@ if [ -d ${srcdir} ]; then
 	fi
 	source ${venvdir}/bin/activate
 
-	${venvdir}/bin/pip install pex requests
+	${venvdir}/bin/python2.7 ${venvdir}/bin/pip install pex requests
 	if [ $? -gt 0 ]; then
 		echo "Failed to install pex utility for building!"
 		exit 1
 	fi
 
 	[ ! -d ${bindir} ] && mkdir -p ${bindir}
-	${venvdir}/bin/pex --disable-cache -o ${output_file} -m ${py_entry_point} -r ${require_file} ${builddir}
+	${venvdir}/bin/python2.7 ${venvdir}/bin/pex --disable-cache -o ${output_file} -m ${py_entry_point} -r ${require_file} ${builddir}
 	if [ $? -lt 1 ] && [ -x ${output_file} ]; then
 		echo "pex executable written to '$output_file'"
 		rm -rf ${builddir}