feat!: integrate obfuscation to the stats exporter [APMSP-2764]

Cargo.toml

@@ -77,6 +77,7 @@ license = "Apache-2.0"
 authors = ["Datadog Inc. <info@datadoghq.com>"]
 
 [workspace.dependencies]
+arc-swap = "1.7.1"
 hyper = { version = "1.6", features = [
   "http1",
   "client",

datadog-sidecar/Cargo.toml

@@ -12,9 +12,11 @@ bench = false
 default = ["tracing"]
 tracing = ["tracing/std", "tracing-log", "tracing-subscriber"]
 tokio-console = ["tokio/full", "tokio/tracing", "console-subscriber"]
+stats-obfuscation = ["libdd-trace-stats/stats-obfuscation", "libdd-data-pipeline/stats-obfuscation", "arc-swap"]
 
 [dependencies]
 anyhow = { version = "1.0" }
+arc-swap = { workspace = true, optional = true }
 arrayref = "0.3.7"
 priority-queue = "2.1.1"
 libdd-common = { path = "../libdd-common" }
@@ -24,7 +26,7 @@ datadog-sidecar-macros = { path = "../datadog-sidecar-macros" }
 libdd-telemetry = { path = "../libdd-telemetry", features = ["tracing"] }
 libdd-data-pipeline = { path = "../libdd-data-pipeline" }
 libdd-trace-utils = { path = "../libdd-trace-utils" }
-libdd-trace-stats = { path = "../libdd-trace-stats" }
+libdd-trace-stats = { path = "../libdd-trace-stats", default-features=false, features = ["https"] }
 datadog-remote-config = { path = "../datadog-remote-config" , features = ["live-debugger"]}
 datadog-live-debugger = { path = "../datadog-live-debugger" }
 libdd-crashtracker = { path = "../libdd-crashtracker" }

datadog-sidecar/src/service/stats_flusher.rs

@@ -110,6 +110,14 @@ fn make_exporter(
         s.meta.clone(),
         endpoint,
         NativeCapabilities::new_client(),
+        // Sidecar does not perform client-side stats obfuscation. Pass a disabled
+        // default so the `datadog-obfuscation-version` header is never sent.
+        #[cfg(feature = "stats-obfuscation")]
+        Arc::new(arc_swap::ArcSwap::from_pointee(
+            libdd_trace_stats::span_concentrator::StatsComputationObfuscationConfig::default(),
+        )),
+        #[cfg(feature = "stats-obfuscation")]
+        "0",
     )
 }
 

libdd-crashtracker/src/receiver/ptrace_collector.rs

@@ -31,9 +31,9 @@ use std::ptr;
 use std::time::{Duration, Instant};
 
 use libdd_libunwind_sys::{
-    UnwAddrSpaceT, UnwCursor, UnwWord, _UPT_accessors, _UPT_create, _UPT_destroy,
-    unw_create_addr_space, unw_destroy_addr_space, unw_get_proc_name_remote, unw_get_reg_remote,
-    unw_init_remote, unw_step_remote, UNW_REG_IP, UNW_REG_SP,
+    _UPT_accessors, _UPT_create, _UPT_destroy, unw_create_addr_space, unw_destroy_addr_space,
+    unw_get_proc_name_remote, unw_get_reg_remote, unw_init_remote, unw_step_remote, UnwAddrSpaceT,
+    UnwCursor, UnwWord, UNW_REG_IP, UNW_REG_SP,
 };
 
 use crate::crash_info::{StackFrame, StackTrace};

libdd-data-pipeline/Cargo.toml

@@ -12,8 +12,8 @@ autobenches = false
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+arc-swap.workspace = true
 anyhow = { version = "1.0" }
-arc-swap = "1.7.1"
 async-trait = "0.1"
 http = "1"
 http-body-util = "0.1"
@@ -38,6 +38,8 @@ libdd-telemetry = { version = "5.0.0", path = "../libdd-telemetry", default-feat
 libdd-trace-protobuf = { version = "3.0.1", path = "../libdd-trace-protobuf" }
 libdd-trace-stats = { version = "2.0.0", path = "../libdd-trace-stats", default-features = false }
 libdd-trace-utils = { version = "3.0.1", path = "../libdd-trace-utils", default-features = false }
+libdd-trace-obfuscation = { version = "2.0.0", path = "../libdd-trace-obfuscation", optional = true }
+libdd-ddsketch = { version = "1.0.1", path = "../libdd-ddsketch" }
 libdd-dogstatsd-client = { version = "2.0.0", path = "../libdd-dogstatsd-client", default-features = false }
 libdd-tinybytes = { version = "1.1.0", path = "../libdd-tinybytes", features = [
     "bytes_string",
@@ -91,6 +93,10 @@ https = [
     "libdd-trace-utils/https",
     "libdd-dogstatsd-client/https",
 ]
+stats-obfuscation = [
+    "libdd-trace-obfuscation",
+    "libdd-trace-stats/stats-obfuscation"
+]
 fips = [
     "libdd-common/fips",
     "libdd-capabilities-impl/fips",

libdd-data-pipeline/src/agent_info/fetcher.rs

@@ -408,12 +408,12 @@ mod single_threaded_tests {
                         },
                         "remove_stack_traces": false,
                         "redis": {
-                                "Enabled": true,
-                                "RemoveAllArgs": false
+                                "enabled": true,
+                                "remove_all_args": false
                         },
                         "memcached": {
-                                "Enabled": true,
-                                "KeepCommand": false
+                                "enabled": true,
+                                "keep_command": false
                         }
                 }
         },
@@ -424,7 +424,7 @@ mod single_threaded_tests {
         format!("{:x}", Sha256::digest(json.as_bytes()))
     }
 
-    const TEST_INFO_HASH: &str = "b7709671827946c15603847bca76c90438579c038ec134eae19c51f1f3e3dfea";
+    const TEST_INFO_HASH: &str = "cce54bf6e7d1bf38088a3ec809bfeec160bc52d37f70bd6b581ce3c2f7be5a65";
 
     #[cfg_attr(miri, ignore)]
     #[tokio::test]

libdd-data-pipeline/src/agent_info/schema.rs

@@ -35,6 +35,8 @@ pub struct AgentInfoStruct {
     pub peer_tags: Option<Vec<String>>,
     /// List of span kinds eligible for stats computation
     pub span_kinds_stats_computed: Option<Vec<String>>,
+    /// Obfuscation version supported by the agent for client-side stats
+    pub obfuscation_version: Option<u32>,
     /// Container tags hash from HTTP response header
     pub container_tags_hash: Option<String>,
     /// Exact-match tag filters applied before stats computation (root span only).
@@ -69,37 +71,41 @@ pub struct Config {
     pub max_memory: Option<f64>,
     pub max_cpu: Option<f64>,
     pub analyzed_spans_by_service: Option<HashMap<String, HashMap<String, f64>>>,
+    pub obfuscation: Option<ObfuscationConfig>,
 }
 
 #[allow(missing_docs)]
-#[derive(Clone, Deserialize, Default, Debug, PartialEq)]
+#[derive(Clone, Serialize, Deserialize, Default, Debug, PartialEq)]
 pub struct ObfuscationConfig {
     pub elastic_search: bool,
     pub mongo: bool,
     pub sql_exec_plan: bool,
     pub sql_exec_plan_normalize: bool,
+    #[cfg(feature = "stats-obfuscation")]
+    // Option because it might not exist with old agents
+    pub sql_obfuscation_mode: Option<libdd_trace_obfuscation::sql::SqlObfuscationMode>,
     pub http: HttpObfuscationConfig,
     pub remove_stack_traces: bool,
     pub redis: RedisObfuscationConfig,
     pub memcached: MemcachedObfuscationConfig,
 }
 
 #[allow(missing_docs)]
-#[derive(Clone, Deserialize, Default, Debug, PartialEq)]
+#[derive(Clone, Serialize, Deserialize, Default, Debug, PartialEq)]
 pub struct HttpObfuscationConfig {
     pub remove_query_string: bool,
     pub remove_path_digits: bool,
 }
 
 #[allow(missing_docs)]
-#[derive(Clone, Deserialize, Default, Debug, PartialEq)]
+#[derive(Clone, Serialize, Deserialize, Default, Debug, PartialEq)]
 pub struct RedisObfuscationConfig {
     pub enabled: bool,
     pub remove_all_args: bool,
 }
 
 #[allow(missing_docs)]
-#[derive(Clone, Deserialize, Default, Debug, PartialEq)]
+#[derive(Clone, Serialize, Deserialize, Default, Debug, PartialEq)]
 pub struct MemcachedObfuscationConfig {
     pub enabled: bool,
     pub keep_command: bool,

libdd-data-pipeline/src/lib.rs

@@ -3,6 +3,7 @@
 #![cfg_attr(not(test), deny(clippy::panic))]
 #![cfg_attr(not(test), deny(clippy::unwrap_used))]
 #![cfg_attr(not(test), deny(clippy::expect_used))]
+#![cfg_attr(not(test), deny(clippy::unreachable))]
 #![cfg_attr(not(test), deny(clippy::todo))]
 #![cfg_attr(not(test), deny(clippy::unimplemented))]
 

libdd-data-pipeline/src/trace_exporter/builder.rs

@@ -53,6 +53,8 @@ pub struct TraceExporterBuilder {
     peer_tags_aggregation: bool,
     compute_stats_by_span_kind: bool,
     peer_tags: Vec<String>,
+    #[cfg(feature = "stats-obfuscation")]
+    client_side_stats_obfuscation_enabled: bool,
     #[cfg(feature = "telemetry")]
     telemetry: Option<TelemetryConfig>,
     telemetry_instrumentation_sessions: TelemetryInstrumentationSessions,
@@ -211,6 +213,17 @@ impl TraceExporterBuilder {
         self
     }
 
+    /// Enable client-side stats obfuscation. Disabled by default.
+    ///
+    /// Final activation also requires the agent to advertise a supported
+    /// `obfuscation_version` via the `/info` endpoint. When disabled, no
+    /// `datadog-obfuscation-version` header is sent on stats payloads.
+    #[cfg(feature = "stats-obfuscation")]
+    pub fn enable_client_side_stats_obfuscation(&mut self) -> &mut Self {
+        self.client_side_stats_obfuscation_enabled = true;
+        self
+    }
+
     #[cfg(feature = "telemetry")]
     /// Enables sending telemetry metrics.
     pub fn enable_telemetry(&mut self, cfg: TelemetryConfig) -> &mut Self {
@@ -324,6 +337,11 @@ impl TraceExporterBuilder {
         // native, workers run on the tokio runtime; on wasm, they run on the JS
         // event loop via `spawn_local`. Telemetry remains native-only for now.
 
+        #[cfg(feature = "stats-obfuscation")]
+        use libdd_trace_stats::span_concentrator::StatsComputationObfuscationConfig;
+
+        use crate::trace_exporter::stats::StatsComputationConfig;
+
         let info_endpoint = Endpoint::from_url(add_path(&agent_url, INFO_ENDPOINT));
         let (info_fetcher, info_response_observer) =
             AgentInfoFetcher::<C>::new(info_endpoint, Duration::from_secs(5 * 60));
@@ -453,7 +471,15 @@ impl TraceExporterBuilder {
             shared_runtime,
             dogstatsd,
             common_stats_tags: vec![libdatadog_version],
-            client_side_stats: ArcSwap::new(stats.into()),
+            client_side_stats: StatsComputationConfig {
+                status: ArcSwap::new(stats.into()),
+                #[cfg(feature = "stats-obfuscation")]
+                obfuscation_config: Arc::new(ArcSwap::from_pointee(
+                    StatsComputationObfuscationConfig::default(),
+                )),
+                #[cfg(feature = "stats-obfuscation")]
+                obfuscation_enabled: self.client_side_stats_obfuscation_enabled,
+            },
             previous_info_state: arc_swap::ArcSwapOption::new(None),
             info_response_observer,
             #[cfg(all(not(target_arch = "wasm32"), feature = "telemetry"))]