unbroke non-replset mode

Author: Tim Vaillancourt

MongoBackup/Methods/Dumper.py

@@ -33,6 +33,32 @@ def __init__(self, secondaries, base_dir, binary, dump_gzip=False, user=None, pa
     def summary(self):
         return self._summary
 
+    def wait(self):
+        # wait for all threads to finish
+        for thread in self.threads:
+            thread.join()
+
+        # sleep for 3 sec to fix logging order
+        sleep(3)
+
+        # get oplog summaries from the queue
+        completed = 0
+        while not self.response_queue.empty():
+            backup = self.response_queue.get()
+            host = backup['host']
+            port = backup['port']
+            if host not in self._summary:
+                self._summary[host] = {}
+            self._summary[host][port] = backup
+            if backup['completed']:
+                completed += 1
+
+        # check if all threads completed
+        if completed == len(self.threads):
+            logging.info("All mongodump backups completed")
+        else:
+            raise Exception, "Not all mongodump threads completed successfully!", None
+
     def run(self):
         # backup a secondary from each shard:
         for shard in self.secondaries:
@@ -51,8 +77,8 @@ def run(self):
             )
             self.threads.append(thread)
 
-        # backup a single config server:
-        if self.config_server and self.config_server['replset']:
+        # backup a single replica-set config server alongside the shards, if exists:
+        if self.config_server and self.config_server['replSet']:
             thread = Dump(
                 self.response_queue,
                 'config',
@@ -70,36 +96,33 @@ def run(self):
         if not len(self.threads) > 0:
             raise Exception, 'No backup threads started!', None
 
-        # start all threads
+        # start all threads and wait
         logging.info(
             "Starting backups in threads using mongodump %s (inline gzip: %s)" % (self.version, str(self.dump_gzip)))
         for thread in self.threads:
             thread.start()
+        self.wait()
 
-        # wait for all threads to finish
-        for thread in self.threads:
-            thread.join()
-
-        # sleep for 3 sec to fix logging order
-        sleep(3)
-
-        # get oplog summaries from the queue
-        completed = 0
-        while not self.response_queue.empty():
-            backup = self.response_queue.get()
-            host = backup['host']
-            port = backup['port']
-            if host not in self._summary:
-                self._summary[host] = {}
-            self._summary[host][port] = backup
-            if backup['completed']:
-                completed += 1
-
-        # check if all threads completed
-        if completed == len(self.threads):
-            logging.info("All mongodump backups completed")
-        else:
-            raise Exception, "Not all mongodump threads completed successfully!", None
+        # backup a single non-replset config server, if exists:
+        if self.config_server and not self.config_server['replSet']:
+            logging.debug("Using non-replset backup method for config server")
+            thread = Dump(
+                self.response_queue,
+                'config',
+                self.config_server['host'],
+                self.user,
+                self.password,
+                self.authdb,
+                self.base_dir,
+                self.binary,
+                self.dump_gzip,
+                self.verbose
+            )
+            self.threads = [thread]
+            self.threads[0].start()
+            if not len(self.threads) == 1:
+                raise Exception, 'No backup threads started!', None
+            self.wait()
 
         return self._summary
 

MongoBackup/Replset.py

@@ -11,6 +11,7 @@ def __init__(self, db, user=None, password=None, authdb='admin', max_lag_secs=5)
         self.authdb       = authdb
         self.max_lag_secs = max_lag_secs
 
+        self.rs_config = None
         self.rs_status = None
         self.primary   = None
         self.secondary = None
@@ -36,21 +37,23 @@ def get_rs_status(self, force=False, quiet=False):
         except Exception, e:
             raise Exception, "Error getting replica set status! Error: %s" % e, None
 
-    def get_rs_config(self, quiet=False):
-        try:
-            if self.db.server_version() >= tuple("3.0.0".split(".")):
-                output = self.db.admin_command('replSetGetConfig', quiet)
-                return output['config']
-            else:
-                return self.connection['local'].system.replset.find_one()
-        except Exception, e:
-            raise Exception, "Error getting replica set config! Error: %s" % e, None
+    def get_rs_config(self, force=False, quiet=False):
+        if force or not self.rs_config:
+            try:
+                if self.db.server_version() >= tuple("3.0.0".split(".")):
+                    output = self.db.admin_command('replSetGetConfig', quiet)
+                    self.rs_config = output['config']
+                else:
+                    self.rs_config = self.connection['local'].system.replset.find_one()
+            except Exception, e:
+                raise Exception, "Error getting replica set config! Error: %s" % e, None
+        return self.rs_config
 
     def rs_name(self):
         return self.get_rs_status['set']
 
-    def find_primary(self, force=False):
-        rs_status = self.get_rs_status(force)
+    def find_primary(self, force=False, quiet=False):
+        rs_status = self.get_rs_status(force, quiet)
         rs_name   = rs_status['set']
         for member in rs_status['members']:
             if member['stateStr'] == 'PRIMARY' and member['health'] > 0:
@@ -71,9 +74,9 @@ def find_primary(self, force=False):
             raise Exception, "Unable to locate a PRIMARY member for replset %s, giving up" % rs_name, None
         return self.primary
 
-    def find_secondary(self):
-        rs_status    = self.get_rs_status()
-        rs_config    = self.get_rs_config()
+    def find_secondary(self, force=False, quiet=False):
+        rs_status    = self.get_rs_status(force, quiet)
+        rs_config    = self.get_rs_config(force, quiet)
         rs_name      = rs_status['set']
         quorum_count = ceil(len(rs_status['members']) / 2.0)
 
@@ -125,6 +128,10 @@ def find_secondary(self):
                 secondary_count,
                 quorum_count
             ))
+
+            import pprint
+            pprint.pprint(rs_status)
+
             raise Exception, "Not enough secondaries in replset %s to safely take backup!" % rs_name, None
 
         logging.info("Choosing SECONDARY %s for replica set %s (score: %i)" % (self.secondary['host'], rs_name, self.secondary['score']))

MongoBackup/ReplsetSharded.py

@@ -1,6 +1,7 @@
 import logging
 
 from Common import DB
+from Replset import Replset
 from Sharding import Sharding
 
 
@@ -52,7 +53,7 @@ def get_replsets(self, force=False):
                     rs_db = self.get_replset_connection(host, port)
                     self.replsets[shard_name] = Replset(rs_db, self.user, self.password, self.authdb, self.max_lag_secs)
                 except Exception, e:
-                    logging.fatal("Could not get Replset class object for replset %s! Error: %s" % (rs_name, e))
+                    logging.fatal("Could not get Replset class object for replset %s! Error: %s" % (shard_name, e))
                     raise e
         return self.replsets
 

MongoBackup/Sharding.py

@@ -119,7 +119,7 @@ def get_configserver(self, force=False):
                     db = DB(config_host, config_port, self.user, self.password, self.authdb)
                     rs = Replset(db, self.user, self.password, self.authdb)
                     try:
-                        self.config_server = rs.find_secondary()
+                        self.config_server = rs.find_secondary(True, True)
                     except Exception:
                         self.config_server = {'host': config_list[0], 'replSet': None}
                 except Exception, e: