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

libdd-data-pipeline/Cargo.toml

@@ -35,6 +35,7 @@ libdd-telemetry = { version = "4.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" }
 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" }
 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 = [

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

libdd-data-pipeline/src/stats_exporter.rs

@@ -31,6 +31,7 @@ pub struct StatsExporter {
     sequence_id: AtomicU64,
     cancellation_token: CancellationToken,
     client: HttpClient,
+    obfuscation_active: bool,
 }
 
 impl StatsExporter {
@@ -49,6 +50,7 @@ impl StatsExporter {
         endpoint: Endpoint,
         cancellation_token: CancellationToken,
         client: HttpClient,
+        obfuscation_active: bool,
     ) -> Self {
         Self {
             flush_interval,
@@ -58,6 +60,7 @@ impl StatsExporter {
             sequence_id: AtomicU64::new(0),
             cancellation_token,
             client,
+            obfuscation_active,
         }
     }
 
@@ -90,6 +93,13 @@ impl StatsExporter {
             libdd_common::header::APPLICATION_MSGPACK,
         );
 
+        if self.obfuscation_active {
+            headers.insert(
+                http::HeaderName::from_static("datadog-obfuscation-version"),
+                http::HeaderValue::from_static("1"),
+            );
+        }
+
         let result = send_with_retry(
             &self.client,
             &self.endpoint,
@@ -274,6 +284,7 @@ mod tests {
             Endpoint::from_url(stats_url_from_agent_url(&server.url("/")).unwrap()),
             CancellationToken::new(),
             new_default_client(),
+            false,
         );
 
         let send_status = stats_exporter.send(true).await;
@@ -302,6 +313,7 @@ mod tests {
             Endpoint::from_url(stats_url_from_agent_url(&server.url("/")).unwrap()),
             CancellationToken::new(),
             new_default_client(),
+            false,
         );
 
         let send_status = stats_exporter.send(true).await;
@@ -336,6 +348,7 @@ mod tests {
             Endpoint::from_url(stats_url_from_agent_url(&server.url("/")).unwrap()),
             CancellationToken::new(),
             new_default_client(),
+            false,
         );
 
         tokio::time::pause();
@@ -378,6 +391,7 @@ mod tests {
             Endpoint::from_url(stats_url_from_agent_url(&server.url("/")).unwrap()),
             cancellation_token.clone(),
             new_default_client(),
+            false,
         );
 
         tokio::spawn(async move {
@@ -415,4 +429,35 @@ mod tests {
             "Non-empty env should be preserved"
         );
     }
+    #[cfg_attr(miri, ignore)]
+    #[tokio::test]
+    async fn test_send_stats_with_obfuscation_header() {
+        let server = MockServer::start_async().await;
+
+        let mock = server
+            .mock_async(|when, then| {
+                when.method(POST)
+                    .header("Content-type", "application/msgpack")
+                    .header("datadog-obfuscation-version", "1")
+                    .path("/v0.6/stats")
+                    .body_includes("libdatadog-test");
+                then.status(200).body("");
+            })
+            .await;
+
+        let stats_exporter = StatsExporter::new(
+            BUCKETS_DURATION,
+            Arc::new(Mutex::new(get_test_concentrator())),
+            get_test_metadata(),
+            Endpoint::from_url(stats_url_from_agent_url(&server.url("/")).unwrap()),
+            CancellationToken::new(),
+            new_default_client(),
+            true,
+        );
+
+        let send_status = stats_exporter.send(true).await;
+        send_status.unwrap();
+
+        mock.assert_async().await;
+    }
 }

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

@@ -430,7 +430,9 @@ impl TraceExporter {
                         );
                     }
                     StatsComputationStatus::Enabled {
-                        stats_concentrator, ..
+                        stats_concentrator,
+                        cancellation_token,
+                        obfuscation_active,
                     } => {
                         let ctx = stats::StatsContext {
                             metadata: &self.metadata,
@@ -441,6 +443,8 @@ impl TraceExporter {
                             &ctx,
                             &agent_info,
                             stats_concentrator,
+                            cancellation_token,
+                            *obfuscation_active,
                             &self.client_side_stats,
                             &self.workers,
                         );