adding functionality to get true/false for checking if a host is a config server

Author: Tim Vaillancourt

MongoBackup/Backup.py

@@ -13,7 +13,8 @@
 from Methods import Dumper
 from Notify import NotifyNSCA
 from Oplog import OplogTailer, OplogResolver
-from Replset import Replset, ReplsetSharded
+from Replset import Replset
+from ReplsetSharded import ReplsetSharded
 from Sharding import Sharding
 from Upload import UploadS3
 

MongoBackup/Common/DB.py

@@ -44,10 +44,10 @@ def auth_if_required(self):
         else:
             pass
 
-    def admin_command(self, admin_command):
+    def admin_command(self, admin_command, retry=True):
         tries  = 0
         status = None
-        while not status and tries < self.retries:
+        while retry and not status and tries < self.retries:
             try:
                 status = self._conn['admin'].command(admin_command)
                 if not status:
@@ -57,7 +57,7 @@ def admin_command(self, admin_command):
                 tries += 1
                 sleep(1)
         if not status:
-            raise Exception, "Could not get output from command: '%s' after %i retries!" % (admin_command, self.retries), None
+            raise Exception, "Could not get output from command: '%s' after %i retries!" % (admin_command, retries), None
         return status
 
     def server_version(self):

MongoBackup/Methods/Dumper.py

@@ -20,6 +20,7 @@ def __init__(self, secondaries, base_dir, binary, dump_gzip=False, user=None, pa
         self.config_server = config_server
         self.verbose       = verbose
 
+        self.config_replset = False
         self.response_queue = Queue()
         self.threads        = []
         self._summary       = {}

MongoBackup/Replset.py

@@ -2,9 +2,6 @@
 
 from math import ceil
 
-from Common import DB
-from Sharding import Sharding
-
 
 class Replset:
     def __init__(self, db, user=None, password=None, authdb='admin', max_lag_secs=5):
@@ -31,18 +28,18 @@ def __init__(self, db, user=None, password=None, authdb='admin', max_lag_secs=5)
     def close(self):
         pass
 
-    def get_rs_status(self, force=False):
+    def get_rs_status(self, force=False, retry=True):
         try:
             if force or not self.rs_status:
-                self.rs_status = self.db.admin_command('replSetGetStatus')
+                self.rs_status = self.db.admin_command('replSetGetStatus', retry)
             return self.rs_status
         except Exception, e:
             raise Exception, "Error getting replica set status! Error: %s" % e, None
 
-    def get_rs_config(self):
+    def get_rs_config(self, retry=True):
         try:
             if self.db.server_version() >= tuple("3.0.0".split(".")):
-                output = self.db.admin_command('replSetGetConfig')
+                output = self.db.admin_command('replSetGetConfig', retry)
                 return output['config']
             else:
                 return self.connection['local'].system.replset.find_one()
@@ -133,76 +130,3 @@ def primary_optime(self):
         rs_primary = self.find_primary(True)
         if 'optime' in rs_primary:
             return rs_primary['optime']
-
-
-class ReplsetSharded:
-    def __init__(self, sharding, db, user=None, password=None, authdb='admin', max_lag_secs=5):
-        self.sharding     = sharding
-        self.db           = db
-        self.user         = user
-        self.password     = password
-        self.authdb       = authdb
-        self.max_lag_secs = max_lag_secs
-
-        self.replsets      = {} 
-        self.replset_conns = {}
-
-        # Check Sharding class:
-        if not self.sharding.__class__.__name__ == "Sharding":
-            raise Exception, "'sharding' field is an instance of %s, not 'Sharding'!" % self.sharding.__class__.__name__, None
-
-        # Get a DB connection
-        try:
-            if self.db.__class__.__name__ == "DB":
-                self.connection = self.db.connection()
-                if not self.connection.is_mongos:
-                    raise Exception, 'MongoDB connection is not to a mongos!', None
-            else:
-                raise Exception, "'db' field is an instance of %s, not 'DB'!" % self.db.__class__.__name__, None
-        except Exception, e:
-            logging.fatal("Could not get DB connection! Error: %s" % e)
-            raise e
-
-    def get_replset_connection(self, host, port, force=False):
-        conn_name = "%s-%i" % (host, port)
-        if force or not conn_name in self.replset_conns:
-            try:
-                self.replset_conns[conn_name] = DB(host, port, self.user, self.password, self.authdb)
-            except Exception, e:
-                logging.fatal("Could not get DB connection to %s:%i! Error: %s" % (host, port, e))
-                raise e
-        return self.replset_conns[conn_name]
-
-    def get_replsets(self, force=False):
-        for shard in self.sharding.shards():
-            shard_name, members = shard['host'].split('/')
-            host, port = members.split(',')[0].split(":")
-            port       = int(port)
-            if force or not shard_name in self.replsets:
-                try:
-                    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))
-                    raise e
-        return self.replsets
-
-    def find_secondaries(self):
-        shard_secondaries = {}
-        for rs_name in self.get_replsets():
-            replset = self.replsets[rs_name]
-            shard_secondaries[rs_name] = replset.find_secondary()
-        return shard_secondaries
-
-    def primary_optimes(self):
-        primary_optimes = {}
-        for rs_name in self.get_replsets():
-            replset = self.replsets[rs_name]
-            primary_optimes[rs_name] = replset.primary_optime()
-        return primary_optimes
-
-    def close(self):
-        for rs_name in self.replsets:
-            self.replsets[rs_name].close()
-        for conn_name in self.replset_conns:
-            self.replset_conns[conn_name].close()

MongoBackup/ReplsetSharded.py

@@ -0,0 +1,77 @@
+import logging
+
+from Common import DB
+from Sharding import Sharding
+
+
+class ReplsetSharded:
+    def __init__(self, sharding, db, user=None, password=None, authdb='admin', max_lag_secs=5):
+        self.sharding     = sharding
+        self.db           = db
+        self.user         = user
+        self.password     = password
+        self.authdb       = authdb
+        self.max_lag_secs = max_lag_secs
+
+        self.replsets      = {} 
+        self.replset_conns = {}
+
+        # Check Sharding class:
+        if not self.sharding.__class__.__name__ == "Sharding":
+            raise Exception, "'sharding' field is an instance of %s, not 'Sharding'!" % self.sharding.__class__.__name__, None
+
+        # Get a DB connection
+        try:
+            if self.db.__class__.__name__ == "DB":
+                self.connection = self.db.connection()
+                if not self.connection.is_mongos:
+                    raise Exception, 'MongoDB connection is not to a mongos!', None
+            else:
+                raise Exception, "'db' field is an instance of %s, not 'DB'!" % self.db.__class__.__name__, None
+        except Exception, e:
+            logging.fatal("Could not get DB connection! Error: %s" % e)
+            raise e
+
+    def get_replset_connection(self, host, port, force=False):
+        conn_name = "%s-%i" % (host, port)
+        if force or not conn_name in self.replset_conns:
+            try:
+                self.replset_conns[conn_name] = DB(host, port, self.user, self.password, self.authdb)
+            except Exception, e:
+                logging.fatal("Could not get DB connection to %s:%i! Error: %s" % (host, port, e))
+                raise e
+        return self.replset_conns[conn_name]
+
+    def get_replsets(self, force=False):
+        for shard in self.sharding.shards():
+            shard_name, members = shard['host'].split('/')
+            host, port = members.split(',')[0].split(":")
+            port       = int(port)
+            if force or not shard_name in self.replsets:
+                try:
+                    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))
+                    raise e
+        return self.replsets
+
+    def find_secondaries(self):
+        shard_secondaries = {}
+        for rs_name in self.get_replsets():
+            replset = self.replsets[rs_name]
+            shard_secondaries[rs_name] = replset.find_secondary()
+        return shard_secondaries
+
+    def primary_optimes(self):
+        primary_optimes = {}
+        for rs_name in self.get_replsets():
+            replset = self.replsets[rs_name]
+            primary_optimes[rs_name] = replset.primary_optime()
+        return primary_optimes
+
+    def close(self):
+        for rs_name in self.replsets:
+            self.replsets[rs_name].close()
+        for conn_name in self.replset_conns:
+            self.replset_conns[conn_name].close()

MongoBackup/Sharding.py

@@ -2,6 +2,9 @@
 
 from time import sleep
 
+from Common import DB
+from Replset import Replset
+
 
 class Sharding:
     def __init__(self, db, user=None, password=None, authdb='admin', balancer_wait_secs=300, balancer_sleep=5):
@@ -12,6 +15,7 @@ def __init__(self, db, user=None, password=None, authdb='admin', balancer_wait_s
         self.balancer_wait_secs = balancer_wait_secs
         self.balancer_sleep     = balancer_sleep
 
+        self.config_server         = None
         self._balancer_state_start = None
 
         # Get a DB connection
@@ -94,20 +98,37 @@ def stop_balancer(self):
         logging.fatal("Could not stop balancer: %s:%i!" % (self.host, self.port))
         raise Exception, "Could not stop balancer: %s:%i" % (self.host, self.port), None
 
-    def get_configserver(self):
-        cmdlineopts = self.db.admin_command("getCmdLineOpts")
-        config_string = None
-        if cmdlineopts.get('parsed').get('configdb'):
-            config_string = cmdlineopts.get('parsed').get('configdb')
-        elif cmdlineopts.get('parsed').get('sharding').get('configDB'):
-            config_string = cmdlineopts.get('parsed').get('sharding').get('configDB')
-        if config_string:
-            # noinspection PyBroadException
+    def get_configserver(self, force=False):
+        if force or not self.config_server:
+            cmdlineopts = self.db.admin_command("getCmdLineOpts")
+            config_string = None
+            if cmdlineopts.get('parsed').get('configdb'):
+                config_string = cmdlineopts.get('parsed').get('configdb')
+            elif cmdlineopts.get('parsed').get('sharding').get('configDB'):
+                config_string = cmdlineopts.get('parsed').get('sharding').get('configDB')
+            if config_string:
+                # noinspection PyBroadException
+                try:
+                    config_list = config_string.split(",")
+                except Exception:
+                    config_list = [config_string]
+                self.config_server = config_list[0]
+            else:
+                logging.fatal("Unable to locate config servers for %s:%i!" % (self.host, self.port))
+                raise Exception, "Unable to locate config servers for %s:%i!" % (self.host, self.port), None
+        return self.config_server
+
+    def is_config_replset(self):
+        try:
+            config_server = self.get_configserver()
+            config_host, config_port = config_server.split(":")
+            db = DB(config_host, config_port, self.user, self.password, self.authdb)
+            rs = Replset(db, self.user, self.password, self.authdb)
             try:
-                config_list = config_string.split(",")
+                rs.get_rs_status(True, False)
+                return 
             except Exception:
-                config_list = [config_string]
-            return config_list[0]
-        else:
-            logging.fatal("Unable to locate config servers for %s:%i!" % (self.host, self.port))
-            raise Exception, "Unable to locate config servers for %s:%i!" % (self.host, self.port), None
+                return False
+        except Exception, e:
+            raise e             
+