Merge pull request #33 from timvaillancourt/replsethandler_to_backup_py

Use consistent replset info throughout backup + class+method renaming + reuse db connection in MainThread

Author: dbmurphy

Makefile

@@ -7,7 +7,7 @@ VERSION=$(shell cat VERSION)
 
 all: bin/mongodb-consistent-backup
 
-bin/mongodb-consistent-backup: setup.py requirements.txt VERSION scripts/build.sh MongoBackup/*.py MongoBackup/Common/*.py MongoBackup/Notify/*.py MongoBackup/Oplog/*.py MongoBackup/Upload/*.py
+bin/mongodb-consistent-backup: setup.py requirements.txt VERSION scripts/build.sh MongoBackup/*.py MongoBackup/Common/*.py MongoBackup/Methods/*.py MongoBackup/Notify/*.py MongoBackup/Oplog/*.py MongoBackup/Upload/*.py
 	PYTHON_BIN=$(PYTHON_BIN) VIRTUALENV_BIN=$(VIRTUALENV_BIN) bash scripts/build.sh
 
 install: bin/mongodb-consistent-backup

MongoBackup/Backup.py

@@ -8,12 +8,13 @@
 from signal import signal, SIGINT, SIGTERM
 from time import time
 
-from Common import DB, Lock
-from ShardingHandler import ShardingHandler
-from Mongodumper import Mongodumper
-from Oplog import OplogTailer, OplogResolver
 from Archiver import Archiver
+from Common import DB, Lock
+from Methods import Dumper
 from Notify import NotifyNSCA
+from Oplog import OplogTailer, OplogResolver
+from Replset import Replset, ReplsetSharded
+from Sharding import Sharding
 from Upload import UploadS3
 
 
@@ -34,7 +35,7 @@ def __init__(self, options):
         self.backup_location = None
         self.dump_gzip = False
         self.balancer_wait_secs = 300
-        self.balancer_sleep = 10
+        self.balancer_sleep = 5 
         self.archiver_threads = 1
         self.resolver_threads = 1
         self.notify_nsca = None
@@ -58,6 +59,7 @@ def __init__(self, options):
         self.upload_s3_chunk_size_mb = None
         self.archiver = None
         self.sharding = None
+        self.replset  = None
         self.mongodumper = None
         self.oplogtailer = None
         self.oplog_resolver = None
@@ -69,6 +71,7 @@ def __init__(self, options):
         self.start_time = time()
         self.oplog_threads = []
         self.oplog_summary = {}
+        self.secondaries   = {}
         self.mongodumper_summary = {}
 
         # Setup options are properies and connection to node
@@ -108,9 +111,9 @@ def __init__(self, options):
 
         # Get a DB connection
         try:
-            connection = DB(self.host, self.port, self.user, self.password, self.authdb).connection()
-            self.is_mongos = connection.is_mongos
-            connection.close()
+            self.db         = DB(self.host, self.port, self.user, self.password, self.authdb)
+            self.connection = self.db.connection()
+            self.is_sharded = self.connection.is_mongos
         except Exception, e:
             raise e
 
@@ -140,7 +143,7 @@ def cleanup_and_exit(self, code, frame):
         if current_process().name == "MainProcess":
             logging.info("Starting cleanup and exit procedure! Killing running threads")
 
-            submodules = ['sharding', 'mongodumper', 'oplogtailer', 'archiver', 'uploader_s3']
+            submodules = ['replset', 'sharding', 'mongodumper', 'oplogtailer', 'archiver', 'uploader_s3']
             for submodule_name in submodules:
                 submodule = getattr(self, submodule_name)
                 if submodule:
@@ -152,6 +155,9 @@ def cleanup_and_exit(self, code, frame):
                     self.backup_name
                 ))
 
+            if self.db:
+                self.db.close()
+
             if self._lock:
                 self._lock.release()
 
@@ -172,22 +178,37 @@ def run(self):
             logging.fatal("Could not acquire lock! Is another %s process running? Exiting" % self.program_name)
             sys.exit(1)
 
-        if not self.is_mongos:
+        if not self.is_sharded:
             logging.info("Running backup of %s:%s in replset mode" % (self.host, self.port))
 
             self.archiver_threads = 1
 
+            # get shard secondary
             try:
-                self.mongodumper = Mongodumper(
-                    self.host,
-                    self.port,
+                self.replset = Replset(
+                    self.db,
                     self.user,
                     self.password,
                     self.authdb,
+                    self.max_repl_lag_secs
+                )
+                secondary    = self.replset.find_secondary()
+                replset_name = secondary['replSet']
+
+                self.secondaries[replset_name] = secondary
+                self.replset.close()
+            except Exception, e:
+                self.exception("Problem getting shard secondaries! Error: %s" % e)
+
+            try:
+                self.mongodumper = Dumper(
+                    self.secondaries,
                     self.backup_root_directory,
                     self.backup_binary,
                     self.dump_gzip,
-                    self.max_repl_lag_secs,
+                    self.user,
+                    self.password,
+                    self.authdb,
                     None,
                     self.verbose
                 )
@@ -200,32 +221,48 @@ def run(self):
 
             # connect to balancer and stop it
             try:
-                self.sharding = ShardingHandler(
-                    self.host,
-                    self.port,
+                self.sharding = Sharding(
+                    self.db,
                     self.user,
                     self.password,
                     self.authdb,
                     self.balancer_wait_secs,
                     self.balancer_sleep
                 )
                 self.sharding.get_start_state()
+            except Exception, e:
+                self.exception("Problem connecting to the balancer! Error: %s" % e)
+
+            # get shard secondaries
+            try:
+                self.replset = ReplsetSharded(
+                    self.sharding,
+                    self.db,
+                    self.user,
+                    self.password,
+                    self.authdb,
+                    self.max_repl_lag_secs
+                )
+                self.secondaries = self.replset.find_secondaries()
+            except Exception, e:
+                self.exception("Problem getting shard secondaries! Error: %s" % e)
+
+            # Stop the balancer:    
+            try:
                 self.sharding.stop_balancer()
             except Exception, e:
-                self.exception("Problem connecting-to and/or stopping balancer! Error: %s" % e)
+                self.exception("Problem stopping the balancer! Error: %s" % e)
 
             # start the oplog tailer threads
             if self.no_oplog_tailer:
                 logging.warning("Oplog tailing disabled! Skipping")
             else:
                 try:
                     self.oplogtailer = OplogTailer(
+                        self.secondaries,
                         self.backup_name,
                         self.backup_root_directory,
-                        self.host,
-                        self.port,
                         self.dump_gzip,
-                        self.max_repl_lag_secs,
                         self.user,
                         self.password,
                         self.authdb
@@ -236,16 +273,14 @@ def run(self):
 
             # start the mongodumper threads
             try:
-                self.mongodumper = Mongodumper(
-                    self.host,
-                    self.port,
-                    self.user,
-                    self.password,
-                    self.authdb,
+                self.mongodumper = Dumper(
+                    self.secondaries, 
                     self.backup_root_directory,
                     self.backup_binary,
                     self.dump_gzip,
-                    self.max_repl_lag_secs,
+                    self.user,
+                    self.password,
+                    self.authdb,
                     self.sharding.get_configserver(),
                     self.verbose
                 )
@@ -312,6 +347,9 @@ def run(self):
             except Exception, e:
                 self.exception("Problem running NSCA notifier! Error: %s" % e)
 
+        if self.db:
+            self.db.close()
+
         self._lock.release()
 
         logging.info("Backup completed in %s sec" % self.backup_duration)

MongoBackup/Mongodump.py renamed to MongoBackup/Methods/Dump.py

@@ -5,12 +5,12 @@
 from signal import signal, SIGINT, SIGTERM
 from time import time
 
-from Common import LocalCommand
-from Oplog import OplogInfo
+from MongoBackup.Common import LocalCommand
+from MongoBackup.Oplog import OplogInfo
 
 
 # noinspection PyStringFormat
-class Mongodump(Process):
+class Dump(Process):
     def __init__(self, response_queue, backup_name, host_port, user, password, authdb, base_dir, binary,
                  dump_gzip=False, verbose=False):
         Process.__init__(self)
@@ -51,7 +51,6 @@ def run(self):
 
         mongodump_flags = ["-h", self.host_port, "--oplog", "-o", "%s/dump" % self.backup_dir]
         if self.dump_gzip:
-            logging.debug("Enabling inline mongodump compression using --gzip flag")
             mongodump_flags.extend(["--gzip"])
         if self.authdb and self.authdb != "admin":
             logging.debug("Using database %s for authentication" % self.authdb)

MongoBackup/Methods/Dumper.py

@@ -0,0 +1,111 @@
+import logging
+
+from fabric.api import hide, settings, local
+from multiprocessing import Process, Queue
+from time import sleep
+
+from MongoBackup.Methods import Dump
+
+
+class Dumper:
+    def __init__(self, secondaries, base_dir, binary, dump_gzip=False, user=None, password=None,
+                 authdb='admin', config_server=None, verbose=False):
+        self.secondaries   = secondaries
+        self.base_dir      = base_dir
+        self.binary        = binary
+        self.dump_gzip     = dump_gzip
+        self.user          = user
+        self.password      = password
+        self.authdb        = authdb
+        self.config_server = config_server
+        self.verbose       = verbose
+
+        self.response_queue = Queue()
+        self.threads        = []
+        self._summary       = {}
+
+        if not isinstance(self.secondaries, dict):
+            raise Exception, "Field 'secondaries' must be a dictionary of secondary info (by shard)!", None
+
+        with hide('running', 'warnings'), settings(warn_only=True):
+            self.version = local("%s --version|awk 'NR >1 {exit}; /version/{print $NF}'" % self.binary, capture=True)
+
+    def summary(self):
+        return self._summary
+
+    def run(self):
+        # backup a secondary from each shard:
+        for shard in self.secondaries:
+            secondary = self.secondaries[shard]
+            thread = Dump(
+                self.response_queue,
+                secondary['replSet'],
+                secondary['host'],
+                self.user,
+                self.password,
+                self.authdb,
+                self.base_dir,
+                self.binary,
+                self.dump_gzip,
+                self.verbose
+            )
+            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
+        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()
+
+        # 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
+
+        return self._summary
+
+    def close(self):
+        logging.info("Killing all mongodump threads...")
+        if len(self.threads) > 0:
+            for thread in self.threads:
+                thread.terminate()
+        logging.info("Killed all mongodump threads")

MongoBackup/Methods/__init__.py

@@ -0,0 +1,2 @@
+from Dump import Dump
+from Dumper import Dumper