chore: fix merge issues

Author: hoolioh

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

@@ -2,7 +2,6 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use clap::Parser;
-use libdd_data_pipeline::trace_exporter::TelemetryConfig;
 use libdd_capabilities_impl::NativeCapabilities;
 use libdd_data_pipeline::trace_exporter::{
     TraceExporter, TraceExporterInputFormat, TraceExporterOutputFormat,
@@ -70,7 +69,7 @@ fn main() {
         .set_output_format(TraceExporterOutputFormat::V04)
         .enable_stats(Duration::from_secs(10));
     #[cfg(feature = "telemetry")]
-    builder.enable_telemetry(TelemetryConfig::default());
+    builder.enable_telemetry(libdd_data_pipeline::telemetry::TelemetryConfig::default());
     let exporter = builder
         .build::<NativeCapabilities>()
         .expect("Failed to build TraceExporter");

libdd-data-pipeline/src/lib.rs

@@ -17,7 +17,6 @@ pub(crate) mod otlp;
 mod pausable_worker;
 #[allow(missing_docs)]
 pub mod stats_exporter;
-#[cfg(feature = "telemetry")]
-pub(crate) mod telemetry;
+pub mod telemetry;
 #[allow(missing_docs)]
 pub mod trace_exporter;

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

@@ -1,6 +1,7 @@
 // Copyright 2024-Present Datadog, Inc. https://www.datadoghq.com/
 // SPDX-License-Identifier: Apache-2.0
 
+use super::error::TelemetryError;
 #[cfg(feature = "telemetry")]
 use super::metrics::Metrics;
 use super::worker;
@@ -9,7 +10,6 @@ use super::TelemetryClient;
 use libdd_telemetry::worker::{TelemetryWorkerBuilder, TelemetryWorkerFlavor};
 use std::time::Duration;
 use tokio::runtime::Handle;
-use super::error::TelemetryError;
 
 /// Structure to build a Telemetry client.
 ///
@@ -96,12 +96,20 @@ impl TelemetryClientBuilder {
 #[cfg(feature = "telemetry")]
 impl TelemetryClientBuilder {
     /// Builds the telemetry client.
-    pub fn build(self, runtime: Handle) -> Result<(TelemetryClient, worker::TelemetryWorker), TelemetryError> {
+    pub fn build(
+        self,
+        runtime: Handle,
+    ) -> Result<(TelemetryClient, worker::TelemetryWorker), TelemetryError> {
         let mut builder = TelemetryWorkerBuilder::new_fetch_host(
-            self.service_name.ok_or_else(|| TelemetryError::Builder("service_name is required".to_string()))?,
-            self.language.ok_or_else(|| TelemetryError::Builder("language is required".to_string()))?,
-            self.language_version.ok_or_else(|| TelemetryError::Builder("language_version is required".to_string()))?,
-            self.tracer_version.ok_or_else(|| TelemetryError::Builder("tracer_version is required".to_string()))?,
+            self.service_name
+                .ok_or_else(|| TelemetryError::Builder("service_name is required".to_string()))?,
+            self.language
+                .ok_or_else(|| TelemetryError::Builder("language is required".to_string()))?,
+            self.language_version.ok_or_else(|| {
+                TelemetryError::Builder("language_version is required".to_string())
+            })?,
+            self.tracer_version
+                .ok_or_else(|| TelemetryError::Builder("tracer_version is required".to_string()))?,
         );
         if let Some(url) = self.url {
             builder
@@ -137,7 +145,10 @@ impl TelemetryClientBuilder {
 #[cfg(not(feature = "telemetry"))]
 impl TelemetryClientBuilder {
     /// Builds a no-op telemetry client.
-    pub fn build(self, _runtime: Handle) -> Result<(TelemetryClient, worker::TelemetryWorker), TelemetryError> {
+    pub fn build(
+        self,
+        _runtime: Handle,
+    ) -> Result<(TelemetryClient, worker::TelemetryWorker), TelemetryError> {
         Ok((TelemetryClient {}, worker::TelemetryWorker {}))
     }
 }

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

@@ -16,12 +16,11 @@ pub struct TelemetryConfig {
     pub debug_enabled: bool,
 }
 
-
 #[cfg(not(feature = "telemetry"))]
 mod inner {
     use super::*;
-    use libdd_trace_utils::trace_utils::SendDataResult;
     use libdd_trace_utils::send_with_retry::SendWithRetryResult;
+    use libdd_trace_utils::trace_utils::SendDataResult;
 
     #[derive(Debug)]
     pub struct TelemetryClient {}
@@ -70,7 +69,6 @@ mod inner {
     use libdd_trace_utils::send_with_retry::SendWithRetryResult;
     use libdd_trace_utils::trace_utils::SendDataResult;
 
-
     /// Telemetry handle used to send metrics to the agent
     #[derive(Debug)]
     pub struct TelemetryClient {
@@ -123,8 +121,11 @@ mod inner {
                 let key = self
                     .metrics
                     .get(metrics::MetricKind::ChunksDroppedSerializationError);
-                self.worker
-                    .add_point(data.chunks_dropped_serialization_error as f64, key, vec![])?;
+                self.worker.add_point(
+                    data.chunks_dropped_serialization_error as f64,
+                    key,
+                    vec![],
+                )?;
             }
             if data.chunks_dropped_send_failure > 0 {
                 let key = self
@@ -160,7 +161,6 @@ mod inner {
         }
     }
 
-
     /// Telemetry describing the sending of a trace payload
     /// It can be produced from a [`SendWithRetryResult`] or from a [`SendDataResult`].
     #[derive(PartialEq, Debug, Default)]
@@ -223,6 +223,11 @@ mod inner {
                         telemetry.errors_network = 1;
                         telemetry.requests_count = *attempts as u64;
                     }
+                    SendWithRetryError::ResponseBody(attempts) => {
+                        telemetry.chunks_dropped_send_failure = chunks;
+                        telemetry.errors_network = 1;
+                        telemetry.requests_count = *attempts as u64;
+                    }
                     SendWithRetryError::Build(attempts) => {
                         telemetry.chunks_dropped_serialization_error = chunks;
                         telemetry.requests_count = *attempts as u64;
@@ -248,12 +253,11 @@ mod inner {
             }
         }
     }
-
 }
 
-pub use inner::*;
 pub(crate) use builder::TelemetryClientBuilder;
 pub(crate) use error::TelemetryError;
+pub use inner::*;
 pub(crate) use worker::TelemetryWorker;
 
 #[cfg(test)]
@@ -262,9 +266,11 @@ mod tests {
     use bytes::Bytes;
     use httpmock::Method::POST;
     use httpmock::MockServer;
-    use libdd_trace_utils::{send_data::send_data_result::SendDataResult, send_with_retry::SendWithRetryError};
     use libdd_capabilities::HttpError;
     use libdd_common::worker::Worker;
+    use libdd_trace_utils::{
+        send_data::send_data_result::SendDataResult, send_with_retry::SendWithRetryError,
+    };
     use regex::Regex;
     use std::collections::HashMap;
     use std::time::Duration;

libdd-data-pipeline/src/trace_exporter/builder.rs

@@ -9,7 +9,6 @@ use crate::pausable_worker::PausableWorker;
 use crate::telemetry::{TelemetryClientBuilder, TelemetryConfig};
 use crate::trace_exporter::agent_response::AgentResponsePayloadVersion;
 use crate::trace_exporter::error::BuilderErrorKind;
-use crate::trace_exporter::TelemetryConfig;
 #[cfg(not(target_arch = "wasm32"))]
 use crate::trace_exporter::TraceExporterWorkers;
 use crate::trace_exporter::{
@@ -296,7 +295,8 @@ impl TraceExporterBuilder {
                 stats = StatsComputationStatus::DisabledByAgent { bucket_size };
             }
 
-            let (telemetry_client, telemetry_worker) = self.telemetry
+            let (telemetry_client, telemetry_worker) = self
+                .telemetry
                 .map(|telemetry_config| -> Result<_, TraceExporterError> {
                     let mut builder = TelemetryClientBuilder::default()
                         .set_language(&self.language)
@@ -312,17 +312,19 @@ impl TraceExporterBuilder {
                         builder = builder.set_runtime_id(&id);
                     }
 
-                    let (client, worker) = builder.build(runtime.handle().clone()).map_err(TraceExporterError::from)?;
+                    let (client, worker) = builder
+                        .build(runtime.handle().clone())
+                        .map_err(TraceExporterError::from)?;
                     let mut telemetry_worker = PausableWorker::new(worker);
                     telemetry_worker.start(&runtime).map_err(|e| {
                         TraceExporterError::Builder(BuilderErrorKind::InvalidConfiguration(
-                                e.to_string(),
+                            e.to_string(),
                         ))
                     })?;
                     runtime.block_on(client.start());
                     Ok((client, telemetry_worker))
                 })
-            .transpose()?
+                .transpose()?
                 .map_or((None, None), |(c, w)| (Some(c), Some(w)));
 
             Ok(TraceExporter {

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

@@ -14,14 +14,10 @@ use self::agent_response::AgentResponse;
 use self::metrics::MetricsEmitter;
 use self::stats::StatsComputationStatus;
 use self::trace_serializer::TraceSerializer;
-#[cfg(not(target_arch = "wasm32"))]
-use crate::agent_info::AgentInfoFetcher;
 use crate::agent_info::ResponseObserver;
 use crate::otlp::{map_traces_to_otlp, send_otlp_traces_http, OtlpResourceInfo, OtlpTraceConfig};
 #[cfg(not(target_arch = "wasm32"))]
 use crate::pausable_worker::PausableWorker;
-#[cfg(not(target_arch = "wasm32"))]
-use crate::stats_exporter::StatsExporter;
 use crate::telemetry::{SendPayloadTelemetry, TelemetryClient};
 use crate::trace_exporter::agent_response::{
     AgentResponsePayloadVersion, DATADOG_RATES_PAYLOAD_VERSION,
@@ -42,8 +38,6 @@ use libdd_capabilities::{HttpClientTrait, MaybeSend};
 use libdd_common::tag::Tag;
 use libdd_common::{Endpoint, MutexExt};
 use libdd_dogstatsd_client::Client;
-#[cfg(not(target_arch = "wasm32"))]
-use libdd_telemetry::worker::TelemetryWorker;
 use libdd_trace_utils::msgpack_decoder;
 use libdd_trace_utils::send_with_retry::{
     send_with_retry, RetryStrategy, SendWithRetryError, SendWithRetryResult,
@@ -182,9 +176,9 @@ impl<'a> From<&'a TracerMetadata> for HeaderMap {
 /// `H` is the HTTP client implementation, see [`HttpClientTrait`].
 #[derive(Debug)]
 pub(crate) struct TraceExporterWorkers<H: HttpClientTrait + MaybeSend + Sync + 'static> {
-    pub info: PausableWorker<AgentInfoFetcher<H>>,
-    pub stats: Option<PausableWorker<StatsExporter<H>>>,
-    pub telemetry: Option<PausableWorker<TelemetryWorker>>,
+    pub info: PausableWorker<crate::agent_info::AgentInfoFetcher<H>>,
+    pub stats: Option<PausableWorker<crate::stats_exporter::StatsExporter<H>>>,
+    pub telemetry: Option<PausableWorker<crate::telemetry::TelemetryWorker>>,
 }
 
 /// The TraceExporter ingest traces from the tracers serialized as messagepack and forward them to
@@ -962,6 +956,7 @@ pub trait ResponseCallback {
 mod tests {
     use self::error::AgentErrorKind;
     use super::*;
+    use crate::telemetry::TelemetryConfig;
     use httpmock::prelude::*;
     use httpmock::MockServer;
     use libdd_capabilities_impl::NativeCapabilities;
@@ -972,7 +967,6 @@ mod tests {
     use std::time::Duration;
     use tokio::time::sleep;
 
-
     #[test]
     fn test_from_tracer_tags_to_tracer_header_tags() {
         let tracer_tags = TracerMetadata {
@@ -1661,7 +1655,10 @@ mod tests {
             true,
         );
 
-        let v5: (Vec<BytesString>, Vec<Vec<libdd_trace_utils::span::v05::Span>>) = (vec![], vec![]);
+        let v5: (
+            Vec<BytesString>,
+            Vec<Vec<libdd_trace_utils::span::v05::Span>>,
+        ) = (vec![], vec![]);
         let traces = rmp_serde::to_vec(&v5).unwrap();
         let result = exporter.send(traces.as_ref()).unwrap();
         let AgentResponse::Changed { body } = result else {