feat(stats_exporter)!: add process tags to CSS payloads

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

@@ -98,6 +98,7 @@ pub struct TraceExporterConfig {
     client_computed_stats: bool,
     telemetry_cfg: Option<TelemetryConfig>,
     health_metrics_enabled: bool,
+    process_tags: Option<String>,
     test_session_token: Option<String>,
     connection_timeout: Option<u64>,
 }
@@ -389,6 +390,26 @@ pub unsafe extern "C" fn ddog_trace_exporter_config_set_client_computed_stats(
     )
 }
 
+/// Sets the process tags to be included in the stats payload.
+#[no_mangle]
+pub unsafe extern "C" fn ddog_trace_exporter_config_set_process_tags(
+    config: Option<&mut TraceExporterConfig>,
+    process_tags: CharSlice,
+) -> Option<Box<ExporterError>> {
+    catch_panic!(
+        if let Option::Some(handle) = config {
+            handle.process_tags = match sanitize_string(process_tags) {
+                Ok(s) => Some(s),
+                Err(e) => return Some(e),
+            };
+            None
+        } else {
+            gen_error!(ErrorCode::InvalidArgument)
+        },
+        gen_error!(ErrorCode::Panic)
+    )
+}
+
 /// Sets the `X-Datadog-Test-Session-Token` header. Only used for testing with the test agent.
 #[no_mangle]
 pub unsafe extern "C" fn ddog_trace_exporter_config_set_test_session_token(
@@ -456,6 +477,7 @@ pub unsafe extern "C" fn ddog_trace_exporter_new(
                 .set_env(config.env.as_ref().unwrap_or(&"".to_string()))
                 .set_app_version(config.version.as_ref().unwrap_or(&"".to_string()))
                 .set_service(config.service.as_ref().unwrap_or(&"".to_string()))
+                .set_process_tags(config.process_tags.as_deref().unwrap_or(""))
                 .set_input_format(config.input_format)
                 .set_output_format(config.output_format)
                 .set_connection_timeout(config.connection_timeout);
@@ -570,6 +592,7 @@ mod tests {
             assert!(!cfg.compute_stats);
             assert!(cfg.telemetry_cfg.is_none());
             assert!(!cfg.health_metrics_enabled);
+            assert!(cfg.process_tags.is_none());
             assert!(cfg.test_session_token.is_none());
             assert!(cfg.connection_timeout.is_none());
 
@@ -759,6 +782,28 @@ mod tests {
         }
     }
 
+    #[test]
+    fn config_process_tags_test() {
+        unsafe {
+            let error =
+                ddog_trace_exporter_config_set_process_tags(None, CharSlice::from("k:v"));
+            assert_eq!(error.as_ref().unwrap().code, ErrorCode::InvalidArgument);
+
+            ddog_trace_exporter_error_free(error);
+
+            let mut config = Some(TraceExporterConfig::default());
+            let error = ddog_trace_exporter_config_set_process_tags(
+                config.as_mut(),
+                CharSlice::from("key1:val1,key2:val2"),
+            );
+
+            assert_eq!(error, None);
+
+            let cfg = config.unwrap();
+            assert_eq!(cfg.process_tags.as_ref().unwrap(), "key1:val1,key2:val2");
+        }
+    }
+
     #[test]
     fn config_client_computed_stats_test() {
         unsafe {

libdd-data-pipeline/src/stats_exporter.rs

@@ -168,13 +168,13 @@ fn encode_stats_payload(
         sequence,
         stats: buckets,
         git_commit_sha: meta.git_commit_sha.clone(),
+        process_tags: meta.process_tags.clone(),
         // These fields are unused or will be set by the Agent
         service: String::new(),
         container_id: String::new(),
         tags: Vec::new(),
         agent_aggregation: String::new(),
         image_tag: String::new(),
-        process_tags: String::new(),
         process_tags_hash: 0,
     }
 }
@@ -217,6 +217,7 @@ mod tests {
             language: "rust".into(),
             tracer_version: "0.0.0".into(),
             runtime_id: "e39d6d12-0752-489f-b488-cf80006c0378".into(),
+            process_tags: "key1:value1,key2:value2".into(),
             ..Default::default()
         }
     }
@@ -257,7 +258,8 @@ mod tests {
                 when.method(POST)
                     .header("Content-type", "application/msgpack")
                     .path("/v0.6/stats")
-                    .body_includes("libdatadog-test");
+                    .body_includes("libdatadog-test")
+                    .body_includes("key1:value1,key2:value2");
                 then.status(200).body("");
             })
             .await;
@@ -318,7 +320,8 @@ mod tests {
                 when.method(POST)
                     .header("Content-type", "application/msgpack")
                     .path("/v0.6/stats")
-                    .body_includes("libdatadog-test");
+                    .body_includes("libdatadog-test")
+                    .body_includes("key1:value1,key2:value2");
                 then.status(200).body("");
             })
             .await;
@@ -356,7 +359,8 @@ mod tests {
                 when.method(POST)
                     .header("Content-type", "application/msgpack")
                     .path("/v0.6/stats")
-                    .body_includes("libdatadog-test");
+                    .body_includes("libdatadog-test")
+                    .body_includes("key1:value1,key2:value2");
                 then.status(200).body("");
             })
             .await;

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

@@ -34,6 +34,7 @@ pub struct TraceExporterBuilder {
     language_interpreter: String,
     language_interpreter_vendor: String,
     git_commit_sha: String,
+    process_tags: String,
     input_format: TraceExporterInputFormat,
     output_format: TraceExporterOutputFormat,
     dogstatsd_url: Option<String>,
@@ -111,6 +112,11 @@ impl TraceExporterBuilder {
         self
     }
 
+    pub fn set_process_tags(&mut self, process_tags: &str) -> &mut Self {
+        process_tags.clone_into(&mut self.process_tags);
+        self
+    }
+
     /// Set the `Datadog-Meta-Tracer-Version` header
     pub fn set_tracer_version(&mut self, tracer_version: &str) -> &mut Self {
         tracer_version.clone_into(&mut self.tracer_version);
@@ -309,6 +315,7 @@ impl TraceExporterBuilder {
                 language_interpreter_vendor: self.language_interpreter_vendor,
                 language: self.language,
                 git_commit_sha: self.git_commit_sha,
+                process_tags: self.process_tags,
                 client_computed_stats: self.client_computed_stats,
                 client_computed_top_level: self.client_computed_top_level,
                 hostname: self.hostname,

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

@@ -128,6 +128,7 @@ pub struct TracerMetadata {
     pub language_interpreter: String,
     pub language_interpreter_vendor: String,
     pub git_commit_sha: String,
+    pub process_tags: String,
     pub client_computed_stats: bool,
     pub client_computed_top_level: bool,
 }

libdd-data-pipeline/tests/snapshots/stats_with_process_tags_snapshot_test.json

@@ -0,0 +1,115 @@
+[[
+  {
+    "name": "stats_process_tags_test_03",
+    "service": "test-service",
+    "resource": "test-resource",
+    "trace_id": 0,
+    "span_id": 1,
+    "parent_id": 0,
+    "type": "web",
+    "meta": {
+      "_dd.origin": "cloudfunction",
+      "env": "test-env",
+      "functionname": "dummy_function_name",
+      "origin": "cloudfunction",
+      "runtime-id": "test-runtime-id-value",
+      "service": "test-service"
+    },
+    "metrics": {
+      "_top_level": 1.0
+    },
+    "duration": 5,
+    "start": 0
+  },
+     {
+       "name": "stats_process_tags_test_01",
+       "service": "test-service",
+       "resource": "test-resource",
+       "trace_id": 0,
+       "span_id": 2,
+       "parent_id": 1,
+       "meta": {
+         "env": "test-env",
+         "runtime-id": "test-runtime-id-value",
+         "service": "test-service"
+       },
+       "metrics": {
+         "_dd_metric1": 1.0,
+         "_dd_metric2": 2.0
+       },
+       "span_events": [
+         {
+           "time_unix_nano": 1727211691770715042,
+           "name": "test_span"
+         },
+         {
+           "time_unix_nano": 1727211691770716000,
+           "name": "exception",
+           "attributes": {
+             "exception.count": {
+               "type": 2,
+               "int_value": 1
+             },
+             "exception.escaped": {
+               "type": 1,
+               "bool_value": true
+             },
+             "exception.lines": {
+               "type": 4,
+               "array_value": {
+                 "values": [
+                   {
+                     "type": 0,
+                     "string_value": "  File \"<string>\", line 1, in <module>"
+                   },
+                   {
+                     "type": 0,
+                     "string_value": "  File \"<string>\", line 1, in divide"
+                   }
+                 ]
+               }
+             },
+             "exception.message": {
+               "type": 0,
+               "string_value": "Cannot divide by zero"
+             },
+             "exception.version": {
+               "type": 3,
+               "double_value": 4.2
+             }
+           }
+         }
+       ],
+       "duration": 5,
+       "start": 1
+     },
+     {
+       "name": "stats_process_tags_test_02",
+       "service": "test-service",
+       "resource": "test-resource",
+       "trace_id": 0,
+       "span_id": 3,
+       "parent_id": 1,
+       "meta": {
+         "env": "test-env",
+         "runtime-id": "test-runtime-id-value",
+         "service": "test-service"
+       },
+       "span_links": [
+         {
+           "trace_id": 13930160852258120406,
+           "trace_id_high": 0,
+           "span_id": 11788048577503494824,
+           "attributes": {
+             "link.name": "Job #123"
+           }
+         },
+         {
+           "trace_id": 12184699170068370717,
+           "trace_id_high": 5943849730528564606,
+           "span_id": 13874630024467741450
+         }
+       ],
+       "duration": 5,
+       "start": 1
+     }]]

libdd-data-pipeline/tests/snapshots/stats_with_process_tags_snapshot_test_tracestats.json

@@ -0,0 +1,22 @@
+[
+  {
+    "Start": 0,
+    "Duration": 10000000000,
+    "Stats": [
+      {
+        "Name": "stats_process_tags_test_03",
+        "Resource": "test-resource",
+        "Service": "test-service",
+        "Type": "web",
+        "HTTPStatusCode": 0,
+        "Synthetics": false,
+        "Hits": 1,
+        "TopLevelHits": 1,
+        "Duration": 5,
+        "Errors": 0,
+        "OkSummary": 37,
+        "ErrorSummary": 24
+      }
+    ]
+  }
+]

libdd-data-pipeline/tests/test_trace_exporter.rs

@@ -1,4 +1,4 @@
 // Copyright 2024-Present Datadog, Inc. https://www.datadoghq.com/
 // SPDX-License-Identifier: Apache-2.0
 #[cfg(test)]
 mod tracing_integration_tests {
@@ -10,6 +10,7 @@
     use libdd_trace_utils::test_utils::datadog_test_agent::DatadogTestAgent;
     use libdd_trace_utils::test_utils::{create_test_json_span, create_test_v05_span};
     use serde_json::json;
+    use std::time::Duration;
     use tokio::task;
 
     fn get_v04_trace_snapshot_test_payload(name_prefix: &str) -> Vec<u8> {
@@ -327,4 +328,53 @@
 
         test_agent.assert_snapshot(snapshot_name).await;
     }
+
+    #[cfg_attr(miri, ignore)]
+    #[tokio::test]
+    async fn stats_snapshot_test() {
+        let relative_snapshot_path = "libdd-data-pipeline/tests/snapshots/";
+        let snapshot_name = "stats_snapshot_test";
+        let test_agent = DatadogTestAgent::new(Some(relative_snapshot_path), None, &[]).await;
+        let url = test_agent.get_base_uri().await;
+        test_agent.start_session(snapshot_name, None).await;
+
+        let task_result = task::spawn_blocking(move || {
+            let mut builder = TraceExporter::builder();
+            builder
+                .set_url(url.to_string().as_ref())
+                .set_language("test-lang")
+                .set_language_version("2.0")
+                .set_language_interpreter_vendor("vendor")
+                .set_language_interpreter("interpreter")
+                .set_tracer_version("1.0")
+                .set_env("test_env")
+                .set_service("test")
+                .set_hostname("test-host")
+                .set_process_tags("key1:val1,key2:val2")
+                .set_test_session_token(snapshot_name)
+                .enable_stats(Duration::from_secs(10));
+
+            let trace_exporter = builder.build().expect("Unable to build TraceExporter");
+
+            // Give the info-fetcher worker time to fetch agent info on its first iteration.
+            // Once the cache is populated, the first send() call will enable stats computation
+            // via check_agent_info() and spans will be recorded in the concentrator.
+            std::thread::sleep(Duration::from_secs(1));
+
+            let data = get_v04_trace_snapshot_test_payload("stats_test");
+            trace_exporter
+                .send(data.as_ref())
+                .expect("Failed to send traces");
+
+            // shutdown force-flushes and sends any remaining stats
+            trace_exporter
+                .shutdown(Some(Duration::from_secs(5)))
+                .expect("Failed to shutdown exporter");
+        })
+        .await;
+
+        assert!(task_result.is_ok());
+
+        test_agent.assert_snapshot(snapshot_name).await;
+    }
 }