From 70a0d4d48a2ec530f97ca16e807ff472fd8e6655 Mon Sep 17 00:00:00 2001 From: paullegranddc Date: Tue, 7 Jul 2026 14:58:17 +0200 Subject: [PATCH 1/4] feat(data-pipeline): add compression option for agentless export # Motivation Since we are sending data to the intake directly, we should be compressing requests as they can be multi MB. The compression uses the zstd crate, which was already used in trace-utils. It links to a C library, so it is not usable everyhwere (WASM for instance) and adds ~500KB of artifact size. For these reasons, it is hidden behind a feature flag and should be enabled as required (implementation using the agentless option mostly). Another option would be rustzstd, an full rust port. Tge library is smaller (300KB) but it is notably slower.. (2x to 5x slower, with worse compression ratios at the same level) # Changes * Add a compression strategy parameter to send_with_retry * implement zstd compression * since compressio is implemented lower down, remove compression option from SendData * Add a "compression" feature flag on data-pipeline to be used in languages where it is needed --- libdd-data-pipeline/Cargo.toml | 2 + libdd-data-pipeline/src/agentless/exporter.rs | 18 ++++- libdd-data-pipeline/src/otlp/exporter.rs | 13 +++- libdd-data-pipeline/src/trace_exporter/mod.rs | 3 +- libdd-trace-stats/src/stats_exporter.rs | 3 +- libdd-trace-utils/Cargo.toml | 8 ++- libdd-trace-utils/src/send_data/mod.rs | 71 ++++--------------- .../src/send_with_retry/compression.rs | 50 +++++++++++++ libdd-trace-utils/src/send_with_retry/mod.rs | 16 ++++- 9 files changed, 118 insertions(+), 66 deletions(-) create mode 100644 libdd-trace-utils/src/send_with_retry/compression.rs diff --git a/libdd-data-pipeline/Cargo.toml b/libdd-data-pipeline/Cargo.toml index 46bc35a720..ac6b0a285b 100644 --- a/libdd-data-pipeline/Cargo.toml +++ b/libdd-data-pipeline/Cargo.toml @@ -111,3 +111,5 @@ fips = [ ] test-utils = [] regex-lite = ["libdd-common/regex-lite"] +# Enable zstd compression for the agentless trace intake sender. +compression = ["libdd-trace-utils/compression"] diff --git a/libdd-data-pipeline/src/agentless/exporter.rs b/libdd-data-pipeline/src/agentless/exporter.rs index bc512433bc..5ac9d05209 100644 --- a/libdd-data-pipeline/src/agentless/exporter.rs +++ b/libdd-data-pipeline/src/agentless/exporter.rs @@ -9,7 +9,7 @@ use http::HeaderMap; use libdd_capabilities::{HttpClientCapability, SleepCapability}; use libdd_common::Endpoint; use libdd_trace_utils::send_with_retry::{ - send_with_retry, RetryBackoffType, RetryStrategy, SendWithRetryError, + send_with_retry, CompressionStrategy, RetryBackoffType, RetryStrategy, SendWithRetryError, }; use tracing::error; @@ -46,7 +46,21 @@ pub async fn send_agentless_traces_http Ok(()), Err(e) => Err(map_send_error(e)), } diff --git a/libdd-data-pipeline/src/otlp/exporter.rs b/libdd-data-pipeline/src/otlp/exporter.rs index 31c952010a..1297a7a25e 100644 --- a/libdd-data-pipeline/src/otlp/exporter.rs +++ b/libdd-data-pipeline/src/otlp/exporter.rs @@ -9,7 +9,7 @@ use http::HeaderMap; use libdd_capabilities::{HttpClientCapability, SleepCapability}; use libdd_common::Endpoint; use libdd_trace_utils::send_with_retry::{ - send_with_retry, RetryBackoffType, RetryStrategy, SendWithRetryError, + send_with_retry, CompressionStrategy, RetryBackoffType, RetryStrategy, SendWithRetryError, }; use std::time::Duration; @@ -67,7 +67,16 @@ pub(crate) async fn send_otlp_http( None, ); - match send_with_retry(capabilities, &target, body, &headers, &retry_strategy).await { + match send_with_retry( + capabilities, + &target, + body, + &headers, + &retry_strategy, + CompressionStrategy::None, + ) + .await + { Ok(_) => Ok(()), Err(e) => Err(map_send_error(e).await), } diff --git a/libdd-data-pipeline/src/trace_exporter/mod.rs b/libdd-data-pipeline/src/trace_exporter/mod.rs index 44ead972e1..dc393ab79a 100644 --- a/libdd-data-pipeline/src/trace_exporter/mod.rs +++ b/libdd-data-pipeline/src/trace_exporter/mod.rs @@ -48,7 +48,7 @@ use libdd_shared_runtime::BlockingRuntime; use libdd_shared_runtime::{SharedRuntime, WorkerHandle}; use libdd_trace_utils::msgpack_decoder; use libdd_trace_utils::send_with_retry::{ - send_with_retry, RetryStrategy, SendWithRetryError, SendWithRetryResult, + send_with_retry, CompressionStrategy, RetryStrategy, SendWithRetryError, SendWithRetryResult, }; use libdd_trace_utils::span::{v04::Span, TraceData}; use libdd_trace_utils::trace_utils::TracerHeaderTags; @@ -733,6 +733,7 @@ impl< mp_payload, &headers, &strategy, + CompressionStrategy::None, ) .await; diff --git a/libdd-trace-stats/src/stats_exporter.rs b/libdd-trace-stats/src/stats_exporter.rs index c771fd5b2b..0aac3e8ec4 100644 --- a/libdd-trace-stats/src/stats_exporter.rs +++ b/libdd-trace-stats/src/stats_exporter.rs @@ -17,7 +17,7 @@ use libdd_capabilities::{HttpClientCapability, MaybeSend, SleepCapability}; use libdd_common::{tag, Endpoint}; use libdd_shared_runtime::Worker; use libdd_trace_protobuf::pb; -use libdd_trace_utils::send_with_retry::{send_with_retry, RetryStrategy}; +use libdd_trace_utils::send_with_retry::{send_with_retry, CompressionStrategy, RetryStrategy}; use libdd_trace_utils::trace_utils::TracerHeaderTags; use libdd_trace_utils::tracer_metadata::TracerMetadata; use std::fmt::Debug; @@ -224,6 +224,7 @@ impl body, &headers, &RetryStrategy::default(), + CompressionStrategy::None, ) .await; diff --git a/libdd-trace-utils/Cargo.toml b/libdd-trace-utils/Cargo.toml index 6c653e1fe4..12580c9619 100644 --- a/libdd-trace-utils/Cargo.toml +++ b/libdd-trace-utils/Cargo.toml @@ -76,7 +76,9 @@ getrandom = { version = "0.2", features = ["js"] } [dev-dependencies] libdd-capabilities-impl = { version = "2.0.0", path = "../libdd-capabilities-impl" } -libdd-common = { path = "../libdd-common", default-features = false, features = ["bench-utils"] } +libdd-common = { path = "../libdd-common", default-features = false, features = [ + "bench-utils", +] } bolero = "0.13" criterion = "0.5.1" httpmock = { version = "0.8.0-alpha.1" } @@ -88,7 +90,7 @@ tempfile = "3.3.0" [features] default = ["https"] https = ["libdd-common/https", "libdd-capabilities-impl/https"] -mini_agent = ["compression", "libdd-common/use_webpki_roots"] +mini_agent = ["compression", "libdd-common/use_webpki_roots", "dep:flate2"] test-utils = [ "hyper/server", "httpmock", @@ -99,6 +101,6 @@ test-utils = [ change-buffer = [] # Opt-in switch for crate-internal microbenchmarks (e.g. `vec_map_bench`). bench-internals = [] -compression = ["zstd", "flate2"] +compression = ["zstd"] # FIPS mode uses the FIPS-compliant cryptographic provider (Unix only) fips = ["libdd-common/fips", "libdd-capabilities-impl/fips"] diff --git a/libdd-trace-utils/src/send_data/mod.rs b/libdd-trace-utils/src/send_data/mod.rs index 53d14d8ecd..a6310f8147 100644 --- a/libdd-trace-utils/src/send_data/mod.rs +++ b/libdd-trace-utils/src/send_data/mod.rs @@ -4,7 +4,9 @@ pub mod send_data_result; use crate::msgpack_encoder; -use crate::send_with_retry::{send_with_retry, RetryStrategy, SendWithRetryResult}; +use crate::send_with_retry::{ + send_with_retry, CompressionStrategy, RetryStrategy, SendWithRetryResult, +}; use crate::trace_utils::TracerHeaderTags; use crate::tracer_payload::TracerPayloadCollection; use anyhow::{anyhow, Context}; @@ -22,10 +24,6 @@ use libdd_common::{ use libdd_trace_protobuf::pb::{AgentPayload, TracerPayload}; use send_data_result::SendDataResult; use std::collections::HashMap; -#[cfg(feature = "compression")] -use std::io::Write; -#[cfg(feature = "compression")] -use zstd::stream::write::Encoder; #[derive(Debug)] /// `SendData` is a structure that holds the data to be sent to a target endpoint. @@ -71,15 +69,7 @@ pub struct SendData { target: Endpoint, headers: HeaderMap, retry_strategy: RetryStrategy, - #[cfg(feature = "compression")] - compression: Compression, -} - -#[cfg(feature = "compression")] -#[derive(Debug, Clone)] -pub enum Compression { - Zstd(i32), - None, + compression: CompressionStrategy, } pub struct SendDataBuilder { @@ -88,8 +78,7 @@ pub struct SendDataBuilder { target: Endpoint, headers: HeaderMap, retry_strategy: RetryStrategy, - #[cfg(feature = "compression")] - compression: Compression, + compression: CompressionStrategy, } impl SendDataBuilder { @@ -107,13 +96,11 @@ impl SendDataBuilder { target: target.clone(), headers, retry_strategy: RetryStrategy::default(), - #[cfg(feature = "compression")] - compression: Compression::None, + compression: CompressionStrategy::None, } } - #[cfg(feature = "compression")] - pub fn with_compression(mut self, compression: Compression) -> SendDataBuilder { + pub fn with_compression(mut self, compression: CompressionStrategy) -> SendDataBuilder { self.compression = compression; self } @@ -135,7 +122,6 @@ impl SendDataBuilder { target: self.target, headers: self.headers, retry_strategy: self.retry_strategy, - #[cfg(feature = "compression")] compression: self.compression, } } @@ -169,8 +155,7 @@ impl SendData { target: target.clone(), headers, retry_strategy: RetryStrategy::default(), - #[cfg(feature = "compression")] - compression: Compression::None, + compression: CompressionStrategy::None, } } @@ -250,6 +235,7 @@ impl SendData { payload: Vec, headers: HeaderMap, endpoint: Option<&Endpoint>, + compression_strategy: CompressionStrategy, ) -> (SendWithRetryResult, u64, u64) { #[allow(clippy::unwrap_used)] let payload_len = u64::try_from(payload.len()).unwrap(); @@ -260,6 +246,7 @@ impl SendData { payload, &headers, &self.retry_strategy, + compression_strategy, ) .await, payload_len, @@ -271,31 +258,6 @@ impl SendData { self.target.api_key.is_some() } - #[cfg(feature = "compression")] - fn compress_payload(&self, payload: Vec, headers: &mut HeaderMap) -> Vec { - match self.compression { - Compression::Zstd(level) => { - let result = (|| -> std::io::Result> { - let mut encoder = Encoder::new(Vec::new(), level)?; - encoder.write_all(&payload)?; - encoder.finish() - })(); - - match result { - Ok(compressed_payload) => { - headers.insert( - http::header::CONTENT_ENCODING, - HeaderValue::from_static("zstd"), - ); - compressed_payload - } - Err(_) => payload, - } - } - _ => payload, - } - } - async fn send_with_protobuf( &self, capabilities: &C, @@ -317,22 +279,16 @@ impl SendData { }; let mut request_headers = self.headers.clone(); - #[cfg(feature = "compression")] - let final_payload = - self.compress_payload(serialized_trace_payload, &mut request_headers); - - #[cfg(not(feature = "compression"))] - let final_payload = serialized_trace_payload; - request_headers.insert(CONTENT_TYPE, APPLICATION_PROTOBUF); let (response, bytes_sent, chunks) = self .send_payload( capabilities, chunks, - final_payload, + serialized_trace_payload, request_headers, endpoint.as_ref(), + self.compression, ) .await; @@ -373,6 +329,7 @@ impl SendData { payload, headers, endpoint.as_ref(), + CompressionStrategy::None, )); } } @@ -392,6 +349,7 @@ impl SendData { payload, headers, endpoint.as_ref(), + CompressionStrategy::None, )); } TracerPayloadCollection::V05(payload) => { @@ -413,6 +371,7 @@ impl SendData { payload, headers, endpoint.as_ref(), + CompressionStrategy::None, )); } } diff --git a/libdd-trace-utils/src/send_with_retry/compression.rs b/libdd-trace-utils/src/send_with_retry/compression.rs new file mode 100644 index 0000000000..447a7ebaef --- /dev/null +++ b/libdd-trace-utils/src/send_with_retry/compression.rs @@ -0,0 +1,50 @@ +// Copyright 2024-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +#[cfg(feature = "compression")] +use std::io::Write as _; + +#[cfg(feature = "compression")] +const CONTENT_ENCODING_ZSTD: http::HeaderValue = http::HeaderValue::from_static("zstd"); + +#[derive(Clone, Copy, Debug)] +pub enum CompressionStrategy { + None, + #[cfg(feature = "compression")] + Zstd { + level: i32, + }, +} + +/// Returns the compressed data, and the actual compression strategy used. +/// If an error happens during compression, defaults to [`CompressionStrategy::None`] +pub fn compress(data: Vec, strategy: CompressionStrategy) -> (Vec, CompressionStrategy) { + match strategy { + CompressionStrategy::None => (data, CompressionStrategy::None), + #[cfg(feature = "compression")] + CompressionStrategy::Zstd { level } => { + // Start with an initial buffer + // Allocate 1/10th of the original buffer, so we shouldn't add too + // much memory usage, and no less than 256 bytes + let writer = Vec::with_capacity((data.len() / 10).max(256)); + zstd::Encoder::new(writer, level) + .and_then(|mut e| { + e.write_all(&data)?; + Ok((e.finish()?, strategy)) + }) + .unwrap_or((data, CompressionStrategy::None)) + } + } +} + +pub fn add_headers(headers: &mut http::HeaderMap, strategy: CompressionStrategy) { + match strategy { + CompressionStrategy::None => { + let _ = headers; + } + #[cfg(feature = "compression")] + CompressionStrategy::Zstd { .. } => { + headers.insert(http::header::CONTENT_ENCODING, CONTENT_ENCODING_ZSTD); + } + } +} diff --git a/libdd-trace-utils/src/send_with_retry/mod.rs b/libdd-trace-utils/src/send_with_retry/mod.rs index 88cedb1764..d20f6ec9bf 100644 --- a/libdd-trace-utils/src/send_with_retry/mod.rs +++ b/libdd-trace-utils/src/send_with_retry/mod.rs @@ -7,6 +7,9 @@ mod retry_strategy; pub use retry_strategy::{RetryBackoffType, RetryStrategy}; +mod compression; +pub use compression::CompressionStrategy; + use bytes::Bytes; use http::HeaderMap; use libdd_capabilities::{HttpClientCapability, HttpError, SleepCapability}; @@ -92,6 +95,7 @@ pub async fn send_with_retry( payload: Vec, headers: &HeaderMap, retry_strategy: &RetryStrategy, + compression_strategy: CompressionStrategy, ) -> SendWithRetryResult { let mut request_attempt = 0; let timeout = Duration::from_millis(target.timeout_ms); @@ -103,7 +107,9 @@ pub async fn send_with_retry( "Sending with retry" ); - let payload = Bytes::from(payload); + let (compressed, compression_strategy) = compression::compress(payload, compression_strategy); + let payload = Bytes::from(compressed); + loop { request_attempt += 1; @@ -121,6 +127,10 @@ pub async fn send_with_retry( for (key, value) in headers { builder = builder.header(key, value); } + // headers_mut is only None if the builder is in an error state + if let Some(h) = builder.headers_mut() { + compression::add_headers(h, compression_strategy); + } let req = match builder.body(payload.clone()) { Ok(r) => r, Err(_) => { @@ -281,6 +291,7 @@ mod tests { vec![0, 1, 2, 3], &HeaderMap::new(), &strategy, + CompressionStrategy::None, ) .await; assert!(result.is_err(), "Expected an error result"); @@ -330,6 +341,7 @@ mod tests { vec![0, 1, 2, 3], &HeaderMap::new(), &strategy, + CompressionStrategy::None, ) .await; assert!( @@ -375,6 +387,7 @@ mod tests { vec![0, 1, 2, 3], &HeaderMap::new(), &strategy, + CompressionStrategy::None, ) .await; assert!( @@ -424,6 +437,7 @@ mod tests { vec![0, 1, 2, 3], &HeaderMap::new(), &strategy, + CompressionStrategy::None, ) .await; assert!( From f5530e26c333185a666689f476f14412b794a419 Mon Sep 17 00:00:00 2001 From: paullegranddc Date: Wed, 8 Jul 2026 00:21:21 +0200 Subject: [PATCH 2/4] fix: rustdoc --- libdd-trace-utils/src/send_with_retry/mod.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/libdd-trace-utils/src/send_with_retry/mod.rs b/libdd-trace-utils/src/send_with_retry/mod.rs index d20f6ec9bf..ff45b23949 100644 --- a/libdd-trace-utils/src/send_with_retry/mod.rs +++ b/libdd-trace-utils/src/send_with_retry/mod.rs @@ -86,7 +86,7 @@ impl std::error::Error for SendWithRetryError {} /// ); /// let retry_strategy = RetryStrategy::new(3, 10, RetryBackoffType::Exponential, Some(5)); /// let capabilities = libdd_capabilities_impl::NativeCapabilities::new_client(); -/// send_with_retry(&capabilities, &target, payload, &headers, &retry_strategy).await +/// send_with_retry(&capabilities, &target, payload, &headers, &retry_strategy, CompressionStrategy::None).await /// # } /// ``` pub async fn send_with_retry( From 0b849f6d570b3dc5a9eaa6c9e3bc62bd7f5a579f Mon Sep 17 00:00:00 2001 From: paullegranddc Date: Thu, 9 Jul 2026 15:19:51 +0200 Subject: [PATCH 3/4] fix(data-pipeline): address compression PR review feedback - Fix rustfmt on the send_with_retry doctest example - Keep test_agentless_export_body_shape compatible with the compression feature by asserting on content-encoding when compression is enabled - Report the compressed byte count in SendData telemetry by compressing in send_payload before measuring bytes_sent - Expose a 'compression' feature through the data-pipeline / profiling FFI crates and the builder so release artifacts can opt in --- builder/Cargo.toml | 3 +++ builder/src/bin/release.rs | 2 ++ libdd-data-pipeline-ffi/Cargo.toml | 2 ++ libdd-data-pipeline/src/trace_exporter/mod.rs | 12 ++++++++++-- libdd-profiling-ffi/Cargo.toml | 2 ++ libdd-trace-utils/src/send_data/mod.rs | 9 +++++++-- libdd-trace-utils/src/send_with_retry/mod.rs | 12 ++++++++++-- 7 files changed, 36 insertions(+), 6 deletions(-) diff --git a/builder/Cargo.toml b/builder/Cargo.toml index 4c07580c7d..0fbf801709 100644 --- a/builder/Cargo.toml +++ b/builder/Cargo.toml @@ -25,6 +25,9 @@ crashtracker = [] profiling = [] telemetry = [] data-pipeline = [] +# Enable zstd compression for the agentless trace intake sender (opt-in: links a C +# library and increases artifact size). +data-pipeline-compression = ["data-pipeline"] symbolizer = [] library-config = [] log = [] diff --git a/builder/src/bin/release.rs b/builder/src/bin/release.rs index ff0904ee7e..be54a3b5e1 100644 --- a/builder/src/bin/release.rs +++ b/builder/src/bin/release.rs @@ -62,6 +62,8 @@ pub fn main() { f.push("ddtelemetry-ffi".to_string()); #[cfg(feature = "data-pipeline")] f.push("data-pipeline-ffi".to_string()); + #[cfg(feature = "data-pipeline-compression")] + f.push("data-pipeline-compression".to_string()); #[cfg(feature = "crashtracker")] f.push("crashtracker-ffi".to_string()); #[cfg(feature = "symbolizer")] diff --git a/libdd-data-pipeline-ffi/Cargo.toml b/libdd-data-pipeline-ffi/Cargo.toml index 300786b331..a673b7699a 100644 --- a/libdd-data-pipeline-ffi/Cargo.toml +++ b/libdd-data-pipeline-ffi/Cargo.toml @@ -20,6 +20,8 @@ default = ["cbindgen", "catch_panic"] catch_panic = [] cbindgen = ["build_common/cbindgen", "libdd-common-ffi/cbindgen"] regex-lite = ["libdd-data-pipeline/regex-lite"] +# Enable zstd compression for the agentless trace intake sender. +compression = ["libdd-data-pipeline/compression"] [build-dependencies] build_common = { path = "../build-common" } diff --git a/libdd-data-pipeline/src/trace_exporter/mod.rs b/libdd-data-pipeline/src/trace_exporter/mod.rs index dc393ab79a..b40a96e92e 100644 --- a/libdd-data-pipeline/src/trace_exporter/mod.rs +++ b/libdd-data-pipeline/src/trace_exporter/mod.rs @@ -2440,8 +2440,15 @@ mod tests { fn test_agentless_export_body_shape() { let server = MockServer::start(); let mock_intake = server.mock(|when, then| { - when.method(POST) - .path("/v1/input") + let when = when.method(POST).path("/v1/input"); + // When the `compression` feature is enabled the agentless sender zstd-compresses + // the body, so the JSON substrings are no longer present in the raw request. In that + // case assert on the `content-encoding` header instead; the body shape itself is + // covered by the default (uncompressed) build. + #[cfg(feature = "compression")] + let when = when.header("content-encoding", "zstd"); + #[cfg(not(feature = "compression"))] + let when = when .body_includes("\"traces\":") .body_includes("\"spans\":") .body_includes("\"hostname\":\"h-1\"") @@ -2450,6 +2457,7 @@ mod tests { .body_includes("\"_top_level\":1") .body_includes("\"_trace_root\":1") .body_includes("\"parent_id\":\"0000000000000000\""); + let _ = when; then.status(200).body(""); }); diff --git a/libdd-profiling-ffi/Cargo.toml b/libdd-profiling-ffi/Cargo.toml index e1b40f7063..977a5d70de 100644 --- a/libdd-profiling-ffi/Cargo.toml +++ b/libdd-profiling-ffi/Cargo.toml @@ -23,6 +23,8 @@ ddtelemetry-ffi = ["dep:libdd-telemetry-ffi"] datadog-log-ffi = ["dep:libdd-log-ffi"] symbolizer = ["symbolizer-ffi"] data-pipeline-ffi = ["dep:libdd-data-pipeline-ffi"] +# Enable zstd compression for the agentless trace intake sender. +data-pipeline-compression = ["data-pipeline-ffi", "libdd-data-pipeline-ffi/compression"] crashtracker-ffi = ["dep:libdd-crashtracker-ffi"] # Enables the in-process collection of crash-info crashtracker-collector = ["crashtracker-ffi", "libdd-crashtracker-ffi/collector"] diff --git a/libdd-trace-utils/src/send_data/mod.rs b/libdd-trace-utils/src/send_data/mod.rs index a6310f8147..d6e7b9db47 100644 --- a/libdd-trace-utils/src/send_data/mod.rs +++ b/libdd-trace-utils/src/send_data/mod.rs @@ -4,6 +4,7 @@ pub mod send_data_result; use crate::msgpack_encoder; +use crate::send_with_retry::compression::{add_headers, compress}; use crate::send_with_retry::{ send_with_retry, CompressionStrategy, RetryStrategy, SendWithRetryResult, }; @@ -233,10 +234,14 @@ impl SendData { capabilities: &C, chunks: u64, payload: Vec, - headers: HeaderMap, + mut headers: HeaderMap, endpoint: Option<&Endpoint>, compression_strategy: CompressionStrategy, ) -> (SendWithRetryResult, u64, u64) { + // Compress here (rather than inside `send_with_retry`) so that the reported + // `bytes_sent` metric reflects the number of bytes actually put on the wire. + let (payload, compression_strategy) = compress(payload, compression_strategy); + add_headers(&mut headers, compression_strategy); #[allow(clippy::unwrap_used)] let payload_len = u64::try_from(payload.len()).unwrap(); ( @@ -246,7 +251,7 @@ impl SendData { payload, &headers, &self.retry_strategy, - compression_strategy, + CompressionStrategy::None, ) .await, payload_len, diff --git a/libdd-trace-utils/src/send_with_retry/mod.rs b/libdd-trace-utils/src/send_with_retry/mod.rs index ff45b23949..30f35db752 100644 --- a/libdd-trace-utils/src/send_with_retry/mod.rs +++ b/libdd-trace-utils/src/send_with_retry/mod.rs @@ -7,7 +7,7 @@ mod retry_strategy; pub use retry_strategy::{RetryBackoffType, RetryStrategy}; -mod compression; +pub(crate) mod compression; pub use compression::CompressionStrategy; use bytes::Bytes; @@ -86,7 +86,15 @@ impl std::error::Error for SendWithRetryError {} /// ); /// let retry_strategy = RetryStrategy::new(3, 10, RetryBackoffType::Exponential, Some(5)); /// let capabilities = libdd_capabilities_impl::NativeCapabilities::new_client(); -/// send_with_retry(&capabilities, &target, payload, &headers, &retry_strategy, CompressionStrategy::None).await +/// send_with_retry( +/// &capabilities, +/// &target, +/// payload, +/// &headers, +/// &retry_strategy, +/// CompressionStrategy::None, +/// ) +/// .await /// # } /// ``` pub async fn send_with_retry( From b2b7fe1ba48a801fc1db8b3b017416278714d541 Mon Sep 17 00:00:00 2001 From: paullegranddc Date: Thu, 9 Jul 2026 16:02:48 +0200 Subject: [PATCH 4/4] fix: assert body also in compression mode --- Cargo.lock | 1 + libdd-data-pipeline/Cargo.toml | 1 + libdd-data-pipeline/src/trace_exporter/mod.rs | 34 ++++++++++++------- 3 files changed, 23 insertions(+), 13 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 7b6e9fbe57..a40f7b9dbb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3026,6 +3026,7 @@ dependencies = [ "tokio-util", "tracing", "uuid", + "zstd", ] [[package]] diff --git a/libdd-data-pipeline/Cargo.toml b/libdd-data-pipeline/Cargo.toml index ac6b0a285b..6323091178 100644 --- a/libdd-data-pipeline/Cargo.toml +++ b/libdd-data-pipeline/Cargo.toml @@ -83,6 +83,7 @@ tokio = { version = "1.23", features = [ "test-util", ], default-features = false } duplicate = "2.0.1" +zstd = { version = "0.13", default-features = false } [features] default = ["https", "telemetry"] diff --git a/libdd-data-pipeline/src/trace_exporter/mod.rs b/libdd-data-pipeline/src/trace_exporter/mod.rs index b40a96e92e..c8bb630394 100644 --- a/libdd-data-pipeline/src/trace_exporter/mod.rs +++ b/libdd-data-pipeline/src/trace_exporter/mod.rs @@ -2440,23 +2440,31 @@ mod tests { fn test_agentless_export_body_shape() { let server = MockServer::start(); let mock_intake = server.mock(|when, then| { + fn check_body(body: &str) -> bool { + body.contains("\"traces\":") + && body.contains("\"spans\":") + && body.contains("\"hostname\":\"h-1\"") + && body.contains("\"languageName\":\"nodejs\"") + && body.contains("\"_dd.compute_stats\":\"1\"") + && body.contains("\"_top_level\":1") + && body.contains("\"_trace_root\":1") + && body.contains("\"parent_id\":\"0000000000000000\"") + } let when = when.method(POST).path("/v1/input"); - // When the `compression` feature is enabled the agentless sender zstd-compresses - // the body, so the JSON substrings are no longer present in the raw request. In that - // case assert on the `content-encoding` header instead; the body shape itself is - // covered by the default (uncompressed) build. #[cfg(feature = "compression")] - let when = when.header("content-encoding", "zstd"); + let when = when.header("content-encoding", "zstd").is_true(|req| { + let Ok(body) = zstd::decode_all(req.body_ref()) else { + return false; + }; + let body = String::from_utf8(body).unwrap(); + check_body(&body) + }); #[cfg(not(feature = "compression"))] let when = when - .body_includes("\"traces\":") - .body_includes("\"spans\":") - .body_includes("\"hostname\":\"h-1\"") - .body_includes("\"languageName\":\"nodejs\"") - .body_includes("\"_dd.compute_stats\":\"1\"") - .body_includes("\"_top_level\":1") - .body_includes("\"_trace_root\":1") - .body_includes("\"parent_id\":\"0000000000000000\""); + .is_true(|req| { + let body = String::from_utf8(req.body_vec()).unwrap(); + check_body(&body) + }); let _ = when; then.status(200).body(""); });