Move 'threads' and 'retries' to top-level upload stage. Added missing retry loop to Gs upload

Author: timvaillancourt

mongodb_consistent_backup/Upload/Gs/Gs.py

@@ -27,12 +27,13 @@ def __init__(self, manager, config, timer, base_dir, backup_dir, **kwargs):
         super(Gs, self).__init__(self.__class__.__name__, manager, config, timer, base_dir, backup_dir, **kwargs)
         self.backup_location = self.config.backup.location
         self.remove_uploaded = self.config.upload.remove_uploaded
+        self.retries         = self.config.upload.retries
         self.project_id      = self.config.upload.gs.project_id
         self.access_key      = self.config.upload.gs.access_key
         self.secret_key      = self.config.upload.gs.secret_key
         self.bucket          = self.config.upload.gs.bucket
 
-        self.threads(self.config.upload.gs.threads)
+        self.threads(self.config.upload.threads)
         self._pool = Pool(processes=self.threads())
 
     def close(self):
@@ -69,7 +70,8 @@ def run(self):
                     self.project_id,
                     self.access_key,
                     self.secret_key,
-                    self.remove_uploaded
+                    self.remove_uploaded,
+                    self.retries
                 ).run)
             self._pool.close()
             self._pool.join()

mongodb_consistent_backup/Upload/Gs/GsUploadThread.py

@@ -7,7 +7,7 @@
 
 
 class GsUploadThread:
-    def __init__(self, backup_dir, file_path, gs_path, bucket, project_id, access_key, secret_key, remove_uploaded=False):
+    def __init__(self, backup_dir, file_path, gs_path, bucket, project_id, access_key, secret_key, remove_uploaded=False, retries=5):
         self.backup_dir      = backup_dir
         self.file_path       = file_path
         self.gs_path         = gs_path
@@ -16,6 +16,7 @@ def __init__(self, backup_dir, file_path, gs_path, bucket, project_id, access_ke
         self.access_key      = access_key
         self.secret_key      = secret_key
         self.remove_uploaded = remove_uploaded
+        self.retries         = retries
 
         self.path          = "%s/%s" % (self.bucket, self.gs_path)
         self.meta_data_dir = "mongodb_consistent_backup-META"
@@ -76,10 +77,21 @@ def run(self):
                 logging.debug("Path %s does not exist, uploading" % self.path)
 
             try:
-                f   = open(self.file_path, 'r')
-                uri = self.get_uri()
-                logging.info("Uploading %s to Google Cloud Storage" % self.path)
-                uri.new_key().set_contents_from_file(f)
+                f     = open(self.file_path, 'r')
+                uri   = self.get_uri()
+                retry = 0
+                error = None
+                while retry < self.retries:
+                    try:
+                        logging.info("Uploading %s to Google Cloud Storage (attempt %i/%i)" % (self.path, retry, self.retries))
+                        uri.new_key().set_contents_from_file(f)
+                    except Exception, e:
+                        logging.error("Received error for Google Cloud Storage upload of %s: %s" % (self.path, e))
+                        error  = e
+                        retry += 1
+                        continue
+                if retry >= self.retries and error:
+                    raise error
             finally:
                 if f:
                     f.close()

mongodb_consistent_backup/Upload/Gs/__init__.py

@@ -7,5 +7,4 @@ def config(parser):
     parser.add_argument("--upload.gs.secret_key", dest="upload.gs.secret_key", help="Google Cloud Storage Uploader Secret Key (required for GS upload)", type=str)
     parser.add_argument("--upload.gs.bucket_name", dest="upload.gs.bucket_name", help="Google Cloud Storage Uploader destination bucket name", type=str)
     parser.add_argument("--upload.gs.bucket_prefix", dest="upload.gs.bucket_prefix", help="Google Cloud Storage Uploader destination bucket path prefix", type=str)
-    parser.add_argument("--upload.gs.threads", dest="upload.gs.threads", help="Google Cloud Storage Uploader worker threads (default: 4)", default=4, type=int)
     return parser

mongodb_consistent_backup/Upload/S3/S3.py

@@ -29,16 +29,16 @@ class S3(Task):
     def __init__(self, manager, config, timer, base_dir, backup_dir, **kwargs):
         super(S3, self).__init__(self.__class__.__name__, manager, config, timer, base_dir, backup_dir, **kwargs)
         self.remove_uploaded = self.config.upload.remove_uploaded
+        self.retries         = self.config.upload.retries
+        self.thread_count    = self.config.upload.threads
         self.region          = self.config.upload.s3.region
         self.bucket_name     = self.config.upload.s3.bucket_name
         self.bucket_prefix   = self.config.upload.s3.bucket_prefix
         self.access_key      = self.config.upload.s3.access_key
         self.secret_key      = self.config.upload.s3.secret_key
-        self.thread_count    = self.config.upload.s3.threads
         self.chunk_size_mb   = self.config.upload.s3.chunk_size_mb
         self.chunk_size      = self.chunk_size_mb * 1024 * 1024
         self.secure          = self.config.upload.s3.secure
-        self.retries         = self.config.upload.s3.retries
         self.s3_acl          = self.config.upload.s3.acl
         self.key_prefix      = base_dir
 

mongodb_consistent_backup/Upload/S3/__init__.py

@@ -7,9 +7,7 @@ def config(parser):
     parser.add_argument("--upload.s3.secret_key", dest="upload.s3.secret_key", help="S3 Uploader AWS Secret Key (required for S3 upload)", type=str)
     parser.add_argument("--upload.s3.bucket_name", dest="upload.s3.bucket_name", help="S3 Uploader destination bucket name", type=str)
     parser.add_argument("--upload.s3.bucket_prefix", dest="upload.s3.bucket_prefix", help="S3 Uploader destination bucket path prefix", type=str)
-    parser.add_argument("--upload.s3.threads", dest="upload.s3.threads", help="S3 Uploader worker threads (default: 4)", default=4, type=int)
     parser.add_argument("--upload.s3.chunk_size_mb", dest="upload.s3.chunk_size_mb", help="S3 Uploader upload chunk size, in megabytes (default: 50)", default=50, type=int)
     parser.add_argument("--upload.s3.secure", dest="upload.s3.secure", help="S3 Uploader connect over SSL (default: true)", default=True, action="store_false")
-    parser.add_argument("--upload.s3.retries", dest="upload.s3.retries", help="S3 Uploader retry times (default: 5)", default=5, type=int)
     parser.add_argument("--upload.s3.acl", dest="upload.s3.acl", help="S3 Uploader ACL associated with objects (default: none)", default=None, type=str)
     return parser

mongodb_consistent_backup/Upload/__init__.py

@@ -4,4 +4,6 @@
 def config(parser):
     parser.add_argument("--upload.method", dest="upload.method", help="Uploader method (default: none)", default='none', choices=['gs', 's3', 'none'])
     parser.add_argument("--upload.remove_uploaded", dest="upload.remove_uploaded", help="Remove source files after successful upload (default: false)", default=False, action="store_true")
+    parser.add_argument("--upload.retries", dest="upload.retries", help="Number of times to retry upload attempts (default: 5)", default=5, type=int)
+    parser.add_argument("--upload.threads", dest="upload.threads", help="Number of threads to use for upload (default: 4)", default=4, type=int)
     return parser