🐛S3: when copying files there is no callback if the multipart threshold is not reached (#6305)

Author: sanderegg

packages/aws-library/src/aws_library/s3/_client.py

@@ -23,7 +23,7 @@
 from types_aiobotocore_s3.literals import BucketLocationConstraintType
 from types_aiobotocore_s3.type_defs import ObjectIdentifierTypeDef
 
-from ._constants import MULTIPART_UPLOADS_MIN_TOTAL_SIZE, PRESIGNED_LINK_MAX_SIZE
+from ._constants import MULTIPART_COPY_THRESHOLD, MULTIPART_UPLOADS_MIN_TOTAL_SIZE
 from ._error_handler import s3_exception_handler, s3_exception_handler_async_gen
 from ._errors import S3DestinationNotEmptyError, S3KeyNotFoundError
 from ._models import (
@@ -407,6 +407,7 @@ async def copy_object(
         src_object_key: S3ObjectKey,
         dst_object_key: S3ObjectKey,
         bytes_transfered_cb: CopiedBytesTransferredCallback | None,
+        object_metadata: S3MetaData | None = None,
     ) -> None:
         """copy a file in S3 using aioboto3 transfer manager (e.g. works >5Gb and creates multiple threads)"""
         copy_options: dict[str, Any] = {
@@ -415,7 +416,7 @@ async def copy_object(
             "Key": dst_object_key,
             "Config": TransferConfig(
                 max_concurrency=self.transfer_max_concurrency,
-                multipart_threshold=PRESIGNED_LINK_MAX_SIZE,
+                multipart_threshold=MULTIPART_COPY_THRESHOLD,
             ),
         }
         if bytes_transfered_cb:
@@ -424,7 +425,16 @@ async def copy_object(
                     bytes_transfered_cb, file_name=f"{dst_object_key}"
                 )
             }
+        # NOTE: boto3 copy function uses copy_object until 'multipart_threshold' is reached then switches to multipart copy
+        # copy_object does not provide any callbacks so we can't track progress so we need to ensure at least the completion
+        # of the object is tracked
         await self._client.copy(**copy_options)
+        if bytes_transfered_cb:
+            if object_metadata is None:
+                object_metadata = await self.get_object_metadata(
+                    bucket=bucket, object_key=dst_object_key
+                )
+            bytes_transfered_cb(object_metadata.size, file_name=f"{dst_object_key}")
 
     @s3_exception_handler(_logger)
     async def copy_objects_recursively(
@@ -448,6 +458,7 @@ async def copy_objects_recursively(
                     src_object_key=s3_object.object_key,
                     dst_object_key=s3_object.object_key.replace(src_prefix, dst_prefix),
                     bytes_transfered_cb=bytes_transfered_cb,
+                    object_metadata=s3_object,
                 )
                 async for s3_object in self._list_all_objects(
                     bucket=bucket, prefix=src_prefix

packages/aws-library/src/aws_library/s3/_constants.py

@@ -4,7 +4,7 @@
 
 # NOTE: AWS S3 upload limits https://docs.aws.amazon.com/AmazonS3/latest/userguide/qfacts.html
 MULTIPART_UPLOADS_MIN_TOTAL_SIZE: Final[ByteSize] = parse_obj_as(ByteSize, "100MiB")
-
+MULTIPART_COPY_THRESHOLD: Final[ByteSize] = parse_obj_as(ByteSize, "100MiB")
 
 PRESIGNED_LINK_MAX_SIZE: Final[ByteSize] = parse_obj_as(ByteSize, "5GiB")
 S3_MAX_FILE_SIZE: Final[ByteSize] = parse_obj_as(ByteSize, "5TiB")

services/storage/src/simcore_service_storage/s3_utils.py

@@ -9,15 +9,15 @@
     TaskProgress,
 )
 
-logger = logging.getLogger(__name__)
+_logger = logging.getLogger(__name__)
 
 
 def update_task_progress(
     task_progress: TaskProgress | None,
     message: ProgressMessage | None = None,
     progress: ProgressPercent | None = None,
 ) -> None:
-    logger.debug("%s [%s]", message or "", progress or "n/a")
+    _logger.debug("%s [%s]", message or "", progress or "n/a")
     if task_progress:
         task_progress.update(message=message, percent=progress)
 
@@ -53,7 +53,7 @@ def finalize_transfer(self) -> None:
         self._update()
 
     def copy_transfer_cb(self, total_bytes_copied: int, *, file_name: str) -> None:
-        logger.debug(
+        _logger.debug(
             "Copied %s of %s",
             parse_obj_as(ByteSize, total_bytes_copied).human_readable(),
             file_name,
@@ -64,6 +64,11 @@ def copy_transfer_cb(self, total_bytes_copied: int, *, file_name: str) -> None:
             self._update()
 
     def upload_transfer_cb(self, bytes_transferred: int, *, file_name: str) -> None:
+        _logger.debug(
+            "Uploaded %s of %s",
+            parse_obj_as(ByteSize, bytes_transferred).human_readable(),
+            file_name,
+        )
         self._file_total_bytes_copied[file_name] += bytes_transferred
         self._total_bytes_copied = sum(self._file_total_bytes_copied.values())
         if self.total_bytes_to_transfer != 0:

services/storage/tests/unit/test_simcore_s3_dsm.py

@@ -28,8 +28,8 @@ def file_size() -> ByteSize:
 
 
 @pytest.fixture
-def mock_copy_transfer_cb() -> Callable[[int], None]:
-    def copy_transfer_cb(copied_bytes: int) -> None:
+def mock_copy_transfer_cb() -> Callable[..., None]:
+    def copy_transfer_cb(total_bytes_copied: int, *, file_name: str) -> None:
         ...
 
     return copy_transfer_cb
@@ -43,7 +43,7 @@ async def test__copy_path_s3_s3(
     upload_file: Callable[[ByteSize, str], Awaitable[tuple[Path, SimcoreS3FileID]]],
     file_size: ByteSize,
     user_id: UserID,
-    mock_copy_transfer_cb: Callable[[int], None],
+    mock_copy_transfer_cb: Callable[..., None],
     aiopg_engine: Engine,
 ):
     def _get_dest_file_id(src: SimcoreS3FileID) -> SimcoreS3FileID: