fix: format

# Conflicts:
#	data-pipeline-ffi/src/trace_exporter.rs
#	data-pipeline/examples/send-traces-with-stats.rs
#	ddtelemetry/src/worker/http_client.rs

Author: Ganesh Jangir

data-pipeline-ffi/src/trace_exporter.rs

@@ -11,7 +11,7 @@ use ddcommon_ffi::{
     {slice::AsBytes, slice::ByteSlice},
 };
 use std::{ptr::NonNull, time::Duration};
-use tracing::error;
+use tracing::{debug, error};
 
 #[cfg(all(feature = "catch_panic", panic = "unwind"))]
 use std::panic::{catch_unwind, AssertUnwindSafe};
@@ -325,16 +325,16 @@ pub unsafe extern "C" fn ddog_trace_exporter_config_enable_telemetry(
     catch_panic!(
         if let Option::Some(config) = config {
             if let Option::Some(telemetry_cfg) = telemetry_cfg {
-                config.telemetry_cfg = Some(TelemetryConfig {
+                let cfg = TelemetryConfig {
                     heartbeat: telemetry_cfg.interval,
                     runtime_id: match sanitize_string(telemetry_cfg.runtime_id) {
                         Ok(s) => Some(s),
                         Err(e) => return Some(e),
                     },
                     debug_enabled: telemetry_cfg.debug_enabled,
-                })
-            } else {
-                config.telemetry_cfg = Some(TelemetryConfig::default());
+                };
+                debug!(telemetry_cfg = ?cfg, "Configuring telemetry");
+                config.telemetry_cfg = Some(cfg);
             }
             None
         } else {
@@ -467,7 +467,7 @@ pub unsafe extern "C" fn ddog_trace_exporter_new(
             }
 
             if let Some(cfg) = &config.telemetry_cfg {
-                builder.enable_telemetry(Some(cfg.clone()));
+                builder.enable_telemetry(cfg.clone());
             }
 
             if let Some(token) = &config.test_session_token {
@@ -1080,7 +1080,8 @@ mod tests {
             let mock_metrics = server.mock(|when, then| {
                 when.method(POST)
                     .path("/telemetry/proxy/api/v2/apmtelemetry")
-                    .body_contains(r#""runtime_id":"foo""#);
+                    .body_includes(r#""runtime_id":"foo""#)
+                    .body_includes(r#""metric":"trace_api."#);
                 then.status(200)
                     .header("content-type", "application/json")
                     .body("");
@@ -1132,8 +1133,8 @@ mod tests {
             );
 
             ddog_trace_exporter_free(exporter);
-            // It should receive 1 payloads: metrics
-            mock_metrics.assert_hits(1);
+            // It should receive 1 metrics payload (excluding heartbeats)
+            mock_metrics.assert_calls(1);
         }
     }
 

data-pipeline/examples/send-traces-with-stats.rs

@@ -1,8 +1,12 @@
 // Copyright 2024-Present Datadog, Inc. https://www.datadoghq.com/
 // SPDX-License-Identifier: Apache-2.0
 
+use clap::Parser;
 use data_pipeline::trace_exporter::{
-    TraceExporter, TraceExporterInputFormat, TraceExporterOutputFormat,
+    TelemetryConfig, TraceExporter, TraceExporterInputFormat, TraceExporterOutputFormat,
+};
+use datadog_log::logger::{
+    logger_configure_std, logger_set_log_level, LogEventLevel, StdConfig, StdTarget,
 };
 use datadog_trace_protobuf::pb;
 use std::{
@@ -29,10 +33,31 @@ fn get_span(now: i64, trace_id: u64, span_id: u64) -> pb::Span {
     }
 }
 
+#[derive(Parser)]
+#[command(name = "send-traces-with-stats")]
+#[command(about = "A data pipeline example for sending traces with statistics")]
+struct Args {
+    #[arg(
+        short = 'u',
+        long = "url",
+        default_value = "http://localhost:8126",
+        help = "Set the trace agent URL\n\nExamples:\n  http://localhost:8126 (default)\n  windows://./pipe/dd-apm-test-agent (Windows named pipe)\n  https://trace.agent.datadoghq.com:443 (custom endpoint)"
+    )]
+    url: String,
+}
+
 fn main() {
+    logger_configure_std(StdConfig {
+        target: StdTarget::Out,
+    })
+    .expect("Failed to configure logger");
+    logger_set_log_level(LogEventLevel::Debug).expect("Failed to set log level");
+
+    let args = Args::parse();
+    let telemetry_cfg = TelemetryConfig::default();
     let mut builder = TraceExporter::builder();
     builder
-        .set_url("http://localhost:8126")
+        .set_url(&args.url)
         .set_hostname("test")
         .set_env("testing")
         .set_app_version(env!("CARGO_PKG_VERSION"))
@@ -42,20 +67,28 @@ fn main() {
         .set_language_version(env!("CARGO_PKG_RUST_VERSION"))
         .set_input_format(TraceExporterInputFormat::V04)
         .set_output_format(TraceExporterOutputFormat::V04)
+        .enable_telemetry(telemetry_cfg)
         .enable_stats(Duration::from_secs(10));
-    let exporter = builder.build().unwrap();
-    let now = UNIX_EPOCH.elapsed().unwrap().as_nanos() as i64;
+    let exporter = builder.build().expect("Failed to build TraceExporter");
+    let now = UNIX_EPOCH
+        .elapsed()
+        .expect("Failed to get time since UNIX_EPOCH")
+        .as_nanos() as i64;
 
     let mut traces = Vec::new();
-    for trace_id in 1..=100 {
+    for trace_id in 1..=2 {
         let mut trace = Vec::new();
-        for span_id in 1..=1000 {
+        for span_id in 1..=2 {
             trace.push(get_span(now, trace_id, span_id));
         }
         traces.push(trace);
     }
-    let data = rmp_serde::to_vec_named(&traces).unwrap();
+    let data = rmp_serde::to_vec_named(&traces).expect("Failed to serialize traces");
 
-    exporter.send(data.as_ref(), 100).unwrap();
-    exporter.shutdown(None).unwrap();
+    exporter
+        .send(data.as_ref(), 2)
+        .expect("Failed to send traces");
+    exporter
+        .shutdown(None)
+        .expect("Failed to shutdown exporter");
 }

ddtelemetry/src/worker/http_client.rs

@@ -12,6 +12,7 @@ use std::{
 };
 
 use crate::config::Config;
+use tracing::{debug, error, info, warn};
 
 pub mod header {
     #![allow(clippy::declare_interior_mutable_const)]
@@ -35,16 +36,29 @@ pub trait HttpClient {
 pub fn request_builder(c: &Config) -> anyhow::Result<HttpRequestBuilder> {
     match &c.endpoint {
         Some(e) => {
+            debug!(
+                endpoint.url = %e.url,
+                endpoint.timeout_ms = e.timeout_ms,
+                telemetry.version = env!("CARGO_PKG_VERSION"),
+                "Building telemetry request"
+            );
             let mut builder =
                 e.to_request_builder(concat!("telemetry/", env!("CARGO_PKG_VERSION")));
             if c.debug_enabled {
+                debug!(
+                    telemetry.debug_enabled = true,
+                    "Telemetry debug mode 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(),
-        )),
+        None => {
+            error!("No valid telemetry endpoint found, cannot build request");
+            Err(anyhow::Error::msg(
+                "no valid endpoint found, can't build the request".to_string(),
+            ))
+        }
     }
 }
 
@@ -54,6 +68,10 @@ pub fn from_config(c: &Config) -> Box<dyn HttpClient + Sync + Send> {
             #[allow(clippy::expect_used)]
             let file_path = ddcommon::decode_uri_path_in_authority(&e.url)
                 .expect("file urls should always have been encoded in authority");
+            info!(
+                file.path = ?file_path,
+                "Using file-based mock telemetry client"
+            );
             return Box::new(MockClient {
                 #[allow(clippy::expect_used)]
                 file: Arc::new(Mutex::new(Box::new(
@@ -65,7 +83,19 @@ pub fn from_config(c: &Config) -> Box<dyn HttpClient + Sync + Send> {
                 ))),
             });
         }
-        Some(_) | None => {}
+        Some(e) => {
+            info!(
+                endpoint.url = %e.url,
+                endpoint.timeout_ms = e.timeout_ms,
+                "Using HTTP telemetry client"
+            );
+        }
+        None => {
+            warn!(
+                endpoint = "default",
+                "No telemetry endpoint configured, using default HTTP client"
+            );
+        }
     };
     Box::new(HyperClient {
         inner: hyper_migration::new_client_periodic(),
@@ -78,8 +108,26 @@ pub struct HyperClient {
 
 impl HttpClient for HyperClient {
     fn request(&self, req: hyper_migration::HttpRequest) -> ResponseFuture {
+        debug!("Sending HTTP request via HyperClient");
         let resp = self.inner.request(req);
-        Box::pin(async { Ok(hyper_migration::into_response(resp.await?)) })
+        Box::pin(async move {
+            match resp.await {
+                Ok(response) => {
+                    debug!(
+                        http.status = response.status().as_u16(),
+                        "HTTP request completed successfully"
+                    );
+                    Ok(hyper_migration::into_response(response))
+                }
+                Err(e) => {
+                    error!(
+                        error = %e,
+                        "HTTP request failed"
+                    );
+                    Err(e.into())
+                }
+            }
+        })
     }
 }
 
@@ -92,16 +140,30 @@ impl HttpClient for MockClient {
     fn request(&self, req: hyper_migration::HttpRequest) -> ResponseFuture {
         let s = self.clone();
         Box::pin(async move {
+            debug!("MockClient writing request to file");
             let mut body = req.collect().await?.to_bytes().to_vec();
             body.push(b'\n');
 
             {
                 #[allow(clippy::expect_used)]
                 let mut writer = s.file.lock().expect("mutex poisoned");
 
-                writer.write_all(body.as_ref())?;
+                match writer.write_all(body.as_ref()) {
+                    Ok(()) => debug!(
+                        file.bytes_written = body.len(),
+                        "Successfully wrote payload to mock file"
+                    ),
+                    Err(e) => {
+                        error!(
+                            error = %e,
+                            "Failed to write to mock file"
+                        );
+                        return Err(e.into());
+                    }
+                }
             }
 
+            debug!(http.status = 202, "MockClient returning success response");
             hyper_migration::empty_response(hyper::Response::builder().status(202))
         })
     }