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/Cargo.toml

@@ -26,6 +26,8 @@ http-body-util = "0.1"
 http = "1"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = { version = "1.0" }
+strum = { version = "0.26", default-features = false }
+strum_macros = "0.26"
 tokio = { version = "1.23", features = ["sync", "io-util"] }
 tokio-util = { version = "0.7", features = ["codec"] }
 tracing = { version = "0.1", default-features = false }

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

@@ -36,8 +36,20 @@ pub enum SerializedSketch {
     B64 { sketch_b64: String },
 }
 
-#[derive(Serialize, Deserialize, Debug, Clone, Copy)]
+#[derive(
+    Serialize,
+    Deserialize,
+    Debug,
+    Clone,
+    Copy,
+    PartialEq,
+    Eq,
+    Hash,
+    strum_macros::Display,
+    strum_macros::EnumIter,
+)]
 #[serde(rename_all = "snake_case")]
+#[strum(serialize_all = "snake_case")]
 #[repr(C)]
 pub enum MetricNamespace {
     Tracers,
@@ -51,13 +63,29 @@ pub enum MetricNamespace {
     Telemetry,
     Apm,
     Sidecar,
+    Civisibility,
+    Mlobs,
+    Ddtraceapi,
 }
 
-#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, Hash)]
+#[derive(
+    Serialize,
+    Deserialize,
+    Debug,
+    Clone,
+    Copy,
+    PartialEq,
+    Eq,
+    Hash,
+    strum_macros::Display,
+    strum_macros::EnumIter,
+)]
 #[serde(rename_all = "snake_case")]
+#[strum(serialize_all = "snake_case")]
 #[repr(C)]
 pub enum MetricType {
     Gauge,
     Count,
+    Rate,
     Distribution,
 }