Merge pull request #34 from timvaillancourt/config_replset

Replset-based config server feature, version 0.3.0

Author: dbmurphy

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
 
@@ -281,7 +282,7 @@ def run(self):
                     self.user,
                     self.password,
                     self.authdb,
-                    self.sharding.get_configserver(),
+                    self.sharding.get_config_server(),
                     self.verbose
                 )
                 self.mongodumper_summary = self.mongodumper.run()

MongoBackup/Common/DB.py

@@ -44,7 +44,7 @@ def auth_if_required(self):
         else:
             pass
 
-    def admin_command(self, admin_command):
+    def admin_command(self, admin_command, quiet=False):
         tries  = 0
         status = None
         while not status and tries < self.retries:
@@ -53,7 +53,8 @@ def admin_command(self, admin_command):
                 if not status:
                     raise e
             except Exception, e:
-                logging.error("Error running admin command '%s': %s" % (admin_command, e))
+                if not quiet:
+                    logging.error("Error running admin command '%s': %s" % (admin_command, e))
                 tries += 1
                 sleep(1)
         if not status:

MongoBackup/Methods/Dumper.py

@@ -20,10 +20,14 @@ 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       = {}
 
+        if not isinstance(self.config_server, dict) and self.config_server in self.secondaries:
+            self.config_replset = True
+
         if not isinstance(self.secondaries, dict):
             raise Exception, "Field 'secondaries' must be a dictionary of secondary info (by shard)!", None
 
@@ -33,6 +37,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,55 +81,33 @@ def run(self):
             )
             self.threads.append(thread)
 
-        # backup a single config server:
-        if self.config_server:
-            thread = Dump(
-                self.response_queue,
-                'config',
-                self.config_server,
-                self.user,
-                self.password,
-                self.authdb,
-                self.base_dir,
-                self.binary,
-                self.dump_gzip,
-                self.verbose
-            )
-            self.threads.append(thread)
-
         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 not self.config_replset and isinstance(self.config_server, dict):
+            logging.info("Using non-replset backup method for config server mongodump")
+            self.threads = [Dump(
+                self.response_queue,
+                'configsvr',
+                self.config_server['host'],
+                self.user,
+                self.password,
+                self.authdb,
+                self.base_dir,
+                self.binary,
+                self.dump_gzip,
+                self.verbose
+            )]
+            self.threads[0].start()
+            self.wait()
 
         return self._summary
 

MongoBackup/Replset.py

@@ -1,20 +1,19 @@
 import logging
 
 from math import ceil
-from time import mktime
 
 from Common import DB
-from Sharding import Sharding
 
 
 class Replset:
-    def __init__(self, db, user, password, authdb, max_lag_secs):
+    def __init__(self, db, user=None, password=None, authdb='admin', max_lag_secs=5):
         self.db           = db
         self.user         = user
         self.password     = password
         self.authdb       = authdb
         self.max_lag_secs = max_lag_secs
 
+        self.rs_config = None
         self.rs_status = None
         self.primary   = None
         self.secondary = None
@@ -32,49 +31,54 @@ def __init__(self, db, user, password, authdb, max_lag_secs):
     def close(self):
         pass
 
-    def get_rs_status(self, force=False):
+    def get_rs_status(self, force=False, quiet=False):
         try:
             if force or not self.rs_status:
-                self.rs_status = self.db.admin_command('replSetGetStatus')
+                self.rs_status = self.db.admin_command('replSetGetStatus', quiet)
             return self.rs_status
         except Exception, e:
             raise Exception, "Error getting replica set status! Error: %s" % e, None
 
-    def get_rs_config(self):
-        try:
-            if self.db.server_version() >= tuple("3.0.0".split(".")):
-                output = self.db.admin_command('replSetGetConfig')
-                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 get_rs_name(self):
+        return self.get_rs_status()['set']
 
-    def find_primary(self):
-        rs_status = self.get_rs_status()
+    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:
+                optime_ts = member['optime']
+                if isinstance(member['optime'], dict) and 'ts' in member['optime']:
+                    optime_ts = member['optime']['ts']
                 self.primary = {
                     'host': member['name'],
-                    'optime': member['optimeDate']
+                    'optime': optime_ts
                 }
-                optime = member['optime']
-                if isinstance(member['optime'], dict) and 'ts' in member['optime']:
-                    optime = member['optime']['ts']
                 logging.info("Found PRIMARY: %s/%s with optime %s" % (
                     rs_name,
                     member['name'],
-                    str(optime)
+                    str(optime_ts)
                 ))
         if self.primary is None:
-            logging.fatal("Unable to locate a PRIMARY member for replset %s, giving up" % rs_name)
+            logging.error("Unable to locate a PRIMARY member for replset %s, giving up" % rs_name)
             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)
 
@@ -96,30 +100,37 @@ def find_secondary(self):
                                 score = score - member_config['priority']
                         break
 
-                rep_lag = (mktime(self.primary_optime().timetuple()) - mktime(member['optimeDate'].timetuple()))
+                optime_ts = member['optime']
+                if isinstance(member['optime'], dict) and 'ts' in member['optime']:
+                    optime_ts = member['optime']['ts']
+
+                rep_lag = (self.primary_optime().time - optime_ts.time)
                 score = ceil((score - rep_lag) * score_scale)
                 if rep_lag < self.max_lag_secs:
                     if self.secondary is None or score > self.secondary['score']:
                         self.secondary = {
                             'replSet': rs_name,
                             'count': 1 if self.secondary is None else self.secondary['count'] + 1,
                             'host': member['name'],
-                            'optime': member['optimeDate'],
+                            'optime': optime_ts,
                             'score': score
                         }
                     log_msg = "Found SECONDARY %s/%s" % (rs_name, member['name'])
                 else:
                     log_msg = "Found SECONDARY %s/%s with too-high replication lag! Skipping" % (rs_name, member['name'])
 
-                log_data['optime'] = member['optime']
-                if isinstance(member['optime'], dict) and 'ts' in member['optime']:
-                    log_data['optime'] = member['optime']['ts']
+                if 'configsvr' in rs_status and rs_status['configsvr']:
+                    log_data['configsvr'] = True
+
+                log_data['lag']    = rep_lag
+                log_data['optime'] = optime_ts
                 log_data['score']  = int(score)
                 logging.info("%s: %s" % (log_msg, str(log_data)))
         if self.secondary is None or (self.secondary['count'] + 1) < quorum_count:
-            logging.fatal("Not enough secondaries in replset %s to take backup! Num replset members: %i, required quorum: %i" % (
+            secondary_count = self.secondary['count'] + 1 if self.secondary else 0
+            logging.error("Not enough secondaries in replset %s to take backup! Num replset members: %i, required quorum: %i" % (
                 rs_name,
-                self.secondary['count'] + 1,
+                secondary_count,
                 quorum_count
             ))
             raise Exception, "Not enough secondaries in replset %s to safely take backup!" % rs_name, None
@@ -128,74 +139,6 @@ def find_secondary(self):
         return self.secondary
 
     def primary_optime(self):
-        rs_status  = self.get_rs_status()
-        rs_primary = self.find_primary()
+        rs_primary = self.find_primary(True)
         if 'optime' in rs_primary:
             return rs_primary['optime']
-
-
-class ReplsetSharded:
-    def __init__(self, sharding, db, user, password, authdb, max_lag_secs):
-        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 isinstance(self.sharding, Sharding):
-            raise Exception, "'sharding' field is not an instance of class: 'Sharding'!", None
-
-        # Get a DB connection
-        try:
-            if isinstance(self.db, 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 not an instance of class: 'DB'!", 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]
-            secondary = replset.find_secondary()
-            shard_secondaries[rs_name] = secondary
-        return shard_secondaries
-
-    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()