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/30] 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/30] 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/30] 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/30] 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/30] 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/30] 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/30] 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/30] 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/30] 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/30] 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/30] 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/30] 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/30] 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/30] 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/30] 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/30] 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. From e08200565a945cd3256341622380bbcf4c00f97d Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Wed, 8 Jul 2026 14:00:57 +0200 Subject: [PATCH 17/30] feat!(stats): per-key cardinality limits --- .../src/trace_exporter/stats.rs | 6 +- .../src/span_concentrator/aggregation.rs | 78 +++++++++-- .../src/span_concentrator/mod.rs | 44 ++++++- .../src/span_concentrator/tests.rs | 121 ++++++++++++++++-- libdd-trace-stats/src/stats_exporter.rs | 9 +- 5 files changed, 228 insertions(+), 30 deletions(-) diff --git a/libdd-data-pipeline/src/trace_exporter/stats.rs b/libdd-data-pipeline/src/trace_exporter/stats.rs index eaff519d3d..33e3ac9362 100644 --- a/libdd-data-pipeline/src/trace_exporter/stats.rs +++ b/libdd-data-pipeline/src/trace_exporter/stats.rs @@ -18,7 +18,7 @@ use libdd_capabilities::{HttpClientCapability, MaybeSend, SleepCapability}; use libdd_common::Endpoint; use libdd_common::MutexExt; use libdd_shared_runtime::{SharedRuntime, WorkerHandle}; -use libdd_trace_stats::span_concentrator::SpanConcentrator; +use libdd_trace_stats::span_concentrator::{CardinalityLimitConfig, SpanConcentrator}; #[cfg(feature = "stats-obfuscation")] use libdd_trace_stats::span_concentrator::{ SharedStatsComputationObfuscationConfig, StatsComputationObfuscationConfig, @@ -49,7 +49,7 @@ pub(crate) struct StatsContext<'a, R: SharedRuntime> { pub metadata: &'a TracerMetadata, pub endpoint_url: &'a http::Uri, pub shared_runtime: &'a R, - pub stats_cardinality_limit: Option, + pub stats_cardinality_limits: Option, /// Optional DogStatsD client forwarded to the [`StatsExporter`]. pub dogstatsd: Option>, /// Optional telemetry handle forwarded to the [`StatsExporter`]. @@ -142,7 +142,7 @@ pub(crate) fn start_stats_computation< std::time::SystemTime::now(), span_kinds, peer_tags, - ctx.stats_cardinality_limit, + ctx.stats_cardinality_limits, #[cfg(feature = "stats-obfuscation")] Some(client_side_stats.obfuscation_config.clone()), ))); diff --git a/libdd-trace-stats/src/span_concentrator/aggregation.rs b/libdd-trace-stats/src/span_concentrator/aggregation.rs index d42a1d9294..dc6a4c2e59 100644 --- a/libdd-trace-stats/src/span_concentrator/aggregation.rs +++ b/libdd-trace-stats/src/span_concentrator/aggregation.rs @@ -5,7 +5,7 @@ //! This includes the aggregation key to group spans together and the computation of stats from a //! span. -use hashbrown::HashMap; +use hashbrown::{HashMap, HashSet}; use libdd_trace_obfuscation::ip_address::quantize_peer_ip_addresses; use libdd_trace_protobuf::pb; use libdd_trace_utils::span::SpanText; @@ -13,6 +13,8 @@ use std::borrow::{Borrow, Cow}; use crate::span_concentrator::StatSpan; +use super::CardinalityLimitConfig; + /// Sentinel value used for cardinality limiting. pub const TRACER_BLOCKED_VALUE: &str = "tracer_blocked_value"; @@ -299,7 +301,7 @@ impl OwnedAggregationKey { is_synthetics_request: false, is_trace_root: pb::Trilean::NotSet, }, - peer_tags: vec![], + peer_tags: vec![(TRACER_BLOCKED_VALUE.to_owned(), "".to_owned())], } } } @@ -428,7 +430,13 @@ pub(super) struct StatsBucket { start: u64, /// Maximum number of distinct aggregation keys this bucket will hold before collapsing new /// ones into the overflow sentinel key. - max_entries: usize, + cardinality_limits: CardinalityLimitConfig, + // FIXME: optimize memory by storing hashes only + distinct_resources: HashSet, + distinct_http_endpoint: HashSet, + distinct_peer_tags: HashSet>, + #[allow(unused, reason = "FIXME: implement stats additional tags")] + distinct_additional_tags: HashSet, /// Number of spans collapsed into the overflow bucket due to cardinality limiting. collapsed_count: u64, /// Indicates if stats obfuscated in this bucket. This is set once at creation and stays @@ -440,20 +448,23 @@ pub(super) struct StatsBucket { impl StatsBucket { /// Return a new StatsBucket starting at `start_timestamp`. /// - /// `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. + /// `cardinality_limits` are the values for whole-key and per-field cardinality limits pub(super) fn new( start_timestamp: u64, - max_entries: usize, + cardinality_limits: CardinalityLimitConfig, #[cfg(feature = "stats-obfuscation")] obfuscation_enabled: bool, ) -> Self { Self { data: HashMap::new(), start: start_timestamp, - max_entries, + cardinality_limits, collapsed_count: 0, #[cfg(feature = "stats-obfuscation")] obfuscated: obfuscation_enabled, + distinct_resources: HashSet::new(), + distinct_http_endpoint: HashSet::new(), + distinct_peer_tags: HashSet::new(), + distinct_additional_tags: HashSet::new(), } } @@ -467,12 +478,14 @@ impl StatsBucket { /// instead merged into the overflow sentinel key. pub(super) fn insert( &mut self, - key: BorrowedAggregationKey<'_>, + mut key: BorrowedAggregationKey<'_>, duration: i64, is_error: bool, is_top_level: bool, ) { - if self.data.len() >= self.max_entries && !self.data.contains_key(&key) { + if self.data.len() >= self.cardinality_limits.whole_key_limit + && !self.data.contains_key(&key) + { self.collapsed_count += 1; self.data .entry(OwnedAggregationKey::overflow_key()) @@ -480,12 +493,51 @@ impl StatsBucket { .insert(duration, is_error, is_top_level); return; } + + self.collapse_key_cardinality(&mut key); + self.data .entry_ref(&key) .or_default() .insert(duration, is_error, is_top_level); } + /// Collapse an aggregation key fields following the bucket's `CardinalityLimitConfig` + fn collapse_key_cardinality(&mut self, key: &mut BorrowedAggregationKey<'_>) { + if self.distinct_resources.len() >= self.cardinality_limits.resource_limit + && !self.distinct_resources.contains(key.fixed.resource_name) + { + key.fixed.resource_name = TRACER_BLOCKED_VALUE; + } else { + self.distinct_resources + .insert(key.fixed.resource_name.to_owned()); + } + + if self.distinct_http_endpoint.len() >= self.cardinality_limits.http_endpoint_limit + && !self + .distinct_http_endpoint + .contains(key.fixed.http_endpoint) + { + key.fixed.http_endpoint = TRACER_BLOCKED_VALUE; + } else { + self.distinct_http_endpoint + .insert(key.fixed.http_endpoint.to_owned()); + } + + let owned_peer_tags = key + .peer_tags + .iter() + .map(|(k, v)| ((*k).to_owned(), v.to_string())) + .collect(); + if self.distinct_peer_tags.len() >= self.cardinality_limits.peer_tags_limit + && !self.distinct_peer_tags.contains(&owned_peer_tags) + { + key.peer_tags = vec![(TRACER_BLOCKED_VALUE, Cow::Borrowed(TRACER_BLOCKED_VALUE))]; + } else { + self.distinct_peer_tags.insert(owned_peer_tags); + } + } + /// Consume the bucket and return a ClientStatsBucket containing the bucket stats. /// `bucket_duration` is the size of buckets for the concentrator containing the bucket. pub(super) fn flush(self, bucket_duration: u64) -> pb::ClientStatsBucket { @@ -550,7 +602,13 @@ fn encode_grouped_stats(key: OwnedAggregationKey, group: GroupedStats) -> pb::Cl peer_tags: key .peer_tags .into_iter() - .map(|(k, v)| format!("{k}:{v}")) + .map(|(k, v)| { + if v.is_empty() { + k.to_string() + } else { + format!("{k}:{v}") + } + }) .collect(), is_trace_root: f.is_trace_root.into(), http_method: f.http_method, diff --git a/libdd-trace-stats/src/span_concentrator/mod.rs b/libdd-trace-stats/src/span_concentrator/mod.rs index 0f233daca2..d1db96b738 100644 --- a/libdd-trace-stats/src/span_concentrator/mod.rs +++ b/libdd-trace-stats/src/span_concentrator/mod.rs @@ -100,6 +100,33 @@ pub type SharedStatsComputationObfuscationConfig = /// over this limit (e.g. adding extra overflow buckets) we set a slightly lower limit. pub const DEFAULT_MAX_ENTRIES_PER_BUCKET: usize = 7_000; +/// Config to override the default stats cardinality limit values +#[derive(Debug, Clone, Copy)] +pub struct CardinalityLimitConfig { + /// The whole-key cardinality limit (defaults to 7000) + pub whole_key_limit: usize, + /// The per-field cardinality limit for the Resource field (defaults to 1024) + pub resource_limit: usize, + /// The per-field cardinality limit for the HttpEndpoint field (defaults to 512) + pub http_endpoint_limit: usize, + /// The per-field cardinality limit for the PeerTags field (defaults to 512) + pub peer_tags_limit: usize, + /// The per-field cardinality limit for the AdditionalTags field (defaults to 100) + pub additional_tags_limit: usize, +} + +impl Default for CardinalityLimitConfig { + fn default() -> Self { + Self { + whole_key_limit: DEFAULT_MAX_ENTRIES_PER_BUCKET, + resource_limit: 1024, + http_endpoint_limit: 512, + peer_tags_limit: 512, + additional_tags_limit: 100, + } + } +} + /// SpanConcentrator compute stats on span aggregated by time and span attributes /// /// # Aggregation @@ -130,8 +157,8 @@ pub struct SpanConcentrator { oldest_timestamp: u64, /// bufferLen is the number stats bucket we keep when flushing. buffer_len: usize, - /// Maximum number of distinct aggregation keys per bucket. - max_entries_per_bucket: usize, + /// Config values for whole-key and per-field cardinality limits + cardinality_limits: CardinalityLimitConfig, /// span.kind fields eligible for stats computation span_kinds_stats_computed: Vec, /// keys for supplementary tags that describe peer.service entities @@ -154,7 +181,7 @@ impl SpanConcentrator { now: SystemTime, span_kinds_stats_computed: Vec, peer_tag_keys: Vec, - override_max_entries_per_bucket: Option, + override_cardinality_limits: Option, #[cfg(feature = "stats-obfuscation")] obfuscation_config: Option< SharedStatsComputationObfuscationConfig, >, @@ -167,8 +194,7 @@ impl SpanConcentrator { bucket_size.as_nanos() as u64, ), buffer_len: 2, - max_entries_per_bucket: override_max_entries_per_bucket - .unwrap_or(DEFAULT_MAX_ENTRIES_PER_BUCKET), + cardinality_limits: override_cardinality_limits.unwrap_or_default(), span_kinds_stats_computed, peer_tag_keys, #[cfg(feature = "stats-obfuscation")] @@ -218,7 +244,7 @@ impl SpanConcentrator { let target_bucket = self.buckets.entry(bucket_timestamp).or_insert_with(|| { StatsBucket::new( bucket_timestamp, - self.max_entries_per_bucket, + self.cardinality_limits, #[cfg(feature = "stats-obfuscation")] self.obfuscation_config.load().enabled, ) @@ -338,7 +364,11 @@ impl SpanConcentrator { }) .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"); + debug!( + max_entries_per_bucket = self.cardinality_limits.whole_key_limit, + 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) } diff --git a/libdd-trace-stats/src/span_concentrator/tests.rs b/libdd-trace-stats/src/span_concentrator/tests.rs index 81d7e94778..9001f0b916 100644 --- a/libdd-trace-stats/src/span_concentrator/tests.rs +++ b/libdd-trace-stats/src/span_concentrator/tests.rs @@ -1647,14 +1647,14 @@ fn test_flush_with_otlp_exact_per_cell_scalars() { } /// Build a minimal concentrator with a tiny `max_entries_per_bucket` for cardinality tests. -fn make_cardinality_concentrator(max_entries: usize) -> SpanConcentrator { +fn make_cardinality_concentrator(cardinality_limits: CardinalityLimitConfig) -> SpanConcentrator { let now = SystemTime::now(); SpanConcentrator::new( Duration::from_nanos(BUCKET_SIZE), now, get_span_kinds(), vec![], - Some(max_entries), + Some(cardinality_limits), #[cfg(feature = "stats-obfuscation")] None, ) @@ -1663,10 +1663,13 @@ fn make_cardinality_concentrator(max_entries: usize) -> SpanConcentrator { /// When the limit is 3 and we insert 5 distinct-resource spans, only 3 normal keys plus one /// overflow key must appear in the flushed stats. Total hits must equal 5. #[test] -fn test_cardinality_limit_collapse() { +fn test_whole_key_cardinality_limit_collapse() { let now = SystemTime::now(); let limit: usize = 3; - let mut concentrator = make_cardinality_concentrator(limit); + let mut concentrator = make_cardinality_concentrator(CardinalityLimitConfig { + whole_key_limit: limit, + ..Default::default() + }); // Insert limit + 2 distinct-resource root spans all in the same time bucket. let resources: Vec = (0..limit + 2).map(|i| format!("resource-{i}")).collect(); @@ -1719,12 +1722,108 @@ fn test_cardinality_limit_collapse() { ); } +/// When the `http_endpoint` cardinality limit is 3 and we insert 5 distinct spans differing only on +/// their `http_endpoint`, only 3 normal keys plus one overflow key must appear in the flushed +/// stats. +/// +/// But then inserting spans differing on another field, it should not be collapsed. +/// So total hits must equal 6. +#[test] +fn test_per_key_cardinality_limit_collapse_http_endpoint() { + let now = SystemTime::now(); + let limit: usize = 3; + let mut concentrator = make_cardinality_concentrator(CardinalityLimitConfig { + http_endpoint_limit: limit, + ..Default::default() + }); + + // Insert limit + 2 distinct `http_endpoint` root spans all in the same time bucket. + let http_endpoints: Vec = (0..limit + 2).map(|i| format!("endpoint-{i}")).collect(); + for (i, http_endpoint) in http_endpoints.iter().enumerate() { + let meta = [("http.endpoint", http_endpoint.as_str())]; + let span = get_test_span_with_meta( + now, + i as u64 + 1, + 0, + 100, + 2, + "svc", + "resource", + 0, + &meta, + &[("_dd.measured", 1.0)], + ); + concentrator.add_span(&span); + } + // Insert a distinct `resource` root span, this one won't get collapsed + { + let meta = [("http.endpoint", "endpoint-0")]; + let span = get_test_span_with_meta( + now, + limit as u64 + 3, + 0, + 100, + 2, + "svc", + "different-resource", + 0, + &meta, + &[("_dd.measured", 1.0)], + ); + concentrator.add_span(&span); + } + + let (buckets, _) = concentrator.flush(SystemTime::now(), true); + assert!(!buckets.is_empty(), "should get at least one time bucket"); + + let stats = &buckets[0].stats; + + // Exactly limit normal keys + 1 overflow key + the distinct `resource` span + assert_eq!( + stats.len(), + limit + 2, + "expected {limit} normal groups + 1 overflow group + 1 for the distinct `resource` span, got {}", + stats.len() + ); + + // Total hits must be preserved. + let total_hits: u64 = stats.iter().map(|g| g.hits).sum(); + assert_eq!( + total_hits, + (limit + 3) as u64, + "total hits must equal the number of inserted spans" + ); + + // No overflow group, identified by the sentinel resource. + let overflow_groups: Vec<_> = stats + .iter() + .filter(|g| g.resource == TRACER_BLOCKED_VALUE) + .collect(); + assert_eq!( + overflow_groups.len(), + 0, + "expected no overflow group, given whole key cardinality limit was not reached" + ); + let http_overflow_groups: Vec<_> = stats + .iter() + .filter(|g| g.http_endpoint == TRACER_BLOCKED_VALUE) + .collect(); + assert_eq!( + http_overflow_groups.len(), + 1, + "expected exactly one overflow key for the http_endpoint field" + ); +} + /// The overflow bucket must correctly aggregate the hits from overflow spans. #[test] fn test_overflow_bucket_counts() { let now = SystemTime::now(); let limit: usize = 1; - let mut concentrator = make_cardinality_concentrator(limit); + let mut concentrator = make_cardinality_concentrator(CardinalityLimitConfig { + whole_key_limit: limit, + ..Default::default() + }); // First span fills the sole slot; the next 4 spans all have distinct keys → all overflow. for i in 0..5usize { @@ -1773,7 +1872,10 @@ fn test_overflow_bucket_counts() { fn test_no_collapse_within_limit() { let now = SystemTime::now(); let limit: usize = 10; - let mut concentrator = make_cardinality_concentrator(limit); + let mut concentrator = make_cardinality_concentrator(CardinalityLimitConfig { + whole_key_limit: limit, + ..Default::default() + }); // Insert exactly `limit` distinct-resource spans — no overflow expected. for i in 0..limit { @@ -1814,7 +1916,10 @@ fn test_no_collapse_within_limit() { fn test_overflow_bucket_key_sentinel_values() { let now = SystemTime::now(); let limit: usize = 1; - let mut concentrator = make_cardinality_concentrator(limit); + let mut concentrator = make_cardinality_concentrator(CardinalityLimitConfig { + whole_key_limit: limit, + ..Default::default() + }); // First span occupies the only slot; second one overflows. let first = get_test_span_with_meta( @@ -1896,7 +2001,7 @@ fn test_overflow_bucket_key_sentinel_values() { overflow.is_trace_root, 0, "is_trace_root must be NOT_SET (0)" ); - assert!(overflow.peer_tags.is_empty(), "peer_tags must be empty"); + assert_eq!(overflow.peer_tags, [TRACER_BLOCKED_VALUE]); // The normal group must be unaffected. let normal = stats diff --git a/libdd-trace-stats/src/stats_exporter.rs b/libdd-trace-stats/src/stats_exporter.rs index 24aa5d4c1b..a8aeeb12e9 100644 --- a/libdd-trace-stats/src/stats_exporter.rs +++ b/libdd-trace-stats/src/stats_exporter.rs @@ -323,6 +323,9 @@ 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 crate::span_concentrator::CardinalityLimitConfig; use httpmock::prelude::*; use httpmock::MockServer; use libdd_capabilities_impl::NativeCapabilities; @@ -589,7 +592,6 @@ 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; @@ -642,7 +644,10 @@ mod tests { SystemTime::now(), vec![], vec![], - Some(1), // max 1 distinct key → second span collapses + Some(CardinalityLimitConfig { + whole_key_limit: 1, // max 1 distinct key → second span collapses + ..Default::default() + }), #[cfg(feature = "stats-obfuscation")] None, ); From 1c9dca962201fdf89ab4d1d0d804117f24e2b656 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Wed, 8 Jul 2026 14:17:59 +0200 Subject: [PATCH 18/30] feat: store distinct hashes instead of values save memory by storing value hashes only, u64 has enough values to make collisions rare and in case we still get a collision storing 1 extra key should be fine --- .../src/span_concentrator/aggregation.rs | 43 ++++++++++--------- 1 file changed, 23 insertions(+), 20 deletions(-) diff --git a/libdd-trace-stats/src/span_concentrator/aggregation.rs b/libdd-trace-stats/src/span_concentrator/aggregation.rs index dc6a4c2e59..51c4905d18 100644 --- a/libdd-trace-stats/src/span_concentrator/aggregation.rs +++ b/libdd-trace-stats/src/span_concentrator/aggregation.rs @@ -9,7 +9,10 @@ use hashbrown::{HashMap, HashSet}; use libdd_trace_obfuscation::ip_address::quantize_peer_ip_addresses; use libdd_trace_protobuf::pb; use libdd_trace_utils::span::SpanText; -use std::borrow::{Borrow, Cow}; +use std::{ + borrow::{Borrow, Cow}, + hash::{DefaultHasher, Hash, Hasher as _}, +}; use crate::span_concentrator::StatSpan; @@ -432,11 +435,11 @@ pub(super) struct StatsBucket { /// ones into the overflow sentinel key. cardinality_limits: CardinalityLimitConfig, // FIXME: optimize memory by storing hashes only - distinct_resources: HashSet, - distinct_http_endpoint: HashSet, - distinct_peer_tags: HashSet>, + distinct_resources: HashSet, + distinct_http_endpoint: HashSet, + distinct_peer_tags: HashSet, #[allow(unused, reason = "FIXME: implement stats additional tags")] - distinct_additional_tags: HashSet, + distinct_additional_tags: HashSet, /// Number of spans collapsed into the overflow bucket due to cardinality limiting. collapsed_count: u64, /// Indicates if stats obfuscated in this bucket. This is set once at creation and stays @@ -504,37 +507,37 @@ impl StatsBucket { /// Collapse an aggregation key fields following the bucket's `CardinalityLimitConfig` fn collapse_key_cardinality(&mut self, key: &mut BorrowedAggregationKey<'_>) { + fn hash(input: &impl Hash) -> u64 { + let mut hasher = DefaultHasher::new(); + input.hash(&mut hasher); + hasher.finish() + } + + let resource_name_hash = hash(&key.fixed.resource_name); if self.distinct_resources.len() >= self.cardinality_limits.resource_limit - && !self.distinct_resources.contains(key.fixed.resource_name) + && !self.distinct_resources.contains(&resource_name_hash) { key.fixed.resource_name = TRACER_BLOCKED_VALUE; } else { - self.distinct_resources - .insert(key.fixed.resource_name.to_owned()); + self.distinct_resources.insert(resource_name_hash); } + let http_endpoint_hash = hash(&key.fixed.http_endpoint); if self.distinct_http_endpoint.len() >= self.cardinality_limits.http_endpoint_limit - && !self - .distinct_http_endpoint - .contains(key.fixed.http_endpoint) + && !self.distinct_http_endpoint.contains(&http_endpoint_hash) { key.fixed.http_endpoint = TRACER_BLOCKED_VALUE; } else { - self.distinct_http_endpoint - .insert(key.fixed.http_endpoint.to_owned()); + self.distinct_http_endpoint.insert(http_endpoint_hash); } - let owned_peer_tags = key - .peer_tags - .iter() - .map(|(k, v)| ((*k).to_owned(), v.to_string())) - .collect(); + let peer_tags_hash = hash(&key.peer_tags); if self.distinct_peer_tags.len() >= self.cardinality_limits.peer_tags_limit - && !self.distinct_peer_tags.contains(&owned_peer_tags) + && !self.distinct_peer_tags.contains(&peer_tags_hash) { key.peer_tags = vec![(TRACER_BLOCKED_VALUE, Cow::Borrowed(TRACER_BLOCKED_VALUE))]; } else { - self.distinct_peer_tags.insert(owned_peer_tags); + self.distinct_peer_tags.insert(peer_tags_hash); } } From 43a1b0bf3d39024b57b16191ce0e6a4f304dd234 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Wed, 8 Jul 2026 14:37:16 +0200 Subject: [PATCH 19/30] fix: libdd-data-pipeline(-ffi) build --- libdd-data-pipeline-ffi/src/trace_exporter.rs | 37 ++++++++++++++----- .../src/trace_exporter/builder.rs | 14 ++++--- libdd-data-pipeline/src/trace_exporter/mod.rs | 6 +-- .../src/trace_exporter/stats.rs | 6 ++- .../src/span_concentrator/mod.rs | 3 +- 5 files changed, 46 insertions(+), 20 deletions(-) diff --git a/libdd-data-pipeline-ffi/src/trace_exporter.rs b/libdd-data-pipeline-ffi/src/trace_exporter.rs index c3a68962a4..f7369386b7 100644 --- a/libdd-data-pipeline-ffi/src/trace_exporter.rs +++ b/libdd-data-pipeline-ffi/src/trace_exporter.rs @@ -9,6 +9,7 @@ use libdd_common_ffi::{ CharSlice, {slice::AsBytes, slice::ByteSlice}, }; +use libdd_data_pipeline::trace_exporter::stats::CardinalityLimitConfig; use libdd_data_pipeline::trace_exporter::{ TelemetryConfig, TelemetryInstrumentationSessions, TraceExporter as GenericTraceExporter, TraceExporterInputFormat, TraceExporterOutputFormat, @@ -90,7 +91,7 @@ pub struct TraceExporterConfig { otlp_protocol: Option, output_to_log: bool, log_max_line_size: Option, - stats_cardinality_limit: Option, + stats_cardinality_limits: Option, } #[no_mangle] @@ -555,11 +556,11 @@ pub unsafe extern "C" fn ddog_trace_exporter_config_set_otlp_protocol( #[no_mangle] pub unsafe extern "C" fn ddog_trace_exporter_config_set_stats_cardinality_limit( config: Option<&mut TraceExporterConfig>, - limit: usize, + limits: CardinalityLimitConfig, ) -> Option> { catch_panic!( if let Some(handle) = config { - handle.stats_cardinality_limit = Some(limit); + handle.stats_cardinality_limits = Some(limits); None } else { gen_error!(ErrorCode::InvalidArgument) @@ -645,7 +646,7 @@ pub unsafe extern "C" fn ddog_trace_exporter_new( .set_output_format(config.output_format) .set_connection_timeout(config.connection_timeout); - if let Some(limit) = config.stats_cardinality_limit { + if let Some(limit) = config.stats_cardinality_limits { builder.set_stats_cardinality_limit(limit); } @@ -782,7 +783,7 @@ mod tests { assert!(cfg.connection_timeout.is_none()); assert!(!cfg.output_to_log); assert_eq!(cfg.log_max_line_size, None); - assert_eq!(cfg.stats_cardinality_limit, None); + assert_eq!(cfg.stats_cardinality_limits, None); ddog_trace_exporter_config_free(cfg); } @@ -1501,16 +1502,34 @@ mod tests { fn config_stats_cardinality_limit_test() { unsafe { // Null config → InvalidArgument - let error = ddog_trace_exporter_config_set_stats_cardinality_limit(None, 100); + let error = ddog_trace_exporter_config_set_stats_cardinality_limit( + None, + CardinalityLimitConfig { + whole_key_limit: 100, + ..Default::default() + }, + ); assert_eq!(error.as_ref().unwrap().code, ErrorCode::InvalidArgument); ddog_trace_exporter_error_free(error); // Valid config → value stored let mut config = Some(TraceExporterConfig::default()); - let error = - ddog_trace_exporter_config_set_stats_cardinality_limit(config.as_mut(), 500); + let error = ddog_trace_exporter_config_set_stats_cardinality_limit( + config.as_mut(), + CardinalityLimitConfig { + whole_key_limit: 500, + ..Default::default() + }, + ); assert_eq!(error, None); - assert_eq!(config.unwrap().stats_cardinality_limit, Some(500)); + assert_eq!( + config + .unwrap() + .stats_cardinality_limits + .unwrap() + .whole_key_limit, + 500 + ); } } diff --git a/libdd-data-pipeline/src/trace_exporter/builder.rs b/libdd-data-pipeline/src/trace_exporter/builder.rs index eef1c9e3f9..628ab71eaf 100644 --- a/libdd-data-pipeline/src/trace_exporter/builder.rs +++ b/libdd-data-pipeline/src/trace_exporter/builder.rs @@ -26,6 +26,7 @@ use libdd_dogstatsd_client::new; use libdd_shared_runtime::SharedRuntime; #[cfg(not(target_arch = "wasm32"))] use libdd_shared_runtime::{BlockingRuntime, ForkSafeRuntime}; +use libdd_trace_stats::span_concentrator::CardinalityLimitConfig; use libdd_trace_utils::trace_filter::TraceFilterer; use std::sync::Arc; use std::time::Duration; @@ -75,7 +76,7 @@ pub struct TraceExporterBuilder { peer_tags_aggregation: bool, compute_stats_by_span_kind: bool, peer_tags: Vec, - stats_cardinality_limit: Option, + stats_cardinality_limits: Option, #[cfg(feature = "stats-obfuscation")] client_side_stats_obfuscation_enabled: bool, #[cfg(feature = "telemetry")] @@ -143,7 +144,7 @@ impl TraceExporterBuilder { peer_tags_aggregation: false, compute_stats_by_span_kind: false, peer_tags: Vec::new(), - stats_cardinality_limit: None, + stats_cardinality_limits: None, #[cfg(feature = "stats-obfuscation")] client_side_stats_obfuscation_enabled: false, #[cfg(feature = "telemetry")] @@ -333,8 +334,11 @@ impl TraceExporterBuilder { /// This bounds memory usage when the trace population has very high cardinality. /// /// Has no effect unless stats computation is enabled. - pub fn set_stats_cardinality_limit(&mut self, cardinality_limit: usize) -> &mut Self { - self.stats_cardinality_limit = Some(cardinality_limit); + pub fn set_stats_cardinality_limit( + &mut self, + cardinality_limits: CardinalityLimitConfig, + ) -> &mut Self { + self.stats_cardinality_limits = Some(cardinality_limits); self } @@ -820,7 +824,7 @@ impl TraceExporterBuilder { common_stats_tags: vec![libdatadog_version], client_side_stats: StatsComputationConfig { status: ArcSwap::new(stats.into()), - stats_cardinality_limit: self.stats_cardinality_limit, + stats_cardinality_limits: self.stats_cardinality_limits, #[cfg(feature = "stats-obfuscation")] obfuscation_config: Arc::new(ArcSwap::from_pointee( StatsComputationObfuscationConfig::default(), diff --git a/libdd-data-pipeline/src/trace_exporter/mod.rs b/libdd-data-pipeline/src/trace_exporter/mod.rs index 08f69e8b99..f0a4e5c9a7 100644 --- a/libdd-data-pipeline/src/trace_exporter/mod.rs +++ b/libdd-data-pipeline/src/trace_exporter/mod.rs @@ -469,7 +469,7 @@ impl< metadata: &self.metadata, endpoint_url: &self.endpoint.url, shared_runtime: &*self.shared_runtime, - stats_cardinality_limit: self.client_side_stats.stats_cardinality_limit, + stats_cardinality_limits: self.client_side_stats.stats_cardinality_limits, dogstatsd: if self.health_metrics_enabled { self.dogstatsd.clone() } else { @@ -875,8 +875,8 @@ impl< && self.v1_active.swap(false, Ordering::Relaxed) { warn!( - "V1 trace send returned 404; agent no longer advertises {V1_TRACES_ENDPOINT} — falling back to V0.4" - ); + "V1 trace send returned 404; agent no longer advertises {V1_TRACES_ENDPOINT} — falling back to V0.4" + ); self.info_response_observer.manual_trigger(); } } diff --git a/libdd-data-pipeline/src/trace_exporter/stats.rs b/libdd-data-pipeline/src/trace_exporter/stats.rs index 33e3ac9362..4ccf3f0fb1 100644 --- a/libdd-data-pipeline/src/trace_exporter/stats.rs +++ b/libdd-data-pipeline/src/trace_exporter/stats.rs @@ -7,6 +7,8 @@ //! including starting/stopping stats workers, managing the span concentrator, //! and processing traces for stats collection. +pub use libdd_trace_stats::span_concentrator::CardinalityLimitConfig; + #[cfg(not(target_arch = "wasm32"))] use super::add_path; use super::TracerMetadata; @@ -18,7 +20,7 @@ use libdd_capabilities::{HttpClientCapability, MaybeSend, SleepCapability}; use libdd_common::Endpoint; use libdd_common::MutexExt; use libdd_shared_runtime::{SharedRuntime, WorkerHandle}; -use libdd_trace_stats::span_concentrator::{CardinalityLimitConfig, SpanConcentrator}; +use libdd_trace_stats::span_concentrator::SpanConcentrator; #[cfg(feature = "stats-obfuscation")] use libdd_trace_stats::span_concentrator::{ SharedStatsComputationObfuscationConfig, StatsComputationObfuscationConfig, @@ -77,7 +79,7 @@ pub(crate) enum StatsComputationStatus { #[cfg_attr(target_arch = "wasm32", allow(dead_code))] pub(crate) struct StatsComputationConfig { pub(crate) status: ArcSwap, - pub(crate) stats_cardinality_limit: Option, + pub(crate) stats_cardinality_limits: Option, #[cfg(feature = "stats-obfuscation")] pub(crate) obfuscation_config: SharedStatsComputationObfuscationConfig, /// Builder-level opt-in. When false, stats obfuscation stays off diff --git a/libdd-trace-stats/src/span_concentrator/mod.rs b/libdd-trace-stats/src/span_concentrator/mod.rs index d1db96b738..c00da5d2d3 100644 --- a/libdd-trace-stats/src/span_concentrator/mod.rs +++ b/libdd-trace-stats/src/span_concentrator/mod.rs @@ -101,7 +101,8 @@ pub type SharedStatsComputationObfuscationConfig = pub const DEFAULT_MAX_ENTRIES_PER_BUCKET: usize = 7_000; /// Config to override the default stats cardinality limit values -#[derive(Debug, Clone, Copy)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[repr(C)] pub struct CardinalityLimitConfig { /// The whole-key cardinality limit (defaults to 7000) pub whole_key_limit: usize, From 586e6d7aff7373bbb62729ac0ef7e39ec9f6509b Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Wed, 8 Jul 2026 14:54:31 +0200 Subject: [PATCH 20/30] fix: wrong peer_tags collapse value --- 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 51c4905d18..f55e76bb31 100644 --- a/libdd-trace-stats/src/span_concentrator/aggregation.rs +++ b/libdd-trace-stats/src/span_concentrator/aggregation.rs @@ -535,7 +535,7 @@ impl StatsBucket { if self.distinct_peer_tags.len() >= self.cardinality_limits.peer_tags_limit && !self.distinct_peer_tags.contains(&peer_tags_hash) { - key.peer_tags = vec![(TRACER_BLOCKED_VALUE, Cow::Borrowed(TRACER_BLOCKED_VALUE))]; + key.peer_tags = vec![(TRACER_BLOCKED_VALUE, Cow::Borrowed(""))]; } else { self.distinct_peer_tags.insert(peer_tags_hash); } From cc5bf3b2bab1010da96b890103328054eaff67a7 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Wed, 8 Jul 2026 14:56:36 +0200 Subject: [PATCH 21/30] fix: fmt --- 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 a8aeeb12e9..302edf75cc 100644 --- a/libdd-trace-stats/src/stats_exporter.rs +++ b/libdd-trace-stats/src/stats_exporter.rs @@ -323,9 +323,9 @@ pub fn stats_url_from_agent_url(agent_url: &str) -> anyhow::Result { #[cfg(test)] mod tests { use super::*; + use crate::span_concentrator::CardinalityLimitConfig; #[cfg(feature = "stats-obfuscation")] use crate::span_concentrator::StatsComputationObfuscationConfig; - use crate::span_concentrator::CardinalityLimitConfig; use httpmock::prelude::*; use httpmock::MockServer; use libdd_capabilities_impl::NativeCapabilities; From 726c57a17205e021c4ec573607bf05f9962d3246 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Wed, 8 Jul 2026 15:34:19 +0200 Subject: [PATCH 22/30] fix: collapse key fields before applying whole-key cardinality limits --- .../src/span_concentrator/aggregation.rs | 9 +- .../src/span_concentrator/tests.rs | 85 ++++++++++++++++++- 2 files changed, 86 insertions(+), 8 deletions(-) diff --git a/libdd-trace-stats/src/span_concentrator/aggregation.rs b/libdd-trace-stats/src/span_concentrator/aggregation.rs index f55e76bb31..680b41220c 100644 --- a/libdd-trace-stats/src/span_concentrator/aggregation.rs +++ b/libdd-trace-stats/src/span_concentrator/aggregation.rs @@ -434,7 +434,6 @@ pub(super) struct StatsBucket { /// Maximum number of distinct aggregation keys this bucket will hold before collapsing new /// ones into the overflow sentinel key. cardinality_limits: CardinalityLimitConfig, - // FIXME: optimize memory by storing hashes only distinct_resources: HashSet, distinct_http_endpoint: HashSet, distinct_peer_tags: HashSet, @@ -486,6 +485,10 @@ impl StatsBucket { is_error: bool, is_top_level: bool, ) { + // Per field cardinality limiting + self.collapse_key_fields_cardinality(&mut key); + + // Whole-key cardinality limiting if self.data.len() >= self.cardinality_limits.whole_key_limit && !self.data.contains_key(&key) { @@ -497,8 +500,6 @@ impl StatsBucket { return; } - self.collapse_key_cardinality(&mut key); - self.data .entry_ref(&key) .or_default() @@ -506,7 +507,7 @@ impl StatsBucket { } /// Collapse an aggregation key fields following the bucket's `CardinalityLimitConfig` - fn collapse_key_cardinality(&mut self, key: &mut BorrowedAggregationKey<'_>) { + fn collapse_key_fields_cardinality(&mut self, key: &mut BorrowedAggregationKey<'_>) { fn hash(input: &impl Hash) -> u64 { let mut hasher = DefaultHasher::new(); input.hash(&mut hasher); diff --git a/libdd-trace-stats/src/span_concentrator/tests.rs b/libdd-trace-stats/src/span_concentrator/tests.rs index 9001f0b916..89507a1eb7 100644 --- a/libdd-trace-stats/src/span_concentrator/tests.rs +++ b/libdd-trace-stats/src/span_concentrator/tests.rs @@ -6,7 +6,7 @@ use crate::span_concentrator::aggregation::{OwnedAggregationKey, TRACER_BLOCKED_ use super::*; use libdd_trace_utils::span::v04::VecMap; use libdd_trace_utils::span::{trace_utils::compute_top_level_span, v04::SpanSlice}; -use rand::{thread_rng, Rng}; +use rand::{Rng, thread_rng}; const BUCKET_SIZE: u64 = Duration::from_secs(2).as_nanos() as u64; @@ -1653,7 +1653,7 @@ fn make_cardinality_concentrator(cardinality_limits: CardinalityLimitConfig) -> Duration::from_nanos(BUCKET_SIZE), now, get_span_kinds(), - vec![], + vec!["peer.hostname".to_owned()], Some(cardinality_limits), #[cfg(feature = "stats-obfuscation")] None, @@ -1773,7 +1773,7 @@ fn test_per_key_cardinality_limit_collapse_http_endpoint() { 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; @@ -1782,7 +1782,7 @@ fn test_per_key_cardinality_limit_collapse_http_endpoint() { assert_eq!( stats.len(), limit + 2, - "expected {limit} normal groups + 1 overflow group + 1 for the distinct `resource` span, got {}", + "expected {limit} normal groups + 1 overflow key + 1 for the distinct `resource` span, got {}", stats.len() ); @@ -1815,6 +1815,83 @@ fn test_per_key_cardinality_limit_collapse_http_endpoint() { ); } +/// When whole-key cardinality limit is reached, check that per-key fields are collapsed before +/// falling back to whole-key +#[test] +fn test_per_key_cardinality_limit_collapse_before_whole_key() { + let now = SystemTime::now(); + let peer_tags_limit = 3; + let whole_key_limit = peer_tags_limit + 1; + let mut concentrator = make_cardinality_concentrator(CardinalityLimitConfig { + whole_key_limit, + peer_tags_limit, + ..Default::default() + }); + + // Insert limit + 2 distinct `peer.hostname` root spans all in the same time bucket. + let inserted_spans = peer_tags_limit + 2; + let peer_tag_values: Vec = (0..inserted_spans).map(|i| format!("peer-{i}")).collect(); + for (i, peer_tag_value) in peer_tag_values.iter().enumerate() { + let meta = [ + ("peer.hostname", peer_tag_value.as_str()), + ("span.kind", "client"), + ]; + let span = get_test_span_with_meta( + now, + i as u64 + 1, + 0, + 100, + 2, + "svc", + "resource", + 0, + &meta, + &[("_dd.measured", 1.0)], + ); + concentrator.add_span(&span); + } + + 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; + + // Exactly peer_tags_limit normal keys + 1 overflow key + assert_eq!( + stats.len(), + peer_tags_limit + 1, + "expected {peer_tags_limit} normal groups + 1 overflow key, got {}", + stats.len() + ); + + // Total hits must be preserved. + let total_hits: u64 = stats.iter().map(|g| g.hits).sum(); + assert_eq!( + total_hits, inserted_spans as u64, + "total hits must equal the number of inserted spans" + ); + + // No overflow group, identified by the sentinel resource. + let overflow_groups: Vec<_> = stats + .iter() + .filter(|g| g.resource == TRACER_BLOCKED_VALUE) + .collect(); + assert_eq!( + overflow_groups.len(), + 0, + "expected no overflow group: whole key cardinality limit was not reached because per-field cardinality limit collapsed keys before overflowing whole key" + ); + let peer_tag_overflow_groups: Vec<_> = stats + .iter() + .filter(|g| g.peer_tags == [TRACER_BLOCKED_VALUE]) + .collect(); + assert_eq!( + peer_tag_overflow_groups.len(), + 1, + "expected exactly one overflow key for the peer_tags field" + ); +} + /// The overflow bucket must correctly aggregate the hits from overflow spans. #[test] fn test_overflow_bucket_counts() { From bd6e3bca16529794e3a4cbd008c674a3cfed8e5d Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Wed, 8 Jul 2026 15:34:45 +0200 Subject: [PATCH 23/30] feat: add a warning when whole-key limit is lower than per-field limits --- libdd-trace-stats/src/span_concentrator/mod.rs | 16 +++++++++++++++- 1 file changed, 15 insertions(+), 1 deletion(-) diff --git a/libdd-trace-stats/src/span_concentrator/mod.rs b/libdd-trace-stats/src/span_concentrator/mod.rs index c00da5d2d3..0b1faee070 100644 --- a/libdd-trace-stats/src/span_concentrator/mod.rs +++ b/libdd-trace-stats/src/span_concentrator/mod.rs @@ -3,7 +3,7 @@ //! This module implements the SpanConcentrator used to aggregate spans into stats use std::collections::HashMap; use std::time::{self, Duration, SystemTime}; -use tracing::debug; +use tracing::{debug, warn}; use libdd_trace_protobuf::pb; @@ -187,6 +187,20 @@ impl SpanConcentrator { SharedStatsComputationObfuscationConfig, >, ) -> SpanConcentrator { + if let Some(cardinality_limit_config) = override_cardinality_limits.as_ref() { + if cardinality_limit_config.whole_key_limit <= cardinality_limit_config.resource_limit + || cardinality_limit_config.whole_key_limit + <= cardinality_limit_config.http_endpoint_limit + || cardinality_limit_config.whole_key_limit + <= cardinality_limit_config.peer_tags_limit + || cardinality_limit_config.whole_key_limit + <= cardinality_limit_config.additional_tags_limit + { + warn!( + "Stats cardinality limit is misconfigured: per-field limits must be lower than whole-key limit otherwise they have no effect and you will get over-collapsed stats!" + ); + } + } SpanConcentrator { bucket_size: bucket_size.as_nanos() as u64, buckets: HashMap::new(), From 1c46a805b4536b32b53822e54549754a5ceb66c8 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Wed, 8 Jul 2026 15:35:51 +0200 Subject: [PATCH 24/30] fix: fmt --- libdd-trace-stats/src/span_concentrator/tests.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/libdd-trace-stats/src/span_concentrator/tests.rs b/libdd-trace-stats/src/span_concentrator/tests.rs index 89507a1eb7..938ef8e8b9 100644 --- a/libdd-trace-stats/src/span_concentrator/tests.rs +++ b/libdd-trace-stats/src/span_concentrator/tests.rs @@ -6,7 +6,7 @@ use crate::span_concentrator::aggregation::{OwnedAggregationKey, TRACER_BLOCKED_ use super::*; use libdd_trace_utils::span::v04::VecMap; use libdd_trace_utils::span::{trace_utils::compute_top_level_span, v04::SpanSlice}; -use rand::{Rng, thread_rng}; +use rand::{thread_rng, Rng}; const BUCKET_SIZE: u64 = Duration::from_secs(2).as_nanos() as u64; From 04befd7a00d24231e1940fe708316fa060d39bae Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Wed, 8 Jul 2026 15:58:19 +0200 Subject: [PATCH 25/30] fix(ffi): add libdd-trace-stats to cbindgen config to generate CardinalityLimitConfig binding --- libdd-data-pipeline-ffi/cbindgen.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/libdd-data-pipeline-ffi/cbindgen.toml b/libdd-data-pipeline-ffi/cbindgen.toml index d3e36b4945..5e18df2f87 100644 --- a/libdd-data-pipeline-ffi/cbindgen.toml +++ b/libdd-data-pipeline-ffi/cbindgen.toml @@ -47,4 +47,4 @@ must_use = "DDOG_CHECK_RETURN" [parse] parse_deps = true -include = ["libdd-common", "libdd-common-ffi", "libdd-shared-runtime", "libdd-data-pipeline", "libdd-trace-utils"] +include = ["libdd-common", "libdd-common-ffi", "libdd-shared-runtime", "libdd-data-pipeline", "libdd-trace-utils", "libdd-trace-stats"] From 74f105a43f7e6b7bfe982f802a4542f89798e690 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Fri, 10 Jul 2026 13:11:47 +0200 Subject: [PATCH 26/30] fix: inline DEFAULT_MAX_ENTRIES_PER_BUCKET --- .../src/span_concentrator/mod.rs | 24 +++++++++---------- 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/libdd-trace-stats/src/span_concentrator/mod.rs b/libdd-trace-stats/src/span_concentrator/mod.rs index 0b1faee070..3b29e6685d 100644 --- a/libdd-trace-stats/src/span_concentrator/mod.rs +++ b/libdd-trace-stats/src/span_concentrator/mod.rs @@ -92,14 +92,6 @@ pub struct StatsComputationObfuscationConfig { pub type SharedStatsComputationObfuscationConfig = std::sync::Arc>; -/// Default maximum number of distinct aggregation keys per time bucket. -/// -/// 7 168 is the limit to exactly saturate hashbrown's internal table at its maximum load factor of -/// 7/8. Any higher limit would immediately force a doubling of the table capacity, wasting -/// half the allocated slots for a modest increase in cardinality. To avoid future changes going -/// over this limit (e.g. adding extra overflow buckets) we set a slightly lower limit. -pub const DEFAULT_MAX_ENTRIES_PER_BUCKET: usize = 7_000; - /// Config to override the default stats cardinality limit values #[derive(Debug, Clone, Copy, PartialEq, Eq)] #[repr(C)] @@ -119,8 +111,16 @@ pub struct CardinalityLimitConfig { impl Default for CardinalityLimitConfig { fn default() -> Self { Self { - whole_key_limit: DEFAULT_MAX_ENTRIES_PER_BUCKET, - resource_limit: 1024, + // Default maximum number of distinct aggregation keys per time bucket. + // + // 7 168 is the limit to exactly saturate hashbrown's internal table at its maximum + // load factor of 7/8. Any higher limit would immediately force a doubling + // of the table capacity, wasting half the allocated slots for a modest + // increase in cardinality. To avoid future changes going over this limit + // (e.g. adding extra overflow buckets) we set a slightly lower limit. + whole_key_limit: 7_000, + // Other defaults from the spec + resource_limit: 1_024, http_endpoint_limit: 512, peer_tags_limit: 512, additional_tags_limit: 100, @@ -174,8 +174,8 @@ impl SpanConcentrator { /// - `now` the current system time, used to define the oldest bucket /// - `span_kinds_stats_computed` list of span kinds eligible for stats computation /// - `peer_tags_keys` list of keys considered as peer tags for aggregation - /// - `override_max_entries_per_bucket` maximum distinct aggregation keys per time bucket before - /// cardinality limiting applies. Pass `None` to use [`DEFAULT_MAX_ENTRIES_PER_BUCKET`]. + /// - `override_cardinality_limits` config values for whole-key and per-field cardinality limi. + /// Pass `None` to use defaults (see [`CardinalityLimitConfig`]). /// - `obfuscation_config` optional and updatable config for resource key obfuscation pub fn new( bucket_size: Duration, From edf1e0cd3a115a22d4b625b655590d05e685da1c Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Fri, 10 Jul 2026 13:22:31 +0200 Subject: [PATCH 27/30] fix: link span derived primary tags ticket and pr https://github.com/DataDog/libdatadog/pull/2170 --- libdd-trace-stats/src/span_concentrator/aggregation.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/libdd-trace-stats/src/span_concentrator/aggregation.rs b/libdd-trace-stats/src/span_concentrator/aggregation.rs index 680b41220c..988a0fd941 100644 --- a/libdd-trace-stats/src/span_concentrator/aggregation.rs +++ b/libdd-trace-stats/src/span_concentrator/aggregation.rs @@ -437,7 +437,10 @@ pub(super) struct StatsBucket { distinct_resources: HashSet, distinct_http_endpoint: HashSet, distinct_peer_tags: HashSet, - #[allow(unused, reason = "FIXME: implement stats additional tags")] + #[allow( + unused, + reason = "FIXME(SVLS-8787|github.com/DataDog/libdatadog/pull/2170): implement stats additional tags" + )] distinct_additional_tags: HashSet, /// Number of spans collapsed into the overflow bucket due to cardinality limiting. collapsed_count: u64, From 5cedcf4a8f3f4c5d57ceace116195276008d3875 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Fri, 10 Jul 2026 13:43:17 +0200 Subject: [PATCH 28/30] fix: avoid useless re-hashes --- .../src/span_concentrator/aggregation.rs | 47 ++++++++++--------- 1 file changed, 25 insertions(+), 22 deletions(-) diff --git a/libdd-trace-stats/src/span_concentrator/aggregation.rs b/libdd-trace-stats/src/span_concentrator/aggregation.rs index 988a0fd941..878d551087 100644 --- a/libdd-trace-stats/src/span_concentrator/aggregation.rs +++ b/libdd-trace-stats/src/span_concentrator/aggregation.rs @@ -5,7 +5,7 @@ //! This includes the aggregation key to group spans together and the computation of stats from a //! span. -use hashbrown::{HashMap, HashSet}; +use hashbrown::{HashMap, HashSet, hash_set::Entry}; use libdd_trace_obfuscation::ip_address::quantize_peer_ip_addresses; use libdd_trace_protobuf::pb; use libdd_trace_utils::span::SpanText; @@ -435,7 +435,7 @@ pub(super) struct StatsBucket { /// ones into the overflow sentinel key. cardinality_limits: CardinalityLimitConfig, distinct_resources: HashSet, - distinct_http_endpoint: HashSet, + distinct_http_endpoints: HashSet, distinct_peer_tags: HashSet, #[allow( unused, @@ -467,7 +467,7 @@ impl StatsBucket { #[cfg(feature = "stats-obfuscation")] obfuscated: obfuscation_enabled, distinct_resources: HashSet::new(), - distinct_http_endpoint: HashSet::new(), + distinct_http_endpoints: HashSet::new(), distinct_peer_tags: HashSet::new(), distinct_additional_tags: HashSet::new(), } @@ -517,31 +517,34 @@ impl StatsBucket { hasher.finish() } - let resource_name_hash = hash(&key.fixed.resource_name); - if self.distinct_resources.len() >= self.cardinality_limits.resource_limit - && !self.distinct_resources.contains(&resource_name_hash) - { - key.fixed.resource_name = TRACER_BLOCKED_VALUE; - } else { - self.distinct_resources.insert(resource_name_hash); + let resource_hash = hash(&key.fixed.resource_name); + let resources_count = self.distinct_resources.len(); + if let Entry::Vacant(slot) = self.distinct_resources.entry(resource_hash) { + if resources_count >= self.cardinality_limits.resource_limit { + key.fixed.resource_name = TRACER_BLOCKED_VALUE; + } else { + slot.insert(); + } } let http_endpoint_hash = hash(&key.fixed.http_endpoint); - if self.distinct_http_endpoint.len() >= self.cardinality_limits.http_endpoint_limit - && !self.distinct_http_endpoint.contains(&http_endpoint_hash) - { - key.fixed.http_endpoint = TRACER_BLOCKED_VALUE; - } else { - self.distinct_http_endpoint.insert(http_endpoint_hash); + let http_endpoints_count = self.distinct_http_endpoints.len(); + if let Entry::Vacant(slot) = self.distinct_http_endpoints.entry(http_endpoint_hash) { + if http_endpoints_count >= self.cardinality_limits.http_endpoint_limit { + key.fixed.http_endpoint = TRACER_BLOCKED_VALUE; + } else { + slot.insert(); + } } let peer_tags_hash = hash(&key.peer_tags); - if self.distinct_peer_tags.len() >= self.cardinality_limits.peer_tags_limit - && !self.distinct_peer_tags.contains(&peer_tags_hash) - { - key.peer_tags = vec![(TRACER_BLOCKED_VALUE, Cow::Borrowed(""))]; - } else { - self.distinct_peer_tags.insert(peer_tags_hash); + let peer_tags_count = self.distinct_peer_tags.len(); + if let Entry::Vacant(slot) = self.distinct_peer_tags.entry(peer_tags_hash) { + if peer_tags_count >= self.cardinality_limits.peer_tags_limit { + key.peer_tags = vec![(TRACER_BLOCKED_VALUE, Cow::Borrowed(""))]; + } else { + slot.insert(); + } } } From 83043a4a3d4a09c794deb5719d918e76890a8067 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Fri, 10 Jul 2026 13:47:22 +0200 Subject: [PATCH 29/30] fix: warn on any cardinality limit = 0 --- libdd-trace-stats/src/span_concentrator/mod.rs | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/libdd-trace-stats/src/span_concentrator/mod.rs b/libdd-trace-stats/src/span_concentrator/mod.rs index 3b29e6685d..6f62b5e53c 100644 --- a/libdd-trace-stats/src/span_concentrator/mod.rs +++ b/libdd-trace-stats/src/span_concentrator/mod.rs @@ -188,6 +188,16 @@ impl SpanConcentrator { >, ) -> SpanConcentrator { if let Some(cardinality_limit_config) = override_cardinality_limits.as_ref() { + if cardinality_limit_config.whole_key_limit == 0 + || cardinality_limit_config.http_endpoint_limit == 0 + || cardinality_limit_config.peer_tags_limit == 0 + || cardinality_limit_config.additional_tags_limit == 0 + { + warn!( + ?cardinality_limit_config, + "Stats cardinality limit is misconfigured: cardinality limits must not be 0 otherwise all the stats get collapsed!" + ); + } if cardinality_limit_config.whole_key_limit <= cardinality_limit_config.resource_limit || cardinality_limit_config.whole_key_limit <= cardinality_limit_config.http_endpoint_limit @@ -197,6 +207,7 @@ impl SpanConcentrator { <= cardinality_limit_config.additional_tags_limit { warn!( + ?cardinality_limit_config, "Stats cardinality limit is misconfigured: per-field limits must be lower than whole-key limit otherwise they have no effect and you will get over-collapsed stats!" ); } From 39427ce03c5b4f0f906043af2d8687f216d7487f Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Fri, 10 Jul 2026 14:22:14 +0200 Subject: [PATCH 30/30] fix: fmt --- 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 878d551087..73873779fb 100644 --- a/libdd-trace-stats/src/span_concentrator/aggregation.rs +++ b/libdd-trace-stats/src/span_concentrator/aggregation.rs @@ -5,7 +5,7 @@ //! This includes the aggregation key to group spans together and the computation of stats from a //! span. -use hashbrown::{HashMap, HashSet, hash_set::Entry}; +use hashbrown::{hash_set::Entry, HashMap, HashSet}; use libdd_trace_obfuscation::ip_address::quantize_peer_ip_addresses; use libdd_trace_protobuf::pb; use libdd_trace_utils::span::SpanText;