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

changelog.d/aws_s3_source_log_processing_error_key.enhancement.md

@@ -0,0 +1 @@
+The `aws_s3` source now logs an error containing the S3 object key whenever it fails to process an S3 event (e.g. a `GetObject` failure, a decode error, or a downstream acknowledgement error), making it easier to identify which object failed without cross-referencing SQS message IDs.

changelog.d/aws_s3_source_s3_timeouts.fix.md

@@ -0,0 +1,3 @@
+The `aws_s3` source now bounds its S3 `GetObject` requests with default client timeouts (`connect_timeout_seconds = 5`, `read_timeout_seconds = 30`) and exposes them as configuration options. Previously the S3 client was created without a read or operation timeout, so a half-open or silently dropped connection — for example one reaped by a NAT gateway idle timeout — could hang a polling task indefinitely, stalling SQS-based S3 ingestion until the process was restarted. The `read_timeout` applies per-read, so slow but steadily progressing transfers of large objects are not cut off. Any dimension can be overridden (or an operation timeout added) via `connect_timeout_seconds`, `read_timeout_seconds`, and `operation_timeout_seconds` on the source.
+
+authors: taylorchandleryoung

src/aws/timeout.rs

@@ -47,6 +47,21 @@ pub struct AwsTimeout {
 }
 
 impl AwsTimeout {
+    /// Creates a new timeout configuration from the given connect, operation, and read
+    /// timeouts, each expressed in seconds. A `None` leaves that dimension unbounded (subject
+    /// to the AWS SDK defaults).
+    pub const fn new(
+        connect_timeout: Option<u64>,
+        operation_timeout: Option<u64>,
+        read_timeout: Option<u64>,
+    ) -> Self {
+        Self {
+            connect_timeout,
+            operation_timeout,
+            read_timeout,
+        }
+    }
+
     /// returns the connection timeout
     pub const fn connect_timeout(&self) -> Option<u64> {
         self.connect_timeout

src/sources/aws_s3/mod.rs

@@ -19,7 +19,10 @@ use vrl::value::{Kind, kind::Collection};
 
 use super::util::MultilineConfig;
 use crate::{
-    aws::{RegionOrEndpoint, auth::AwsAuthentication, create_client, create_client_and_region},
+    aws::{
+        AwsTimeout, RegionOrEndpoint, auth::AwsAuthentication, create_client,
+        create_client_and_region,
+    },
     codecs::DecodingConfig,
     common::{s3::S3ClientBuilder, sqs::SqsClientBuilder},
     config::{
@@ -117,6 +120,19 @@ pub struct AwsS3Config {
     #[configurable(derived)]
     tls_options: Option<TlsConfig>,
 
+    /// Client timeout configuration for S3 operations.
+    ///
+    /// These settings bound the S3 `GetObject` request and the read of its response body. Any
+    /// dimension left unset falls back to a safe default (`connect_timeout_seconds = 5`,
+    /// `read_timeout_seconds = 30`) so that a half-open or silently dropped connection — for
+    /// example one reaped by a NAT gateway idle timeout — cannot hang a polling task
+    /// indefinitely and stall S3 ingestion. `read_timeout_seconds` applies per-read rather than
+    /// to the whole request, so a slow but steadily progressing transfer of a large object is
+    /// not cut off.
+    #[configurable(derived)]
+    #[serde(default)]
+    timeout: Option<AwsTimeout>,
+
     /// The namespace to use for logs. This overrides the global setting.
     #[configurable(metadata(docs::hidden))]
     #[serde(default)]
@@ -151,6 +167,33 @@ const fn default_true() -> bool {
     true
 }
 
+/// Default connect timeout for the S3 client, in seconds.
+const DEFAULT_S3_CONNECT_TIMEOUT_SECONDS: u64 = 5;
+
+/// Default read (per-read inactivity) timeout for the S3 client, in seconds.
+///
+/// This bounds the response-body read of `GetObject` so a stalled or silently dropped
+/// connection cannot hang a polling task indefinitely. It is applied per-read, so a slow but
+/// steadily progressing transfer of a large object is not cut off. No operation timeout is set
+/// by default, to avoid capping legitimately long large-object transfers.
+const DEFAULT_S3_READ_TIMEOUT_SECONDS: u64 = 30;
+
+/// Resolves the effective S3 client timeout, applying default bounds to any dimension the
+/// configuration leaves unset. Connect and read timeouts default to bounded values; the
+/// operation timeout is left unset so large-object transfers are not capped.
+fn resolve_s3_timeout(configured: Option<AwsTimeout>) -> AwsTimeout {
+    let configured = configured.unwrap_or_default();
+    AwsTimeout::new(
+        configured
+            .connect_timeout()
+            .or(Some(DEFAULT_S3_CONNECT_TIMEOUT_SECONDS)),
+        configured.operation_timeout(),
+        configured
+            .read_timeout()
+            .or(Some(DEFAULT_S3_READ_TIMEOUT_SECONDS)),
+    )
+}
+
 impl_generate_config_from_default!(AwsS3Config);
 
 #[async_trait::async_trait]
@@ -243,6 +286,10 @@ impl AwsS3Config {
         let region = self.region.region();
         let endpoint = self.region.endpoint();
 
+        // Bound the S3 client's requests. Historically this was `None`, which left the
+        // response-body read of `GetObject` unbounded, so a half-open or silently dropped
+        // connection could hang a polling task indefinitely and stall SQS-based ingestion.
+        let s3_timeout = resolve_s3_timeout(self.timeout);
         let s3_client = create_client::<S3ClientBuilder>(
             &S3ClientBuilder {
                 force_path_style: Some(self.force_path_style),
@@ -252,7 +299,7 @@ impl AwsS3Config {
             endpoint.clone(),
             proxy,
             self.tls_options.as_ref(),
-            None,
+            Some(&s3_timeout),
         )
         .await?;
 

src/sources/aws_s3/sqs.rs

@@ -273,6 +273,23 @@ pub enum ProcessingError {
     },
 }
 
+impl ProcessingError {
+    /// Returns the S3 object key associated with this error, if the failure occurred after the
+    /// S3 event record (and thus its key) was known.
+    fn key(&self) -> Option<&str> {
+        match self {
+            ProcessingError::GetObject { key, .. }
+            | ProcessingError::ReadObject { key, .. }
+            | ProcessingError::PipelineSend { key, .. }
+            | ProcessingError::WrongRegion { key, .. }
+            | ProcessingError::ErrorAcknowledgement { key, .. }
+            | ProcessingError::FileTooOld { key, .. } => Some(key),
+            ProcessingError::InvalidSqsMessage { .. }
+            | ProcessingError::UnsupportedS3EventVersion { .. } => None,
+        }
+    }
+}
+
 pub struct State {
     region: Region,
 
@@ -524,6 +541,11 @@ impl IngestorProcess {
                             }
                         }
                         _ => {
+                            error!(
+                                message = "Failed to process S3 event.",
+                                key = err.key().unwrap_or("<unknown>"),
+                                error = %err,
+                            );
                             emit!(SqsMessageProcessingError {
                                 message_id: &message_id,
                                 error: &err,