feat(telemetry): allow debug mode for the telemetry client (#1001)

* feat(telemetry): allow debug mode for the telemetry client

* Fixed the typo in `telemetry_hearbeat_interval` to `telemetry_heartbeat_interval` across multiple files.
* Added `debug_enabled` flag to the telemetry configuration.
* Updated request builder to conditionally include the debug header based on the new flag.

Author: Ganesh Jangir

data-pipeline-ffi/src/trace_exporter.rs

@@ -41,6 +41,11 @@ pub struct TelemetryClientConfig<'a> {
     ///   a single instrumented app should share the same runtime_id)
     /// - Be associated with traces to allow correlation between traces and telemetry data
     pub runtime_id: CharSlice<'a>,
+
+    /// Whether to enable debug mode for telemetry.
+    /// When enabled, sets the DD-Telemetry-Debug-Enabled header to true.
+    /// Defaults to false.
+    pub debug_enabled: bool,
 }
 
 /// The TraceExporterConfig object will hold the configuration properties for the TraceExporter.
@@ -244,6 +249,7 @@ pub unsafe extern "C" fn ddog_trace_exporter_config_enable_telemetry(
                     Ok(s) => Some(s),
                     Err(e) => return Some(e),
                 },
+                debug_enabled: telemetry_cfg.debug_enabled,
             })
         } else {
             config.telemetry_cfg = Some(TelemetryConfig::default());
@@ -596,6 +602,7 @@ mod tests {
                 Some(&TelemetryClientConfig {
                     interval: 1000,
                     runtime_id: CharSlice::from("id"),
+                    debug_enabled: false,
                 }),
             );
             assert_eq!(error.as_ref().unwrap().code, ErrorCode::InvalidArgument);
@@ -613,6 +620,7 @@ mod tests {
                 Some(&TelemetryClientConfig {
                     interval: 1000,
                     runtime_id: CharSlice::from("foo"),
+                    debug_enabled: true,
                 }),
             );
             assert!(error.is_none());
@@ -627,6 +635,7 @@ mod tests {
                     .unwrap(),
                 "foo"
             );
+            assert!(cfg.telemetry_cfg.as_ref().unwrap().debug_enabled);
         }
     }
 
@@ -735,7 +744,7 @@ mod tests {
                     r#"{
                     "rate_by_service": {
                         "service:foo,env:staging": 1.0,
-                        "service:,env:": 0.8 
+                        "service:,env:": 0.8
                     }
                 }"#,
                 );
@@ -780,7 +789,7 @@ mod tests {
                 r#"{
                     "rate_by_service": {
                         "service:foo,env:staging": 1.0,
-                        "service:,env:": 0.8 
+                        "service:,env:": 0.8
                     }
                 }"#,
             );
@@ -796,10 +805,10 @@ mod tests {
         // (.NET) ping the agent with the aforementioned data type.
         unsafe {
             let server = MockServer::start();
-            let response_body = r#"{ 
+            let response_body = r#"{
                         "rate_by_service": {
                             "service:foo,env:staging": 1.0,
-                            "service:,env:": 0.8 
+                            "service:,env:": 0.8
                         }
                     }"#;
 
@@ -859,10 +868,10 @@ mod tests {
     fn exporter_send_telemetry_test() {
         unsafe {
             let server = MockServer::start();
-            let response_body = r#"{ 
+            let response_body = r#"{
                         "rate_by_service": {
                             "service:foo,env:staging": 1.0,
-                            "service:,env:": 0.8 
+                            "service:,env:": 0.8
                         }
                     }"#;
             let mock_traces = server.mock(|when, then| {
@@ -897,6 +906,7 @@ mod tests {
                 telemetry_cfg: Some(TelemetryConfig {
                     heartbeat: 50,
                     runtime_id: Some("foo".to_string()),
+                    debug_enabled: true,
                 }),
             };
 

data-pipeline/src/telemetry/mod.rs

@@ -65,9 +65,9 @@ impl TelemetryClientBuilder {
     }
 
     /// Sets the heartbeat notification interval in millis.
-    pub fn set_hearbeat(mut self, interval: u64) -> Self {
+    pub fn set_heartbeat(mut self, interval: u64) -> Self {
         if interval > 0 {
-            self.config.telemetry_hearbeat_interval = Duration::from_millis(interval);
+            self.config.telemetry_heartbeat_interval = Duration::from_millis(interval);
         }
         self
     }
@@ -78,6 +78,12 @@ impl TelemetryClientBuilder {
         self
     }
 
+    /// Sets the debug enabled flag for the telemetry client.
+    pub fn set_debug_enabled(mut self, debug: bool) -> Self {
+        self.config.debug_enabled = debug;
+        self
+    }
+
     /// Builds the telemetry client.
     pub async fn build(self) -> Result<TelemetryClient, TelemetryError> {
         #[allow(clippy::unwrap_used)]
@@ -276,7 +282,8 @@ mod tests {
             .set_language_version("test_language_version")
             .set_tracer_version("test_tracer_version")
             .set_url(url)
-            .set_hearbeat(100)
+            .set_heartbeat(100)
+            .set_debug_enabled(true)
             .build()
             .await
             .unwrap()
@@ -290,18 +297,20 @@ mod tests {
             .set_language_version("test_language_version")
             .set_tracer_version("test_tracer_version")
             .set_url("http://localhost")
-            .set_hearbeat(30);
+            .set_debug_enabled(true)
+            .set_heartbeat(30);
 
         assert_eq!(&builder.service_name.unwrap(), "test_service");
         assert_eq!(&builder.language.unwrap(), "test_language");
         assert_eq!(&builder.language_version.unwrap(), "test_language_version");
         assert_eq!(&builder.tracer_version.unwrap(), "test_tracer_version");
+        assert!(builder.config.debug_enabled);
         assert_eq!(
             <String as AsRef<str>>::as_ref(&builder.config.endpoint.unwrap().url.to_string()),
             "http://localhost/telemetry/proxy/api/v2/apmtelemetry"
         );
         assert_eq!(
-            builder.config.telemetry_hearbeat_interval,
+            builder.config.telemetry_heartbeat_interval,
             Duration::from_millis(30)
         );
     }
@@ -660,7 +669,7 @@ mod tests {
             .set_language_version("test_language_version")
             .set_tracer_version("test_tracer_version")
             .set_url(&server.url("/"))
-            .set_hearbeat(100)
+            .set_heartbeat(100)
             .set_runtime_id("foo")
             .build()
             .await;

data-pipeline/src/trace_exporter/mod.rs

@@ -779,6 +779,7 @@ const DEFAULT_AGENT_URL: &str = "http://127.0.0.1:8126";
 pub struct TelemetryConfig {
     pub heartbeat: u64,
     pub runtime_id: Option<String>,
+    pub debug_enabled: bool,
 }
 
 #[allow(missing_docs)]
@@ -1020,8 +1021,9 @@ impl TraceExporterBuilder {
                     .set_language_version(&self.language_version)
                     .set_service_name(&self.service)
                     .set_tracer_version(&self.tracer_version)
-                    .set_hearbeat(telemetry_config.heartbeat)
-                    .set_url(base_url);
+                    .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);
                 }
@@ -1124,6 +1126,7 @@ mod tests {
             .enable_telemetry(Some(TelemetryConfig {
                 heartbeat: 1000,
                 runtime_id: None,
+                debug_enabled: false,
             }));
         let exporter = builder.build().unwrap();
 

ddtelemetry/examples/tm-metrics-worker-test.rs

@@ -36,7 +36,7 @@ fn main() -> Result<(), Box<dyn Error>> {
     builder.config.endpoint = Some(ddcommon::Endpoint::from_slice(
         "file://./tm-metrics-worker-test.output",
     ));
-    builder.config.telemetry_hearbeat_interval = Some(Duration::from_secs(1));
+    builder.config.telemetry_heartbeat_interval = Some(Duration::from_secs(1));
 
     let handle = builder.run_metrics_logs()?;
 

ddtelemetry/examples/tm-send-sketch.rs

@@ -45,7 +45,6 @@ pub async fn push_telemetry(config: &Config, telemetry: &Telemetry<'_>) -> anyho
     let req = request_builder(config)?
         .method(http::Method::POST)
         .header(CONTENT_TYPE, ddcommon::header::APPLICATION_JSON)
-        .header("dd-telemetry-debug-enabled", "true")
         .body(serde_json::to_string(telemetry)?.into())?;
 
     let resp = client.request(req).await?;
@@ -116,5 +115,6 @@ async fn async_main() {
         api_key: Some(Cow::Owned(std::env::var("DD_API_KEY").unwrap())),
         ..Default::default()
     });
+    config.debug_enabled = true;
     push_telemetry(&config, &req).await.unwrap();
 }

ddtelemetry/examples/tm-worker-test.rs

@@ -40,7 +40,7 @@ fn main() -> Result<(), Box<dyn Error>> {
         url: ddcommon::parse_uri("file://./tm-worker-test.output").unwrap(),
         ..Default::default()
     });
-    builder.config.telemetry_hearbeat_interval = Some(Duration::from_secs(1));
+    builder.config.telemetry_heartbeat_interval = Some(Duration::from_secs(1));
 
     let handle = builder.run()?;
 

ddtelemetry/src/config.rs

@@ -31,11 +31,13 @@ pub struct Config {
     pub endpoint: Option<Endpoint>,
     /// Enables debug logging
     pub telemetry_debug_logging_enabled: bool,
-    pub telemetry_hearbeat_interval: Duration,
+    pub telemetry_heartbeat_interval: Duration,
     pub direct_submission_enabled: bool,
     /// Prevents LifecycleAction::Stop from terminating the worker (except if the WorkerHandle is
     /// dropped)
     pub restartable: bool,
+
+    pub debug_enabled: bool,
 }
 
 fn endpoint_with_telemetry_path(
@@ -161,9 +163,10 @@ impl Default for Config {
         Self {
             endpoint: None,
             telemetry_debug_logging_enabled: false,
-            telemetry_hearbeat_interval: Duration::from_secs(60),
+            telemetry_heartbeat_interval: Duration::from_secs(60),
             direct_submission_enabled: false,
             restartable: false,
+            debug_enabled: false,
         }
     }
 }
@@ -233,9 +236,10 @@ impl Config {
         let mut this = Self {
             endpoint: None,
             telemetry_debug_logging_enabled: settings.shared_lib_debug,
-            telemetry_hearbeat_interval: settings.telemetry_heartbeat_interval,
+            telemetry_heartbeat_interval: settings.telemetry_heartbeat_interval,
             direct_submission_enabled: settings.direct_submission_enabled,
             restartable: false,
+            debug_enabled: false,
         };
         if let Ok(url) = parse_uri(&trace_agent_url) {
             let _res = this.set_endpoint(Endpoint {

ddtelemetry/src/worker/builder.rs

@@ -10,7 +10,7 @@ use ddcommon::Endpoint;
 pub struct ConfigBuilder {
     pub endpoint: Option<Endpoint>,
     pub telemetry_debug_logging_enabled: Option<bool>,
-    pub telemetry_hearbeat_interval: Option<Duration>,
+    pub telemetry_heartbeat_interval: Option<Duration>,
 }
 
 impl ConfigBuilder {
@@ -20,11 +20,12 @@ impl ConfigBuilder {
             telemetry_debug_logging_enabled: self
                 .telemetry_debug_logging_enabled
                 .unwrap_or(other.telemetry_debug_logging_enabled),
-            telemetry_hearbeat_interval: self
-                .telemetry_hearbeat_interval
-                .unwrap_or(other.telemetry_hearbeat_interval),
+            telemetry_heartbeat_interval: self
+                .telemetry_heartbeat_interval
+                .unwrap_or(other.telemetry_heartbeat_interval),
             direct_submission_enabled: other.direct_submission_enabled,
             restartable: other.restartable,
+            debug_enabled: false,
         }
     }
 }
@@ -38,7 +39,7 @@ mod tests {
         let builder = ConfigBuilder {
             telemetry_debug_logging_enabled: Some(true),
             endpoint: None,
-            telemetry_hearbeat_interval: None,
+            telemetry_heartbeat_interval: None,
         };
 
         let merged = builder.merge(Config::default());

ddtelemetry/src/worker/http_client.rs

@@ -20,6 +20,9 @@ pub mod header {
     pub const API_VERSION: HeaderName = HeaderName::from_static("dd-telemetry-api-version");
     pub const LIBRARY_LANGUAGE: HeaderName = HeaderName::from_static("dd-client-library-language");
     pub const LIBRARY_VERSION: HeaderName = HeaderName::from_static("dd-client-library-version");
+
+    /// Header key for whether to enable debug mode of telemetry.
+    pub const DEBUG_ENABLED: HeaderName = HeaderName::from_static("dd-telemetry-debug-enabled");
 }
 
 pub type ResponseFuture =
@@ -31,7 +34,14 @@ pub trait HttpClient {
 
 pub fn request_builder(c: &Config) -> anyhow::Result<HttpRequestBuilder> {
     match &c.endpoint {
-        Some(e) => e.to_request_builder(concat!("telemetry/", env!("CARGO_PKG_VERSION"))),
+        Some(e) => {
+            let mut builder =
+                e.to_request_builder(concat!("telemetry/", env!("CARGO_PKG_VERSION")));
+            if c.debug_enabled {
+                builder = Ok(builder?.header(header::DEBUG_ENABLED, "true"))
+            }
+            builder
+        }
         None => Err(anyhow::Error::msg(
             "no valid endpoint found, can't build the request".to_string(),
         )),

ddtelemetry/src/worker/mod.rs

@@ -975,7 +975,7 @@ impl TelemetryWorkerBuilder {
         let contexts = MetricContexts::default();
         let token = CancellationToken::new();
         let config = self.config.merge(external_config);
-        let telemetry_hearbeat_interval = config.telemetry_hearbeat_interval;
+        let telemetry_heartbeat_interval = config.telemetry_heartbeat_interval;
         let client = http_client::from_config(&config);
 
         #[allow(clippy::unwrap_used)]
@@ -1003,7 +1003,7 @@ impl TelemetryWorkerBuilder {
                     MetricBuckets::METRICS_FLUSH_INTERVAL,
                     LifecycleAction::FlushMetricAggr,
                 ),
-                (telemetry_hearbeat_interval, LifecycleAction::FlushData),
+                (telemetry_heartbeat_interval, LifecycleAction::FlushData),
                 (
                     time::Duration::from_secs(60 * 60 * 24),
                     LifecycleAction::ExtendedHeartbeat,