fix(aws provider): Allow configuring the load timeout for AWS credent…

…ials (#12422)

* fix(aws provider): Allow configuring the load timeout for AWS credentials

Technically an enhancement, but it allows working around a regression in
behavior (the default changing from 30s to 5s as part of the SDK
upgrade) so marking as a fix.

Closes: #12421

Signed-off-by: Jesse Szwedko <jesse@szwedko.me>

* Fix the rest of the usages

Signed-off-by: Jesse Szwedko <jesse@szwedko.me>

* Fix test

Signed-off-by: Jesse Szwedko <jesse@szwedko.me>

src/aws/auth.rs

@@ -3,6 +3,11 @@ use aws_config::{
 };
 use aws_types::{credentials::SharedCredentialsProvider, region::Region, Credentials};
 use serde::{Deserialize, Serialize};
+use std::time::Duration;
+
+// matches default load timeout from the SDK as of 0.10.1, but lets us confidently document the
+// default rather than relying on the SDK default to not change
+const DEFAULT_LOAD_TIMEOUT: Duration = Duration::from_secs(5);
 
 /// Configuration for configuring authentication strategy for AWS.
 #[derive(Serialize, Deserialize, Clone, Debug, Derivative)]
@@ -20,13 +25,12 @@ pub enum AwsAuthentication {
     },
     Role {
         assume_role: String,
+        load_timeout_secs: Option<u64>,
     },
     // Default variant is used instead of Option<AWSAuthentication> since even for
     // None we need to build `AwsCredentialsProvider`.
-    //
-    // {} is required to work around a bug in serde. https://github.com/serde-rs/serde/issues/1374
     #[derivative(Default)]
-    Default {},
+    Default { load_timeout_secs: Option<u64> },
 }
 
 impl AwsAuthentication {
@@ -46,15 +50,18 @@ impl AwsAuthentication {
             AwsAuthentication::File { .. } => {
                 Err("Overriding the credentials file is not supported.".into())
             }
-            AwsAuthentication::Role { assume_role } => {
+            AwsAuthentication::Role {
+                assume_role,
+                load_timeout_secs,
+            } => {
                 let provider = AssumeRoleProviderBuilder::new(assume_role)
                     .region(region.clone())
-                    .build(default_credentials_provider(region).await);
+                    .build(default_credentials_provider(region, *load_timeout_secs).await);
 
                 Ok(SharedCredentialsProvider::new(provider))
             }
-            AwsAuthentication::Default {} => Ok(SharedCredentialsProvider::new(
-                default_credentials_provider(region).await,
+            AwsAuthentication::Default { load_timeout_secs } => Ok(SharedCredentialsProvider::new(
+                default_credentials_provider(region, *load_timeout_secs).await,
             )),
         }
     }
@@ -68,13 +75,19 @@ impl AwsAuthentication {
     }
 }
 
-async fn default_credentials_provider(region: Region) -> SharedCredentialsProvider {
+async fn default_credentials_provider(
+    region: Region,
+    load_timeout_secs: Option<u64>,
+) -> SharedCredentialsProvider {
     let chain = DefaultCredentialsChain::builder()
         .region(region)
-        .build()
-        .await;
+        .load_timeout(
+            load_timeout_secs
+                .map(Duration::from_secs)
+                .unwrap_or(DEFAULT_LOAD_TIMEOUT),
+        );
 
-    SharedCredentialsProvider::new(chain)
+    SharedCredentialsProvider::new(chain.build().await)
 }
 
 #[cfg(test)]
@@ -96,7 +109,24 @@ mod tests {
         )
         .unwrap();
 
-        assert!(matches!(config.auth, AwsAuthentication::Default {}));
+        assert!(matches!(config.auth, AwsAuthentication::Default { .. }));
+    }
+
+    #[test]
+    fn parsing_default_with_load_timeout() {
+        let config = toml::from_str::<ComponentConfig>(
+            r#"
+            auth.load_timeout_secs = 10
+        "#,
+        )
+        .unwrap();
+
+        assert!(matches!(
+            config.auth,
+            AwsAuthentication::Default {
+                load_timeout_secs: Some(10)
+            }
+        ));
     }
 
     #[test]
@@ -108,14 +138,15 @@ mod tests {
         )
         .unwrap();
 
-        assert!(matches!(config.auth, AwsAuthentication::Default {}));
+        assert!(matches!(config.auth, AwsAuthentication::Default { .. }));
     }
 
     #[test]
     fn parsing_assume_role() {
         let config = toml::from_str::<ComponentConfig>(
             r#"
             auth.assume_role = "root"
+            auth.load_timeout_secs = 10
         "#,
         )
         .unwrap();
@@ -129,12 +160,19 @@ mod tests {
             r#"
             assume_role = "root"
             auth.assume_role = "auth.root"
+            auth.load_timeout_secs = 10
         "#,
         )
         .unwrap();
 
         match config.auth {
-            AwsAuthentication::Role { assume_role } => assert_eq!(&assume_role, "auth.root"),
+            AwsAuthentication::Role {
+                assume_role,
+                load_timeout_secs,
+            } => {
+                assert_eq!(&assume_role, "auth.root");
+                assert_eq!(load_timeout_secs, Some(10));
+            }
             _ => panic!(),
         }
     }

src/sinks/aws_kinesis_firehose/integration_tests.rs

@@ -78,7 +78,9 @@ async fn firehose_put_records() {
     components::SINK_TESTS.assert(&AWS_SINK_TAGS);
 
     let config = ElasticsearchConfig {
-        auth: Some(ElasticsearchAuth::Aws(AwsAuthentication::Default {})),
+        auth: Some(ElasticsearchAuth::Aws(AwsAuthentication::Default {
+            load_timeout_secs: Some(5),
+        })),
         endpoint: elasticsearch_address(),
         bulk: Some(BulkConfig {
             index: Some(stream.clone()),

src/sinks/elasticsearch/integration_tests.rs

@@ -241,7 +241,9 @@ async fn insert_events_on_aws() {
 
     run_insert_tests(
         ElasticsearchConfig {
-            auth: Some(ElasticsearchAuth::Aws(AwsAuthentication::Default {})),
+            auth: Some(ElasticsearchAuth::Aws(AwsAuthentication::Default {
+                load_timeout_secs: Some(5),
+            })),
             endpoint: aws_server(),
             aws: Some(RegionOrEndpoint::with_region(String::from("localstack"))),
             ..config()
@@ -258,7 +260,9 @@ async fn insert_events_on_aws_with_compression() {
 
     run_insert_tests(
         ElasticsearchConfig {
-            auth: Some(ElasticsearchAuth::Aws(AwsAuthentication::Default {})),
+            auth: Some(ElasticsearchAuth::Aws(AwsAuthentication::Default {
+                load_timeout_secs: Some(5),
+            })),
             endpoint: aws_server(),
             aws: Some(RegionOrEndpoint::with_region(String::from("localstack"))),
             compression: Compression::gzip_default(),

website/cue/reference/components/aws.cue

@@ -39,6 +39,17 @@ components: _aws: {
 							examples: ["arn:aws:iam::123456789098:role/my_role"]
 						}
 					}
+					load_timeout_secs: {
+						category:    "Auth"
+						common:      false
+						description: "The timeout for loading credentials. Relevant when the default credentials chain is used or `assume_role`."
+						required:    false
+						type: uint: {
+							unit:    "seconds"
+							default: 5
+							examples: [30]
+						}
+					}
 					profile: {
 						category:    "Auth"
 						common:      false