feat(data-pipeline): propagate session ids through telemetry FFI

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

@@ -48,6 +48,22 @@ pub struct TelemetryClientConfig<'a> {
     /// When enabled, sets the DD-Telemetry-Debug-Enabled header to true.
     /// Defaults to false.
     pub debug_enabled: bool,
+
+    /// Stable session identifier. Sent as the `DD-Session-ID` HTTP header on
+    /// telemetry requests. An empty `CharSlice` is treated as unset.
+    pub session_id: CharSlice<'a>,
+
+    /// Identifier of the root process in a hierarchy of instrumented
+    /// processes. Sent as the `DD-Root-Session-ID` HTTP header on telemetry
+    /// requests when distinct from the session id. An empty `CharSlice` is
+    /// treated as unset.
+    pub root_session_id: CharSlice<'a>,
+
+    /// Identifier of the immediate parent process. Sent as the
+    /// `DD-Parent-Session-ID` HTTP header on telemetry requests when
+    /// distinct from the session id. An empty `CharSlice` is treated as
+    /// unset.
+    pub parent_session_id: CharSlice<'a>,
 }
 
 /// The TraceExporterConfig object will hold the configuration properties for the TraceExporter.
@@ -300,13 +316,29 @@ pub unsafe extern "C" fn ddog_trace_exporter_config_enable_telemetry(
     catch_panic!(
         if let Option::Some(config) = config {
             if let Option::Some(telemetry_cfg) = telemetry_cfg {
+                let session_id = match telemetry_cfg.session_id.try_to_string_option() {
+                    Ok(s) => s,
+                    Err(_) => return gen_error!(ErrorCode::InvalidInput),
+                };
+                let root_session_id = match telemetry_cfg.root_session_id.try_to_string_option() {
+                    Ok(s) => s,
+                    Err(_) => return gen_error!(ErrorCode::InvalidInput),
+                };
+                let parent_session_id = match telemetry_cfg.parent_session_id.try_to_string_option()
+                {
+                    Ok(s) => s,
+                    Err(_) => return gen_error!(ErrorCode::InvalidInput),
+                };
                 let cfg = TelemetryConfig {
                     heartbeat: telemetry_cfg.interval,
                     runtime_id: match sanitize_string(telemetry_cfg.runtime_id) {
                         Ok(s) => Some(s),
                         Err(e) => return Some(e),
                     },
                     debug_enabled: telemetry_cfg.debug_enabled,
+                    session_id,
+                    root_session_id,
+                    parent_session_id,
                 };
                 debug!(telemetry_cfg = ?cfg, "Configuring telemetry");
                 config.telemetry_cfg = Some(cfg);
@@ -871,6 +903,9 @@ mod tests {
                     interval: 1000,
                     runtime_id: CharSlice::from("id"),
                     debug_enabled: false,
+                    session_id: CharSlice::default(),
+                    root_session_id: CharSlice::default(),
+                    parent_session_id: CharSlice::default(),
                 }),
             );
             assert_eq!(error.as_ref().unwrap().code, ErrorCode::InvalidArgument);
@@ -888,6 +923,9 @@ mod tests {
                     interval: 1000,
                     runtime_id: CharSlice::from("foo"),
                     debug_enabled: true,
+                    session_id: CharSlice::default(),
+                    root_session_id: CharSlice::default(),
+                    parent_session_id: CharSlice::default(),
                 }),
             );
             assert!(error.is_none());
@@ -903,6 +941,46 @@ mod tests {
                 "foo"
             );
             assert!(cfg.telemetry_cfg.as_ref().unwrap().debug_enabled);
+            // Empty session CharSlices must be treated as unset.
+            assert!(cfg.telemetry_cfg.as_ref().unwrap().session_id.is_none());
+            assert!(cfg
+                .telemetry_cfg
+                .as_ref()
+                .unwrap()
+                .root_session_id
+                .is_none());
+            assert!(cfg
+                .telemetry_cfg
+                .as_ref()
+                .unwrap()
+                .parent_session_id
+                .is_none());
+        }
+    }
+
+    #[test]
+    fn config_telemetry_session_ids_test() {
+        unsafe {
+            let mut cfg = TraceExporterConfig::default();
+            let error = ddog_trace_exporter_config_enable_telemetry(
+                Some(&mut cfg),
+                Some(&TelemetryClientConfig {
+                    interval: 1000,
+                    runtime_id: CharSlice::from("runtime"),
+                    debug_enabled: false,
+                    session_id: CharSlice::from("session-abc"),
+                    root_session_id: CharSlice::from("root-xyz"),
+                    parent_session_id: CharSlice::from("parent-123"),
+                }),
+            );
+            assert!(error.is_none());
+            let telemetry_cfg = cfg.telemetry_cfg.as_ref().unwrap();
+            assert_eq!(telemetry_cfg.session_id.as_deref(), Some("session-abc"));
+            assert_eq!(telemetry_cfg.root_session_id.as_deref(), Some("root-xyz"));
+            assert_eq!(
+                telemetry_cfg.parent_session_id.as_deref(),
+                Some("parent-123")
+            );
         }
     }
 
@@ -1183,6 +1261,7 @@ mod tests {
                     heartbeat: 10000,
                     runtime_id: Some("foo".to_string()),
                     debug_enabled: true,
+                    ..Default::default()
                 }),
                 ..Default::default()
             };

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

@@ -98,6 +98,29 @@ impl TelemetryClientBuilder {
         self
     }
 
+    /// Sets the stable session identifier for the telemetry client. Sent as
+    /// the `DD-Session-ID` HTTP header on telemetry requests.
+    pub fn set_session_id(mut self, id: &str) -> Self {
+        self.config.session_id = Some(id.to_string());
+        self
+    }
+
+    /// Sets the root session identifier for the telemetry client. Sent as
+    /// the `DD-Root-Session-ID` HTTP header on telemetry requests when
+    /// distinct from the session id.
+    pub fn set_root_session_id(mut self, id: &str) -> Self {
+        self.config.root_session_id = Some(id.to_string());
+        self
+    }
+
+    /// Sets the parent session identifier for the telemetry client. Sent as
+    /// the `DD-Parent-Session-ID` HTTP header on telemetry requests when
+    /// distinct from the session id.
+    pub fn set_parent_session_id(mut self, id: &str) -> Self {
+        self.config.parent_session_id = Some(id.to_string());
+        self
+    }
+
     /// Builds the telemetry client.
     pub fn build(self) -> (TelemetryClient, TelemetryWorker) {
         #[allow(clippy::unwrap_used)]
@@ -363,6 +386,34 @@ mod tests {
         );
     }
 
+    #[test]
+    fn builder_session_ids_test() {
+        let builder = TelemetryClientBuilder::default()
+            .set_session_id("session-abc")
+            .set_root_session_id("root-xyz")
+            .set_parent_session_id("parent-123");
+
+        assert_eq!(builder.config.session_id.as_deref(), Some("session-abc"));
+        assert_eq!(builder.config.root_session_id.as_deref(), Some("root-xyz"));
+        assert_eq!(
+            builder.config.parent_session_id.as_deref(),
+            Some("parent-123")
+        );
+    }
+
+    #[test]
+    fn builder_session_ids_unset_by_default_test() {
+        let builder = TelemetryClientBuilder::default()
+            .set_service_name("test_service")
+            .set_language("test_language")
+            .set_language_version("1.0")
+            .set_tracer_version("1.0");
+
+        assert!(builder.config.session_id.is_none());
+        assert!(builder.config.root_session_id.is_none());
+        assert!(builder.config.parent_session_id.is_none());
+    }
+
     #[cfg_attr(miri, ignore)]
     #[test]
     fn api_bytes_test() {

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

@@ -331,6 +331,15 @@ impl TraceExporterBuilder {
                     if let Some(id) = telemetry_config.runtime_id {
                         builder = builder.set_runtime_id(&id);
                     }
+                    if let Some(id) = telemetry_config.session_id {
+                        builder = builder.set_session_id(&id);
+                    }
+                    if let Some(id) = telemetry_config.root_session_id {
+                        builder = builder.set_root_session_id(&id);
+                    }
+                    if let Some(id) = telemetry_config.parent_session_id {
+                        builder = builder.set_parent_session_id(&id);
+                    }
                     Ok(builder.build())
                 });
                 match telemetry {
@@ -547,6 +556,7 @@ mod tests {
                 heartbeat: 1000,
                 runtime_id: None,
                 debug_enabled: false,
+                ..Default::default()
             });
         let exporter = builder.build::<NativeCapabilities>().unwrap();
 

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

@@ -840,6 +840,17 @@ pub struct TelemetryConfig {
     pub heartbeat: u64,
     pub runtime_id: Option<String>,
     pub debug_enabled: bool,
+    /// Stable session identifier for the instrumented process. Sent as the
+    /// `DD-Session-ID` HTTP header on telemetry requests.
+    pub session_id: Option<String>,
+    /// Identifier of the root process in a hierarchy of instrumented
+    /// processes. Sent as the `DD-Root-Session-ID` HTTP header on telemetry
+    /// requests when distinct from `session_id`.
+    pub root_session_id: Option<String>,
+    /// Identifier of the immediate parent process. Sent as the
+    /// `DD-Parent-Session-ID` HTTP header on telemetry requests when
+    /// distinct from `session_id`.
+    pub parent_session_id: Option<String>,
 }
 
 #[allow(missing_docs)]