From 857e4f86a3c74ece1cafe53057e627263ff31f6a Mon Sep 17 00:00:00 2001 From: Bob Weinand Date: Wed, 8 Jul 2026 20:23:39 +0200 Subject: [PATCH] Add Installation signature and AppProduct changes payloads And some smaller stuff like more metric namespaces. --- libdd-data-pipeline/src/telemetry/mod.rs | 10 ++ .../src/trace_exporter/builder.rs | 21 +++- libdd-telemetry-ffi/src/worker_handle.rs | 2 +- libdd-telemetry/src/config.rs | 11 ++ libdd-telemetry/src/data/metrics.rs | 3 + libdd-telemetry/src/data/payload.rs | 105 ++++++++++++++++- libdd-telemetry/src/data/payloads.rs | 45 ++++++++ libdd-telemetry/src/lib.rs | 3 + libdd-telemetry/src/worker/mod.rs | 107 ++++++++++++++++-- 9 files changed, 293 insertions(+), 14 deletions(-) diff --git a/libdd-data-pipeline/src/telemetry/mod.rs b/libdd-data-pipeline/src/telemetry/mod.rs index f4c186e079..28989b0d71 100644 --- a/libdd-data-pipeline/src/telemetry/mod.rs +++ b/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)] diff --git a/libdd-data-pipeline/src/trace_exporter/builder.rs b/libdd-data-pipeline/src/trace_exporter/builder.rs index 5582c934cd..b7132b61bf 100644 --- a/libdd-data-pipeline/src/trace_exporter/builder.rs +++ b/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 { client_side_stats_obfuscation_enabled: bool, #[cfg(feature = "telemetry")] telemetry: Option, + #[cfg(feature = "telemetry")] + telemetry_handle: Option, telemetry_instrumentation_sessions: TelemetryInstrumentationSessions, shared_runtime: Option>, health_metrics_enabled: bool, @@ -148,6 +152,8 @@ impl TraceExporterBuilder { 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 TraceExporterBuilder { 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 TraceExporterBuilder { } #[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 diff --git a/libdd-telemetry-ffi/src/worker_handle.rs b/libdd-telemetry-ffi/src/worker_handle.rs index 16380a971c..ab76a1abc7 100644 --- a/libdd-telemetry-ffi/src/worker_handle.rs +++ b/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 } diff --git a/libdd-telemetry/src/config.rs b/libdd-telemetry/src/config.rs index 026d4bc8ef..54bb7d3f3c 100644 --- a/libdd-telemetry/src/config.rs +++ b/libdd-telemetry/src/config.rs @@ -58,6 +58,15 @@ pub struct Config { pub parent_session_id: Option, #[serde(default)] pub root_session_id: Option, + + /// 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 { diff --git a/libdd-telemetry/src/data/metrics.rs b/libdd-telemetry/src/data/metrics.rs index e131298a1a..28f7ed398f 100644 --- a/libdd-telemetry/src/data/metrics.rs +++ b/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)] diff --git a/libdd-telemetry/src/data/payload.rs b/libdd-telemetry/src/data/payload.rs index 609969ce38..815d12a995 100644 --- a/libdd-telemetry/src/data/payload.rs +++ b/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(); diff --git a/libdd-telemetry/src/data/payloads.rs b/libdd-telemetry/src/data/payloads.rs index fb5c97dc93..17c6800f1b 100644 --- a/libdd-telemetry/src/data/payloads.rs +++ b/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, + #[serde(skip_serializing_if = "Option::is_none")] + pub hash: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub metadata: Option>, +} + +#[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,6 +47,7 @@ pub struct Configuration { #[serde(rename_all = "snake_case")] pub enum ConfigurationOrigin { EnvVar, + OtelEnvVar, Code, DdConfig, RemoteConfig, @@ -43,6 +55,13 @@ pub enum ConfigurationOrigin { LocalStableConfig, FleetStableConfig, Calculated, + Unknown, +} + +#[derive(Serialize, Deserialize, Debug, PartialEq, Eq, Clone)] +pub struct Error { + pub code: Option, + pub message: Option, } #[derive(Serialize, Debug)] @@ -50,6 +69,19 @@ pub struct AppStarted { pub configuration: Vec, pub dependencies: Vec, pub integrations: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + pub install_signature: Option, + #[serde(skip_serializing_if = "HashMap::is_empty")] + pub products: HashMap, + #[serde(skip_serializing_if = "Option::is_none")] + pub error: Option, +} + +#[derive(Serialize, Debug, Clone)] +pub struct InstallSignature { + pub install_id: Option, + pub install_type: Option, + pub install_time: Option, } #[derive(Serialize, Debug)] @@ -57,6 +89,19 @@ pub struct AppDependenciesLoaded { pub dependencies: Vec, } +#[derive(Serialize, Deserialize, Debug, PartialEq, Eq, Clone)] +pub struct ProductState { + pub enabled: bool, + pub version: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub error: Option, +} + +#[derive(Serialize, Debug)] +pub struct AppProductChange { + pub products: HashMap, +} + #[derive(Serialize, Debug)] pub struct AppIntegrationsChange { pub integrations: Vec, diff --git a/libdd-telemetry/src/lib.rs b/libdd-telemetry/src/lib.rs index 09b4390f08..6dc9672f98 100644 --- a/libdd-telemetry/src/lib.rs +++ b/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")); diff --git a/libdd-telemetry/src/worker/mod.rs b/libdd-telemetry/src/worker/mod.rs index b4c4bc1955..95aeed01f5 100644 --- a/libdd-telemetry/src/worker/mod.rs +++ b/libdd-telemetry/src/worker/mod.rs @@ -7,7 +7,10 @@ pub mod store; use crate::{ config::Config, - data::{self, Application, Dependency, Endpoint, Host, Integration, Log, Payload, Telemetry}, + data::{ + self, Application, Dependency, Endpoint, Host, Integration, Log, Payload, ProductState, + Telemetry, + }, metrics::{ContextKey, MetricBuckets, MetricContexts}, }; @@ -78,7 +81,7 @@ macro_rules! telemetry_worker_log { $($arg)* ); if $worker.config.telemetry_debug_logging_enabled { - println!(concat!("{}: Telemetry worker DEBUG: ", $fmt_str), time_now(), $($arg)*); + eprintln!(concat!("{}: Telemetry worker DEBUG: ", $fmt_str), time_now(), $($arg)*); } } }; @@ -90,6 +93,7 @@ pub enum TelemetryActions { AddConfig(data::Configuration), AddDependency(Dependency), AddIntegration(Integration), + AddProductChange((String, ProductState)), AddLog((LogIdentifier, Log)), AddEndpoint(Endpoint), Lifecycle(LifecycleAction), @@ -124,11 +128,14 @@ struct TelemetryWorkerData { configurations: store::Store, integrations: store::Store, endpoints: HashSet, + products: std::collections::HashMap, + products_pending: HashSet, logs: store::QueueHashMap, metric_contexts: MetricContexts, metric_buckets: MetricBuckets, host: Host, app: Application, + install_signature: Option, } pub struct TelemetryWorker { @@ -223,6 +230,8 @@ impl Worker for TelemetryWorker { self.data.integrations.clear(); self.data.configurations.clear(); self.data.endpoints.clear(); + self.data.products.clear(); + self.data.products_pending.clear(); } async fn shutdown(&mut self) { @@ -381,6 +390,7 @@ impl TelemetryWorker { AddConfig(_) | AddDependency(_) | AddIntegration(_) + | AddProductChange(_) | AddEndpoint(_) | Lifecycle(ExtendedHeartbeat) => {} Lifecycle(Stop) => { @@ -423,10 +433,12 @@ impl TelemetryWorker { match action { Lifecycle(Start) => { if !self.data.started { - let app_started = data::Payload::AppStarted(self.build_app_started()); - match self.send_payload(&app_started).await { - Ok(()) => self.payload_sent_success(&app_started), - Err(err) => self.log_err(&err), + if self.config.emit_app_lifecycle { + let app_started = data::Payload::AppStarted(self.build_app_started()); + match self.send_payload(&app_started).await { + Ok(()) => self.payload_sent_success(&app_started), + Err(err) => self.log_err(&err), + } } #[allow(clippy::unwrap_used)] @@ -449,6 +461,10 @@ impl TelemetryWorker { } AddDependency(dep) => self.data.dependencies.insert(dep), AddIntegration(integration) => self.data.integrations.insert(integration), + AddProductChange((name, state)) => { + self.data.products.insert(name.clone(), state); + self.data.products_pending.insert(name); + } AddConfig(cfg) => self.data.configurations.insert(cfg), AddEndpoint(endpoint) => { self.data.endpoints.insert(endpoint); @@ -511,6 +527,21 @@ impl TelemetryWorker { Ok(()) => self.payload_sent_success(&extended_hb), Err(err) => self.log_err(&err), } + + if !self.data.products.is_empty() { + let products = self + .data + .products + .iter() + .map(|(name, state)| (name.clone(), state.clone())) + .collect(); + let product_change = + data::Payload::AppProductChange(data::AppProductChange { products }); + match self.send_payload(&product_change).await { + Ok(()) => self.payload_sent_success(&product_change), + Err(err) => self.log_err(&err), + } + } // Only re-schedule self. Resetting `FlushData` here would replace its // existing deadline with `now + heartbeat_interval`, starving FlushData // when `extended_heartbeat_interval < heartbeat_interval` because each @@ -527,7 +558,9 @@ impl TelemetryWorker { self.data.metric_buckets.flush_aggregates(); let mut app_events = self.build_app_events_batch(); - app_events.push(data::Payload::AppClosing(())); + if self.config.emit_app_lifecycle { + app_events.push(data::Payload::AppClosing(())); + } let observability_events = self.build_observability_batch(); @@ -588,6 +621,22 @@ impl TelemetryWorker { }, )) } + if !self.data.products_pending.is_empty() { + let products = self + .data + .products_pending + .iter() + .filter_map(|name| { + self.data + .products + .get(name) + .map(|state| (name.clone(), state.clone())) + }) + .collect(); + payloads.push(data::Payload::AppProductChange(data::AppProductChange { + products, + })) + } if self.data.configurations.flush_not_empty() { payloads.push(data::Payload::AppClientConfigurationChange( data::AppClientConfigurationChange { @@ -687,6 +736,9 @@ impl TelemetryWorker { configuration: self.data.configurations.unflushed().cloned().collect(), dependencies: self.data.dependencies.unflushed().cloned().collect(), integrations: self.data.integrations.unflushed().cloned().collect(), + install_signature: self.data.install_signature.clone(), + products: self.data.products.clone(), + error: None, } } @@ -709,6 +761,11 @@ impl TelemetryWorker { AppIntegrationsChange(p) => { self.data.integrations.removed_flushed(p.integrations.len()) } + AppProductChange(p) => { + for name in p.products.keys() { + self.data.products_pending.remove(name); + } + } AppClientConfigurationChange(p) => self .data .configurations @@ -1039,15 +1096,40 @@ impl TelemetryWorkerHandle { self.wait_for_shutdown() } - pub fn add_dependency(&self, name: String, version: Option) -> anyhow::Result<()> { + pub fn add_dependency( + &self, + name: String, + version: Option, + metadata: Option>, + ) -> anyhow::Result<()> { self.sender .try_send(TelemetryActions::AddDependency(Dependency { name, version, + hash: None, + metadata, }))?; Ok(()) } + pub fn add_product_change( + &self, + product: String, + enabled: bool, + version: Option, + ) -> anyhow::Result<()> { + self.sender + .try_send(TelemetryActions::AddProductChange(( + product, + ProductState { + enabled, + version, + error: None, + }, + )))?; + Ok(()) + } + pub fn add_integration( &self, name: String, @@ -1141,6 +1223,7 @@ pub struct TelemetryWorkerBuilder { pub rust_shared_lib_deps: bool, pub config: Config, pub flavor: TelemetryWorkerFlavor, + pub install_signature: Option, } impl TelemetryWorkerBuilder { @@ -1192,6 +1275,7 @@ impl TelemetryWorkerBuilder { rust_shared_lib_deps: false, config: Config::default(), flavor: TelemetryWorkerFlavor::default(), + install_signature: None, } } @@ -1228,11 +1312,14 @@ impl TelemetryWorkerBuilder { integrations: self.integrations, configurations: self.configurations, endpoints: self.endpoints, + products: std::collections::HashMap::new(), + products_pending: HashSet::new(), logs: store::QueueHashMap::default(), metric_contexts: contexts.clone(), metric_buckets: MetricBuckets::default(), host: self.host, app: self.application, + install_signature: self.install_signature, }, config, mailbox, @@ -1620,7 +1707,7 @@ mod tests { // Populate every data field that reset() should clear. worker.data.dependencies.insert(Dependency { name: "dep".to_string(), - version: None, + ..Default::default() }); worker.data.integrations.insert(Integration { name: "integration".to_string(), @@ -1707,7 +1794,7 @@ mod tests { handle .try_send_msg(TelemetryActions::AddDependency(Dependency { name: "dep".to_string(), - version: None, + ..Default::default() })) .unwrap(); let (id, log) = make_log(1, "pre-fork log");