Add retries to gs upload

py/utils/gs_utils.py

 #!/usr/bin/python
 
 # pylint: disable=C0301
 """
 Copyright 2014 Google Inc.
 
 Use of this source code is governed by a BSD-style license that can be
 found in the LICENSE file.
 
 Utilities for accessing Google Cloud Storage, using the boto library (wrapper
 for the XML API).
 
 API/library references:
 - https://developers.google.com/storage/docs/reference-guide
 - http://googlecloudstorage.blogspot.com/2012/09/google-cloud-storage-tutorial-using-boto.html
 """
 # pylint: enable=C0301
 
 # 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(
     os.path.dirname(__file__), os.pardir, os.pardir))
 for import_subdir in ['boto']:
   import_dirpath = os.path.join(
       TRUNK_DIRECTORY, 'third_party', 'externals', import_subdir)
   if import_dirpath not in sys.path:
     # We need to insert at the beginning of the path, to make sure that our
     # imported versions are favored over others that might be in the path.
@@ skipping 341 matching lines @@
     if num_files_to_upload == 0:
       return
     if num_threads > num_files_to_upload:
       num_threads = num_files_to_upload
 
     # Create a work queue with all files that need to be uploaded.
     q = Queue.Queue(maxsize=num_files_to_upload)
     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:
         try:
           rel_path = q.get(block=False)
         except Queue.Empty:
           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
       t.start()
 
     # 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.
 
     Args:
       source_bucket: GS bucket to download the file from
       source_path: full path (Posix-style) within that bucket
       dest_path: full path (local-OS-style) on local disk to copy the file to
       create_subdirs_if_needed: boolean; whether to create subdirectories as
           needed to create dest_path
@@ skipping 281 matching lines @@
 
 def _get_local_md5(path):
   """Returns the MD5 hash of a file on local disk."""
   hasher = hashlib.md5()
   with open(path, 'rb') as f:
     while True:
       data = f.read(64*1024)
       if not data:
         return hasher.hexdigest()
       hasher.update(data)