From c45e38d248fdc7e0caadd86e68977802184c6080 Mon Sep 17 00:00:00 2001 From: Steve Yoo Date: Thu, 10 Sep 2026 16:03:05 -0400 Subject: [PATCH] Route auto resolve to classic for unsupported workloads --- awscli/customizations/s3/factory.py | 57 +++++++- tests/unit/customizations/s3/test_factory.py | 135 +++++++++++++++++++ 2 files changed, 185 insertions(+), 7 deletions(-) diff --git a/awscli/customizations/s3/factory.py b/awscli/customizations/s3/factory.py index b866f7ec7894..6deb79740eeb 100644 --- a/awscli/customizations/s3/factory.py +++ b/awscli/customizations/s3/factory.py @@ -11,6 +11,7 @@ # 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 awscrt.s3 from botocore.client import Config @@ -33,6 +34,8 @@ LOGGER = logging.getLogger(__name__) +ADAPTIVE_RETRY_MODE = 'adaptive' + CRT_CLIENT_KWARG_MAP = { 'multipart_chunksize': 'part_size', 'multipart_threshold': 'multipart_upload_threshold', @@ -88,16 +91,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', @@ -110,6 +111,48 @@ 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)}' + ) + 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 = [] + if runtime_config.is_explicitly_set('max_bandwidth'): + unsupported.append('max_bandwidth') + if ( + self._session.get_config_variable('retry_mode') + == 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') + return unsupported + + 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 diff --git a/tests/unit/customizations/s3/test_factory.py b/tests/unit/customizations/s3/test_factory.py index 69c7a2c10118..f23ae2155876 100644 --- a/tests/unit/customizations/s3/test_factory.py +++ b/tests/unit/customizations/s3/test_factory.py @@ -702,6 +702,141 @@ def test_part_size_configured_when_matching_default(self, mock_crt_client): ) +@pytest.fixture +def auto_resolve_session(): + session = mock.Mock(Session) + session.get_config_variable.return_value = None + 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_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 + ): + auto_resolve_session.get_config_variable.return_value = '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 + ): + auto_resolve_session.get_config_variable.return_value = '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 + + @pytest.mark.parametrize( 'preferred_transfer_client,extra_params,' 'crt_is_optimized_for_system,crt_running_in_other_process,'