chore: improve logging in telemetry crate

# Conflicts:
#	data-pipeline/Cargo.toml
#	data-pipeline/examples/send-traces-with-stats.rs
#	examples/ffi/README.md

Author: ganeshnj

Cargo.lock

@@ -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 {

data-pipeline-ffi/src/trace_exporter.rs

@@ -51,6 +51,8 @@ harness = false
 path = "benches/main.rs"
 
 [dev-dependencies]
+datadog-log = { path = "../datadog-log" }
+clap = { version = "4.0", features = ["derive"] }
 criterion = "0.5.1"
 datadog-trace-utils = { path = "../datadog-trace-utils", features = ["test-utils"] }
 httpmock = "0.7.0"

data-pipeline/Cargo.toml

@@ -1,6 +1,8 @@
 // Copyright 2024-Present Datadog, Inc. https://www.datadoghq.com/
 // SPDX-License-Identifier: Apache-2.0
 
+use clap::Parser;
+use data_pipeline::trace_exporter::{TelemetryConfig, TraceExporter, TraceExporterInputFormat, TraceExporterOutputFormat};
 use data_pipeline::trace_exporter::{
     TraceExporter, TraceExporterInputFormat, TraceExporterOutputFormat,
 };
@@ -9,6 +11,7 @@ use std::{
     collections::HashMap,
     time::{Duration, UNIX_EPOCH},
 };
+use datadog_log::logger::{logger_configure_std, logger_set_log_level, LogEventLevel, StdConfig, StdTarget};
 
 fn get_span(now: i64, trace_id: u64, span_id: u64) -> pb::Span {
     pb::Span {
@@ -30,6 +33,14 @@ fn get_span(now: i64, trace_id: u64, span_id: u64) -> pb::Span {
 }
 
 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")
@@ -42,20 +53,27 @@ 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");
 }

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

@@ -193,12 +193,8 @@ impl TraceExporterBuilder {
     }
 
     /// Enables sending telemetry metrics.
-    pub fn enable_telemetry(&mut self, cfg: Option<TelemetryConfig>) -> &mut Self {
-        if let Some(cfg) = cfg {
-            self.telemetry = Some(cfg);
-        } else {
-            self.telemetry = Some(TelemetryConfig::default());
-        }
+    pub fn enable_telemetry(&mut self, cfg: TelemetryConfig) -> &mut Self {
+        self.telemetry = Some(cfg);
         self
     }
 

data-pipeline/src/trace_exporter/builder.rs

@@ -10,7 +10,7 @@ license.workspace = true
 bench = false
 
 [features]
-default = []
+default = ["tracing"]
 tracing = ["tracing/std"]
 
 [dependencies]

ddtelemetry/Cargo.toml

@@ -4,6 +4,7 @@
 use ddcommon::{config::parse_env, parse_uri, Endpoint};
 use http::{uri::PathAndQuery, Uri};
 use std::{borrow::Cow, time::Duration};
+use tracing::debug;
 
 pub const DEFAULT_DD_SITE: &str = "datadoghq.com";
 pub const PROD_INTAKE_SUBDOMAIN: &str = "instrumentation-telemetry-intake";
@@ -119,6 +120,10 @@ impl Settings {
     const _DD_SHARED_LIB_DEBUG: &'static str = "_DD_SHARED_LIB_DEBUG";
 
     pub fn from_env() -> Self {
+        debug!(
+            config.source = "environment",
+            "Loading telemetry settings from environment variables"
+        );
         let default = Self::default();
         Self {
             agent_host: parse_env::str_not_empty(Self::DD_AGENT_HOST),

ddtelemetry/src/config.rs

@@ -10,6 +10,7 @@
 #![cfg_attr(not(test), deny(clippy::unimplemented))]
 
 use ddcommon::entity_id;
+use tracing::debug;
 
 pub mod config;
 pub mod data;
@@ -18,11 +19,24 @@ pub mod metrics;
 pub mod worker;
 
 pub fn build_host() -> data::Host {
+    debug!("Building telemetry host information");
+    let hostname = info::os::real_hostname().unwrap_or_else(|_| String::from("unknown_hostname"));
+    let container_id = entity_id::get_container_id().map(|f| f.to_string());
+    let os_version = info::os::os_version().ok();
+
+    debug!(
+        host.hostname = %hostname,
+        host.container_id = ?container_id,
+        host.os = info::os::os_name(),
+        host.os_version = ?os_version,
+        "Built telemetry host information"
+    );
+
     data::Host {
-        hostname: info::os::real_hostname().unwrap_or_else(|_| String::from("unknown_hostname")),
-        container_id: entity_id::get_container_id().map(|f| f.to_string()),
+        hostname,
+        container_id,
         os: Some(String::from(info::os::os_name())),
-        os_version: info::os::os_version().ok(),
+        os_version,
         kernel_name: None,
         kernel_release: None,
         kernel_version: None,

ddtelemetry/src/lib.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,28 @@ 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 +142,35 @@ 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))
         })
     }