feat(data-pipeline)!: add compression option for agentless export

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 = []

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")]

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" }

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"]
@@ -111,3 +112,5 @@ fips = [
 ]
 test-utils = []
 regex-lite = ["libdd-common/regex-lite"]
+# Enable zstd compression for the agentless trace intake sender.
+compression = ["libdd-trace-utils/compression"]

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<C: HttpClientCapability + SleepCapabilit
         None,
     );
 
-    match send_with_retry(capabilities, &target, json_body, &headers, &retry_strategy).await {
+    #[cfg(feature = "compression")]
+    let compression_strategy = CompressionStrategy::Zstd { level: 1 };
+    #[cfg(not(feature = "compression"))]
+    let compression_strategy = CompressionStrategy::None;
+
+    match send_with_retry(
+        capabilities,
+        &target,
+        json_body,
+        &headers,
+        &retry_strategy,
+        compression_strategy,
+    )
+    .await
+    {
         Ok(_) => Ok(()),
         Err(e) => Err(map_send_error(e)),
     }

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<C: HttpClientCapability + SleepCapability>(
         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),
     }

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;
 
@@ -2270,16 +2271,32 @@ 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")
-                .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\"");
+            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");
+            #[cfg(feature = "compression")]
+            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
+                .is_true(|req| {
+                let body = String::from_utf8(req.body_vec()).unwrap();
+                check_body(&body)
+            });
+            let _ = when;
             then.status(200).body("");
         });
 

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"]

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<Cap: HttpClientCapability + SleepCapability, Con: FlushableConcentrator>
             body,
             &headers,
             &RetryStrategy::default(),
+            CompressionStrategy::None,
         )
         .await;
 

libdd-trace-utils/Cargo.toml

@@ -76,7 +76,9 @@ getrandom = { version = "0.2", features = ["js"] }
 
 [dev-dependencies]
 libdd-capabilities-impl = { version = "3.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"]