Rethrow Isolate gRPC proxy exceptions as IOError

client/isolate_storage.py

 #!/usr/bin/env python
 # Copyright 2016 The LUCI Authors. All rights reserved.
 # Use of this source code is governed under the Apache License, Version 2.0
 # that can be found in the LICENSE file.
 
 """A low-level blob storage/retrieval interface to the Isolate server"""
 
 import base64
 import binascii
 import collections
@@ skipping 524 matching lines @@
 
   def fetch(self, digest, offset=0):
     # The gRPC APIs only work with an offset of 0
     assert offset == 0
     request = isolate_bot_pb2.FetchBlobsRequest()
     req_digest = request.digest.add()
     # Convert the utf-8 encoded hexidecimal string (like '012abc') to a byte
     # array (like [0x01, 0x2a, 0xbc]).
     req_digest.digest = binascii.unhexlify(digest)
     expected_offset = 0
-    for response in self._stub.FetchBlobs(request,
-                                          timeout=DOWNLOAD_READ_TIMEOUT):
-      if not response.status.succeeded:
-        raise IOError(
-            'Error while fetching %s: %s' % (digest, response.status))
-      if not expected_offset == response.data.offset:
-        raise IOError(
-            'Error while fetching %s: expected offset %d, got %d' % (
-                digest, expected_offset, response.data.offset))
-      expected_offset += len(response.data.data)
-      yield response.data.data
+    try:
+      for response in self._stub.FetchBlobs(request,
+                                            timeout=DOWNLOAD_READ_TIMEOUT):
+        if not response.status.succeeded:
+          raise IOError(
+              'Error while fetching %s: %s' % (digest, response.status))
+        if not expected_offset == response.data.offset:
+          raise IOError(
+              'Error while fetching %s: expected offset %d, got %d' % (
+                  digest, expected_offset, response.data.offset))
+        expected_offset += len(response.data.data)
+        yield response.data.data
+    except grpc.RpcError as g:
+      logging.error('gRPC error during fetch: re-throwing as IOError (%s)' % g)
+      raise IOError(g)
 
   def push(self, item, push_state, content=None):
     assert isinstance(item, Item)
     assert item.digest is not None
     assert item.size is not None
     assert isinstance(push_state, _IsolateServerGrpcPushState)
 
     # Default to item.content().
     content = item.content() if content is None else content
     guard_memory_use(self, content, item.size)
@@ skipping 21 matching lines @@
             slice_len = min(len(chunk), NET_IO_FILE_CHUNK)
             request.data.data = chunk[:slice_len]
             yield request
             has_sent_anything = True
             request.data.offset += slice_len
             # The proxy only expects the first chunk to have the digest
             request.data.ClearField("digest")
             chunk = chunk[slice_len:]
 
       # TODO(aludwin): batch up several requests to reuse TCP connections
-      response = self._stub.PushBlobs(slicer())
+      try:
+        response = self._stub.PushBlobs(slicer())
+      except grpc.RpcError as g:
+        logging.error('gRPC error during push: re-throwing as IOError (%s)' % g)
+        raise IOError(g)
+
       if not response.status.succeeded:
         raise IOError(
             'Error while uploading %s: %s' % (
                 item.digest, response.status.error_detail))
 
     finally:
       with self._lock:
         self._memory_use -= item.size
 
   def contains(self, items):
     """Returns the set of all missing items."""
     # Ensure all items were initialized with 'prepare' call. Storage does that.
     assert all(i.digest is not None and i.size is not None for i in items)
     request = isolate_bot_pb2.ContainsRequest()
     items_by_digest = {}
     for item in items:
       cd = request.digest.add()
       cd.digest = binascii.unhexlify(item.digest)
       items_by_digest[cd.digest] = item
-    response = self._stub.Contains(request)
+    try:
+      response = self._stub.Contains(request)
+    except grpc.RpcError as g:
+      logging.error('gRPC error during contains: re-throwing as IOError (%s)'
+                    % g)
+      raise IOError(g)
 
     # If everything's present, return the empty set.
     if response.status.succeeded:
       return {}
 
     if not response.status.error == isolate_bot_pb2.BlobStatus.MISSING_DIGEST:
       raise IOError('Unknown response during lookup: %s' % response.status)
 
     # Pick Items that are missing, attach _PushState to them. The gRPC
     # implementation doesn't actually have a push state, we just attach
@@ skipping 27 matching lines @@
     url: URL of isolate service to use shared cloud based storage.
     namespace: isolate namespace to operate in, also defines hashing and
         compression scheme used, i.e. namespace names that end with '-gzip'
         store compressed data.
 
   Returns:
     Instance of StorageApi subclass.
   """
   cls = _storage_api_cls or IsolateServer
   return cls(url, namespace)