From bbf2140b853cba0531e2c612daeed18a3407b40e Mon Sep 17 00:00:00 2001 From: Armand Thibaudon Date: Fri, 3 Jul 2026 19:12:00 +0200 Subject: [PATCH 1/4] enhancement(sources): pre-size HTTP request body buffer from Content-Length --- ...ze_http_request_body_buffer.enhancement.md | 3 +++ src/sources/util/http/encoding.rs | 26 +++++++++++++------ 2 files changed, 21 insertions(+), 8 deletions(-) create mode 100644 changelog.d/presize_http_request_body_buffer.enhancement.md diff --git a/changelog.d/presize_http_request_body_buffer.enhancement.md b/changelog.d/presize_http_request_body_buffer.enhancement.md new file mode 100644 index 0000000000000..8b58a1d718cc4 --- /dev/null +++ b/changelog.d/presize_http_request_body_buffer.enhancement.md @@ -0,0 +1,3 @@ +The HTTP-based sources (e.g. `http_server`, `opentelemetry`) now pre-size the request-body buffer from the `Content-Length` header, collecting each body in a single allocation instead of growing an unsized buffer. This removes a per-request reallocation/copy cost on the ingestion path and restores throughput on high-volume HTTP sources, while keeping the streaming decompressed-size cap. + +authors: armleth diff --git a/src/sources/util/http/encoding.rs b/src/sources/util/http/encoding.rs index 5d90ca0c20189..8b08e80d54f8d 100644 --- a/src/sources/util/http/encoding.rs +++ b/src/sources/util/http/encoding.rs @@ -60,13 +60,12 @@ pub(crate) fn limited_body(max_body_size: usize) -> BoxedFilter<(Bytes,)> { max_body_size, ))) } else { - Ok(()) + Ok(declared) } }) - .untuple_one() .and(warp::body::stream()) - .and_then(move |body| async move { - collect_body_with_limit(body, max_body_size) + .and_then(move |declared: Option, body| async move { + collect_body_with_limit(body, max_body_size, declared) .await .map_err(warp::reject::custom) }) @@ -183,14 +182,25 @@ fn decompress_snappy( feature = "sources-opentelemetry", test ))] -async fn collect_body_with_limit(body: S, max_body_size: usize) -> Result +async fn collect_body_with_limit( + body: S, + max_body_size: usize, + declared_len: Option, +) -> Result where S: futures_util::Stream>, B: Buf, { futures_util::pin_mut!(body); - let mut bytes = BytesMut::new(); + // Pre-size from the Content-Length header (capped at the limit) so the body is collected with + // a single allocation and one copy per chunk, matching the previous `warp::body::bytes()` + // cost. Without this the buffer starts empty and grows by repeated realloc+recopy on every + // request, which shows up as a per-request throughput/CPU regression on high-volume sources. + let capacity = declared_len + .and_then(|len| usize::try_from(len).ok()) + .map_or(0, |len| len.min(max_body_size)); + let mut bytes = BytesMut::with_capacity(capacity); while let Some(chunk) = body.next().await { let chunk = chunk.map_err(|error| { ErrorMessage::new( @@ -368,7 +378,7 @@ mod tests { Ok::<_, warp::Error>(Bytes::from_static(b" world")), ]); - let collected = collect_body_with_limit(body, 11).await.unwrap(); + let collected = collect_body_with_limit(body, 11, Some(11)).await.unwrap(); assert_eq!(collected, Bytes::from_static(b"hello world")); } @@ -379,7 +389,7 @@ mod tests { Ok::<_, warp::Error>(Bytes::from_static(b" world")), ]); - let err = collect_body_with_limit(body, 5) + let err = collect_body_with_limit(body, 5, None) .await .expect_err("should reject"); assert_eq!(err.status_code(), StatusCode::PAYLOAD_TOO_LARGE); From 2c6c5afde61799eba7777f98c2832059088d4d29 Mon Sep 17 00:00:00 2001 From: Armand Thibaudon Date: Fri, 3 Jul 2026 19:22:50 +0200 Subject: [PATCH 2/4] chore: drop changelog fragment (using no-changelog label) --- changelog.d/presize_http_request_body_buffer.enhancement.md | 3 --- 1 file changed, 3 deletions(-) delete mode 100644 changelog.d/presize_http_request_body_buffer.enhancement.md diff --git a/changelog.d/presize_http_request_body_buffer.enhancement.md b/changelog.d/presize_http_request_body_buffer.enhancement.md deleted file mode 100644 index 8b58a1d718cc4..0000000000000 --- a/changelog.d/presize_http_request_body_buffer.enhancement.md +++ /dev/null @@ -1,3 +0,0 @@ -The HTTP-based sources (e.g. `http_server`, `opentelemetry`) now pre-size the request-body buffer from the `Content-Length` header, collecting each body in a single allocation instead of growing an unsized buffer. This removes a per-request reallocation/copy cost on the ingestion path and restores throughput on high-volume HTTP sources, while keeping the streaming decompressed-size cap. - -authors: armleth From 428db8bebe7679c3dcf644ac6a5c62675b73b347 Mon Sep 17 00:00:00 2001 From: Armand Thibaudon Date: Fri, 3 Jul 2026 19:23:47 +0200 Subject: [PATCH 3/4] chore: drop verbose comment --- src/sources/util/http/encoding.rs | 4 ---- 1 file changed, 4 deletions(-) diff --git a/src/sources/util/http/encoding.rs b/src/sources/util/http/encoding.rs index 8b08e80d54f8d..9f1e2692d044b 100644 --- a/src/sources/util/http/encoding.rs +++ b/src/sources/util/http/encoding.rs @@ -193,10 +193,6 @@ where { futures_util::pin_mut!(body); - // Pre-size from the Content-Length header (capped at the limit) so the body is collected with - // a single allocation and one copy per chunk, matching the previous `warp::body::bytes()` - // cost. Without this the buffer starts empty and grows by repeated realloc+recopy on every - // request, which shows up as a per-request throughput/CPU regression on high-volume sources. let capacity = declared_len .and_then(|len| usize::try_from(len).ok()) .map_or(0, |len| len.min(max_body_size)); From 3e16af70e1b6d091dd537d5b5c3640b090aafd2a Mon Sep 17 00:00:00 2001 From: Armand Thibaudon Date: Mon, 6 Jul 2026 12:05:54 +0200 Subject: [PATCH 4/4] fix(sources): cap HTTP body buffer preallocation at 8 MiB --- src/sources/util/http/encoding.rs | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/src/sources/util/http/encoding.rs b/src/sources/util/http/encoding.rs index 9f1e2692d044b..beb418c5a7b38 100644 --- a/src/sources/util/http/encoding.rs +++ b/src/sources/util/http/encoding.rs @@ -177,6 +177,15 @@ fn decompress_snappy( Ok(decoded.into()) } +/// Cap on the initial buffer allocation when pre-sizing from a `Content-Length` header. +/// Prevents stalling clients from reserving large amounts of memory before sending any bytes. +#[cfg(any( + feature = "sources-utils-http-prelude", + feature = "sources-opentelemetry", + test +))] +const MAX_INITIAL_BODY_CAPACITY: usize = 8 * 1024 * 1024; // 8 MiB + #[cfg(any( feature = "sources-utils-http-prelude", feature = "sources-opentelemetry", @@ -195,7 +204,7 @@ where let capacity = declared_len .and_then(|len| usize::try_from(len).ok()) - .map_or(0, |len| len.min(max_body_size)); + .map_or(0, |len| len.min(max_body_size).min(MAX_INITIAL_BODY_CAPACITY)); let mut bytes = BytesMut::with_capacity(capacity); while let Some(chunk) = body.next().await { let chunk = chunk.map_err(|error| {