From 52710cf7a3af3e9a13998e04409417eb618c1e44 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Mon, 6 Jul 2026 19:04:26 +0200 Subject: [PATCH 01/16] fix: attach obfuscation flag to stats buckets, fix obfuscation version condition --- .../src/trace_exporter/stats.rs | 3 +- .../src/span_concentrator/aggregation.rs | 11 ++++- .../src/span_concentrator/mod.rs | 42 ++++++++++++------- 3 files changed, 40 insertions(+), 16 deletions(-) diff --git a/libdd-data-pipeline/src/trace_exporter/stats.rs b/libdd-data-pipeline/src/trace_exporter/stats.rs index 2fdaa29728..3c6c27d51a 100644 --- a/libdd-data-pipeline/src/trace_exporter/stats.rs +++ b/libdd-data-pipeline/src/trace_exporter/stats.rs @@ -107,7 +107,7 @@ fn is_obfuscation_active(agent_info: &AgentInfo) -> bool { agent_info .info .obfuscation_version - .is_some_and(|v| v >= 1 && v <= SUPPORTED_OBFUSCATION_VERSION) + .is_some_and(|v| v >= 1 && v == SUPPORTED_OBFUSCATION_VERSION) } #[cfg(not(target_arch = "wasm32"))] @@ -258,6 +258,7 @@ fn update_obfuscation_config( ) { let obfuscation_active = client_side_stats.obfuscation_enabled && is_obfuscation_active(agent_info); + // FIXME: there is more than this to obfuscation config let sql_obfuscation_mode = (|| { agent_info .info diff --git a/libdd-trace-stats/src/span_concentrator/aggregation.rs b/libdd-trace-stats/src/span_concentrator/aggregation.rs index 5d17d54056..aaa94d2ab1 100644 --- a/libdd-trace-stats/src/span_concentrator/aggregation.rs +++ b/libdd-trace-stats/src/span_concentrator/aggregation.rs @@ -431,6 +431,9 @@ pub(super) struct StatsBucket { max_entries: usize, /// Number of spans collapsed into the overflow bucket due to cardinality limiting. collapsed_count: u64, + #[cfg(feature = "stats-obfuscation")] + /// Indicates if stats obfuscated in this bucket. This is set once at creation and stays constant per bucket + pub(super) obfuscated: bool, } impl StatsBucket { @@ -438,12 +441,18 @@ impl StatsBucket { /// /// `max_entries` is the maximum number of distinct aggregation keys the bucket will hold. /// Once the limit is reached, new distinct keys are collapsed into the overflow sentinel key. - pub(super) fn new(start_timestamp: u64, max_entries: usize) -> Self { + pub(super) fn new( + start_timestamp: u64, + max_entries: usize, + #[cfg(feature = "stats-obfuscation")] obfuscation_enabled: bool, + ) -> Self { Self { data: HashMap::new(), start: start_timestamp, max_entries, collapsed_count: 0, + #[cfg(feature = "stats-obfuscation")] + obfuscated: obfuscation_enabled, } } diff --git a/libdd-trace-stats/src/span_concentrator/mod.rs b/libdd-trace-stats/src/span_concentrator/mod.rs index f1beb5eab9..f555f2c865 100644 --- a/libdd-trace-stats/src/span_concentrator/mod.rs +++ b/libdd-trace-stats/src/span_concentrator/mod.rs @@ -190,7 +190,23 @@ impl SpanConcentrator { if bucket_timestamp < self.oldest_timestamp { bucket_timestamp = self.oldest_timestamp; } - let obfuscated_resource = self.compute_obfuscated_span(span); + + let target_bucket = self.buckets.entry(bucket_timestamp).or_insert_with(|| { + StatsBucket::new( + bucket_timestamp, + self.max_entries_per_bucket, + #[cfg(feature = "stats-obfuscation")] + self.obfuscation_config.load().enabled, + ) + }); + #[cfg(feature = "stats-obfuscation")] + let obfuscated_resource = Self::compute_obfuscated_span( + target_bucket.obfuscated, + self.obfuscation_config.load().sql_obfuscation_mode, + span, + ); + #[cfg(not(feature = "stats-obfuscation"))] + let obfuscated_resource: Option = None; let agg_key = match obfuscated_resource.as_deref() { Some(res) => BorrowedAggregationKey::from_obfuscated_span( res, @@ -199,29 +215,27 @@ impl SpanConcentrator { ), None => BorrowedAggregationKey::from_span(span, self.peer_tag_keys.as_slice()), }; - self.buckets - .entry(bucket_timestamp) - .or_insert_with(|| StatsBucket::new(bucket_timestamp, self.max_entries_per_bucket)) - .insert( - agg_key, - span.duration(), - span.is_error(), - span.has_top_level(), - ); + target_bucket.insert( + agg_key, + span.duration(), + span.is_error(), + span.has_top_level(), + ); } + #[cfg(feature = "stats-obfuscation")] fn compute_obfuscated_span<'a>( - &self, + obfuscate: bool, + sql_obfuscation_mode: libdd_trace_obfuscation::sql::SqlObfuscationMode, #[allow(unused)] span: &'a impl StatSpan<'a>, ) -> Option { - #[cfg(feature = "stats-obfuscation")] - if self.obfuscation_config.load().enabled { + if obfuscate { let dbms_hint: Option<&str> = span.get_meta("db.type"); return libdd_trace_obfuscation::obfuscate::obfuscate_resource_for_stats( span.r#type(), span.resource(), dbms_hint, - self.obfuscation_config.load().sql_obfuscation_mode, + sql_obfuscation_mode, ); } None From a62a15009192ce23211f6f4ddea79c12d7c8f0bd Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Tue, 7 Jul 2026 11:48:21 +0200 Subject: [PATCH 02/16] fix: fmt --- libdd-trace-stats/src/span_concentrator/aggregation.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/libdd-trace-stats/src/span_concentrator/aggregation.rs b/libdd-trace-stats/src/span_concentrator/aggregation.rs index aaa94d2ab1..12be89b535 100644 --- a/libdd-trace-stats/src/span_concentrator/aggregation.rs +++ b/libdd-trace-stats/src/span_concentrator/aggregation.rs @@ -432,7 +432,8 @@ pub(super) struct StatsBucket { /// Number of spans collapsed into the overflow bucket due to cardinality limiting. collapsed_count: u64, #[cfg(feature = "stats-obfuscation")] - /// Indicates if stats obfuscated in this bucket. This is set once at creation and stays constant per bucket + /// Indicates if stats obfuscated in this bucket. This is set once at creation and stays + /// constant per bucket pub(super) obfuscated: bool, } From 8f65b1d65ce68c9bde61b5ca16add5fa5ff671ce Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Tue, 7 Jul 2026 13:55:46 +0200 Subject: [PATCH 03/16] fix: flush buckets which have the same obfuscation indicator together --- datadog-ipc/src/shm_stats.rs | 4 +- datadog-sidecar/src/service/stats_flusher.rs | 6 - libdd-data-pipeline/src/otlp/metrics.rs | 2 +- .../src/trace_exporter/stats.rs | 2 - .../src/span_concentrator/mod.rs | 46 +++++-- libdd-trace-stats/src/stats_exporter.rs | 127 ++++++++---------- 6 files changed, 92 insertions(+), 95 deletions(-) diff --git a/datadog-ipc/src/shm_stats.rs b/datadog-ipc/src/shm_stats.rs index 6a7c513fda..bc5351e142 100644 --- a/datadog-ipc/src/shm_stats.rs +++ b/datadog-ipc/src/shm_stats.rs @@ -820,8 +820,8 @@ impl ShmSpanConcentrator { } impl FlushableConcentrator for ShmSpanConcentrator { - fn flush_buckets(&mut self, force: bool) -> (Vec, u64) { - (self.drain_buckets(force), 0) + fn flush_buckets(&mut self, force: bool) -> (Vec, u64, bool) { + (self.drain_buckets(force), 0, false) } } diff --git a/datadog-sidecar/src/service/stats_flusher.rs b/datadog-sidecar/src/service/stats_flusher.rs index 13332bbb82..ac144e669d 100644 --- a/datadog-sidecar/src/service/stats_flusher.rs +++ b/datadog-sidecar/src/service/stats_flusher.rs @@ -111,12 +111,6 @@ fn make_exporter( s.meta.clone(), endpoint, NativeCapabilities::new_client(), - // Sidecar does not perform client-side stats obfuscation. Pass a disabled - // default so the `datadog-obfuscation-version` header is never sent. - #[cfg(feature = "stats-obfuscation")] - Arc::new(arc_swap::ArcSwap::from_pointee( - libdd_trace_stats::span_concentrator::StatsComputationObfuscationConfig::default(), - )), #[cfg(feature = "stats-obfuscation")] "0", None, diff --git a/libdd-data-pipeline/src/otlp/metrics.rs b/libdd-data-pipeline/src/otlp/metrics.rs index 46104b12f1..a02fbd0b14 100644 --- a/libdd-data-pipeline/src/otlp/metrics.rs +++ b/libdd-data-pipeline/src/otlp/metrics.rs @@ -255,7 +255,7 @@ pub struct OtlpStatsExporter { impl OtlpStatsExporter { /// Flush the concentrator and export stats; returns `Ok(true)` if anything was sent. async fn send(&self, force_flush: bool, max_retries: u32) -> anyhow::Result { - let (buckets, _collapsed_count) = { + let (buckets, _collapsed_count, _buckets_obfuscated) = { #[allow(clippy::unwrap_used)] let mut c = self.concentrator.lock().unwrap(); c.flush_with_otlp_exact(SystemTime::now(), force_flush) diff --git a/libdd-data-pipeline/src/trace_exporter/stats.rs b/libdd-data-pipeline/src/trace_exporter/stats.rs index 3c6c27d51a..fdb66760e6 100644 --- a/libdd-data-pipeline/src/trace_exporter/stats.rs +++ b/libdd-data-pipeline/src/trace_exporter/stats.rs @@ -170,8 +170,6 @@ fn create_and_start_stats_worker< Endpoint::from_url(add_path(ctx.endpoint_url, STATS_ENDPOINT)), capabilities.clone(), #[cfg(feature = "stats-obfuscation")] - client_side_stats.obfuscation_config.clone(), - #[cfg(feature = "stats-obfuscation")] SUPPORTED_OBFUSCATION_VERSION_STR, #[cfg(feature = "telemetry")] ctx.telemetry.clone(), diff --git a/libdd-trace-stats/src/span_concentrator/mod.rs b/libdd-trace-stats/src/span_concentrator/mod.rs index f555f2c865..a1b504aae0 100644 --- a/libdd-trace-stats/src/span_concentrator/mod.rs +++ b/libdd-trace-stats/src/span_concentrator/mod.rs @@ -21,13 +21,20 @@ pub use stat_span::StatSpan; /// `StatsExporter` is generic over `C: FlushableConcentrator` so it can work with /// both the in-process [`SpanConcentrator`] and the SHM-backed `ShmSpanConcentrator`. pub trait FlushableConcentrator { - /// Flush time buckets and return them together with the number of spans that were - /// collapsed into the overflow sentinel bucket due to cardinality limiting. - fn flush_buckets(&mut self, force: bool) -> (Vec, u64); + /// Flush time buckets and return them together some metadata. + /// + /// Returns a triplet `(buckets, collapsed_spans, buckets_obfuscated)` + /// Where + /// - `buckets` are the encoded stats bucket + /// - `collapsed_spans` is the number of spans that were collapsed into the overflow sentinel + /// bucket due to cardinality limiting + /// - `buckets_obfuscated` indicates whether the returned buckets have been obfuscated + /// client-side + fn flush_buckets(&mut self, force: bool) -> (Vec, u64, bool); } impl FlushableConcentrator for SpanConcentrator { - fn flush_buckets(&mut self, force: bool) -> (Vec, u64) { + fn flush_buckets(&mut self, force: bool) -> (Vec, u64, bool) { self.flush(SystemTime::now(), force) } } @@ -247,7 +254,11 @@ impl SpanConcentrator { /// Returns a tuple of `(buckets, collapsed_spans)` where `collapsed_spans` is the total number /// of spans that were collapsed into the overflow sentinel bucket due to cardinality limiting /// across all flushed time buckets. - pub fn flush(&mut self, now: SystemTime, force: bool) -> (Vec, u64) { + pub fn flush( + &mut self, + now: SystemTime, + force: bool, + ) -> (Vec, u64, bool) { self.drain_due_buckets(now, force, StatsBucket::flush) } @@ -262,7 +273,7 @@ impl SpanConcentrator { &mut self, now: SystemTime, force: bool, - ) -> (Vec, u64) { + ) -> (Vec, u64, bool) { self.drain_due_buckets(now, force, StatsBucket::flush_with_otlp_exact) } @@ -271,10 +282,18 @@ impl SpanConcentrator { now: SystemTime, force: bool, encode: impl Fn(StatsBucket, u64) -> T, - ) -> (Vec, u64) { + ) -> (Vec, u64, bool) { // TODO: Wait for HashMap::extract_if to be stabilized to avoid a full drain let now_timestamp = system_time_to_unix_duration(now).as_nanos() as u64; - let buckets: Vec<(u64, StatsBucket)> = self.buckets.drain().collect(); + let mut buckets: Vec<(u64, StatsBucket)> = self.buckets.drain().collect(); + buckets.sort_unstable_by_key(|(timestamp, _)| *timestamp); + #[cfg(not(feature = "stats-obfuscation"))] + let first_bucket_is_obfuscated = false; + #[cfg(feature = "stats-obfuscation")] + let first_bucket_is_obfuscated = buckets + .first() + .map(|(_, bucket)| bucket.obfuscated) + .unwrap_or(false); self.oldest_timestamp = if force { align_timestamp(now_timestamp, self.bucket_size) } else { @@ -292,8 +311,13 @@ impl SpanConcentrator { // if the tracer stops while the latest buckets aren't old enough to be flushed. // The "force" boolean skips the delay and flushes all buckets, typically on // shutdown. - if !force && timestamp > (now_timestamp - self.buffer_len as u64 * self.bucket_size) - { + let keep = !force + && timestamp > (now_timestamp - self.buffer_len as u64 * self.bucket_size); + // Even when forcing to flush, we refuse to mix obfuscated buckets from + // un-obfuscated buckets. This means you need to flush twice to flush it all + #[cfg(feature = "stats-obfuscation")] + let keep = keep || bucket.obfuscated != first_bucket_is_obfuscated; + if keep { self.buckets.insert(timestamp, bucket); return None; } @@ -304,7 +328,7 @@ impl SpanConcentrator { if total_collapsed > 0 { debug!(max_entries_per_bucket = self.max_entries_per_bucket, total_collapsed, "Client-side stats values have been collapsed to 'tracer_blocked_value'. This is due to the cardinality exceeding DD_TRACE_STATS_CARDINALITY_LIMIT"); } - (buckets_pb, total_collapsed) + (buckets_pb, total_collapsed, first_bucket_is_obfuscated) } } diff --git a/libdd-trace-stats/src/stats_exporter.rs b/libdd-trace-stats/src/stats_exporter.rs index c771fd5b2b..3cbd804750 100644 --- a/libdd-trace-stats/src/stats_exporter.rs +++ b/libdd-trace-stats/src/stats_exporter.rs @@ -9,8 +9,6 @@ use std::{ time, }; -#[cfg(feature = "stats-obfuscation")] -use crate::span_concentrator::SharedStatsComputationObfuscationConfig; use crate::span_concentrator::{FlushableConcentrator, SpanConcentrator}; use async_trait::async_trait; use libdd_capabilities::{HttpClientCapability, MaybeSend, SleepCapability}; @@ -96,8 +94,6 @@ pub struct StatsExporter< sequence_id: AtomicU64, capabilities: Cap, #[cfg(feature = "stats-obfuscation")] - obfuscation_config: SharedStatsComputationObfuscationConfig, - #[cfg(feature = "stats-obfuscation")] supported_obfuscation_version: &'static str, /// Optional telemetry handle and context key. #[cfg(feature = "telemetry")] @@ -127,8 +123,6 @@ impl meta: StatsMetadata, endpoint: Endpoint, capabilities: Cap, - #[cfg(feature = "stats-obfuscation")] - obfuscation_config: SharedStatsComputationObfuscationConfig, #[cfg(feature = "stats-obfuscation")] supported_obfuscation_version: &'static str, #[cfg(feature = "telemetry")] telemetry: Option< libdd_telemetry::worker::TelemetryWorkerHandle, @@ -154,8 +148,6 @@ impl sequence_id: AtomicU64::new(0), capabilities, #[cfg(feature = "stats-obfuscation")] - obfuscation_config, - #[cfg(feature = "stats-obfuscation")] supported_obfuscation_version, #[cfg(feature = "telemetry")] telemetry, @@ -181,59 +173,67 @@ impl /// case stats cannot be flushed since the concentrator might be corrupted. /// Returns `Ok(true)` if stats were sent, `Ok(false)` if the concentrator had nothing to send. pub async fn send(&self, force_flush: bool) -> anyhow::Result { - let (payload, collapsed_spans) = self.flush(force_flush); - - if collapsed_spans > 0 { - #[cfg(feature = "telemetry")] - if let Some((handle, key)) = &self.telemetry { - let _ = handle.add_point(collapsed_spans as f64, key, vec![]); + // We flush twice with `force_flush` so that if we have a mix of obfuscated and unobfuscated + // buckets, both get flushed in separate payloads + let flush_count = if force_flush { 2 } else { 1 }; + for _ in 0..flush_count { + let (payload, collapsed_spans, buckets_obfuscated) = self.flush(force_flush); + + if collapsed_spans > 0 { + #[cfg(feature = "telemetry")] + if let Some((handle, key)) = &self.telemetry { + let _ = handle.add_point(collapsed_spans as f64, key, vec![]); + } + #[cfg(feature = "dogstatsd")] + if let Some(client) = &self.dogstatsd { + client.send(vec![libdd_dogstatsd_client::DogStatsDAction::Count( + COLLAPSED_SPANS_HEALTH_METRIC, + collapsed_spans as i64, + [tag!("collapsed_spans", "whole_key")].iter(), + )]); + } } - #[cfg(feature = "dogstatsd")] - if let Some(client) = &self.dogstatsd { - client.send(vec![libdd_dogstatsd_client::DogStatsDAction::Count( - COLLAPSED_SPANS_HEALTH_METRIC, - collapsed_spans as i64, - [tag!("collapsed_spans", "whole_key")].iter(), - )]); - } - } - if payload.stats.is_empty() { - return Ok(false); - } - let body = rmp_serde::encode::to_vec_named(&payload)?; - - let mut headers: http::HeaderMap = TracerHeaderTags::from(&self.meta).into(); + if payload.stats.is_empty() { + return Ok(false); + } + let body = rmp_serde::encode::to_vec_named(&payload)?; - headers.insert( - http::header::CONTENT_TYPE, - libdd_common::header::APPLICATION_MSGPACK, - ); + let mut headers: http::HeaderMap = TracerHeaderTags::from(&self.meta).into(); - #[cfg(feature = "stats-obfuscation")] - if self.obfuscation_config.load().enabled { headers.insert( - http::HeaderName::from_static("datadog-obfuscation-version"), - http::HeaderValue::from_static(self.supported_obfuscation_version), + http::header::CONTENT_TYPE, + libdd_common::header::APPLICATION_MSGPACK, ); - } - let result = send_with_retry( - &self.capabilities, - &self.endpoint, - body, - &headers, - &RetryStrategy::default(), - ) - .await; + #[cfg(feature = "stats-obfuscation")] + if buckets_obfuscated { + headers.insert( + http::HeaderName::from_static("datadog-obfuscation-version"), + http::HeaderValue::from_static(self.supported_obfuscation_version), + ); + } + #[cfg(not(feature = "stats-obfuscation"))] + let _ = buckets_obfuscated; + + let result = send_with_retry( + &self.capabilities, + &self.endpoint, + body, + &headers, + &RetryStrategy::default(), + ) + .await; - match result { - Ok(_) => Ok(true), - Err(err) => { - error!(?err, "Error with the StateExporter when sending stats"); - anyhow::bail!("Failed to send stats: {err}"); + match result { + Ok(_) => {} + Err(err) => { + error!(?err, "Error with the StateExporter when sending stats"); + anyhow::bail!("Failed to send stats: {err}"); + } } } + Ok(true) } /// Flush stats from the concentrator into a payload @@ -245,13 +245,13 @@ impl /// # Panic /// Will panic if another thread panicked while holding the concentrator lock in which /// case stats cannot be flushed since the concentrator might be corrupted. - fn flush(&self, force_flush: bool) -> (pb::ClientStatsPayload, u64) { + fn flush(&self, force_flush: bool) -> (pb::ClientStatsPayload, u64, bool) { let sequence = self.sequence_id.fetch_add(1, Ordering::Relaxed); #[allow(clippy::unwrap_used)] - let (buckets, collapsed_spans) = + let (buckets, collapsed_spans, buckets_obfuscated) = self.concentrator.lock().unwrap().flush_buckets(force_flush); let payload = encode_stats_payload(&self.meta, sequence, buckets); - (payload, collapsed_spans) + (payload, collapsed_spans, buckets_obfuscated) } } @@ -404,8 +404,6 @@ mod tests { Endpoint::from_url(stats_url_from_agent_url(&server.url("/")).unwrap()), NativeCapabilities::new_client(), #[cfg(feature = "stats-obfuscation")] - StatsComputationObfuscationConfig::disabled(), - #[cfg(feature = "stats-obfuscation")] "1", #[cfg(feature = "telemetry")] None, @@ -439,8 +437,6 @@ mod tests { Endpoint::from_url(stats_url_from_agent_url(&server.url("/")).unwrap()), NativeCapabilities::new_client(), #[cfg(feature = "stats-obfuscation")] - StatsComputationObfuscationConfig::disabled(), - #[cfg(feature = "stats-obfuscation")] "1", #[cfg(feature = "telemetry")] None, @@ -482,8 +478,6 @@ mod tests { Endpoint::from_url(stats_url_from_agent_url(&server.url("/")).unwrap()), caps.clone(), #[cfg(feature = "stats-obfuscation")] - StatsComputationObfuscationConfig::disabled(), - #[cfg(feature = "stats-obfuscation")] "1", #[cfg(feature = "telemetry")] None, @@ -531,8 +525,6 @@ mod tests { Endpoint::from_url(stats_url_from_agent_url(&server.url("/")).unwrap()), caps.clone(), #[cfg(feature = "stats-obfuscation")] - StatsComputationObfuscationConfig::disabled(), - #[cfg(feature = "stats-obfuscation")] "1", #[cfg(feature = "telemetry")] None, @@ -603,11 +595,6 @@ mod tests { Endpoint::from_url(stats_url_from_agent_url(&server.url("/")).unwrap()), NativeCapabilities::new_client(), #[cfg(feature = "stats-obfuscation")] - Arc::new(ArcSwap::from_pointee(StatsComputationObfuscationConfig { - enabled: true, - ..Default::default() - })), - #[cfg(feature = "stats-obfuscation")] "1", #[cfg(feature = "telemetry")] None, @@ -703,8 +690,6 @@ mod tests { Endpoint::from_url(stats_url_from_agent_url(&server.url("/")).unwrap()), NativeCapabilities::new_client(), #[cfg(feature = "stats-obfuscation")] - StatsComputationObfuscationConfig::disabled(), - #[cfg(feature = "stats-obfuscation")] "1", #[cfg(feature = "telemetry")] None, @@ -754,8 +739,6 @@ mod tests { Endpoint::from_url(stats_url_from_agent_url(&server.url("/")).unwrap()), NativeCapabilities::new_client(), #[cfg(feature = "stats-obfuscation")] - StatsComputationObfuscationConfig::disabled(), - #[cfg(feature = "stats-obfuscation")] "1", #[cfg(feature = "telemetry")] None, @@ -806,8 +789,6 @@ mod tests { Endpoint::from_url(stats_url_from_agent_url(&server.url("/")).unwrap()), NativeCapabilities::new_client(), #[cfg(feature = "stats-obfuscation")] - StatsComputationObfuscationConfig::disabled(), - #[cfg(feature = "stats-obfuscation")] "1", #[cfg(feature = "telemetry")] Some(handle), From a0bfe8c77972892a267154e1d19e5740812262a7 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Tue, 7 Jul 2026 13:56:46 +0200 Subject: [PATCH 04/16] fix(sidecar): remove useless stats-obfuscation feature --- datadog-sidecar/Cargo.toml | 1 - datadog-sidecar/src/service/stats_flusher.rs | 2 -- 2 files changed, 3 deletions(-) diff --git a/datadog-sidecar/Cargo.toml b/datadog-sidecar/Cargo.toml index b3f8e4b28b..905919c4b0 100644 --- a/datadog-sidecar/Cargo.toml +++ b/datadog-sidecar/Cargo.toml @@ -12,7 +12,6 @@ bench = false default = ["tracing"] tracing = ["tracing/std", "tracing-log", "tracing-subscriber", "arc-swap"] tokio-console = ["tokio/full", "tokio/tracing", "console-subscriber"] -stats-obfuscation = ["libdd-trace-stats/stats-obfuscation", "libdd-data-pipeline/stats-obfuscation", "arc-swap"] [dependencies] anyhow = { version = "1.0" } diff --git a/datadog-sidecar/src/service/stats_flusher.rs b/datadog-sidecar/src/service/stats_flusher.rs index ac144e669d..583f3ca7d9 100644 --- a/datadog-sidecar/src/service/stats_flusher.rs +++ b/datadog-sidecar/src/service/stats_flusher.rs @@ -111,8 +111,6 @@ fn make_exporter( s.meta.clone(), endpoint, NativeCapabilities::new_client(), - #[cfg(feature = "stats-obfuscation")] - "0", None, None, ) From 6c6f0586ef60c428c42af14c4b44b56d492e6e5c Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Tue, 7 Jul 2026 14:06:02 +0200 Subject: [PATCH 05/16] Revert "fix(sidecar): remove useless stats-obfuscation feature" This reverts commit a0bfe8c77972892a267154e1d19e5740812262a7. --- datadog-sidecar/Cargo.toml | 1 + datadog-sidecar/src/service/stats_flusher.rs | 2 ++ 2 files changed, 3 insertions(+) diff --git a/datadog-sidecar/Cargo.toml b/datadog-sidecar/Cargo.toml index 905919c4b0..b3f8e4b28b 100644 --- a/datadog-sidecar/Cargo.toml +++ b/datadog-sidecar/Cargo.toml @@ -12,6 +12,7 @@ bench = false default = ["tracing"] tracing = ["tracing/std", "tracing-log", "tracing-subscriber", "arc-swap"] tokio-console = ["tokio/full", "tokio/tracing", "console-subscriber"] +stats-obfuscation = ["libdd-trace-stats/stats-obfuscation", "libdd-data-pipeline/stats-obfuscation", "arc-swap"] [dependencies] anyhow = { version = "1.0" } diff --git a/datadog-sidecar/src/service/stats_flusher.rs b/datadog-sidecar/src/service/stats_flusher.rs index 583f3ca7d9..ac144e669d 100644 --- a/datadog-sidecar/src/service/stats_flusher.rs +++ b/datadog-sidecar/src/service/stats_flusher.rs @@ -111,6 +111,8 @@ fn make_exporter( s.meta.clone(), endpoint, NativeCapabilities::new_client(), + #[cfg(feature = "stats-obfuscation")] + "0", None, None, ) From a44467a6fe7f056136a95f538f7302c15a956db9 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Tue, 7 Jul 2026 14:08:12 +0200 Subject: [PATCH 06/16] fix: outdated comment, remove metadata return from otlp flush --- libdd-data-pipeline/src/otlp/metrics.rs | 2 +- .../src/span_concentrator/mod.rs | 27 +++++++++---------- 2 files changed, 14 insertions(+), 15 deletions(-) diff --git a/libdd-data-pipeline/src/otlp/metrics.rs b/libdd-data-pipeline/src/otlp/metrics.rs index a02fbd0b14..21c30c1785 100644 --- a/libdd-data-pipeline/src/otlp/metrics.rs +++ b/libdd-data-pipeline/src/otlp/metrics.rs @@ -255,7 +255,7 @@ pub struct OtlpStatsExporter { impl OtlpStatsExporter { /// Flush the concentrator and export stats; returns `Ok(true)` if anything was sent. async fn send(&self, force_flush: bool, max_retries: u32) -> anyhow::Result { - let (buckets, _collapsed_count, _buckets_obfuscated) = { + let buckets = { #[allow(clippy::unwrap_used)] let mut c = self.concentrator.lock().unwrap(); c.flush_with_otlp_exact(SystemTime::now(), force_flush) diff --git a/libdd-trace-stats/src/span_concentrator/mod.rs b/libdd-trace-stats/src/span_concentrator/mod.rs index a1b504aae0..6898101f7d 100644 --- a/libdd-trace-stats/src/span_concentrator/mod.rs +++ b/libdd-trace-stats/src/span_concentrator/mod.rs @@ -21,7 +21,8 @@ pub use stat_span::StatSpan; /// `StatsExporter` is generic over `C: FlushableConcentrator` so it can work with /// both the in-process [`SpanConcentrator`] and the SHM-backed `ShmSpanConcentrator`. pub trait FlushableConcentrator { - /// Flush time buckets and return them together some metadata. + /// Flush time buckets and return them together some metadata. If `force` is true, flush + /// all buckets. /// /// Returns a triplet `(buckets, collapsed_spans, buckets_obfuscated)` /// Where @@ -251,9 +252,13 @@ impl SpanConcentrator { /// Flush all stats bucket except for the `buffer_len` most recent. If `force` is true, flush /// all buckets. /// - /// Returns a tuple of `(buckets, collapsed_spans)` where `collapsed_spans` is the total number - /// of spans that were collapsed into the overflow sentinel bucket due to cardinality limiting - /// across all flushed time buckets. + /// Returns a triplet `(buckets, collapsed_spans, buckets_obfuscated)` + /// Where + /// - `buckets` are the encoded stats bucket + /// - `collapsed_spans` is the number of spans that were collapsed into the overflow sentinel + /// bucket due to cardinality limiting + /// - `buckets_obfuscated` indicates whether the returned buckets have been obfuscated + /// client-side pub fn flush( &mut self, now: SystemTime, @@ -265,16 +270,10 @@ impl SpanConcentrator { /// Like [`Self::flush`], but also emits exact per-cell scalars alongside each bucket for the /// OTLP trace-metrics path. The protobuf bucket inside each [`OtlpStatsBucket`] is identical /// to what [`Self::flush`] would produce, so the /v0.6/stats agent path is unaffected. - /// - /// Returns a tuple of `(buckets, collapsed_spans)` where `collapsed_spans` is the total number - /// of spans that were collapsed into the overflow sentinel bucket due to cardinality limiting - /// across all flushed time buckets. - pub fn flush_with_otlp_exact( - &mut self, - now: SystemTime, - force: bool, - ) -> (Vec, u64, bool) { - self.drain_due_buckets(now, force, StatsBucket::flush_with_otlp_exact) + pub fn flush_with_otlp_exact(&mut self, now: SystemTime, force: bool) -> Vec { + let (buckets, _, _) = + self.drain_due_buckets(now, force, StatsBucket::flush_with_otlp_exact); + buckets } fn drain_due_buckets( From 3113a7e503636c4ccc04e5e0c0768346afebf7b7 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Tue, 7 Jul 2026 14:29:20 +0200 Subject: [PATCH 07/16] fix: tests lints --- .../src/span_concentrator/tests.rs | 44 +++++++++---------- libdd-trace-stats/src/stats_exporter.rs | 4 -- 2 files changed, 22 insertions(+), 26 deletions(-) diff --git a/libdd-trace-stats/src/span_concentrator/tests.rs b/libdd-trace-stats/src/span_concentrator/tests.rs index 3fd088e09b..20021c862c 100644 --- a/libdd-trace-stats/src/span_concentrator/tests.rs +++ b/libdd-trace-stats/src/span_concentrator/tests.rs @@ -126,12 +126,12 @@ fn test_concentrator_oldest_timestamp_cold() { // Assert we didn't insert spans in older buckets for _ in 0..concentrator.buffer_len { - let (stats, _) = concentrator.flush(flushtime, false); + let (stats, _, _) = concentrator.flush(flushtime, false); assert_eq!(stats.len(), 0, "We should get 0 time buckets"); flushtime += Duration::from_nanos(concentrator.bucket_size); } - let (stats, _) = concentrator.flush(flushtime, false); + let (stats, _, _) = concentrator.flush(flushtime, false); assert_eq!(stats.len(), 1, "We should get exactly one time bucket"); @@ -188,12 +188,12 @@ fn test_concentrator_oldest_timestamp_hot() { } for _ in 0..(concentrator.buffer_len - 1) { - let (stats, _) = concentrator.flush(flushtime, false); + let (stats, _, _) = concentrator.flush(flushtime, false); assert!(stats.is_empty(), "We should get 0 time buckets"); flushtime += Duration::from_nanos(concentrator.bucket_size); } - let (stats, _) = concentrator.flush(flushtime, false); + let (stats, _, _) = concentrator.flush(flushtime, false); assert_eq!(stats.len(), 1, "We should get exactly one time bucket"); // First oldest bucket aggregates, it should have it all except the @@ -213,7 +213,7 @@ fn test_concentrator_oldest_timestamp_hot() { assert_counts_equal(expected, stats.first().unwrap().stats.clone()); flushtime += Duration::from_nanos(concentrator.bucket_size); - let (stats, _) = concentrator.flush(flushtime, false); + let (stats, _, _) = concentrator.flush(flushtime, false); assert_eq!(stats.len(), 1, "We should get exactly one time bucket"); // Stats of the last four spans. @@ -278,7 +278,7 @@ fn test_concentrator_stats_totals() { let mut flushtime = now; for _ in 0..=concentrator.buffer_len { - let (stats, _) = concentrator.flush(flushtime, false); + let (stats, _, _) = concentrator.flush(flushtime, false); if stats.is_empty() { continue; } @@ -528,7 +528,7 @@ fn test_concentrator_stats_counts() { let mut flushtime = now; for _ in 0..=concentrator.buffer_len + 2 { - let (stats, _) = concentrator.flush(flushtime, false); + let (stats, _, _) = concentrator.flush(flushtime, false); let expected_flushed_timestamps = align_timestamp( system_time_to_unix_duration(flushtime).as_nanos() as u64, concentrator.bucket_size, @@ -553,7 +553,7 @@ fn test_concentrator_stats_counts() { stats.first().unwrap().stats.clone(), ); - let (stats, _) = concentrator.flush(flushtime, false); + let (stats, _, _) = concentrator.flush(flushtime, false); assert_eq!( stats.len(), 0, @@ -658,7 +658,7 @@ fn test_span_should_be_included_in_stats() { }, ]; - let (stats, _) = concentrator.flush( + let (stats, _, _) = concentrator.flush( now + Duration::from_nanos(concentrator.bucket_size * concentrator.buffer_len as u64), false, ); @@ -696,7 +696,7 @@ fn test_ignore_partial_spans() { concentrator.add_span(span); } - let (stats, _) = concentrator.flush( + let (stats, _, _) = concentrator.flush( now + Duration::from_nanos(concentrator.bucket_size * concentrator.buffer_len as u64), false, ); @@ -727,10 +727,10 @@ fn test_force_flush() { let flushtime = now - Duration::from_secs(3600); // Bucket should not be flushed without force flush - let (stats, _) = concentrator.flush(flushtime, false); + let (stats, _, _) = concentrator.flush(flushtime, false); assert_eq!(0, stats.len()); - let (stats, _) = concentrator.flush(flushtime, true); + let (stats, _, _) = concentrator.flush(flushtime, true); assert_eq!(1, stats.len()); } @@ -900,7 +900,7 @@ fn test_peer_tags_aggregation() { }, ]; - let (stats_with_peer_tags, _) = concentrator_with_peer_tags.flush(flushtime, false); + let (stats_with_peer_tags, _, _) = concentrator_with_peer_tags.flush(flushtime, false); assert_counts_equal( expected_with_peer_tags, stats_with_peer_tags @@ -910,7 +910,7 @@ fn test_peer_tags_aggregation() { .clone(), ); - let (stats_without_peer_tags, _) = concentrator_without_peer_tags.flush(flushtime, false); + let (stats_without_peer_tags, _, _) = concentrator_without_peer_tags.flush(flushtime, false); assert_counts_equal( expected_without_peer_tags, stats_without_peer_tags @@ -1023,7 +1023,7 @@ fn test_peer_tags_quantization_aggregation() { ..Default::default() }]; - let (stats_with_peer_tags, _) = concentrator_with_peer_tags.flush(flushtime, false); + let (stats_with_peer_tags, _, _) = concentrator_with_peer_tags.flush(flushtime, false); assert_counts_equal( expected_with_peer_tags, stats_with_peer_tags @@ -1193,7 +1193,7 @@ fn test_base_service_peer_tag() { }, ]; - let (stats, _) = concentrator.flush(flushtime, false); + let (stats, _, _) = concentrator.flush(flushtime, false); assert_counts_equal( expected, stats @@ -1497,7 +1497,7 @@ fn test_pb_span() { // Flush and get stats let flushtime = now + Duration::from_nanos(concentrator.bucket_size * concentrator.buffer_len as u64); - let (stats, _) = concentrator.flush(flushtime, false); + let (stats, _, _) = concentrator.flush(flushtime, false); assert_eq!(stats.len(), 1, "Should get exactly one time bucket"); let bucket = &stats[0]; @@ -1610,7 +1610,7 @@ fn test_flush_with_otlp_exact_per_cell_scalars() { concentrator.add_span(s); } - let flushed = concentrator.flush_with_otlp_exact(now, true).0; + let flushed = concentrator.flush_with_otlp_exact(now, true); assert_eq!(flushed.len(), 1); let b = &flushed[0]; assert_eq!(b.exact.len(), 1); @@ -1676,7 +1676,7 @@ fn test_cardinality_limit_collapse() { concentrator.add_span(&span); } - let (buckets, _) = concentrator.flush(SystemTime::now(), true); + let (buckets, _, _) = concentrator.flush(SystemTime::now(), true); assert!(!buckets.is_empty(), "should get at least one time bucket"); let stats = &buckets[0].stats; @@ -1734,7 +1734,7 @@ fn test_overflow_bucket_counts() { concentrator.add_span(&span); } - let (buckets, _) = concentrator.flush(SystemTime::now(), true); + let (buckets, _, _) = concentrator.flush(SystemTime::now(), true); assert!(!buckets.is_empty()); let stats = &buckets[0].stats; @@ -1783,7 +1783,7 @@ fn test_no_collapse_within_limit() { concentrator.add_span(&span); } - let (buckets, _) = concentrator.flush(SystemTime::now(), true); + let (buckets, _, _) = concentrator.flush(SystemTime::now(), true); assert!(!buckets.is_empty()); let stats = &buckets[0].stats; @@ -1835,7 +1835,7 @@ fn test_overflow_bucket_key_sentinel_values() { concentrator.add_span(&first); concentrator.add_span(&second); - let (buckets, _) = concentrator.flush(SystemTime::now(), true); + let (buckets, _, _) = concentrator.flush(SystemTime::now(), true); assert!(!buckets.is_empty()); let stats = &buckets[0].stats; diff --git a/libdd-trace-stats/src/stats_exporter.rs b/libdd-trace-stats/src/stats_exporter.rs index 3cbd804750..6c45c369b1 100644 --- a/libdd-trace-stats/src/stats_exporter.rs +++ b/libdd-trace-stats/src/stats_exporter.rs @@ -316,8 +316,6 @@ pub fn stats_url_from_agent_url(agent_url: &str) -> anyhow::Result { #[cfg(test)] mod tests { use super::*; - #[cfg(feature = "stats-obfuscation")] - use crate::span_concentrator::StatsComputationObfuscationConfig; use httpmock::prelude::*; use httpmock::MockServer; use libdd_capabilities_impl::NativeCapabilities; @@ -573,8 +571,6 @@ mod tests { #[cfg_attr(miri, ignore)] #[tokio::test] async fn test_send_stats_with_obfuscation_header() { - use arc_swap::ArcSwap; - let server = MockServer::start_async().await; let mock = server From 909d654d1906b4bf9bf60b5ac36dcd59be90cb6b Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Tue, 7 Jul 2026 14:35:45 +0200 Subject: [PATCH 08/16] fix: local review suggestions --- libdd-data-pipeline/src/trace_exporter/stats.rs | 2 +- libdd-trace-stats/src/stats_exporter.rs | 7 ++++--- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/libdd-data-pipeline/src/trace_exporter/stats.rs b/libdd-data-pipeline/src/trace_exporter/stats.rs index fdb66760e6..03b9a4da02 100644 --- a/libdd-data-pipeline/src/trace_exporter/stats.rs +++ b/libdd-data-pipeline/src/trace_exporter/stats.rs @@ -256,7 +256,7 @@ fn update_obfuscation_config( ) { let obfuscation_active = client_side_stats.obfuscation_enabled && is_obfuscation_active(agent_info); - // FIXME: there is more than this to obfuscation config + // FIXME(APMSP-3720): there is more than this to obfuscation config let sql_obfuscation_mode = (|| { agent_info .info diff --git a/libdd-trace-stats/src/stats_exporter.rs b/libdd-trace-stats/src/stats_exporter.rs index 6c45c369b1..f0eae78441 100644 --- a/libdd-trace-stats/src/stats_exporter.rs +++ b/libdd-trace-stats/src/stats_exporter.rs @@ -176,6 +176,7 @@ impl // We flush twice with `force_flush` so that if we have a mix of obfuscated and unobfuscated // buckets, both get flushed in separate payloads let flush_count = if force_flush { 2 } else { 1 }; + let mut sent_stats = false; for _ in 0..flush_count { let (payload, collapsed_spans, buckets_obfuscated) = self.flush(force_flush); @@ -195,7 +196,7 @@ impl } if payload.stats.is_empty() { - return Ok(false); + continue; } let body = rmp_serde::encode::to_vec_named(&payload)?; @@ -226,14 +227,14 @@ impl .await; match result { - Ok(_) => {} + Ok(_) => {sent_stats = true;} Err(err) => { error!(?err, "Error with the StateExporter when sending stats"); anyhow::bail!("Failed to send stats: {err}"); } } } - Ok(true) + Ok(sent_stats) } /// Flush stats from the concentrator into a payload From 85eaf0cd6f807a2ce8986d19d3bcb9e95ee69be1 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Tue, 7 Jul 2026 14:44:37 +0200 Subject: [PATCH 09/16] fix: fmt --- libdd-trace-stats/src/stats_exporter.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/libdd-trace-stats/src/stats_exporter.rs b/libdd-trace-stats/src/stats_exporter.rs index f0eae78441..469ae2107a 100644 --- a/libdd-trace-stats/src/stats_exporter.rs +++ b/libdd-trace-stats/src/stats_exporter.rs @@ -227,7 +227,9 @@ impl .await; match result { - Ok(_) => {sent_stats = true;} + Ok(_) => { + sent_stats = true; + } Err(err) => { error!(?err, "Error with the StateExporter when sending stats"); anyhow::bail!("Failed to send stats: {err}"); From 05d2364a9fe097fc3e49fbd13e4fd738b4fdd3bd Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Tue, 7 Jul 2026 15:55:00 +0200 Subject: [PATCH 10/16] fix(test): put back obfuscation enabled in SpanConcentrator instead of StatsExporter --- libdd-trace-stats/src/stats_exporter.rs | 25 +++++++++++++++++++++++-- 1 file changed, 23 insertions(+), 2 deletions(-) diff --git a/libdd-trace-stats/src/stats_exporter.rs b/libdd-trace-stats/src/stats_exporter.rs index 469ae2107a..7698bf92af 100644 --- a/libdd-trace-stats/src/stats_exporter.rs +++ b/libdd-trace-stats/src/stats_exporter.rs @@ -354,6 +354,17 @@ mod tests { } fn get_test_concentrator() -> SpanConcentrator { + get_test_concentrator_with_obfuscation_config( + #[cfg(feature = "stats-obfuscation")] + None, + ) + } + + fn get_test_concentrator_with_obfuscation_config( + #[cfg(feature = "stats-obfuscation")] obfuscation_config: Option< + crate::span_concentrator::SharedStatsComputationObfuscationConfig, + >, + ) -> SpanConcentrator { let mut concentrator = SpanConcentrator::new( BUCKETS_DURATION, // Make sure the oldest bucket will be flushed on next send @@ -362,7 +373,7 @@ mod tests { vec![], None, #[cfg(feature = "stats-obfuscation")] - None, + obfuscation_config, ); let mut trace = vec![]; @@ -574,6 +585,9 @@ mod tests { #[cfg_attr(miri, ignore)] #[tokio::test] async fn test_send_stats_with_obfuscation_header() { + use crate::span_concentrator::StatsComputationObfuscationConfig; + use arc_swap::ArcSwap; + let server = MockServer::start_async().await; let mock = server @@ -587,9 +601,16 @@ mod tests { }) .await; + let concentrator = get_test_concentrator_with_obfuscation_config(Some(Arc::new( + ArcSwap::from_pointee(StatsComputationObfuscationConfig { + enabled: true, + ..Default::default() + }), + ))); + let stats_exporter = StatsExporter::new( BUCKETS_DURATION, - Arc::new(Mutex::new(get_test_concentrator())), + Arc::new(Mutex::new(concentrator)), get_test_metadata(), Endpoint::from_url(stats_url_from_agent_url(&server.url("/")).unwrap()), NativeCapabilities::new_client(), From ed0277a1cc71871744ab42ec769df139fc465b2f Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Tue, 7 Jul 2026 15:55:17 +0200 Subject: [PATCH 11/16] fix: remove unused file results.xml it got introduced by mistake in 3d8ec6c --- libdd-data-pipeline/results.xml | 0 1 file changed, 0 insertions(+), 0 deletions(-) delete mode 100644 libdd-data-pipeline/results.xml diff --git a/libdd-data-pipeline/results.xml b/libdd-data-pipeline/results.xml deleted file mode 100644 index e69de29bb2..0000000000 From 888e56c7e6dc739c3c319e31d9291268df0a7f51 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin <90446228+Eldolfin@users.noreply.github.com> Date: Tue, 7 Jul 2026 17:28:43 +0200 Subject: [PATCH 12/16] fix error message StateExporter -> StatsExporter Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- libdd-trace-stats/src/stats_exporter.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/libdd-trace-stats/src/stats_exporter.rs b/libdd-trace-stats/src/stats_exporter.rs index 7698bf92af..400a874bfa 100644 --- a/libdd-trace-stats/src/stats_exporter.rs +++ b/libdd-trace-stats/src/stats_exporter.rs @@ -231,7 +231,7 @@ impl sent_stats = true; } Err(err) => { - error!(?err, "Error with the StateExporter when sending stats"); + error!(?err, "Error with the StatsExporter when sending stats"); anyhow::bail!("Failed to send stats: {err}"); } } From 9d91a3c34a01b31df27fd0f5dfc87302ab4dbbc5 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Tue, 7 Jul 2026 18:06:42 +0200 Subject: [PATCH 13/16] fix nit comment --- libdd-trace-stats/src/span_concentrator/aggregation.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/libdd-trace-stats/src/span_concentrator/aggregation.rs b/libdd-trace-stats/src/span_concentrator/aggregation.rs index 12be89b535..d42a1d9294 100644 --- a/libdd-trace-stats/src/span_concentrator/aggregation.rs +++ b/libdd-trace-stats/src/span_concentrator/aggregation.rs @@ -431,9 +431,9 @@ pub(super) struct StatsBucket { max_entries: usize, /// Number of spans collapsed into the overflow bucket due to cardinality limiting. collapsed_count: u64, - #[cfg(feature = "stats-obfuscation")] /// Indicates if stats obfuscated in this bucket. This is set once at creation and stays /// constant per bucket + #[cfg(feature = "stats-obfuscation")] pub(super) obfuscated: bool, } From 2342ce154c7712dbd23fe887e820a7f6a9bab7ca Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Wed, 8 Jul 2026 10:43:46 +0200 Subject: [PATCH 14/16] refacto: cleaner flush result struct, simplify span obfuscation path --- datadog-ipc/src/shm_stats.rs | 13 +- .../src/span_concentrator/mod.rs | 135 +++++++++-------- .../src/span_concentrator/tests.rs | 64 ++++---- libdd-trace-stats/src/stats_exporter.rs | 141 +++++++++--------- 4 files changed, 189 insertions(+), 164 deletions(-) diff --git a/datadog-ipc/src/shm_stats.rs b/datadog-ipc/src/shm_stats.rs index bc5351e142..87b55981da 100644 --- a/datadog-ipc/src/shm_stats.rs +++ b/datadog-ipc/src/shm_stats.rs @@ -56,7 +56,9 @@ use zwohash::ZwoHasher; use libdd_ddsketch::DDSketch; use libdd_trace_protobuf::pb; -use libdd_trace_stats::span_concentrator::{FixedAggregationKey, FlushableConcentrator}; +use libdd_trace_stats::span_concentrator::{ + FixedAggregationKey, FlushResult, FlushableConcentrator, +}; use crate::platform::{FileBackedHandle, MappedMem, NamedShmHandle}; @@ -820,8 +822,13 @@ impl ShmSpanConcentrator { } impl FlushableConcentrator for ShmSpanConcentrator { - fn flush_buckets(&mut self, force: bool) -> (Vec, u64, bool) { - (self.drain_buckets(force), 0, false) + fn flush_buckets(&mut self, force: bool) -> FlushResult { + // The SHM concentrator does not perform client-side obfuscation. + FlushResult { + obfuscated_buckets: vec![], + unobfuscated_buckets: self.drain_buckets(force), + collapsed_spans: 0, + } } } diff --git a/libdd-trace-stats/src/span_concentrator/mod.rs b/libdd-trace-stats/src/span_concentrator/mod.rs index 6898101f7d..0f233daca2 100644 --- a/libdd-trace-stats/src/span_concentrator/mod.rs +++ b/libdd-trace-stats/src/span_concentrator/mod.rs @@ -16,26 +16,42 @@ pub use aggregation::{FixedAggregationKey, OtlpExactCell, OtlpExactGroup, OtlpSt pub mod stat_span; pub use stat_span::StatSpan; +/// Result of flushing a concentrator. +/// +/// Obfuscated and un-obfuscated buckets are kept separate because they must be sent in distinct +/// stats payloads: only the obfuscated payload carries the `datadog-obfuscation-version` header. +pub struct FlushResult { + /// Buckets whose resource names were obfuscated client-side. + pub obfuscated_buckets: Vec, + /// Buckets whose resource names were left as-is. + pub unobfuscated_buckets: Vec, + /// Total number of spans that were collapsed into the overflow sentinel bucket due to + /// cardinality limiting across all flushed time buckets. + pub collapsed_spans: u64, +} + +impl FlushResult { + /// All flushed buckets regardless of obfuscation. + #[cfg(test)] + pub fn all_buckets(self) -> Vec { + let mut buckets = self.obfuscated_buckets; + buckets.extend(self.unobfuscated_buckets); + buckets + } +} + /// Concentrators that can provide raw time buckets for export implement this trait. /// /// `StatsExporter` is generic over `C: FlushableConcentrator` so it can work with /// both the in-process [`SpanConcentrator`] and the SHM-backed `ShmSpanConcentrator`. pub trait FlushableConcentrator { - /// Flush time buckets and return them together some metadata. If `force` is true, flush - /// all buckets. - /// - /// Returns a triplet `(buckets, collapsed_spans, buckets_obfuscated)` - /// Where - /// - `buckets` are the encoded stats bucket - /// - `collapsed_spans` is the number of spans that were collapsed into the overflow sentinel - /// bucket due to cardinality limiting - /// - `buckets_obfuscated` indicates whether the returned buckets have been obfuscated - /// client-side - fn flush_buckets(&mut self, force: bool) -> (Vec, u64, bool); + /// Flush time buckets and return them together with flush metadata. If `force` is true, flush + /// all buckets. See [`FlushResult`] for the returned data. + fn flush_buckets(&mut self, force: bool) -> FlushResult; } impl FlushableConcentrator for SpanConcentrator { - fn flush_buckets(&mut self, force: bool) -> (Vec, u64, bool) { + fn flush_buckets(&mut self, force: bool) -> FlushResult { self.flush(SystemTime::now(), force) } } @@ -208,11 +224,11 @@ impl SpanConcentrator { ) }); #[cfg(feature = "stats-obfuscation")] - let obfuscated_resource = Self::compute_obfuscated_span( - target_bucket.obfuscated, - self.obfuscation_config.load().sql_obfuscation_mode, - span, - ); + let obfuscated_resource = if target_bucket.obfuscated { + Self::compute_obfuscated_span(self.obfuscation_config.load().sql_obfuscation_mode, span) + } else { + None + }; #[cfg(not(feature = "stats-obfuscation"))] let obfuscated_resource: Option = None; let agg_key = match obfuscated_resource.as_deref() { @@ -233,66 +249,63 @@ impl SpanConcentrator { #[cfg(feature = "stats-obfuscation")] fn compute_obfuscated_span<'a>( - obfuscate: bool, sql_obfuscation_mode: libdd_trace_obfuscation::sql::SqlObfuscationMode, - #[allow(unused)] span: &'a impl StatSpan<'a>, + span: &'a impl StatSpan<'a>, ) -> Option { - if obfuscate { - let dbms_hint: Option<&str> = span.get_meta("db.type"); - return libdd_trace_obfuscation::obfuscate::obfuscate_resource_for_stats( - span.r#type(), - span.resource(), - dbms_hint, - sql_obfuscation_mode, - ); - } - None + let dbms_hint: Option<&str> = span.get_meta("db.type"); + libdd_trace_obfuscation::obfuscate::obfuscate_resource_for_stats( + span.r#type(), + span.resource(), + dbms_hint, + sql_obfuscation_mode, + ) } /// Flush all stats bucket except for the `buffer_len` most recent. If `force` is true, flush /// all buckets. /// - /// Returns a triplet `(buckets, collapsed_spans, buckets_obfuscated)` - /// Where - /// - `buckets` are the encoded stats bucket - /// - `collapsed_spans` is the number of spans that were collapsed into the overflow sentinel - /// bucket due to cardinality limiting - /// - `buckets_obfuscated` indicates whether the returned buckets have been obfuscated - /// client-side - pub fn flush( - &mut self, - now: SystemTime, - force: bool, - ) -> (Vec, u64, bool) { - self.drain_due_buckets(now, force, StatsBucket::flush) + /// Obfuscated and un-obfuscated buckets are returned separately, see [`FlushResult`]. + pub fn flush(&mut self, now: SystemTime, force: bool) -> FlushResult { + let (buckets, collapsed_spans) = self.drain_due_buckets(now, force, StatsBucket::flush); + let mut obfuscated_buckets = Vec::new(); + let mut unobfuscated_buckets = Vec::new(); + for (obfuscated, bucket) in buckets { + if obfuscated { + obfuscated_buckets.push(bucket); + } else { + unobfuscated_buckets.push(bucket); + } + } + FlushResult { + obfuscated_buckets, + unobfuscated_buckets, + collapsed_spans, + } } /// Like [`Self::flush`], but also emits exact per-cell scalars alongside each bucket for the /// OTLP trace-metrics path. The protobuf bucket inside each [`OtlpStatsBucket`] is identical /// to what [`Self::flush`] would produce, so the /v0.6/stats agent path is unaffected. pub fn flush_with_otlp_exact(&mut self, now: SystemTime, force: bool) -> Vec { - let (buckets, _, _) = - self.drain_due_buckets(now, force, StatsBucket::flush_with_otlp_exact); - buckets + let (buckets, _) = self.drain_due_buckets(now, force, StatsBucket::flush_with_otlp_exact); + buckets.into_iter().map(|(_, bucket)| bucket).collect() } + /// Drain the buckets that are due for flushing, encoding each with `encode`. + /// + /// Returns a tuple `(buckets, collapsed_spans)` where each encoded bucket is paired with a + /// boolean indicating whether it was obfuscated client-side (always `false` when the + /// `stats-obfuscation` feature is disabled), and `collapsed_spans` is the total number of + /// spans collapsed into the overflow sentinel bucket due to cardinality limiting. fn drain_due_buckets( &mut self, now: SystemTime, force: bool, encode: impl Fn(StatsBucket, u64) -> T, - ) -> (Vec, u64, bool) { + ) -> (Vec<(bool, T)>, u64) { // TODO: Wait for HashMap::extract_if to be stabilized to avoid a full drain let now_timestamp = system_time_to_unix_duration(now).as_nanos() as u64; - let mut buckets: Vec<(u64, StatsBucket)> = self.buckets.drain().collect(); - buckets.sort_unstable_by_key(|(timestamp, _)| *timestamp); - #[cfg(not(feature = "stats-obfuscation"))] - let first_bucket_is_obfuscated = false; - #[cfg(feature = "stats-obfuscation")] - let first_bucket_is_obfuscated = buckets - .first() - .map(|(_, bucket)| bucket.obfuscated) - .unwrap_or(false); + let buckets: Vec<(u64, StatsBucket)> = self.buckets.drain().collect(); self.oldest_timestamp = if force { align_timestamp(now_timestamp, self.bucket_size) } else { @@ -312,22 +325,22 @@ impl SpanConcentrator { // shutdown. let keep = !force && timestamp > (now_timestamp - self.buffer_len as u64 * self.bucket_size); - // Even when forcing to flush, we refuse to mix obfuscated buckets from - // un-obfuscated buckets. This means you need to flush twice to flush it all - #[cfg(feature = "stats-obfuscation")] - let keep = keep || bucket.obfuscated != first_bucket_is_obfuscated; if keep { self.buckets.insert(timestamp, bucket); return None; } total_collapsed += bucket.collapsed_count(); - Some(encode(bucket, self.bucket_size)) + #[cfg(feature = "stats-obfuscation")] + let obfuscated = bucket.obfuscated; + #[cfg(not(feature = "stats-obfuscation"))] + let obfuscated = false; + Some((obfuscated, encode(bucket, self.bucket_size))) }) .collect(); if total_collapsed > 0 { debug!(max_entries_per_bucket = self.max_entries_per_bucket, total_collapsed, "Client-side stats values have been collapsed to 'tracer_blocked_value'. This is due to the cardinality exceeding DD_TRACE_STATS_CARDINALITY_LIMIT"); } - (buckets_pb, total_collapsed, first_bucket_is_obfuscated) + (buckets_pb, total_collapsed) } } diff --git a/libdd-trace-stats/src/span_concentrator/tests.rs b/libdd-trace-stats/src/span_concentrator/tests.rs index 20021c862c..81d7e94778 100644 --- a/libdd-trace-stats/src/span_concentrator/tests.rs +++ b/libdd-trace-stats/src/span_concentrator/tests.rs @@ -126,12 +126,12 @@ fn test_concentrator_oldest_timestamp_cold() { // Assert we didn't insert spans in older buckets for _ in 0..concentrator.buffer_len { - let (stats, _, _) = concentrator.flush(flushtime, false); + let stats = concentrator.flush(flushtime, false).all_buckets(); assert_eq!(stats.len(), 0, "We should get 0 time buckets"); flushtime += Duration::from_nanos(concentrator.bucket_size); } - let (stats, _, _) = concentrator.flush(flushtime, false); + let stats = concentrator.flush(flushtime, false).all_buckets(); assert_eq!(stats.len(), 1, "We should get exactly one time bucket"); @@ -188,12 +188,12 @@ fn test_concentrator_oldest_timestamp_hot() { } for _ in 0..(concentrator.buffer_len - 1) { - let (stats, _, _) = concentrator.flush(flushtime, false); + let stats = concentrator.flush(flushtime, false).all_buckets(); assert!(stats.is_empty(), "We should get 0 time buckets"); flushtime += Duration::from_nanos(concentrator.bucket_size); } - let (stats, _, _) = concentrator.flush(flushtime, false); + let stats = concentrator.flush(flushtime, false).all_buckets(); assert_eq!(stats.len(), 1, "We should get exactly one time bucket"); // First oldest bucket aggregates, it should have it all except the @@ -213,7 +213,7 @@ fn test_concentrator_oldest_timestamp_hot() { assert_counts_equal(expected, stats.first().unwrap().stats.clone()); flushtime += Duration::from_nanos(concentrator.bucket_size); - let (stats, _, _) = concentrator.flush(flushtime, false); + let stats = concentrator.flush(flushtime, false).all_buckets(); assert_eq!(stats.len(), 1, "We should get exactly one time bucket"); // Stats of the last four spans. @@ -278,7 +278,7 @@ fn test_concentrator_stats_totals() { let mut flushtime = now; for _ in 0..=concentrator.buffer_len { - let (stats, _, _) = concentrator.flush(flushtime, false); + let stats = concentrator.flush(flushtime, false).all_buckets(); if stats.is_empty() { continue; } @@ -528,7 +528,7 @@ fn test_concentrator_stats_counts() { let mut flushtime = now; for _ in 0..=concentrator.buffer_len + 2 { - let (stats, _, _) = concentrator.flush(flushtime, false); + let stats = concentrator.flush(flushtime, false).all_buckets(); let expected_flushed_timestamps = align_timestamp( system_time_to_unix_duration(flushtime).as_nanos() as u64, concentrator.bucket_size, @@ -553,7 +553,7 @@ fn test_concentrator_stats_counts() { stats.first().unwrap().stats.clone(), ); - let (stats, _, _) = concentrator.flush(flushtime, false); + let stats = concentrator.flush(flushtime, false).all_buckets(); assert_eq!( stats.len(), 0, @@ -658,10 +658,12 @@ fn test_span_should_be_included_in_stats() { }, ]; - let (stats, _, _) = concentrator.flush( - now + Duration::from_nanos(concentrator.bucket_size * concentrator.buffer_len as u64), - false, - ); + let stats = concentrator + .flush( + now + Duration::from_nanos(concentrator.bucket_size * concentrator.buffer_len as u64), + false, + ) + .all_buckets(); assert_counts_equal( expected, stats @@ -696,10 +698,12 @@ fn test_ignore_partial_spans() { concentrator.add_span(span); } - let (stats, _, _) = concentrator.flush( - now + Duration::from_nanos(concentrator.bucket_size * concentrator.buffer_len as u64), - false, - ); + let stats = concentrator + .flush( + now + Duration::from_nanos(concentrator.bucket_size * concentrator.buffer_len as u64), + false, + ) + .all_buckets(); assert_eq!(0, stats.len()); } @@ -727,10 +731,10 @@ fn test_force_flush() { let flushtime = now - Duration::from_secs(3600); // Bucket should not be flushed without force flush - let (stats, _, _) = concentrator.flush(flushtime, false); + let stats = concentrator.flush(flushtime, false).all_buckets(); assert_eq!(0, stats.len()); - let (stats, _, _) = concentrator.flush(flushtime, true); + let stats = concentrator.flush(flushtime, true).all_buckets(); assert_eq!(1, stats.len()); } @@ -900,7 +904,9 @@ fn test_peer_tags_aggregation() { }, ]; - let (stats_with_peer_tags, _, _) = concentrator_with_peer_tags.flush(flushtime, false); + let stats_with_peer_tags = concentrator_with_peer_tags + .flush(flushtime, false) + .all_buckets(); assert_counts_equal( expected_with_peer_tags, stats_with_peer_tags @@ -910,7 +916,9 @@ fn test_peer_tags_aggregation() { .clone(), ); - let (stats_without_peer_tags, _, _) = concentrator_without_peer_tags.flush(flushtime, false); + let stats_without_peer_tags = concentrator_without_peer_tags + .flush(flushtime, false) + .all_buckets(); assert_counts_equal( expected_without_peer_tags, stats_without_peer_tags @@ -1023,7 +1031,9 @@ fn test_peer_tags_quantization_aggregation() { ..Default::default() }]; - let (stats_with_peer_tags, _, _) = concentrator_with_peer_tags.flush(flushtime, false); + let stats_with_peer_tags = concentrator_with_peer_tags + .flush(flushtime, false) + .all_buckets(); assert_counts_equal( expected_with_peer_tags, stats_with_peer_tags @@ -1193,7 +1203,7 @@ fn test_base_service_peer_tag() { }, ]; - let (stats, _, _) = concentrator.flush(flushtime, false); + let stats = concentrator.flush(flushtime, false).all_buckets(); assert_counts_equal( expected, stats @@ -1497,7 +1507,7 @@ fn test_pb_span() { // Flush and get stats let flushtime = now + Duration::from_nanos(concentrator.bucket_size * concentrator.buffer_len as u64); - let (stats, _, _) = concentrator.flush(flushtime, false); + let stats = concentrator.flush(flushtime, false).all_buckets(); assert_eq!(stats.len(), 1, "Should get exactly one time bucket"); let bucket = &stats[0]; @@ -1676,7 +1686,7 @@ fn test_cardinality_limit_collapse() { concentrator.add_span(&span); } - let (buckets, _, _) = concentrator.flush(SystemTime::now(), true); + let buckets = concentrator.flush(SystemTime::now(), true).all_buckets(); assert!(!buckets.is_empty(), "should get at least one time bucket"); let stats = &buckets[0].stats; @@ -1734,7 +1744,7 @@ fn test_overflow_bucket_counts() { concentrator.add_span(&span); } - let (buckets, _, _) = concentrator.flush(SystemTime::now(), true); + let buckets = concentrator.flush(SystemTime::now(), true).all_buckets(); assert!(!buckets.is_empty()); let stats = &buckets[0].stats; @@ -1783,7 +1793,7 @@ fn test_no_collapse_within_limit() { concentrator.add_span(&span); } - let (buckets, _, _) = concentrator.flush(SystemTime::now(), true); + let buckets = concentrator.flush(SystemTime::now(), true).all_buckets(); assert!(!buckets.is_empty()); let stats = &buckets[0].stats; @@ -1835,7 +1845,7 @@ fn test_overflow_bucket_key_sentinel_values() { concentrator.add_span(&first); concentrator.add_span(&second); - let (buckets, _, _) = concentrator.flush(SystemTime::now(), true); + let buckets = concentrator.flush(SystemTime::now(), true).all_buckets(); assert!(!buckets.is_empty()); let stats = &buckets[0].stats; diff --git a/libdd-trace-stats/src/stats_exporter.rs b/libdd-trace-stats/src/stats_exporter.rs index 400a874bfa..47c6bfe935 100644 --- a/libdd-trace-stats/src/stats_exporter.rs +++ b/libdd-trace-stats/src/stats_exporter.rs @@ -12,7 +12,7 @@ use std::{ use crate::span_concentrator::{FlushableConcentrator, SpanConcentrator}; use async_trait::async_trait; use libdd_capabilities::{HttpClientCapability, MaybeSend, SleepCapability}; -use libdd_common::{tag, Endpoint}; +use libdd_common::Endpoint; use libdd_shared_runtime::Worker; use libdd_trace_protobuf::pb; use libdd_trace_utils::send_with_retry::{send_with_retry, RetryStrategy}; @@ -133,7 +133,7 @@ impl let telemetry = telemetry.map(|handle| { let key = handle.register_metric_context( COLLAPSED_SPANS_TELEMETRY_METRIC.to_string(), - vec![tag!("collapsed_spans", "whole_key")], + vec![libdd_common::tag!("collapsed_spans", "whole_key")], libdd_telemetry::data::metrics::MetricType::Count, true, libdd_telemetry::data::metrics::MetricNamespace::Tracers, @@ -173,88 +173,83 @@ impl /// case stats cannot be flushed since the concentrator might be corrupted. /// Returns `Ok(true)` if stats were sent, `Ok(false)` if the concentrator had nothing to send. pub async fn send(&self, force_flush: bool) -> anyhow::Result { - // We flush twice with `force_flush` so that if we have a mix of obfuscated and unobfuscated - // buckets, both get flushed in separate payloads - let flush_count = if force_flush { 2 } else { 1 }; - let mut sent_stats = false; - for _ in 0..flush_count { - let (payload, collapsed_spans, buckets_obfuscated) = self.flush(force_flush); - - if collapsed_spans > 0 { - #[cfg(feature = "telemetry")] - if let Some((handle, key)) = &self.telemetry { - let _ = handle.add_point(collapsed_spans as f64, key, vec![]); - } - #[cfg(feature = "dogstatsd")] - if let Some(client) = &self.dogstatsd { - client.send(vec![libdd_dogstatsd_client::DogStatsDAction::Count( - COLLAPSED_SPANS_HEALTH_METRIC, - collapsed_spans as i64, - [tag!("collapsed_spans", "whole_key")].iter(), - )]); - } - } + let flush = { + #[allow(clippy::unwrap_used)] + let mut concentrator = self.concentrator.lock().unwrap(); + concentrator.flush_buckets(force_flush) + }; - if payload.stats.is_empty() { - continue; + if flush.collapsed_spans > 0 { + #[cfg(feature = "telemetry")] + if let Some((handle, key)) = &self.telemetry { + let _ = handle.add_point(flush.collapsed_spans as f64, key, vec![]); } - let body = rmp_serde::encode::to_vec_named(&payload)?; - - let mut headers: http::HeaderMap = TracerHeaderTags::from(&self.meta).into(); - - headers.insert( - http::header::CONTENT_TYPE, - libdd_common::header::APPLICATION_MSGPACK, - ); - - #[cfg(feature = "stats-obfuscation")] - if buckets_obfuscated { - headers.insert( - http::HeaderName::from_static("datadog-obfuscation-version"), - http::HeaderValue::from_static(self.supported_obfuscation_version), - ); + #[cfg(feature = "dogstatsd")] + if let Some(client) = &self.dogstatsd { + client.send(vec![libdd_dogstatsd_client::DogStatsDAction::Count( + COLLAPSED_SPANS_HEALTH_METRIC, + flush.collapsed_spans as i64, + [libdd_common::tag!("collapsed_spans", "whole_key")].iter(), + )]); } - #[cfg(not(feature = "stats-obfuscation"))] - let _ = buckets_obfuscated; - - let result = send_with_retry( - &self.capabilities, - &self.endpoint, - body, - &headers, - &RetryStrategy::default(), - ) - .await; + } - match result { - Ok(_) => { - sent_stats = true; - } - Err(err) => { - error!(?err, "Error with the StatsExporter when sending stats"); - anyhow::bail!("Failed to send stats: {err}"); - } - } + // Obfuscated and un-obfuscated buckets must be sent in separate payloads because only + // the obfuscated one carries the `datadog-obfuscation-version` header. + let mut sent_stats = false; + if !flush.obfuscated_buckets.is_empty() { + self.send_payload(flush.obfuscated_buckets, true).await?; + sent_stats = true; + } + if !flush.unobfuscated_buckets.is_empty() { + self.send_payload(flush.unobfuscated_buckets, false).await?; + sent_stats = true; } Ok(sent_stats) } - /// Flush stats from the concentrator into a payload + /// Encode the given buckets into a stats payload and send it to the agent. /// - /// # Arguments - /// - `force_flush` if true, triggers a force flush on the concentrator causing all buckets to - /// be flushed regardless of their age. - /// - /// # Panic - /// Will panic if another thread panicked while holding the concentrator lock in which - /// case stats cannot be flushed since the concentrator might be corrupted. - fn flush(&self, force_flush: bool) -> (pb::ClientStatsPayload, u64, bool) { + /// `obfuscated` indicates whether the buckets were obfuscated client-side, in which case the + /// `datadog-obfuscation-version` header is added. + async fn send_payload( + &self, + buckets: Vec, + #[cfg_attr(not(feature = "stats-obfuscation"), allow(unused))] obfuscated: bool, + ) -> anyhow::Result<()> { let sequence = self.sequence_id.fetch_add(1, Ordering::Relaxed); - #[allow(clippy::unwrap_used)] - let (buckets, collapsed_spans, buckets_obfuscated) = - self.concentrator.lock().unwrap().flush_buckets(force_flush); let payload = encode_stats_payload(&self.meta, sequence, buckets); - (payload, collapsed_spans, buckets_obfuscated) + let body = rmp_serde::encode::to_vec_named(&payload)?; + + let mut headers: http::HeaderMap = TracerHeaderTags::from(&self.meta).into(); + headers.insert( + http::header::CONTENT_TYPE, + libdd_common::header::APPLICATION_MSGPACK, + ); + #[cfg(feature = "stats-obfuscation")] + if obfuscated { + headers.insert( + http::HeaderName::from_static("datadog-obfuscation-version"), + http::HeaderValue::from_static(self.supported_obfuscation_version), + ); + } + + let result = send_with_retry( + &self.capabilities, + &self.endpoint, + body, + &headers, + &RetryStrategy::default(), + ) + .await; + + match result { + Ok(_) => Ok(()), + Err(err) => { + error!(?err, "Error with the StatsExporter when sending stats"); + anyhow::bail!("Failed to send stats: {err}"); + } + } } } From 2d4eebc99b132934533f3b3f519cfd8c2045ceab Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Thu, 9 Jul 2026 16:50:11 +0200 Subject: [PATCH 15/16] feat: send obfuscated/unobfuscated stats concurrently --- libdd-trace-stats/src/stats_exporter.rs | 24 ++++++++++++++---------- 1 file changed, 14 insertions(+), 10 deletions(-) diff --git a/libdd-trace-stats/src/stats_exporter.rs b/libdd-trace-stats/src/stats_exporter.rs index 47c6bfe935..d4855ee192 100644 --- a/libdd-trace-stats/src/stats_exporter.rs +++ b/libdd-trace-stats/src/stats_exporter.rs @@ -194,18 +194,22 @@ impl } } - // Obfuscated and un-obfuscated buckets must be sent in separate payloads because only - // the obfuscated one carries the `datadog-obfuscation-version` header. - let mut sent_stats = false; - if !flush.obfuscated_buckets.is_empty() { + if !flush.obfuscated_buckets.is_empty() && !flush.unobfuscated_buckets.is_empty() { + let (res_obfuscated, res_unobfuscated) = tokio::join!( + self.send_payload(flush.obfuscated_buckets, true), + self.send_payload(flush.unobfuscated_buckets, false) + ); + res_obfuscated?; + res_unobfuscated?; + return Ok(true); + } else if !flush.obfuscated_buckets.is_empty() { self.send_payload(flush.obfuscated_buckets, true).await?; - sent_stats = true; - } - if !flush.unobfuscated_buckets.is_empty() { - self.send_payload(flush.unobfuscated_buckets, false).await?; - sent_stats = true; + return Ok(true); + } else if !flush.unobfuscated_buckets.is_empty() { + self.send_payload(flush.unobfuscated_buckets, true).await?; + return Ok(true); } - Ok(sent_stats) + Ok(false) } /// Encode the given buckets into a stats payload and send it to the agent. From 35d265b29b28e3baad3ce3f05747b9006ad47aa9 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Fri, 10 Jul 2026 14:03:12 +0200 Subject: [PATCH 16/16] fix: FuturesUnordered over tokio::join! --- Cargo.lock | 38 ++++++++++++------------- libdd-trace-stats/Cargo.toml | 1 + libdd-trace-stats/src/stats_exporter.rs | 35 +++++++++++++---------- 3 files changed, 40 insertions(+), 34 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 2917cbba83..3fd5276f74 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1954,9 +1954,9 @@ checksum = "673464e1e314dd67a0fd9544abc99e8eb28d0c7e3b69b033bcff9b2d00b87333" [[package]] name = "futures" -version = "0.3.31" +version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "65bc07b1a8bc7c85c5f2e110c476c7389b4554ba72af57d8445ea63a576b0876" +checksum = "8b147ee9d1f6d097cef9ce628cd2ee62288d963e16fb287bd9286455b241382d" dependencies = [ "futures-channel", "futures-core", @@ -1969,9 +1969,9 @@ dependencies = [ [[package]] name = "futures-channel" -version = "0.3.31" +version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2dff15bf788c671c1934e366d07e30c1814a8ef514e1af724a602e8a2fbe1b10" +checksum = "07bbe89c50d7a535e539b8c17bc0b49bdb77747034daa8087407d655f3f7cc1d" dependencies = [ "futures-core", "futures-sink", @@ -1979,15 +1979,15 @@ dependencies = [ [[package]] name = "futures-core" -version = "0.3.31" +version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "05f29059c0c2090612e8d742178b0580d2dc940c837851ad723096f87af6663e" +checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d" [[package]] name = "futures-executor" -version = "0.3.31" +version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1e28d1d997f585e54aebc3f97d39e72338912123a67330d723fdbb564d646c9f" +checksum = "baf29c38818342a3b26b5b923639e7b1f4a61fc5e76102d4b1981c6dc7a7579d" dependencies = [ "futures-core", "futures-task", @@ -1996,9 +1996,9 @@ dependencies = [ [[package]] name = "futures-io" -version = "0.3.31" +version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9e5c1b78ca4aae1ac06c48a526a655760685149f0d465d21f37abfe57ce075c6" +checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718" [[package]] name = "futures-lite" @@ -2015,9 +2015,9 @@ dependencies = [ [[package]] name = "futures-macro" -version = "0.3.31" +version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "162ee34ebcb7c64a8abebc059ce0fee27c2262618d7b60ed8faf72fef13c3650" +checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b" dependencies = [ "proc-macro2", "quote", @@ -2026,15 +2026,15 @@ dependencies = [ [[package]] name = "futures-sink" -version = "0.3.31" +version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e575fab7d1e0dcb8d0c7bcf9a63ee213816ab51902e6d244a95819acacf1d4f7" +checksum = "c39754e157331b013978ec91992bde1ac089843443c49cbc7f46150b0fad0893" [[package]] name = "futures-task" -version = "0.3.31" +version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f90f7dce0722e95104fcb095585910c0977252f286e354b5e3bd38902cd99988" +checksum = "037711b3d59c33004d3856fbdc83b99d4ff37a24768fa1be9ce3538a1cde4393" [[package]] name = "futures-timer" @@ -2044,9 +2044,9 @@ checksum = "f288b0a4f20f9a56b5d1da57e2227c661b7b16168e2f72365f57b63326e29b24" [[package]] name = "futures-util" -version = "0.3.31" +version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9fa08315bb612088cc391249efdc3bc77536f16c91f6cf495e6fbe85b20a4a81" +checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6" dependencies = [ "futures-channel", "futures-core", @@ -2056,7 +2056,6 @@ dependencies = [ "futures-task", "memchr", "pin-project-lite", - "pin-utils", "slab", ] @@ -3495,6 +3494,7 @@ dependencies = [ "arc-swap", "async-trait", "criterion", + "futures", "hashbrown 0.15.1", "http", "httpmock", diff --git a/libdd-trace-stats/Cargo.toml b/libdd-trace-stats/Cargo.toml index fd2cf45b78..bb87e1ee0f 100644 --- a/libdd-trace-stats/Cargo.toml +++ b/libdd-trace-stats/Cargo.toml @@ -29,6 +29,7 @@ tokio = { version = "1.23", features = ["macros", "time"], default-features = fa tokio-util = "0.7.11" tracing = { version = "0.1", default-features = false } async-trait = "0.1.85" +futures = { version = "0.3.32", default-features = false, features = ["alloc"] } [target.'cfg(not(target_arch = "wasm32"))'.dependencies] libdd-capabilities-impl = { version = "3.0.0", path = "../libdd-capabilities-impl", default-features = false } diff --git a/libdd-trace-stats/src/stats_exporter.rs b/libdd-trace-stats/src/stats_exporter.rs index d4855ee192..24aa5d4c1b 100644 --- a/libdd-trace-stats/src/stats_exporter.rs +++ b/libdd-trace-stats/src/stats_exporter.rs @@ -11,6 +11,8 @@ use std::{ use crate::span_concentrator::{FlushableConcentrator, SpanConcentrator}; use async_trait::async_trait; +use futures::stream::FuturesUnordered; +use futures::StreamExt as _; use libdd_capabilities::{HttpClientCapability, MaybeSend, SleepCapability}; use libdd_common::Endpoint; use libdd_shared_runtime::Worker; @@ -194,22 +196,25 @@ impl } } - if !flush.obfuscated_buckets.is_empty() && !flush.unobfuscated_buckets.is_empty() { - let (res_obfuscated, res_unobfuscated) = tokio::join!( - self.send_payload(flush.obfuscated_buckets, true), - self.send_payload(flush.unobfuscated_buckets, false) - ); - res_obfuscated?; - res_unobfuscated?; - return Ok(true); - } else if !flush.obfuscated_buckets.is_empty() { - self.send_payload(flush.obfuscated_buckets, true).await?; - return Ok(true); - } else if !flush.unobfuscated_buckets.is_empty() { - self.send_payload(flush.unobfuscated_buckets, true).await?; - return Ok(true); + let futures = FuturesUnordered::new(); + + if !flush.obfuscated_buckets.is_empty() { + futures.push(self.send_payload(flush.obfuscated_buckets, true)); + } + + if !flush.unobfuscated_buckets.is_empty() { + futures.push(self.send_payload(flush.unobfuscated_buckets, false)); } - Ok(false) + + let sent_stats = !futures.is_empty(); + + futures + .collect::>>() + .await + .into_iter() + .collect::>()?; + + Ok(sent_stats) } /// Encode the given buckets into a stats payload and send it to the agent.