Route auto resolve to classic for unsupported workloads

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

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,'