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/utils.py b/awscli/botocore/utils.py
index a3ac03e745e8..8a43ecaac75a 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,49 +1616,9 @@ 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,
- ]
- ):
- return
-
- bucket = request_dict['context']['s3_redirect']['bucket']
- client_region = request_dict['context'].get('client_region')
- new_region = self.get_bucket_region(bucket, response)
+ bucket = redirect_ctx.get('bucket')
+ client_region = request_dict.get('context', {}).get('client_region')
+ new_region = self.get_redirect_region(bucket, response, operation)
if new_region is None:
logger.debug(
@@ -1702,6 +1662,52 @@ 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):
+ if bucket is None:
+ return None
+ if ArnParser.is_arn(bucket):
+ return None
+ 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,
+ ]
+ ):
+ return None
+ return self.get_bucket_region(bucket, response)
+
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..5a618a121f93 100644
--- a/awscli/customizations/s3/factory.py
+++ b/awscli/customizations/s3/factory.py
@@ -20,6 +20,7 @@
BotocoreCRTRequestSerializer,
CRTTransferManager,
acquire_crt_s3_process_lock,
+ create_crt_client_bootstrap,
create_s3_crt_client,
)
from s3transfer.manager import TransferManager
@@ -114,16 +115,34 @@ 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()
+ transfer_manager = CRTTransferManager(
+ crt_client_factory=lambda client_region=None: (
+ self._create_crt_client(
+ params,
+ runtime_config,
+ region=client_region or region,
+ bootstrap=bootstrap,
+ )
+ ),
+ crt_request_serializer=self._create_crt_request_serializer(params),
)
-
- def _create_crt_client(self, params, runtime_config):
+ # 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_client(
+ self, params, runtime_config, region=None, bootstrap=None
+ ):
config_file_params = self._session.get_scoped_config().get('s3', {})
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':
@@ -161,6 +180,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/s3transfer/crt.py b/awscli/s3transfer/crt.py
index ae4e5195a4b2..6d9d25622bb7 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
@@ -44,7 +46,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
@@ -89,6 +97,7 @@ def create_s3_crt_client(
use_ssl=True,
verify=None,
fio_options=None,
+ bootstrap=None,
):
"""
:type region: str
@@ -135,11 +144,14 @@ 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.
"""
- 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 = (
@@ -183,6 +195,12 @@ def create_s3_crt_client(
)
+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()
@@ -201,17 +219,181 @@ 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
+ # Bucket regions are held under one lock 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, and so that a
+ # request cannot be built for a region that changes while it is built.
+ # One lock covers every bucket because a command generally transfers
+ # to or from a single one.
+ self._region_lock = threading.Lock()
+
+ def get_cached_bucket_region(self, bucket):
+ """Return a region already discovered for a bucket, if any."""
+ return self._crt_request_serializer.get_cached_bucket_region(bucket)
+
+ @contextlib.contextmanager
+ def locked_bucket_region(self, bucket):
+ """Hold a bucket's region steady while a request is built for it.
+
+ Serializing a request resolves its endpoint from the region cached for
+ its bucket. The caller has to select a client for that same region, so
+ the region must not change in between.
+ """
+ with self._region_lock:
+ yield self.get_cached_bucket_region(bucket)
+
+ def is_error_redirect_candidate(
+ self,
+ bucket,
+ is_region_redirect,
+ bytes_transferred,
+ cancelled,
+ is_replayable,
+ ):
+ """Return whether a failed request may be worth redirecting.
+
+ Callers only ask this about a request that failed, so this inspects
+ the state of the transfer rather than the error itself. These checks
+ are cheap and never do I/O, so a caller running on a CRT completion
+ thread can use them to decide whether discovering a region is worth
+ handing off to another thread.
+ """
+ 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
+ return True
+
+ 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._region_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.get_cached_bucket_region(bucket)
+ if cached_region is None:
+ return None
+ if cached_region == request_region:
+ # The failed request already used this region, so the cached
+ # region is stale and retrying there would fail the same way.
+ return None
+ return cached_region
+
+ def _discover_bucket_region(
+ self, bucket, transfer_type, error, request_region=None
+ ):
+ 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 CRTTransferManager:
- def __init__(self, crt_s3_client, crt_request_serializer, osutil=None):
+ def __init__(
+ self, crt_client_factory, crt_request_serializer, osutil=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
@@ -219,13 +401,18 @@ def __init__(self, crt_s3_client, crt_request_serializer, osutil=None):
"""
if osutil is None:
self._osutil = OSUtils()
- self._crt_s3_client = crt_s3_client
self._s3_args_creator = S3ClientArgsCreator(
crt_request_serializer, self._osutil
)
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.
@@ -327,12 +514,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),
@@ -341,6 +541,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()
@@ -348,22 +564,45 @@ 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)
+ policy = self._region_redirect_policy
+ with policy.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
@@ -447,9 +686,55 @@ 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."""
+
+ 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
@@ -460,12 +745,43 @@ 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'
+ region_redirect_client_factory = lambda: 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)
+
+ # Share the bucket-region cache between redirect discovery and later
+ # request serialization. The real fallback client remains lazy.
+ self._region_redirect_client = LazyHeadBucketClient(
+ region_redirect_client_factory
+ )
+ self._region_cache = {}
+ 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
)
@@ -478,6 +794,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,
+ )
def _resolve_client_config(self, session, client_kwargs):
user_provided_config = None
@@ -612,6 +932,48 @@ def _remove_checksum_context(self, params, **kwargs):
if "checksum" in request_context:
del request_context["checksum"]
+ def cache_bucket_region(self, bucket, region):
+ self._region_cache[bucket] = region
+
+ def get_cached_bucket_region(self, bucket):
+ return self._region_cache.get(bucket)
+
+ def get_configured_region(self):
+ return self._client.meta.region_name
+
+ 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.
+ """
+ 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''),
+ )
+ response = (http_response, translated_error.response)
+ # 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,
+ response,
+ operation,
+ )
+
class FakeRawResponse(BytesIO):
def stream(self, amt=1024, decode_content=None):
@@ -642,18 +1004,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):
@@ -671,17 +1051,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)
@@ -705,16 +1093,172 @@ 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(
+ 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, is_region_redirect, bytes_transferred):
+ try:
+ return self._region_redirect_policy.is_error_redirect_candidate(
+ bucket=self._bucket,
+ 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
+ if error is None:
+ completion_future.set_result(None)
+ else:
+ completion_future.set_exception(error)
class S3ClientArgsCreator:
+ _DOWNLOAD_TEMP_FILENAME = '_crt_download_temp_filename'
+
def __init__(self, crt_request_serializer, os_utils):
self._request_serializer = crt_request_serializer
self._os_utils = os_utils
@@ -779,15 +1323,23 @@ def _get_make_request_args_put_object(
call_args.extra_args["Body"] = call_args.fileobj
checksum_config = None
+ provided_checksum_algorithm = None
if not any(
checksum_arg in call_args.extra_args
for checksum_arg in FULL_OBJECT_CHECKSUM_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
@@ -796,14 +1348,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
@@ -822,7 +1380,21 @@ def _get_make_request_args_get_object(
checksum_config = awscrt.s3.S3ChecksumConfig(validate_response=True)
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
diff --git a/awscli/topics/s3-config.rst b/awscli/topics/s3-config.rst
index defa8b2da5ec..47ce57dea195 100644
--- a/awscli/topics/s3-config.rst
+++ b/awscli/topics/s3-config.rst
@@ -376,8 +376,12 @@ 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
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 d092982f082c..64053183eccb 100644
--- a/tests/functional/s3transfer/test_crt.py
+++ b/tests/functional/s3transfer/test_crt.py
@@ -16,6 +16,7 @@
import time
from concurrent.futures import Future
+from botocore.exceptions import ClientError
from botocore.session import Session
from s3transfer.subscribers import BaseSubscriber
@@ -33,6 +34,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 +54,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 +106,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()
@@ -218,7 +229,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(
@@ -236,6 +247,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]
@@ -262,6 +363,305 @@ 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_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(
@@ -667,13 +1067,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:
@@ -701,7 +1102,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(
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/customizations/s3/test_factory.py b/tests/unit/customizations/s3/test_factory.py
index 9fafd4a9d588..23c66580934d 100644
--- a/tests/unit/customizations/s3/test_factory.py
+++ b/tests/unit/customizations/s3/test_factory.py
@@ -16,6 +16,7 @@
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
@@ -284,6 +285,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
@@ -392,6 +424,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(
diff --git a/tests/unit/s3transfer/test_crt.py b/tests/unit/s3transfer/test_crt.py
index 740131c8be81..01362f0ed778 100644
--- a/tests/unit/s3transfer/test_crt.py
+++ b/tests/unit/s3transfer/test_crt.py
@@ -11,12 +11,14 @@
# ANY KIND, either express or implied. See the License for the specific
# language governing permissions and limitations under the License.
import io
+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
@@ -169,14 +171,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 +184,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 +235,98 @@ def test_translate_unknown_operation_404(self):
boto_err = self.request_serializer.translate_crt_exception(crt_exc)
self.assertIsInstance(boto_err, ClientError)
+ def test_cached_bucket_region_changes_serialized_endpoint(self):
+ self.request_serializer.cache_bucket_region(
+ self.bucket, 'eu-central-1'
+ )
+ 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
+ )
+
+ crt_request = self.request_serializer.serialize_http_request(
+ "get_object", future
+ )
+
+ self.assertEqual(
+ crt_request.headers.get("host"),
+ "s3.eu-central-1.amazonaws.com",
+ )
+
+ 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 +406,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 +431,197 @@ 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.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,
+ '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_get_cached_bucket_region(self):
+ self.serializer.get_cached_bucket_region.return_value = 'eu-west-1'
+ region = self.policy.get_cached_bucket_region(self.bucket)
+ assert region == 'eu-west-1'
+
+
@requires_crt()
class TestOnBodyFileObjWriter(unittest.TestCase):
def test_call(self):