Add retries to gs upload

py/utils/gs_utils.py

Index: py/utils/gs_utils.py
diff --git a/py/utils/gs_utils.py b/py/utils/gs_utils.py
index a125b68a633e28d7d8161af242a2e91c4359e869..2973f7446b4314a03ca1f66ea8d340a5941a676e 100644
--- a/py/utils/gs_utils.py
+++ b/py/utils/gs_utils.py
@@ -19,12 +19,14 @@ API/library references:
 # System-level imports
 import errno
 import hashlib
+import math
 import os
 import posixpath
 import Queue
 import re
 import sys
 import threading
+import time
 
 # Imports from third-party code
 TRUNK_DIRECTORY = os.path.abspath(os.path.join(
@@ -386,6 +388,8 @@ class GSUtils(object):
     for rel_path in source_fileset:
       q.put(rel_path)
 
+    err = {}
+
     # Spin up worker threads to read from the task queue.
     def worker():
       while True:
@@ -395,13 +399,25 @@ class GSUtils(object):
           return  # no more tasks in the queue, so exit
         print (' Uploading file %d/%d: %s' % (
             num_files_to_upload - q.qsize(), num_files_to_upload, rel_path))
-        self.upload_file(
-            source_path=os.path.join(source_dir, rel_path),
-            dest_bucket=b,
-            dest_path=posixpath.join(dest_dir, rel_path),
-            upload_if=self.UploadIf.ALWAYS,
-            **kwargs)
-        q.task_done()
+
+        retries = 5
+        for retry in range(retries):
+          try:
+            self.upload_file(
+                source_path=os.path.join(source_dir, rel_path),
+                dest_bucket=b,
+                dest_path=posixpath.join(dest_dir, rel_path),
+                upload_if=self.UploadIf.ALWAYS,
+                **kwargs)
+            q.task_done()
+            break
+          except Exception as error:
+            if retry < retries - 1:
+              print '  Retrying upload, attempt #%d' % retry + 1
+              time.sleep(math.pow(2, range))

[mtklein, 2015/04/14 17:17:14]

time.sleep(2 ** retry)  ?

[borenet, 2015/04/14 17:20:19]

Done.

+            else:
+              err[rel_path] = error
+
     for _ in range(num_threads):
       t = threading.Thread(target=worker)
       t.daemon = True
@@ -410,6 +426,12 @@ class GSUtils(object):
     # Block until all files have been uploaded and all workers have exited.
     q.join()
 
+    if err:
+      errMsg = 'Failed to upload the following: \n\n'
+      for rel_path, e in err.iteritems():
+        errMsg += '%s: %s\n' % (rel_path, e)
+      raise Exception(errMsg)
+
   def download_file(self, source_bucket, source_path, dest_path,
                     create_subdirs_if_needed=False, source_generation=None):
     """Downloads a single file from Google Cloud Storage to local disk.