diff --git a/libdd-data-pipeline-core/src/agentless/exporter.rs b/libdd-data-pipeline-core/src/agentless/exporter.rs index 5c97fb0918..bf4c3709d3 100644 --- a/libdd-data-pipeline-core/src/agentless/exporter.rs +++ b/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( + capabilities: &C, + traces: Vec>>, + 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( capabilities: &C, mut traces: Vec>>, 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( +async fn send_agentless_json( capabilities: &C, config: &AgentlessTraceConfig, mut headers: HeaderMap, json_body: Vec, + 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,7 +144,7 @@ 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, @@ -122,9 +152,11 @@ where &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)); diff --git a/libdd-data-pipeline-core/src/agentless/mod.rs b/libdd-data-pipeline-core/src/agentless/mod.rs index 2e25e90c13..9aa272813d 100644 --- a/libdd-data-pipeline-core/src/agentless/mod.rs +++ b/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}; diff --git a/libdd-data-pipeline-core/src/lib.rs b/libdd-data-pipeline-core/src/lib.rs index a274761075..34dae8610a 100644 --- a/libdd-data-pipeline-core/src/lib.rs +++ b/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; diff --git a/libdd-data-pipeline/src/agentless/exporter.rs b/libdd-data-pipeline/src/agentless/exporter.rs index fa8102b2b9..fda230ddb7 100644 --- a/libdd-data-pipeline/src/agentless/exporter.rs +++ b/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( +pub(crate) async fn send_agentless_traces_with_observer( capabilities: &C, traces: Vec>>, 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 { diff --git a/libdd-data-pipeline/src/otlp/exporter.rs b/libdd-data-pipeline/src/otlp/exporter.rs index 1297a7a25e..a05a04cc6a 100644 --- a/libdd-data-pipeline/src/otlp/exporter.rs +++ b/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( content_type: http::HeaderValue, body: Vec, 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, + max_retries: u32, + observer: F, ) -> Result<(), TraceExporterError> { let url = libdd_common::parse_uri(endpoint_url).map_err(|e| { TraceExporterError::Internal(InternalErrorKind::InvalidWorkerState(format!( @@ -67,7 +97,7 @@ pub(crate) async fn send_otlp_http( None, ); - match send_with_retry( + let result = send_with_retry( capabilities, &target, body, @@ -75,8 +105,9 @@ pub(crate) async fn send_otlp_http( &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( /// /// `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( capabilities: &C, config: &OtlpTraceConfig, diff --git a/libdd-data-pipeline/src/otlp/mod.rs b/libdd-data-pipeline/src/otlp/mod.rs index 0bda6b1b7e..073372ba02 100644 --- a/libdd-data-pipeline/src/otlp/mod.rs +++ b/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; diff --git a/libdd-data-pipeline/src/telemetry/metrics.rs b/libdd-data-pipeline/src/telemetry/metrics.rs index c003dfbecd..2e11a185ec 100644 --- a/libdd-data-pipeline/src/telemetry/metrics.rs +++ b/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. diff --git a/libdd-data-pipeline/src/telemetry/mod.rs b/libdd-data-pipeline/src/telemetry/mod.rs index c371436996..94a081b407 100644 --- a/libdd-data-pipeline/src/telemetry/mod.rs +++ b/libdd-data-pipeline/src/telemetry/mod.rs @@ -212,6 +212,9 @@ pub struct SendPayloadTelemetry { chunks_sent: u64, chunks_dropped_serialization_error: u64, chunks_dropped_send_failure: u64, + spans_enqueued_for_serialization: u64, + spans_dropped_serialization_error: u64, + spans_dropped_api_error: u64, responses_count_per_code: HashMap, } @@ -281,6 +284,24 @@ impl SendPayloadTelemetry { }; telemetry } + + pub(crate) fn from_retry_result_with_spans( + value: &SendWithRetryResult, + bytes_sent: u64, + chunks: u64, + spans: u64, + ) -> Self { + let mut telemetry = Self::from_retry_result(value, bytes_sent, chunks); + telemetry.spans_enqueued_for_serialization = spans; + match value { + Err(SendWithRetryError::Build(_)) => { + telemetry.spans_dropped_serialization_error = spans; + } + Err(_) => telemetry.spans_dropped_api_error = spans, + Ok(_) => {} + } + telemetry + } } impl TelemetryClient { @@ -333,6 +354,25 @@ impl Tel self.worker .add_point(data.chunks_dropped_send_failure as f64, key, vec![])?; } + if data.spans_enqueued_for_serialization > 0 { + let key = self + .metrics + .get(metrics::MetricKind::SpansEnqueuedForSerialization); + self.worker + .add_point(data.spans_enqueued_for_serialization as f64, key, vec![])?; + } + if data.spans_dropped_serialization_error > 0 { + let key = self + .metrics + .get(metrics::MetricKind::SpansDroppedSerializationError); + self.worker + .add_point(data.spans_dropped_serialization_error as f64, key, vec![])?; + } + if data.spans_dropped_api_error > 0 { + let key = self.metrics.get(metrics::MetricKind::SpansDroppedApiError); + self.worker + .add_point(data.spans_dropped_api_error as f64, key, vec![])?; + } if !data.responses_count_per_code.is_empty() { let key = self.metrics.get(metrics::MetricKind::ApiResponses); for (status_code, count) in &data.responses_count_per_code { @@ -789,11 +829,13 @@ mod tests { .body(Bytes::new()) .unwrap(); let result = Err(SendWithRetryError::Http(error_response, 5)); - let telemetry = SendPayloadTelemetry::from_retry_result(&result, 1, 2); + let telemetry = SendPayloadTelemetry::from_retry_result_with_spans(&result, 1, 2, 7); assert_eq!( telemetry, SendPayloadTelemetry { chunks_dropped_send_failure: 2, + spans_enqueued_for_serialization: 7, + spans_dropped_api_error: 7, requests_count: 5, errors_status_code: 1, responses_count_per_code: HashMap::from([(400, 1)]), diff --git a/libdd-data-pipeline/src/trace_exporter/mod.rs b/libdd-data-pipeline/src/trace_exporter/mod.rs index 83c8b84048..f0dc3b0d43 100644 --- a/libdd-data-pipeline/src/trace_exporter/mod.rs +++ b/libdd-data-pipeline/src/trace_exporter/mod.rs @@ -18,9 +18,10 @@ use self::metrics::MetricsEmitter; use self::stats::StatsComputationStatus; use self::trace_serializer::TraceSerializer; use crate::agent_info::ResponseObserver; -use crate::agentless::exporter::send_agentless_traces; +use crate::agentless::exporter::send_agentless_traces_with_observer; use crate::agentless::AgentlessTraceConfig; -use crate::otlp::{map_traces_to_otlp, send_otlp_traces_http, OtlpResourceInfo, OtlpTraceConfig}; +use crate::otlp::exporter::{send_otlp_http_with_observer, OTLP_MAX_RETRIES}; +use crate::otlp::{map_traces_to_otlp, OtlpResourceInfo, OtlpTraceConfig}; #[cfg(feature = "telemetry")] use crate::telemetry::{SendPayloadTelemetry, TelemetryClient}; use crate::trace_exporter::agent_response::{ @@ -69,6 +70,23 @@ const V04_TRACES_ENDPOINT: &str = "/v0.4/traces"; const V05_TRACES_ENDPOINT: &str = "/v0.5/traces"; const V1_TRACES_ENDPOINT: &str = "/v1.0/traces"; +#[derive(Clone, Copy)] +struct PayloadCounts { + chunks: usize, + #[cfg(feature = "telemetry")] + spans: usize, +} + +impl PayloadCounts { + fn from_traces(traces: &[Vec>]) -> Self { + Self { + chunks: traces.len(), + #[cfg(feature = "telemetry")] + spans: traces.iter().map(Vec::len).sum(), + } + } +} + /// Values for optional telemetry HTTP session headers (`dd-session-id`, root/parent). #[derive(Debug, Default, Clone)] pub struct TelemetryInstrumentationSessions { @@ -266,6 +284,26 @@ impl< .store(handle.map(|h| Arc::new(TelemetryClient::with_handle(h)))); } + #[cfg(feature = "telemetry")] + fn emit_serialization_drop(&self, counts: PayloadCounts) { + self.emit_retry_result(&Err(SendWithRetryError::Build(0)), 0, counts); + } + + #[cfg(feature = "telemetry")] + fn emit_retry_result(&self, result: &SendWithRetryResult, bytes: usize, counts: PayloadCounts) { + if let Some(telemetry) = self.telemetry.load_full().as_deref() { + let payload = SendPayloadTelemetry::from_retry_result_with_spans( + result, + bytes as u64, + counts.chunks as u64, + counts.spans as u64, + ); + if let Err(e) = telemetry.send(&payload) { + error!(?e, "Error sending telemetry"); + } + } + } + /// Stop the background workers owned by this exporter. /// /// Sync facade over [`Self::shutdown_async`]; panics inside an existing tokio context. @@ -617,15 +655,22 @@ impl< config: &AgentlessTraceConfig, client_side_stats: bool, ) -> Result { - // When local stats computation is active (agentless stats path), the - // intake must not also compute stats for these traces. Suppressing - // `_dd.compute_stats=1` prevents double-counting. - send_agentless_traces( + #[cfg(feature = "telemetry")] + let counts = PayloadCounts::from_traces(&traces); + send_agentless_traces_with_observer( &self.capabilities, traces, &self.metadata, config, client_side_stats, + |_result, _payload_len| { + #[cfg(feature = "telemetry")] + self.emit_retry_result(_result, _payload_len, counts); + }, + || { + #[cfg(feature = "telemetry")] + self.emit_serialization_drop(counts); + }, ) .await?; Ok(AgentResponse::Unchanged) @@ -637,6 +682,8 @@ impl< traces: Vec>>, config: &OtlpTraceConfig, ) -> Result { + #[cfg(feature = "telemetry")] + let counts = PayloadCounts::from_traces(&traces); let resource_info = { let mut r = OtlpResourceInfo::default(); r.service = self.metadata.service.clone(); @@ -658,6 +705,8 @@ impl< map_traces_to_otlp(traces, &resource_info, config.otel_trace_semantics_enabled); let body = config.protocol.encode(&request).map_err(|e| { error!("OTLP serialization error: {e}"); + #[cfg(feature = "telemetry")] + self.emit_serialization_drop(counts); TraceExporterError::Internal(InternalErrorKind::InvalidWorkerState(format!( "failed to encode OTLP request: {e}" ))) @@ -677,13 +726,24 @@ impl< } else { config }; - send_otlp_traces_http( + #[cfg(feature = "telemetry")] + let payload_len = body.len(); + let result = send_otlp_http_with_observer( &self.capabilities, - config_to_use, + &config_to_use.endpoint_url, + &config_to_use.headers, + config_to_use.timeout, self.endpoint.test_token.as_deref(), + config_to_use.protocol.content_type(), body, + OTLP_MAX_RETRIES, + |_result| { + #[cfg(feature = "telemetry")] + self.emit_retry_result(_result, payload_len, counts); + }, ) - .await?; + .await; + result?; Ok(AgentResponse::Unchanged) } @@ -693,11 +753,10 @@ impl< endpoint: &Endpoint, mp_payload: Vec, headers: HeaderMap, - chunks: usize, + counts: PayloadCounts, ) -> Result { let strategy = RetryStrategy::default(); let payload_len = mp_payload.len(); - // Send traces to the agent let result = send_with_retry( &self.capabilities, @@ -710,17 +769,10 @@ impl< .await; #[cfg(feature = "telemetry")] - if let Some(telemetry) = self.telemetry.load_full().as_deref() { - if let Err(e) = telemetry.send(&SendPayloadTelemetry::from_retry_result( - &result, - payload_len as u64, - chunks as u64, - )) { - error!(?e, "Error sending telemetry"); - } - } + self.emit_retry_result(&result, payload_len, counts); - self.handle_send_result(result, chunks, payload_len).await + self.handle_send_result(result, counts.chunks, payload_len) + .await } /// Synchronous log-export path: encode every span to newline-delimited @@ -799,6 +851,7 @@ impl< // Snapshot the effective format once so the serializer and the URL agree even if // `v1_active` flips mid-send (the background `/info` fetcher can race us otherwise). let effective_format = self.effective_output_format(); + let counts = PayloadCounts::from_traces(&traces); let prepared = match self.serializer.prepare_traces_payload( traces, @@ -814,9 +867,15 @@ impl< HealthMetric::Count(health_metrics::SERIALIZE_TRACES_ERRORS, 1), None, ); + #[cfg(feature = "telemetry")] + self.emit_serialization_drop(counts); return Err(e); } }; + let counts = PayloadCounts { + chunks: prepared.chunk_count, + ..counts + }; let endpoint = Endpoint { url: effective_format.add_path(&self.endpoint.url), @@ -824,12 +883,7 @@ impl< }; let result = self - .send_traces_with_telemetry( - &endpoint, - prepared.data, - prepared.headers, - prepared.chunk_count, - ) + .send_traces_with_telemetry(&endpoint, prepared.data, prepared.headers, counts) .await; // State-hash trap mitigation: the agent does not return a `Datadog-Agent-State` @@ -2398,7 +2452,8 @@ mod telemetry_metrics_tests { use libdd_capabilities_impl::NativeCapabilities; use libdd_shared_runtime::ForkSafeRuntime; use libdd_tinybytes::BytesString; - use libdd_trace_utils::span::v05; + use libdd_trace_utils::msgpack_encoder; + use libdd_trace_utils::span::{v04::SpanBytes, v05}; // v05 messagepack empty payload -> [[""], []] const V5_EMPTY: [u8; 4] = [0x92, 0x91, 0xA0, 0x90]; @@ -2423,6 +2478,7 @@ mod telemetry_metrics_tests { let metrics_endpoint = server.mock(|when, then| { when.method(POST) .body_includes("\"metric\":\"trace_api.bytes\"") + .body_includes("\"metric\":\"spans_enqueued_for_serialization\"") .path("/telemetry/proxy/api/v2/apmtelemetry"); then.status(200) .header("content-type", "application/json") @@ -2444,7 +2500,7 @@ mod telemetry_metrics_tests { }); let exporter = builder.build::().unwrap(); - let traces = vec![0x90]; + let traces = msgpack_encoder::v04::to_vec_from_v04(&[vec![SpanBytes::default()]]); let result = exporter.send(traces.as_ref()).unwrap(); let AgentResponse::Changed { body } = result else { panic!("Expected Changed response"); diff --git a/libdd-trace-utils/src/send_with_retry/mod.rs b/libdd-trace-utils/src/send_with_retry/mod.rs index e88937e431..4cb3b98f1d 100644 --- a/libdd-trace-utils/src/send_with_retry/mod.rs +++ b/libdd-trace-utils/src/send_with_retry/mod.rs @@ -107,6 +107,28 @@ pub async fn send_with_retry( retry_strategy: &RetryStrategy, compression_strategy: CompressionStrategy, ) -> SendWithRetryResult { + send_with_retry_and_size( + capabilities, + target, + payload, + headers, + retry_strategy, + compression_strategy, + ) + .await + .0 +} + +/// Send a payload with retries and return its post-compression size. +#[allow(clippy::result_large_err)] +pub async fn send_with_retry_and_size( + capabilities: &C, + target: &Endpoint, + payload: Vec, + headers: &HeaderMap, + retry_strategy: &RetryStrategy, + compression_strategy: CompressionStrategy, +) -> (SendWithRetryResult, usize) { let mut request_attempt = 0; let timeout = Duration::from_millis(target.timeout_ms); @@ -119,8 +141,9 @@ pub async fn send_with_retry( let (compressed, compression_strategy) = compression::compress(payload, compression_strategy); let payload = Bytes::from(compressed); + let payload_size = payload.len(); - loop { + let result = loop { request_attempt += 1; debug!( @@ -144,7 +167,7 @@ pub async fn send_with_retry( let req = match builder.body(payload.clone()) { Ok(r) => r, Err(_) => { - return Err(SendWithRetryError::Build(request_attempt)); + break Err(SendWithRetryError::Build(request_attempt)); } }; @@ -188,7 +211,7 @@ pub async fn send_with_retry( attempts = request_attempt, "Max retries exceeded, returning HTTP error" ); - return Err(SendWithRetryError::Http(response, request_attempt)); + break Err(SendWithRetryError::Http(response, request_attempt)); } } else { debug!( @@ -196,7 +219,7 @@ pub async fn send_with_retry( attempts = request_attempt, "Request succeeded" ); - return Ok((response, request_attempt)); + break Ok((response, request_attempt)); } } Ok(Err(e)) => { @@ -230,7 +253,7 @@ pub async fn send_with_retry( attempts = request_attempt, "Max retries exceeded, returning request error" ); - return Err(classified_error); + break Err(classified_error); } } Err(_) => { @@ -254,11 +277,12 @@ pub async fn send_with_retry( attempts = request_attempt, "Max retries exceeded, returning timeout error" ); - return Err(SendWithRetryError::Timeout(request_attempt)); + break Err(SendWithRetryError::Timeout(request_attempt)); } } } - } + }; + (result, payload_size) } #[cfg(test)]