feat(data-pipeline): emit native trace export telemetry

libdd-data-pipeline-core/src/agentless/exporter.rs

@@ -8,7 +8,8 @@ use http::HeaderMap;
 use libdd_capabilities::{HttpClientCapability, SleepCapability};
 use libdd_common::Endpoint;
 use libdd_trace_utils::send_with_retry::{
-    send_with_retry, CompressionStrategy, RetryBackoffType, RetryStrategy, SendWithRetryError,
+    send_with_retry_and_size, CompressionStrategy, RetryBackoffType, RetryStrategy,
+    SendWithRetryError, SendWithRetryResult,
 };
 use libdd_trace_utils::span::{trace_utils::compute_top_level_span, TraceData};
 use libdd_trace_utils::tracer_metadata::TracerMetadata;
@@ -41,15 +42,42 @@ pub enum AgentlessError {
 /// the caller is already computing and exporting stats locally so that the intake
 /// does not double-count the same traces.
 pub async fn send_agentless_traces<C, T>(
+    capabilities: &C,
+    traces: Vec<Vec<libdd_trace_utils::span::v04::Span<T>>>,
+    metadata: &TracerMetadata,
+    config: &AgentlessTraceConfig,
+    client_side_stats: bool,
+) -> Result<(), AgentlessError>
+where
+    C: HttpClientCapability + SleepCapability,
+    T: TraceData,
+{
+    send_agentless_traces_with_observer(
+        capabilities,
+        traces,
+        metadata,
+        config,
+        client_side_stats,
+        |_, _| {},
+    )
+    .await
+}
+
+/// Encodes and sends already-decoded v0.4 traces, reporting the final send result.
+///
+/// The observer is invoked once after the retry loop with the post-compression payload size.
+pub async fn send_agentless_traces_with_observer<C, T, F>(
     capabilities: &C,
     mut traces: Vec<Vec<libdd_trace_utils::span::v04::Span<T>>>,
     metadata: &TracerMetadata,
     config: &AgentlessTraceConfig,
     client_side_stats: bool,
+    observer: F,
 ) -> Result<(), AgentlessError>
 where
     C: HttpClientCapability + SleepCapability,
     T: TraceData,
+    F: FnOnce(&SendWithRetryResult, usize),
 {
     // Top-level tagging is already done before sending when client-side stats
     // are active
@@ -76,20 +104,22 @@ where
         libdd_trace_utils::agentless_encoder::encode_payload(&traces, metadata, client_side_stats)
             .map_err(AgentlessError::Serialization)?;
     let headers = build_agentless_headers(metadata, trace_count);
-    send_agentless_json(capabilities, config, headers, json_body).await
+    send_agentless_json(capabilities, config, headers, json_body, observer).await
 }
 
 /// Sends an encoded agentless JSON request.
 ///
 /// The configured API key replaces any `dd-api-key` value in `headers`.
-async fn send_agentless_json<C>(
+async fn send_agentless_json<C, F>(
     capabilities: &C,
     config: &AgentlessTraceConfig,
     mut headers: HeaderMap,
     json_body: Vec<u8>,
+    observer: F,
 ) -> Result<(), AgentlessError>
 where
     C: HttpClientCapability + SleepCapability,
+    F: FnOnce(&SendWithRetryResult, usize),
 {
     let api_key =
         http::HeaderValue::from_str(&config.api_key).map_err(|_| AgentlessError::InvalidApiKey)?;
@@ -114,17 +144,19 @@ where
     #[cfg(not(feature = "compression"))]
     let compression_strategy = CompressionStrategy::None;
 
-    send_with_retry(
+    let (result, payload_size) = send_with_retry_and_size(
         capabilities,
         &target,
         json_body,
         &headers,
         &retry_strategy,
         compression_strategy,
     )
-    .await
-    .map(|_| ())
-    .map_err(|error| AgentlessError::Send(Box::new(error)))
+    .await;
+    observer(&result, payload_size);
+    result
+        .map(|_| ())
+        .map_err(|error| AgentlessError::Send(Box::new(error)))
 }
 
 fn build_agentless_headers(metadata: &TracerMetadata, trace_count: usize) -> HeaderMap {
@@ -258,6 +290,25 @@ mod tests {
             .contains("\"_top_level\":1"));
     }
 
+    #[test]
+    fn reports_final_send_result_and_payload_size() {
+        let capabilities = TestCapabilities::default();
+        let mut observation = None;
+        let result = futures::executor::block_on(send_agentless_traces_with_observer(
+            &capabilities,
+            v04_traces(),
+            &metadata(),
+            &config(),
+            false,
+            |result, payload_size| observation = Some((result.is_ok(), payload_size)),
+        ));
+
+        assert!(result.is_ok());
+        let (send_succeeded, payload_size) = observation.unwrap();
+        assert!(send_succeeded);
+        assert!(payload_size > 0);
+    }
+
     #[test]
     fn json_request_uses_configured_api_key() {
         let capabilities = TestCapabilities::default();
@@ -272,6 +323,7 @@ mod tests {
             &config(),
             headers,
             b"[]".to_vec(),
+            |_, _| {},
         ));
         assert!(result.is_ok());
         let requests = capabilities.requests.lock().unwrap();
@@ -289,6 +341,7 @@ mod tests {
             &invalid_config,
             HeaderMap::new(),
             b"[]".to_vec(),
+            |_, _| {},
         ))
         .unwrap_err();
         assert!(matches!(error, AgentlessError::InvalidApiKey));

libdd-data-pipeline-core/src/agentless/mod.rs

@@ -7,4 +7,4 @@ mod config;
 mod exporter;
 
 pub use config::{AgentlessTraceConfig, DEFAULT_AGENTLESS_TIMEOUT};
-pub use exporter::{send_agentless_traces, AgentlessError};
+pub use exporter::{send_agentless_traces, send_agentless_traces_with_observer, AgentlessError};

libdd-data-pipeline-core/src/lib.rs

@@ -10,6 +10,7 @@
 mod agentless;
 
 pub use agentless::{
-    send_agentless_traces, AgentlessError, AgentlessTraceConfig, DEFAULT_AGENTLESS_TIMEOUT,
+    send_agentless_traces, send_agentless_traces_with_observer, AgentlessError,
+    AgentlessTraceConfig, DEFAULT_AGENTLESS_TIMEOUT,
 };
 pub use libdd_trace_utils::tracer_metadata::TracerMetadata;

libdd-data-pipeline/src/agentless/exporter.rs

@@ -6,27 +6,41 @@
 use crate::trace_exporter::error::{InternalErrorKind, RequestError, TraceExporterError};
 use libdd_capabilities::{HttpClientCapability, SleepCapability};
 use libdd_data_pipeline_core::{
-    send_agentless_traces as send_traces, AgentlessError, AgentlessTraceConfig,
+    send_agentless_traces_with_observer as send_traces, AgentlessError, AgentlessTraceConfig,
 };
-use libdd_trace_utils::send_with_retry::SendWithRetryError;
+use libdd_trace_utils::send_with_retry::{SendWithRetryError, SendWithRetryResult};
 use libdd_trace_utils::span::TraceData;
 use libdd_trace_utils::tracer_metadata::TracerMetadata;
 use tracing::error;
 
-pub(crate) async fn send_agentless_traces<T, C>(
+pub(crate) async fn send_agentless_traces_with_observer<T, C, F, S>(
     capabilities: &C,
     traces: Vec<Vec<libdd_trace_utils::span::v04::Span<T>>>,
     metadata: &TracerMetadata,
     config: &AgentlessTraceConfig,
     client_side_stats: bool,
+    observer: F,
+    serialization_error_observer: S,
 ) -> Result<(), TraceExporterError>
 where
     T: TraceData,
     C: HttpClientCapability + SleepCapability,
+    F: FnOnce(&SendWithRetryResult, usize),
+    S: FnOnce(),
 {
-    send_traces(capabilities, traces, metadata, config, client_side_stats)
-        .await
-        .map_err(map_agentless_error)
+    let result = send_traces(
+        capabilities,
+        traces,
+        metadata,
+        config,
+        client_side_stats,
+        observer,
+    )
+    .await;
+    if matches!(&result, Err(AgentlessError::Serialization(_))) {
+        serialization_error_observer();
+    }
+    result.map_err(map_agentless_error)
 }
 
 fn map_agentless_error(error: AgentlessError) -> TraceExporterError {

libdd-data-pipeline/src/otlp/exporter.rs

@@ -10,6 +10,7 @@ use libdd_capabilities::{HttpClientCapability, SleepCapability};
 use libdd_common::Endpoint;
 use libdd_trace_utils::send_with_retry::{
     send_with_retry, CompressionStrategy, RetryBackoffType, RetryStrategy, SendWithRetryError,
+    SendWithRetryResult,
 };
 use std::time::Duration;
 
@@ -35,6 +36,35 @@ pub(crate) async fn send_otlp_http<C: HttpClientCapability + SleepCapability>(
     content_type: http::HeaderValue,
     body: Vec<u8>,
     max_retries: u32,
+) -> Result<(), TraceExporterError> {
+    send_otlp_http_with_observer(
+        capabilities,
+        endpoint_url,
+        config_headers,
+        timeout,
+        test_token,
+        content_type,
+        body,
+        max_retries,
+        |_| {},
+    )
+    .await
+}
+
+#[allow(clippy::too_many_arguments)]
+pub(crate) async fn send_otlp_http_with_observer<
+    C: HttpClientCapability + SleepCapability,
+    F: FnOnce(&SendWithRetryResult),
+>(
+    capabilities: &C,
+    endpoint_url: &str,
+    config_headers: &HeaderMap,
+    timeout: Duration,
+    test_token: Option<&str>,
+    content_type: http::HeaderValue,
+    body: Vec<u8>,
+    max_retries: u32,
+    observer: F,
 ) -> Result<(), TraceExporterError> {
     let url = libdd_common::parse_uri(endpoint_url).map_err(|e| {
         TraceExporterError::Internal(InternalErrorKind::InvalidWorkerState(format!(
@@ -67,16 +97,17 @@ pub(crate) async fn send_otlp_http<C: HttpClientCapability + SleepCapability>(
         None,
     );
 
-    match send_with_retry(
+    let result = send_with_retry(
         capabilities,
         &target,
         body,
         &headers,
         &retry_strategy,
         CompressionStrategy::None,
     )
-    .await
-    {
+    .await;
+    observer(&result);
+    match result {
         Ok(_) => Ok(()),
         Err(e) => Err(map_send_error(e).await),
     }
@@ -87,6 +118,7 @@ pub(crate) async fn send_otlp_http<C: HttpClientCapability + SleepCapability>(
 ///
 /// `test_token` is forwarded as `X-Datadog-Test-Session-Token` when set, enabling snapshot tests
 /// against the Datadog test agent's OTLP endpoint.
+#[allow(dead_code)]
 pub async fn send_otlp_traces_http<C: HttpClientCapability + SleepCapability>(
     capabilities: &C,
     config: &OtlpTraceConfig,

libdd-data-pipeline/src/otlp/mod.rs

@@ -32,6 +32,7 @@ pub mod exporter;
 pub mod metrics;
 
 pub use config::{OtlpMetricsConfig, OtlpProtocol, OtlpTraceConfig};
+#[allow(unused_imports)]
 pub use exporter::send_otlp_traces_http;
 pub use libdd_trace_utils::otlp_encoder::{map_traces_to_otlp, OtlpResourceInfo};
 pub use metrics::OtlpStatsExporter;

libdd-data-pipeline/src/telemetry/metrics.rs

@@ -34,6 +34,12 @@ pub enum MetricKind {
     ChunksDroppedSerializationError,
     /// trace_chunks_dropped metric (reason: send_failure)
     ChunksDroppedSendFailure,
+    /// spans_enqueued_for_serialization metric
+    SpansEnqueuedForSerialization,
+    /// spans_dropped metric (reason: serialization_error)
+    SpansDroppedSerializationError,
+    /// spans_dropped metric (reason: api_error)
+    SpansDroppedApiError,
 }
 
 /// Constants for metric names
@@ -44,6 +50,8 @@ const API_BYTES_STR: &str = "trace_api.bytes";
 const API_RESPONSES_STR: &str = "trace_api.responses";
 const CHUNKS_SENT_STR: &str = "trace_chunks_sent";
 const CHUNKS_DROPPED_STR: &str = "trace_chunks_dropped";
+const SPANS_ENQUEUED_FOR_SERIALIZATION_STR: &str = "spans_enqueued_for_serialization";
+const SPANS_DROPPED_STR: &str = "spans_dropped";
 
 #[derive(Debug)]
 struct Metric {
@@ -132,6 +140,24 @@ const METRICS: &[Metric] = &[
             tag!["reason", "send_failure"],
         ],
     },
+    Metric {
+        name: SPANS_ENQUEUED_FOR_SERIALIZATION_STR,
+        metric_type: MetricType::Count,
+        namespace: MetricNamespace::Tracers,
+        tags: &[],
+    },
+    Metric {
+        name: SPANS_DROPPED_STR,
+        metric_type: MetricType::Count,
+        namespace: MetricNamespace::Tracers,
+        tags: &[tag!["reason", "serialization_error"]],
+    },
+    Metric {
+        name: SPANS_DROPPED_STR,
+        metric_type: MetricType::Count,
+        namespace: MetricNamespace::Tracers,
+        tags: &[tag!["reason", "api_error"]],
+    },
 ];
 
 /// Structure to accumulate partial results coming from sending traces to the agent.