fix(aws_S3 source): Add timeout for aws operations.

changelog.d/aws_timeout_all_components.enhancement.md

@@ -0,0 +1,5 @@
+All AWS-backed components now bound their AWS API requests with client timeouts (`connect_timeout_seconds = 5`, `read_timeout_seconds = 30` by default), exposed as configurable `connect_timeout_seconds`, `operation_timeout_seconds`, and `read_timeout_seconds` options.
+
+The options are now available on the `aws_cloudwatch_logs`, `aws_cloudwatch_metrics`, `aws_kinesis_firehose`, `aws_kinesis_streams`, `aws_s3`, `aws_sns`, and `aws_sqs` sinks, the `aws_s3` and `aws_sqs` sources, and the `aws_secrets_manager` secrets backend (as `client_timeout`, to avoid colliding with the unrelated `timeout` option on the `exec` secrets backend).
+
+authors: petere-datadog

src/aws/mod.rs

@@ -172,7 +172,7 @@ pub async fn create_client<T>(
     endpoint: Option<String>,
     proxy: &ProxyConfig,
     tls_options: Option<&TlsConfig>,
-    timeout: Option<&AwsTimeout>,
+    timeout: &AwsTimeout,
 ) -> crate::Result<T::Client>
 where
     T: ClientBuilder,
@@ -190,7 +190,7 @@ pub async fn create_client_and_region<T>(
     endpoint: Option<String>,
     proxy: &ProxyConfig,
     tls_options: Option<&TlsConfig>,
-    timeout: Option<&AwsTimeout>,
+    timeout: &AwsTimeout,
 ) -> crate::Result<(T::Client, Region)>
 where
     T: ClientBuilder,
@@ -238,20 +238,18 @@ where
         config_builder = config_builder.use_fips(use_fips);
     }
 
-    if let Some(timeout) = timeout {
-        let mut timeout_config_builder = TimeoutConfig::builder();
+    let mut timeout_config_builder = TimeoutConfig::builder();
 
-        let operation_timeout = timeout.operation_timeout();
-        let connect_timeout = timeout.connect_timeout();
-        let read_timeout = timeout.read_timeout();
+    let operation_timeout = timeout.operation_timeout();
+    let connect_timeout = timeout.connect_timeout();
+    let read_timeout = timeout.read_timeout();
 
-        timeout_config_builder
-            .set_operation_timeout(operation_timeout.map(Duration::from_secs))
-            .set_connect_timeout(connect_timeout.map(Duration::from_secs))
-            .set_read_timeout(read_timeout.map(Duration::from_secs));
+    timeout_config_builder
+        .set_operation_timeout(operation_timeout.map(Duration::from_secs))
+        .set_connect_timeout(Some(Duration::from_secs(connect_timeout)))
+        .set_read_timeout(Some(Duration::from_secs(read_timeout)));
 
-        config_builder = config_builder.timeout_config(timeout_config_builder.build());
-    }
+    config_builder = config_builder.timeout_config(timeout_config_builder.build());
 
     let config = config_builder.build();
 

src/aws/timeout.rs

@@ -2,6 +2,12 @@
 //use std::time::Duration;
 use serde_with::serde_as;
 use vector_lib::configurable::configurable_component;
+const fn default_aws_connect_timeout_seconds() -> u64 {
+    5
+}
+const fn default_aws_read_timeout_seconds() -> u64 {
+    30
+}
 
 /// Client timeout configuration for AWS operations.
 #[serde_as]
@@ -16,9 +22,10 @@ pub struct AwsTimeout {
     #[configurable(metadata(docs::examples = 20))]
     #[configurable(metadata(docs::human_name = "Connect Timeout"))]
     #[configurable(metadata(docs::type_unit = "seconds"))]
-    #[serde(skip_serializing_if = "Option::is_none")]
     #[serde(rename = "connect_timeout_seconds")]
-    connect_timeout: Option<u64>,
+    #[serde(default = "default_aws_connect_timeout_seconds")]
+    #[derivative(Default(value = "default_aws_connect_timeout_seconds()"))]
+    connect_timeout: u64,
 
     /// The operation timeout for AWS requests
     ///
@@ -41,14 +48,15 @@ pub struct AwsTimeout {
     #[configurable(metadata(docs::examples = 20))]
     #[configurable(metadata(docs::human_name = "Read Timeout"))]
     #[configurable(metadata(docs::type_unit = "seconds"))]
-    #[serde(skip_serializing_if = "Option::is_none")]
     #[serde(rename = "read_timeout_seconds")]
-    read_timeout: Option<u64>,
+    #[serde(default = "default_aws_read_timeout_seconds")]
+    #[derivative(Default(value = "default_aws_read_timeout_seconds()"))]
+    read_timeout: u64,
 }
 
 impl AwsTimeout {
     /// returns the connection timeout
-    pub const fn connect_timeout(&self) -> Option<u64> {
+    pub const fn connect_timeout(&self) -> u64 {
         self.connect_timeout
     }
 
@@ -58,7 +66,7 @@ impl AwsTimeout {
     }
 
     /// returns the read timeout
-    pub const fn read_timeout(&self) -> Option<u64> {
+    pub const fn read_timeout(&self) -> u64 {
         self.read_timeout
     }
 }
@@ -76,8 +84,20 @@ mod tests {
         "})
         .unwrap();
 
-        assert_eq!(config.connect_timeout, Some(20));
+        assert_eq!(config.connect_timeout, 20);
         assert_eq!(config.operation_timeout, Some(20));
-        assert_eq!(config.read_timeout, Some(60));
+        assert_eq!(config.read_timeout, 60);
+    }
+
+    #[test]
+    fn default_matches_serde_defaults() {
+        let default = AwsTimeout::default();
+
+        assert_eq!(
+            default.connect_timeout,
+            default_aws_connect_timeout_seconds()
+        );
+        assert_eq!(default.operation_timeout, None);
+        assert_eq!(default.read_timeout, default_aws_read_timeout_seconds());
     }
 }

src/secrets/aws_secrets_manager.rs

@@ -4,7 +4,7 @@ use aws_sdk_secretsmanager::{Client, config};
 use vector_lib::configurable::{component::GenerateConfig, configurable_component};
 
 use crate::{
-    aws::{AwsAuthentication, ClientBuilder, RegionOrEndpoint, create_client},
+    aws::{AwsAuthentication, AwsTimeout, ClientBuilder, RegionOrEndpoint, create_client},
     config::{ProxyConfig, SecretBackend},
     signal,
     tls::TlsConfig,
@@ -38,6 +38,15 @@ pub struct AwsSecretsManagerBackend {
 
     #[configurable(derived)]
     pub tls: Option<TlsConfig>,
+
+    /// Client timeout configuration for AWS requests.
+    ///
+    /// These settings bound how long the client waits when connecting to and reading from the
+    /// AWS API. Any dimension left unset falls back to a default (`connect_timeout_seconds = 5`,
+    /// `read_timeout_seconds = 30`).
+    #[configurable(derived)]
+    #[serde(default)]
+    pub client_timeout: AwsTimeout,
 }
 
 impl GenerateConfig for AwsSecretsManagerBackend {
@@ -47,6 +56,7 @@ impl GenerateConfig for AwsSecretsManagerBackend {
             region: Default::default(),
             auth: Default::default(),
             tls: None,
+            client_timeout: Default::default(),
         })
         .unwrap()
     }
@@ -65,7 +75,7 @@ impl SecretBackend for AwsSecretsManagerBackend {
             self.region.endpoint(),
             &ProxyConfig::default(),
             self.tls.as_ref(),
-            None,
+            &self.client_timeout,
         )
         .await?;
 

src/sinks/aws_cloudwatch_logs/config.rs

@@ -8,7 +8,7 @@ use vector_lib::{codecs::JsonSerializerConfig, configurable::configurable_compon
 use vrl::value::Kind;
 
 use crate::{
-    aws::{AwsAuthentication, ClientBuilder, RegionOrEndpoint, create_client},
+    aws::{AwsAuthentication, AwsTimeout, ClientBuilder, RegionOrEndpoint, create_client},
     codecs::{Encoder, EncodingConfig},
     config::{
         AcknowledgementsConfig, DataType, GenerateConfig, Input, ProxyConfig, SinkConfig,
@@ -157,6 +157,15 @@ pub struct CloudwatchLogsSinkConfig {
     #[serde(default)]
     pub auth: AwsAuthentication,
 
+    /// Client timeout configuration for AWS requests.
+    ///
+    /// These settings bound how long the client waits when connecting to and reading from the
+    /// AWS API. Any dimension left unset falls back to a default (`connect_timeout_seconds = 5`,
+    /// `read_timeout_seconds = 30`).
+    #[configurable(derived)]
+    #[serde(default)]
+    pub timeout: AwsTimeout,
+
     #[configurable(derived)]
     #[serde(
         default,
@@ -197,7 +206,7 @@ impl CloudwatchLogsSinkConfig {
             self.region.endpoint(),
             proxy,
             self.tls.as_ref(),
-            None,
+            &self.timeout,
         )
         .await
     }
@@ -279,6 +288,7 @@ fn default_config(encoding: EncodingConfig) -> CloudwatchLogsSinkConfig {
         tls: Default::default(),
         assume_role: Default::default(),
         auth: Default::default(),
+        timeout: Default::default(),
         acknowledgements: Default::default(),
         kms_key: Default::default(),
         tags: Default::default(),

src/sinks/aws_cloudwatch_logs/integration_tests.rs

@@ -11,7 +11,7 @@ use vrl::event_path;
 
 use super::*;
 use crate::{
-    aws::{AwsAuthentication, ClientBuilder, RegionOrEndpoint, create_client},
+    aws::{AwsAuthentication, AwsTimeout, ClientBuilder, RegionOrEndpoint, create_client},
     config::{ProxyConfig, SinkConfig, SinkContext, log_schema},
     event::{Event, LogEvent, Value},
     sinks::{aws_cloudwatch_logs::config::CloudwatchLogsClientBuilder, util::BatchConfig},
@@ -63,6 +63,7 @@ async fn cloudwatch_insert_log_event() {
         tls: Default::default(),
         assume_role: None,
         auth: Default::default(),
+        timeout: Default::default(),
         acknowledgements: Default::default(),
         kms_key: None,
         tags: None,
@@ -117,6 +118,7 @@ async fn cloudwatch_insert_log_events_sorted() {
         tls: Default::default(),
         assume_role: None,
         auth: Default::default(),
+        timeout: Default::default(),
         acknowledgements: Default::default(),
         kms_key: None,
         tags: None,
@@ -196,6 +198,7 @@ async fn cloudwatch_insert_out_of_range_timestamp() {
         tls: Default::default(),
         assume_role: None,
         auth: Default::default(),
+        timeout: Default::default(),
         acknowledgements: Default::default(),
         kms_key: None,
         tags: None,
@@ -276,6 +279,7 @@ async fn cloudwatch_dynamic_group_and_stream_creation() {
         tls: Default::default(),
         assume_role: None,
         auth: Default::default(),
+        timeout: Default::default(),
         acknowledgements: Default::default(),
         kms_key: None,
         tags: None,
@@ -330,6 +334,7 @@ async fn cloudwatch_dynamic_group_and_stream_creation_with_kms_key_and_tags() {
         tls: Default::default(),
         assume_role: None,
         auth: Default::default(),
+        timeout: Default::default(),
         acknowledgements: Default::default(),
         kms_key: Some(
             create_kms_client_test()
@@ -420,6 +425,7 @@ async fn cloudwatch_insert_log_event_batched() {
         tls: Default::default(),
         assume_role: None,
         auth: Default::default(),
+        timeout: Default::default(),
         acknowledgements: Default::default(),
         kms_key: None,
         tags: None,
@@ -474,6 +480,7 @@ async fn cloudwatch_insert_log_event_partitioned() {
         tls: Default::default(),
         assume_role: None,
         auth: Default::default(),
+        timeout: Default::default(),
         acknowledgements: Default::default(),
         kms_key: None,
         tags: None,
@@ -570,6 +577,7 @@ async fn cloudwatch_healthcheck() {
         tls: Default::default(),
         assume_role: None,
         auth: Default::default(),
+        timeout: Default::default(),
         acknowledgements: Default::default(),
         kms_key: None,
         tags: None,
@@ -593,7 +601,7 @@ async fn create_client_test() -> CloudwatchLogsClient {
         endpoint,
         &proxy,
         None,
-        None,
+        &AwsTimeout::default(),
     )
     .await
     .unwrap()
@@ -612,7 +620,7 @@ async fn create_kms_client_test() -> KMSClient {
         endpoint,
         &proxy,
         None,
-        None,
+        &AwsTimeout::default(),
     )
     .await
     .unwrap()

src/sinks/aws_cloudwatch_metrics/mod.rs

@@ -24,7 +24,8 @@ use vector_lib::{
 use super::util::service::TowerRequestConfigDefaults;
 use crate::{
     aws::{
-        ClientBuilder, RegionOrEndpoint, auth::AwsAuthentication, create_client, is_retriable_error,
+        AwsTimeout, ClientBuilder, RegionOrEndpoint, auth::AwsAuthentication, create_client,
+        is_retriable_error,
     },
     config::{AcknowledgementsConfig, Input, ProxyConfig, SinkConfig, SinkContext},
     event::{
@@ -107,6 +108,15 @@ pub struct CloudWatchMetricsSinkConfig {
     #[serde(default)]
     pub auth: AwsAuthentication,
 
+    /// Client timeout configuration for AWS requests.
+    ///
+    /// These settings bound how long the client waits when connecting to and reading from the
+    /// AWS API. Any dimension left unset falls back to a default (`connect_timeout_seconds = 5`,
+    /// `read_timeout_seconds = 30`).
+    #[configurable(derived)]
+    #[serde(default)]
+    pub timeout: AwsTimeout,
+
     #[configurable(derived)]
     #[serde(
         default,
@@ -191,7 +201,7 @@ impl CloudWatchMetricsSinkConfig {
             self.region.endpoint(),
             proxy,
             self.tls.as_ref(),
-            None,
+            &self.timeout,
         )
         .await
     }

src/sinks/aws_kinesis/config.rs

@@ -9,7 +9,7 @@ use super::{
     sink::{BatchKinesisRequest, KinesisSink},
 };
 use crate::{
-    aws::{AwsAuthentication, RegionOrEndpoint},
+    aws::{AwsAuthentication, AwsTimeout, RegionOrEndpoint},
     sinks::{
         prelude::*,
         util::{TowerRequestConfig, retries::RetryLogic},
@@ -51,6 +51,15 @@ pub struct KinesisSinkBaseConfig {
     #[serde(default)]
     pub auth: AwsAuthentication,
 
+    /// Client timeout configuration for AWS requests.
+    ///
+    /// These settings bound how long the client waits when connecting to and reading from the
+    /// AWS API. Any dimension left unset falls back to a default (`connect_timeout_seconds = 5`,
+    /// `read_timeout_seconds = 30`).
+    #[configurable(derived)]
+    #[serde(default)]
+    pub timeout: AwsTimeout,
+
     /// Whether or not to retry successful requests containing partial failures.
     #[serde(default)]
     #[configurable(metadata(docs::advanced))]

src/sinks/aws_kinesis/firehose/config.rs

@@ -109,7 +109,7 @@ impl KinesisFirehoseSinkConfig {
             self.base.region.endpoint(),
             proxy,
             self.base.tls.as_ref(),
-            None,
+            &self.base.timeout,
         )
         .await
     }