diff --git a/.changes/next-release/bugfix-crt-24627.json b/.changes/next-release/bugfix-crt-24627.json
new file mode 100644
index 000000000000..bd7f9d50bff4
--- /dev/null
+++ b/.changes/next-release/bugfix-crt-24627.json
@@ -0,0 +1,5 @@
+{
+ "type": "bugfix",
+ "category": "crt",
+ "description": "Return error when final rename task fails on downloads"
+}
diff --git a/.changes/next-release/enhancement-crt-31704.json b/.changes/next-release/enhancement-crt-31704.json
new file mode 100644
index 000000000000..d57b54c3782d
--- /dev/null
+++ b/.changes/next-release/enhancement-crt-31704.json
@@ -0,0 +1,5 @@
+{
+ "type": "enhancement",
+ "category": "crt",
+ "description": "Support ``multipart_threshold`` (upload only) and ``max_concurrent_requests`` config options for CRT client."
+}
diff --git a/.changes/next-release/enhancement-crt-44299.json b/.changes/next-release/enhancement-crt-44299.json
new file mode 100644
index 000000000000..8588ab3341b4
--- /dev/null
+++ b/.changes/next-release/enhancement-crt-44299.json
@@ -0,0 +1,5 @@
+{
+ "type": "enhancement",
+ "category": "crt",
+ "description": "Enforce minimum 10gbps target throughput for explicitly configured crt environments"
+}
diff --git a/.changes/next-release/enhancement-crt-53269.json b/.changes/next-release/enhancement-crt-53269.json
new file mode 100644
index 000000000000..6401268c6786
--- /dev/null
+++ b/.changes/next-release/enhancement-crt-53269.json
@@ -0,0 +1,5 @@
+{
+ "type": "enhancement",
+ "category": "crt",
+ "description": "Set lower 4gbps target throughput default for non-EC2 hosts."
+}
diff --git a/.changes/next-release/enhancement-crt-69912.json b/.changes/next-release/enhancement-crt-69912.json
new file mode 100644
index 000000000000..0725537f4642
--- /dev/null
+++ b/.changes/next-release/enhancement-crt-69912.json
@@ -0,0 +1,5 @@
+{
+ "type": "enhancement",
+ "category": "crt",
+ "description": "Configure CRT client to download with single GET when object size is below ``multipart_threshold``"
+}
diff --git a/.changes/next-release/enhancement-crt-76250.json b/.changes/next-release/enhancement-crt-76250.json
new file mode 100644
index 000000000000..f97371337080
--- /dev/null
+++ b/.changes/next-release/enhancement-crt-76250.json
@@ -0,0 +1,5 @@
+{
+ "type": "enhancement",
+ "category": "crt",
+ "description": "Pass resolved ``max_attempts`` value to the CRT client."
+}
diff --git a/.changes/next-release/enhancement-s3-42614.json b/.changes/next-release/enhancement-s3-42614.json
new file mode 100644
index 000000000000..ba821821fb0c
--- /dev/null
+++ b/.changes/next-release/enhancement-s3-42614.json
@@ -0,0 +1,5 @@
+{
+ "type": "enhancement",
+ "category": "``s3``",
+ "description": "Follow bucket region redirects for the ``s3`` commands when CRT is enabled."
+}
diff --git a/awscli/botocore/configprovider.py b/awscli/botocore/configprovider.py
index 56a25bd2aa87..01f04a402912 100644
--- a/awscli/botocore/configprovider.py
+++ b/awscli/botocore/configprovider.py
@@ -21,6 +21,7 @@
from botocore import utils
from botocore.exceptions import InvalidConfigError
+
def _resolve_new_retries():
_env_new_retries = os.environ.get('AWS_NEW_RETRIES_2026')
if _env_new_retries is not None:
@@ -451,6 +452,24 @@ def get_config_variable(self, logical_name):
provider = self._mapping[logical_name]
return provider.provide()
+ def is_explicitly_set(self, logical_name):
+ """
+ Determine whether a value was configured for the specified
+ logical_name, as opposed to resolving to a built-in default.
+
+ :type logical_name: str
+ :param logical_name: The logical name of the session variable
+ you want to check.
+
+ :returns: True if a value was configured, False otherwise.
+ """
+ if logical_name in self._overrides:
+ return True
+ if logical_name not in self._mapping:
+ return False
+ resolved = self._mapping[logical_name].resolve()
+ return resolved is not None and resolved.is_configured
+
def get_config_provider(self, logical_name):
"""
Retrieve the provider associated with the specified logical_name.
@@ -525,6 +544,14 @@ def set_config_provider(self, logical_name, provider):
self._mapping[logical_name] = provider
+class ConfigValue:
+ """A resolved config value and whether it came from a configured source."""
+
+ def __init__(self, value, is_configured=True):
+ self.value = value
+ self.is_configured = is_configured
+
+
class BaseProvider:
"""Base class for configuration value providers.
@@ -536,6 +563,17 @@ def provide(self):
"""Provide a config value."""
raise NotImplementedError('provide')
+ def resolve(self):
+ """Provide a config value along with where it came from.
+
+ :rtype: Optional[ConfigValue]
+ :returns: The resolved value, or None if this provider has none.
+ """
+ value = self.provide()
+ if value is None:
+ return None
+ return ConfigValue(value)
+
class ChainProvider(BaseProvider):
"""This provider wraps one or more other providers.
@@ -568,10 +606,16 @@ def provide(self):
one in the chain to return a non-None value is the returned from the
ChainProvider. When no non-None value is found, None is returned.
"""
+ resolved = self.resolve()
+ return resolved.value if resolved is not None else None
+
+ def resolve(self):
for provider in self._providers:
- value = provider.provide()
- if value is not None:
- return self._convert_type(value)
+ resolved = provider.resolve()
+ if resolved is not None:
+ return ConfigValue(
+ self._convert_type(resolved.value), resolved.is_configured
+ )
return None
def _convert_type(self, value):
@@ -717,6 +761,9 @@ def provide(self):
"""Provide the constant value given during initialization."""
return self._value
+ def resolve(self):
+ return ConfigValue(self._value, is_configured=False)
+
def __repr__(self):
return f'ConstantProvider(value={self._value})'
diff --git a/awscli/botocore/utils.py b/awscli/botocore/utils.py
index a3ac03e745e8..da7fe682324c 100644
--- a/awscli/botocore/utils.py
+++ b/awscli/botocore/utils.py
@@ -1576,7 +1576,7 @@ class S3RegionRedirectorv2:
"""
def __init__(self, endpoint_bridge, client, cache=None):
- self._cache = cache or {}
+ self._cache = {} if cache is None else cache
self._client = weakref.proxy(client)
def register(self, event_emitter=None):
@@ -1616,48 +1616,13 @@ def redirect_from_error(self, request_dict, response, operation, **kwargs):
)
return
- error = response[1].get('Error', {})
- error_code = error.get('Code')
- response_metadata = response[1].get('ResponseMetadata', {})
-
- # We have to account for 400 responses because
- # if we sign a Head* request with the wrong region,
- # we'll get a 400 Bad Request but we won't get a
- # body saying it's an "AuthorizationHeaderMalformed".
- is_special_head_object = (
- error_code in ('301', '400') and operation.name == 'HeadObject'
- )
- is_special_head_bucket = (
- error_code in ('301', '400')
- and operation.name == 'HeadBucket'
- and 'x-amz-bucket-region'
- in response_metadata.get('HTTPHeaders', {})
- )
- is_wrong_signing_region = (
- error_code == 'AuthorizationHeaderMalformed' and 'Region' in error
- )
- is_redirect_status = response[0] is not None and response[
- 0
- ].status_code in (301, 302, 307)
- is_permanent_redirect = error_code == 'PermanentRedirect'
- is_opt_in_region_redirect = (
- error_code == 'IllegalLocationConstraintException'
- and operation.name != 'CreateBucket'
- )
- if not any(
- [
- is_special_head_object,
- is_wrong_signing_region,
- is_permanent_redirect,
- is_special_head_bucket,
- is_redirect_status,
- is_opt_in_region_redirect,
- ]
- ):
+ bucket = redirect_ctx.get('bucket')
+ if bucket is None:
+ return
+ if not self.is_redirect_response(response, operation):
return
- bucket = request_dict['context']['s3_redirect']['bucket']
- client_region = request_dict['context'].get('client_region')
+ client_region = request_dict.get('context', {}).get('client_region')
new_region = self.get_bucket_region(bucket, response)
if new_region is None:
@@ -1702,6 +1667,55 @@ def redirect_from_error(self, request_dict, response, operation, **kwargs):
# Return 0 so it doesn't wait to retry
return 0
+ def get_redirect_region(self, bucket, response, operation):
+ """Return the region a response redirects a bucket to, if any."""
+ if bucket is None or ArnParser.is_arn(bucket):
+ return None
+ if not self.is_redirect_response(response, operation):
+ return None
+ return self.get_bucket_region(bucket, response)
+
+ def is_redirect_response(self, response, operation):
+ """Return whether a response says the bucket is in another region."""
+ error = response[1].get('Error', {})
+ error_code = error.get('Code')
+ response_metadata = response[1].get('ResponseMetadata', {})
+
+ # We have to account for 400 responses because
+ # if we sign a Head* request with the wrong region,
+ # we'll get a 400 Bad Request but we won't get a
+ # body saying it's an "AuthorizationHeaderMalformed".
+ is_special_head_object = (
+ error_code in ('301', '400') and operation.name == 'HeadObject'
+ )
+ is_special_head_bucket = (
+ error_code in ('301', '400')
+ and operation.name == 'HeadBucket'
+ and 'x-amz-bucket-region'
+ in response_metadata.get('HTTPHeaders', {})
+ )
+ is_wrong_signing_region = (
+ error_code == 'AuthorizationHeaderMalformed' and 'Region' in error
+ )
+ is_redirect_status = response[0] is not None and response[
+ 0
+ ].status_code in (301, 302, 307)
+ is_permanent_redirect = error_code == 'PermanentRedirect'
+ is_opt_in_region_redirect = (
+ error_code == 'IllegalLocationConstraintException'
+ and operation.name != 'CreateBucket'
+ )
+ return any(
+ [
+ is_special_head_object,
+ is_wrong_signing_region,
+ is_permanent_redirect,
+ is_special_head_bucket,
+ is_redirect_status,
+ is_opt_in_region_redirect,
+ ]
+ )
+
def get_bucket_region(self, bucket, response):
"""
There are multiple potential sources for the new region to redirect to,
diff --git a/awscli/customizations/s3/factory.py b/awscli/customizations/s3/factory.py
index 50ba57294400..7e99adcd775d 100644
--- a/awscli/customizations/s3/factory.py
+++ b/awscli/customizations/s3/factory.py
@@ -11,6 +11,8 @@
# ANY KIND, either express or implied. See the License for the specific
# language governing permissions and limitations under the License.
import logging
+import os
+import sys
import awscrt.s3
from botocore.client import Config
@@ -18,20 +20,77 @@
from s3transfer.crt import (
BotocoreCRTCredentialsWrapper,
BotocoreCRTRequestSerializer,
+ CRTTransferConfig,
CRTTransferManager,
acquire_crt_s3_process_lock,
+ create_crt_client_bootstrap,
create_s3_crt_client,
+ release_crt_s3_process_lock,
)
from s3transfer.manager import TransferManager
from awscli.compat import urlparse
from awscli.customizations.s3 import constants
from awscli.customizations.s3.transferconfig import (
+ DEFAULTS,
+ InvalidConfigError,
create_transfer_config_from_runtime_config,
)
+from awscli.customizations.utils import uni_print
LOGGER = logging.getLogger(__name__)
+ADAPTIVE_RETRY_MODE = 'adaptive'
+
+# A max_retries of 0 configures the crt client's own retry count instead of
+# disabling retries, so it cannot honor a single attempt. It also rejects a
+# max_retries of 64 or more outright.
+MIN_CRT_MAX_ATTEMPTS = 2
+MAX_CRT_MAX_ATTEMPTS = 64
+
+# Throughput target, in gigabits per second, for hosts the crt client has not
+# been tuned for. Staying at 4 keeps it in its smallest memory pool tier.
+UNTUNED_TARGET_THROUGHPUT_GBPS = 4.0
+
+# Throughput target, in gigabits per second, to fall back to rather than
+# accepting a lower recommendation from the crt.
+MINIMUM_TARGET_THROUGHPUT_GBPS = 10.0
+
+# The crt client rejects a part size over half of its memory pool while it is
+# being constructed. The pool is sized from the throughput target, and neither
+# the sizing nor the limit is exposed, so the only way to know a multipart
+# chunksize does not fit is to build the client and see. awscrt raises a plain
+# RuntimeError for this, leaving the error code as the only thing to match on.
+CRT_PART_SIZE_EXCEEDS_MEMORY_LIMIT = 14371
+
+WARN_IGNORED = 'warn_ignored'
+
+EXCLUDE_FROM_AUTO = 'exclude_from_auto'
+
+UNSUPPORTED_OPTIONS = {
+ constants.CRT_TRANSFER_CLIENT: {
+ 'max_bandwidth': EXCLUDE_FROM_AUTO,
+ 'max_queue_size': WARN_IGNORED,
+ 'io_chunksize': WARN_IGNORED,
+ },
+ constants.CLASSIC_TRANSFER_CLIENT: {
+ 'target_bandwidth': WARN_IGNORED,
+ 'should_stream': WARN_IGNORED,
+ 'disk_throughput': WARN_IGNORED,
+ 'direct_io': WARN_IGNORED,
+ },
+}
+
+CRT_CLIENT_KWARG_MAP = {
+ 'multipart_chunksize': 'part_size',
+ 'multipart_threshold': 'multipart_upload_threshold',
+ 'max_concurrent_requests': 'max_active_connections_override',
+}
+
+
+def _gbps_to_bytes_per_sec(gbps):
+ return int(gbps * 1_000_000_000 / 8)
+
class ClientFactory:
def __init__(self, session):
@@ -68,11 +127,38 @@ def create_transfer_manager(
params, runtime_config
)
if client_type == constants.CRT_TRANSFER_CLIENT:
+ transfer_manager = self._try_create_crt_transfer_manager(
+ params, runtime_config
+ )
+ if transfer_manager is not None:
+ self.warn_unsupported_settings(client_type, runtime_config)
+ return transfer_manager
+ client_type = constants.CLASSIC_TRANSFER_CLIENT
+ self.warn_unsupported_settings(client_type, runtime_config)
+ return self._create_classic_transfer_manager(
+ params, runtime_config, botocore_client
+ )
+
+ def _try_create_crt_transfer_manager(self, params, runtime_config):
+ try:
return self._create_crt_transfer_manager(params, runtime_config)
- else:
- return self._create_classic_transfer_manager(
- params, runtime_config, botocore_client
+ except RuntimeError as e:
+ if str(CRT_PART_SIZE_EXCEEDS_MEMORY_LIMIT) not in str(e):
+ raise
+ if self._is_preferring_crt_client(runtime_config):
+ raise InvalidConfigError(
+ f'The configured multipart_chunksize is too large for the '
+ f"'{constants.CRT_TRANSFER_CLIENT}' s3 transfer client. "
+ f'Lower multipart_chunksize or raise the '
+ f'memory available to the transfer client by setting the '
+ f'AWS_CRT_S3_MEMORY_LIMIT_IN_GIB environment variable.'
+ ) from e
+ LOGGER.debug(
+ f'Not using the crt s3 transfer client because the configured '
+ f'multipart_chunksize does not fit its memory pool: {e}'
)
+ release_crt_s3_process_lock()
+ return None
def _compute_transfer_client_type(self, params, runtime_config):
if params.get('paths_type') == 's3s3':
@@ -81,16 +167,14 @@ def _compute_transfer_client_type(self, params, runtime_config):
'preferred_transfer_client', constants.AUTO_RESOLVE_TRANSFER_CLIENT
)
if preferred_transfer_client == constants.AUTO_RESOLVE_TRANSFER_CLIENT:
- return self._resolve_transfer_client_type_for_system()
+ return self._resolve_transfer_client_type_for_system(
+ params, runtime_config
+ )
return preferred_transfer_client
- def _resolve_transfer_client_type_for_system(self):
+ def _resolve_transfer_client_type_for_system(self, params, runtime_config):
transfer_client_type = constants.CLASSIC_TRANSFER_CLIENT
- is_optimized_for_system = awscrt.s3.is_optimized_for_system()
- LOGGER.debug(
- 'S3 CRT client optimized for system: %s', is_optimized_for_system
- )
- if is_optimized_for_system:
+ if self._is_eligible_for_crt_client(params, runtime_config):
is_running = self._is_crt_client_running_in_other_aws_cli_process()
LOGGER.debug(
'S3 CRT client running in different AWS CLI process: %s',
@@ -103,6 +187,129 @@ def _resolve_transfer_client_type_for_system(self):
)
return transfer_client_type
+ def _is_eligible_for_crt_client(self, params, runtime_config):
+ is_optimized_for_system = awscrt.s3.is_optimized_for_system()
+ LOGGER.debug(
+ f'S3 CRT client optimized for system: {is_optimized_for_system}'
+ )
+ if is_optimized_for_system:
+ return True
+ if not self._is_crt_auto_resolve_enabled():
+ return False
+ unsupported = self._get_unsupported_settings(params, runtime_config)
+ if unsupported:
+ LOGGER.debug(
+ f'Not auto resolving to the crt s3 transfer client because '
+ f'it does not support: {", ".join(unsupported)}'
+ )
+ self._warn_classic_only_settings(runtime_config)
+ return False
+ return True
+
+ def _is_crt_auto_resolve_enabled(self):
+ return (
+ os.environ.get('AWS_CLI_AUTO_RESOLVE_CLIENT')
+ == constants.CRT_TRANSFER_CLIENT
+ )
+
+ def _get_unsupported_settings(self, params, runtime_config):
+ unsupported = self._get_classic_only_settings(runtime_config)
+ if self._is_adaptive_retry_mode():
+ unsupported.append(f'retry_mode = {ADAPTIVE_RETRY_MODE}')
+ if self._is_non_seekable_stream_upload(params):
+ unsupported.append('uploads from a non-seekable stream')
+ if unsupported_attempts := self._get_unsupported_max_attempts(
+ runtime_config
+ ):
+ unsupported.append(unsupported_attempts)
+ return unsupported
+
+ def _get_unsupported_max_attempts(self, runtime_config):
+ max_attempts = self._resolve_max_attempts(runtime_config)
+ if max_attempts is None or (
+ MIN_CRT_MAX_ATTEMPTS <= max_attempts <= MAX_CRT_MAX_ATTEMPTS
+ ):
+ return None
+ return (
+ f'max_attempts = {max_attempts} (must be between '
+ f'{MIN_CRT_MAX_ATTEMPTS} and {MAX_CRT_MAX_ATTEMPTS})'
+ )
+
+ def _resolve_max_attempts(self, runtime_config):
+ config_store = self._session.get_component('config_store')
+ if config_store.is_explicitly_set('max_attempts') or (
+ self._should_use_transfer_config_defaults(runtime_config)
+ ):
+ return self._session.get_config_variable('max_attempts')
+ return None
+
+ def _get_classic_only_settings(self, runtime_config):
+ return self._get_unsupported_options(
+ constants.CRT_TRANSFER_CLIENT,
+ runtime_config,
+ action=EXCLUDE_FROM_AUTO,
+ )
+
+ def _get_unsupported_options(
+ self, client_type, runtime_config, action=None
+ ):
+ return [
+ name
+ for name, option_action in UNSUPPORTED_OPTIONS.get(
+ client_type, {}
+ ).items()
+ if (action is None or option_action == action)
+ and runtime_config.is_explicitly_set(name)
+ ]
+
+ def _is_adaptive_retry_mode(self):
+ return (
+ self._session.get_config_variable('retry_mode')
+ == ADAPTIVE_RETRY_MODE
+ )
+
+ def _warn_classic_only_settings(self, runtime_config):
+ classic_only = self._get_classic_only_settings(runtime_config)
+ if not classic_only:
+ return
+ uni_print(
+ f"warning: Using the '{constants.CLASSIC_TRANSFER_CLIENT}' s3 "
+ f"transfer client because the "
+ f"'{constants.CRT_TRANSFER_CLIENT}' s3 transfer client does not "
+ f"support: {', '.join(classic_only)}. A future version of the AWS "
+ f"CLI will use the '{constants.CRT_TRANSFER_CLIENT}' s3 transfer "
+ f"client by default, at which point these values will be "
+ f"ignored. Set the preferred_transfer_client configuration value "
+ f"to '{constants.CLASSIC_TRANSFER_CLIENT}' to continue using the "
+ f"'{constants.CLASSIC_TRANSFER_CLIENT}' s3 transfer client.\n",
+ sys.stderr,
+ )
+
+ def warn_unsupported_settings(self, client_type, runtime_config):
+ unsupported = self._get_unsupported_options(
+ client_type, runtime_config
+ )
+ if client_type == constants.CRT_TRANSFER_CLIENT:
+ if self._is_adaptive_retry_mode():
+ unsupported.append(f'retry_mode = {ADAPTIVE_RETRY_MODE}')
+ if unsupported_attempts := self._get_unsupported_max_attempts(
+ runtime_config
+ ):
+ unsupported.append(unsupported_attempts)
+ if not unsupported:
+ return
+ uni_print(
+ f"warning: The following configuration values are not supported "
+ f"by the '{client_type}' s3 transfer client and will be ignored: "
+ f"{', '.join(unsupported)}.\n",
+ sys.stderr,
+ )
+
+ def _is_non_seekable_stream_upload(self, params):
+ return bool(
+ params.get('is_stream') and params.get('paths_type') == 'locals3'
+ )
+
def _is_crt_client_running_in_other_aws_cli_process(self):
# If None is returned from acquiring the CRT process lock, it
# means the CRT S3 client is currently being used in a different
@@ -114,29 +321,58 @@ def _acquire_crt_s3_process_lock(self):
def _create_crt_transfer_manager(self, params, runtime_config):
self._acquire_crt_s3_process_lock()
- return CRTTransferManager(
- self._create_crt_client(params, runtime_config),
- self._create_crt_request_serializer(params),
+ region = self._resolve_region(params)
+ bootstrap = create_crt_client_bootstrap()
+ config_kwargs = self._resolve_crt_client_config_kwargs(runtime_config)
+
+ transfer_manager = CRTTransferManager(
+ crt_client_factory=lambda client_region=None: (
+ self._create_crt_client(
+ params,
+ runtime_config,
+ config_kwargs,
+ region=client_region or region,
+ bootstrap=bootstrap,
+ )
+ ),
+ crt_request_serializer=self._create_crt_request_serializer(params),
+ transfer_config=self._create_crt_transfer_config(config_kwargs),
+ )
+ # Clients for redirected regions are created on demand, but create the
+ # one for the configured region now. Otherwise invalid client
+ # configuration is not reported until a transfer is submitted, which
+ # reports it once per object instead of once for the command.
+ transfer_manager.get_crt_client()
+ return transfer_manager
+
+ def _create_crt_transfer_config(self, config_kwargs):
+ # The crt client only applies its multipart threshold to uploads, so
+ # downloads rely on the transfer config to match it. Leaving the
+ # threshold unset keeps the client's own download behavior.
+ return CRTTransferConfig(
+ multipart_threshold=config_kwargs.get('multipart_upload_threshold')
)
- def _create_crt_client(self, params, runtime_config):
- config_file_params = self._session.get_scoped_config().get('s3', {})
+ def _create_crt_client(
+ self,
+ params,
+ runtime_config,
+ config_kwargs,
+ region=None,
+ bootstrap=None,
+ ):
create_crt_client_kwargs = {
- 'region': self._resolve_region(params),
+ 'region': region or self._resolve_region(params),
'verify': self._resolve_verify(params),
+ 'bootstrap': bootstrap,
}
endpoint_url = params.get('endpoint_url')
if endpoint_url and urlparse.urlparse(endpoint_url).scheme == 'http':
create_crt_client_kwargs['use_ssl'] = False
- target_throughput = runtime_config.get('target_bandwidth', None)
+ target_throughput = self._resolve_target_throughput(runtime_config)
if target_throughput:
create_crt_client_kwargs['target_throughput'] = target_throughput
- multipart_chunksize = runtime_config.get('multipart_chunksize', None)
- # User didn't explicitly configure `multipart_chunksize`. Set it to
- # `None` and let CRT dynamically calculate the part size.
- if 'multipart_chunksize' not in config_file_params:
- multipart_chunksize = None
- create_crt_client_kwargs['part_size'] = multipart_chunksize
+ create_crt_client_kwargs.update(config_kwargs)
if params.get('sign_request', True):
crt_credentials_provider = self._get_crt_credentials_provider()
create_crt_client_kwargs['crt_credentials_provider'] = (
@@ -154,6 +390,70 @@ def _create_crt_client(self, params, runtime_config):
return create_s3_crt_client(**create_crt_client_kwargs)
+ def _resolve_crt_client_config_kwargs(self, runtime_config):
+ use_defaults = self._should_use_transfer_config_defaults(
+ runtime_config
+ )
+ kwargs = {}
+ for config_name, crt_name in CRT_CLIENT_KWARG_MAP.items():
+ if runtime_config.is_explicitly_set(config_name):
+ kwargs[crt_name] = runtime_config[config_name]
+ elif use_defaults:
+ kwargs[crt_name] = DEFAULTS[config_name]
+ if 'part_size' not in kwargs:
+ # `create_s3_crt_client` defaults this to 8MB, so `None` has to be
+ # passed to opt into the CRT's dynamic part size calculation.
+ kwargs['part_size'] = None
+ max_attempts = self._resolve_max_attempts(runtime_config)
+ if max_attempts is not None and (
+ MIN_CRT_MAX_ATTEMPTS <= max_attempts <= MAX_CRT_MAX_ATTEMPTS
+ ):
+ kwargs['retry_options'] = {'max_retries': max_attempts - 1}
+ return kwargs
+
+ def _resolve_target_throughput(self, runtime_config):
+ target_throughput = runtime_config.get('target_bandwidth')
+ if target_throughput is not None:
+ return target_throughput
+ if self._is_preferring_crt_client(runtime_config):
+ # Users who opted into the crt transfer client keep the throughput
+ # they get today, even on hosts the crt recommends less for.
+ recommended = awscrt.s3.get_recommended_throughput_target_gbps()
+ return _gbps_to_bytes_per_sec(
+ max(recommended or 0, MINIMUM_TARGET_THROUGHPUT_GBPS)
+ )
+ if self._is_newly_eligible_for_crt_client(runtime_config) and (
+ self._is_untuned_system()
+ ):
+ # The crt client sizes its memory pool from the throughput target.
+ # Without a recommendation it assumes 10gbps, which maps to a max
+ # pool size of 2GiB. Newly-eligible hosts that auto-resolve to crt
+ # may not be able to afford 2GiB, so it sets the maximum throughput
+ # that maps to the smallest 256MiB tier.
+ return _gbps_to_bytes_per_sec(UNTUNED_TARGET_THROUGHPUT_GBPS)
+ return None
+
+ def _is_untuned_system(self):
+ # The crt client has no throughput recommendation for systems it has
+ # not been tuned for.
+ return awscrt.s3.get_recommended_throughput_target_gbps() is None
+
+ def _is_preferring_crt_client(self, runtime_config):
+ return (
+ runtime_config.get('preferred_transfer_client')
+ == constants.CRT_TRANSFER_CLIENT
+ )
+
+ def _is_newly_eligible_for_crt_client(self, runtime_config):
+ if self._is_preferring_crt_client(runtime_config):
+ return False
+ return not awscrt.s3.is_optimized_for_system()
+
+ def _should_use_transfer_config_defaults(self, runtime_config):
+ # Configurations that already resolve to the crt transfer client keep
+ # its defaults so their behavior is unchanged.
+ return self._is_newly_eligible_for_crt_client(runtime_config)
+
def _create_crt_request_serializer(self, params):
return BotocoreCRTRequestSerializer(
self._session,
@@ -161,6 +461,9 @@ def _create_crt_request_serializer(self, params):
'region_name': self._resolve_region(params),
'endpoint_url': params.get('endpoint_url'),
},
+ region_redirect_client_factory=lambda: (
+ self._botocore_client_factory.create_client(params)
+ ),
)
def _create_classic_transfer_manager(
diff --git a/awscli/customizations/s3/transferconfig.py b/awscli/customizations/s3/transferconfig.py
index 5502ea93b0cc..a4edfdde07ba 100644
--- a/awscli/customizations/s3/transferconfig.py
+++ b/awscli/customizations/s3/transferconfig.py
@@ -42,6 +42,29 @@ class InvalidConfigError(Exception):
pass
+class ResolvedRuntimeConfig(dict):
+ """A runtime config that tracks which values the user supplied.
+
+ A value matching its default says nothing about whether the user
+ configured it, so callers needing that distinction cannot infer it from
+ the resolved value alone.
+ """
+
+ def __init__(self, values, explicit_keys):
+ super().__init__(values)
+ self._explicit_keys = frozenset(explicit_keys)
+
+ @property
+ def explicit_keys(self):
+ return self._explicit_keys
+
+ def is_explicitly_set(self, name):
+ return name in self._explicit_keys
+
+ def copy(self):
+ return ResolvedRuntimeConfig(dict(self), self._explicit_keys)
+
+
class RuntimeConfig:
POSITIVE_INTEGERS = [
'multipart_chunksize',
@@ -89,9 +112,11 @@ def build_config(self, **kwargs):
that use this runtime config.
:param kwargs: Any key in the ``DEFAULTS`` dict.
- :return: A dictionary of the merged and converted values.
+ :return: A ``ResolvedRuntimeConfig`` of the merged and converted
+ values, which also tracks which keys were explicitly provided.
"""
+ explicit_keys = set(kwargs)
runtime_config = DEFAULTS.copy()
if kwargs:
runtime_config.update(kwargs)
@@ -100,7 +125,7 @@ def build_config(self, **kwargs):
self._convert_booleans(runtime_config)
self._resolve_choice_aliases(runtime_config)
self._validate_config(runtime_config)
- return runtime_config
+ return ResolvedRuntimeConfig(runtime_config, explicit_keys)
def _convert_human_readable_sizes(self, runtime_config):
for attr in self.HUMAN_READABLE_SIZES:
diff --git a/awscli/s3transfer/crt.py b/awscli/s3transfer/crt.py
index efd91b490217..b441d4dbccf5 100644
--- a/awscli/s3transfer/crt.py
+++ b/awscli/s3transfer/crt.py
@@ -10,9 +10,11 @@
# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
# ANY KIND, either express or implied. See the License for the specific
# language governing permissions and limitations under the License.
+import contextlib
import logging
import re
import threading
+from concurrent.futures import CancelledError, Future
from io import BytesIO
import awscrt.http
@@ -37,6 +39,7 @@
S3RequestTlsMode,
S3RequestType,
S3ResponseError,
+ S3RetryConfig,
get_recommended_throughput_target_gbps,
)
from botocore import UNSIGNED
@@ -44,7 +47,13 @@
from botocore.config import Config
from botocore.exceptions import InvalidConfigError, NoCredentialsError
from botocore.useragent import register_feature_id
-from botocore.utils import ArnParser, InvalidArnException, is_s3express_bucket
+from botocore.utils import (
+ ArnParser,
+ InvalidArnException,
+ S3RegionRedirectorv2,
+ is_s3express_bucket,
+)
+from s3transfer.compat import seekable
from s3transfer.constants import FULL_OBJECT_CHECKSUM_ARGS, MB
from s3transfer.exceptions import TransferNotDoneError
from s3transfer.futures import BaseTransferFuture, BaseTransferMeta
@@ -87,6 +96,17 @@ def acquire_crt_s3_process_lock(name):
return CRT_S3_PROCESS_LOCK
+def release_crt_s3_process_lock():
+ # Acquiring the lock signals to other processes that this one is using the
+ # CRT S3 client, so a process that acquired it and then did not use the
+ # client has to release it. Otherwise it denies the client to every other
+ # process of the same application for the rest of its lifetime.
+ global CRT_S3_PROCESS_LOCK
+ if CRT_S3_PROCESS_LOCK is not None:
+ CRT_S3_PROCESS_LOCK.release()
+ CRT_S3_PROCESS_LOCK = None
+
+
def create_s3_crt_client(
region,
crt_credentials_provider=None,
@@ -96,6 +116,10 @@ def create_s3_crt_client(
use_ssl=True,
verify=None,
fio_options=None,
+ bootstrap=None,
+ multipart_upload_threshold=None,
+ max_active_connections_override=None,
+ retry_options=None,
):
"""
:type region: str
@@ -142,11 +166,29 @@ def create_s3_crt_client(
:type fio_options: Optional[dict]
:param fio_options: Kwargs to use to build an `awscrt.s3.S3FileIoOptions`.
+
+ :type bootstrap: Optional[awscrt.io.ClientBootstrap]
+ :param bootstrap: Shared I/O bootstrap to use for the client. If not
+ provided, a new bootstrap is created.
+
+ :type multipart_upload_threshold: Optional[int]
+ :param multipart_upload_threshold: Size, in bytes, above which uploads use
+ a multipart upload rather than a single request. Only affects uploads.
+ If not set, the maximum of ``part_size`` and 5 MiB is used.
+
+ :type max_active_connections_override: Optional[int]
+ :param max_active_connections_override: Caps the number of active
+ connections. Only applies when lower than the connection count derived
+ from ``target_throughput``. If not set, the derived value is used.
+
+ :type retry_options: Optional[dict]
+ :param retry_options: Kwargs to use to build an
+ `awscrt.s3.S3RetryConfig`. If not set, the CRT's own retry
+ configuration is used.
"""
- event_loop_group = EventLoopGroup(num_threads)
- host_resolver = DefaultHostResolver(event_loop_group)
- bootstrap = ClientBootstrap(event_loop_group, host_resolver)
+ if bootstrap is None:
+ bootstrap = create_crt_client_bootstrap(num_threads)
tls_connection_options = None
tls_mode = (
@@ -177,6 +219,9 @@ def create_s3_crt_client(
crt_fio_options = None
if fio_options:
crt_fio_options = S3FileIoOptions(**fio_options)
+ crt_retry_config = None
+ if retry_options:
+ crt_retry_config = S3RetryConfig(**retry_options)
return S3Client(
bootstrap=bootstrap,
region=region,
@@ -187,9 +232,18 @@ def create_s3_crt_client(
throughput_target_gbps=target_gbps,
enable_s3express=True,
fio_options=crt_fio_options,
+ multipart_upload_threshold=multipart_upload_threshold,
+ max_active_connections_override=max_active_connections_override,
+ retry_config=crt_retry_config,
)
+def create_crt_client_bootstrap(num_threads=None):
+ event_loop_group = EventLoopGroup(num_threads)
+ host_resolver = DefaultHostResolver(event_loop_group)
+ return ClientBootstrap(event_loop_group, host_resolver)
+
+
def _get_crt_throughput_target_gbps(provided_throughput_target_bytes=None):
if provided_throughput_target_bytes is None:
target_gbps = get_recommended_throughput_target_gbps()
@@ -208,31 +262,227 @@ def _get_crt_throughput_target_gbps(provided_throughput_target_bytes=None):
return target_gbps
+class CRTS3RegionRedirectPolicy:
+ """Decides which region a CRT transfer to a bucket should use.
+
+ Region discovery is delegated to the request serializer, which reuses
+ botocore's ``S3RegionRedirectorv2`` classification. This class owns only
+ the conditions under which a redirect may be attempted at all.
+ """
+
+ def __init__(self, crt_request_serializer):
+ self._crt_request_serializer = crt_request_serializer
+ # Held while a failed transfer looks up its bucket's region, so that a
+ # burst of transfers failing at once shares a single lookup instead of
+ # each paying for its own, potentially a HeadBucket request each.
+ # Request construction uses the serializer's separate cache lock, so a
+ # slow lookup does not block other transfers.
+ self._redirect_lock = threading.Lock()
+
+ def is_error_redirect_candidate(
+ self,
+ bucket,
+ transfer_type,
+ error,
+ is_region_redirect,
+ bytes_transferred,
+ cancelled,
+ is_replayable,
+ ):
+ """Return whether a failed request may be worth redirecting.
+
+ These checks never make a request, so a caller running on a CRT
+ completion thread can use them to decide whether looking up a region
+ is worth handing off to another thread. Finding the region is what may
+ need a ``HeadBucket``, and that is left to ``get_retry_region``.
+ """
+ if is_region_redirect:
+ logger.debug(
+ 'Transfer for bucket %s was already redirected, not '
+ 'redirecting again.',
+ bucket,
+ )
+ return False
+ if cancelled:
+ return False
+ if not is_replayable:
+ logger.debug(
+ 'Not redirecting transfer for bucket %s because its stream '
+ 'cannot be replayed.',
+ bucket,
+ )
+ return False
+ if bytes_transferred:
+ # Replaying a request that moved data would either duplicate
+ # bytes or double-count progress.
+ logger.debug(
+ 'Not redirecting transfer for bucket %s because it already '
+ 'transferred %s bytes.',
+ bucket,
+ bytes_transferred,
+ )
+ return False
+ if is_s3express_bucket(bucket):
+ return False
+ # Parsing the error response is the most expensive check here, so it
+ # runs only once the transfer is otherwise worth redirecting.
+ return self._crt_request_serializer.is_redirect_error(
+ transfer_type, error
+ )
+
+ def get_retry_region(
+ self, bucket, transfer_type, error, request_region=None
+ ):
+ """Return the region to retry a failed request in, or ``None``.
+
+ A returned region has been cached, so both the retried request and
+ later transfers to the same bucket use it. This discovers the region
+ of a bucket, which may require an additional ``HeadBucket`` request,
+ so it must not be called from a CRT completion thread.
+
+ :type request_region: Optional[str]
+ :param request_region: The region the failed request was made in, or
+ ``None`` if it used the configured region.
+ """
+ retry_region = self._get_cached_retry_region(bucket, request_region)
+ if retry_region is not None:
+ return retry_region
+ with self._redirect_lock:
+ # Another transfer may have discovered the region while this one
+ # waited for the lock.
+ retry_region = self._get_cached_retry_region(
+ bucket, request_region
+ )
+ if retry_region is not None:
+ return retry_region
+ return self._discover_bucket_region(
+ bucket, transfer_type, error, request_region
+ )
+
+ def _get_cached_retry_region(self, bucket, request_region):
+ """Return an already discovered region the failed request did not use.
+
+ A region another transfer discovered is worth retrying in, but the one
+ the request just failed in is not.
+ """
+ cached_region = self._crt_request_serializer.get_cached_bucket_region(
+ bucket
+ )
+ if cached_region is None:
+ return None
+ if cached_region == request_region:
+ # The failed request was already made in this region, so retrying
+ # there would fail the same way. A bucket does not move between
+ # regions, so the redirect points somewhere else or nowhere.
+ return None
+ return cached_region
+
+ def _discover_bucket_region(
+ self, bucket, transfer_type, error, request_region=None
+ ):
+ """Look up and cache a bucket's region.
+
+ Callers must hold ``_redirect_lock``. Every write to the region cache
+ happens here, so holding it is what lets ``get_retry_region`` treat its
+ second cache read as final rather than racing another lookup.
+ """
+ try:
+ new_region = self._crt_request_serializer.get_bucket_region(
+ bucket, transfer_type, error
+ )
+ except Exception as redirect_error:
+ logger.debug(
+ 'Unable to determine S3 redirect region.',
+ exc_info=redirect_error,
+ )
+ return None
+ if new_region is None:
+ return None
+ if new_region == (
+ request_region
+ or self._crt_request_serializer.get_configured_region()
+ ):
+ # The failed request was already made in this region, so retrying
+ # it there would fail the same way. Leaving it out of the cache
+ # also keeps later transfers on the client they already use.
+ logger.debug(
+ 'Not redirecting transfer for bucket %s because it was '
+ 'already made in region %s.',
+ bucket,
+ new_region,
+ )
+ return None
+ logger.debug(
+ 'Redirecting CRT S3 transfer for bucket %s to region %s',
+ bucket,
+ new_region,
+ )
+ self._crt_request_serializer.cache_bucket_region(bucket, new_region)
+ return new_region
+
+
+class CRTTransferConfig:
+ def __init__(self, multipart_threshold=None):
+ """Configuration the CRT transfer manager applies itself
+
+ This only covers configuration that the CRT s3 client cannot apply
+ on its own. Everything else is configured on the client, so it
+ deliberately has no equivalent here.
+
+ :type multipart_threshold: Optional[int]
+ :param multipart_threshold: The size, in bytes, that a download must
+ exceed to be split into ranged requests. The CRT s3 client only
+ applies its own threshold to uploads. If not set, the client
+ decides how to split every download.
+ """
+ self.multipart_threshold = multipart_threshold
+
+
class CRTTransferManager:
- def __init__(self, crt_s3_client, crt_request_serializer, osutil=None):
+ def __init__(
+ self,
+ crt_client_factory,
+ crt_request_serializer,
+ osutil=None,
+ transfer_config=None,
+ ):
"""A transfer manager interface for Amazon S3 on CRT s3 client.
- :type crt_s3_client: awscrt.s3.S3Client
- :param crt_s3_client: The CRT s3 client, handling all the
- HTTP requests and functions under then hood
+ :type crt_client_factory:
+ Callable[[Optional[str]], awscrt.s3.S3Client]
+ :param crt_client_factory: Creates a CRT client. ``None`` selects the
+ configured region; a region string selects a redirected region.
:type crt_request_serializer: s3transfer.crt.BaseCRTRequestSerializer
- :param crt_request_serializer: Serializer, generates unsigned crt HTTP
- request.
+ :param crt_request_serializer: Serializer, generates unsigned CRT HTTP
+ requests.
:type osutil: s3transfer.utils.OSUtils
:param osutil: OSUtils object to use for os-related behavior when
using with transfer manager.
+
+ :type transfer_config: s3transfer.crt.CRTTransferConfig
+ :param transfer_config: The transfer configuration to apply. If not
+ provided, the CRT s3 client's own configuration applies to every
+ transfer.
"""
if osutil is None:
self._osutil = OSUtils()
- self._crt_s3_client = crt_s3_client
self._s3_args_creator = S3ClientArgsCreator(
- crt_request_serializer, self._osutil
+ crt_request_serializer,
+ self._osutil,
+ transfer_config=transfer_config,
)
+ self._crt_request_serializer = crt_request_serializer
self._crt_exception_translator = (
crt_request_serializer.translate_crt_exception
)
+ self._crt_client_factory = crt_client_factory
+ self._crt_clients = {}
+ self._crt_client_lock = threading.Lock()
+ self._region_redirect_policy = CRTS3RegionRedirectPolicy(
+ crt_request_serializer
+ )
self._future_coordinators = []
self._semaphore = threading.Semaphore(128) # not configurable
# A counter to create unique id's for each transfer submitted.
@@ -334,12 +584,25 @@ def _shutdown(self, cancel=False):
def _release_semaphore(self, **kwargs):
self._semaphore.release()
+ def get_crt_client(self, region=None):
+ with self._crt_client_lock:
+ crt_client = self._crt_clients.get(region)
+ if crt_client is None:
+ logger.debug(
+ 'Creating CRT S3 client for region %s',
+ region if region is not None else 'default',
+ )
+ crt_client = self._crt_client_factory(region)
+ self._crt_clients[region] = crt_client
+ return crt_client
+
def _submit_transfer(self, request_type, call_args):
register_feature_id('S3_TRANSFER')
on_done_after_calls = [self._release_semaphore]
coordinator = CRTTransferCoordinator(
transfer_id=self._id_counter,
exception_translator=self._crt_exception_translator,
+ completion_future=Future(),
)
components = {
'meta': CRTTransferMeta(self._id_counter, call_args),
@@ -348,6 +611,22 @@ def _submit_transfer(self, request_type, call_args):
future = CRTTransferFuture(**components)
afterdone = AfterDoneHandler(coordinator)
on_done_after_calls.append(afterdone)
+ # Serialization can rewrite an ARN in call_args, so retain the
+ # caller-provided bucket for redirect eligibility and caching.
+ bucket = call_args.bucket
+ # Record the current stream position and if its replayable,
+ # in the event of a region redirect, we need to reset first
+ is_replayable = True
+ upload_stream_position = None
+ if request_type == 'put_object' and not isinstance(
+ call_args.fileobj, str
+ ):
+ try:
+ is_replayable = seekable(call_args.fileobj)
+ if is_replayable:
+ upload_stream_position = call_args.fileobj.tell()
+ except (AttributeError, OSError, ValueError):
+ is_replayable = False
try:
self._semaphore.acquire()
@@ -355,22 +634,44 @@ def _submit_transfer(self, request_type, call_args):
future, 'queued'
)
on_queued()
- crt_callargs = self._s3_args_creator.get_make_request_args(
+
+ def create_request(is_region_redirect):
+ # Reset the stream if we're redirecting due to bucket region
+ if is_region_redirect and upload_stream_position is not None:
+ call_args.fileobj.seek(upload_stream_position)
+ with self._crt_request_serializer.locked_bucket_region(
+ bucket
+ ) as region:
+ if region is not None:
+ logger.debug(
+ 'Using cached region %s for S3 bucket %s',
+ region,
+ bucket,
+ )
+ crt_callargs = self._s3_args_creator.get_make_request_args(
+ request_type,
+ call_args,
+ coordinator,
+ future,
+ on_done_after_calls,
+ )
+ crt_client = self.get_crt_client(region)
+ return crt_client, crt_callargs, region
+
+ coordinator.submit(
+ create_request,
+ self._region_redirect_policy,
+ bucket,
request_type,
- call_args,
- coordinator,
- future,
- on_done_after_calls,
+ is_replayable=is_replayable,
)
- crt_s3_request = self._crt_s3_client.make_request(**crt_callargs)
except Exception as e:
coordinator.set_exception(e, True)
on_done = self._s3_args_creator.get_crt_callback(
future, 'done', after_subscribers=on_done_after_calls
)
+ coordinator.complete(e)
on_done(error=e)
- else:
- coordinator.set_s3_request(crt_s3_request)
self._future_coordinators.append(coordinator)
self._id_counter += 1
@@ -384,6 +685,7 @@ def __init__(self, transfer_id=None, call_args=None):
self._transfer_id = transfer_id
self._call_args = call_args
self._user_context = {}
+ self._size = None
@property
def call_args(self):
@@ -397,6 +699,13 @@ def transfer_id(self):
def user_context(self):
return self._user_context
+ @property
+ def size(self):
+ return self._size
+
+ def provide_transfer_size(self, size):
+ self._size = size
+
class CRTTransferFuture(BaseTransferFuture):
def __init__(self, meta=None, coordinator=None):
@@ -470,9 +779,64 @@ def serialize_http_request(self, transfer_type, future):
def translate_crt_exception(self, exception):
raise NotImplementedError('translate_crt_exception()')
+ def get_cached_bucket_region(self, bucket):
+ """Return the region already discovered for a bucket, if any.
+
+ Serializers that do not support bucket region redirects never have a
+ region to report, which keeps their transfers in the configured
+ region.
+ """
+ return None
+
+ def cache_bucket_region(self, bucket, region):
+ """Remember the region a bucket was found in."""
+
+ @contextlib.contextmanager
+ def locked_bucket_region(self, bucket):
+ """Hold a bucket's cached region steady while a request is built."""
+ yield self.get_cached_bucket_region(bucket)
+
+ def is_redirect_error(self, transfer_type, error):
+ """Return whether a failed request reports a region redirect."""
+ return False
+
+ def get_bucket_region(self, bucket, transfer_type, error):
+ """Return the region a failed request should be retried in, if any."""
+ return None
+
+ def get_configured_region(self):
+ """Return the region requests are serialized for by default."""
+ return None
+
+
+class LazyHeadBucketClient:
+ """
+ A boto client that is lazily initialized when needing to call head_bucket.
+ The existing region redirect logic supports falling back to this when the
+ region cannot be parsed from the S3 exception. We use this boto client even
+ when CRT is being used for transfers
+ """
+
+ def __init__(self, client_factory):
+ self._client_factory = client_factory
+ self._client = None
+ self._lock = threading.Lock()
+
+ def head_bucket(self, **kwargs):
+ with self._lock:
+ if self._client is None:
+ self._client = self._client_factory()
+ client = self._client
+ return client.head_bucket(**kwargs)
+
class BotocoreCRTRequestSerializer(BaseCRTRequestSerializer):
- def __init__(self, session, client_kwargs=None):
+ def __init__(
+ self,
+ session,
+ client_kwargs=None,
+ region_redirect_client_factory=None,
+ ):
"""Serialize CRT HTTP request using botocore logic
It also takes into account configuration from both the session
and any keyword arguments that could be passed to
@@ -483,12 +847,45 @@ def __init__(self, session, client_kwargs=None):
:type client_kwargs: Optional[Dict[str, str]])
:param client_kwargs: The kwargs for the botocore
s3 client initialization.
+
+ :type region_redirect_client_factory: Optional[Callable]
+ :param region_redirect_client_factory: Creates a real botocore S3
+ client if a CRT redirect response requires a HeadBucket fallback.
"""
self._session = session
if client_kwargs is None:
client_kwargs = {}
+ client_kwargs = client_kwargs.copy()
+
+ # Preserve the caller's client configuration for the real botocore
+ # client used only by the HeadBucket fallback. The serializer client
+ # below is reconfigured as unsigned and never sends its request.
+ if region_redirect_client_factory is None:
+ redirect_client_kwargs = client_kwargs.copy()
+ redirect_client_kwargs['service_name'] = 's3'
+
+ def region_redirect_client_factory():
+ return session.create_client(**redirect_client_kwargs)
+
+ # Build the botocore client that converts API arguments into the
+ # unsigned HTTP requests consumed by CRT.
self._resolve_client_config(session, client_kwargs)
self._client = session.create_client(**client_kwargs)
+
+ # Store discovered bucket regions beside the lock that serializes
+ # updates against the requests being built from them. The real
+ # fallback client remains lazy.
+ self._region_redirect_client = LazyHeadBucketClient(
+ region_redirect_client_factory
+ )
+ self._region_cache = {}
+ self._region_cache_lock = threading.Lock()
+ self._region_redirector = S3RegionRedirectorv2(
+ None, self._region_redirect_client, cache=self._region_cache
+ )
+
+ # Convert normal botocore calls into serialized requests without
+ # allowing the serializer client to perform network I/O.
self._client.meta.events.register(
'request-created.s3.*', self._capture_http_request
)
@@ -501,6 +898,10 @@ def __init__(self, session, client_kwargs=None):
self._client.meta.events.register(
'before-call.s3.*', self._remove_checksum_context
)
+ self._client.meta.events.register(
+ 'before-endpoint-resolution.s3',
+ self._region_redirector.redirect_from_cache,
+ )
@property
def client_config(self):
@@ -598,6 +999,9 @@ def _get_botocore_http_request(self, client_method, call_args):
)['HTTPRequest']
def serialize_http_request(self, transfer_type, future):
+ # Resolving the endpoint reads the bucket region cache, so callers
+ # have to run this inside locked_bucket_region() to be able to send
+ # the request on a client for the region it was built for.
botocore_http_request = self._get_botocore_http_request(
transfer_type, future.meta.call_args
)
@@ -643,6 +1047,80 @@ def _remove_checksum_context(self, params, **kwargs):
if "checksum" in request_context:
del request_context["checksum"]
+ def cache_bucket_region(self, bucket, region):
+ with self._region_cache_lock:
+ self._region_cache[bucket] = region
+
+ def get_cached_bucket_region(self, bucket):
+ with self.locked_bucket_region(bucket) as region:
+ return region
+
+ def get_configured_region(self):
+ return self._client.meta.region_name
+
+ @contextlib.contextmanager
+ def locked_bucket_region(self, bucket):
+ """Hold a bucket's cached region steady while a request is built.
+
+ Endpoint resolution reads the cache again while a request is being
+ serialized, so the caller has to hold this open until the request is
+ built for the yielded region to be the one it was built for.
+ """
+ with self._region_cache_lock:
+ yield self._region_cache.get(bucket)
+
+ def is_redirect_error(self, transfer_type, error):
+ """Return whether a CRT response error reports a region redirect.
+
+ This parses the error response but never makes a request, so unlike
+ ``get_bucket_region`` it is safe to call from a CRT completion thread.
+ """
+ redirect_args = self._get_redirect_args(transfer_type, error)
+ if redirect_args is None:
+ return False
+ return self._region_redirector.is_redirect_response(*redirect_args)
+
+ def get_bucket_region(self, bucket, transfer_type, error):
+ """Extract a redirect region from a CRT response error.
+ This adapts the CRT error for S3RegionRedirectorv2.
+ """
+ redirect_args = self._get_redirect_args(transfer_type, error)
+ if redirect_args is None:
+ return None
+ # The redirector checks the CRT response first. It only uses this
+ # serializer's real botocore client for HeadBucket when the response
+ # identifies a redirect but omits the target region.
+ return self._region_redirector.get_redirect_region(
+ bucket, *redirect_args
+ )
+
+ def _get_redirect_args(self, transfer_type, error):
+ """Adapt a CRT response error for S3RegionRedirectorv2.
+
+ Returns the ``(response, operation)`` pair the redirector expects, or
+ ``None`` if the error is not one it can be asked about.
+ """
+ if not isinstance(error, S3ResponseError):
+ return None
+ translated_error = self._translate_crt_s3_response_error(error)
+ if translated_error is None:
+ return None
+
+ operation_name = (
+ translated_error.operation_name
+ or self._client.meta.method_to_api_mapping[transfer_type]
+ )
+ operation = self._client.meta.service_model.operation_model(
+ operation_name
+ )
+ http_response = botocore.awsrequest.AWSResponse(
+ None,
+ error.status_code,
+ dict(error.headers or []),
+ FakeRawResponse(error.body or b''),
+ )
+ return (http_response, translated_error.response), operation
+
class FakeRawResponse(BytesIO):
def stream(self, amt=1024, decode_content=None):
@@ -673,18 +1151,36 @@ def _get_credentials(self):
class CRTTransferCoordinator:
- """A helper class for managing CRTTransferFuture"""
+ """
+ Coordinates one logical transfer across its native CRT request(s), which
+ can make two if following a bucket region redirect
+ """
def __init__(
- self, transfer_id=None, s3_request=None, exception_translator=None
+ self,
+ transfer_id=None,
+ s3_request=None,
+ exception_translator=None,
+ completion_future=None,
):
self.transfer_id = transfer_id
self._exception_translator = exception_translator
self._s3_request = s3_request
self._lock = threading.Lock()
self._exception = None
- self._crt_future = None
+ # This future represents the entire transfer,
+ # which could include a retry for a region redirect.
+ self._completion_future = completion_future or Future()
+ self._completion_started = False
self._done_event = threading.Event()
+ self._cancelled = False
+ self._redirect_retry_started = False
+ # Set by submit(), and the same for every request the transfer makes.
+ self._request_factory = None
+ self._region_redirect_policy = None
+ self._bucket = None
+ self._transfer_type = None
+ self._is_replayable = True
@property
def s3_request(self):
@@ -702,17 +1198,25 @@ def set_exception(self, exception, override=False):
self._exception = exception
def cancel(self):
- if self._s3_request:
- self._s3_request.cancel()
+ with self._lock:
+ self._cancelled = True
+ s3_request = self._s3_request
+ if s3_request:
+ s3_request.cancel()
+
+ @property
+ def cancelled(self):
+ with self._lock:
+ return self._cancelled
def result(self, timeout=None):
if self._exception:
raise self._exception
try:
- self._crt_future.result(timeout)
+ self._completion_future.result(timeout)
except KeyboardInterrupt:
self.cancel()
- self._crt_future.result(timeout)
+ self._completion_future.result(timeout)
raise
except Exception as e:
self.handle_exception(e)
@@ -736,19 +1240,183 @@ def handle_exception(self, exc):
raise exc
def done(self):
- if self._crt_future is None:
+ return self._completion_future.done()
+
+ def submit(
+ self,
+ request_factory,
+ region_redirect_policy,
+ bucket,
+ transfer_type,
+ is_replayable=True,
+ ):
+ """Submit the transfer's CRT request.
+
+ A request that failed because it was made in the wrong region for its
+ bucket is resubmitted in the bucket's region, which makes a second
+ request for the same transfer.
+ """
+ self._request_factory = request_factory
+ self._region_redirect_policy = region_redirect_policy
+ self._bucket = bucket
+ self._transfer_type = transfer_type
+ self._is_replayable = is_replayable
+ self._start_request(is_region_redirect=False)
+
+ def _start_request(self, is_region_redirect):
+ with self._lock:
+ if self._cancelled:
+ raise CancelledError()
+ if is_region_redirect:
+ self._redirect_retry_started = True
+ crt_client, crt_callargs, request_region = self._request_factory(
+ is_region_redirect
+ )
+ on_done = crt_callargs['on_done']
+ on_progress = crt_callargs['on_progress']
+ bytes_transferred = 0
+
+ def track_progress(transferred):
+ nonlocal bytes_transferred
+ bytes_transferred += transferred
+ on_progress(transferred)
+
+ def finish(error, kwargs):
+ self.complete(error)
+ on_done(error=error, **kwargs)
+
+ def redirect_and_finish(error, kwargs):
+ # Any failure deciding on or starting a redirect must still
+ # complete the transfer. Otherwise the transfer is never marked
+ # done and anything waiting on its result blocks forever.
+ try:
+ new_region = self._region_redirect_policy.get_retry_region(
+ self._bucket,
+ self._transfer_type,
+ error,
+ request_region,
+ )
+ if new_region is not None:
+ try:
+ self._start_request(is_region_redirect=True)
+ return
+ except Exception as retry_error:
+ retry_error.__cause__ = error
+ error = retry_error
+ self.set_exception(retry_error, True)
+ except Exception as redirect_error:
+ logger.debug(
+ 'Unable to determine whether to redirect transfer for '
+ 'bucket %s.',
+ self._bucket,
+ exc_info=redirect_error,
+ )
+ if error is None:
+ error = redirect_error
+ self.set_exception(redirect_error, True)
+ finish(error, kwargs)
+
+ def request_done(error=None, **kwargs):
+ if error is not None and self._can_redirect(
+ error, is_region_redirect, bytes_transferred
+ ):
+ # Discovering a region and serializing the retry can both
+ # block, and this runs on a CRT completion thread, where
+ # blocking stalls every other transfer sharing the event loop.
+ self._dispatch_redirect(redirect_and_finish, error, kwargs)
+ return
+ # Nothing to discover, so finish on this thread rather than paying
+ # for a handoff on every completed transfer.
+ finish(error, kwargs)
+
+ crt_callargs['on_done'] = request_done
+ crt_callargs['on_progress'] = track_progress
+ s3_request = crt_client.make_request(**crt_callargs)
+ self.set_s3_request(s3_request, is_region_redirect=is_region_redirect)
+
+ def _can_redirect(self, error, is_region_redirect, bytes_transferred):
+ try:
+ return self._region_redirect_policy.is_error_redirect_candidate(
+ bucket=self._bucket,
+ transfer_type=self._transfer_type,
+ error=error,
+ is_region_redirect=is_region_redirect,
+ bytes_transferred=bytes_transferred,
+ cancelled=self.cancelled,
+ is_replayable=self._is_replayable,
+ )
+ except Exception as redirect_error:
+ logger.debug(
+ 'Unable to determine whether transfer for bucket %s can be '
+ 'redirected.',
+ self._bucket,
+ exc_info=redirect_error,
+ )
return False
- return self._crt_future.done()
- def set_s3_request(self, s3_request):
- self._s3_request = s3_request
- self._crt_future = self._s3_request.finished_future
+ def _dispatch_redirect(self, fn, *args):
+ """Run a region redirect off of the CRT completion thread.
+
+ Discovering a region and serializing the retry can both block, which
+ would stall the event loop shared by every in-flight transfer. A
+ transfer is redirected at most once and only when it fails, so these
+ threads are few and short lived.
+ """
+ try:
+ threading.Thread(
+ target=fn, args=args, name='crt-s3-region-redirect'
+ ).start()
+ except RuntimeError as thread_error:
+ # The OS refused a new thread. Blocking this thread is still
+ # better than stranding the transfer.
+ logger.debug(
+ 'Unable to hand off S3 region redirect, handling it inline.',
+ exc_info=thread_error,
+ )
+ fn(*args)
+
+ def set_s3_request(self, s3_request, is_region_redirect=False):
+ """Make a CRT request the one the transfer acts on."""
+ with self._lock:
+ if not is_region_redirect and self._redirect_retry_started:
+ # The redirect is already active. The original request
+ # completed and redirected before make_request() returned.
+ return
+ if is_region_redirect:
+ self._redirect_retry_started = True
+ self._s3_request = s3_request
+ cancelled = self._cancelled
+ if cancelled:
+ s3_request.cancel()
+
+ def complete(self, error=None):
+ """Complete the logical transfer after the original or retry request.
+
+ This is separate from the native CRT request completion callbacks.
+ """
+ with self._lock:
+ if self._completion_started or self._completion_future.done():
+ return
+ self._completion_started = True
+ completion_future = self._completion_future
+ # No further request will be started for this transfer, so release
+ # the factory. It holds the transfer's call args, including an
+ # upload's stream, and the manager keeps every coordinator until
+ # it shuts down.
+ self._request_factory = None
+ if error is None:
+ completion_future.set_result(None)
+ else:
+ completion_future.set_exception(error)
class S3ClientArgsCreator:
- def __init__(self, crt_request_serializer, os_utils):
+ _DOWNLOAD_TEMP_FILENAME = '_crt_download_temp_filename'
+
+ def __init__(self, crt_request_serializer, os_utils, transfer_config=None):
self._request_serializer = crt_request_serializer
self._os_utils = os_utils
+ self._transfer_config = transfer_config
self._client_config = crt_request_serializer.client_config
self._service_model = crt_request_serializer.service_model
@@ -812,14 +1480,22 @@ def _get_make_request_args_put_object(
call_args.extra_args["Body"] = call_args.fileobj
checksum_config = None
+ provided_checksum_algorithm = None
if self._should_calculate_upload_checksum(
request_type, call_args.extra_args
):
- checksum_algorithm = call_args.extra_args.pop(
- 'ChecksumAlgorithm', 'CRC64NVME'
+ # CRT applies this checksum itself, so we hide it from botocore
+ # while serializing but store it for a possible redirected attempt
+ provided_checksum_algorithm = call_args.extra_args.pop(
+ 'ChecksumAlgorithm', None
+ )
+ applied_checksum_algorithm = (
+ provided_checksum_algorithm or 'CRC64NVME'
).upper()
checksum_config = awscrt.s3.S3ChecksumConfig(
- algorithm=awscrt.s3.S3ChecksumAlgorithm[checksum_algorithm],
+ algorithm=awscrt.s3.S3ChecksumAlgorithm[
+ applied_checksum_algorithm
+ ],
location=awscrt.s3.S3ChecksumLocation.TRAILER,
)
# Suppress botocore's automatic MD5 calculation by setting an override
@@ -828,14 +1504,20 @@ def _get_make_request_args_put_object(
# automatically add trailing checksums to its uploads.
call_args.extra_args["ContentMD5"] = "override-to-be-removed"
- make_request_args = self._default_get_make_request_args(
- request_type=request_type,
- call_args=call_args,
- coordinator=coordinator,
- future=future,
- on_done_before_calls=on_done_before_calls,
- on_done_after_calls=on_done_after_calls,
- )
+ try:
+ make_request_args = self._default_get_make_request_args(
+ request_type=request_type,
+ call_args=call_args,
+ coordinator=coordinator,
+ future=future,
+ on_done_before_calls=on_done_before_calls,
+ on_done_after_calls=on_done_after_calls,
+ )
+ finally:
+ if provided_checksum_algorithm is not None:
+ call_args.extra_args['ChecksumAlgorithm'] = (
+ provided_checksum_algorithm
+ )
make_request_args['send_filepath'] = send_filepath
make_request_args['checksum_config'] = checksum_config
return make_request_args
@@ -858,7 +1540,21 @@ def _get_make_request_args_get_object(
)
if isinstance(call_args.fileobj, str):
final_filepath = call_args.fileobj
- recv_filepath = self._os_utils.get_temp_filename(final_filepath)
+ # A redirected download creates more than one CRT request for the
+ # same logical transfer. Keep one temp path so the final done
+ # callback handles the file used by every attempt.
+ recv_filepath = future.meta.user_context.get(
+ self._DOWNLOAD_TEMP_FILENAME
+ )
+ if recv_filepath is None:
+ # Store the path before the first request so a redirected
+ # attempt reuses it.
+ recv_filepath = self._os_utils.get_temp_filename(
+ final_filepath
+ )
+ future.meta.user_context[self._DOWNLOAD_TEMP_FILENAME] = (
+ recv_filepath
+ )
on_done_before_calls.append(
RenameTempFileHandler(
coordinator, final_filepath, recv_filepath, self._os_utils
@@ -878,8 +1574,19 @@ def _get_make_request_args_get_object(
make_request_args['recv_filepath'] = recv_filepath
make_request_args['on_body'] = on_body
make_request_args['checksum_config'] = checksum_config
+ if self._should_download_in_single_request(future.meta.size):
+ make_request_args['type'] = S3RequestType.DEFAULT
+ make_request_args['operation_name'] = _get_operation_name(
+ request_type
+ )
return make_request_args
+ def _should_download_in_single_request(self, size):
+ if self._transfer_config is None or size is None:
+ return False
+ threshold = self._transfer_config.multipart_threshold
+ return threshold is not None and size <= threshold
+
def _should_calculate_upload_checksum(self, request_type, extra_args):
if any(
checksum_arg in extra_args
@@ -992,8 +1699,11 @@ def __call__(self, **kwargs):
)
except Exception as e:
self._osutil.remove_file(self._temp_filename)
- # the CRT future has done already at this point
- self._coordinator.set_exception(e)
+ # This runs as an on_done callback, so the transfer is already
+ # marked complete and the exception has to override that
+ # result. Otherwise the download reports success having
+ # written nothing.
+ self._coordinator.set_exception(e, override=True)
class AfterDoneHandler:
diff --git a/awscli/topics/s3-config.rst b/awscli/topics/s3-config.rst
index defa8b2da5ec..31cd99b40b4c 100644
--- a/awscli/topics/s3-config.rst
+++ b/awscli/topics/s3-config.rst
@@ -376,12 +376,23 @@ files to and from S3. Valid choices are:
* S3 to S3 copies - Falls back to using the ``classic`` transfer client
- * Region redirects - Transfers fail for requests sent to a region that does
- not match the region of the targeted S3 bucket.
+ * Some region redirects - Transfers to a region that does not match the
+ region of the targeted S3 bucket are retried in the bucket's region, but
+ only if the transfer can be replayed from the start. A transfer fails
+ instead of being redirected if it is an upload whose source cannot be
+ rewound or if any of its data was already transferred. The ``classic``
+ transfer client is able to redirect in both of those cases.
- * ``max_concurrent_requests``, ``max_queue_size``, ``multipart_threshold``,
- and ``max_bandwidth`` configuration values - Ignores these configuration
- values.
+
+ * ``max_queue_size`` and ``max_bandwidth`` configuration values - Ignores
+ these configuration values.
+
+ * ``max_attempts`` configuration value - Honors values greater than one.
+ Ignores a value of one because the ``crt`` transfer client cannot disable
+ retries.
+
+ When a configured value is not supported by the transfer client being used,
+ the AWS CLI emits a warning naming the values it is ignoring.
target_bandwidth
diff --git a/tests/functional/s3/__init__.py b/tests/functional/s3/__init__.py
index 76bb8d076ccc..f9a6cff6e6b8 100644
--- a/tests/functional/s3/__init__.py
+++ b/tests/functional/s3/__init__.py
@@ -505,6 +505,7 @@ def simulate_make_request_side_effect(self, *args, **kwargs):
s3_request = FakeCRTS3Request(
future=FakeCRTFuture(kwargs.get('on_done'))
)
+ s3_request.finished_future.result()
return s3_request
def simulate_file_download(self, recv_filepath):
diff --git a/tests/functional/s3transfer/test_crt.py b/tests/functional/s3transfer/test_crt.py
index 790d80c46452..eb7e1c0f0e12 100644
--- a/tests/functional/s3transfer/test_crt.py
+++ b/tests/functional/s3transfer/test_crt.py
@@ -16,12 +16,15 @@
import time
from concurrent.futures import Future
+import pytest
+from botocore.exceptions import ClientError
from botocore.session import Session
from s3transfer.subscribers import BaseSubscriber
from tests import (
HAS_CRT,
FileCreator,
+ FileSizeProvider,
NonSeekableReader,
NonSeekableWriter,
mock,
@@ -33,6 +36,10 @@
import awscrt
import s3transfer.crt
+# Bound on waiting for a transfer that completes from another thread, so a
+# transfer that never completes fails the test instead of hanging it.
+RESULT_TIMEOUT = 20
+
class submitThread(threading.Thread):
def __init__(self, transfer_manager, futures, callargs):
@@ -49,16 +56,20 @@ class RecordingSubscriber(BaseSubscriber):
def __init__(self):
self.on_queued_called = False
self.on_done_called = False
+ self.on_queued_calls = 0
+ self.on_done_calls = 0
self.bytes_transferred = 0
self.on_queued_future = None
self.on_done_future = None
def on_queued(self, future, **kwargs):
self.on_queued_called = True
+ self.on_queued_calls += 1
self.on_queued_future = future
def on_done(self, future, **kwargs):
self.on_done_called = True
+ self.on_done_calls += 1
self.on_done_future = future
@@ -97,11 +108,13 @@ def setUp(self):
self.request_serializer = s3transfer.crt.BotocoreCRTRequestSerializer(
self.session
)
+ self.crt_client_factory = mock.Mock(return_value=self.s3_crt_client)
self.transfer_manager = s3transfer.crt.CRTTransferManager(
- crt_s3_client=self.s3_crt_client,
+ crt_client_factory=self.crt_client_factory,
crt_request_serializer=self.request_serializer,
)
self.record_subscriber = RecordingSubscriber()
+ self.completion_threads = []
def tearDown(self):
self.files.remove_all()
@@ -231,7 +244,7 @@ def _set_checksum_config_variables(
self.session
)
self.transfer_manager = s3transfer.crt.CRTTransferManager(
- crt_s3_client=self.s3_crt_client,
+ crt_client_factory=self.crt_client_factory,
crt_request_serializer=self.request_serializer,
)
@@ -329,7 +342,7 @@ def _invoke_done_callbacks(self, **kwargs):
callargs = self.s3_crt_client.make_request.call_args
callargs_kwargs = callargs[1]
on_done = callargs_kwargs["on_done"]
- on_done(error=None)
+ on_done(error=kwargs.get('error'))
def _simulate_file_download(self, recv_filepath):
self.files.create_file(
@@ -347,6 +360,96 @@ def _simulate_make_request_side_effect(self, **kwargs):
self._invoke_done_callbacks()
return self.s3_request
+ def _create_redirect_error(self, region=None):
+ headers = [] if region is None else [('x-amz-bucket-region', region)]
+ return awscrt.s3.S3ResponseError(
+ code=14343,
+ name='AWS_ERROR_S3_INVALID_RESPONSE_STATUS',
+ message='Invalid response status from request',
+ status_code=301,
+ headers=headers,
+ body=b'PermanentRedirect',
+ operation_name='PutObject',
+ )
+
+ def _create_redirect_transfer_manager(
+ self, initial_client, client_factory
+ ):
+ def create_client(region=None):
+ if region is None:
+ return initial_client
+ return client_factory(region)
+
+ return s3transfer.crt.CRTTransferManager(
+ crt_client_factory=create_client,
+ crt_request_serializer=self.request_serializer,
+ )
+
+ def _create_redirecting_transfer_manager(
+ self, initial_make_request, redirected_make_request=None
+ ):
+ """Create a manager whose initial region and redirected region differ.
+
+ The clients for both regions and the factory that creates the
+ redirected one are recorded as ``self.initial_client``,
+ ``self.redirected_client``, and ``self.redirected_client_factory``.
+ """
+ self.initial_client = mock.Mock(awscrt.s3.S3Client)
+ self.initial_client.make_request.side_effect = initial_make_request
+ self.redirected_client = mock.Mock(awscrt.s3.S3Client)
+ self.redirected_client.make_request.side_effect = (
+ redirected_make_request or self._succeed_make_request
+ )
+ self.redirected_client_factory = mock.Mock(
+ return_value=self.redirected_client
+ )
+ return self._create_redirect_transfer_manager(
+ self.initial_client, self.redirected_client_factory
+ )
+
+ def _upload_and_wait(self, transfer_manager, subscribers=None):
+ future = transfer_manager.upload(
+ self.filename,
+ self.bucket,
+ self.key,
+ {},
+ subscribers if subscribers is not None else [],
+ )
+ future.result(timeout=RESULT_TIMEOUT)
+ return future
+
+ def _fail_make_request(self, error):
+ def make_request(**kwargs):
+ kwargs['on_done'](error=error)
+ return mock.Mock(awscrt.s3.S3Request)
+
+ return make_request
+
+ def _fail_make_request_on_other_thread(self, error):
+ """Fail a request from another thread, like a CRT completion thread.
+
+ The thread the request completed on is recorded in
+ ``self.completion_threads``.
+ """
+
+ def complete_request(on_done):
+ self.completion_threads.append(threading.get_ident())
+ on_done(error=error)
+
+ def make_request(**kwargs):
+ thread = threading.Thread(
+ target=complete_request, args=(kwargs['on_done'],)
+ )
+ self.addCleanup(thread.join)
+ thread.start()
+ return mock.Mock(awscrt.s3.S3Request)
+
+ return make_request
+
+ def _succeed_make_request(self, **kwargs):
+ kwargs['on_done'](error=None)
+ return mock.Mock(awscrt.s3.S3Request)
+
def test_upload(self):
future = self.transfer_manager.upload(
self.filename, self.bucket, self.key, {}, [self.record_subscriber]
@@ -373,6 +476,330 @@ def test_upload(self):
)
self._assert_subscribers_called(future)
+ def test_upload_redirects_and_reuses_cached_region(self):
+ redirected_region = 'eu-central-1'
+ transfer_manager = self._create_redirecting_transfer_manager(
+ self._fail_make_request(
+ self._create_redirect_error(redirected_region)
+ )
+ )
+
+ first_subscriber = RecordingSubscriber()
+ first_future = transfer_manager.upload(
+ self.filename,
+ self.bucket,
+ self.key,
+ {},
+ [first_subscriber],
+ )
+ first_future.result()
+
+ self.assertEqual(self.initial_client.make_request.call_count, 1)
+ self.assertEqual(self.redirected_client.make_request.call_count, 1)
+ self.redirected_client_factory.assert_called_once_with(
+ redirected_region
+ )
+ initial_call = self.initial_client.make_request.call_args_list[
+ 0
+ ].kwargs
+ redirected_call = self.redirected_client.make_request.call_args_list[
+ 0
+ ].kwargs
+ self.assertEqual(
+ initial_call['request'].headers.get('host'),
+ f's3.{self.region}.amazonaws.com',
+ )
+ self.assertEqual(
+ redirected_call['request'].headers.get('host'),
+ f's3.{redirected_region}.amazonaws.com',
+ )
+ # The redirect is internal to one logical transfer, so subscribers
+ # only see it once.
+ self.assertEqual(first_subscriber.on_queued_calls, 1)
+ self.assertEqual(first_subscriber.on_done_calls, 1)
+
+ second_subscriber = RecordingSubscriber()
+ second_future = transfer_manager.upload(
+ self.filename,
+ self.bucket,
+ self.key,
+ {},
+ [second_subscriber],
+ )
+ second_future.result()
+
+ self.assertEqual(self.initial_client.make_request.call_count, 1)
+ self.assertEqual(self.redirected_client.make_request.call_count, 2)
+ self.redirected_client_factory.assert_called_once_with(
+ redirected_region
+ )
+ self.assertEqual(second_subscriber.on_queued_calls, 1)
+ self.assertEqual(second_subscriber.on_done_calls, 1)
+
+ def test_upload_redirect_restores_seekable_stream_position(self):
+ redirected_region = 'eu-central-1'
+ redirect_error = self._create_redirect_error(redirected_region)
+ attempt_bodies = []
+
+ def consume_body_and_finish(error):
+ def make_request(**kwargs):
+ attempt_bodies.append(
+ kwargs['request'].body_stream._stream.read()
+ )
+ kwargs['on_done'](error=error)
+ return mock.Mock(awscrt.s3.S3Request)
+
+ return make_request
+
+ transfer_manager = self._create_redirecting_transfer_manager(
+ consume_body_and_finish(redirect_error),
+ consume_body_and_finish(None),
+ )
+
+ future = transfer_manager.upload(
+ io.BytesIO(self.expected_content), self.bucket, self.key, {}, []
+ )
+ future.result()
+
+ self.assertEqual(
+ attempt_bodies, [self.expected_content, self.expected_content]
+ )
+
+ def test_successful_upload_does_not_consult_redirect_policy(self):
+ # A transfer that did not fail is never a redirect candidate.
+ with mock.patch.object(
+ self.transfer_manager._region_redirect_policy,
+ 'is_error_redirect_candidate',
+ ) as is_error_redirect_candidate:
+ future = self.transfer_manager.upload(
+ self.filename, self.bucket, self.key, {}, []
+ )
+ future.result(timeout=RESULT_TIMEOUT)
+
+ is_error_redirect_candidate.assert_not_called()
+
+ def test_upload_does_not_redirect_to_configured_region(self):
+ # A redirect naming the region the request already used, e.g. from an
+ # accelerate or dualstack endpoint, is not worth retrying.
+ transfer_manager = self._create_redirecting_transfer_manager(
+ self._fail_make_request(self._create_redirect_error(self.region))
+ )
+
+ with self.assertRaises(ClientError):
+ self._upload_and_wait(transfer_manager)
+
+ # No duplicate client for a region the transfer already used, and no
+ # retry that would just fail again.
+ self.redirected_client_factory.assert_not_called()
+ self.assertEqual(self.initial_client.make_request.call_count, 1)
+
+ def test_upload_does_not_redirect_unrelated_error(self):
+ # An error that is not a redirect is ruled out on the completion
+ # thread, without handing anything off to a new thread.
+ access_denied = awscrt.s3.S3ResponseError(
+ code=14343,
+ name='AWS_ERROR_S3_INVALID_RESPONSE_STATUS',
+ message='Invalid response status from request',
+ status_code=403,
+ headers=[],
+ body=b'AccessDenied',
+ operation_name='PutObject',
+ )
+ transfer_manager = self._create_redirecting_transfer_manager(
+ self._fail_make_request(access_denied)
+ )
+
+ with mock.patch.object(
+ transfer_manager._region_redirect_policy, 'get_retry_region'
+ ) as get_retry_region:
+ with self.assertRaises(ClientError):
+ self._upload_and_wait(transfer_manager)
+
+ get_retry_region.assert_not_called()
+ self.redirected_client_factory.assert_not_called()
+
+ def test_upload_does_not_redirect_nonseekable_stream(self):
+ transfer_manager = self._create_redirecting_transfer_manager(
+ self._fail_make_request(
+ self._create_redirect_error('eu-central-1')
+ )
+ )
+
+ future = transfer_manager.upload(
+ NonSeekableReader(self.expected_content),
+ self.bucket,
+ self.key,
+ {},
+ [],
+ )
+
+ with self.assertRaises(ClientError):
+ future.result()
+ self.redirected_client_factory.assert_not_called()
+ self.redirected_client.make_request.assert_not_called()
+
+ def test_upload_does_not_redirect_after_progress(self):
+ redirect_error = self._create_redirect_error('eu-central-1')
+
+ def fail_after_progress(**kwargs):
+ kwargs['on_progress'](1)
+ kwargs['on_done'](error=redirect_error)
+ return mock.Mock(awscrt.s3.S3Request)
+
+ transfer_manager = self._create_redirecting_transfer_manager(
+ fail_after_progress
+ )
+
+ with self.assertRaises(ClientError):
+ self._upload_and_wait(transfer_manager)
+ self.redirected_client_factory.assert_not_called()
+ self.redirected_client.make_request.assert_not_called()
+
+ def test_concurrent_redirects_discover_region_once(self):
+ # Transfers redirected at the same time share one region lookup, and
+ # each request is sent on a client for the region it was signed for.
+ redirected_region = 'eu-central-1'
+ redirect_error = self._create_redirect_error(redirected_region)
+ release = threading.Event()
+
+ def fail_when_released(**kwargs):
+ def complete_request():
+ release.wait(RESULT_TIMEOUT)
+ kwargs['on_done'](error=redirect_error)
+
+ thread = threading.Thread(target=complete_request)
+ self.addCleanup(thread.join)
+ thread.start()
+ return mock.Mock(awscrt.s3.S3Request)
+
+ transfer_manager = self._create_redirecting_transfer_manager(
+ fail_when_released
+ )
+
+ with mock.patch.object(
+ self.request_serializer,
+ 'get_bucket_region',
+ wraps=self.request_serializer.get_bucket_region,
+ ) as discover_region:
+ # Both transfers are in flight before either has a region to
+ # reuse, then both fail with a redirect at once.
+ futures = [
+ transfer_manager.upload(
+ self.filename, self.bucket, f'{self.key}-{i}', {}, []
+ )
+ for i in range(2)
+ ]
+ release.set()
+ for future in futures:
+ future.result(timeout=RESULT_TIMEOUT)
+
+ # The region is discovered once and reused, rather than every
+ # redirected transfer paying for its own lookup.
+ self.assertEqual(discover_region.call_count, 1)
+ self.assertEqual(self.redirected_client.make_request.call_count, 2)
+ # Sending a request signed for one region on a client configured for
+ # another fails with SignatureDoesNotMatch, so every request has to
+ # agree with the client it was sent on.
+ for call in self.initial_client.make_request.call_args_list:
+ self.assertEqual(
+ call.kwargs['request'].headers.get('host'),
+ self.expected_host,
+ )
+ for call in self.redirected_client.make_request.call_args_list:
+ self.assertEqual(
+ call.kwargs['request'].headers.get('host'),
+ f's3.{redirected_region}.amazonaws.com',
+ )
+
+ def test_upload_redirect_does_not_block_completion_thread(self):
+ # Redirecting must not run on the CRT thread that reported the
+ # failure, since it can block on the network.
+ redirect_threads = []
+
+ def succeed_and_record_thread(**kwargs):
+ redirect_threads.append(threading.get_ident())
+ return self._succeed_make_request(**kwargs)
+
+ transfer_manager = self._create_redirecting_transfer_manager(
+ self._fail_make_request_on_other_thread(
+ self._create_redirect_error('eu-central-1')
+ ),
+ succeed_and_record_thread,
+ )
+
+ self._upload_and_wait(transfer_manager)
+
+ # Discovering the region and serializing the retry can both block on
+ # the network, so they must not run on the thread the CRT completed
+ # the original request on.
+ self.assertEqual(len(redirect_threads), 1)
+ self.assertEqual(len(self.completion_threads), 1)
+ self.assertNotEqual(redirect_threads[0], self.completion_threads[0])
+
+ def test_cancel_cancels_retry_started_before_original_request_returned(
+ self,
+ ):
+ # A redirect can start before the original request registers, so a
+ # cancel has to reach the retry rather than the finished request.
+ redirect_error = self._create_redirect_error('eu-central-1')
+ original_request = mock.Mock(awscrt.s3.S3Request)
+ retry_request = mock.Mock(awscrt.s3.S3Request)
+ retry_started = threading.Event()
+ retry_callbacks = {}
+
+ def start_retry(**kwargs):
+ # Leave the retry in flight so it is the request a cancel has to
+ # reach.
+ retry_callbacks['on_done'] = kwargs['on_done']
+ retry_started.set()
+ return retry_request
+
+ def redirect_before_returning(**kwargs):
+ kwargs['on_done'](error=redirect_error)
+ # The redirect is handled on another thread, so wait for the retry
+ # to register before this request reports its own native request.
+ self.assertTrue(retry_started.wait(RESULT_TIMEOUT))
+ return original_request
+
+ transfer_manager = self._create_redirecting_transfer_manager(
+ redirect_before_returning, start_retry
+ )
+
+ future = transfer_manager.upload(
+ self.filename, self.bucket, self.key, {}, []
+ )
+ future.cancel()
+
+ # The original request completed and was replaced by the retry, so
+ # cancelling must not target the request that already finished.
+ retry_request.cancel.assert_called_once_with()
+ original_request.cancel.assert_not_called()
+
+ retry_callbacks['on_done'](error=None)
+ future.result(timeout=RESULT_TIMEOUT)
+
+ def test_upload_completes_when_redirect_decision_raises(self):
+ # A redirect decision that raises must still finish the transfer.
+ # The CRT invokes on_done from one of its own threads, so raising
+ # there strands the transfer instead of failing make_request().
+ transfer_manager = self._create_redirecting_transfer_manager(
+ self._fail_make_request_on_other_thread(
+ self._create_redirect_error('eu-central-1')
+ )
+ )
+ with mock.patch.object(
+ transfer_manager._region_redirect_policy,
+ 'get_retry_region',
+ side_effect=RuntimeError('Unexpected redirect failure'),
+ ):
+ # The transfer must still finish, and surface the error from the
+ # transfer itself instead of the one from the redirect decision.
+ with self.assertRaises(ClientError):
+ self._upload_and_wait(transfer_manager)
+ transfer_manager.shutdown()
+
+ self.redirected_client_factory.assert_not_called()
+
def test_upload_from_seekable_stream(self):
with open(self.filename, 'rb') as f:
future = self.transfer_manager.upload(
@@ -778,13 +1205,14 @@ def test_blocks_when_max_requests_processes_reached(self):
def _cancel_function(self):
self.cancel_called = True
- self.s3_request.finished_future.set_exception(
- awscrt.exceptions.from_code(0)
- )
- self._invoke_done_callbacks()
+ error = awscrt.exceptions.from_code(0)
+ self.s3_request.finished_future.set_exception(error)
+ self._invoke_done_callbacks(error=error)
def test_cancel(self):
self.s3_request.finished_future = Future()
+ self.s3_crt_client.make_request.side_effect = None
+ self.s3_crt_client.make_request.return_value = self.s3_request
self.cancel_called = False
self.s3_request.cancel = self._cancel_function
try:
@@ -812,7 +1240,7 @@ def serialize_http_request(self, transfer_type, future):
not_impl_serializer = ExceptionRaisingSerializer()
transfer_manager = s3transfer.crt.CRTTransferManager(
- crt_s3_client=self.s3_crt_client,
+ crt_client_factory=self.crt_client_factory,
crt_request_serializer=not_impl_serializer,
)
future = transfer_manager.upload(
@@ -831,3 +1259,88 @@ def test_crt_s3_client_error_handling(self):
)
with self.assertRaises(awscrt.exceptions.AwsCrtError):
future.result()
+
+
+MULTIPART_THRESHOLD = 8 * 1024 * 1024
+DOWNLOADED_CONTENT = 'content'
+
+
+@pytest.fixture
+def files():
+ file_creator = FileCreator()
+ yield file_creator
+ file_creator.remove_all()
+
+
+@pytest.fixture
+def filename(files):
+ return files.full_path('myfile')
+
+
+@pytest.fixture
+def crt_client(files):
+ client = mock.Mock(awscrt.s3.S3Client)
+
+ def simulate_make_request(**kwargs):
+ files.create_file(
+ kwargs['recv_filepath'], DOWNLOADED_CONTENT, mode='w'
+ )
+ kwargs['on_done'](error=None)
+ return mock.Mock(awscrt.s3.S3Request)
+
+ client.make_request.side_effect = simulate_make_request
+ return client
+
+
+@pytest.fixture
+def request_serializer():
+ session = Session()
+ session.set_config_variable('region', 'us-west-2')
+ return s3transfer.crt.BotocoreCRTRequestSerializer(session)
+
+
+@pytest.fixture
+def download(crt_client, request_serializer, filename):
+ """Downloads an object of a given size and returns the crt request args"""
+
+ def _download(size, multipart_threshold):
+ transfer_manager = s3transfer.crt.CRTTransferManager(
+ crt_client_factory=lambda region=None: crt_client,
+ crt_request_serializer=request_serializer,
+ transfer_config=s3transfer.crt.CRTTransferConfig(
+ multipart_threshold=multipart_threshold
+ ),
+ )
+ subscribers = [FileSizeProvider(size)]
+ transfer_manager.download(
+ 'test_bucket', 'test_key', filename, {}, subscribers
+ ).result()
+ return crt_client.make_request.call_args[1]
+
+ return _download
+
+
+class TestDownloadMultipartThreshold:
+ @pytest.mark.parametrize(
+ 'size', [MULTIPART_THRESHOLD - 1, MULTIPART_THRESHOLD]
+ )
+ def test_within_threshold_downloads_in_single_request(
+ self, download, size
+ ):
+ request_args = download(size, MULTIPART_THRESHOLD)
+ assert request_args['type'] == awscrt.s3.S3RequestType.DEFAULT
+ assert request_args['operation_name'] == 'GetObject'
+
+ def test_above_threshold_splits_download(self, download):
+ request_args = download(MULTIPART_THRESHOLD + 1, MULTIPART_THRESHOLD)
+ assert request_args['type'] == awscrt.s3.S3RequestType.GET_OBJECT
+ assert 'operation_name' not in request_args
+
+ def test_unset_threshold_splits_download(self, download):
+ request_args = download(1, None)
+ assert request_args['type'] == awscrt.s3.S3RequestType.GET_OBJECT
+
+ def test_single_request_download_writes_file(self, download, filename):
+ download(MULTIPART_THRESHOLD - 1, MULTIPART_THRESHOLD)
+ with open(filename) as f:
+ assert f.read() == DOWNLOADED_CONTENT
diff --git a/tests/integration/s3transfer/test_crt.py b/tests/integration/s3transfer/test_crt.py
index 2603cfeac6ac..7dba76a7089c 100644
--- a/tests/integration/s3transfer/test_crt.py
+++ b/tests/integration/s3transfer/test_crt.py
@@ -58,17 +58,25 @@ def setUp(self):
self.s3_key = 's3key.txt'
self.download_path = os.path.join(self.files.rootdir, 'download.txt')
- def _create_s3_transfer(self):
+ def _create_s3_transfer(self, region=None):
+ if region is None:
+ region = self.region
self.request_serializer = s3transfer.crt.BotocoreCRTRequestSerializer(
- self.session, client_kwargs={'region_name': self.region}
- )
- self.s3_crt_client = s3transfer.crt.create_s3_crt_client(
- self.region, self._get_crt_credentials_provider()
+ self.session, client_kwargs={'region_name': region}
)
+ credentials_provider = self._get_crt_credentials_provider()
+ bootstrap = s3transfer.crt.create_crt_client_bootstrap()
self.record_subscriber = RecordingSubscriber()
self.osutil = OSUtils()
return s3transfer.crt.CRTTransferManager(
- self.s3_crt_client, self.request_serializer
+ crt_client_factory=lambda client_region=None: (
+ s3transfer.crt.create_s3_crt_client(
+ client_region or region,
+ credentials_provider,
+ bootstrap=bootstrap,
+ )
+ ),
+ crt_request_serializer=self.request_serializer,
)
def _get_crt_credentials_provider(self):
@@ -150,6 +158,22 @@ def test_upload_below_multipart_chunksize(self):
self.assertTrue(self.object_exists('foo.txt'))
self._assert_subscribers_called(file_size)
+ def test_upload_redirects_from_wrong_region(self):
+ wrong_region = (
+ 'us-east-1' if self.region != 'us-east-1' else 'us-west-2'
+ )
+ transfer = self._create_s3_transfer(region=wrong_region)
+ filename = self.files.create_file('redirect.txt', 'content')
+ self.addCleanup(self.delete_object, 'redirect.txt')
+
+ with transfer:
+ future = transfer.upload(
+ filename, self.bucket_name, 'redirect.txt'
+ )
+ future.result()
+
+ self.assertTrue(self.object_exists('redirect.txt'))
+
def test_upload_above_multipart_chunksize(self):
transfer = self._create_s3_transfer()
file_size = 20 * 1024 * 1024
diff --git a/tests/unit/botocore/test_config_provider.py b/tests/unit/botocore/test_config_provider.py
index caf78bac38f2..2dc162e840b7 100644
--- a/tests/unit/botocore/test_config_provider.py
+++ b/tests/unit/botocore/test_config_provider.py
@@ -499,6 +499,7 @@ def test_provide_nested_value_but_not_section(self):
def _make_provider_that_returns(return_value):
provider = mock.Mock(spec=BaseProvider)
provider.provide.return_value = return_value
+ provider.resolve.side_effect = lambda: BaseProvider.resolve(provider)
return provider
@@ -824,3 +825,90 @@ def test_is_deepcopyable(self):
env['AWS_ENDPOINT_URL_BATCH'] = 'https://another-new-endpoint-override'
assert provider.provide() == 'https://another-new-endpoint-override'
assert provider_deepcopy.provide() == 'https://endpoint-override'
+
+
+class TestIsExplicitlySet:
+ def test_constant_only_is_not_explicitly_set(self):
+ store = ConfigValueStore({'foo': ChainProvider([ConstantProvider(3)])})
+ assert store.get_config_variable('foo') == 3
+ assert not store.is_explicitly_set('foo')
+
+ def test_value_from_environment_is_explicitly_set(self):
+ store = ConfigValueStore(
+ {
+ 'foo': ChainProvider(
+ [
+ EnvironmentProvider('FOO', {'FOO': '1'}),
+ ConstantProvider(3),
+ ]
+ )
+ }
+ )
+ assert store.is_explicitly_set('foo')
+
+ def test_value_matching_constant_is_still_explicitly_set(self):
+ store = ConfigValueStore(
+ {
+ 'foo': ChainProvider(
+ [
+ EnvironmentProvider('FOO', {'FOO': '3'}),
+ ConstantProvider(3),
+ ]
+ )
+ }
+ )
+ assert store.is_explicitly_set('foo')
+
+ def test_override_is_explicitly_set(self):
+ store = ConfigValueStore({'foo': ConstantProvider(3)})
+ store.set_config_variable('foo', 1)
+ assert store.is_explicitly_set('foo')
+
+ def test_unknown_variable_is_not_explicitly_set(self):
+ assert not ConfigValueStore({}).is_explicitly_set('foo')
+
+ def test_conversion_func_applies_to_explicit_value(self):
+ store = ConfigValueStore(
+ {
+ 'foo': ChainProvider(
+ [EnvironmentProvider('FOO', {'FOO': '3'})],
+ conversion_func=int,
+ )
+ }
+ )
+ assert store.is_explicitly_set('foo')
+
+
+class TestResolve:
+ def test_constant_provider_resolves_as_unconfigured(self):
+ resolved = ConstantProvider(3).resolve()
+ assert resolved.value == 3
+ assert not resolved.is_configured
+
+ def test_environment_provider_resolves_as_configured(self):
+ resolved = EnvironmentProvider('FOO', {'FOO': '3'}).resolve()
+ assert resolved.value == '3'
+ assert resolved.is_configured
+
+ def test_provider_without_value_resolves_to_none(self):
+ assert EnvironmentProvider('FOO', {}).resolve() is None
+
+ def test_chain_preserves_origin_of_winning_provider(self):
+ chain = ChainProvider(
+ [EnvironmentProvider('FOO', {}), ConstantProvider(3)]
+ )
+ assert not chain.resolve().is_configured
+
+ def test_chain_applies_conversion_func_to_resolved_value(self):
+ chain = ChainProvider(
+ [EnvironmentProvider('FOO', {'FOO': '3'})], conversion_func=int
+ )
+ resolved = chain.resolve()
+ assert resolved.value == 3
+ assert resolved.is_configured
+
+ def test_provide_still_returns_the_value(self):
+ chain = ChainProvider(
+ [EnvironmentProvider('FOO', {}), ConstantProvider(3)]
+ )
+ assert chain.provide() == 3
diff --git a/tests/unit/botocore/test_utils.py b/tests/unit/botocore/test_utils.py
index ed7cc1a30480..318daf3e9c96 100644
--- a/tests/unit/botocore/test_utils.py
+++ b/tests/unit/botocore/test_utils.py
@@ -1839,6 +1839,26 @@ def test_does_not_redirect_unless_permanentredirect_recieved(self):
self.assertIsNone(redirect_response)
self.assertEqual(request_dict, {})
+ def test_does_not_redirect_without_bucket(self):
+ request_dict = {
+ 'url': 'https://s3.us-west-2.amazonaws.com',
+ 'context': {},
+ }
+ response = (
+ None,
+ {
+ 'Error': {'Code': 'PermanentRedirect'},
+ 'ResponseMetadata': {'HTTPHeaders': {}},
+ },
+ )
+
+ redirect_response = self.redirector.redirect_from_error(
+ request_dict, response, self.operation
+ )
+
+ self.assertIsNone(redirect_response)
+ self.client.head_bucket.assert_not_called()
+
def test_does_not_redirect_if_region_cannot_be_found(self):
request_dict = {
'url': 'https://us-west-2.amazonaws.com/foo',
diff --git a/tests/unit/customizations/s3/test_factory.py b/tests/unit/customizations/s3/test_factory.py
index 9fafd4a9d588..4b0e3600d596 100644
--- a/tests/unit/customizations/s3/test_factory.py
+++ b/tests/unit/customizations/s3/test_factory.py
@@ -10,21 +10,32 @@
# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
# ANY KIND, either express or implied. See the License for the specific
# language governing permissions and limitations under the License.
+import awscrt.exceptions
import awscrt.s3
import pytest
import s3transfer.crt
from awscrt.s3 import S3FileIoOptions, S3RequestTlsMode
from botocore.config import Config
from botocore.credentials import Credentials
+from botocore.exceptions import InvalidConfigError
from botocore.httpsession import DEFAULT_CA_BUNDLE
from botocore.session import Session
-from s3transfer.crt import CRTTransferManager
+from s3transfer.crt import CRTTransferManager, create_s3_crt_client
from s3transfer.manager import TransferManager
+from awscli.customizations.s3 import constants
from awscli.customizations.s3.factory import (
+ ADAPTIVE_RETRY_MODE,
+ CRT_PART_SIZE_EXCEEDS_MEMORY_LIMIT,
+ MAX_CRT_MAX_ATTEMPTS,
+ MIN_CRT_MAX_ATTEMPTS,
+ MINIMUM_TARGET_THROUGHPUT_GBPS,
ClientFactory,
TransferManagerFactory,
)
+from awscli.customizations.s3.transferconfig import (
+ InvalidConfigError as InvalidTransferConfigError,
+)
from awscli.customizations.s3.transferconfig import RuntimeConfig
from awscli.testutils import FileCreator, mock, unittest
@@ -54,10 +65,23 @@ def mock_crt_s3_client():
yield mock_client
+def stub_config_variables(session, **values):
+ """Resolves the named config variables and everything else to None"""
+ session.get_config_variable.side_effect = values.get
+
+
+def stub_configured_variables(session, *names):
+ """Marks the named config variables as explicitly configured"""
+ session.get_component.return_value.is_explicitly_set.side_effect = (
+ lambda name: name in names
+ )
+
+
@pytest.fixture
def transfer_manager_factory():
session = mock.Mock(Session)
- session.get_config_variable.return_value = None
+ stub_config_variables(session)
+ stub_configured_variables(session)
session.get_default_client_config.return_value = None
session.get_scoped_config.return_value = {}
return TransferManagerFactory(session)
@@ -161,7 +185,8 @@ def test_create_client_respects_source_region_for_copies(self):
class TestTransferManagerFactory(unittest.TestCase):
def setUp(self):
self.session = mock.Mock(Session)
- self.session.get_config_variable.return_value = None
+ stub_config_variables(self.session)
+ stub_configured_variables(self.session)
self.session.get_default_client_config.return_value = None
self.session.get_scoped_config.return_value = {}
self.factory = TransferManagerFactory(self.session)
@@ -284,6 +309,37 @@ def test_uses_region_parameter_for_crt_manager(self, mock_crt_client):
'param-region',
)
+ @mock.patch('s3transfer.crt.S3Client')
+ def test_creates_crt_client_for_redirected_region(self, mock_crt_client):
+ self.runtime_config = self.get_runtime_config(
+ preferred_transfer_client='crt'
+ )
+ transfer_manager = self.factory.create_transfer_manager(
+ self.params, self.runtime_config
+ )
+
+ # The client for the configured region is created up front, and each
+ # selected region is cached.
+ self.assertEqual(mock_crt_client.call_count, 1)
+ self.assertIs(
+ transfer_manager.get_crt_client(),
+ transfer_manager.get_crt_client(),
+ )
+ self.assertIs(
+ transfer_manager.get_crt_client('eu-central-1'),
+ transfer_manager.get_crt_client('eu-central-1'),
+ )
+
+ self.assertEqual(mock_crt_client.call_count, 2)
+ self.assertEqual(
+ mock_crt_client.call_args_list[0].kwargs['region'],
+ 'us-west-2',
+ )
+ self.assertEqual(
+ mock_crt_client.call_args_list[1].kwargs['region'],
+ 'eu-central-1',
+ )
+
@mock.patch('s3transfer.crt.S3Client')
def test_falls_back_to_session_region_for_crt_manager(
self, mock_crt_client
@@ -292,7 +348,7 @@ def test_falls_back_to_session_region_for_crt_manager(
preferred_transfer_client='crt'
)
params = {'verify_ssl': DEFAULT_CA_BUNDLE}
- self.session.get_config_variable.return_value = 'config-region'
+ stub_config_variables(self.session, region='config-region')
transfer_manager = self.factory.create_transfer_manager(
params, self.runtime_config
)
@@ -392,6 +448,21 @@ def test_disable_botocore_credentials_for_crt_manager(
self.session.get_credentials.assert_not_called()
self.assertIsNone(mock_crt_client.call_args[1]['credential_provider'])
+ @mock.patch('s3transfer.crt.S3Client')
+ def test_invalid_client_config_raises_when_creating_crt_manager(
+ self, mock_crt_client
+ ):
+ self.runtime_config = self.get_runtime_config(
+ preferred_transfer_client='crt'
+ )
+ self.params['verify_ssl'] = ' '
+
+ # Reported once here rather than once per submitted transfer.
+ with self.assertRaises(InvalidConfigError):
+ self.factory.create_transfer_manager(
+ self.params, self.runtime_config
+ )
+
@mock.patch('s3transfer.crt.S3Client')
@mock.patch('s3transfer.crt.ClientTlsContext')
def test_use_verify_ssl_parameter_for_crt_manager(
@@ -427,7 +498,7 @@ def test_use_ca_bundle_from_session_for_crt_manager(
fake_ca_bundle = self.files.create_file(
"fake_ca", fake_ca_contents, mode='wb'
)
- self.session.get_config_variable.return_value = fake_ca_bundle
+ stub_config_variables(self.session, ca_bundle=fake_ca_bundle)
transfer_manager = self.factory.create_transfer_manager(
self.params, self.runtime_config
)
@@ -510,7 +581,7 @@ def test_fio_options_configure_for_crt_manager(self, mock_crt_client):
mock_crt_client.call_args[1]['fio_options'], expected_fio_options
)
- @mock.patch('s3transfer.crt.get_recommended_throughput_target_gbps')
+ @mock.patch('awscrt.s3.get_recommended_throughput_target_gbps')
@mock.patch('s3transfer.crt.S3Client')
def test_target_bandwidth_uses_crt_recommended_throughput(
self, mock_crt_client, mock_get_target_gbps
@@ -547,11 +618,6 @@ def test_multipart_chunksize_configure_for_crt_manager(
self, mock_crt_client
):
part_size = 16 * (1024**2)
- self.session.get_scoped_config.return_value = {
- 's3': {
- 'multipart_chunksize': part_size,
- }
- }
self.runtime_config = self.get_runtime_config(
preferred_transfer_client='crt', multipart_chunksize=part_size
)
@@ -563,12 +629,10 @@ def test_multipart_chunksize_configure_for_crt_manager(
@mock.patch('s3transfer.crt.S3Client')
def test_default_part_size_for_crt_manager(self, mock_crt_client):
- part_size = 16 * (1024**2)
- # Explicitly showing that the user has not configured
- # `multipart_chunksize`.
- self.session.get_scoped_config.return_value = {'s3': {}}
+ # `multipart_chunksize` is not provided, so it is not explicitly
+ # configured even though the runtime config still resolves a default.
self.runtime_config = self.get_runtime_config(
- preferred_transfer_client='crt', multipart_chunksize=part_size
+ preferred_transfer_client='crt'
)
transfer_manager = self.factory.create_transfer_manager(
self.params, self.runtime_config
@@ -578,6 +642,490 @@ def test_default_part_size_for_crt_manager(self, mock_crt_client):
# `part_size` to `None`.
self.assertEqual(mock_crt_client.call_args[1]['part_size'], None)
+ @mock.patch('s3transfer.crt.S3Client')
+ def test_multipart_threshold_configure_for_crt_manager(
+ self, mock_crt_client
+ ):
+ threshold = 64 * (1024**2)
+ self.runtime_config = self.get_runtime_config(
+ preferred_transfer_client='crt', multipart_threshold=threshold
+ )
+ transfer_manager = self.factory.create_transfer_manager(
+ self.params, self.runtime_config
+ )
+ self.assert_is_crt_manager(transfer_manager)
+ self.assertEqual(
+ mock_crt_client.call_args[1]['multipart_upload_threshold'],
+ threshold,
+ )
+
+ @mock.patch('s3transfer.crt.S3Client')
+ def test_max_concurrent_requests_configure_for_crt_manager(
+ self, mock_crt_client
+ ):
+ self.runtime_config = self.get_runtime_config(
+ preferred_transfer_client='crt', max_concurrent_requests=3
+ )
+ transfer_manager = self.factory.create_transfer_manager(
+ self.params, self.runtime_config
+ )
+ self.assert_is_crt_manager(transfer_manager)
+ self.assertEqual(
+ mock_crt_client.call_args[1]['max_active_connections_override'], 3
+ )
+
+ def test_optimized_system_does_not_use_transfer_config_defaults(self):
+ runtime_config = self.get_runtime_config()
+ with mock.patch(
+ 'awscrt.s3.is_optimized_for_system', return_value=True
+ ):
+ self.assertFalse(
+ self.factory._should_use_transfer_config_defaults(
+ runtime_config
+ )
+ )
+
+ def test_explicit_crt_does_not_use_transfer_config_defaults(self):
+ runtime_config = self.get_runtime_config(
+ preferred_transfer_client='crt'
+ )
+ with mock.patch(
+ 'awscrt.s3.is_optimized_for_system', return_value=False
+ ):
+ self.assertFalse(
+ self.factory._should_use_transfer_config_defaults(
+ runtime_config
+ )
+ )
+
+ def test_newly_eligible_system_uses_transfer_config_defaults(self):
+ runtime_config = self.get_runtime_config()
+ with mock.patch(
+ 'awscrt.s3.is_optimized_for_system', return_value=False
+ ):
+ self.assertTrue(
+ self.factory._should_use_transfer_config_defaults(
+ runtime_config
+ )
+ )
+
+ @mock.patch('awscrt.s3.is_optimized_for_system', return_value=False)
+ @mock.patch('s3transfer.crt.S3Client')
+ def test_transfer_config_defaults_passed_for_newly_eligible_system(
+ self, mock_crt_client, mock_is_optimized
+ ):
+ self.runtime_config = self.get_runtime_config()
+ with mock.patch.object(
+ self.factory,
+ '_resolve_transfer_client_type_for_system',
+ return_value=constants.CRT_TRANSFER_CLIENT,
+ ):
+ transfer_manager = self.factory.create_transfer_manager(
+ self.params, self.runtime_config
+ )
+ self.assert_is_crt_manager(transfer_manager)
+ call_kwargs = mock_crt_client.call_args[1]
+ defaults = RuntimeConfig.defaults()
+ self.assertEqual(
+ call_kwargs['part_size'], defaults['multipart_chunksize']
+ )
+ self.assertEqual(
+ call_kwargs['multipart_upload_threshold'],
+ defaults['multipart_threshold'],
+ )
+ self.assertEqual(
+ call_kwargs['max_active_connections_override'],
+ defaults['max_concurrent_requests'],
+ )
+
+ @mock.patch('s3transfer.crt.S3Client')
+ def test_unconfigured_options_not_passed_to_crt_manager(
+ self, mock_crt_client
+ ):
+ self.runtime_config = self.get_runtime_config(
+ preferred_transfer_client='crt'
+ )
+ transfer_manager = self.factory.create_transfer_manager(
+ self.params, self.runtime_config
+ )
+ self.assert_is_crt_manager(transfer_manager)
+ call_kwargs = mock_crt_client.call_args[1]
+ self.assertIsNone(call_kwargs['part_size'])
+ self.assertIsNone(call_kwargs['multipart_upload_threshold'])
+ self.assertIsNone(call_kwargs['max_active_connections_override'])
+
+ @mock.patch('s3transfer.crt.S3Client')
+ def test_part_size_configured_when_matching_default(self, mock_crt_client):
+ # Explicitly configuring the same value as the default still counts
+ # as explicitly configured.
+ default_chunksize = RuntimeConfig.defaults()['multipart_chunksize']
+ self.runtime_config = self.get_runtime_config(
+ preferred_transfer_client='crt',
+ multipart_chunksize=default_chunksize,
+ )
+ transfer_manager = self.factory.create_transfer_manager(
+ self.params, self.runtime_config
+ )
+ self.assert_is_crt_manager(transfer_manager)
+ self.assertEqual(
+ mock_crt_client.call_args[1]['part_size'], default_chunksize
+ )
+
+
+@pytest.fixture
+def auto_resolve_session():
+ session = mock.Mock(Session)
+ stub_config_variables(session)
+ stub_configured_variables(session)
+ session.get_default_client_config.return_value = None
+ session.get_scoped_config.return_value = {}
+ return session
+
+
+@pytest.fixture
+def auto_resolve_factory(auto_resolve_session, monkeypatch):
+ monkeypatch.setenv('AWS_CLI_AUTO_RESOLVE_CLIENT', 'crt')
+ return TransferManagerFactory(auto_resolve_session)
+
+
+@pytest.fixture
+def mock_crt_lock_held(auto_resolve_factory):
+ with mock.patch.object(
+ auto_resolve_factory,
+ '_is_crt_client_running_in_other_aws_cli_process',
+ return_value=False,
+ ) as mock_lock_held:
+ yield mock_lock_held
+
+
+@pytest.fixture
+def resolve_client_type(
+ auto_resolve_factory,
+ s3_params,
+ mock_crt_is_optimized_for_system,
+ mock_crt_recommended_throughput,
+ mock_crt_lock_held,
+):
+ def _resolve(**kwargs):
+ runtime_config = RuntimeConfig().build_config(**kwargs)
+ return auto_resolve_factory._compute_transfer_client_type(
+ s3_params, runtime_config
+ )
+
+ return _resolve
+
+
+class TestAutoResolveCrtClient:
+ def test_resolves_to_crt_when_enabled(self, resolve_client_type):
+ assert resolve_client_type() == constants.CRT_TRANSFER_CLIENT
+
+ def test_resolves_to_classic_when_env_var_unset(
+ self, resolve_client_type, monkeypatch
+ ):
+ monkeypatch.delenv('AWS_CLI_AUTO_RESOLVE_CLIENT')
+ assert resolve_client_type() == constants.CLASSIC_TRANSFER_CLIENT
+
+ def test_resolves_to_classic_when_env_var_is_other_value(
+ self, resolve_client_type, monkeypatch
+ ):
+ monkeypatch.setenv('AWS_CLI_AUTO_RESOLVE_CLIENT', 'classic')
+ assert resolve_client_type() == constants.CLASSIC_TRANSFER_CLIENT
+
+ def test_optimized_system_resolves_to_crt_without_env_var(
+ self,
+ resolve_client_type,
+ monkeypatch,
+ mock_crt_is_optimized_for_system,
+ ):
+ monkeypatch.delenv('AWS_CLI_AUTO_RESOLVE_CLIENT')
+ mock_crt_is_optimized_for_system.return_value = True
+ assert resolve_client_type() == constants.CRT_TRANSFER_CLIENT
+
+ def test_resolves_to_classic_when_max_bandwidth_configured(
+ self, resolve_client_type
+ ):
+ assert (
+ resolve_client_type(max_bandwidth=1024)
+ == constants.CLASSIC_TRANSFER_CLIENT
+ )
+
+ def test_resolves_to_classic_for_adaptive_retry_mode(
+ self, resolve_client_type, auto_resolve_session
+ ):
+ stub_config_variables(auto_resolve_session, retry_mode='adaptive')
+ assert resolve_client_type() == constants.CLASSIC_TRANSFER_CLIENT
+
+ def test_resolves_to_crt_for_standard_retry_mode(
+ self, resolve_client_type, auto_resolve_session
+ ):
+ stub_config_variables(auto_resolve_session, retry_mode='standard')
+ assert resolve_client_type() == constants.CRT_TRANSFER_CLIENT
+
+ def test_resolves_to_classic_for_stream_upload(
+ self, resolve_client_type, s3_params
+ ):
+ s3_params['is_stream'] = True
+ s3_params['paths_type'] = 'locals3'
+ assert resolve_client_type() == constants.CLASSIC_TRANSFER_CLIENT
+
+ def test_resolves_to_crt_for_stream_download(
+ self, resolve_client_type, s3_params
+ ):
+ s3_params['is_stream'] = True
+ s3_params['paths_type'] = 's3local'
+ assert resolve_client_type() == constants.CRT_TRANSFER_CLIENT
+
+ def test_optimized_system_resolves_to_crt_for_stream_upload(
+ self,
+ resolve_client_type,
+ s3_params,
+ mock_crt_is_optimized_for_system,
+ ):
+ mock_crt_is_optimized_for_system.return_value = True
+ s3_params['is_stream'] = True
+ s3_params['paths_type'] = 'locals3'
+ assert resolve_client_type() == constants.CRT_TRANSFER_CLIENT
+
+ def test_resolves_to_classic_when_lock_held(
+ self, resolve_client_type, mock_crt_lock_held
+ ):
+ mock_crt_lock_held.return_value = True
+ assert resolve_client_type() == constants.CLASSIC_TRANSFER_CLIENT
+
+ def test_explicit_crt_ignores_unsupported_settings(
+ self, resolve_client_type
+ ):
+ assert (
+ resolve_client_type(
+ preferred_transfer_client='crt', max_bandwidth=1024
+ )
+ == constants.CRT_TRANSFER_CLIENT
+ )
+
+ def test_s3s3_always_resolves_to_classic(
+ self, resolve_client_type, s3_params
+ ):
+ s3_params['paths_type'] = 's3s3'
+ assert resolve_client_type() == constants.CLASSIC_TRANSFER_CLIENT
+
+
+class TestClassicOnlySettingsWarning:
+ def test_warns_when_routed_away_for_max_bandwidth(
+ self, resolve_client_type, capsys
+ ):
+ resolve_client_type(max_bandwidth=1024)
+ warning = capsys.readouterr().err
+ assert 'max_bandwidth' in warning
+ assert 'A future version of the AWS CLI' in warning
+ assert 'preferred_transfer_client' in warning
+
+ def test_does_not_warn_for_adaptive_retry_mode(
+ self, resolve_client_type, auto_resolve_session, capsys
+ ):
+ # The crt transfer client will eventually support adaptive retries, so
+ # there is nothing for the user to act on.
+ stub_config_variables(auto_resolve_session, retry_mode='adaptive')
+ resolve_client_type()
+ assert capsys.readouterr().err == ''
+
+ def test_does_not_warn_for_stream_upload_fallback(
+ self, resolve_client_type, s3_params, capsys
+ ):
+ s3_params['is_stream'] = True
+ s3_params['paths_type'] = 'locals3'
+ resolve_client_type()
+ assert capsys.readouterr().err == ''
+
+ def test_does_not_warn_when_crt_is_resolved(
+ self, resolve_client_type, capsys
+ ):
+ resolve_client_type()
+ assert capsys.readouterr().err == ''
+
+ def test_does_not_warn_when_auto_resolve_disabled(
+ self, resolve_client_type, monkeypatch, capsys
+ ):
+ monkeypatch.delenv('AWS_CLI_AUTO_RESOLVE_CLIENT')
+ resolve_client_type(max_bandwidth=1024)
+ assert capsys.readouterr().err == ''
+
+ def test_does_not_warn_on_optimized_system(
+ self,
+ resolve_client_type,
+ mock_crt_is_optimized_for_system,
+ capsys,
+ ):
+ mock_crt_is_optimized_for_system.return_value = True
+ resolve_client_type(max_bandwidth=1024)
+ assert capsys.readouterr().err == ''
+
+ def test_does_not_warn_when_classic_explicitly_preferred(
+ self, resolve_client_type, capsys
+ ):
+ resolve_client_type(
+ preferred_transfer_client='classic', max_bandwidth=1024
+ )
+ assert capsys.readouterr().err == ''
+
+
+@pytest.fixture
+def mock_crt_get_ec2_instance_type():
+ with mock.patch('awscrt.s3.get_ec2_instance_type') as mock_instance_type:
+ mock_instance_type.return_value = None
+ yield mock_instance_type
+
+
+@pytest.fixture
+def mock_crt_recommended_throughput():
+ # The factory and s3transfer each hold their own reference, and which one
+ # resolves the target depends on the transfer client being created.
+ with (
+ mock.patch(
+ 'awscrt.s3.get_recommended_throughput_target_gbps'
+ ) as mock_recommended,
+ mock.patch(
+ 's3transfer.crt.get_recommended_throughput_target_gbps',
+ new=mock_recommended,
+ ),
+ ):
+ mock_recommended.return_value = None
+ yield mock_recommended
+
+
+@pytest.fixture
+def crt_s3_client_kwargs(
+ auto_resolve_factory,
+ s3_params,
+ mock_crt_is_optimized_for_system,
+ mock_crt_get_ec2_instance_type,
+ mock_crt_recommended_throughput,
+ mock_crt_s3_client,
+ mock_crt_process_lock,
+):
+ """Creates a crt transfer manager and returns the S3Client kwargs"""
+
+ def _create(**kwargs):
+ runtime_config = RuntimeConfig().build_config(**kwargs)
+ auto_resolve_factory._create_crt_transfer_manager(
+ s3_params, runtime_config
+ )
+ return mock_crt_s3_client.call_args[1]
+
+ return _create
+
+
+@pytest.fixture
+def crt_client_kwargs(auto_resolve_factory, mock_crt_is_optimized_for_system):
+ def _resolve(**kwargs):
+ runtime_config = RuntimeConfig().build_config(**kwargs)
+ return auto_resolve_factory._resolve_crt_client_config_kwargs(
+ runtime_config
+ )
+
+ return _resolve
+
+
+@pytest.fixture
+def warn_unsupported_settings(auto_resolve_factory, capsys):
+ def _warn(client_type, **kwargs):
+ runtime_config = RuntimeConfig().build_config(**kwargs)
+ auto_resolve_factory.warn_unsupported_settings(
+ client_type, runtime_config
+ )
+ return capsys.readouterr().err
+
+ return _warn
+
+
+class TestWarnUnsupportedSettings:
+ def test_warns_for_options_crt_ignores(self, warn_unsupported_settings):
+ warning = warn_unsupported_settings(
+ constants.CRT_TRANSFER_CLIENT,
+ max_queue_size=500,
+ io_chunksize=1024,
+ )
+ assert 'max_queue_size' in warning
+ assert 'io_chunksize' in warning
+ assert constants.CRT_TRANSFER_CLIENT in warning
+
+ def test_warns_for_options_classic_ignores(
+ self, warn_unsupported_settings
+ ):
+ warning = warn_unsupported_settings(
+ constants.CLASSIC_TRANSFER_CLIENT,
+ target_bandwidth=1024,
+ direct_io=True,
+ )
+ assert 'target_bandwidth' in warning
+ assert 'direct_io' in warning
+ assert constants.CLASSIC_TRANSFER_CLIENT in warning
+
+ def test_warns_for_max_bandwidth_when_crt_resolved(
+ self, warn_unsupported_settings
+ ):
+ warning = warn_unsupported_settings(
+ constants.CRT_TRANSFER_CLIENT, max_bandwidth=1024
+ )
+ assert 'max_bandwidth' in warning
+
+ def test_does_not_warn_for_max_bandwidth_when_classic_resolved(
+ self, warn_unsupported_settings
+ ):
+ assert (
+ warn_unsupported_settings(
+ constants.CLASSIC_TRANSFER_CLIENT, max_bandwidth=1024
+ )
+ == ''
+ )
+
+ def test_warns_for_adaptive_retry_mode_under_crt(
+ self, warn_unsupported_settings, auto_resolve_session
+ ):
+ stub_config_variables(auto_resolve_session, retry_mode='adaptive')
+ warning = warn_unsupported_settings(constants.CRT_TRANSFER_CLIENT)
+ assert f'retry_mode = {ADAPTIVE_RETRY_MODE}' in warning
+
+ def test_does_not_warn_for_supported_retry_mode(
+ self, warn_unsupported_settings, auto_resolve_session
+ ):
+ stub_config_variables(auto_resolve_session, retry_mode='standard')
+ assert warn_unsupported_settings(constants.CRT_TRANSFER_CLIENT) == ''
+
+ def test_does_not_warn_for_adaptive_retry_mode_under_classic(
+ self, warn_unsupported_settings, auto_resolve_session
+ ):
+ stub_config_variables(auto_resolve_session, retry_mode='adaptive')
+ assert (
+ warn_unsupported_settings(constants.CLASSIC_TRANSFER_CLIENT) == ''
+ )
+
+ def test_does_not_warn_when_nothing_configured(
+ self, warn_unsupported_settings
+ ):
+ assert warn_unsupported_settings(constants.CRT_TRANSFER_CLIENT) == ''
+
+ def test_does_not_warn_for_supported_options(
+ self, warn_unsupported_settings
+ ):
+ assert (
+ warn_unsupported_settings(
+ constants.CRT_TRANSFER_CLIENT,
+ multipart_chunksize=8 * (1024**2),
+ max_concurrent_requests=5,
+ )
+ == ''
+ )
+
+ def test_does_not_warn_across_clients(self, warn_unsupported_settings):
+ assert (
+ warn_unsupported_settings(
+ constants.CLASSIC_TRANSFER_CLIENT,
+ max_queue_size=500,
+ io_chunksize=1024,
+ )
+ == ''
+ )
+
@pytest.mark.parametrize(
'preferred_transfer_client,extra_params,'
@@ -701,3 +1249,295 @@ def _create_transfer_manager_from_factory(
return transfer_manager_factory.create_transfer_manager(
params, runtime_config
)
+
+
+class TestMaxAttempts:
+ def test_resolves_to_classic_when_retries_disabled(
+ self, resolve_client_type, auto_resolve_session
+ ):
+ stub_config_variables(auto_resolve_session, max_attempts=1)
+ stub_configured_variables(auto_resolve_session, 'max_attempts')
+ assert resolve_client_type() == constants.CLASSIC_TRANSFER_CLIENT
+
+ def test_resolves_to_crt_when_retries_enabled(
+ self, resolve_client_type, auto_resolve_session
+ ):
+ stub_config_variables(auto_resolve_session, max_attempts=2)
+ stub_configured_variables(auto_resolve_session, 'max_attempts')
+ assert resolve_client_type() == constants.CRT_TRANSFER_CLIENT
+
+ def test_does_not_warn_when_falling_back_for_disabled_retries(
+ self, resolve_client_type, auto_resolve_session, capsys
+ ):
+ stub_config_variables(auto_resolve_session, max_attempts=1)
+ stub_configured_variables(auto_resolve_session, 'max_attempts')
+ resolve_client_type()
+ assert capsys.readouterr().err == ''
+
+ def test_warns_when_crt_explicitly_preferred_with_retries_disabled(
+ self, warn_unsupported_settings, auto_resolve_session
+ ):
+ stub_config_variables(auto_resolve_session, max_attempts=1)
+ stub_configured_variables(auto_resolve_session, 'max_attempts')
+ warning = warn_unsupported_settings(
+ constants.CRT_TRANSFER_CLIENT,
+ preferred_transfer_client=constants.CRT_TRANSFER_CLIENT,
+ )
+ assert 'max_attempts = 1' in warning
+
+ def test_maps_configured_max_attempts_to_crt_client(
+ self, crt_client_kwargs, auto_resolve_session
+ ):
+ stub_config_variables(auto_resolve_session, max_attempts=5)
+ stub_configured_variables(auto_resolve_session, 'max_attempts')
+ assert crt_client_kwargs()['retry_options'] == {'max_retries': 4}
+
+ def test_applies_default_max_attempts_for_newly_eligible_hosts(
+ self, crt_client_kwargs, auto_resolve_session
+ ):
+ stub_config_variables(auto_resolve_session, max_attempts=3)
+ assert crt_client_kwargs()['retry_options'] == {'max_retries': 2}
+
+ def test_omits_max_attempts_when_crt_explicitly_preferred(
+ self, crt_client_kwargs, auto_resolve_session
+ ):
+ stub_config_variables(auto_resolve_session, max_attempts=3)
+ kwargs = crt_client_kwargs(
+ preferred_transfer_client=constants.CRT_TRANSFER_CLIENT
+ )
+ assert 'retry_options' not in kwargs
+
+ def test_maps_configured_max_attempts_when_crt_explicitly_preferred(
+ self, crt_client_kwargs, auto_resolve_session
+ ):
+ stub_config_variables(auto_resolve_session, max_attempts=5)
+ stub_configured_variables(auto_resolve_session, 'max_attempts')
+ kwargs = crt_client_kwargs(
+ preferred_transfer_client=constants.CRT_TRANSFER_CLIENT
+ )
+ assert kwargs['retry_options'] == {'max_retries': 4}
+
+
+class TestTargetThroughput:
+ def test_targets_less_when_crt_has_no_recommendation(
+ self, crt_s3_client_kwargs
+ ):
+ assert crt_s3_client_kwargs()['throughput_target_gbps'] == 4.0
+
+ def test_defers_to_crt_recommendation_when_it_has_one(
+ self, crt_s3_client_kwargs, mock_crt_recommended_throughput
+ ):
+ mock_crt_recommended_throughput.return_value = 50.0
+ assert crt_s3_client_kwargs()['throughput_target_gbps'] == 50.0
+
+ def test_targets_less_on_ec2_hosts_crt_cannot_recommend_for(
+ self, crt_s3_client_kwargs, mock_crt_get_ec2_instance_type
+ ):
+ # Being on EC2 does not mean the crt client sized a pool for this
+ # host, so the instance type must not decide the throughput target.
+ mock_crt_get_ec2_instance_type.return_value = 't3.micro'
+ assert crt_s3_client_kwargs()['throughput_target_gbps'] == 4.0
+
+ def test_configured_target_bandwidth_wins(self, crt_s3_client_kwargs):
+ kwargs = crt_s3_client_kwargs(target_bandwidth=1_250_000_000)
+ assert kwargs['throughput_target_gbps'] == 10.0
+
+ def test_floors_throughput_when_crt_explicitly_preferred(
+ self, crt_s3_client_kwargs, mock_crt_recommended_throughput
+ ):
+ mock_crt_recommended_throughput.return_value = 3.0
+
+ kwargs = crt_s3_client_kwargs(
+ preferred_transfer_client=constants.CRT_TRANSFER_CLIENT
+ )
+ assert (
+ kwargs['throughput_target_gbps'] == MINIMUM_TARGET_THROUGHPUT_GBPS
+ )
+
+ def test_keeps_higher_recommendation_when_crt_explicitly_preferred(
+ self, crt_s3_client_kwargs, mock_crt_recommended_throughput
+ ):
+ mock_crt_recommended_throughput.return_value = 50.0
+ kwargs = crt_s3_client_kwargs(
+ preferred_transfer_client=constants.CRT_TRANSFER_CLIENT
+ )
+ assert kwargs['throughput_target_gbps'] == 50.0
+
+ def test_does_not_floor_configured_target_bandwidth(
+ self, crt_s3_client_kwargs
+ ):
+ kwargs = crt_s3_client_kwargs(
+ preferred_transfer_client=constants.CRT_TRANSFER_CLIENT,
+ target_bandwidth=125_000_000,
+ )
+ assert kwargs['throughput_target_gbps'] == 1.0
+
+ def test_defers_to_crt_on_optimized_host(
+ self,
+ crt_s3_client_kwargs,
+ mock_crt_is_optimized_for_system,
+ mock_crt_recommended_throughput,
+ ):
+ mock_crt_is_optimized_for_system.return_value = True
+ mock_crt_recommended_throughput.return_value = 3.0
+ assert crt_s3_client_kwargs()['throughput_target_gbps'] == 3.0
+
+
+class TestChunksizeExceedingCrtMemoryPool:
+ """The crt client only reports an oversized chunksize while constructing."""
+
+ @pytest.fixture
+ def part_size_error(self):
+ return RuntimeError(
+ f'{CRT_PART_SIZE_EXCEEDS_MEMORY_LIMIT} '
+ f'(AWS_ERROR_S3_PART_SIZE_EXCEEDS_MEMORY_LIMIT): Part size '
+ f'exceeds the configured memory limit.'
+ )
+
+ @pytest.fixture
+ def create_manager(self, auto_resolve_factory, s3_params):
+ def _create(**kwargs):
+ runtime_config = RuntimeConfig().build_config(**kwargs)
+ return auto_resolve_factory.create_transfer_manager(
+ s3_params, runtime_config, mock.Mock()
+ )
+
+ return _create
+
+ @pytest.fixture
+ def crt_manager_raises(self, auto_resolve_factory, part_size_error):
+ with mock.patch.object(
+ auto_resolve_factory,
+ '_create_crt_transfer_manager',
+ side_effect=part_size_error,
+ ) as mock_create:
+ yield mock_create
+
+ def test_error_code_still_means_what_we_match_on(self):
+ # Guards against awscrt renumbering the code out from under us.
+ assert (
+ awscrt.exceptions.from_code(
+ CRT_PART_SIZE_EXCEEDS_MEMORY_LIMIT
+ ).name
+ == 'AWS_ERROR_S3_PART_SIZE_EXCEEDS_MEMORY_LIMIT'
+ )
+
+ def test_falls_back_to_classic_when_auto_resolved(
+ self, create_manager, crt_manager_raises, mock_crt_lock_held
+ ):
+ assert isinstance(create_manager(), TransferManager)
+
+ def test_releases_process_lock_when_falling_back(
+ self, create_manager, crt_manager_raises, mock_crt_lock_held
+ ):
+ # Holding the lock while running classic denies the crt client to
+ # every other process of the same application.
+ with mock.patch(
+ 'awscli.customizations.s3.factory.release_crt_s3_process_lock'
+ ) as mock_release:
+ create_manager()
+ assert mock_release.called
+
+ def test_does_not_warn_when_falling_back(
+ self, create_manager, crt_manager_raises, mock_crt_lock_held, capsys
+ ):
+ # Classic honors the configured chunksize, so nothing is lost.
+ create_manager()
+ assert capsys.readouterr().err == ''
+
+ def test_raises_when_crt_explicitly_preferred(
+ self, create_manager, crt_manager_raises
+ ):
+ with pytest.raises(InvalidTransferConfigError) as excinfo:
+ create_manager(
+ preferred_transfer_client=constants.CRT_TRANSFER_CLIENT
+ )
+ message = str(excinfo.value)
+ assert 'multipart_chunksize' in message
+ assert 'AWS_CRT_S3_MEMORY_LIMIT_IN_GIB' in message
+ # Explicit crt must never be told to switch to classic.
+ assert constants.CLASSIC_TRANSFER_CLIENT not in message
+
+ def test_reraises_unrelated_runtime_errors(
+ self, create_manager, auto_resolve_factory, mock_crt_lock_held
+ ):
+ with mock.patch.object(
+ auto_resolve_factory,
+ '_create_crt_transfer_manager',
+ side_effect=RuntimeError('something else entirely'),
+ ):
+ with pytest.raises(RuntimeError, match='something else entirely'):
+ create_manager()
+
+ def test_uses_crt_when_the_chunksize_fits(
+ self, create_manager, mock_crt_lock_held, mock_crt_s3_client
+ ):
+ manager = create_manager(multipart_chunksize=8 * 1024 * 1024)
+ assert not isinstance(manager, TransferManager)
+
+
+class TestMaxAttemptsBounds:
+ """The crt rejects max_retries of 0 and of 64 or more."""
+
+ @pytest.mark.parametrize(
+ 'max_attempts', [MIN_CRT_MAX_ATTEMPTS, 3, MAX_CRT_MAX_ATTEMPTS]
+ )
+ def test_resolves_to_crt_within_bounds(
+ self, resolve_client_type, auto_resolve_session, max_attempts
+ ):
+ stub_config_variables(auto_resolve_session, max_attempts=max_attempts)
+ stub_configured_variables(auto_resolve_session, 'max_attempts')
+ assert resolve_client_type() == constants.CRT_TRANSFER_CLIENT
+
+ @pytest.mark.parametrize(
+ 'max_attempts',
+ [MIN_CRT_MAX_ATTEMPTS - 1, MAX_CRT_MAX_ATTEMPTS + 1, 1000],
+ )
+ def test_resolves_to_classic_outside_bounds(
+ self, resolve_client_type, auto_resolve_session, max_attempts
+ ):
+ stub_config_variables(auto_resolve_session, max_attempts=max_attempts)
+ stub_configured_variables(auto_resolve_session, 'max_attempts')
+ assert resolve_client_type() == constants.CLASSIC_TRANSFER_CLIENT
+
+ @pytest.mark.parametrize(
+ 'max_attempts', [MAX_CRT_MAX_ATTEMPTS, MAX_CRT_MAX_ATTEMPTS + 1]
+ )
+ def test_only_maps_retries_within_bounds(
+ self, crt_client_kwargs, auto_resolve_session, max_attempts
+ ):
+ stub_config_variables(auto_resolve_session, max_attempts=max_attempts)
+ stub_configured_variables(auto_resolve_session, 'max_attempts')
+ kwargs = crt_client_kwargs()
+ if max_attempts == MAX_CRT_MAX_ATTEMPTS:
+ assert kwargs['retry_options'] == {'max_retries': max_attempts - 1}
+ else:
+ assert 'retry_options' not in kwargs
+
+ def test_warns_when_crt_explicitly_preferred_above_bounds(
+ self, warn_unsupported_settings, auto_resolve_session
+ ):
+ too_many = MAX_CRT_MAX_ATTEMPTS + 1
+ stub_config_variables(auto_resolve_session, max_attempts=too_many)
+ stub_configured_variables(auto_resolve_session, 'max_attempts')
+ warning = warn_unsupported_settings(
+ constants.CRT_TRANSFER_CLIENT,
+ preferred_transfer_client=constants.CRT_TRANSFER_CLIENT,
+ )
+ assert f'max_attempts = {too_many}' in warning
+ assert (
+ f'must be between {MIN_CRT_MAX_ATTEMPTS} and '
+ f'{MAX_CRT_MAX_ATTEMPTS}'
+ ) in warning
+
+ def test_upper_bound_matches_what_awscrt_accepts(self):
+ # Guards against awscrt moving the limit out from under us.
+ create_s3_crt_client(
+ region='us-west-2',
+ retry_options={'max_retries': MAX_CRT_MAX_ATTEMPTS - 1},
+ )
+ with pytest.raises(RuntimeError):
+ create_s3_crt_client(
+ region='us-west-2',
+ retry_options={'max_retries': MAX_CRT_MAX_ATTEMPTS},
+ )
diff --git a/tests/unit/customizations/s3/test_transferconfig.py b/tests/unit/customizations/s3/test_transferconfig.py
index 0deecf860997..bcac818b7976 100644
--- a/tests/unit/customizations/s3/test_transferconfig.py
+++ b/tests/unit/customizations/s3/test_transferconfig.py
@@ -193,3 +193,56 @@ def test_convert(self):
assert result.max_bandwidth == 1024 * 1024
assert result.io_chunksize == 1024 * 1024
assert result.max_in_memory_upload_chunks != 1000
+
+
+class TestResolvedRuntimeConfig:
+ def build(self, **kwargs):
+ return transferconfig.RuntimeConfig().build_config(**kwargs)
+
+ def test_tracks_explicitly_provided_keys(self):
+ config = self.build(max_concurrent_requests=5, io_chunksize='1MB')
+ assert config.explicit_keys == {
+ 'max_concurrent_requests',
+ 'io_chunksize',
+ }
+
+ def test_nothing_is_explicit_when_no_values_provided(self):
+ config = self.build()
+ assert config.explicit_keys == set()
+ assert not config.is_explicitly_set('max_concurrent_requests')
+
+ def test_value_matching_default_is_still_explicit(self):
+ default = transferconfig.DEFAULTS['max_concurrent_requests']
+ config = self.build(max_concurrent_requests=default)
+ assert config.is_explicitly_set('max_concurrent_requests')
+
+ def test_defaults_are_not_explicit(self):
+ config = self.build(max_concurrent_requests=5)
+ assert not config.is_explicitly_set('multipart_threshold')
+ # The default value is still resolved and available.
+ assert (
+ config['multipart_threshold']
+ == transferconfig.DEFAULTS['multipart_threshold']
+ )
+
+ def test_unknown_keys_are_tracked_as_explicit(self):
+ config = self.build(not_a_real_option='foo')
+ assert config.is_explicitly_set('not_a_real_option')
+ assert 'not_a_real_option' not in transferconfig.DEFAULTS
+
+ def test_copy_preserves_provenance(self):
+ config = self.build(io_chunksize='1MB').copy()
+ assert config.is_explicitly_set('io_chunksize')
+ assert not config.is_explicitly_set('multipart_threshold')
+
+ def test_provenance_records_keys_not_converted_values(self):
+ # Values are converted after provenance is captured.
+ config = self.build(multipart_chunksize='16MB')
+ assert config.is_explicitly_set('multipart_chunksize')
+ assert config['multipart_chunksize'] == 16 * 1024 * 1024
+
+ def test_behaves_like_a_dict(self):
+ config = self.build(max_queue_size=10)
+ assert config['max_queue_size'] == 10
+ assert config.get('max_queue_size') == 10
+ assert 'multipart_threshold' in dict(config)
diff --git a/tests/unit/s3transfer/test_crt.py b/tests/unit/s3transfer/test_crt.py
index 740131c8be81..6d1240608b61 100644
--- a/tests/unit/s3transfer/test_crt.py
+++ b/tests/unit/s3transfer/test_crt.py
@@ -11,18 +11,21 @@
# ANY KIND, either express or implied. See the License for the specific
# language governing permissions and limitations under the License.
import io
+import threading
+from concurrent.futures import Future
import pytest
from botocore.credentials import Credentials, ReadOnlyCredentials
from botocore.exceptions import (
ClientError,
InvalidConfigError,
+ InvalidRegionError,
NoCredentialsError,
)
from botocore.session import Session
from s3transfer.constants import GB
from s3transfer.exceptions import TransferNotDoneError
-from s3transfer.utils import CallArgs
+from s3transfer.utils import CallArgs, OSUtils
from tests import HAS_CRT, FileCreator, mock, requires_crt, unittest
@@ -169,14 +172,10 @@ def test_delete_request(self):
self.assertIsNone(crt_request.headers.get("Authorization"))
def _create_crt_response_error(
- self, status_code, body, operation_name=None
+ self, status_code, body, operation_name=None, headers=None
):
- return awscrt.s3.S3ResponseError(
- code=14343,
- name='AWS_ERROR_S3_INVALID_RESPONSE_STATUS',
- message='Invalid response status from request',
- status_code=status_code,
- headers=[
+ if headers is None:
+ headers = [
('x-amz-request-id', 'QSJHJJZR2EDYD4GQ'),
(
'x-amz-id-2',
@@ -186,11 +185,26 @@ def _create_crt_response_error(
('Transfer-Encoding', 'chunked'),
('Date', 'Fri, 10 Nov 2023 23:22:47 GMT'),
('Server', 'AmazonS3'),
- ],
+ ]
+ return awscrt.s3.S3ResponseError(
+ code=14343,
+ name='AWS_ERROR_S3_INVALID_RESPONSE_STATUS',
+ message='Invalid response status from request',
+ status_code=status_code,
+ headers=headers,
body=body,
operation_name=operation_name,
)
+ def _create_serializer_with_redirect_client(self):
+ redirect_client = mock.Mock()
+ client_factory = mock.Mock(return_value=redirect_client)
+ serializer = s3transfer.crt.BotocoreCRTRequestSerializer(
+ self.session,
+ region_redirect_client_factory=client_factory,
+ )
+ return serializer, redirect_client, client_factory
+
def test_translate_get_object_404(self):
body = (
b'\n'
@@ -222,6 +236,225 @@ def test_translate_unknown_operation_404(self):
boto_err = self.request_serializer.translate_crt_exception(crt_exc)
self.assertIsInstance(boto_err, ClientError)
+ def _serialize_get_object(self):
+ callargs = CallArgs(
+ bucket=self.bucket,
+ key=self.key,
+ fileobj=self.filename,
+ extra_args={},
+ subscribers=[],
+ )
+ coordinator = s3transfer.crt.CRTTransferCoordinator()
+ future = s3transfer.crt.CRTTransferFuture(
+ s3transfer.crt.CRTTransferMeta(call_args=callargs), coordinator
+ )
+ return self.request_serializer.serialize_http_request(
+ "get_object", future
+ )
+
+ def test_cached_bucket_region_changes_serialized_endpoint(self):
+ self.request_serializer.cache_bucket_region(
+ self.bucket, 'eu-central-1'
+ )
+ with self.request_serializer.locked_bucket_region(
+ self.bucket
+ ) as region:
+ self.assertEqual(region, 'eu-central-1')
+ crt_request = self._serialize_get_object()
+
+ self.assertEqual(
+ crt_request.headers.get("host"),
+ "s3.eu-central-1.amazonaws.com",
+ )
+
+ def test_is_redirect_error_for_permanent_redirect(self):
+ error = self._create_crt_response_error(
+ 301,
+ b'PermanentRedirect',
+ operation_name='PutObject',
+ )
+
+ self.assertTrue(
+ self.request_serializer.is_redirect_error('put_object', error)
+ )
+
+ def test_is_not_redirect_error_for_access_denied(self):
+ error = self._create_crt_response_error(
+ 403,
+ b'AccessDenied',
+ operation_name='PutObject',
+ )
+
+ self.assertFalse(
+ self.request_serializer.is_redirect_error('put_object', error)
+ )
+
+ def test_is_redirect_error_does_not_look_up_region(self):
+ # This is decided on a CRT completion thread, so it must not make the
+ # HeadBucket request that finding the region can fall back to.
+ serializer, redirect_client, client_factory = (
+ self._create_serializer_with_redirect_client()
+ )
+ # The fallback would work if it were used, so the assertions below
+ # show that it is not reached rather than that it failed.
+ redirect_client.head_bucket.return_value = {
+ 'ResponseMetadata': {
+ 'HTTPHeaders': {'x-amz-bucket-region': 'eu-central-1'}
+ }
+ }
+ error = self._create_crt_response_error(
+ 301,
+ b'PermanentRedirect',
+ operation_name='PutObject',
+ )
+
+ self.assertTrue(serializer.is_redirect_error('put_object', error))
+ client_factory.assert_not_called()
+ redirect_client.head_bucket.assert_not_called()
+
+ def test_locked_bucket_region_blocks_cache_updates(self):
+ # Endpoint resolution reads the cache again while a request is being
+ # serialized. If a region could be cached in between, the request
+ # would be signed for one region and sent on a client for another.
+ caching = threading.Event()
+ cached = threading.Event()
+
+ def cache_region():
+ caching.set()
+ self.request_serializer.cache_bucket_region(
+ self.bucket, 'eu-central-1'
+ )
+ cached.set()
+
+ writer = threading.Thread(target=cache_region, daemon=True)
+ self.addCleanup(writer.join)
+ with self.request_serializer.locked_bucket_region(self.bucket):
+ writer.start()
+ self.assertTrue(caching.wait(5))
+ self.assertFalse(cached.wait(0.1))
+
+ writer.join(5)
+ self.assertEqual(
+ self.request_serializer.get_cached_bucket_region(self.bucket),
+ 'eu-central-1',
+ )
+
+ def test_region_lookup_does_not_block_building_a_request(self):
+ # A lookup can make a HeadBucket request, which may be slow or
+ # retried, so it must not hold up requests being built.
+ policy = s3transfer.crt.CRTS3RegionRedirectPolicy(
+ self.request_serializer
+ )
+ lookup_started = threading.Event()
+ release_lookup = threading.Event()
+ built = threading.Event()
+
+ def blocking_get_bucket_region(*args, **kwargs):
+ lookup_started.set()
+ release_lookup.wait(5)
+ return 'eu-central-1'
+
+ def build_request():
+ with self.request_serializer.locked_bucket_region(self.bucket):
+ pass
+ built.set()
+
+ with mock.patch.object(
+ self.request_serializer,
+ 'get_bucket_region',
+ blocking_get_bucket_region,
+ ):
+ # Daemons so that a regression leaving either thread blocked
+ # fails this test rather than wedging the interpreter on exit.
+ lookup = threading.Thread(
+ target=policy.get_retry_region,
+ args=(self.bucket, 'put_object', None),
+ daemon=True,
+ )
+ lookup.start()
+ builder = None
+ try:
+ self.assertTrue(lookup_started.wait(5))
+ # Building runs on its own thread so that sharing a lock with
+ # the lookup fails this test instead of hanging it.
+ builder = threading.Thread(target=build_request, daemon=True)
+ builder.start()
+ self.assertTrue(built.wait(1))
+ finally:
+ # Released before joining, so a failure does not leave the
+ # lookup and the builder blocked on each other.
+ release_lookup.set()
+ if builder is not None:
+ builder.join(5)
+ lookup.join(5)
+
+ def test_redirect_region_does_not_create_fallback_client(self):
+ serializer, redirect_client, client_factory = (
+ self._create_serializer_with_redirect_client()
+ )
+ error = self._create_crt_response_error(
+ 301,
+ None,
+ operation_name='GetObject',
+ headers=[('x-amz-bucket-region', 'eu-central-1')],
+ )
+
+ region = serializer.get_bucket_region(self.bucket, 'get_object', error)
+
+ self.assertEqual(region, 'eu-central-1')
+ client_factory.assert_not_called()
+ redirect_client.head_bucket.assert_not_called()
+
+ def test_redirect_region_creates_fallback_client(self):
+ serializer, redirect_client, client_factory = (
+ self._create_serializer_with_redirect_client()
+ )
+ redirect_client.head_bucket.return_value = {
+ 'ResponseMetadata': {
+ 'HTTPHeaders': {'x-amz-bucket-region': 'eu-central-1'}
+ }
+ }
+ error = self._create_crt_response_error(
+ 301,
+ b'PermanentRedirect',
+ operation_name='DeleteObject',
+ )
+
+ region = serializer.get_bucket_region(
+ self.bucket, 'delete_object', error
+ )
+
+ self.assertEqual(region, 'eu-central-1')
+ client_factory.assert_called_once_with()
+ redirect_client.head_bucket.assert_called_once_with(Bucket=self.bucket)
+
+ def test_redirect_region_rejects_invalid_region(self):
+ error = self._create_crt_response_error(
+ 301,
+ b'PermanentRedirect',
+ operation_name='GetObject',
+ headers=[('x-amz-bucket-region', 'invalid region!')],
+ )
+ with self.assertRaises(InvalidRegionError):
+ self.request_serializer.get_bucket_region(
+ self.bucket, 'get_object', error
+ )
+
+ def test_redirect_region_ignores_arn_bucket(self):
+ error = self._create_crt_response_error(
+ 301,
+ b'PermanentRedirect',
+ operation_name='GetObject',
+ headers=[('x-amz-bucket-region', 'eu-central-1')],
+ )
+ bucket = 'arn:aws:s3:us-west-2:123456789012:accesspoint/myendpoint'
+
+ region = self.request_serializer.get_bucket_region(
+ bucket, 'get_object', error
+ )
+
+ self.assertIsNone(region)
+
@requires_crt_pytest
class TestBotocoreCRTCredentialsWrapper:
@@ -301,7 +534,9 @@ def setUp(self):
self.mock_s3_request = mock.Mock(awscrt.s3.S3RequestType)
self.mock_crt_future = mock.Mock(awscrt.s3.Future)
self.mock_s3_request.finished_future = self.mock_crt_future
- self.coordinator = s3transfer.crt.CRTTransferCoordinator()
+ self.coordinator = s3transfer.crt.CRTTransferCoordinator(
+ completion_future=self.mock_crt_future
+ )
self.coordinator.set_s3_request(self.mock_s3_request)
self.future = s3transfer.crt.CRTTransferFuture(
coordinator=self.coordinator
@@ -324,6 +559,209 @@ def test_set_exception_can_override_previous_exception(self):
self.future.result()
+@requires_crt_pytest
+class TestCRTTransferCoordinator:
+ def setup_method(self):
+ self.completion_future = Future()
+ self.coordinator = s3transfer.crt.CRTTransferCoordinator(
+ completion_future=self.completion_future
+ )
+
+ def create_s3_request(self):
+ s3_request = mock.Mock(awscrt.s3.S3Request)
+ s3_request.finished_future = Future()
+ return s3_request
+
+ def test_set_s3_request(self):
+ s3_request = self.create_s3_request()
+ self.coordinator.set_s3_request(s3_request)
+ assert self.coordinator.s3_request is s3_request
+
+ def test_original_request_cannot_replace_redirect(self):
+ first_request = self.create_s3_request()
+ second_request = self.create_s3_request()
+ # The redirect started before the original request registered its
+ # native request, so the original request must not become active.
+ self.coordinator.set_s3_request(
+ second_request, is_region_redirect=True
+ )
+ self.coordinator.set_s3_request(first_request)
+
+ assert self.coordinator.s3_request is second_request
+
+ def test_cancel_cancels_redirected_request(self):
+ first_request = self.create_s3_request()
+ second_request = self.create_s3_request()
+ self.coordinator.set_s3_request(first_request)
+ self.coordinator.set_s3_request(
+ second_request, is_region_redirect=True
+ )
+
+ self.coordinator.cancel()
+
+ second_request.cancel.assert_called_once_with()
+ first_request.cancel.assert_not_called()
+
+ def test_cancel_before_request_cancels_request(self):
+ self.coordinator.cancel()
+ s3_request = self.create_s3_request()
+
+ self.coordinator.set_s3_request(s3_request)
+
+ assert self.coordinator.cancelled
+ s3_request.cancel.assert_called_once_with()
+
+ def test_default_completion_future_can_complete_without_request(self):
+ coordinator = s3transfer.crt.CRTTransferCoordinator()
+
+ coordinator.complete()
+
+ assert coordinator.done()
+ assert coordinator.result() is None
+
+ def test_complete_resolves_completion_future(self):
+ self.coordinator.complete()
+ assert self.completion_future.done()
+ assert self.coordinator.result() is None
+
+ def test_complete_with_error(self):
+ self.coordinator.complete(CustomFutureException())
+ with pytest.raises(CustomFutureException):
+ self.coordinator.result()
+
+ def test_complete_is_idempotent(self):
+ self.coordinator.complete()
+ self.coordinator.complete(CustomFutureException())
+ assert self.coordinator.result() is None
+
+ def test_not_done_until_completed(self):
+ s3_request = self.create_s3_request()
+ self.coordinator.set_s3_request(s3_request)
+ # A native CRT request failing does not complete the transfer,
+ # since it may still be redirected to another region.
+ s3_request.finished_future.set_exception(CustomFutureException())
+
+ assert not self.coordinator.done()
+
+ self.coordinator.complete()
+ assert self.coordinator.done()
+
+
+@requires_crt_pytest
+class TestS3RegionRedirectPolicy:
+ def setup_method(self):
+ self.bucket = 'mybucket'
+ self.error = Exception('wrong region')
+ self.serializer = mock.Mock(
+ s3transfer.crt.BotocoreCRTRequestSerializer
+ )
+ self.serializer.get_cached_bucket_region.return_value = None
+ self.serializer.is_redirect_error.return_value = True
+ self.serializer.get_bucket_region.return_value = 'eu-central-1'
+ self.serializer.get_configured_region.return_value = 'us-west-2'
+ self.policy = s3transfer.crt.CRTS3RegionRedirectPolicy(self.serializer)
+
+ def is_error_redirect_candidate(self, **overrides):
+ kwargs = {
+ 'bucket': self.bucket,
+ 'transfer_type': 'put_object',
+ 'error': self.error,
+ 'is_region_redirect': False,
+ 'bytes_transferred': 0,
+ 'cancelled': False,
+ 'is_replayable': True,
+ }
+ kwargs.update(overrides)
+ return self.policy.is_error_redirect_candidate(**kwargs)
+
+ def get_retry_region(self, request_region=None):
+ return self.policy.get_retry_region(
+ self.bucket, 'put_object', self.error, request_region
+ )
+
+ def test_returns_and_caches_discovered_region(self):
+ assert self.get_retry_region() == 'eu-central-1'
+ self.serializer.cache_bucket_region.assert_called_once_with(
+ self.bucket, 'eu-central-1'
+ )
+
+ def test_returns_none_when_region_not_discovered(self):
+ self.serializer.get_bucket_region.return_value = None
+ assert self.get_retry_region() is None
+ self.serializer.cache_bucket_region.assert_not_called()
+
+ def test_returns_none_when_discovery_raises(self):
+ self.serializer.get_bucket_region.side_effect = InvalidRegionError(
+ region_name='not a region!'
+ )
+ assert self.get_retry_region() is None
+ self.serializer.cache_bucket_region.assert_not_called()
+
+ def test_returns_none_when_discovered_region_is_configured_region(self):
+ # Retrying in the region the request already used would fail the same
+ # way, and caching it would build a duplicate client for that region.
+ self.serializer.get_bucket_region.return_value = 'us-west-2'
+ assert self.get_retry_region() is None
+ self.serializer.cache_bucket_region.assert_not_called()
+
+ def test_returns_none_when_discovered_region_is_request_region(self):
+ self.serializer.get_bucket_region.return_value = 'eu-west-1'
+ assert self.get_retry_region(request_region='eu-west-1') is None
+ self.serializer.cache_bucket_region.assert_not_called()
+
+ def test_reuses_region_discovered_by_another_transfer(self):
+ # A transfer that failed in the configured region does not need to
+ # rediscover a region another transfer already cached.
+ self.serializer.get_cached_bucket_region.return_value = 'eu-west-1'
+ assert self.get_retry_region() == 'eu-west-1'
+ self.serializer.get_bucket_region.assert_not_called()
+
+ def test_rediscovers_region_when_cached_region_failed(self):
+ # The failed request already used the cached region, so the cache is
+ # stale and retrying there again would just fail the same way.
+ self.serializer.get_cached_bucket_region.return_value = 'eu-west-1'
+ assert self.get_retry_region(request_region='eu-west-1') == (
+ 'eu-central-1'
+ )
+ self.serializer.get_bucket_region.assert_called_once_with(
+ self.bucket, 'put_object', self.error
+ )
+
+ def test_is_candidate_for_failed_replayable_transfer(self):
+ assert self.is_error_redirect_candidate()
+
+ def test_not_candidate_after_redirect(self):
+ assert not self.is_error_redirect_candidate(is_region_redirect=True)
+
+ def test_not_candidate_after_bytes_transferred(self):
+ assert not self.is_error_redirect_candidate(bytes_transferred=1)
+
+ def test_not_candidate_when_cancelled(self):
+ assert not self.is_error_redirect_candidate(cancelled=True)
+
+ def test_not_candidate_when_stream_is_not_replayable(self):
+ assert not self.is_error_redirect_candidate(is_replayable=False)
+
+ def test_not_candidate_for_s3express_bucket(self):
+ assert not self.is_error_redirect_candidate(
+ bucket='mybucket--usw2-az5--x-s3'
+ )
+
+ def test_candidate_checks_do_not_discover_region(self):
+ self.is_error_redirect_candidate()
+ self.serializer.get_bucket_region.assert_not_called()
+
+ def test_not_candidate_when_error_is_not_a_redirect(self):
+ self.serializer.is_redirect_error.return_value = False
+ assert not self.is_error_redirect_candidate()
+
+ def test_error_is_classified_only_when_transfer_is_eligible(self):
+ # Classifying the error parses its response, so it is the last check
+ # rather than one every failed transfer pays for.
+ assert not self.is_error_redirect_candidate(cancelled=True)
+ self.serializer.is_redirect_error.assert_not_called()
+
+
@requires_crt()
class TestOnBodyFileObjWriter(unittest.TestCase):
def test_call(self):
@@ -438,3 +876,42 @@ def test_fio_options(
mock_s3_crt_client.call_args[1]['fio_options'].direct_io
is direct_io
)
+
+
+@requires_crt_pytest
+class TestRenameTempFileHandler:
+ @pytest.fixture
+ def coordinator(self):
+ return s3transfer.crt.CRTTransferCoordinator()
+
+ @pytest.fixture
+ def osutil(self):
+ return mock.Mock(spec=OSUtils)
+
+ @pytest.fixture
+ def handler(self, coordinator, osutil):
+ return s3transfer.crt.RenameTempFileHandler(
+ coordinator, 'final', 'temp', osutil
+ )
+
+ def test_renames_temp_file(self, handler, osutil):
+ handler(error=None)
+ osutil.rename_file.assert_called_once_with('temp', 'final')
+
+ def test_removes_temp_file_on_transfer_error(self, handler, osutil):
+ handler(error=Exception('transfer failed'))
+ osutil.remove_file.assert_called_once_with('temp')
+ assert not osutil.rename_file.called
+
+ def test_surfaces_rename_error(self, coordinator, handler, osutil):
+ osutil.rename_file.side_effect = OSError('Is a directory')
+ # The handler runs as an on done callback, so the transfer is already
+ # complete by the time the rename fails.
+ coordinator.complete()
+ assert coordinator.done()
+
+ handler(error=None)
+
+ osutil.remove_file.assert_called_once_with('temp')
+ with pytest.raises(OSError):
+ coordinator.result()