From c9743ec510b43cd7c7498bb5a3663a8b0cc8adba Mon Sep 17 00:00:00 2001 From: Nuwan Goonasekera <2070605+nuwang@users.noreply.github.com> Date: Fri, 10 Jul 2026 23:36:00 +0530 Subject: [PATCH] Add cross-provider ranged, parallel downloads BucketObject.download_to_file fetches objects larger than the configured threshold as ranged reads of part_size bytes, up to max_concurrency parts in parallel, so large downloads are not bound to a single connection and the whole object is never held in memory. AWS delegates to boto3's TransferManager and Azure to azure-storage-blob's concurrent downloader; GCP and OpenStack Swift use the generic driver, which fetches ranges in parallel via cloned providers (memory bounded to ~concurrency * part_size; the partial file is removed on failure). A new BucketObjectService.download_range primitive is also usable directly for partial reads. Since the same threshold/part-size/concurrency knobs now tune transfers in both directions, UploadConfig (shipped only in 4.2.x, unused downstream) is renamed to TransferConfig with no alias kept. --- CHANGELOG.rst | 19 +- cloudbridge/base/resources.py | 100 +++++++- cloudbridge/interfaces/resources.py | 62 +++-- cloudbridge/interfaces/services.py | 27 +++ cloudbridge/providers/aws/resources.py | 24 +- cloudbridge/providers/aws/services.py | 10 + cloudbridge/providers/azure/azure_client.py | 6 + cloudbridge/providers/azure/resources.py | 14 +- cloudbridge/providers/azure/services.py | 8 + cloudbridge/providers/gcp/resources.py | 4 +- cloudbridge/providers/gcp/services.py | 11 + cloudbridge/providers/openstack/resources.py | 4 +- cloudbridge/providers/openstack/services.py | 10 + tests/test_download_driver.py | 236 +++++++++++++++++++ tests/test_multipart_driver.py | 4 +- tests/test_object_store_service.py | 105 ++++++++- 16 files changed, 597 insertions(+), 47 deletions(-) create mode 100644 tests/test_download_driver.py diff --git a/CHANGELOG.rst b/CHANGELOG.rst index c2c726fb..91402610 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -1,3 +1,20 @@ +4.3.0 - unreleased +------------------ + +## What's new +* **Cross-provider ranged, parallel downloads.** The new + ``BucketObject.download_to_file`` fetches objects larger than the configured + threshold as ranged reads of ``part_size`` bytes, up to ``max_concurrency`` + parts in parallel, so large downloads are no longer bound to a single + connection (and the whole object is never held in memory). AWS delegates to + boto3's TransferManager and Azure to azure-storage-blob's concurrent + downloader; GCP and OpenStack Swift use CloudBridge's generic driver, which + fetches ranges in parallel via cloned providers. A new + ``BucketObjectService.download_range`` primitive is also available directly + for partial reads. The per-call ``TransferConfig`` knobs (threshold, part + size, concurrency) now tune transfers in both directions — pass a different + instance per call to tune uploads and downloads differently. + 4.2.1 - July 10, 2026 (sha f06765d1100ec461908396475a4510460843a65c) -------------------------------------------------------------------- @@ -42,7 +59,7 @@ APIs it supported. underlying cloud SDKs are untyped. A ``mypy`` check runs in CI — the interface and base layers under a strict bar, and the SDK-wrapping providers under a pragmatic tier. * **Cross-provider multipart upload.** Large object uploads now use multipart transfer - with per-call ``UploadConfig`` knobs (threshold, part size, concurrency), with parts + with per-call ``TransferConfig`` knobs (threshold, part size, concurrency), with parts uploaded in parallel via cloned providers. ## Breaking changes diff --git a/cloudbridge/base/resources.py b/cloudbridge/base/resources.py index 343d5367..911aebae 100644 --- a/cloudbridge/base/resources.py +++ b/cloudbridge/base/resources.py @@ -59,7 +59,7 @@ from cloudbridge.interfaces.resources import SnapshotState from cloudbridge.interfaces.resources import Subnet from cloudbridge.interfaces.resources import SubnetState -from cloudbridge.interfaces.resources import UploadConfig +from cloudbridge.interfaces.resources import TransferConfig from cloudbridge.interfaces.resources import UploadPart from cloudbridge.interfaces.resources import VMFirewall from cloudbridge.interfaces.resources import VMFirewallRule @@ -884,23 +884,107 @@ def save_content(self, target_stream: IO[bytes]) -> None: shutil.copyfileobj( cast("SupportsRead[bytes]", self.iter_content()), target_stream) + def download_to_file(self, path: str, + config: TransferConfig | None = None) -> None: + size = self.size + if size <= self._multipart_threshold(config): + with open(path, 'wb') as f: + self.save_content(f) + return + self._download_ranged(path, size, config) + + def _download_ranged(self, path: str, size: int, + config: TransferConfig | None = None) -> None: + """ + Fetch the object as ranged reads across a bounded thread pool, + writing each range at its offset into a preallocated file. + + To stay safe even on providers whose SDK client/connection is not + thread-safe, each worker reads through its own cloned provider (see + :meth:`.CloudProvider.clone`), so no provider state is shared between + threads. Memory is bounded to ~concurrency * part_size. On any + failure the partial file is removed and the error re-raised. + + Providers with an efficient, thread-safe native downloader (e.g. AWS + via boto3's ``download_file``, Azure via ``download_blob``) override + ``download_to_file`` to use it directly. + """ + part_size = self._multipart_part_size(config) + if part_size < 1: + raise InvalidValueException('part_size', part_size) + concurrency = max(1, self._multipart_max_concurrency(config)) + ranges = [(offset, min(part_size, size - offset)) + for offset in range(0, size, part_size)] + try: + with open(path, 'wb') as f: + f.truncate(size) + if concurrency == 1: + bucket_objects = self._bucket_objects + with open(path, 'r+b') as f: + for offset, length in ranges: + f.seek(offset) + f.write(bucket_objects.download_range( + self.bucket, self.name, offset, length)) + else: + self._download_ranges_concurrently(path, ranges, concurrency) + except Exception: + try: + os.remove(path) + except OSError: + pass + raise + + def _download_ranges_concurrently( + self, path: str, ranges: list[tuple[int, int]], + concurrency: int) -> None: + # A pool of cloned bucket-object services, one per worker, so each + # thread touches an isolated provider/connection. + clones: "queue.Queue[BucketObjectService]" = queue.Queue() + for _ in range(concurrency): + storage = cast("BaseStorageService", + self._provider.clone().storage) + clones.put(storage._bucket_objects) + + bucket = self.bucket + name = self.name + + def fetch_one(offset: int, length: int) -> None: + service = clones.get() + try: + data = service.download_range(bucket, name, offset, length) + finally: + clones.put(service) + # Each worker writes through its own handle at its own offset; + # ranges never overlap, so no locking is needed. Data is released + # as soon as it is written, bounding memory to + # ~concurrency * part_size. + with open(path, 'r+b') as f: + f.seek(offset) + f.write(data) + + with ThreadPoolExecutor(max_workers=concurrency) as executor: + futures = [executor.submit(fetch_one, offset, length) + for offset, length in ranges] + for future in futures: + future.result() + # The three resolvers below pick, in order of precedence: an explicit - # per-call UploadConfig field, the provider/global config, then the class + # per-call TransferConfig field, the provider/global config, then the class # default constant. - def _multipart_threshold(self, config: UploadConfig | None = None) -> int: + def _multipart_threshold(self, config: TransferConfig | None = None) -> int: if config is not None and config.threshold is not None: return int(config.threshold) return int(self._provider._get_config_value( 'multipart_threshold', self.CB_MULTIPART_THRESHOLD)) - def _multipart_part_size(self, config: UploadConfig | None = None) -> int: + def _multipart_part_size(self, config: TransferConfig | None = None) -> int: if config is not None and config.part_size is not None: return int(config.part_size) return int(self._provider._get_config_value( 'multipart_part_size', self.CB_MULTIPART_PART_SIZE)) def _multipart_max_concurrency( - self, config: UploadConfig | None = None) -> int: + self, config: TransferConfig | None = None) -> int: if config is not None and config.max_concurrency is not None: return int(config.max_concurrency) return int(self._provider._get_config_value( @@ -936,7 +1020,7 @@ def _as_stream(data: str | bytes | IO[bytes]) -> IO[bytes]: return data def upload(self, data: str | bytes | IO[bytes], - config: UploadConfig | None = None) -> BucketObject: + config: TransferConfig | None = None) -> BucketObject: size = self._data_size(data) if size is not None and size > self._multipart_threshold(config): return self._upload_multipart(self._as_stream(data), config) @@ -944,14 +1028,14 @@ def upload(self, data: str | bytes | IO[bytes], def upload_from_file( self, path: str, - config: UploadConfig | None = None) -> BucketObject: + config: TransferConfig | None = None) -> BucketObject: if os.path.getsize(path) > self._multipart_threshold(config): with open(path, 'rb') as f: return self._upload_multipart(f, config) return self._upload_from_file_single_shot(path) def _upload_multipart(self, stream: IO[bytes], - config: UploadConfig | None = None) -> BucketObject: + config: TransferConfig | None = None) -> BucketObject: """ Drive the explicit multipart lifecycle over a stream, reading it one part at a time so the whole payload is never held in memory. diff --git a/cloudbridge/interfaces/resources.py b/cloudbridge/interfaces/resources.py index 4f0346b8..09d12fe9 100644 --- a/cloudbridge/interfaces/resources.py +++ b/cloudbridge/interfaces/resources.py @@ -2161,14 +2161,18 @@ def delete(self) -> None: pass -class UploadConfig(object): +class TransferConfig(object): """ - Provider-agnostic, per-call tuning for an object upload. + Provider-agnostic, per-call tuning for an object transfer in either + direction. - Passed optionally to :meth:`.BucketObject.upload` and - :meth:`.BucketObject.upload_from_file`. Any field left as ``None`` falls - back to the provider/global configuration (the ``CB_MULTIPART_*`` settings). - Each provider maps these fields onto its native transfer mechanism. + Passed optionally to :meth:`.BucketObject.upload`, + :meth:`.BucketObject.upload_from_file` and + :meth:`.BucketObject.download_to_file`. Any field left as ``None`` falls + back to the provider/global configuration (the ``CB_MULTIPART_*`` + settings). Each provider maps these fields onto its native transfer + mechanism. To tune uploads and downloads differently, pass a different + instance to each call. """ def __init__(self, threshold: int | None = None, @@ -2176,22 +2180,24 @@ def __init__(self, threshold: int | None = None, max_concurrency: int | None = None) -> None: """ :type threshold: ``int`` - :param threshold: Size in bytes above which the upload is split into - multiple parts. + :param threshold: Size in bytes above which the transfer is split + into multiple parts. :type part_size: ``int`` - :param part_size: Size in bytes of each part. Must be at least the - provider minimum (5 MiB on S3) for all but the final part. + :param part_size: Size in bytes of each part. For uploads this must + be at least the provider minimum (5 MiB on S3) for all but the + final part; downloads have no minimum. :type max_concurrency: ``int`` - :param max_concurrency: Maximum number of parts to upload in parallel. + :param max_concurrency: Maximum number of parts to transfer in + parallel. """ self.threshold = threshold self.part_size = part_size self.max_concurrency = max_concurrency def __repr__(self) -> str: - return ("".format( self.threshold, self.part_size, self.max_concurrency)) @@ -2368,13 +2374,37 @@ def save_content(self, target_stream: IO[bytes]) -> None: """ pass + @abstractmethod + def download_to_file(self, path: str, + config: TransferConfig | None = None) -> None: + """ + Download this object's content to a local file. + + Objects larger than the configured transfer threshold are fetched as + ranged reads of ``part_size`` bytes, up to ``max_concurrency`` parts + in parallel, so large downloads are not bound to a single connection + (and the whole object is never held in memory). Smaller objects are + streamed in a single request. ``iter_content``/``save_content`` + remain single-stream alternatives for arbitrary target streams. + + :type path: ``str`` + :param path: Local path to write the object's content to. An existing + file is overwritten; on failure no partial file is left behind. + + :type config: :class:`.TransferConfig` + :param config: Optional per-call transfer tuning (threshold, part + size, concurrency). Any field left unset falls back to the + provider/global configuration. + """ + pass + @abstractmethod def upload(self, source_stream: IO[bytes], - config: UploadConfig | None = None) -> BucketObject | None: + config: TransferConfig | None = None) -> BucketObject | None: """ Set the contents of the object to the data read from the source stream. - :type config: :class:`.UploadConfig` + :type config: :class:`.TransferConfig` :param config: Optional per-call upload tuning (multipart threshold, part size, concurrency). Any field left unset falls back to the provider/global configuration. @@ -2386,7 +2416,7 @@ def upload(self, source_stream: IO[bytes], @abstractmethod def upload_from_file(self, path: str, - config: UploadConfig | None = None) -> BucketObject | None: + config: TransferConfig | None = None) -> BucketObject | None: """ Store the contents of the file pointed by the "path" variable. @@ -2396,7 +2426,7 @@ def upload_from_file(self, path: str, :type path: ``str`` :param path: Absolute path to the file to be uploaded to S3. - :type config: :class:`.UploadConfig` + :type config: :class:`.TransferConfig` :param config: Optional per-call upload tuning (multipart threshold, part size, concurrency). Any field left unset falls back to the provider/global configuration. diff --git a/cloudbridge/interfaces/services.py b/cloudbridge/interfaces/services.py index 9570e038..e195bff0 100644 --- a/cloudbridge/interfaces/services.py +++ b/cloudbridge/interfaces/services.py @@ -1383,6 +1383,33 @@ def abort_multipart_upload(self, bucket: Bucket | str, """ pass + @abstractmethod + def download_range(self, bucket: Bucket | str, object_name: str, + offset: int, length: int) -> bytes: + """ + Read a byte range of an object. + + This is the primitive behind + :meth:`.BucketObject.download_to_file`'s parallel ranged downloads; + it is also usable directly for partial reads. + + :type bucket: :class:`.Bucket` + :param bucket: The bucket containing the object. + + :type object_name: ``str`` + :param object_name: The key of the object to read. + + :type offset: ``int`` + :param offset: Zero-based byte offset to start reading at. + + :type length: ``int`` + :param length: Number of bytes to read. + + :rtype: ``bytes`` + :return: The requested byte range of the object's content. + """ + pass + class SecurityService(CloudService): diff --git a/cloudbridge/providers/aws/resources.py b/cloudbridge/providers/aws/resources.py index e5994167..8299c9f7 100644 --- a/cloudbridge/providers/aws/resources.py +++ b/cloudbridge/providers/aws/resources.py @@ -13,7 +13,7 @@ from typing import TYPE_CHECKING from typing import cast -from boto3.s3.transfer import TransferConfig +from boto3.s3.transfer import TransferConfig as S3TransferConfig from botocore.exceptions import ClientError @@ -58,7 +58,7 @@ from cloudbridge.interfaces.resources import Subnet from cloudbridge.interfaces.resources import SubnetState from cloudbridge.interfaces.resources import TrafficDirection -from cloudbridge.interfaces.resources import UploadConfig +from cloudbridge.interfaces.resources import TransferConfig from cloudbridge.interfaces.resources import VMFirewall from cloudbridge.interfaces.resources import VMType from cloudbridge.interfaces.resources import Volume @@ -920,11 +920,11 @@ def _upload_single_shot(self, def _upload_multipart( self, stream: IO[bytes], - config: UploadConfig | None = None) -> BucketObject: + config: TransferConfig | None = None) -> BucketObject: # boto3's TransferManager uploads parts concurrently with a thread-safe # client, so the transparent multipart path delegates to it rather than # CloudBridge's generic clone-pool driver. - transfer_config = TransferConfig( + transfer_config = S3TransferConfig( multipart_threshold=self._multipart_part_size(config), multipart_chunksize=self._multipart_part_size(config), max_concurrency=self._multipart_max_concurrency(config)) @@ -932,18 +932,30 @@ def _upload_multipart( return self def upload_from_file(self, path: str, - config: UploadConfig | None = None) -> BucketObject: + config: TransferConfig | None = None) -> BucketObject: # boto3's upload_file streams large files in parts via its # TransferManager. Drive it with CloudBridge's multipart knobs so that # upload_from_file and upload() honour the same configuration rather # than boto3's own defaults. - transfer_config = TransferConfig( + transfer_config = S3TransferConfig( multipart_threshold=self._multipart_threshold(config), multipart_chunksize=self._multipart_part_size(config), max_concurrency=self._multipart_max_concurrency(config)) self._obj.upload_file(path, Config=transfer_config) return self + def download_to_file(self, path: str, + config: TransferConfig | None = None) -> None: + # boto3's TransferManager downloads large objects as parallel ranged + # GETs with a thread-safe client, so the transparent ranged path + # delegates to it rather than CloudBridge's generic clone-pool driver. + transfer_config = S3TransferConfig( + multipart_threshold=self._multipart_threshold(config), + multipart_chunksize=self._multipart_part_size(config), + max_concurrency=self._multipart_max_concurrency(config)) + self._obj.meta.client.download_file( + self._obj.bucket_name, self.id, path, Config=transfer_config) + def delete(self) -> None: self._obj.delete() diff --git a/cloudbridge/providers/aws/services.py b/cloudbridge/providers/aws/services.py index 2a9f03cf..53fd9e34 100644 --- a/cloudbridge/providers/aws/services.py +++ b/cloudbridge/providers/aws/services.py @@ -733,6 +733,16 @@ def abort_multipart_upload(self, bucket: Bucket | str, Bucket=cast(Any, bucket).name, Key=upload.object_name, UploadId=upload.id) + @dispatch(event="provider.storage._bucket_objects.download_range", + priority=BaseBucketObjectService.STANDARD_EVENT_PRIORITY) + def download_range(self, bucket: Bucket | str, object_name: str, + offset: int, length: int) -> bytes: + client = cast("AWSCloudProvider", self.provider).s3_conn.meta.client + response = client.get_object( + Bucket=cast(Any, bucket).name, Key=object_name, + Range='bytes=%d-%d' % (offset, offset + length - 1)) + return cast(bytes, response['Body'].read()) + class AWSComputeService(BaseComputeService): diff --git a/cloudbridge/providers/azure/azure_client.py b/cloudbridge/providers/azure/azure_client.py index dddd6e38..ed532de0 100644 --- a/cloudbridge/providers/azure/azure_client.py +++ b/cloudbridge/providers/azure/azure_client.py @@ -507,6 +507,12 @@ def delete_blob(self, container_name: str, blob_name: str, blob_client = self.blob_client(container_name, blob_name) blob_client.delete_blob(delete_snapshots) + def get_blob_range(self, container_name: str, blob_name: str, + offset: int, length: int) -> bytes: + blob_client = self.blob_client(container_name, blob_name) + downloader = blob_client.download_blob(offset=offset, length=length) + return downloader.readall() + def get_blob_url(self, container_name: Any, blob_name: str, expiry_time: int, writable: bool, content_disposition: str | None = None, diff --git a/cloudbridge/providers/azure/resources.py b/cloudbridge/providers/azure/resources.py index aa7ce9ef..251d76b6 100644 --- a/cloudbridge/providers/azure/resources.py +++ b/cloudbridge/providers/azure/resources.py @@ -62,7 +62,7 @@ from cloudbridge.interfaces.resources import Subnet from cloudbridge.interfaces.resources import SubnetState from cloudbridge.interfaces.resources import TrafficDirection -from cloudbridge.interfaces.resources import UploadConfig +from cloudbridge.interfaces.resources import TransferConfig from cloudbridge.interfaces.resources import VMFirewall from cloudbridge.interfaces.resources import VMType from cloudbridge.interfaces.resources import Volume @@ -293,7 +293,7 @@ def _upload_single_shot(self, data: str | bytes | IO[bytes]) -> BucketObject: return self def _upload_multipart(self, stream: IO[bytes], - config: UploadConfig | None = None) -> BucketObject: + config: TransferConfig | None = None) -> BucketObject: # The Azure SDK's upload_blob stages blocks concurrently (max_concurrency # workers) over a thread-safe client, so the transparent multipart path # delegates to it rather than CloudBridge's generic clone-pool driver. @@ -302,6 +302,16 @@ def _upload_multipart(self, stream: IO[bytes], max_concurrency=self._multipart_max_concurrency(config)) return self + def download_to_file(self, path: str, + config: TransferConfig | None = None) -> None: + # azure-storage-blob's downloader fetches block ranges concurrently + # with a thread-safe client, so delegate to it rather than + # CloudBridge's generic clone-pool driver. + with open(path, 'wb') as f: + self._blob_client.download_blob( + max_concurrency=self._multipart_max_concurrency(config) + ).readinto(f) + def delete(self) -> None: """ Delete this object. diff --git a/cloudbridge/providers/azure/services.py b/cloudbridge/providers/azure/services.py index 62cbc325..9eeb8dd0 100644 --- a/cloudbridge/providers/azure/services.py +++ b/cloudbridge/providers/azure/services.py @@ -766,6 +766,14 @@ def abort_multipart_upload(self, bucket: Bucket | str, "%s/%s will expire automatically.", cast("Bucket", bucket).name, upload.object_name) + @dispatch(event="provider.storage._bucket_objects.download_range", + priority=BaseBucketObjectService.STANDARD_EVENT_PRIORITY) + def download_range(self, bucket: Bucket | str, object_name: str, + offset: int, length: int) -> bytes: + azure_client = cast("AzureCloudProvider", self.provider).azure_client + return cast(bytes, azure_client.get_blob_range( + cast("Bucket", bucket).name, object_name, offset, length)) + class AzureComputeService(BaseComputeService): def __init__(self, provider: CloudProvider) -> None: diff --git a/cloudbridge/providers/gcp/resources.py b/cloudbridge/providers/gcp/resources.py index 5dbd06a8..6464ba81 100644 --- a/cloudbridge/providers/gcp/resources.py +++ b/cloudbridge/providers/gcp/resources.py @@ -81,7 +81,7 @@ from cloudbridge.interfaces.resources import Snapshot from cloudbridge.interfaces.resources import Subnet from cloudbridge.interfaces.resources import SubnetSubService - from cloudbridge.interfaces.resources import UploadConfig + from cloudbridge.interfaces.resources import TransferConfig from cloudbridge.interfaces.resources import VMFirewall from cloudbridge.interfaces.resources import VMFirewallRuleSubService from cloudbridge.interfaces.resources import VMType @@ -2169,7 +2169,7 @@ def _upload_single_shot(self, data: str | bytes | IO[bytes] return self def upload_from_file(self, path: str, - config: UploadConfig | None = None) -> BucketObject: + config: TransferConfig | None = None) -> BucketObject: """ Upload a binary file. diff --git a/cloudbridge/providers/gcp/services.py b/cloudbridge/providers/gcp/services.py index 3eb72a6b..1bb4fc19 100644 --- a/cloudbridge/providers/gcp/services.py +++ b/cloudbridge/providers/gcp/services.py @@ -1769,6 +1769,17 @@ def abort_multipart_upload(self, bucket: Bucket | str, self._delete_objects(bucket, self._list_temp_objects(bucket, upload), ignore_missing=True) + @dispatch(event="provider.storage._bucket_objects.download_range", + priority=BaseBucketObjectService.STANDARD_EVENT_PRIORITY) + def download_range(self, bucket: Bucket | str, object_name: str, + offset: int, length: int) -> bytes: + provider = cast("GCPCloudProvider", self.provider) + request = provider.gcp_storage.objects().get_media( + bucket=cast(Any, bucket).name, object=object_name) + request.headers['range'] = 'bytes=%d-%d' % ( + offset, offset + length - 1) + return cast(bytes, request.execute()) + def _compose(self, bucket: Bucket | str, upload: MultipartUpload, destination: str, sources: builtins.list[str]) -> builtins.list[str]: """ diff --git a/cloudbridge/providers/openstack/resources.py b/cloudbridge/providers/openstack/resources.py index d2f58419..c32a3a2e 100644 --- a/cloudbridge/providers/openstack/resources.py +++ b/cloudbridge/providers/openstack/resources.py @@ -64,7 +64,7 @@ from cloudbridge.interfaces.resources import Subnet from cloudbridge.interfaces.resources import SubnetState from cloudbridge.interfaces.resources import TrafficDirection -from cloudbridge.interfaces.resources import UploadConfig +from cloudbridge.interfaces.resources import TransferConfig from cloudbridge.interfaces.resources import VMFirewall from cloudbridge.interfaces.resources import VMType from cloudbridge.interfaces.resources import Volume @@ -1413,7 +1413,7 @@ def _upload_single_shot( return self def upload_from_file(self, path: str, - config: UploadConfig | None = None) -> BucketObject: + config: TransferConfig | None = None) -> BucketObject: """ Stores the contents of the file pointed by the ``path`` variable. If the file is bigger than 5 Gig, it will be broken into segments. diff --git a/cloudbridge/providers/openstack/services.py b/cloudbridge/providers/openstack/services.py index 5e2a713a..3864cc92 100644 --- a/cloudbridge/providers/openstack/services.py +++ b/cloudbridge/providers/openstack/services.py @@ -808,6 +808,16 @@ def abort_multipart_upload(self, bucket: Bucket | str, except SwiftClientException: pass # idempotent: ignore already-deleted segments + @dispatch(event="provider.storage._bucket_objects.download_range", + priority=BaseBucketObjectService.STANDARD_EVENT_PRIORITY) + def download_range(self, bucket: Bucket | str, object_name: str, + offset: int, length: int) -> bytes: + provider = cast("OpenStackCloudProvider", self.provider) + _, data = provider.swift.get_object( + cast(Any, bucket).name, object_name, + headers={'Range': 'bytes=%d-%d' % (offset, offset + length - 1)}) + return cast(bytes, data) + class OpenStackComputeService(BaseComputeService): diff --git a/tests/test_download_driver.py b/tests/test_download_driver.py new file mode 100644 index 00000000..773185c9 --- /dev/null +++ b/tests/test_download_driver.py @@ -0,0 +1,236 @@ +""" +Provider-agnostic unit tests for the base ranged download driver +(``BaseBucketObject.download_to_file`` / ``_download_ranged``). + +The driver is the engine behind transparent large downloads on providers that +do not override it (GCP, OpenStack Swift). Because the mock provider is +AWS-backed and AWS overrides the driver with boto3's native downloader, the +driver is exercised here directly against in-memory fakes so it has coverage +in CI without cloud credentials. +""" +import os +import tempfile +import threading +import unittest + +from cloudbridge.base.resources import BaseBucketObject +from cloudbridge.interfaces.exceptions import InvalidValueException +from cloudbridge.interfaces.resources import TransferConfig + + +class _Recorder: + """Thread-safe range log shared by the original and cloned fake + services.""" + + def __init__(self, content): + self.content = content + self._lock = threading.Lock() + self.ranges = [] # (offset, length) served + self.services_used = set() # id() of each service that served a range + self.clone_count = 0 + self.single_shot = False + self.active = 0 + self.max_active = 0 + self.fail_on_offset = None # offset that should raise + + def serve_range(self, service, offset, length): + with self._lock: + self.active += 1 + self.max_active = max(self.max_active, self.active) + try: + if self.fail_on_offset == offset: + raise RuntimeError("boom at offset %d" % offset) + # Hold briefly so concurrent fetches genuinely overlap. + threading.Event().wait(0.02) + with self._lock: + self.ranges.append((offset, length)) + self.services_used.add(id(service)) + return self.content[offset:offset + length] + finally: + with self._lock: + self.active -= 1 + + +class _FakeService: + def __init__(self, recorder, provider): + self._recorder = recorder + self._provider = provider + + def download_range(self, bucket, object_name, offset, length): + return self._recorder.serve_range(self, offset, length) + + +class _FakeStorage: + def __init__(self, service): + self._bucket_objects = service + + +class _FakeProvider: + def __init__(self, recorder): + self._recorder = recorder + self.storage = _FakeStorage(_FakeService(recorder, self)) + + def clone(self, zone=None): + self._recorder.clone_count += 1 + return _FakeProvider(self._recorder) + + def _get_config_value(self, key, default_value=None): + return default_value + + +class _DriverObject(BaseBucketObject): + """A BaseBucketObject wired to fakes with tiny transfer sizes.""" + + def __init__(self, provider, threshold, part_size, concurrency): + super(_DriverObject, self).__init__(provider) + self._threshold = threshold + self._part_size = part_size + self._concurrency = concurrency + + @property + def id(self): + return "obj" + + @property + def name(self): + return "obj" + + @property + def size(self): + return len(self._provider._recorder.content) + + @property + def bucket(self): + return "BUCKET" + + def save_content(self, target_stream): + self._provider._recorder.single_shot = True + target_stream.write(self._provider._recorder.content) + + def _multipart_threshold(self, config=None): + if config is not None and config.threshold is not None: + return config.threshold + return self._threshold + + def _multipart_part_size(self, config=None): + if config is not None and config.part_size is not None: + return config.part_size + return self._part_size + + def _multipart_max_concurrency(self, config=None): + if config is not None and config.max_concurrency is not None: + return config.max_concurrency + return self._concurrency + + +class DownloadDriverTestCase(unittest.TestCase): + + def _driver(self, recorder, threshold, part_size, concurrency): + return _DriverObject( + _FakeProvider(recorder), threshold, part_size, concurrency) + + def _download(self, driver, config=None): + fd, path = tempfile.mkstemp() + os.close(fd) + os.remove(path) + try: + driver.download_to_file(path, config) + with open(path, 'rb') as f: + return f.read() + finally: + if os.path.exists(path): + os.remove(path) + + def test_reassembles_content_in_order(self): + content = b"abcdefghijABCDEFGHIJ0123456789x" # 31 bytes -> 8 ranges + recorder = _Recorder(content) + driver = self._driver( + recorder, threshold=10, part_size=4, concurrency=3) + self.assertEqual(self._download(driver), content) + self.assertFalse(recorder.single_shot) + # Ranges tile the object exactly: no gaps, no overlap, short tail. + self.assertEqual( + sorted(recorder.ranges), + [(offset, min(4, 31 - offset)) for offset in range(0, 31, 4)]) + + def test_below_threshold_uses_single_shot(self): + content = b"tiny content" + recorder = _Recorder(content) + driver = self._driver( + recorder, threshold=100, part_size=4, concurrency=3) + self.assertEqual(self._download(driver), content) + self.assertTrue(recorder.single_shot) + self.assertEqual(recorder.ranges, []) + + def test_downloads_ranges_concurrently_via_cloned_services(self): + concurrency = 4 + content = bytes(range(12)) # 12 ranges of one byte each + recorder = _Recorder(content) + driver = self._driver( + recorder, threshold=1, part_size=1, concurrency=concurrency) + self.assertEqual(self._download(driver), content) + + # A clone per worker, reused across ranges. + self.assertEqual(recorder.clone_count, concurrency) + self.assertEqual(len(recorder.services_used), concurrency) + # Real parallelism happened, bounded by the configured concurrency. + self.assertGreater(recorder.max_active, 1) + self.assertLessEqual(recorder.max_active, concurrency) + + def test_single_concurrency_does_not_clone(self): + content = b"abcdefghij" + recorder = _Recorder(content) + driver = self._driver( + recorder, threshold=1, part_size=4, concurrency=1) + self.assertEqual(self._download(driver), content) + self.assertEqual(recorder.clone_count, 0) + self.assertEqual(recorder.max_active, 1) + + def test_per_call_config_overrides_concurrency(self): + content = bytes(range(12)) + recorder = _Recorder(content) + driver = self._driver( + recorder, threshold=1, part_size=1, concurrency=1) + result = None + fd, path = tempfile.mkstemp() + os.close(fd) + try: + driver.download_to_file(path, TransferConfig(max_concurrency=3)) + with open(path, 'rb') as f: + result = f.read() + finally: + os.remove(path) + self.assertEqual(result, content) + self.assertEqual(recorder.clone_count, 3) + self.assertGreater(recorder.max_active, 1) + self.assertLessEqual(recorder.max_active, 3) + + def test_removes_partial_file_and_raises_on_range_failure(self): + content = bytes(range(16)) + recorder = _Recorder(content) + recorder.fail_on_offset = 8 + driver = self._driver( + recorder, threshold=1, part_size=4, concurrency=2) + fd, path = tempfile.mkstemp() + os.close(fd) + os.remove(path) + try: + with self.assertRaises(Exception): + driver.download_to_file(path) + self.assertFalse(os.path.exists(path)) + finally: + if os.path.exists(path): + os.remove(path) + + def test_part_size_must_be_positive(self): + content = bytes(range(16)) + recorder = _Recorder(content) + driver = self._driver( + recorder, threshold=1, part_size=4, concurrency=2) + with self.assertRaises(InvalidValueException): + self._download(driver, TransferConfig(part_size=0)) + self.assertEqual(recorder.ranges, []) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_multipart_driver.py b/tests/test_multipart_driver.py index 2348b21d..a7b91b43 100644 --- a/tests/test_multipart_driver.py +++ b/tests/test_multipart_driver.py @@ -16,7 +16,7 @@ from cloudbridge.base.resources import BaseMultipartUpload from cloudbridge.base.resources import BaseUploadPart from cloudbridge.interfaces.exceptions import InvalidValueException -from cloudbridge.interfaces.resources import UploadConfig +from cloudbridge.interfaces.resources import TransferConfig class _Recorder: @@ -193,7 +193,7 @@ def test_per_call_config_overrides_concurrency(self): # bumps it to 3, engaging the clone-per-worker pool. driver = self._driver(recorder, part_size=1, concurrency=1) driver._upload_multipart(BytesIO(b"0123456789ab"), - UploadConfig(max_concurrency=3)) + TransferConfig(max_concurrency=3)) self.assertEqual(recorder.clone_count, 3) self.assertGreater(recorder.max_active, 1) self.assertLessEqual(recorder.max_active, 3) diff --git a/tests/test_object_store_service.py b/tests/test_object_store_service.py index 9fc6e31a..5382514a 100644 --- a/tests/test_object_store_service.py +++ b/tests/test_object_store_service.py @@ -14,7 +14,7 @@ from cloudbridge.interfaces.provider import TestMockHelperMixin from cloudbridge.interfaces.resources import Bucket from cloudbridge.interfaces.resources import BucketObject -from cloudbridge.interfaces.resources import UploadConfig +from cloudbridge.interfaces.resources import TransferConfig from tests import helpers from tests.helpers import ProviderTestBase @@ -452,7 +452,7 @@ def test_upload_from_file_uses_multipart_config(self): # knobs (not boto3's own defaults). This wiring is AWS-specific. if self.provider.PROVIDER_ID not in ('aws', 'mock'): self.skipTest("TransferConfig wiring is specific to AWS") - from boto3.s3.transfer import TransferConfig + from boto3.s3.transfer import TransferConfig as S3TransferConfig name = "cbtest-mpu-{0}".format(helpers.get_uuid()) test_bucket = self.provider.storage.buckets.create(name) @@ -476,16 +476,16 @@ def test_upload_from_file_uses_multipart_config(self): spy.assert_called_once() config = spy.call_args.kwargs['Config'] - self.assertIsInstance(config, TransferConfig) + self.assertIsInstance(config, S3TransferConfig) self.assertEqual(config.multipart_chunksize, 7 * 1024 * 1024) self.assertEqual(config.max_concurrency, 3) @helpers.skipIfNoService(['storage.buckets']) def test_per_call_upload_config_overrides_defaults(self): - # A per-call UploadConfig takes precedence over the global knobs. + # A per-call TransferConfig takes precedence over the global knobs. if self.provider.PROVIDER_ID not in ('aws', 'mock'): self.skipTest("TransferConfig wiring is specific to AWS") - from boto3.s3.transfer import TransferConfig + from boto3.s3.transfer import TransferConfig as S3TransferConfig name = "cbtest-mpu-{0}".format(helpers.get_uuid()) test_bucket = self.provider.storage.buckets.create(name) @@ -496,8 +496,8 @@ def test_per_call_upload_config_overrides_defaults(self): with cb_helpers.cleanup_action(lambda: obj.delete()): test_file = os.path.join( helpers.get_test_fixtures_folder(), 'logo.jpg') - cfg = UploadConfig(part_size=8 * 1024 * 1024, - max_concurrency=2) + cfg = TransferConfig(part_size=8 * 1024 * 1024, + max_concurrency=2) # pylint:disable=protected-access with mock.patch.object( obj._obj, 'upload_file', @@ -505,10 +505,99 @@ def test_per_call_upload_config_overrides_defaults(self): obj.upload_from_file(test_file, config=cfg) config = spy.call_args.kwargs['Config'] - self.assertIsInstance(config, TransferConfig) + self.assertIsInstance(config, S3TransferConfig) self.assertEqual(config.multipart_chunksize, 8 * 1024 * 1024) self.assertEqual(config.max_concurrency, 2) + @helpers.skipIfNoService(['storage.buckets']) + def test_download_to_file_roundtrip(self): + name = "cbtest-dl-{0}".format(helpers.get_uuid()) + test_bucket = self.provider.storage.buckets.create(name) + + with cb_helpers.cleanup_action(lambda: test_bucket.delete()): + obj = test_bucket.objects.create("roundtrip.txt") + + with cb_helpers.cleanup_action(lambda: obj.delete()): + content = b"a small payload below the download threshold" + obj.upload(content) + + stored = test_bucket.objects.get("roundtrip.txt") + fd, path = tempfile.mkstemp() + os.close(fd) + with cb_helpers.cleanup_action(lambda: os.remove(path)): + stored.download_to_file(path) + with open(path, 'rb') as f: + self.assertEqual(f.read(), content) + + @helpers.skipIfNoService(['storage.buckets']) + def test_transparent_download_large_uses_ranged(self): + name = "cbtest-dl-{0}".format(helpers.get_uuid()) + test_bucket = self.provider.storage.buckets.create(name) + + with cb_helpers.cleanup_action(lambda: test_bucket.delete()): + obj = test_bucket.objects.create("ranged.bin") + + with cb_helpers.cleanup_action(lambda: obj.delete()): + # Distinct byte pattern so out-of-order reassembly would show. + content = bytes(range(256)) * ((MIN_PART_SIZE * 2) // 256) + obj.upload(BytesIO(content), + TransferConfig(threshold=MIN_PART_SIZE, + part_size=MIN_PART_SIZE)) + + stored = test_bucket.objects.get("ranged.bin") + fd, path = tempfile.mkstemp() + os.close(fd) + with cb_helpers.cleanup_action(lambda: os.remove(path)): + # Lower the threshold so the download crosses it (each + # provider routes its own way underneath) and assert the + # content round-trips exactly. + stored.download_to_file( + path, TransferConfig(threshold=MIN_PART_SIZE, + part_size=MIN_PART_SIZE, + max_concurrency=2)) + with open(path, 'rb') as f: + self.assertEqual(f.read(), content) + + @helpers.skipIfNoService(['storage.buckets']) + def test_download_to_file_uses_transfer_config(self): + # AWS drives boto3's TransferManager with CloudBridge's transfer + # knobs (not boto3's own defaults). This wiring is AWS-specific. + if self.provider.PROVIDER_ID not in ('aws', 'mock'): + self.skipTest("TransferConfig wiring is specific to AWS") + from boto3.s3.transfer import TransferConfig as S3TransferConfig + + name = "cbtest-dl-{0}".format(helpers.get_uuid()) + test_bucket = self.provider.storage.buckets.create(name) + + with cb_helpers.cleanup_action(lambda: test_bucket.delete()): + obj = test_bucket.objects.create("dlconfig.bin") + + with cb_helpers.cleanup_action(lambda: obj.delete()): + obj.upload(b"some downloadable content") + stored = test_bucket.objects.get("dlconfig.bin") + + fd, path = tempfile.mkstemp() + os.close(fd) + with cb_helpers.cleanup_action(lambda: os.remove(path)): + cfg = TransferConfig(part_size=8 * 1024 * 1024, + max_concurrency=2) + # pylint:disable=protected-access + with mock.patch.object( + stored._obj.meta.client, 'download_file', + wraps=stored._obj.meta.client + .download_file) as spy: + stored.download_to_file(path, config=cfg) + + spy.assert_called_once() + config = spy.call_args.kwargs['Config'] + self.assertIsInstance(config, S3TransferConfig) + self.assertEqual( + config.multipart_chunksize, 8 * 1024 * 1024) + self.assertEqual(config.max_concurrency, 2) + with open(path, 'rb') as f: + self.assertEqual( + f.read(), b"some downloadable content") + @skip("Skip unless you want to test objects bigger than 5GB") @helpers.skipIfNoService(['storage.buckets']) def test_upload_download_bucket_content_with_large_file(self):