Merge pull request #89 from timvaillancourt/MCB_1.0-cleanup_v6

Mcb 1.0 cleanup v6

Author: timvaillancourt

mongodb_consistent_backup/Archive/Archive.py

@@ -1,7 +1,7 @@
 import logging
 
 from Tar import Tar
-from mongodb_consistent_backup.Common import config_to_string, parse_method
+from mongodb_consistent_backup.Common import Timer, config_to_string, parse_method
 
 
 class Archive:
@@ -11,6 +11,7 @@ def __init__(self, config, backup_dir):
 
         self.method    = None
         self._archiver = None
+        self.timer     = Timer()
         self.init()
 
     def init(self):
@@ -42,7 +43,12 @@ def archive(self):
         if self._archiver:
             config_string = config_to_string(self.config.archive[self.method])
             logging.info("Archiving with method: %s (options: %s)" % (self.method, config_string))
-            return self._archiver.run()
+            self.timer.start()
+
+            self._archiver.run()
+
+            self.timer.stop()
+            logging.info("Archiving completed in %s seconds" % self.timer.duration())
 
     def close(self):
         if self._archiver:

mongodb_consistent_backup/Archive/Tar/Tar.py

@@ -67,15 +67,14 @@ def run(self):
                     self._pool.apply_async(TarThread(subdir_name, output_file, self.do_gzip(), self.verbose, self.binary).run)
             except Exception, e:
                 self._pool.terminate()
-                logging.fatal("Could not create archiving thread! Error: %s" % e)
+                logging.fatal("Could not create tar archiving thread! Error: %s" % e)
                 raise e
             self._pool.close()
             self._pool.join()
-        logging.info("Archiver threads completed")
 
     def close(self):
-        logging.debug("Stopping Archiver threads")
+        logging.debug("Stopping tar archiving threads")
         if self._pool is not None:
             self._pool.terminate()
             self._pool.join()
-        logging.info("Stopped all Archiver threads")
+        logging.info("Stopped all tar archiving threads")

mongodb_consistent_backup/Archive/Tar/__init__.py

@@ -3,7 +3,7 @@
 
 def config(parser):
     parser.add_argument("--archive.tar.compression", dest="archive.tar.compression",
-                        help="Archiver compression method (default: gzip)", default='gzip', choices=['gzip', 'none'])
+                        help="Tar archiver compression method (default: gzip)", default='gzip', choices=['gzip', 'none'])
     parser.add_argument("--archive.tar.threads", dest="archive.tar.threads", 
-                        help="Number of threads to use in archive phase (default: 1-per-CPU)", default=0, type=int)
+                        help="Number of Tar archiver threads to use (default: 1-per-CPU)", default=0, type=int)
     return parser

mongodb_consistent_backup/Backup/Backup.py

@@ -1,7 +1,7 @@
 import logging
 
 from Mongodump import Mongodump
-from mongodb_consistent_backup.Common import config_to_string, parse_method
+from mongodb_consistent_backup.Common import Timer, config_to_string, parse_method
 
 
 class Backup:
@@ -13,15 +13,14 @@ def __init__(self, config, backup_dir, secondaries, config_server=None):
 
         self.method  = None
         self._method = None
+        self.timer   = Timer()
         self.init()
 
     def init(self):
         backup_method = self.config.backup.method
         if not backup_method or parse_method(backup_method) == "none":
             raise Exception, 'Must specify a backup method!', None
-        self.method   = parse_method(backup_method)
-        config_string = config_to_string(self.config.backup[self.method])
-        logging.info("Using backup method: %s (options: %s)" % (self.method, config_string))
+        self.method = parse_method(backup_method)
         try:
             self._method = globals()[self.method.capitalize()](
                 self.config,
@@ -40,7 +39,16 @@ def is_compressed(self):
 
     def backup(self):
         if self._method:
-            return self._method.run()
+            config_string = config_to_string(self.config.backup[self.method])
+            logging.info("Using backup method: %s (options: %s)" % (self.method, config_string))
+            self.timer.start()
+
+            info = self._method.run()
+
+            self.timer.stop()
+            logging.info("Backup completed in %s seconds" % self.timer.duration())
+
+            return info
 
     def close(self):
         if self._method:

mongodb_consistent_backup/Backup/Mongodump/MongodumpThread.py

@@ -3,9 +3,8 @@
 
 from multiprocessing import Process
 from signal import signal, SIGINT, SIGTERM
-from time import time
 
-from mongodb_consistent_backup.Common import LocalCommand
+from mongodb_consistent_backup.Common import LocalCommand, Timer
 from mongodb_consistent_backup.Oplog import Oplog
 
 
@@ -28,12 +27,12 @@ def __init__(self, response_queue, backup_name, host_port, user, password, authd
         self.dump_gzip      = dump_gzip
         self.verbose        = verbose
 
+        self.timer      = Timer()
         self._command   = None
         self.completed  = False 
         self.backup_dir = "%s/%s" % (self.base_dir, self.backup_name)
         self.dump_dir   = "%s/dump" % self.backup_dir
         self.oplog_file = "%s/oplog.bson" % self.dump_dir
-        self.start_time = time()
 
         signal(SIGINT, self.close)
         signal(SIGTERM, self.close)
@@ -52,6 +51,8 @@ def run(self):
             self.port
         ))
 
+        self.timer.start()
+
         mongodump_flags = ["-h", self.host_port, "--oplog", "-o", "%s/dump" % self.backup_dir]
         if self.threads > 0:
             mongodump_flags.extend(["--numParallelCollections="+str(self.threads)])
@@ -89,7 +90,8 @@ def run(self):
             'completed': self.completed
         })
 
-        time_diff = time() - self.start_time
+        self.timer.stop()
+
         logging.info("Backup for %s/%s:%s completed in %s sec with %i oplog changes captured to: %s" % (
-            self.backup_name, self.host, self.port, time_diff, oplog.count(), str(oplog.last_ts())
+            self.backup_name, self.host, self.port, self.timer.duration(), oplog.count(), str(oplog.last_ts())
         ))

mongodb_consistent_backup/Common/Config.py

@@ -1,3 +1,4 @@
+import mongodb_consistent_backup
 import sys
 
 from argparse import Action
@@ -71,7 +72,7 @@ def _get(self, keys, data=None):
             return data[keys]
 
     def parse_submodules(self):
-        for _, modname, ispkg in walk_packages(path="."):
+        for _, modname, ispkg in walk_packages(path=mongodb_consistent_backup.__path__, prefix=mongodb_consistent_backup.__name__+'.'):
             if ispkg:
                 try:
                     components = modname.split('.')

mongodb_consistent_backup/Common/Timer.py

@@ -0,0 +1,25 @@
+from time import time
+
+
+class Timer:
+    def __init__(self):
+        self.count  = 0
+        self.rounds = {}
+
+    def start(self):
+        self.count += 1
+        self.rounds[self.count] = { 'start': time(), 'started': True }
+
+    def stop(self):
+        if self.rounds[self.count] and self.rounds[self.count]['started']:
+            self.rounds[self.count]['started'] = False
+            self.rounds[self.count]['end']     = time()
+
+    def duration(self):
+        if 'start' in self.rounds[self.count]:
+            if 'end' in self.rounds[self.count]:
+                end = self.rounds[self.count]['end']
+            else:
+                end = time()
+            return end - self.rounds[self.count]['start']
+        return -1

mongodb_consistent_backup/Common/__init__.py

@@ -2,4 +2,5 @@
 from DB import DB
 from LocalCommand import LocalCommand
 from Lock import Lock
+from Timer import Timer
 from Util import config_to_string, parse_method, validate_hostname 

mongodb_consistent_backup/Main.py

@@ -4,11 +4,10 @@
 from datetime import datetime
 from multiprocessing import current_process
 from signal import signal, SIGINT, SIGTERM
-from time import time
 
 from Archive import Archive
 from Backup import Backup
-from Common import Config, DB, Lock, validate_hostname
+from Common import Config, DB, Lock, Timer, validate_hostname
 from Notify import Notify
 from Oplog import Tailer, Resolver
 from Replication import Replset, ReplsetSharded
@@ -29,18 +28,16 @@ def __init__(self, prog_name="mongodb-consistent-backup"):
         self.oplog_resolver           = None
         self.upload                   = None
         self.lock                     = None
-        self.start_time               = time()
-        self.end_time                 = None
-        self.backup_duration          = None
         self.backup_time              = None
         self.backup_root_directory    = None
         self.backup_root_subdirectory = None
         self.db                       = None
         self.is_sharded               = False
+        self.log_level                = None
+        self.timer                    = Timer()
         self.secondaries              = {}
         self.oplog_summary            = {}
         self.backup_summary           = {}
-        self.log_level                = None
 
         self.setup_config()
         self.setup_logger()
@@ -149,6 +146,7 @@ def run(self):
         logging.info("Starting %s version %s (git commit hash: %s)" % (self.program_name, self.config.version, self.config.git_commit))
 
         self.get_lock()
+        self.timer.start()
 
         # Setup the notifier
         try:
@@ -285,8 +283,9 @@ def run(self):
                 self.exception("Problem restoring balancer lock! Error: %s" % e)
 
             # resolve/merge tailed oplog into mongodump oplog.bson to a consistent point for all shards
-            if self.config.backup.method == "mongodump" and self.oplogtailer:
+            if self.backup.method == "mongodump" and self.oplogtailer:
                 self.oplog_resolver = Resolver(self.config, self.oplog_summary, self.backup_summary)
+                self.oplog_resolver.compression(self.oplogtailer.compression())
                 self.oplog_resolver.run()
 
         # archive backup directories
@@ -295,21 +294,20 @@ def run(self):
         except Exception, e:
             self.exception("Problem performing archiving! Error: %s" % e)
 
-        self.end_time = time()
-        self.backup_duration = self.end_time - self.start_time
-
         # upload backup
         try:
             self.upload.upload()
         except Exception, e:
             self.exception("Problem performing upload of backup! Error: %s" % e)
 
+        self.timer.stop()
+
         # send notifications of backup state
         try:
             self.notify.notify("%s: backup '%s' succeeded in %s secs" % (
                 self.program_name,
                 self.config.backup.name,
-                self.backup_duration
+                self.timer.duration()
             ), True)
         except Exception, e:
             self.exception("Problem running Notifier! Error: %s" % e)
@@ -319,4 +317,4 @@ def run(self):
 
         self.release_lock()
 
-        logging.info("Backup completed in %s sec" % self.backup_duration)
+        logging.info("Completed %s in %s sec" % (self.program_name, self.timer.duration()))

mongodb_consistent_backup/Oplog/Resolver/Resolver.py

@@ -9,6 +9,7 @@
 from types import MethodType
 
 from ResolverThread import ResolverThread
+from mongodb_consistent_backup.Common import Timer, parse_method
 
 
 # Allows pooled .apply_async()s to work on Class-methods:
@@ -26,24 +27,35 @@ def __init__(self, config, tailed_oplogs_summary, backup_oplogs_summary):
         self.config        = config
         self.tailed_oplogs = tailed_oplogs_summary
         self.backup_oplogs = backup_oplogs_summary
-        self.thread_count  = self.config.oplog.resolver.threads
-
-        self.dump_gzip = False
-        if self.config.oplog.compression == 'gzip':
-            self.dump_gzip = True
 
+        self.timer  = Timer()
         self.end_ts = None
         self.delete_oplogs = {}
 
-        if self.thread_count is None or self.thread_count < 1:
-            self.thread_count = cpu_count() * 2
-
         try:
-            self._pool = Pool(processes=self.thread_count)
+            self._pool = Pool(processes=self.threads())
         except Exception, e:
-            logging.fatal("Could not start pool! Error: %s" % e)
+            logging.fatal("Could not start oplog resolver pool! Error: %s" % e)
             raise e
 
+    def compression(self, method=None):
+        if method:
+            logging.debug("Setting oplog resolver compression to: %s" % method)
+            self.config.oplog.compression = parse_method(method)
+        return parse_method(self.config.oplog.compression)
+
+    def do_gzip(self):
+        if self.compression() == 'gzip':
+           return True
+        return False
+
+    def threads(self, threads=None):
+        if threads:
+            self.config.oplog.resolver.threads = int(threads)
+        if self.config.oplog.resolver.threads is None or self.config.oplog.resolver.threads < 1:
+            self.config.oplog.resolver.threads = int(cpu_count() * 2)
+        return int(self.config.oplog.resolver.threads)
+
     def get_consistent_end_ts(self):
         ts = None
         for host in self.tailed_oplogs:
@@ -55,7 +67,8 @@ def get_consistent_end_ts(self):
         return ts
 
     def run(self):
-        logging.info("Resolving oplogs using %i threads max" % self.thread_count)
+        logging.info("Resolving oplogs using %i threads max" % self.threads())
+        self.timer.start()
 
         self.end_ts   = self.get_consistent_end_ts()
         for host in self.backup_oplogs:
@@ -84,7 +97,7 @@ def run(self):
                                 backup_oplog['file'],
                                 backup_oplog['last_ts'],
                                 self.end_ts,
-                                self.dump_gzip
+                                self.do_gzip()
                             ).run)
                         except Exception, e:
                             logging.fatal("Resolve failed for %s:%s! Error: %s" % (host, port, e))
@@ -105,4 +118,5 @@ def run(self):
                 logging.fatal("Deleting of tailed oplog file %s failed! Error: %s" % (oplog_file, e))
                 raise e
 
-        logging.info("Done resolving oplogs")
+        self.timer.stop()
+        logging.info("Oplog resolving completed in %s seconds" % self.timer.duration())