use the response queue to check for success instead of a method call (which seems to work inconsistently with threading)

Author: Tim Vaillancourt

MongoBackup/Mongodump.py

@@ -1,7 +1,7 @@
 import os
 import logging
 
-from multiprocessing import Process, Event
+from multiprocessing import Process
 from signal import signal, SIGINT, SIGTERM
 from time import time
 
@@ -28,7 +28,7 @@ def __init__(self, response_queue, backup_name, host_port, user, password, authd
         self.verbose        = verbose
 
         self._command   = None
-        self._complete  = Event()
+        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
@@ -37,13 +37,18 @@ def __init__(self, response_queue, backup_name, host_port, user, password, authd
         signal(SIGINT, self.close)
         signal(SIGTERM, self.close)
 
-    def complete(self):
-        return self._complete.is_set()
-
     def close(self, exit_code=None, frame=None):
         if self._command:
             logging.debug("Killing running subprocess/command: %s" % self._command.command)
             self._command.close()
+            self.response_queue.put({
+                'host': self.host,
+                'port': self.port,
+                'file': self.oplog_file,
+                'count': oplog.count(),
+                'last_ts': oplog.last_ts(),
+                'first_ts': oplog.first_ts()
+            })
 
     def run(self):
         logging.info("Starting mongodump (with oplog) backup of %s/%s:%i" % (
@@ -77,15 +82,16 @@ def run(self):
             return None
 
         oplog = OplogInfo(self.oplog_file, self.dump_gzip)
+        self.completed = True
         self.response_queue.put({
             'host': self.host,
             'port': self.port,
             'file': self.oplog_file,
             'count': oplog.count(),
             'last_ts': oplog.last_ts(),
-            'first_ts': oplog.first_ts()
+            'first_ts': oplog.first_ts(),
+            'completed': self.completed
         })
-        self._complete.set()
 
         time_diff = time() - self.start_time
         logging.info("Backup for %s/%s:%s completed in %s sec with %i oplog changes captured to: %s" % (

MongoBackup/Mongodumper.py

@@ -113,23 +113,28 @@ def run(self):
         for thread in self.threads:
             thread.start()
 
-        # wait for all threads to finish and check their status
+        # wait for all threads to finish
         for thread in self.threads:
             thread.join()
-            if not thread.complete():
-                raise Exception, "Not all mongodump threads completed successfully!", None
 
         # 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
+
+        # fail if all threads did not complete
+        if not completed == len(self.threads):
+            raise Exception, "Not all mongodump threads completed successfully!", None
 
         logging.info("All mongodump backups completed")