add session id support to trace export

mabdinur · mabdinur · commit d01347e48d9e · 2026-03-30T14:32:07.000-04:00
diff --git a/examples/ffi/trace_exporter.c b/examples/ffi/trace_exporter.c
@@ -99,7 +99,10 @@ int main(int argc, char** argv)
     ddog_TelemetryClientConfig telemetry_config = {
         .interval = 60000,
         .runtime_id = DDOG_CHARSLICE_C("12345678-1234-1234-1234-123456789abc"),
-        .debug_enabled = true
+        .debug_enabled = true,
+        .session_id = DDOG_CHARSLICE_C("12345678-1234-1234-1234-123456789abc"),
+        .root_session_id = DDOG_CHARSLICE_C("87654321-1234-1234-1234-123456789abc"),
+        .parent_session_id = DDOG_CHARSLICE_C(""),
     };
 
     ret = ddog_trace_exporter_config_enable_telemetry(config, &telemetry_config);
diff --git a/libdd-data-pipeline-ffi/src/trace_exporter.rs b/libdd-data-pipeline-ffi/src/trace_exporter.rs
@@ -10,7 +10,8 @@ use libdd_common_ffi::{
 };
 
 use libdd_data_pipeline::trace_exporter::{
-    TelemetryConfig, TraceExporter, TraceExporterInputFormat, TraceExporterOutputFormat,
+    TelemetryConfig, TelemetryInstrumentationSessions, TraceExporter, TraceExporterInputFormat,
+    TraceExporterOutputFormat,
 };
 use std::{ptr::NonNull, time::Duration};
 use tracing::{debug, error};
@@ -43,6 +44,13 @@ pub struct TelemetryClientConfig<'a> {
     /// When enabled, sets the DD-Telemetry-Debug-Enabled header to true.
     /// Defaults to false.
     pub debug_enabled: bool,
+
+    /// HTTP header `dd-session-id` (empty = omitted).
+    pub session_id: CharSlice<'a>,
+    /// HTTP header `dd-root-session-id` (empty = omitted).
+    pub root_session_id: CharSlice<'a>,
+    /// HTTP header `dd-parent-session-id` (empty = omitted).
+    pub parent_session_id: CharSlice<'a>,
 }
 
 /// The TraceExporterConfig object will hold the configuration properties for the TraceExporter.
@@ -64,6 +72,7 @@ pub struct TraceExporterConfig {
     compute_stats: bool,
     client_computed_stats: bool,
     telemetry_cfg: Option<TelemetryConfig>,
+    telemetry_instrumentation_sessions: TelemetryInstrumentationSessions,
     health_metrics_enabled: bool,
     process_tags: Option<String>,
     test_session_token: Option<String>,
@@ -302,8 +311,23 @@ pub unsafe extern "C" fn ddog_trace_exporter_config_enable_telemetry(
                     },
                     debug_enabled: telemetry_cfg.debug_enabled,
                 };
-                debug!(telemetry_cfg = ?cfg, "Configuring telemetry");
+                let sessions = TelemetryInstrumentationSessions {
+                    session_id: match sanitize_string(telemetry_cfg.session_id) {
+                        Ok(s) => Some(s),
+                        Err(e) => return Some(e),
+                    },
+                    root_session_id: match sanitize_string(telemetry_cfg.root_session_id) {
+                        Ok(s) => Some(s),
+                        Err(e) => return Some(e),
+                    },
+                    parent_session_id: match sanitize_string(telemetry_cfg.parent_session_id) {
+                        Ok(s) => Some(s),
+                        Err(e) => return Some(e),
+                    },
+                };
+                debug!(telemetry_cfg = ?cfg, telemetry_sessions = ?sessions, "Configuring telemetry");
                 config.telemetry_cfg = Some(cfg);
+                config.telemetry_instrumentation_sessions = sessions;
             }
             None
         } else {
@@ -488,6 +512,9 @@ pub unsafe extern "C" fn ddog_trace_exporter_new(
             if let Some(cfg) = &config.telemetry_cfg {
                 builder.enable_telemetry(cfg.clone());
             }
+            builder.set_telemetry_instrumentation_sessions(
+                config.telemetry_instrumentation_sessions.clone(),
+            );
 
             if let Some(token) = &config.test_session_token {
                 builder.set_test_session_token(token);
@@ -831,6 +858,9 @@ mod tests {
                     interval: 1000,
                     runtime_id: CharSlice::from("id"),
                     debug_enabled: false,
+                    session_id: CharSlice::empty(),
+                    root_session_id: CharSlice::empty(),
+                    parent_session_id: CharSlice::empty(),
                 }),
             );
             assert_eq!(error.as_ref().unwrap().code, ErrorCode::InvalidArgument);
@@ -848,6 +878,9 @@ mod tests {
                     interval: 1000,
                     runtime_id: CharSlice::from("foo"),
                     debug_enabled: true,
+                    session_id: CharSlice::empty(),
+                    root_session_id: CharSlice::empty(),
+                    parent_session_id: CharSlice::empty(),
                 }),
             );
             assert!(error.is_none());
@@ -863,6 +896,40 @@ mod tests {
                 "foo"
             );
             assert!(cfg.telemetry_cfg.as_ref().unwrap().debug_enabled);
+            assert_eq!(
+                cfg.telemetry_instrumentation_sessions.session_id.as_deref(),
+                Some("")
+            );
+            assert_eq!(
+                cfg.telemetry_instrumentation_sessions
+                    .root_session_id
+                    .as_deref(),
+                Some("")
+            );
+            assert_eq!(
+                cfg.telemetry_instrumentation_sessions
+                    .parent_session_id
+                    .as_deref(),
+                Some("")
+            );
+
+            let mut cfg = TraceExporterConfig::default();
+            let error = ddog_trace_exporter_config_enable_telemetry(
+                Some(&mut cfg),
+                Some(&TelemetryClientConfig {
+                    interval: 500,
+                    runtime_id: CharSlice::from("rid"),
+                    debug_enabled: false,
+                    session_id: CharSlice::from("sess-z"),
+                    root_session_id: CharSlice::from("root-z"),
+                    parent_session_id: CharSlice::from("par-z"),
+                }),
+            );
+            assert!(error.is_none());
+            let s = &cfg.telemetry_instrumentation_sessions;
+            assert_eq!(s.session_id.as_deref(), Some("sess-z"));
+            assert_eq!(s.root_session_id.as_deref(), Some("root-z"));
+            assert_eq!(s.parent_session_id.as_deref(), Some("par-z"));
         }
     }
 
diff --git a/libdd-data-pipeline/src/telemetry/mod.rs b/libdd-data-pipeline/src/telemetry/mod.rs
@@ -93,6 +93,24 @@ impl TelemetryClientBuilder {
         self
     }
 
+    /// `dd-session-id` header (with non-empty session id).
+    pub fn set_session_id(mut self, id: &str) -> Self {
+        self.config.session_id = Some(id.to_string());
+        self
+    }
+
+    /// `dd-root-session-id` (omitted if equal to session id).
+    pub fn set_root_session_id(mut self, id: &str) -> Self {
+        self.config.root_session_id = Some(id.to_string());
+        self
+    }
+
+    /// `dd-parent-session-id` (omitted if equal to session id).
+    pub fn set_parent_session_id(mut self, id: &str) -> Self {
+        self.config.parent_session_id = Some(id.to_string());
+        self
+    }
+
     /// Sets the debug enabled flag for the telemetry client.
     pub fn set_debug_enabled(mut self, debug: bool) -> Self {
         self.config.debug_enabled = debug;
@@ -808,53 +826,17 @@ mod tests {
 
     #[cfg_attr(miri, ignore)]
     #[tokio::test]
-    async fn runtime_id_test() {
-        let server = MockServer::start_async().await;
-
-        let telemetry_srv = server
-            .mock_async(|when, then| {
-                when.method(POST).body_includes(r#""runtime_id":"foo""#);
-                then.status(200).body("");
-            })
-            .await;
-
-        let (client, mut worker) = TelemetryClientBuilder::default()
-            .set_service_name("test_service")
-            .set_service_version("test_version")
-            .set_env("test_env")
-            .set_language("test_language")
-            .set_language_version("test_language_version")
-            .set_tracer_version("test_tracer_version")
-            .set_url(&server.url("/"))
-            .set_heartbeat(100)
-            .set_runtime_id("foo")
-            .build(Handle::current());
-        tokio::spawn(async move { worker.run().await });
-
-        client.start().await;
-        client
-            .send(&SendPayloadTelemetry {
-                requests_count: 1,
-                ..Default::default()
-            })
-            .unwrap();
-        client.shutdown().await;
-        while telemetry_srv.calls_async().await == 0 {
-            sleep(Duration::from_millis(10)).await;
-        }
-        // One payload generate-metrics
-        telemetry_srv.assert_calls_async(1).await;
-    }
-
-    #[cfg_attr(miri, ignore)]
-    #[tokio::test]
-    async fn application_metadata_test() {
+    async fn runtime_id_and_session_headers_test() {
         let server = MockServer::start_async().await;
 
         let telemetry_srv = server
             .mock_async(|when, then| {
                 when.method(POST)
-                    .body_includes(r#""application":{"service_name":"test_service","service_version":"test_version","env":"test_env","language_name":"test_language","language_version":"test_language_version","tracer_version":"test_tracer_version"}"#);
+                    .body_includes(r#""runtime_id":"foo""#)
+                    .body_includes(r#""application":{"service_name":"test_service","service_version":"test_version","env":"test_env","language_name":"test_language","language_version":"test_language_version","tracer_version":"test_tracer_version"}"#)
+                    .header("dd-session-id", "sess-e2e")
+                    .header("dd-root-session-id", "root-e2e")
+                    .header("dd-parent-session-id", "parent-e2e");
                 then.status(200).body("");
             })
             .await;
@@ -869,6 +851,9 @@ mod tests {
             .set_url(&server.url("/"))
             .set_heartbeat(100)
             .set_runtime_id("foo")
+            .set_session_id("sess-e2e")
+            .set_root_session_id("root-e2e")
+            .set_parent_session_id("parent-e2e")
             .build(Handle::current());
         tokio::spawn(async move { worker.run().await });
 
diff --git a/libdd-data-pipeline/src/trace_exporter/builder.rs b/libdd-data-pipeline/src/trace_exporter/builder.rs
@@ -9,9 +9,9 @@ use crate::telemetry::TelemetryClientBuilder;
 use crate::trace_exporter::agent_response::AgentResponsePayloadVersion;
 use crate::trace_exporter::error::BuilderErrorKind;
 use crate::trace_exporter::{
-    add_path, StatsComputationStatus, TelemetryConfig, TraceExporter, TraceExporterError,
-    TraceExporterInputFormat, TraceExporterOutputFormat, TraceExporterWorkers, TracerMetadata,
-    INFO_ENDPOINT,
+    add_path, StatsComputationStatus, TelemetryConfig, TelemetryInstrumentationSessions,
+    TraceExporter, TraceExporterError, TraceExporterInputFormat, TraceExporterOutputFormat,
+    TraceExporterWorkers, TracerMetadata, INFO_ENDPOINT,
 };
 use arc_swap::ArcSwap;
 use libdd_common::http_common::new_default_client;
@@ -49,6 +49,7 @@ pub struct TraceExporterBuilder {
     compute_stats_by_span_kind: bool,
     peer_tags: Vec<String>,
     telemetry: Option<TelemetryConfig>,
+    telemetry_instrumentation_sessions: TelemetryInstrumentationSessions,
     health_metrics_enabled: bool,
     test_session_token: Option<String>,
     agent_rates_payload_version_enabled: bool,
@@ -209,6 +210,15 @@ impl TraceExporterBuilder {
         self
     }
 
+    /// Sets optional instrumentation session headers on telemetry requests (`dd-session-id`, etc.).
+    pub fn set_telemetry_instrumentation_sessions(
+        &mut self,
+        sessions: TelemetryInstrumentationSessions,
+    ) -> &mut Self {
+        self.telemetry_instrumentation_sessions = sessions;
+        self
+    }
+
     /// Enables health metrics emission.
     pub fn enable_health_metrics(&mut self) -> &mut Self {
         self.health_metrics_enabled = true;
@@ -294,6 +304,7 @@ impl TraceExporterBuilder {
             stats = StatsComputationStatus::DisabledByAgent { bucket_size };
         }
 
+        let sessions = self.telemetry_instrumentation_sessions.clone();
         let telemetry = self.telemetry.map(|telemetry_config| {
             let mut builder = TelemetryClientBuilder::default()
                 .set_language(&self.language)
@@ -308,6 +319,15 @@ impl TraceExporterBuilder {
             if let Some(id) = telemetry_config.runtime_id {
                 builder = builder.set_runtime_id(&id);
             }
+            if let Some(ref id) = sessions.session_id {
+                builder = builder.set_session_id(id);
+            }
+            if let Some(ref id) = sessions.root_session_id {
+                builder = builder.set_root_session_id(id);
+            }
+            if let Some(ref id) = sessions.parent_session_id {
+                builder = builder.set_parent_session_id(id);
+            }
             builder.build(runtime.handle().clone())
         });
 
diff --git a/libdd-data-pipeline/src/trace_exporter/mod.rs b/libdd-data-pipeline/src/trace_exporter/mod.rs
@@ -10,6 +10,14 @@ mod trace_serializer;
 // Re-export the builder
 pub use builder::TraceExporterBuilder;
 
+/// Values for optional telemetry HTTP session headers (`dd-session-id`, root/parent).
+#[derive(Debug, Default, Clone)]
+pub struct TelemetryInstrumentationSessions {
+    pub session_id: Option<String>,
+    pub root_session_id: Option<String>,
+    pub parent_session_id: Option<String>,
+}
+
 use self::agent_response::AgentResponse;
 use self::metrics::MetricsEmitter;
 use self::stats::StatsComputationStatus;
