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

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,9 @@ 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};
+#[cfg(all(not(target_arch = "wasm32"), feature = "telemetry"))]
+use libdd_telemetry::worker::TelemetryWorkerHandle;
 use crate::trace_exporter::agent_response::AgentResponsePayloadVersion;
 use crate::trace_exporter::error::BuilderErrorKind;
 use crate::trace_exporter::log_writer::DEFAULT_LOG_MAX_LINE_SIZE;
@@ -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,7 +648,12 @@ impl<R: SharedRuntime> TraceExporterBuilder<R> {
         }
 
         #[cfg(all(not(target_arch = "wasm32"), feature = "telemetry"))]
-        let (telemetry_client, telemetry_handle) = {
+        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

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/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();

libdd-telemetry/src/data/payloads.rs

@@ -1,6 +1,7 @@
 // Copyright 2021-Present Datadog, Inc. https://www.datadoghq.com/
 // SPDX-License-Identifier: Apache-2.0
 
+use std::collections::HashMap;
 use std::hash::Hasher;
 
 use crate::data::metrics;
@@ -11,6 +12,16 @@ use serde::{Deserialize, Serialize};
 pub struct Dependency {
     pub name: String,
     pub version: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub hash: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub metadata: Option<Vec<DependencyMetadata>>,
+}
+
+#[derive(Serialize, Deserialize, Debug, Hash, PartialEq, Eq, Clone, Default)]
+pub struct DependencyMetadata {
+    pub r#type: String,
+    pub value: String,
 }
 
 #[derive(Serialize, Deserialize, Debug, Hash, PartialEq, Eq, Clone, Default)]
@@ -36,27 +47,61 @@ pub struct Configuration {
 #[serde(rename_all = "snake_case")]
 pub enum ConfigurationOrigin {
     EnvVar,
+    OtelEnvVar,
     Code,
     DdConfig,
     RemoteConfig,
     Default,
     LocalStableConfig,
     FleetStableConfig,
     Calculated,
+    Unknown,
+}
+
+#[derive(Serialize, Deserialize, Debug, PartialEq, Eq, Clone)]
+pub struct Error {
+    pub code: Option<i64>,
+    pub message: Option<String>,
 }
 
 #[derive(Serialize, Debug)]
 pub struct AppStarted {
     pub configuration: Vec<Configuration>,
     pub dependencies: Vec<Dependency>,
     pub integrations: Vec<Integration>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub install_signature: Option<InstallSignature>,
+    #[serde(skip_serializing_if = "HashMap::is_empty")]
+    pub products: HashMap<String, ProductState>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub error: Option<Error>,
+}
+
+#[derive(Serialize, Debug, Clone)]
+pub struct InstallSignature {
+    pub install_id: Option<String>,
+    pub install_type: Option<String>,
+    pub install_time: Option<String>,
 }
 
 #[derive(Serialize, Debug)]
 pub struct AppDependenciesLoaded {
     pub dependencies: Vec<Dependency>,
 }
 
+#[derive(Serialize, Deserialize, Debug, PartialEq, Eq, Clone)]
+pub struct ProductState {
+    pub enabled: bool,
+    pub version: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub error: Option<Error>,
+}
+
+#[derive(Serialize, Debug)]
+pub struct AppProductChange {
+    pub products: HashMap<String, ProductState>,
+}
+
 #[derive(Serialize, Debug)]
 pub struct AppIntegrationsChange {
     pub integrations: Vec<Integration>,

libdd-telemetry/src/lib.rs

@@ -18,6 +18,9 @@ pub mod info;
 pub mod metrics;
 pub mod worker;
 
+pub use libdd_common::tag::{parse_tags, Tag};
+pub use libdd_common::{parse_uri, Endpoint};
+
 pub fn build_host() -> data::Host {
     debug!("Building telemetry host information");
     let hostname = info::os::real_hostname().unwrap_or_else(|_| String::from("unknown_hostname"));