Fixes for problems when sending Ctrl+C

Author: timvaillancourt

mongodb_consistent_backup/Archive/Tar/Tar.py

@@ -26,7 +26,10 @@ def __init__(self, manager, config, timer, base_dir, backup_dir, **kwargs):
         self.backup_base_dir = backup_dir
         self.verbose         = self.config.verbose
         self.binary          = "tar"
-        self._pool           = None
+
+        self._pool     = None
+        self.stopped   = False
+        self.completed = False
 
     def compression(self, method=None):
         if method:
@@ -72,10 +75,12 @@ def run(self):
                 raise Error(e)
             self._pool.close()
             self._pool.join()
+            self.completed = True
 
     def close(self):
         logging.debug("Stopping tar archiving threads")
-        if self._pool is not None:
+        if not self.stopped and self._pool is not None:
             self._pool.terminate()
             self._pool.join()
             logging.info("Stopped all tar archiving threads")
+            self.stopped = True

mongodb_consistent_backup/Backup/Mongodump/Mongodump.py

@@ -40,6 +40,7 @@ def __init__(self, manager, config, timer, base_dir, backup_dir, **kwargs):
         signal(SIGINT, SIG_IGN)
         signal(SIGINT, self.close)
 
+        self.completed            = False
         self.timer_name           = self.__class__.__name__
         self.threads_per_dump_max = 16
         self.config_replset       = False
@@ -174,6 +175,7 @@ def run(self):
                 self.threads[0].start()
                 self.wait()
 
+        self.completed = True
         return self._summary
 
     def close(self):

mongodb_consistent_backup/Backup/Mongodump/MongodumpThread.py

@@ -105,11 +105,16 @@ def run(self):
             self._process = Popen(mongodump_cmd, stderr=PIPE)
             self.wait()
             self.exit_code = self._process.returncode
+            if self.exit_code > 0:
+                sys.exit(self.exit_code)
         except Exception, e:
             logging.exception("Error performing mongodump: %s" % e)
 
-        oplog = Oplog(self.oplog_file, self.dump_gzip)
-        oplog.load()
+        try:
+            oplog = Oplog(self.oplog_file, self.dump_gzip)
+            oplog.load()
+        except Exception, e:
+            logging.exception("Error loading oplog: %s" % e)
 
         self.state.set('running', False)
         self.state.set('completed', True)
@@ -122,5 +127,3 @@ def run(self):
         if oplog.last_ts():
             log_msg_extra = "%s, end ts: %s" % (log_msg_extra, oplog.last_ts())
         logging.info("Backup %s completed in %.2f seconds, %s" % (self.uri, self.timer.duration(self.timer_name), log_msg_extra))
-
-        sys.exit(self.exit_code)

mongodb_consistent_backup/Common/Timer.py

@@ -11,20 +11,25 @@ def start(self, timer_name):
         self.timers[timer_name] = { 'start': time(), 'started': True }
 
     def stop(self, timer_name):
-        if timer_name in self.timers and 'started' in self.timers[timer_name]:
-            timer = self.timers.copy()[timer_name]
-            del timer['started']
-            timer['end'] = time()
-            timer['stopped'] = True
-            timer['duration'] = timer['end'] - timer['start']
-            self.timers[timer_name] = timer
-        else:
-            raise OperationError("No started timer named %s to stop!" % timer_name)
+        try:
+            if timer_name in self.timers and 'started' in self.timers[timer_name]:
+                timer = self.timers.copy()[timer_name]
+                del timer['started']
+                timer['end'] = time()
+                timer['stopped'] = True
+                timer['duration'] = timer['end'] - timer['start']
+                self.timers[timer_name] = timer
+            else:
+                raise OperationError("No started timer named %s to stop!" % timer_name)
+        except IOError:
+            pass            
 
     def duration(self, timer_name):
-        if timer_name in self.timers and 'duration' in self.timers[timer_name]:
-            return self.timers[timer_name]['duration']
-        else:
+        try:
+            if timer_name in self.timers and 'duration' in self.timers[timer_name]:
+                return self.timers[timer_name]['duration']
+            return 0
+        except IOError:
             return 0
 
     def dump(self, timer_name=None):

mongodb_consistent_backup/Notify/Notify.py

@@ -9,6 +9,7 @@ def __init__(self, manager, config, timer, base_dir, backup_dir):
         super(Notify, self).__init__(self.__class__.__name__, manager, config, timer, base_dir, backup_dir)
         self.method = self.config.notify.method
 
+        self.completed = False
         self.notifications = []
         self.init()
 
@@ -30,6 +31,7 @@ def run(self, *args):
                 except:
                     continue
             self.timers.stop(self.stage)
+        self.completed = True
 
     def close(self):
         if self._method:

mongodb_consistent_backup/Oplog/Resolver/Resolver.py

@@ -65,25 +65,13 @@ def threads(self, threads=None):
         return int(self.config.oplog.resolver.threads)
 
     def get_consistent_end_ts(self):
-        end_ts        = None
-        min_tailed_ts = None
-        max_backup_ts = None
+        ts = None
         for shard in self.tailed_oplogs:
             instance = self.tailed_oplogs[shard]
             if 'last_ts' in instance and instance['last_ts'] is not None:
-                if min_tailed_ts is None or instance['last_ts'].time < min_tailed_ts.time:
-                    min_tailed_ts = instance['last_ts']
-        end_ts = min_tailed_ts
-        for shard in self.backup_oplogs:
-            instance = self.backup_oplogs[shard]
-            if 'last_ts' in instance and instance['last_ts'] is not None:
-                if max_backup_ts is None or instance['last_ts'].time > max_backup_ts.time:
-                    max_backup_ts = instance['last_ts']
-        if max_backup_ts:
-            end_ts = Timestamp(max_backup_ts.time + 1, 0)
-        if end_ts > min_tailed_ts:
-            return Error("Backup maximum end time is greater than tailed oplog minimum end time. This should not happen!")
-        return end_ts
+                if ts is None or instance['last_ts'].time < ts.time:
+                    ts = Timestamp(instance['last_ts'].time, 0)
+        return ts
 
     def run(self):
         logging.info("Resolving oplogs (options: threads=%s,compression=%s)" % (self.threads(), self.compression()))

mongodb_consistent_backup/Pipeline/Stage.py

@@ -15,8 +15,9 @@ def __init__(self, stage_name, manager, config, timers, base_dir, backup_dir, **
         self.backup_dir = backup_dir
         self.args       = kwargs
 
-        self.running  = False
-        self.stopped  = False
+        self.running   = False
+        self.stopped   = False
+        self.completed = False
 
         self.stage   = "mongodb_consistent_backup.%s" % self.stage_name
         self.module  = None
@@ -72,14 +73,19 @@ def run(self):
             try:
                 self.timers.start(self.stage)
                 self.running = True
-                logging.info("Running backup stage %s with method: %s" % (self.stage, self.method.capitalize()))
+                logging.info("Running stage %s with method: %s" % (self.stage, self.method.capitalize()))
                 data = self._method.run()
             except Exception, e:
                 raise OperationError(e)
             finally:
                 self.running = False
                 self.stopped = True
                 self.timers.stop(self.stage)
-                logging.info("Completed running backup stage %s with method %s in %.2f seconds" % (self.stage, self.method.capitalize(), self.timers.duration(self.stage)))
+                if self._method.completed:
+                    logging.info("Completed running stage %s with method %s in %.2f seconds" % (self.stage, self.method.capitalize(), self.timers.duration(self.stage)))
+                    self.completed = True
+                else:
+                    logging.error("Stage %s did not complete!" % self.stage)
+                    raise OperationError("Stage %s did not complete!" % self.stage)
                 self.close()
             return data

mongodb_consistent_backup/Upload/S3/S3.py

@@ -36,6 +36,7 @@ def __init__(self, config, source_dir, key_prefix):
         self.chunk_size_mb   = self.config.upload.s3.chunk_size_mb
         self.chunk_size      = self.chunk_size_mb * 1024 * 1024
 
+        self.completed    = False
         self.timer_name   = self.__class__.__name__
         self._pool        = None
         self._multipart   = None
@@ -51,74 +52,75 @@ def __init__(self, config, source_dir, key_prefix):
     def run(self):
         if not os.path.isdir(self.source_dir):
             logging.error("The source directory: %s does not exist or is not a directory! Skipping AWS S3 Upload!" % self.source_dir)
-        else:
-            try:
-                self.timer.start(self.timer_name)
-                for file_name in os.listdir(self.source_dir):
-                    if self.bucket_prefix == "/":
-                        key_name = "/%s/%s" % (self.key_prefix, file_name)
-                    else:
-                        key_name = "%s/%s/%s" % (self.bucket_prefix, self.key_prefix, file_name)
-
-                    file_path = os.path.join(self.source_dir, file_name)
-                    file_size = os.stat(file_path).st_size
-                    chunk_count = int(ceil(file_size / float(self.chunk_size)))
-
-                    logging.info("Starting multipart AWS S3 upload to key: %s%s using %i threads, %imb chunks, %i retries" % (
+            return
+        try:
+            self.timer.start(self.timer_name)
+            for file_name in os.listdir(self.source_dir):
+                if self.bucket_prefix == "/":
+                    key_name = "/%s/%s" % (self.key_prefix, file_name)
+                else:
+                    key_name = "%s/%s/%s" % (self.bucket_prefix, self.key_prefix, file_name)
+
+                file_path = os.path.join(self.source_dir, file_name)
+                file_size = os.stat(file_path).st_size
+                chunk_count = int(ceil(file_size / float(self.chunk_size)))
+
+                logging.info("Starting multipart AWS S3 upload to key: %s%s using %i threads, %imb chunks, %i retries" % (
+                    self.bucket_name,
+                    key_name,
+                    self.thread_count,
+                    self.chunk_size_mb,
+                    self.retries
+                ))
+                self._multipart = self.bucket.initiate_multipart_upload(key_name)
+                self._pool      = Pool(processes=self.thread_count)
+
+                for i in range(chunk_count):
+                    offset = self.chunk_size * i
+                    byte_count = min(self.chunk_size, file_size - offset)
+                    part_num = i + 1
+                    self._pool.apply_async(S3UploadThread(
                         self.bucket_name,
-                        key_name,
-                        self.thread_count,
-                        self.chunk_size_mb,
-                        self.retries
-                    ))
-                    self._multipart = self.bucket.initiate_multipart_upload(key_name)
-                    self._pool      = Pool(processes=self.thread_count)
-
-                    for i in range(chunk_count):
-                        offset = self.chunk_size * i
-                        byte_count = min(self.chunk_size, file_size - offset)
-                        part_num = i + 1
-                        self._pool.apply_async(S3UploadThread(
-                            self.bucket_name,
-                            self.access_key,
-                            self.secret_key,
-                            self.s3_host,
-                            self._multipart.id,
-                            part_num,
-                            file_path,
-                            offset,
-                            byte_count,
-                            self.retries,
-                            self.secure
-                        ).run)
-                    self._pool.close()
-                    self._pool.join()
-    
-                    if len(self._multipart.get_all_parts()) == chunk_count:
-                        self._multipart.complete_upload()
-                        key = self.bucket.get_key(key_name)
-                        key.set_acl(self.s3_acl)
-                        self._upload_done = True
-
-                        if self.remove_uploaded:
-                            logging.info("Uploaded AWS S3 key: %s%s successfully. Removing local file" % (self.bucket_name, key_name))
-                            os.remove(os.path.join(self.source_dir, file_name))
-                        else:
-                            logging.info("Uploaded AWS S3 key: %s%s successfully" % (self.bucket_name, key_name))
+                        self.access_key,
+                        self.secret_key,
+                        self.s3_host,
+                        self._multipart.id,
+                        part_num,
+                        file_path,
+                        offset,
+                        byte_count,
+                        self.retries,
+                        self.secure
+                    ).run)
+                self._pool.close()
+                self._pool.join()
+
+                if len(self._multipart.get_all_parts()) == chunk_count:
+                    self._multipart.complete_upload()
+                    key = self.bucket.get_key(key_name)
+                    key.set_acl(self.s3_acl)
+                    self._upload_done = True
+
+                    if self.remove_uploaded:
+                        logging.info("Uploaded AWS S3 key: %s%s successfully. Removing local file" % (self.bucket_name, key_name))
+                        os.remove(os.path.join(self.source_dir, file_name))
                     else:
-                        self._multipart.cancel_upload()
-                        logging.error("Failed to upload all multiparts for key: %s%s! Upload cancelled" % (self.bucket_name, key_name))
-                        raise OperationError("Failed to upload all multiparts for key: %s%s! Upload cancelled" % (self.bucket_name, key_name))
-
-                if self.remove_uploaded:
-                    logging.info("Removing backup source dir after successful AWS S3 upload of all backups")
-                    os.rmdir(self.source_dir)
-                self.timer.stop(self.timer_name)
-            except Exception, e:
-                logging.error("Uploading to AWS S3 failed! Error: %s" % e)
-                if self._multipart:
+                        logging.info("Uploaded AWS S3 key: %s%s successfully" % (self.bucket_name, key_name))
+                else:
                     self._multipart.cancel_upload()
-                raise OperationError(e)
+                    logging.error("Failed to upload all multiparts for key: %s%s! Upload cancelled" % (self.bucket_name, key_name))
+                    raise OperationError("Failed to upload all multiparts for key: %s%s! Upload cancelled" % (self.bucket_name, key_name))
+
+            if self.remove_uploaded:
+                logging.info("Removing backup source dir after successful AWS S3 upload of all backups")
+                os.rmdir(self.source_dir)
+            self.timer.stop(self.timer_name)
+        except Exception, e:
+            logging.error("Uploading to AWS S3 failed! Error: %s" % e)
+            if self._multipart:
+                self._multipart.cancel_upload()
+            raise OperationError(e)
+        self.completed = True
 
     def close(self):
         if self._pool: