Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions changelog.d/aws_s3_source_s3_timeouts.fix.md
Original file line number Diff line number Diff line change
@@ -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
15 changes: 15 additions & 0 deletions src/aws/timeout.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
51 changes: 49 additions & 2 deletions src/sources/aws_s3/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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::{
Expand Down Expand Up @@ -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)]
Expand Down Expand Up @@ -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]
Expand Down Expand Up @@ -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),
Expand All @@ -252,7 +299,7 @@ impl AwsS3Config {
endpoint.clone(),
proxy,
self.tls_options.as_ref(),
None,
Some(&s3_timeout),
)
.await?;

Expand Down
Loading