Create CredentialsCache and DefaultResolver only once in test/bench (#2620)

## Motivation and Context
Addresses
#2593 (comment)
and
#2593 (comment).

Also ports the changes made in #2592.

## Description
Prior to this PR, `sra_manual_test` and the bench
`middleware_vs_orchestrator` created `CredentialsCache` and
`aws_sdk_s3::endpoint::DefaultResolver` every time a request for
`ListObjectsV2` was dispatched. This is not the case in production where
those two types are created once during the construction of a service
config
([here](https://github.com/awslabs/aws-sdk-rust/blob/main/sdk/s3/src/config.rs#L623-L625)
and
[here](https://github.com/awslabs/aws-sdk-rust/blob/main/sdk/s3/src/config.rs#L635-L652))
and reused for subsequent request dispatches.

The PR will make `sra_manual_test` and `middleware_vs_orchestrator` do
the same, creating `CredentialsCache` and
`aws_sdk_s3::endpoint::DefaultResolver` only once before the
test/benchmark starts running and reusing them thereafter.

The change will help bring the performance for `orchestrator` closer to
that for `middleware`.

- In the `main` branch with `DefaultEndpointResolver` commented in and
`StaticUriEndpointResolver` commented out:
```
middleware              time:   [20.924 µs 20.943 µs 20.964 µs]
                        change: [-1.0107% -0.7357% -0.4827%] (p = 0.00 < 0.05)
                        Change within noise threshold.
Found 7 outliers among 100 measurements (7.00%)
  1 (1.00%) high mild
  6 (6.00%) high severe

orchestrator            time:   [933.68 µs 940.11 µs 945.82 µs]
                        change: [+2735.7% +2754.5% +2770.9%] (p = 0.00 < 0.05)
                        Performance has regressed.
Found 17 outliers among 100 measurements (17.00%)
  14 (14.00%) low mild
  2 (2.00%) high mild
  1 (1.00%) high severe
```

- With the change in this PR:
```
middleware              time:   [21.161 µs 21.194 µs 21.232 µs]
                        change: [-0.8973% -0.6397% -0.3758%] (p = 0.00 < 0.05)
                        Change within noise threshold.
Found 8 outliers among 100 measurements (8.00%)
  5 (5.00%) high mild
  3 (3.00%) high severe

orchestrator            time:   [56.038 µs 56.182 µs 56.349 µs]
                        change: [-0.7921% -0.5552% -0.3157%] (p = 0.00 < 0.05)
                        Change within noise threshold.
Found 5 outliers among 100 measurements (5.00%)
  2 (2.00%) high mild
  3 (3.00%) high severe

```


## Testing
Executed the following without any errors:
```
➜  smithy-rs git:(ysaito/create-creds-cache-and-ep-resolver-only-once) ✗ ./gradlew aws:sra-test:assemble
➜  aws-sdk-s3 git:(ysaito/create-creds-cache-and-ep-resolver-only-once) cargo t
➜  aws-sdk-s3 git:(ysaito/create-creds-cache-and-ep-resolver-only-once) cargo bench
```

----

_By submitting this pull request, I confirm that you can use, modify,
copy, and redistribute this contribution, under the terms of your
choice._

---------

Co-authored-by: Yuki Saito <awsaito@amazon.com>

Author: ysaito1001

aws/sra-test/integration-tests/aws-sdk-s3/benches/middleware_vs_orchestrator.rs

@@ -5,9 +5,13 @@
 
 #[macro_use]
 extern crate criterion;
+use aws_credential_types::cache::{CredentialsCache, SharedCredentialsCache};
+use aws_credential_types::provider::SharedCredentialsProvider;
+use aws_credential_types::Credentials;
 use aws_sdk_s3 as s3;
 use aws_smithy_client::erase::DynConnector;
 use aws_smithy_client::test_connection::infallible_connection_fn;
+use aws_smithy_http::endpoint::SharedEndpointResolver;
 use aws_smithy_runtime_api::type_erasure::TypedBox;
 use criterion::Criterion;
 use s3::operation::list_objects_v2::{ListObjectsV2Error, ListObjectsV2Input, ListObjectsV2Output};
@@ -22,10 +26,20 @@ async fn middleware(client: &s3::Client) {
         .expect("successful execution");
 }
 
-async fn orchestrator(connector: &DynConnector) {
+async fn orchestrator(
+    connector: &DynConnector,
+    endpoint_resolver: SharedEndpointResolver<s3::endpoint::Params>,
+    credentials_cache: SharedCredentialsCache,
+) {
+    let service_runtime_plugin = orchestrator::ManualServiceRuntimePlugin {
+        connector: connector.clone(),
+        endpoint_resolver: endpoint_resolver.clone(),
+        credentials_cache: credentials_cache.clone(),
+    };
+
     // TODO(enableNewSmithyRuntime): benchmark with `send_v2` directly once it works
     let runtime_plugins = aws_smithy_runtime_api::client::runtime_plugin::RuntimePlugins::new()
-        .with_client_plugin(orchestrator::ManualServiceRuntimePlugin(connector.clone()))
+        .with_client_plugin(service_runtime_plugin)
         .with_operation_plugin(aws_sdk_s3::operation::list_objects_v2::ListObjectsV2::new())
         .with_operation_plugin(orchestrator::ManualOperationRuntimePlugin);
     let input = ListObjectsV2Input::builder()
@@ -95,25 +109,32 @@ fn middleware_bench(c: &mut Criterion) {
 
 fn orchestrator_bench(c: &mut Criterion) {
     let conn = test_connection();
+    let endpoint_resolver = SharedEndpointResolver::new(s3::endpoint::DefaultResolver::new());
+    let credentials_cache = SharedCredentialsCache::new(
+        CredentialsCache::lazy()
+            .create_cache(SharedCredentialsProvider::new(Credentials::for_tests())),
+    );
 
     c.bench_function("orchestrator", move |b| {
         b.to_async(tokio::runtime::Runtime::new().unwrap())
-            .iter(|| async { orchestrator(&conn).await })
+            .iter(|| async {
+                orchestrator(&conn, endpoint_resolver.clone(), credentials_cache.clone()).await
+            })
     });
 }
 
 mod orchestrator {
-    use aws_credential_types::cache::{CredentialsCache, SharedCredentialsCache};
-    use aws_credential_types::provider::SharedCredentialsProvider;
-    use aws_credential_types::Credentials;
+    use aws_credential_types::cache::SharedCredentialsCache;
     use aws_http::user_agent::{ApiMetadata, AwsUserAgent};
     use aws_runtime::recursion_detection::RecursionDetectionInterceptor;
     use aws_runtime::user_agent::UserAgentInterceptor;
     use aws_sdk_s3::config::Region;
+    use aws_sdk_s3::endpoint::Params;
     use aws_sdk_s3::operation::list_objects_v2::ListObjectsV2Input;
     use aws_smithy_client::erase::DynConnector;
+    use aws_smithy_http::endpoint::SharedEndpointResolver;
     use aws_smithy_runtime::client::connections::adapter::DynConnectorAdapter;
-    use aws_smithy_runtime_api::client::endpoints::StaticUriEndpointResolver;
+    use aws_smithy_runtime::client::orchestrator::endpoints::DefaultEndpointResolver;
     use aws_smithy_runtime_api::client::interceptors::{
         Interceptor, InterceptorContext, InterceptorError, Interceptors,
     };
@@ -124,10 +145,13 @@ mod orchestrator {
     use aws_smithy_runtime_api::config_bag::ConfigBag;
     use aws_types::region::SigningRegion;
     use aws_types::SigningService;
-    use http::Uri;
     use std::sync::Arc;
 
-    pub struct ManualServiceRuntimePlugin(pub DynConnector);
+    pub struct ManualServiceRuntimePlugin {
+        pub connector: DynConnector,
+        pub endpoint_resolver: SharedEndpointResolver<Params>,
+        pub credentials_cache: SharedCredentialsCache,
+    }
 
     impl RuntimePlugin for ManualServiceRuntimePlugin {
         fn configure(&self, cfg: &mut ConfigBag) -> Result<(), BoxError> {
@@ -136,9 +160,7 @@ mod orchestrator {
                     .identity_resolver(
                         aws_runtime::auth::sigv4::SCHEME_ID,
                         aws_runtime::identity::credentials::CredentialsIdentityResolver::new(
-                            SharedCredentialsCache::new(CredentialsCache::lazy().create_cache(
-                                SharedCredentialsProvider::new(Credentials::for_tests()),
-                            )),
+                            self.credentials_cache.clone(),
                         ),
                     )
                     .identity_resolver(
@@ -163,14 +185,7 @@ mod orchestrator {
                 ),
             );
 
-            //cfg.set_endpoint_resolver(DefaultEndpointResolver::new(
-            //    aws_smithy_http::endpoint::SharedEndpointResolver::new(
-            //        aws_sdk_s3::endpoint::DefaultResolver::new(),
-            //    ),
-            //));
-            cfg.set_endpoint_resolver(StaticUriEndpointResolver::uri(Uri::from_static(
-                "https://test-bucket.s3.us-east-1.amazonaws.com/",
-            )));
+            cfg.set_endpoint_resolver(DefaultEndpointResolver::new(self.endpoint_resolver.clone()));
 
             let params_builder = aws_sdk_s3::endpoint::Params::builder()
                 .set_region(Some("us-east-1".to_owned()))
@@ -182,7 +197,7 @@ mod orchestrator {
             );
 
             let connection: Box<dyn Connection> =
-                Box::new(DynConnectorAdapter::new(self.0.clone()));
+                Box::new(DynConnectorAdapter::new(self.connector.clone()));
             cfg.set_connection(connection);
 
             cfg.set_trace_probe({
@@ -227,12 +242,10 @@ mod orchestrator {
                     let input = context.input()?;
                     let input = input
                         .downcast_ref::<ListObjectsV2Input>()
-                        .ok_or_else(|| InterceptorError::invalid_input_access())?;
+                        .ok_or_else(|| "failed to downcast to ListObjectsV2Input")?;
                     let mut params_builder = cfg
                         .get::<aws_sdk_s3::endpoint::ParamsBuilder>()
-                        .ok_or(InterceptorError::read_before_execution(
-                            "missing endpoint params builder",
-                        ))?
+                        .ok_or_else(|| "missing endpoint params builder")?
                         .clone();
                     params_builder = params_builder.set_bucket(input.bucket.clone());
                     cfg.put(params_builder);
@@ -251,13 +264,11 @@ mod orchestrator {
                 ) -> Result<(), BoxError> {
                     let params_builder = cfg
                         .get::<aws_sdk_s3::endpoint::ParamsBuilder>()
-                        .ok_or(InterceptorError::read_before_execution(
-                            "missing endpoint params builder",
-                        ))?
+                        .ok_or_else(|| "missing endpoint params builder")?
                         .clone();
-                    let params = params_builder
-                        .build()
-                        .map_err(InterceptorError::read_before_execution)?;
+                    let params = params_builder.build().map_err(|err| {
+                        ContextAttachedError::new("endpoint params could not be built", err)
+                    })?;
                     cfg.put(
                         aws_smithy_runtime_api::client::orchestrator::EndpointResolverParams::new(
                             params,

aws/sra-test/integration-tests/aws-sdk-s3/tests/sra_test.rs

@@ -10,17 +10,18 @@ use aws_runtime::auth::sigv4::SigV4OperationSigningConfig;
 use aws_runtime::recursion_detection::RecursionDetectionInterceptor;
 use aws_runtime::user_agent::UserAgentInterceptor;
 use aws_sdk_s3::config::{Credentials, Region};
+use aws_sdk_s3::endpoint::Params;
 use aws_sdk_s3::operation::list_objects_v2::{
     ListObjectsV2Error, ListObjectsV2Input, ListObjectsV2Output,
 };
 use aws_sdk_s3::primitives::SdkBody;
 use aws_smithy_client::erase::DynConnector;
 use aws_smithy_client::test_connection::TestConnection;
+use aws_smithy_http::endpoint::SharedEndpointResolver;
 use aws_smithy_runtime::client::connections::adapter::DynConnectorAdapter;
 use aws_smithy_runtime::client::orchestrator::endpoints::DefaultEndpointResolver;
-use aws_smithy_runtime_api::client::interceptors::{
-    Interceptor, InterceptorContext, InterceptorError, Interceptors,
-};
+use aws_smithy_runtime_api::client::interceptors::error::ContextAttachedError;
+use aws_smithy_runtime_api::client::interceptors::{Interceptor, InterceptorContext, Interceptors};
 use aws_smithy_runtime_api::client::orchestrator::{
     BoxError, ConfigBagAccessors, Connection, HttpRequest, HttpResponse, TraceProbe,
 };
@@ -74,7 +75,11 @@ async fn sra_test() {
 async fn sra_manual_test() {
     tracing_subscriber::fmt::init();
 
-    struct ManualServiceRuntimePlugin(TestConnection<&'static str>);
+    struct ManualServiceRuntimePlugin {
+        connector: TestConnection<&'static str>,
+        endpoint_resolver: SharedEndpointResolver<Params>,
+        credentials_cache: SharedCredentialsCache,
+    }
 
     impl RuntimePlugin for ManualServiceRuntimePlugin {
         fn configure(&self, cfg: &mut ConfigBag) -> Result<(), BoxError> {
@@ -83,9 +88,7 @@ async fn sra_manual_test() {
                     .identity_resolver(
                         aws_runtime::auth::sigv4::SCHEME_ID,
                         aws_runtime::identity::credentials::CredentialsIdentityResolver::new(
-                            SharedCredentialsCache::new(CredentialsCache::lazy().create_cache(
-                                SharedCredentialsProvider::new(Credentials::for_tests()),
-                            )),
+                            self.credentials_cache.clone(),
                         ),
                     )
                     .identity_resolver(
@@ -110,13 +113,9 @@ async fn sra_manual_test() {
                 ),
             );
 
-            cfg.set_endpoint_resolver(DefaultEndpointResolver::new(
-                aws_smithy_http::endpoint::SharedEndpointResolver::new(
-                    aws_sdk_s3::endpoint::DefaultResolver::new(),
-                ),
-            ));
+            cfg.set_endpoint_resolver(DefaultEndpointResolver::new(self.endpoint_resolver.clone()));
 
-            let params_builder = aws_sdk_s3::endpoint::Params::builder()
+            let params_builder = Params::builder()
                 .set_region(Some("us-east-1".to_owned()))
                 .set_endpoint(Some("https://s3.us-east-1.amazonaws.com/".to_owned()));
             cfg.put(params_builder);
@@ -125,8 +124,9 @@ async fn sra_manual_test() {
                 aws_smithy_runtime_api::client::retries::NeverRetryStrategy::new(),
             );
 
-            let connection: Box<dyn Connection> =
-                Box::new(DynConnectorAdapter::new(DynConnector::new(self.0.clone())));
+            let connection: Box<dyn Connection> = Box::new(DynConnectorAdapter::new(
+                DynConnector::new(self.connector.clone()),
+            ));
             cfg.set_connection(connection);
 
             cfg.set_trace_probe({
@@ -189,12 +189,10 @@ async fn sra_manual_test() {
                     let input = context.input()?;
                     let input = input
                         .downcast_ref::<ListObjectsV2Input>()
-                        .ok_or_else(|| InterceptorError::invalid_input_access())?;
+                        .ok_or_else(|| "failed to downcast to ListObjectsV2Input")?;
                     let mut params_builder = cfg
                         .get::<aws_sdk_s3::endpoint::ParamsBuilder>()
-                        .ok_or(InterceptorError::read_before_execution(
-                            "missing endpoint params builder",
-                        ))?
+                        .ok_or_else(|| "missing endpoint params builder")?
                         .clone();
                     params_builder = params_builder.set_bucket(input.bucket.clone());
                     cfg.put(params_builder);
@@ -213,13 +211,11 @@ async fn sra_manual_test() {
                 ) -> Result<(), BoxError> {
                     let params_builder = cfg
                         .get::<aws_sdk_s3::endpoint::ParamsBuilder>()
-                        .ok_or(InterceptorError::read_before_execution(
-                            "missing endpoint params builder",
-                        ))?
+                        .ok_or_else(|| "missing endpoint params builder")?
                         .clone();
-                    let params = params_builder
-                        .build()
-                        .map_err(InterceptorError::read_before_execution)?;
+                    let params = params_builder.build().map_err(|err| {
+                        ContextAttachedError::new("endpoint params could not be built", err)
+                    })?;
                     cfg.put(
                         aws_smithy_runtime_api::client::orchestrator::EndpointResolverParams::new(
                             params,
@@ -265,8 +261,21 @@ async fn sra_manual_test() {
 "#).unwrap(),
             )]);
 
+    let endpoint_resolver =
+        SharedEndpointResolver::new(aws_sdk_s3::endpoint::DefaultResolver::new());
+    let credentials_cache = SharedCredentialsCache::new(
+        CredentialsCache::lazy()
+            .create_cache(SharedCredentialsProvider::new(Credentials::for_tests())),
+    );
+
+    let service_runtime_plugin = ManualServiceRuntimePlugin {
+        connector: conn.clone(),
+        endpoint_resolver,
+        credentials_cache,
+    };
+
     let runtime_plugins = aws_smithy_runtime_api::client::runtime_plugin::RuntimePlugins::new()
-        .with_client_plugin(ManualServiceRuntimePlugin(conn.clone()))
+        .with_client_plugin(service_runtime_plugin)
         .with_operation_plugin(aws_sdk_s3::operation::list_objects_v2::ListObjectsV2::new())
         .with_operation_plugin(ManualOperationRuntimePlugin);