parent
4d6f96cd6c
commit
cbf7c2bf44
4 changed files with 58 additions and 55 deletions
|
@ -101,10 +101,6 @@ class BaseStorage(StoragePaths):
|
|||
raise NotImplementedError
|
||||
|
||||
|
||||
class InvalidChunkException(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
class BaseStorageV2(BaseStorage):
|
||||
def initiate_chunked_upload(self):
|
||||
""" Start a new chunked upload, returning the uuid and any associated storage metadata
|
||||
|
@ -113,9 +109,9 @@ class BaseStorageV2(BaseStorage):
|
|||
|
||||
def stream_upload_chunk(self, uuid, offset, length, in_fp, storage_metadata, content_type=None):
|
||||
""" Upload the specified amount of data from the given file pointer to the chunked destination
|
||||
specified, starting at the given offset. Returns the number of bytes uploaded, and a new
|
||||
version of the storage_metadata. Raises InvalidChunkException if the offset or length can
|
||||
not be accepted. Pass length as -1 to upload as much data from the in_fp as possible.
|
||||
specified, starting at the given offset. Returns the number of bytes uploaded, a new
|
||||
version of the storage_metadata and an error object (if one occurred or None if none).
|
||||
Pass length as -1 to upload as much data from the in_fp as possible.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
|
|
|
@ -174,7 +174,7 @@ class _CloudStorage(BaseStorageV2):
|
|||
|
||||
def _stream_write_internal(self, path, fp, content_type=None, content_encoding=None,
|
||||
cancel_on_error=True, size=filelike.READ_UNTIL_END):
|
||||
error = None
|
||||
write_error = None
|
||||
mp = self.__initiate_multipart_upload(path, content_type, content_encoding)
|
||||
|
||||
# We are going to reuse this but be VERY careful to only read the number of bytes written to it
|
||||
|
@ -199,9 +199,9 @@ class _CloudStorage(BaseStorageV2):
|
|||
mp.upload_part_from_file(buf, num_part, size=bytes_staged)
|
||||
total_bytes_written += bytes_staged
|
||||
num_part += 1
|
||||
except IOError as ex:
|
||||
logger.warn('stream write error: %s', ex)
|
||||
error = ex
|
||||
except IOError as e:
|
||||
logger.warn('Error when writing to stream in stream_write_internal at path %s: %s', path, e)
|
||||
write_error = e
|
||||
|
||||
if self._metric_queue is not None:
|
||||
self._metric_queue.put_deprecated('MultipartUploadFailure', 1)
|
||||
|
@ -209,7 +209,7 @@ class _CloudStorage(BaseStorageV2):
|
|||
|
||||
if cancel_on_error:
|
||||
mp.cancel_upload()
|
||||
return 0, error
|
||||
return 0, write_error
|
||||
else:
|
||||
break
|
||||
|
||||
|
@ -220,7 +220,7 @@ class _CloudStorage(BaseStorageV2):
|
|||
|
||||
mp.complete_upload()
|
||||
|
||||
return total_bytes_written, error
|
||||
return total_bytes_written, write_error
|
||||
|
||||
def exists(self, path):
|
||||
self._initialize_cloud_conn()
|
||||
|
@ -295,8 +295,9 @@ class _CloudStorage(BaseStorageV2):
|
|||
|
||||
# We are going to upload each chunk to a separate key
|
||||
chunk_path = self._rel_upload_path(str(uuid4()))
|
||||
bytes_written, error = self._stream_write_internal(chunk_path, in_fp, cancel_on_error=False,
|
||||
size=length, content_type=content_type)
|
||||
bytes_written, write_error = self._stream_write_internal(chunk_path, in_fp,
|
||||
cancel_on_error=False, size=length,
|
||||
content_type=content_type)
|
||||
|
||||
new_metadata = copy.deepcopy(storage_metadata)
|
||||
|
||||
|
@ -304,7 +305,7 @@ class _CloudStorage(BaseStorageV2):
|
|||
if bytes_written > 0:
|
||||
new_metadata[_CHUNKS_KEY].append(_PartUploadMetadata(chunk_path, offset, bytes_written))
|
||||
|
||||
return bytes_written, new_metadata, error
|
||||
return bytes_written, new_metadata, write_error
|
||||
|
||||
def _chunk_generator(self, chunk_list):
|
||||
for chunk in chunk_list:
|
||||
|
|
|
@ -91,8 +91,8 @@ class SwiftStorage(BaseStorage):
|
|||
_, obj = self._get_connection().get_object(self._swift_container, path,
|
||||
resp_chunk_size=chunk_size)
|
||||
return obj
|
||||
except Exception:
|
||||
logger.exception('Could not get object: %s', path)
|
||||
except Exception as ex:
|
||||
logger.exception('Could not get object at path %s: %s', path, ex)
|
||||
raise IOError('Path %s not found' % path)
|
||||
|
||||
def _put_object(self, path, content, chunk=None, content_type=None, content_encoding=None,
|
||||
|
@ -111,16 +111,16 @@ class SwiftStorage(BaseStorage):
|
|||
# We re-raise client exception here so that validation of config during setup can see
|
||||
# the client exception messages.
|
||||
raise
|
||||
except Exception:
|
||||
logger.exception('Could not put object: %s', path)
|
||||
except Exception as ex:
|
||||
logger.exception('Could not put object at path %s: %s', path, ex)
|
||||
raise IOError("Could not put content: %s" % path)
|
||||
|
||||
def _head_object(self, path):
|
||||
path = self._normalize_path(path)
|
||||
try:
|
||||
return self._get_connection().head_object(self._swift_container, path)
|
||||
except Exception:
|
||||
logger.exception('Could not head object: %s', path)
|
||||
except Exception as ex:
|
||||
logger.exception('Could not head object at path %s: %s', path, ex)
|
||||
return None
|
||||
|
||||
def get_direct_download_url(self, object_path, expires_in=60, requires_cors=False, head=False):
|
||||
|
@ -233,22 +233,24 @@ class SwiftStorage(BaseStorage):
|
|||
return random_uuid, metadata
|
||||
|
||||
def stream_upload_chunk(self, uuid, offset, length, in_fp, storage_metadata, content_type=None):
|
||||
error = None
|
||||
|
||||
if length == 0:
|
||||
return 0, storage_metadata, error
|
||||
return 0, storage_metadata, None
|
||||
|
||||
# Note: Swift limits segments to a maximum of 5GB, so we keep writing segments until we
|
||||
# are finished hitting the data limit.
|
||||
total_bytes_written = 0
|
||||
upload_error = None
|
||||
|
||||
while True:
|
||||
try:
|
||||
bytes_written, storage_metadata = self._stream_upload_segment(uuid, offset, length, in_fp,
|
||||
storage_metadata,
|
||||
content_type)
|
||||
except IOError as ex:
|
||||
logger.warn('stream write error: %s', ex)
|
||||
error = ex
|
||||
message = ('Error writing to stream in stream_upload_chunk for uuid %s (offset %s' +
|
||||
', length %s, metadata: %s): %s')
|
||||
logger.exception(message, uuid, offset, length, storage_metadata, ex)
|
||||
upload_error = ex
|
||||
break
|
||||
|
||||
if length != filelike.READ_UNTIL_END:
|
||||
|
@ -257,9 +259,9 @@ class SwiftStorage(BaseStorage):
|
|||
offset = offset + bytes_written
|
||||
total_bytes_written = total_bytes_written + bytes_written
|
||||
if bytes_written == 0 or length <= 0:
|
||||
return total_bytes_written, storage_metadata, error
|
||||
return total_bytes_written, storage_metadata, upload_error
|
||||
|
||||
return total_bytes_written, storage_metadata, error
|
||||
return total_bytes_written, storage_metadata, upload_error
|
||||
|
||||
def _stream_upload_segment(self, uuid, offset, length, in_fp, storage_metadata, content_type):
|
||||
updated_metadata = copy.deepcopy(storage_metadata)
|
||||
|
|
Reference in a new issue