Merge pull request #110 from timvaillancourt/MCB_1.0-bugfix19

MCB_1.0: Bugfixes 19

Author: timvaillancourt

mongodb_consistent_backup/Archive/Tar/Tar.py

@@ -8,7 +8,7 @@
 
 from TarThread import TarThread
 from mongodb_consistent_backup.Common import parse_method
-from mongodb_consistent_backup.Errors import Error
+from mongodb_consistent_backup.Errors import Error, OperationError
 from mongodb_consistent_backup.Pipeline import Task
 
 
@@ -68,7 +68,7 @@ def run(self):
                     output_file = "%s.tar" % subdir_name
                     if self.do_gzip():
                         output_file  = "%s.tgz" % subdir_name
-                    self._pool.apply_async(TarThread(subdir_name, output_file, self.do_gzip(), self.verbose, self.binary).run, callback=self.done)
+                    self._pool.apply_async(TarThread(subdir_name, output_file, self.compression(), self.verbose, self.binary).run, callback=self.done)
                     self._pooled.append(subdir_name)
             except Exception, e:
                 self._pool.terminate()

mongodb_consistent_backup/Archive/Tar/TarThread.py

@@ -1,31 +1,28 @@
 import os
 import logging
-import sys
-
-from signal import signal, SIGINT, SIGTERM
 
 from mongodb_consistent_backup.Common import LocalCommand
+from mongodb_consistent_backup.Pipeline import PoolThread
 
 
-class TarThread:
-    def __init__(self, backup_dir, output_file, do_gzip=False, verbose=False, binary="tar"):
-        self.backup_dir  = backup_dir
-        self.output_file = output_file
-        self.do_gzip     = do_gzip
-        self.verbose     = verbose
-        self.binary      = binary
-
-        self._command  = None
+class TarThread(PoolThread):
+    def __init__(self, backup_dir, output_file, compression='none', verbose=False, binary="tar"):
+        super(TarThread, self).__init__(self.__class__.__name__, compression)
+        self.compression_method = compression
+        self.backup_dir         = backup_dir
+        self.output_file        = output_file
+        self.verbose            = verbose
+        self.binary             = binary
 
-        signal(SIGINT, self.close)
-        signal(SIGTERM, self.close)
+        self._command = None
 
     def close(self, exit_code=None, frame=None):
-        if self._command:
-            logging.debug("Killing running subprocess/command: %s" % self._command.command)
+        if self._command and not self.stopped:
+            logging.debug("Stopping running tar command: %s" % self._command.command)
             del exit_code
             del frame
             self._command.close()
+            self.stopped = True
 
     def run(self):
         if os.path.isdir(self.backup_dir):
@@ -37,17 +34,20 @@ def run(self):
                      log_msg   = "Archiving directory: %s" % self.backup_dir
                      cmd_flags = ["-C", backup_base_dir, "-cf", self.output_file, "--remove-files", backup_base_name]
 
-                     if self.do_gzip:
+                     if self.do_gzip():
                          log_msg   = "Archiving and compressing directory: %s" % self.backup_dir
                          cmd_flags = ["-C", backup_base_dir, "-czf", self.output_file, "--remove-files", backup_base_name]
 
                      logging.info(log_msg)
+                     self.running  = True
                      self._command = LocalCommand(self.binary, cmd_flags, self.verbose)
-                     self._command.run()
+                     self.exit_code = self._command.run()
                 except Exception, e:
                     logging.fatal("Failed archiving file: %s! Error: %s" % (self.output_file, e))
-                    sys.exit(1)
+                finally:
+                    self.running   = False
+                    self.stopped   = True
+                    self.completed = True
             else:
                 logging.fatal("Output file: %s already exists!" % self.output_file)
-                sys.exit(1)
             return self.backup_dir

mongodb_consistent_backup/Main.py

@@ -269,8 +269,9 @@ def run(self):
                     self.db
                 )
                 replset_name = self.replset.get_rs_name()
+                replset_dir  = os.path.join(self.backup_directory, replset_name)
                 self.replsets[replset_name] = self.replset
-                state = StateBackupReplset(self.backup_directory, self.config, self.backup_time, replset_name)
+                state = StateBackupReplset(replset_dir, self.config, self.backup_time, replset_name)
                 state.load_state(self.replset.summary())
                 state.write()
             except Exception, e:
@@ -392,7 +393,8 @@ def run(self):
             try:
                 rs_sharded_summary = self.replset_sharded.summary()
                 for shard in rs_sharded_summary:
-                    state = StateBackupReplset(self.backup_directory, self.config, self.backup_time, shard)
+                    shard_dir = os.path.join(self.backup_directory, shard)
+                    state = StateBackupReplset(shard_dir, self.config, self.backup_time, shard)
                     state.load_state(rs_sharded_summary[shard])
                     state.write()
                 self.replset_sharded.close()
@@ -417,7 +419,8 @@ def run(self):
                 self.resolver.compression(self.oplogtailer.compression())
                 resolver_summary = self.resolver.run()
                 for shard in resolver_summary:
-                    state = StateOplog(self.backup_directory, self.config, self.backup_time, shard)
+                    shard_dir = os.path.join(self.backup_directory, shard)
+                    state = StateOplog(shard_dir, self.config, self.backup_time, shard)
                     state.load_state(resolver_summary[shard])
                     state.write()
                 self.resolver.close()

mongodb_consistent_backup/Oplog/Resolver/Resolver.py

@@ -106,7 +106,7 @@ def run(self):
                             tailed_oplog.copy(),
                             backup_oplog.copy(),
                             self.get_consistent_end_ts(),
-                            self.do_gzip()
+                            self.compression()
                         ).run, callback=self.done)
                         self._pooled.append(uri.str())
                     except Exception, e:

mongodb_consistent_backup/Oplog/Resolver/ResolverThread.py

@@ -5,24 +5,26 @@
 from bson import decode_file_iter
 
 from mongodb_consistent_backup.Oplog import Oplog
+from mongodb_consistent_backup.Pipeline import PoolThread
 
 
-class ResolverThread:
-    def __init__(self, state, uri, tailed_oplog, mongodump_oplog, max_end_ts, dump_gzip=False):
-        self.state           = state
-        self.uri             = uri
-        self.tailed_oplog    = tailed_oplog
-        self.mongodump_oplog = mongodump_oplog
-        self.max_end_ts      = max_end_ts
-        self.dump_gzip       = dump_gzip
+class ResolverThread(PoolThread):
+    def __init__(self, state, uri, tailed_oplog, mongodump_oplog, max_end_ts, compression='none'):
+        super(ResolverThread, self).__init__(self.__class__.__name__, compression)
+        self.state              = state
+        self.uri                = uri
+        self.tailed_oplog       = tailed_oplog
+        self.mongodump_oplog    = mongodump_oplog
+        self.max_end_ts         = max_end_ts
+        self.compression_method = compression
 
         self.oplogs  = {}
         self.changes = 0
         self.stopped = True
 
     def run(self):
-        self.oplogs['backup'] = Oplog(self.mongodump_oplog['file'], self.dump_gzip, 'a+')
-        self.oplogs['tailed'] = Oplog(self.tailed_oplog['file'], self.dump_gzip)
+        self.oplogs['backup'] = Oplog(self.mongodump_oplog['file'], self.do_gzip(), 'a+')
+        self.oplogs['tailed'] = Oplog(self.tailed_oplog['file'], self.do_gzip())
 
         logging.info("Resolving oplog for %s to max ts: %s" % (self.uri, self.max_end_ts))
         try:

mongodb_consistent_backup/Pipeline/PoolThread.py

@@ -0,0 +1,31 @@
+import logging
+
+from mongodb_consistent_backup.Errors import Error
+
+
+class PoolThread(object):
+    def __init__(self, thread_name, compression_method='none'):
+        self.thread_name        = thread_name
+        self.compression_method = compression_method
+
+        self.timer_name = self.__class__.__name__
+        self.stopped    = False
+        self.running    = False
+        self.completed  = False
+        self.exit_code  = 255
+
+    def compression(self, method=None):
+        if method:
+            self.compression_method = method
+        return self.compression_method
+
+    def do_gzip(self):
+        if self.compression() == 'gzip':
+            return True
+        return False
+
+    def run(self):
+        raise Error("Must define a .run() method when using %s class!" % self.__class__.__name__)
+
+    def close(self, code=None, frame=None):
+        raise Error("Must define a .close() method when using %s class!" % self.__class__.__name__)

mongodb_consistent_backup/Pipeline/__init__.py

@@ -1,2 +1,3 @@
+from PoolThread import PoolThread
 from Stage import Stage
 from Task import Task

mongodb_consistent_backup/State.py

@@ -73,12 +73,12 @@ def load_state(self, replset):
 
 class StateBackupReplset(StateBaseReplset):
     def __init__(self, base_dir, config, backup_time, set_name):
-        StateBaseReplset.__init__(self, base_dir, config, backup_time, set_name, "set.%s.bson" % set_name)
+        StateBaseReplset.__init__(self, base_dir, config, backup_time, set_name, "replset.bson")
 
 
 class StateOplog(StateBaseReplset):
     def __init__(self, base_dir, config, backup_time, set_name):
-        StateBaseReplset.__init__(self, base_dir, config, backup_time, set_name, "oplog.%s.bson" % set_name)
+        StateBaseReplset.__init__(self, base_dir, config, backup_time, set_name, "oplog.bson")
 
 
 class StateBackup(StateBase):