feat(telemetry)!: Add Installation signature and AppProduct changes payloads

datadog-sidecar-ffi/src/lib.rs

@@ -515,6 +515,8 @@ pub unsafe extern "C" fn ddog_sidecar_telemetry_addDependency(
     let dependency = TelemetryActions::AddDependency(Dependency {
         name: dependency_name.to_utf8_lossy().into_owned(),
         version,
+        hash: None,
+        metadata: None,
     });
 
     try_c!(blocking::enqueue_actions(

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

@@ -157,6 +157,16 @@ pub struct TelemetryClient {
     worker: TelemetryWorkerHandle,
 }
 
+impl TelemetryClient {
+    /// Allow sharing a telemetry worker with data-pipeline
+    pub fn with_handle(handle: TelemetryWorkerHandle) -> Self {
+        TelemetryClient {
+            metrics: Metrics::new(&handle),
+            worker: handle,
+        }
+    }
+}
+
 /// Telemetry describing the sending of a trace payload
 /// It can be produced from a [`SendWithRetryResult`] or from a [`SendDataResult`].
 #[derive(PartialEq, Debug, Default)]

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

@@ -6,7 +6,7 @@ use crate::agentless::config::{AgentlessTraceConfig, DEFAULT_AGENTLESS_TIMEOUT};
 use crate::otlp::config::{OtlpProtocol, DEFAULT_OTLP_TIMEOUT};
 use crate::otlp::{OtlpMetricsConfig, OtlpResourceInfo, OtlpTraceConfig};
 #[cfg(all(not(target_arch = "wasm32"), feature = "telemetry"))]
-use crate::telemetry::TelemetryClientBuilder;
+use crate::telemetry::{TelemetryClient, TelemetryClientBuilder};
 use crate::trace_exporter::agent_response::AgentResponsePayloadVersion;
 use crate::trace_exporter::error::BuilderErrorKind;
 use crate::trace_exporter::log_writer::DEFAULT_LOG_MAX_LINE_SIZE;
@@ -26,6 +26,8 @@ use libdd_dogstatsd_client::new;
 use libdd_shared_runtime::SharedRuntime;
 #[cfg(not(target_arch = "wasm32"))]
 use libdd_shared_runtime::{BlockingRuntime, ForkSafeRuntime};
+#[cfg(all(not(target_arch = "wasm32"), feature = "telemetry"))]
+use libdd_telemetry::worker::TelemetryWorkerHandle;
 use libdd_trace_utils::trace_filter::TraceFilterer;
 use std::sync::Arc;
 use std::time::Duration;
@@ -80,6 +82,8 @@ pub struct TraceExporterBuilder<R: SharedRuntime> {
     client_side_stats_obfuscation_enabled: bool,
     #[cfg(feature = "telemetry")]
     telemetry: Option<TelemetryConfig>,
+    #[cfg(feature = "telemetry")]
+    telemetry_handle: Option<TelemetryWorkerHandle>,
     telemetry_instrumentation_sessions: TelemetryInstrumentationSessions,
     shared_runtime: Option<Arc<R>>,
     health_metrics_enabled: bool,
@@ -148,6 +152,8 @@ impl<R: SharedRuntime> TraceExporterBuilder<R> {
             client_side_stats_obfuscation_enabled: false,
             #[cfg(feature = "telemetry")]
             telemetry: None,
+            #[cfg(feature = "telemetry")]
+            telemetry_handle: None,
             telemetry_instrumentation_sessions: TelemetryInstrumentationSessions::default(),
             shared_runtime: None,
             health_metrics_enabled: false,
@@ -356,6 +362,12 @@ impl<R: SharedRuntime> TraceExporterBuilder<R> {
         self
     }
 
+    #[cfg(feature = "telemetry")]
+    pub fn set_telemetry_handle(&mut self, handle: TelemetryWorkerHandle) -> &mut Self {
+        self.telemetry_handle = Some(handle);
+        self
+    }
+
     /// Sets optional instrumentation session headers on telemetry requests (`dd-session-id`, etc.).
     pub fn set_telemetry_instrumentation_sessions(
         &mut self,
@@ -636,54 +648,59 @@ impl<R: SharedRuntime> TraceExporterBuilder<R> {
         }
 
         #[cfg(all(not(target_arch = "wasm32"), feature = "telemetry"))]
-        let (telemetry_client, telemetry_handle) = {
-            let sessions = self.telemetry_instrumentation_sessions;
-            let telemetry = self
-                .telemetry
-                .filter(|_| {
-                    // no agent endpoint to talk to, so we skip the
-                    // telemetry worker
-                    !(agentless_enabled || self.output_to_log)
-                })
-                .map(|telemetry_config| {
-                    let mut builder = TelemetryClientBuilder::default()
-                        .set_language(&self.language)
-                        .set_language_version(&self.language_version)
-                        .set_service_name(&self.service)
-                        .set_service_version(&self.app_version)
-                        .set_env(&self.env)
-                        .set_tracer_version(&self.tracer_version)
-                        .set_heartbeat(telemetry_config.heartbeat)
-                        .set_url(base_url)
-                        .set_debug_enabled(telemetry_config.debug_enabled);
-                    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);
+        let (telemetry_client, telemetry_handle) =
+            if let Some(shared_handle) = self.telemetry_handle {
+                // Consolidated path: report health metrics through the externally-owned worker.
+                // We do not own it (no spawn, no start, no shutdown handle).
+                (Some(TelemetryClient::with_handle(shared_handle)), None)
+            } else {
+                let sessions = self.telemetry_instrumentation_sessions;
+                let telemetry = self
+                    .telemetry
+                    .filter(|_| {
+                        // no agent endpoint to talk to, so we skip the
+                        // telemetry worker
+                        !(agentless_enabled || self.output_to_log)
+                    })
+                    .map(|telemetry_config| {
+                        let mut builder = TelemetryClientBuilder::default()
+                            .set_language(&self.language)
+                            .set_language_version(&self.language_version)
+                            .set_service_name(&self.service)
+                            .set_service_version(&self.app_version)
+                            .set_env(&self.env)
+                            .set_tracer_version(&self.tracer_version)
+                            .set_heartbeat(telemetry_config.heartbeat)
+                            .set_url(base_url)
+                            .set_debug_enabled(telemetry_config.debug_enabled);
+                        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);
+                        }
+                        Ok(builder.build())
+                    });
+                match telemetry {
+                    Some(Ok((client_tel, worker))) => {
+                        let handle = shared_runtime.spawn_worker(worker, false).map_err(|e| {
+                            TraceExporterError::Builder(BuilderErrorKind::InvalidConfiguration(
+                                e.to_string(),
+                            ))
+                        })?;
+                        client_tel.start().await;
+                        (Some(client_tel), Some(handle))
                     }
-                    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);
-                    }
-                    Ok(builder.build())
-                });
-            match telemetry {
-                Some(Ok((client_tel, worker))) => {
-                    let handle = shared_runtime.spawn_worker(worker, false).map_err(|e| {
-                        TraceExporterError::Builder(BuilderErrorKind::InvalidConfiguration(
-                            e.to_string(),
-                        ))
-                    })?;
-                    client_tel.start().await;
-                    (Some(client_tel), Some(handle))
+                    Some(Err(e)) => return Err(e),
+                    None => (None, None),
                 }
-                Some(Err(e)) => return Err(e),
-                None => (None, None),
-            }
-        };
+            };
 
         // Transport selection: agentless is mutually exclusive with both OTLP and a
         // user-supplied agent URL; OTLP and the agent URL may coexist. All exclusion

libdd-telemetry-ffi/src/worker_handle.rs

@@ -21,7 +21,7 @@ pub unsafe extern "C" fn ddog_telemetry_handle_add_dependency(
 ) -> MaybeError {
     let name = crate::try_c!(dependency_name.try_to_string());
     let version = crate::try_c!(dependency_version.try_to_string_option());
-    crate::try_c!(handle.add_dependency(name, version));
+    crate::try_c!(handle.add_dependency(name, version, None));
     MaybeError::None
 }
 

libdd-telemetry/examples/tm-ping.rs

@@ -19,6 +19,9 @@ fn build_app_started_payload() -> AppStarted {
         configuration: Vec::new(),
         dependencies: Vec::new(),
         integrations: Vec::new(),
+        install_signature: None,
+        products: Default::default(),
+        error: None,
     }
 }
 

libdd-telemetry/src/config.rs

@@ -58,6 +58,15 @@ pub struct Config {
     pub parent_session_id: Option<String>,
     #[serde(default)]
     pub root_session_id: Option<String>,
+
+    /// Whether to emit the `app-started`/`app-closing` lifecycle payloads.
+    /// Forked processes may not be required to emit these.
+    #[serde(default = "default_true")]
+    pub emit_app_lifecycle: bool,
+}
+
+fn default_true() -> bool {
+    true
 }
 
 fn endpoint_with_telemetry_path(
@@ -195,6 +204,7 @@ impl Default for Config {
             session_id: None,
             parent_session_id: None,
             root_session_id: None,
+            emit_app_lifecycle: true,
         }
     }
 }
@@ -324,6 +334,7 @@ impl Config {
             session_id: None,
             parent_session_id: None,
             root_session_id: None,
+            emit_app_lifecycle: true,
         };
 
         _ = this.set_endpoint(TelemetryEndpoint {

libdd-telemetry/src/data/metrics.rs

@@ -51,6 +51,9 @@ pub enum MetricNamespace {
     Telemetry,
     Apm,
     Sidecar,
+    Civisibility,
+    Mlobs,
+    Ddtraceapi,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, Hash)]

libdd-telemetry/src/data/payload.rs

@@ -11,6 +11,7 @@ pub enum Payload {
     AppStarted(AppStarted),
     AppDependenciesLoaded(AppDependenciesLoaded),
     AppIntegrationsChange(AppIntegrationsChange),
+    AppProductChange(AppProductChange),
     AppClientConfigurationChange(AppClientConfigurationChange),
     AppEndpoints(AppEndpoints),
     AppHeartbeat(#[serde(skip_serializing)] ()),
@@ -29,6 +30,7 @@ impl Payload {
             AppStarted(_) => "app-started",
             AppDependenciesLoaded(_) => "app-dependencies-loaded",
             AppIntegrationsChange(_) => "app-integrations-change",
+            AppProductChange(_) => "app-product-change",
             AppClientConfigurationChange(_) => "app-client-configuration-change",
             AppEndpoints(_) => "app-endpoints",
             AppHeartbeat(_) => "app-heartbeat",
@@ -68,6 +70,9 @@ mod tests {
             ],
             dependencies: Vec::new(),
             integrations: Vec::new(),
+            install_signature: None,
+            products: Default::default(),
+            error: None,
         });
 
         let serialized = serde_json::to_value(&payload).unwrap();
@@ -106,10 +111,11 @@ mod tests {
                 Dependency {
                     name: "tokio".to_string(),
                     version: Some("1.32.0".to_string()),
+                    ..Default::default()
                 },
                 Dependency {
                     name: "serde".to_string(),
-                    version: None,
+                    ..Default::default()
                 },
             ],
         });
@@ -484,6 +490,100 @@ mod tests {
         assert_eq!(serialized, expected);
     }
 
+    #[test]
+    fn test_app_product_change_serialization() {
+        let mut products = std::collections::HashMap::new();
+        products.insert(
+            "appsec".to_string(),
+            ProductState {
+                enabled: true,
+                version: Some("1.2.3".to_string()),
+                error: None,
+            },
+        );
+        let payload = Payload::AppProductChange(AppProductChange { products });
+
+        let serialized = serde_json::to_value(&payload).unwrap();
+
+        let expected = json!({
+            "request_type": "app-product-change",
+            "payload": {
+                "products": {
+                    "appsec": {
+                        "enabled": true,
+                        "version": "1.2.3"
+                    }
+                }
+            }
+        });
+
+        assert_eq!(serialized, expected);
+    }
+
+    #[test]
+    fn test_dependency_metadata_serialization() {
+        // A plain dependency (no metadata) omits both `hash` and `metadata`.
+        let plain = Payload::AppDependenciesLoaded(AppDependenciesLoaded {
+            dependencies: vec![Dependency {
+                name: "requests".to_string(),
+                version: Some("2.0".to_string()),
+                ..Default::default()
+            }],
+        });
+        assert_eq!(
+            serde_json::to_value(&plain).unwrap(),
+            json!({
+                "request_type": "app-dependencies-loaded",
+                "payload": { "dependencies": [{ "name": "requests", "version": "2.0" }] }
+            })
+        );
+
+        // With SCA metadata: per the telemetry schema, `type` names the metadata kind and
+        // `value` is an opaque stringified-JSON payload that consumers must parse.
+        let with_sca = Payload::AppDependenciesLoaded(AppDependenciesLoaded {
+            dependencies: vec![Dependency {
+                name: "requests".to_string(),
+                version: Some("2.0".to_string()),
+                metadata: Some(vec![DependencyMetadata {
+                    r#type: "reachability".to_string(),
+                    value: "{\"id\":\"CVE-2024-1\",\"reached\":true}".to_string(),
+                }]),
+                ..Default::default()
+            }],
+        });
+        assert_eq!(
+            serde_json::to_value(&with_sca).unwrap(),
+            json!({
+                "request_type": "app-dependencies-loaded",
+                "payload": { "dependencies": [{
+                    "name": "requests",
+                    "version": "2.0",
+                    "metadata": [{
+                        "type": "reachability",
+                        "value": "{\"id\":\"CVE-2024-1\",\"reached\":true}"
+                    }]
+                }] }
+            })
+        );
+
+        // SCA enabled with no findings: `metadata` is present-but-empty (distinct from omitted).
+        let empty_meta = Payload::AppDependenciesLoaded(AppDependenciesLoaded {
+            dependencies: vec![Dependency {
+                name: "requests".to_string(),
+                version: Some("2.0".to_string()),
+                metadata: Some(vec![]),
+                ..Default::default()
+            }],
+        });
+        assert_eq!(
+            serde_json::to_value(&empty_meta).unwrap(),
+            json!({
+                "request_type": "app-dependencies-loaded",
+                "payload": { "dependencies": [{ "name": "requests", "version": "2.0", "metadata": [] }] }
+            })
+        );
+    }
+
     #[test]
     fn test_app_extended_heartbeat_serialization() {
         let payload = Payload::AppExtendedHeartbeat(AppStarted {
@@ -496,6 +596,9 @@ mod tests {
             }],
             dependencies: Vec::new(),
             integrations: Vec::new(),
+            install_signature: None,
+            products: Default::default(),
+            error: None,
         });
 
         let serialized = serde_json::to_value(&payload).unwrap();