From c6af6a2511d88426b267dc8e7257a2dc6cff04bc Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Fri, 10 Jul 2026 16:45:13 +0200 Subject: [PATCH 01/15] feat: fields combination metrics, dogstatsd only --- .../src/span_concentrator/aggregation.rs | 91 ++++++++++++++++++- .../src/span_concentrator/mod.rs | 17 +++- libdd-trace-stats/src/stats_exporter.rs | 5 + 3 files changed, 107 insertions(+), 6 deletions(-) diff --git a/libdd-trace-stats/src/span_concentrator/aggregation.rs b/libdd-trace-stats/src/span_concentrator/aggregation.rs index 002d60e7e7..dc199ddc37 100644 --- a/libdd-trace-stats/src/span_concentrator/aggregation.rs +++ b/libdd-trace-stats/src/span_concentrator/aggregation.rs @@ -6,6 +6,7 @@ //! span. use hashbrown::{HashMap, HashSet}; +use libdd_common::tag::const_assert; use libdd_trace_obfuscation::ip_address::quantize_peer_ip_addresses; use libdd_trace_protobuf::pb; use libdd_trace_utils::span::SpanText; @@ -527,14 +528,90 @@ pub(super) struct StatsBucket { distinct_http_endpoints: HashSet, distinct_peer_tags: HashSet, distinct_additional_tags: HashSet, - /// Number of spans collapsed into the overflow bucket due to cardinality limiting. + /// Number of spans collapsed into the overflow bucket due to whole-key cardinality limiting. collapsed_count: u64, + collapsed_fields_metrics: CollapsedFieldsMetrics, /// 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, } +#[repr(transparent)] +pub struct CollapsedField; +impl CollapsedField { + pub const RESOURCE_NAME: usize = 1 << 1; + pub const HTTP_ENDPOINT: usize = 1 << 2; + pub const PEER_TAGS: usize = 1 << 3; + #[allow( + unused, + reason = "FIXME(SVLS-8787|github.com/DataDog/libdatadog/pull/2170): implement stats additional tags" + )] + pub const ADDITIONAL_TAGS: usize = 1 << 4; + pub const COUNT: u8 = 5; +} + +const COLLAPSED_FIELD_METRIC_SIZE: usize = 1 << CollapsedField::COUNT; +#[derive(Debug, Clone, Default, Copy)] +pub struct CollapsedFieldsMetrics([usize; COLLAPSED_FIELD_METRIC_SIZE]); + +const_assert!(COLLAPSED_FIELD_METRIC_SIZE <= 32); // Metrics table is of reasonable size + +impl CollapsedFieldsMetrics { + pub(crate) fn zero() -> Self { + Self::default() + } + + #[cfg(feature = "dogstatsd")] + pub fn emit_dogstatsd( + &self, + dogstatsd: &std::sync::Arc, + ) { + for (mask, &count) in self.0.iter().enumerate() { + if count > 0 { + let mut tags = Vec::new(); + for field in 0..CollapsedField::COUNT { + let field_value = 1 << field; + let has_field = (mask | field_value) != 0; + if !has_field { + continue; + } + let field_tag = match field_value { + CollapsedField::RESOURCE_NAME => { + libdd_common::tag!("collapsed_spans", "resource") + } + CollapsedField::HTTP_ENDPOINT => { + libdd_common::tag!("collapsed_spans", "http_endpoint") + } + CollapsedField::PEER_TAGS => { + libdd_common::tag!("collapsed_spans", "peer_tags") + } + CollapsedField::ADDITIONAL_TAGS => { + libdd_common::tag!("collapsed_spans", "additional_metric_tags") + } + #[allow(clippy::unreachable, reason = "I swear it is unreachable")] + _ => unreachable!(), + }; + tags.push(field_tag); + } + dogstatsd.send(vec![libdd_dogstatsd_client::DogStatsDAction::Count( + "datadog.tracer.stats.collapsed_spans", + count as i64, + tags.iter(), + )]); + } + } + } +} + +impl std::ops::AddAssign for CollapsedFieldsMetrics { + fn add_assign(&mut self, rhs: Self) { + for i in 0..self.0.len() { + self.0[i] += rhs.0[i]; + } + } +} + impl StatsBucket { /// Return a new StatsBucket starting at `start_timestamp`. /// @@ -555,9 +632,15 @@ impl StatsBucket { distinct_http_endpoints: HashSet::new(), distinct_peer_tags: HashSet::new(), distinct_additional_tags: HashSet::new(), + collapsed_fields_metrics: CollapsedFieldsMetrics([0; COLLAPSED_FIELD_METRIC_SIZE]), } } + /// Returns metrics on spans field collapse with reasons. + pub fn collapsed_fields_metrics(&self) -> CollapsedFieldsMetrics { + self.collapsed_fields_metrics + } + /// Return the number of spans collapsed into the overflow bucket. pub(super) fn collapsed_count(&self) -> u64 { self.collapsed_count @@ -614,11 +697,14 @@ impl StatsBucket { hasher.finish() } + let mut collapsed_fields = 0; + 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; + collapsed_fields |= CollapsedField::RESOURCE_NAME; } else { slot.insert(); } @@ -629,6 +715,7 @@ impl StatsBucket { 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; + collapsed_fields |= CollapsedField::HTTP_ENDPOINT; } else { slot.insert(); } @@ -639,6 +726,7 @@ impl StatsBucket { 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(""))]; + collapsed_fields |= CollapsedField::PEER_TAGS; } else { slot.insert(); } @@ -653,6 +741,7 @@ impl StatsBucket { slot.insert(); } } + self.collapsed_fields_metrics.0[collapsed_fields] += 1; } /// Consume the bucket and return a ClientStatsBucket containing the bucket stats. diff --git a/libdd-trace-stats/src/span_concentrator/mod.rs b/libdd-trace-stats/src/span_concentrator/mod.rs index 3091dea4ad..a630097059 100644 --- a/libdd-trace-stats/src/span_concentrator/mod.rs +++ b/libdd-trace-stats/src/span_concentrator/mod.rs @@ -9,7 +9,7 @@ use web_time::{SystemTime, UNIX_EPOCH}; use libdd_trace_protobuf::pb; -use aggregation::StatsBucket; +use aggregation::{CollapsedFieldsMetrics, StatsBucket}; mod aggregation; use aggregation::BorrowedAggregationKey; @@ -49,6 +49,7 @@ pub struct FlushResult { /// 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, + pub collapsed_fields_metrics: CollapsedFieldsMetrics, } impl FlushResult { @@ -377,7 +378,8 @@ impl SpanConcentrator { /// /// 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 (buckets, collapsed_spans, collapsed_fields_metrics) = + 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 { @@ -391,6 +393,7 @@ impl SpanConcentrator { obfuscated_buckets, unobfuscated_buckets, collapsed_spans, + collapsed_fields_metrics, } } @@ -398,7 +401,9 @@ impl SpanConcentrator { /// 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); + let buckets = self + .drain_due_buckets(now, force, StatsBucket::flush_with_otlp_exact) + .0; buckets.into_iter().map(|(_, bucket)| bucket).collect() } @@ -413,7 +418,7 @@ impl SpanConcentrator { now: SystemTime, force: bool, encode: impl Fn(StatsBucket, u64) -> T, - ) -> (Vec<(bool, T)>, u64) { + ) -> (Vec<(bool, T)>, u64, CollapsedFieldsMetrics) { // 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(); @@ -424,6 +429,7 @@ impl SpanConcentrator { - (self.buffer_len as u64 - 1) * self.bucket_size }; let mut total_collapsed = 0; + let mut total_collapsed_fields = CollapsedFieldsMetrics::zero(); let buckets_pb = buckets .into_iter() .filter_map(|(timestamp, bucket)| { @@ -441,6 +447,7 @@ impl SpanConcentrator { return None; } total_collapsed += bucket.collapsed_count(); + total_collapsed_fields += bucket.collapsed_fields_metrics(); #[cfg(feature = "stats-obfuscation")] let obfuscated = bucket.obfuscated; #[cfg(not(feature = "stats-obfuscation"))] @@ -455,7 +462,7 @@ impl SpanConcentrator { "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, total_collapsed_fields) } } diff --git a/libdd-trace-stats/src/stats_exporter.rs b/libdd-trace-stats/src/stats_exporter.rs index a7a9efbada..adbe1eace2 100644 --- a/libdd-trace-stats/src/stats_exporter.rs +++ b/libdd-trace-stats/src/stats_exporter.rs @@ -197,6 +197,11 @@ impl< } } + #[cfg(feature = "dogstatsd")] + if let Some(client) = &self.dogstatsd { + flush.collapsed_fields_metrics.emit_dogstatsd(client); + } + let futures = FuturesUnordered::new(); if !flush.obfuscated_buckets.is_empty() { From a8bc23a0a4ca6a4e8cf50b3aef6e011bc591f69e Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Fri, 10 Jul 2026 17:08:02 +0200 Subject: [PATCH 02/15] feat: cherry pick simple 1 collapsed field tests from old pr --- libdd-trace-stats/src/stats_exporter.rs | 37 +++++++++++++++++-------- 1 file changed, 26 insertions(+), 11 deletions(-) diff --git a/libdd-trace-stats/src/stats_exporter.rs b/libdd-trace-stats/src/stats_exporter.rs index adbe1eace2..3375b3366d 100644 --- a/libdd-trace-stats/src/stats_exporter.rs +++ b/libdd-trace-stats/src/stats_exporter.rs @@ -647,7 +647,9 @@ mod tests { /// Build a concentrator with `max_entries_per_bucket = 1` pre-seeded with four distinct spans /// so that three spans are collapsed into the overflow bucket. + #[cfg(any(feature = "telemetry", feature = "dogstatsd"))] fn get_collapsed_concentrator() -> SpanConcentrator { + use crate::span_concentrator::CardinalityLimitConfig; use libdd_trace_utils::span::{trace_utils, v04::SpanSlice}; let mut concentrator = SpanConcentrator::new( @@ -656,7 +658,8 @@ mod tests { vec![], vec![], Some(CardinalityLimitConfig { - whole_key_limit: 1, // max 1 distinct key → second span collapses + whole_key_limit: 2, // max 2 distinct key → third distinct span collapses + resource_limit: 1, // max 1 distinct resource values ..Default::default() }), vec![], @@ -666,26 +669,26 @@ mod tests { let mut trace = vec![ SpanSlice { - service: "svc", + service: "svc-a", resource: "resource-a", duration: 10, ..Default::default() }, SpanSlice { - service: "svc", + service: "svc-a", resource: "resource-b", duration: 20, ..Default::default() }, SpanSlice { - service: "svc", - resource: "resource-c", + service: "svc-b", + resource: "resource-a", duration: 20, ..Default::default() }, SpanSlice { - service: "svc", - resource: "resource-d", + service: "svc-b", + resource: "resource-b", duration: 20, ..Default::default() }, @@ -743,7 +746,8 @@ mod tests { let result = socket.recv(&mut buf); assert!( result.is_err(), - "No DogStatsD datagram expected when collapsed_spans == 0" + "No DogStatsD datagram expected when collapsed_spans == 0. Got {}", + std::str::from_utf8(&buf[..result.unwrap()]).unwrap() ); } @@ -787,12 +791,23 @@ mod tests { stats_exporter.send(true).await.unwrap(); let mut buf = [0u8; 256]; + // Whole-key metric emitted first + let n = socket + .recv(&mut buf) + .expect("expected a DogStatsD datagram"); + let datagram = std::str::from_utf8(&buf[..n]).expect("valid utf-8"); + assert_eq!( + datagram, "datadog.tracer.stats.collapsed_spans:2|c|#collapsed_spans:whole_key", + "DogStatsD datagram must match the expected format" + ); + + // Then comes the per-field collapse metrics let n = socket .recv(&mut buf) .expect("expected a DogStatsD datagram"); let datagram = std::str::from_utf8(&buf[..n]).expect("valid utf-8"); assert_eq!( - datagram, "datadog.tracer.stats.collapsed_spans:3|c|#collapsed_spans:whole_key", + datagram, "datadog.tracer.stats.collapsed_spans:2|c|#collapsed_spans:resource", "DogStatsD datagram must match the expected format" ); } @@ -853,8 +868,8 @@ mod tests { "exactly one metric context (COLLAPSED_SPANS_METRIC) should be registered" ); assert_eq!( - stats.metric_buckets.buckets, 1, - "exactly one metric bucket expected after one collapsed-spans emission" + stats.metric_buckets.buckets, 2, + "exactly one metric bucket expected after collapsing both on whole-key and on the resource field" ); } } From f93f809f2ddcddf8b6f796f731c552f52450d744 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Fri, 10 Jul 2026 17:09:01 +0200 Subject: [PATCH 03/15] fix: bitshift operations are hard... --- .../src/span_concentrator/aggregation.rs | 23 ++++++++++++------- 1 file changed, 15 insertions(+), 8 deletions(-) diff --git a/libdd-trace-stats/src/span_concentrator/aggregation.rs b/libdd-trace-stats/src/span_concentrator/aggregation.rs index dc199ddc37..6628ce5e23 100644 --- a/libdd-trace-stats/src/span_concentrator/aggregation.rs +++ b/libdd-trace-stats/src/span_concentrator/aggregation.rs @@ -553,6 +553,7 @@ impl CollapsedField { const COLLAPSED_FIELD_METRIC_SIZE: usize = 1 << CollapsedField::COUNT; #[derive(Debug, Clone, Default, Copy)] +// note slot 0 is a counter for non_collapsed spans. useless pub struct CollapsedFieldsMetrics([usize; COLLAPSED_FIELD_METRIC_SIZE]); const_assert!(COLLAPSED_FIELD_METRIC_SIZE <= 32); // Metrics table is of reasonable size @@ -563,16 +564,21 @@ impl CollapsedFieldsMetrics { } #[cfg(feature = "dogstatsd")] - pub fn emit_dogstatsd( - &self, - dogstatsd: &std::sync::Arc, - ) { - for (mask, &count) in self.0.iter().enumerate() { + pub fn emit_dogstatsd(&self, dogstatsd: &libdd_dogstatsd_client::DogStatsDClient) { + // skip the first slot that is used to count span which have no collapsed fields. + for (mask, &count) in self.0.iter().enumerate().skip(1) { if count > 0 { let mut tags = Vec::new(); - for field in 0..CollapsedField::COUNT { - let field_value = 1 << field; - let has_field = (mask | field_value) != 0; + for field_pow in 1..CollapsedField::COUNT { + let field_value = 1 << field_pow; + assert!([ + CollapsedField::RESOURCE_NAME, + CollapsedField::HTTP_ENDPOINT, + CollapsedField::PEER_TAGS, + CollapsedField::ADDITIONAL_TAGS + ] + .contains(&field_value)); + let has_field = dbg!(mask & field_value) != 0; if !has_field { continue; } @@ -594,6 +600,7 @@ impl CollapsedFieldsMetrics { }; tags.push(field_tag); } + assert!(!tags.is_empty()); dogstatsd.send(vec![libdd_dogstatsd_client::DogStatsDAction::Count( "datadog.tracer.stats.collapsed_spans", count as i64, From 08ba59cab4e7825b921bddbd01cc3a4a7fb0d941 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Fri, 10 Jul 2026 17:46:08 +0200 Subject: [PATCH 04/15] cleanup --- libdd-trace-stats/src/span_concentrator/aggregation.rs | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/libdd-trace-stats/src/span_concentrator/aggregation.rs b/libdd-trace-stats/src/span_concentrator/aggregation.rs index 6628ce5e23..d968652fee 100644 --- a/libdd-trace-stats/src/span_concentrator/aggregation.rs +++ b/libdd-trace-stats/src/span_concentrator/aggregation.rs @@ -578,7 +578,7 @@ impl CollapsedFieldsMetrics { CollapsedField::ADDITIONAL_TAGS ] .contains(&field_value)); - let has_field = dbg!(mask & field_value) != 0; + let has_field = (mask & field_value) != 0; if !has_field { continue; } @@ -595,7 +595,10 @@ impl CollapsedFieldsMetrics { CollapsedField::ADDITIONAL_TAGS => { libdd_common::tag!("collapsed_spans", "additional_metric_tags") } - #[allow(clippy::unreachable, reason = "I swear it is unreachable")] + #[allow( + clippy::unreachable, + reason = "field pow is between 1..CollapsedField::COUNT, so field_value is a valid CollapsedField value. (Asserted just above)" + )] _ => unreachable!(), }; tags.push(field_tag); @@ -639,7 +642,7 @@ impl StatsBucket { distinct_http_endpoints: HashSet::new(), distinct_peer_tags: HashSet::new(), distinct_additional_tags: HashSet::new(), - collapsed_fields_metrics: CollapsedFieldsMetrics([0; COLLAPSED_FIELD_METRIC_SIZE]), + collapsed_fields_metrics: CollapsedFieldsMetrics::zero(), } } From 8cf0a18d8366ac835ab94245aabff3dadcea00bb Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Tue, 21 Jul 2026 13:16:56 +0200 Subject: [PATCH 05/15] wip telemetry --- datadog-ipc/src/shm_stats.rs | 5 +- .../src/span_concentrator/aggregation.rs | 60 ++++++++++++++++++- .../src/span_concentrator/mod.rs | 6 +- libdd-trace-stats/src/stats_exporter.rs | 27 +++++---- 4 files changed, 80 insertions(+), 18 deletions(-) diff --git a/datadog-ipc/src/shm_stats.rs b/datadog-ipc/src/shm_stats.rs index 87b55981da..951327cace 100644 --- a/datadog-ipc/src/shm_stats.rs +++ b/datadog-ipc/src/shm_stats.rs @@ -57,7 +57,7 @@ use zwohash::ZwoHasher; use libdd_ddsketch::DDSketch; use libdd_trace_protobuf::pb; use libdd_trace_stats::span_concentrator::{ - FixedAggregationKey, FlushResult, FlushableConcentrator, + CollapsedFieldsMetrics, FixedAggregationKey, FlushResult, FlushableConcentrator, }; use crate::platform::{FileBackedHandle, MappedMem, NamedShmHandle}; @@ -823,11 +823,12 @@ impl ShmSpanConcentrator { impl FlushableConcentrator for ShmSpanConcentrator { fn flush_buckets(&mut self, force: bool) -> FlushResult { - // The SHM concentrator does not perform client-side obfuscation. + // The SHM concentrator does not perform client-side obfuscation nor emits telemetry. FlushResult { obfuscated_buckets: vec![], unobfuscated_buckets: self.drain_buckets(force), collapsed_spans: 0, + collapsed_fields_metrics: CollapsedFieldsMetrics::zero(), } } } diff --git a/libdd-trace-stats/src/span_concentrator/aggregation.rs b/libdd-trace-stats/src/span_concentrator/aggregation.rs index d968652fee..dae8437ec8 100644 --- a/libdd-trace-stats/src/span_concentrator/aggregation.rs +++ b/libdd-trace-stats/src/span_concentrator/aggregation.rs @@ -553,13 +553,13 @@ impl CollapsedField { const COLLAPSED_FIELD_METRIC_SIZE: usize = 1 << CollapsedField::COUNT; #[derive(Debug, Clone, Default, Copy)] -// note slot 0 is a counter for non_collapsed spans. useless +// Note: slot 0 is a counter for non_collapsed spans. It's not used for emitting telemetry pub struct CollapsedFieldsMetrics([usize; COLLAPSED_FIELD_METRIC_SIZE]); const_assert!(COLLAPSED_FIELD_METRIC_SIZE <= 32); // Metrics table is of reasonable size impl CollapsedFieldsMetrics { - pub(crate) fn zero() -> Self { + pub fn zero() -> Self { Self::default() } @@ -612,6 +612,62 @@ impl CollapsedFieldsMetrics { } } } + + #[cfg(feature = "telemetry")] + pub fn emit_telemetry< + Cap: libdd_capabilities::HttpClientCapability + + libdd_capabilities::SleepCapability + + libdd_capabilities::MaybeSend + + Sync + + 'static, + >( + &self, + handle: &libdd_telemetry::worker::TelemetryWorkerHandle, + context_key: &libdd_telemetry::metrics::ContextKey, + ) { + // skip the first slot that is used to count span which have no collapsed fields. + for (mask, &count) in self.0.iter().enumerate().skip(1) { + if count > 0 { + let mut tags = Vec::new(); + for field_pow in 1..CollapsedField::COUNT { + let field_value = 1 << field_pow; + assert!([ + CollapsedField::RESOURCE_NAME, + CollapsedField::HTTP_ENDPOINT, + CollapsedField::PEER_TAGS, + CollapsedField::ADDITIONAL_TAGS + ] + .contains(&field_value)); + let has_field = (mask & field_value) != 0; + if !has_field { + continue; + } + let field_tag = match field_value { + CollapsedField::RESOURCE_NAME => { + libdd_common::tag!("collapsed_spans", "resource") + } + CollapsedField::HTTP_ENDPOINT => { + libdd_common::tag!("collapsed_spans", "http_endpoint") + } + CollapsedField::PEER_TAGS => { + libdd_common::tag!("collapsed_spans", "peer_tags") + } + CollapsedField::ADDITIONAL_TAGS => { + libdd_common::tag!("collapsed_spans", "additional_metric_tags") + } + #[allow( + clippy::unreachable, + reason = "field pow is between 1..CollapsedField::COUNT, so field_value is a valid CollapsedField value. (Asserted just above)" + )] + _ => unreachable!(), + }; + tags.push(field_tag); + } + assert!(!tags.is_empty()); + let _ = handle.add_point(count as f64, context_key, tags); + } + } + } } impl std::ops::AddAssign for CollapsedFieldsMetrics { diff --git a/libdd-trace-stats/src/span_concentrator/mod.rs b/libdd-trace-stats/src/span_concentrator/mod.rs index a630097059..f8740eb6fb 100644 --- a/libdd-trace-stats/src/span_concentrator/mod.rs +++ b/libdd-trace-stats/src/span_concentrator/mod.rs @@ -9,11 +9,13 @@ use web_time::{SystemTime, UNIX_EPOCH}; use libdd_trace_protobuf::pb; -use aggregation::{CollapsedFieldsMetrics, StatsBucket}; +use aggregation::StatsBucket; mod aggregation; use aggregation::BorrowedAggregationKey; -pub use aggregation::{FixedAggregationKey, OtlpExactCell, OtlpExactGroup, OtlpStatsBucket}; +pub use aggregation::{ + CollapsedFieldsMetrics, FixedAggregationKey, OtlpExactCell, OtlpExactGroup, OtlpStatsBucket, +}; pub mod stat_span; pub use stat_span::StatSpan; diff --git a/libdd-trace-stats/src/stats_exporter.rs b/libdd-trace-stats/src/stats_exporter.rs index 3375b3366d..c709f3173f 100644 --- a/libdd-trace-stats/src/stats_exporter.rs +++ b/libdd-trace-stats/src/stats_exporter.rs @@ -137,7 +137,7 @@ impl< let telemetry = telemetry.map(|handle| { let key = handle.register_metric_context( COLLAPSED_SPANS_TELEMETRY_METRIC.to_string(), - vec![libdd_common::tag!("collapsed_spans", "whole_key")], + vec![], libdd_telemetry::data::metrics::MetricType::Count, true, libdd_telemetry::data::metrics::MetricNamespace::Tracers, @@ -182,23 +182,27 @@ impl< concentrator.flush_buckets(force_flush) }; - 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![]); + #[cfg(feature = "telemetry")] + if let Some((handle, key)) = &self.telemetry { + if flush.collapsed_spans > 0 { + let _ = handle.add_point( + flush.collapsed_spans as f64, + key, + vec![libdd_common::tag!("collapsed_spans", "whole_key")], + ); } - #[cfg(feature = "dogstatsd")] - if let Some(client) = &self.dogstatsd { + flush.collapsed_fields_metrics.emit_telemetry(handle, key); + } + + #[cfg(feature = "dogstatsd")] + if let Some(client) = &self.dogstatsd { + if flush.collapsed_spans > 0 { 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(feature = "dogstatsd")] - if let Some(client) = &self.dogstatsd { flush.collapsed_fields_metrics.emit_dogstatsd(client); } @@ -334,7 +338,6 @@ pub fn stats_url_from_agent_url(agent_url: &str) -> anyhow::Result { #[cfg(test)] mod tests { use super::*; - use crate::span_concentrator::CardinalityLimitConfig; use httpmock::prelude::*; use httpmock::MockServer; use libdd_capabilities_impl::NativeCapabilities; From 8c4dc36d6f8aecd5b78764bf279c4daf73e3e924 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Tue, 21 Jul 2026 14:01:23 +0200 Subject: [PATCH 06/15] add tests for multiple per-key collapse --- libdd-trace-stats/src/stats_exporter.rs | 104 ++++++++++++++++++++---- 1 file changed, 90 insertions(+), 14 deletions(-) diff --git a/libdd-trace-stats/src/stats_exporter.rs b/libdd-trace-stats/src/stats_exporter.rs index c709f3173f..857187c33f 100644 --- a/libdd-trace-stats/src/stats_exporter.rs +++ b/libdd-trace-stats/src/stats_exporter.rs @@ -650,21 +650,31 @@ mod tests { /// Build a concentrator with `max_entries_per_bucket = 1` pre-seeded with four distinct spans /// so that three spans are collapsed into the overflow bucket. + /// + /// per_key_collapsed: enable small per-key limits to also collapse on `resource` and + /// `http_endpoint` #[cfg(any(feature = "telemetry", feature = "dogstatsd"))] - fn get_collapsed_concentrator() -> SpanConcentrator { + fn get_collapsed_concentrator(per_key_collapsed: bool) -> SpanConcentrator { use crate::span_concentrator::CardinalityLimitConfig; - use libdd_trace_utils::span::{trace_utils, v04::SpanSlice}; + use libdd_trace_utils::span::{ + trace_utils, + v04::{SpanSlice, VecMap}, + }; + let mut cardinality_limit_config = CardinalityLimitConfig { + whole_key_limit: 2, // max 2 distinct key → third distinct span collapses + ..Default::default() + }; + if per_key_collapsed { + cardinality_limit_config.resource_limit = 1; + cardinality_limit_config.http_endpoint_limit = 1; + } let mut concentrator = SpanConcentrator::new( BUCKETS_DURATION, SystemTime::now(), vec![], vec![], - Some(CardinalityLimitConfig { - whole_key_limit: 2, // max 2 distinct key → third distinct span collapses - resource_limit: 1, // max 1 distinct resource values - ..Default::default() - }), + Some(cardinality_limit_config), vec![], #[cfg(feature = "stats-obfuscation")] None, @@ -675,20 +685,26 @@ mod tests { service: "svc-a", resource: "resource-a", duration: 10, + meta: VecMap::from_iter([("http.endpoint", "/")]), ..Default::default() }, + // only resource get collapsed if per-key limits are enabled SpanSlice { service: "svc-a", resource: "resource-b", duration: 20, + meta: VecMap::from_iter([("http.endpoint", "/")]), ..Default::default() }, + // both resource and http_endpoint get collapsed if per-key limits are enabled SpanSlice { service: "svc-b", - resource: "resource-a", + resource: "resource-c", duration: 20, + meta: VecMap::from_iter([("http.endpoint", "/hello.txt")]), ..Default::default() }, + // both resource and http_endpoint get collapsed if per-key limits are enabled SpanSlice { service: "svc-b", resource: "resource-b", @@ -780,7 +796,7 @@ mod tests { let stats_exporter = StatsExporter::::new( BUCKETS_DURATION, - Arc::new(Mutex::new(get_collapsed_concentrator())), + Arc::new(Mutex::new(get_collapsed_concentrator(false))), get_test_metadata(), Endpoint::from_url(stats_url_from_agent_url(&server.url("/")).unwrap()), NativeCapabilities::new_client(), @@ -794,6 +810,58 @@ mod tests { stats_exporter.send(true).await.unwrap(); let mut buf = [0u8; 256]; + let n = socket + .recv(&mut buf) + .expect("expected a DogStatsD datagram"); + let datagram = std::str::from_utf8(&buf[..n]).expect("valid utf-8"); + assert_eq!( + datagram, "datadog.tracer.stats.collapsed_spans:2|c|#collapsed_spans:whole_key", + "DogStatsD datagram must match the expected format" + ); + } + + /// Verify that `COLLAPSED_SPANS_METRIC` is emitted to DogStatsD when spans are collapsed by + /// per-key limits. + #[cfg(feature = "dogstatsd")] + #[cfg_attr(miri, ignore)] + #[tokio::test] + async fn test_collapsed_spans_per_key_dogstatsd() { + use std::net; + + let server = MockServer::start_async().await; + server + .mock_async(|_when, then| { + then.status(200).body(""); + }) + .await; + + let socket = net::UdpSocket::bind("127.0.0.1:0").expect("failed to bind UDP socket"); + socket + .set_read_timeout(Some(std::time::Duration::from_millis(500))) + .unwrap(); + let addr = socket.local_addr().unwrap().to_string(); + + let dogstatsd_client = + libdd_dogstatsd_client::DogStatsDClient::new(libdd_common::Endpoint::from_slice(&addr)) + .expect("failed to create dogstatsd client"); + + let stats_exporter = StatsExporter::::new( + BUCKETS_DURATION, + Arc::new(Mutex::new(get_collapsed_concentrator(true))), + get_test_metadata(), + Endpoint::from_url(stats_url_from_agent_url(&server.url("/")).unwrap()), + NativeCapabilities::new_client(), + #[cfg(feature = "stats-obfuscation")] + "1", + #[cfg(feature = "telemetry")] + None, + Some(dogstatsd_client), + ); + + stats_exporter.send(true).await.unwrap(); + + let mut buf = [0u8; 256]; + // Whole-key metric emitted first let n = socket .recv(&mut buf) @@ -804,13 +872,21 @@ mod tests { "DogStatsD datagram must match the expected format" ); - // Then comes the per-field collapse metrics + // Then comes the per-key collapse telemetry + let n = socket + .recv(&mut buf) + .expect("expected a DogStatsD datagram"); + let datagram = std::str::from_utf8(&buf[..n]).expect("valid utf-8"); + assert_eq!( + datagram, "datadog.tracer.stats.collapsed_spans:1|c|#collapsed_spans:resource", + "DogStatsD datagram must match the expected format" + ); let n = socket .recv(&mut buf) .expect("expected a DogStatsD datagram"); let datagram = std::str::from_utf8(&buf[..n]).expect("valid utf-8"); assert_eq!( - datagram, "datadog.tracer.stats.collapsed_spans:2|c|#collapsed_spans:resource", + datagram, "datadog.tracer.stats.collapsed_spans:2|c|#collapsed_spans:resource,collapsed_spans:http_endpoint", "DogStatsD datagram must match the expected format" ); } @@ -841,7 +917,7 @@ mod tests { let stats_exporter = StatsExporter::::new( BUCKETS_DURATION, - Arc::new(Mutex::new(get_collapsed_concentrator())), + Arc::new(Mutex::new(get_collapsed_concentrator(true))), get_test_metadata(), Endpoint::from_url(stats_url_from_agent_url(&server.url("/")).unwrap()), NativeCapabilities::new_client(), @@ -871,8 +947,8 @@ mod tests { "exactly one metric context (COLLAPSED_SPANS_METRIC) should be registered" ); assert_eq!( - stats.metric_buckets.buckets, 2, - "exactly one metric bucket expected after collapsing both on whole-key and on the resource field" + stats.metric_buckets.buckets, 3, + "exactly 3 metric bucket expected after one whole-key collapsed-spans, one resource key collapsed-span and one resource key+http_endpoint key emissions" ); } } From 141be304dac9962552d3f0a4d52ba9f853068c05 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Tue, 21 Jul 2026 14:21:41 +0200 Subject: [PATCH 07/15] refacto: extract telemetry stuff into its own module feat: doc new module --- datadog-ipc/src/shm_stats.rs | 3 +- .../src/span_concentrator/aggregation.rs | 159 ++---------------- .../cardinality_limit_telemetry.rs | 119 +++++++++++++ .../src/span_concentrator/mod.rs | 11 +- 4 files changed, 137 insertions(+), 155 deletions(-) create mode 100644 libdd-trace-stats/src/span_concentrator/cardinality_limit_telemetry.rs diff --git a/datadog-ipc/src/shm_stats.rs b/datadog-ipc/src/shm_stats.rs index 951327cace..38b6dfd059 100644 --- a/datadog-ipc/src/shm_stats.rs +++ b/datadog-ipc/src/shm_stats.rs @@ -57,7 +57,8 @@ use zwohash::ZwoHasher; use libdd_ddsketch::DDSketch; use libdd_trace_protobuf::pb; use libdd_trace_stats::span_concentrator::{ - CollapsedFieldsMetrics, FixedAggregationKey, FlushResult, FlushableConcentrator, + cardinality_limit_telemetry::CollapsedFieldsMetrics, FixedAggregationKey, FlushResult, + FlushableConcentrator, }; use crate::platform::{FileBackedHandle, MappedMem, NamedShmHandle}; diff --git a/libdd-trace-stats/src/span_concentrator/aggregation.rs b/libdd-trace-stats/src/span_concentrator/aggregation.rs index dae8437ec8..0c09d8a203 100644 --- a/libdd-trace-stats/src/span_concentrator/aggregation.rs +++ b/libdd-trace-stats/src/span_concentrator/aggregation.rs @@ -6,7 +6,6 @@ //! span. use hashbrown::{HashMap, HashSet}; -use libdd_common::tag::const_assert; use libdd_trace_obfuscation::ip_address::quantize_peer_ip_addresses; use libdd_trace_protobuf::pb; use libdd_trace_utils::span::SpanText; @@ -16,9 +15,12 @@ use std::{ }; use tracing::warn; -use crate::span_concentrator::StatSpan; +use crate::span_concentrator::{cardinality_limit_telemetry::collapsed_field, StatSpan}; -use super::CardinalityLimitConfig; +use super::{ + cardinality_limit_telemetry::{self, CollapsedFieldsMetrics}, + CardinalityLimitConfig, +}; /// Sentinel value used for cardinality limiting. pub const TRACER_BLOCKED_VALUE: &str = "tracer_blocked_value"; @@ -537,147 +539,6 @@ pub(super) struct StatsBucket { pub(super) obfuscated: bool, } -#[repr(transparent)] -pub struct CollapsedField; -impl CollapsedField { - pub const RESOURCE_NAME: usize = 1 << 1; - pub const HTTP_ENDPOINT: usize = 1 << 2; - pub const PEER_TAGS: usize = 1 << 3; - #[allow( - unused, - reason = "FIXME(SVLS-8787|github.com/DataDog/libdatadog/pull/2170): implement stats additional tags" - )] - pub const ADDITIONAL_TAGS: usize = 1 << 4; - pub const COUNT: u8 = 5; -} - -const COLLAPSED_FIELD_METRIC_SIZE: usize = 1 << CollapsedField::COUNT; -#[derive(Debug, Clone, Default, Copy)] -// Note: slot 0 is a counter for non_collapsed spans. It's not used for emitting telemetry -pub struct CollapsedFieldsMetrics([usize; COLLAPSED_FIELD_METRIC_SIZE]); - -const_assert!(COLLAPSED_FIELD_METRIC_SIZE <= 32); // Metrics table is of reasonable size - -impl CollapsedFieldsMetrics { - pub fn zero() -> Self { - Self::default() - } - - #[cfg(feature = "dogstatsd")] - pub fn emit_dogstatsd(&self, dogstatsd: &libdd_dogstatsd_client::DogStatsDClient) { - // skip the first slot that is used to count span which have no collapsed fields. - for (mask, &count) in self.0.iter().enumerate().skip(1) { - if count > 0 { - let mut tags = Vec::new(); - for field_pow in 1..CollapsedField::COUNT { - let field_value = 1 << field_pow; - assert!([ - CollapsedField::RESOURCE_NAME, - CollapsedField::HTTP_ENDPOINT, - CollapsedField::PEER_TAGS, - CollapsedField::ADDITIONAL_TAGS - ] - .contains(&field_value)); - let has_field = (mask & field_value) != 0; - if !has_field { - continue; - } - let field_tag = match field_value { - CollapsedField::RESOURCE_NAME => { - libdd_common::tag!("collapsed_spans", "resource") - } - CollapsedField::HTTP_ENDPOINT => { - libdd_common::tag!("collapsed_spans", "http_endpoint") - } - CollapsedField::PEER_TAGS => { - libdd_common::tag!("collapsed_spans", "peer_tags") - } - CollapsedField::ADDITIONAL_TAGS => { - libdd_common::tag!("collapsed_spans", "additional_metric_tags") - } - #[allow( - clippy::unreachable, - reason = "field pow is between 1..CollapsedField::COUNT, so field_value is a valid CollapsedField value. (Asserted just above)" - )] - _ => unreachable!(), - }; - tags.push(field_tag); - } - assert!(!tags.is_empty()); - dogstatsd.send(vec![libdd_dogstatsd_client::DogStatsDAction::Count( - "datadog.tracer.stats.collapsed_spans", - count as i64, - tags.iter(), - )]); - } - } - } - - #[cfg(feature = "telemetry")] - pub fn emit_telemetry< - Cap: libdd_capabilities::HttpClientCapability - + libdd_capabilities::SleepCapability - + libdd_capabilities::MaybeSend - + Sync - + 'static, - >( - &self, - handle: &libdd_telemetry::worker::TelemetryWorkerHandle, - context_key: &libdd_telemetry::metrics::ContextKey, - ) { - // skip the first slot that is used to count span which have no collapsed fields. - for (mask, &count) in self.0.iter().enumerate().skip(1) { - if count > 0 { - let mut tags = Vec::new(); - for field_pow in 1..CollapsedField::COUNT { - let field_value = 1 << field_pow; - assert!([ - CollapsedField::RESOURCE_NAME, - CollapsedField::HTTP_ENDPOINT, - CollapsedField::PEER_TAGS, - CollapsedField::ADDITIONAL_TAGS - ] - .contains(&field_value)); - let has_field = (mask & field_value) != 0; - if !has_field { - continue; - } - let field_tag = match field_value { - CollapsedField::RESOURCE_NAME => { - libdd_common::tag!("collapsed_spans", "resource") - } - CollapsedField::HTTP_ENDPOINT => { - libdd_common::tag!("collapsed_spans", "http_endpoint") - } - CollapsedField::PEER_TAGS => { - libdd_common::tag!("collapsed_spans", "peer_tags") - } - CollapsedField::ADDITIONAL_TAGS => { - libdd_common::tag!("collapsed_spans", "additional_metric_tags") - } - #[allow( - clippy::unreachable, - reason = "field pow is between 1..CollapsedField::COUNT, so field_value is a valid CollapsedField value. (Asserted just above)" - )] - _ => unreachable!(), - }; - tags.push(field_tag); - } - assert!(!tags.is_empty()); - let _ = handle.add_point(count as f64, context_key, tags); - } - } - } -} - -impl std::ops::AddAssign for CollapsedFieldsMetrics { - fn add_assign(&mut self, rhs: Self) { - for i in 0..self.0.len() { - self.0[i] += rhs.0[i]; - } - } -} - impl StatsBucket { /// Return a new StatsBucket starting at `start_timestamp`. /// @@ -698,12 +559,12 @@ impl StatsBucket { distinct_http_endpoints: HashSet::new(), distinct_peer_tags: HashSet::new(), distinct_additional_tags: HashSet::new(), - collapsed_fields_metrics: CollapsedFieldsMetrics::zero(), + collapsed_fields_metrics: cardinality_limit_telemetry::CollapsedFieldsMetrics::zero(), } } /// Returns metrics on spans field collapse with reasons. - pub fn collapsed_fields_metrics(&self) -> CollapsedFieldsMetrics { + pub fn collapsed_fields_metrics(&self) -> cardinality_limit_telemetry::CollapsedFieldsMetrics { self.collapsed_fields_metrics } @@ -770,7 +631,7 @@ impl StatsBucket { 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; - collapsed_fields |= CollapsedField::RESOURCE_NAME; + collapsed_fields |= collapsed_field::RESOURCE_NAME; } else { slot.insert(); } @@ -781,7 +642,7 @@ impl StatsBucket { 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; - collapsed_fields |= CollapsedField::HTTP_ENDPOINT; + collapsed_fields |= collapsed_field::HTTP_ENDPOINT; } else { slot.insert(); } @@ -792,7 +653,7 @@ impl StatsBucket { 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(""))]; - collapsed_fields |= CollapsedField::PEER_TAGS; + collapsed_fields |= collapsed_field::PEER_TAGS; } else { slot.insert(); } diff --git a/libdd-trace-stats/src/span_concentrator/cardinality_limit_telemetry.rs b/libdd-trace-stats/src/span_concentrator/cardinality_limit_telemetry.rs new file mode 100644 index 0000000000..e5e65cfb16 --- /dev/null +++ b/libdd-trace-stats/src/span_concentrator/cardinality_limit_telemetry.rs @@ -0,0 +1,119 @@ +// Copyright 2024-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +//! Each collapsed field is assigned a bit; a span's mask is the OR of the bits for the +//! fields collapsed on it, so masks double as an index into +//! [`CollapsedFieldsMetrics`]'s counter array (one counter per distinct field combination). + +#[cfg(feature = "telemetry")] +use libdd_capabilities::{HttpClientCapability, MaybeSend, SleepCapability}; +use libdd_common::tag::const_assert; + +pub(super) mod collapsed_field { + pub const RESOURCE_NAME: usize = 1 << 1; + pub const HTTP_ENDPOINT: usize = 1 << 2; + pub const PEER_TAGS: usize = 1 << 3; + #[allow( + unused, + reason = "FIXME(SVLS-8787|github.com/DataDog/libdatadog/pull/2170): implement stats additional tags" + )] + pub const ADDITIONAL_TAGS: usize = 1 << 4; + pub const COUNT: u8 = 5; +} + +pub(super) const COLLAPSED_FIELD_METRIC_SIZE: usize = 1 << collapsed_field::COUNT; + +#[derive(Debug, Clone, Default, Copy)] +// Note: slot 0 is a counter for non_collapsed spans. It's not used for emitting telemetry +pub struct CollapsedFieldsMetrics(pub(super) [usize; COLLAPSED_FIELD_METRIC_SIZE]); + +const_assert!(COLLAPSED_FIELD_METRIC_SIZE <= 32); + +impl CollapsedFieldsMetrics { + pub fn zero() -> Self { + Self::default() + } + + #[cfg(feature = "dogstatsd")] + pub fn emit_dogstatsd(&self, dogstatsd: &libdd_dogstatsd_client::DogStatsDClient) { + // skip the first slot that is used to count span which have no collapsed fields. + for (mask, &count) in self.0.iter().enumerate().skip(1) { + if count > 0 { + let tags = Self::fields_mask_to_list(mask); + dogstatsd.send(vec![libdd_dogstatsd_client::DogStatsDAction::Count( + "datadog.tracer.stats.collapsed_spans", + count as i64, + tags.iter(), + )]); + } + } + } + + #[cfg(feature = "telemetry")] + pub fn emit_telemetry< + Cap: HttpClientCapability + SleepCapability + MaybeSend + Sync + 'static, + >( + &self, + handle: &libdd_telemetry::worker::TelemetryWorkerHandle, + context_key: &libdd_telemetry::metrics::ContextKey, + ) { + // skip the first slot that is used to count span which have no collapsed fields. + for (mask, &count) in self.0.iter().enumerate().skip(1) { + if count > 0 { + let tags = Self::fields_mask_to_list(mask); + let _ = handle.add_point(count as f64, context_key, tags); + } + } + } + + /// Given a bitmask of collapsed fields, returns the list of tags to attach to + /// telemetry/dogstatsd + #[cfg(any(feature = "telemetry", feature = "dogstatsd"))] + fn fields_mask_to_list(mask: usize) -> Vec { + let mut tags = Vec::new(); + for field_pow in 1..collapsed_field::COUNT { + let field_value = 1 << field_pow; + assert!([ + collapsed_field::RESOURCE_NAME, + collapsed_field::HTTP_ENDPOINT, + collapsed_field::PEER_TAGS, + collapsed_field::ADDITIONAL_TAGS + ] + .contains(&field_value)); + let has_field = (mask & field_value) != 0; + if !has_field { + continue; + } + let field_tag = match field_value { + collapsed_field::RESOURCE_NAME => { + libdd_common::tag!("collapsed_spans", "resource") + } + collapsed_field::HTTP_ENDPOINT => { + libdd_common::tag!("collapsed_spans", "http_endpoint") + } + collapsed_field::PEER_TAGS => { + libdd_common::tag!("collapsed_spans", "peer_tags") + } + collapsed_field::ADDITIONAL_TAGS => { + libdd_common::tag!("collapsed_spans", "additional_metric_tags") + } + #[allow( + clippy::unreachable, + reason = "field pow is between 1..CollapsedField::COUNT, so field_value is a valid CollapsedField value. (Asserted just above)" + )] + _ => unreachable!(), + }; + tags.push(field_tag); + } + assert!(!tags.is_empty()); + tags + } +} + +impl std::ops::AddAssign for CollapsedFieldsMetrics { + fn add_assign(&mut self, rhs: Self) { + for i in 0..self.0.len() { + self.0[i] += rhs.0[i]; + } + } +} diff --git a/libdd-trace-stats/src/span_concentrator/mod.rs b/libdd-trace-stats/src/span_concentrator/mod.rs index f8740eb6fb..aafc2a871d 100644 --- a/libdd-trace-stats/src/span_concentrator/mod.rs +++ b/libdd-trace-stats/src/span_concentrator/mod.rs @@ -1,6 +1,10 @@ // Copyright 2024-Present Datadog, Inc. https://www.datadoghq.com/ // SPDX-License-Identifier: Apache-2.0 //! This module implements the SpanConcentrator used to aggregate spans into stats +mod aggregation; +pub mod cardinality_limit_telemetry; +pub mod stat_span; + use std::collections::HashMap; use std::time::Duration; use tracing::{debug, warn}; @@ -11,13 +15,10 @@ use libdd_trace_protobuf::pb; use aggregation::StatsBucket; -mod aggregation; use aggregation::BorrowedAggregationKey; -pub use aggregation::{ - CollapsedFieldsMetrics, FixedAggregationKey, OtlpExactCell, OtlpExactGroup, OtlpStatsBucket, -}; +pub use aggregation::{FixedAggregationKey, OtlpExactCell, OtlpExactGroup, OtlpStatsBucket}; +use cardinality_limit_telemetry::CollapsedFieldsMetrics; -pub mod stat_span; pub use stat_span::StatSpan; const ADDITIONAL_METRIC_TAGS_MAX_KEYS: usize = 4; From 83ac8b12c2170a8a1d43d31fc0a5e2994c84058d Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Tue, 21 Jul 2026 14:54:22 +0200 Subject: [PATCH 08/15] fix: assert -> debug_assert, avoid useless unreachable! --- .../src/span_concentrator/cardinality_limit_telemetry.rs | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/libdd-trace-stats/src/span_concentrator/cardinality_limit_telemetry.rs b/libdd-trace-stats/src/span_concentrator/cardinality_limit_telemetry.rs index e5e65cfb16..8a3f95ae17 100644 --- a/libdd-trace-stats/src/span_concentrator/cardinality_limit_telemetry.rs +++ b/libdd-trace-stats/src/span_concentrator/cardinality_limit_telemetry.rs @@ -73,7 +73,7 @@ impl CollapsedFieldsMetrics { let mut tags = Vec::new(); for field_pow in 1..collapsed_field::COUNT { let field_value = 1 << field_pow; - assert!([ + debug_assert!([ collapsed_field::RESOURCE_NAME, collapsed_field::HTTP_ENDPOINT, collapsed_field::PEER_TAGS, @@ -97,11 +97,8 @@ impl CollapsedFieldsMetrics { collapsed_field::ADDITIONAL_TAGS => { libdd_common::tag!("collapsed_spans", "additional_metric_tags") } - #[allow( - clippy::unreachable, - reason = "field pow is between 1..CollapsedField::COUNT, so field_value is a valid CollapsedField value. (Asserted just above)" - )] - _ => unreachable!(), + // Unreachable: asserted just above that field is one of the 4 possible values + _ => continue, }; tags.push(field_tag); } From f0dfddbd5ea1535174bc32d0547a91c1272e8dda Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Wed, 22 Jul 2026 13:01:53 +0200 Subject: [PATCH 09/15] feat: telemetry for additional metric tags collapse --- libdd-trace-stats/src/span_concentrator/aggregation.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/libdd-trace-stats/src/span_concentrator/aggregation.rs b/libdd-trace-stats/src/span_concentrator/aggregation.rs index 0c09d8a203..1907f13018 100644 --- a/libdd-trace-stats/src/span_concentrator/aggregation.rs +++ b/libdd-trace-stats/src/span_concentrator/aggregation.rs @@ -664,6 +664,7 @@ impl StatsBucket { if let Entry::Vacant(slot) = self.distinct_additional_tags.entry(additional_tags_hash) { if additional_tags_count >= self.cardinality_limits.additional_tags_limit { key.additional_metric_tags = vec![(TRACER_BLOCKED_VALUE, "")]; + collapsed_fields |= collapsed_field::ADDITIONAL_TAGS; } else { slot.insert(); } From 57a4a837e73021d29e4f9081aaf970215b62f63e Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Wed, 22 Jul 2026 13:16:54 +0200 Subject: [PATCH 10/15] remove old fixme --- .../src/span_concentrator/cardinality_limit_telemetry.rs | 4 ---- 1 file changed, 4 deletions(-) diff --git a/libdd-trace-stats/src/span_concentrator/cardinality_limit_telemetry.rs b/libdd-trace-stats/src/span_concentrator/cardinality_limit_telemetry.rs index 8a3f95ae17..e1edce006a 100644 --- a/libdd-trace-stats/src/span_concentrator/cardinality_limit_telemetry.rs +++ b/libdd-trace-stats/src/span_concentrator/cardinality_limit_telemetry.rs @@ -13,10 +13,6 @@ pub(super) mod collapsed_field { pub const RESOURCE_NAME: usize = 1 << 1; pub const HTTP_ENDPOINT: usize = 1 << 2; pub const PEER_TAGS: usize = 1 << 3; - #[allow( - unused, - reason = "FIXME(SVLS-8787|github.com/DataDog/libdatadog/pull/2170): implement stats additional tags" - )] pub const ADDITIONAL_TAGS: usize = 1 << 4; pub const COUNT: u8 = 5; } From 17a23546f2baf7dabaa7b2e936ade8e06aae3a1b Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Thu, 23 Jul 2026 11:51:00 +0200 Subject: [PATCH 11/15] fix: another assert -> debug_assert --- .../src/span_concentrator/cardinality_limit_telemetry.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/libdd-trace-stats/src/span_concentrator/cardinality_limit_telemetry.rs b/libdd-trace-stats/src/span_concentrator/cardinality_limit_telemetry.rs index e1edce006a..82366e076b 100644 --- a/libdd-trace-stats/src/span_concentrator/cardinality_limit_telemetry.rs +++ b/libdd-trace-stats/src/span_concentrator/cardinality_limit_telemetry.rs @@ -98,7 +98,7 @@ impl CollapsedFieldsMetrics { }; tags.push(field_tag); } - assert!(!tags.is_empty()); + debug_assert!(!tags.is_empty()); tags } } From 9b34da9591d8559278758729cd5cab6b200962c4 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Tue, 28 Jul 2026 14:24:33 +0200 Subject: [PATCH 12/15] refacto: make telemetry field set api nicer + fix not using the 1<<0 (1) flag --- .../src/span_concentrator/aggregation.rs | 14 ++--- .../cardinality_limit_telemetry.rs | 63 ++++++++++++------- 2 files changed, 47 insertions(+), 30 deletions(-) diff --git a/libdd-trace-stats/src/span_concentrator/aggregation.rs b/libdd-trace-stats/src/span_concentrator/aggregation.rs index 1907f13018..588e5ca7b6 100644 --- a/libdd-trace-stats/src/span_concentrator/aggregation.rs +++ b/libdd-trace-stats/src/span_concentrator/aggregation.rs @@ -15,7 +15,7 @@ use std::{ }; use tracing::warn; -use crate::span_concentrator::{cardinality_limit_telemetry::collapsed_field, StatSpan}; +use crate::span_concentrator::{cardinality_limit_telemetry::CollapsedFieldSet, StatSpan}; use super::{ cardinality_limit_telemetry::{self, CollapsedFieldsMetrics}, @@ -624,14 +624,14 @@ impl StatsBucket { hasher.finish() } - let mut collapsed_fields = 0; + let mut collapsed_fields = CollapsedFieldSet::empty(); 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; - collapsed_fields |= collapsed_field::RESOURCE_NAME; + collapsed_fields.add(CollapsedFieldSet::RESOURCE_NAME); } else { slot.insert(); } @@ -642,7 +642,7 @@ impl StatsBucket { 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; - collapsed_fields |= collapsed_field::HTTP_ENDPOINT; + collapsed_fields.add(CollapsedFieldSet::HTTP_ENDPOINT); } else { slot.insert(); } @@ -653,7 +653,7 @@ impl StatsBucket { 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(""))]; - collapsed_fields |= collapsed_field::PEER_TAGS; + collapsed_fields.add(CollapsedFieldSet::PEER_TAGS); } else { slot.insert(); } @@ -664,12 +664,12 @@ impl StatsBucket { if let Entry::Vacant(slot) = self.distinct_additional_tags.entry(additional_tags_hash) { if additional_tags_count >= self.cardinality_limits.additional_tags_limit { key.additional_metric_tags = vec![(TRACER_BLOCKED_VALUE, "")]; - collapsed_fields |= collapsed_field::ADDITIONAL_TAGS; + collapsed_fields.add(CollapsedFieldSet::ADDITIONAL_TAGS); } else { slot.insert(); } } - self.collapsed_fields_metrics.0[collapsed_fields] += 1; + self.collapsed_fields_metrics.increment(collapsed_fields); } /// Consume the bucket and return a ClientStatsBucket containing the bucket stats. diff --git a/libdd-trace-stats/src/span_concentrator/cardinality_limit_telemetry.rs b/libdd-trace-stats/src/span_concentrator/cardinality_limit_telemetry.rs index 82366e076b..6482374a4f 100644 --- a/libdd-trace-stats/src/span_concentrator/cardinality_limit_telemetry.rs +++ b/libdd-trace-stats/src/span_concentrator/cardinality_limit_telemetry.rs @@ -9,21 +9,37 @@ use libdd_capabilities::{HttpClientCapability, MaybeSend, SleepCapability}; use libdd_common::tag::const_assert; -pub(super) mod collapsed_field { - pub const RESOURCE_NAME: usize = 1 << 1; - pub const HTTP_ENDPOINT: usize = 1 << 2; - pub const PEER_TAGS: usize = 1 << 3; - pub const ADDITIONAL_TAGS: usize = 1 << 4; - pub const COUNT: u8 = 5; +pub struct CollapsedFieldSet(usize); +impl CollapsedFieldSet { + pub const RESOURCE_NAME: usize = 1 << 0; + pub const HTTP_ENDPOINT: usize = 1 << 1; + pub const PEER_TAGS: usize = 1 << 2; + pub const ADDITIONAL_TAGS: usize = 1 << 3; + const FIELDS: [usize; 4] = [ + Self::RESOURCE_NAME, + Self::HTTP_ENDPOINT, + Self::PEER_TAGS, + Self::ADDITIONAL_TAGS, + ]; + + pub fn empty() -> CollapsedFieldSet { + Self(0) + } + + pub fn add(&mut self, field: usize) { + debug_assert!(Self::FIELDS.contains(&field)); + self.0 |= field; + } } -pub(super) const COLLAPSED_FIELD_METRIC_SIZE: usize = 1 << collapsed_field::COUNT; +pub const COLLAPSED_FIELD_METRIC_SIZE: usize = 1 << CollapsedFieldSet::FIELDS.len(); -#[derive(Debug, Clone, Default, Copy)] -// Note: slot 0 is a counter for non_collapsed spans. It's not used for emitting telemetry -pub struct CollapsedFieldsMetrics(pub(super) [usize; COLLAPSED_FIELD_METRIC_SIZE]); +// Verify the array is of a reasonable size +const_assert!(COLLAPSED_FIELD_METRIC_SIZE <= 16); -const_assert!(COLLAPSED_FIELD_METRIC_SIZE <= 32); +// Note: slot 0 is a counter for non_collapsed spans. It's not used for emitting telemetry +#[derive(Debug, Clone, Default, Copy)] +pub struct CollapsedFieldsMetrics([usize; COLLAPSED_FIELD_METRIC_SIZE]); impl CollapsedFieldsMetrics { pub fn zero() -> Self { @@ -67,30 +83,27 @@ impl CollapsedFieldsMetrics { #[cfg(any(feature = "telemetry", feature = "dogstatsd"))] fn fields_mask_to_list(mask: usize) -> Vec { let mut tags = Vec::new(); - for field_pow in 1..collapsed_field::COUNT { + for field_pow in 0..CollapsedFieldSet::FIELDS.len() { let field_value = 1 << field_pow; - debug_assert!([ - collapsed_field::RESOURCE_NAME, - collapsed_field::HTTP_ENDPOINT, - collapsed_field::PEER_TAGS, - collapsed_field::ADDITIONAL_TAGS - ] - .contains(&field_value)); + debug_assert!( + CollapsedFieldSet::FIELDS.contains(&field_value), + "{field_value} is an invalid value for a CollapsedFieldSet" + ); let has_field = (mask & field_value) != 0; if !has_field { continue; } let field_tag = match field_value { - collapsed_field::RESOURCE_NAME => { + CollapsedFieldSet::RESOURCE_NAME => { libdd_common::tag!("collapsed_spans", "resource") } - collapsed_field::HTTP_ENDPOINT => { + CollapsedFieldSet::HTTP_ENDPOINT => { libdd_common::tag!("collapsed_spans", "http_endpoint") } - collapsed_field::PEER_TAGS => { + CollapsedFieldSet::PEER_TAGS => { libdd_common::tag!("collapsed_spans", "peer_tags") } - collapsed_field::ADDITIONAL_TAGS => { + CollapsedFieldSet::ADDITIONAL_TAGS => { libdd_common::tag!("collapsed_spans", "additional_metric_tags") } // Unreachable: asserted just above that field is one of the 4 possible values @@ -101,6 +114,10 @@ impl CollapsedFieldsMetrics { debug_assert!(!tags.is_empty()); tags } + + pub fn increment(&mut self, field_set: CollapsedFieldSet) { + self.0[field_set.0] += 1; + } } impl std::ops::AddAssign for CollapsedFieldsMetrics { From 2eda225df022d735655e81effbb1d2abfcfec355 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Tue, 28 Jul 2026 14:45:29 +0200 Subject: [PATCH 13/15] better doc --- .../cardinality_limit_telemetry.rs | 22 ++++++++++++------- 1 file changed, 14 insertions(+), 8 deletions(-) diff --git a/libdd-trace-stats/src/span_concentrator/cardinality_limit_telemetry.rs b/libdd-trace-stats/src/span_concentrator/cardinality_limit_telemetry.rs index 6482374a4f..dc1bc01c3b 100644 --- a/libdd-trace-stats/src/span_concentrator/cardinality_limit_telemetry.rs +++ b/libdd-trace-stats/src/span_concentrator/cardinality_limit_telemetry.rs @@ -1,20 +1,23 @@ // Copyright 2024-Present Datadog, Inc. https://www.datadoghq.com/ // SPDX-License-Identifier: Apache-2.0 -//! Each collapsed field is assigned a bit; a span's mask is the OR of the bits for the -//! fields collapsed on it, so masks double as an index into -//! [`CollapsedFieldsMetrics`]'s counter array (one counter per distinct field combination). +//! This module implement the logic for sending telemetry/dogstatsd related to cardinality limits +// Because the cardinality limit RFC requires one point of telemetry with a set of collapsed fields +// and not one point for each collapsed field, this solution was found as an alternative to adding +// telemetry and dogstatsd client in the SpanConcentrator to keep it as "pure computation logic" #[cfg(feature = "telemetry")] use libdd_capabilities::{HttpClientCapability, MaybeSend, SleepCapability}; use libdd_common::tag::const_assert; +/// Bitset of collapsed stats key fields pub struct CollapsedFieldSet(usize); impl CollapsedFieldSet { pub const RESOURCE_NAME: usize = 1 << 0; pub const HTTP_ENDPOINT: usize = 1 << 1; pub const PEER_TAGS: usize = 1 << 2; pub const ADDITIONAL_TAGS: usize = 1 << 3; + const FIELDS: [usize; 4] = [ Self::RESOURCE_NAME, Self::HTTP_ENDPOINT, @@ -22,33 +25,35 @@ impl CollapsedFieldSet { Self::ADDITIONAL_TAGS, ]; + /// Default value with none of the fields set (no collapsed fields) pub fn empty() -> CollapsedFieldSet { Self(0) } + /// Enable the bit corresponding to `field` pub fn add(&mut self, field: usize) { debug_assert!(Self::FIELDS.contains(&field)); self.0 |= field; } } -pub const COLLAPSED_FIELD_METRIC_SIZE: usize = 1 << CollapsedFieldSet::FIELDS.len(); - +const COLLAPSED_FIELD_METRIC_SIZE: usize = 1 << CollapsedFieldSet::FIELDS.len(); // Verify the array is of a reasonable size const_assert!(COLLAPSED_FIELD_METRIC_SIZE <= 16); - +/// Counter of combination of collapsed stats key fields // Note: slot 0 is a counter for non_collapsed spans. It's not used for emitting telemetry #[derive(Debug, Clone, Default, Copy)] pub struct CollapsedFieldsMetrics([usize; COLLAPSED_FIELD_METRIC_SIZE]); impl CollapsedFieldsMetrics { + /// Default value every combination of collapsed fields set to 0 pub fn zero() -> Self { Self::default() } #[cfg(feature = "dogstatsd")] pub fn emit_dogstatsd(&self, dogstatsd: &libdd_dogstatsd_client::DogStatsDClient) { - // skip the first slot that is used to count span which have no collapsed fields. + // skip the first slot that is used to count span which have no collapsed fields for (mask, &count) in self.0.iter().enumerate().skip(1) { if count > 0 { let tags = Self::fields_mask_to_list(mask); @@ -69,7 +74,7 @@ impl CollapsedFieldsMetrics { handle: &libdd_telemetry::worker::TelemetryWorkerHandle, context_key: &libdd_telemetry::metrics::ContextKey, ) { - // skip the first slot that is used to count span which have no collapsed fields. + // skip the first slot that is used to count span which have no collapsed fields for (mask, &count) in self.0.iter().enumerate().skip(1) { if count > 0 { let tags = Self::fields_mask_to_list(mask); @@ -115,6 +120,7 @@ impl CollapsedFieldsMetrics { tags } + /// Increment the telemetry counter corresponding to this collapsed field combination pub fn increment(&mut self, field_set: CollapsedFieldSet) { self.0[field_set.0] += 1; } From 67cd3163606e7991affe86fc5b10c428ad7c86a3 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Tue, 28 Jul 2026 14:58:00 +0200 Subject: [PATCH 14/15] refacto: make drain_due_buckets return a FlushResult --- datadog-ipc/src/shm_stats.rs | 2 +- .../src/span_concentrator/mod.rs | 75 +++++++++---------- 2 files changed, 37 insertions(+), 40 deletions(-) diff --git a/datadog-ipc/src/shm_stats.rs b/datadog-ipc/src/shm_stats.rs index 38b6dfd059..72687985a2 100644 --- a/datadog-ipc/src/shm_stats.rs +++ b/datadog-ipc/src/shm_stats.rs @@ -823,7 +823,7 @@ impl ShmSpanConcentrator { } impl FlushableConcentrator for ShmSpanConcentrator { - fn flush_buckets(&mut self, force: bool) -> FlushResult { + fn flush_buckets(&mut self, force: bool) -> FlushResult { // The SHM concentrator does not perform client-side obfuscation nor emits telemetry. FlushResult { obfuscated_buckets: vec![], diff --git a/libdd-trace-stats/src/span_concentrator/mod.rs b/libdd-trace-stats/src/span_concentrator/mod.rs index aafc2a871d..461fa1ff7a 100644 --- a/libdd-trace-stats/src/span_concentrator/mod.rs +++ b/libdd-trace-stats/src/span_concentrator/mod.rs @@ -44,21 +44,20 @@ fn normalize_additional_metric_tag_keys(mut keys: Vec) -> Vec { /// /// 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 { +pub struct FlushResult { /// Buckets whose resource names were obfuscated client-side. - pub obfuscated_buckets: Vec, + pub obfuscated_buckets: Vec, /// Buckets whose resource names were left as-is. - pub unobfuscated_buckets: Vec, + 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, pub collapsed_fields_metrics: CollapsedFieldsMetrics, } -impl FlushResult { +impl FlushResult { /// All flushed buckets regardless of obfuscation. - #[cfg(test)] - pub fn all_buckets(self) -> Vec { + pub fn all_buckets(self) -> Vec { let mut buckets = self.obfuscated_buckets; buckets.extend(self.unobfuscated_buckets); buckets @@ -72,11 +71,11 @@ impl FlushResult { pub trait FlushableConcentrator { /// 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; + fn flush_buckets(&mut self, force: bool) -> FlushResult; } impl FlushableConcentrator for SpanConcentrator { - fn flush_buckets(&mut self, force: bool) -> FlushResult { + fn flush_buckets(&mut self, force: bool) -> FlushResult { self.flush(SystemTime::now(), force) } } @@ -380,34 +379,16 @@ impl SpanConcentrator { /// all buckets. /// /// Obfuscated and un-obfuscated buckets are returned separately, see [`FlushResult`]. - pub fn flush(&mut self, now: SystemTime, force: bool) -> FlushResult { - let (buckets, collapsed_spans, collapsed_fields_metrics) = - 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, - collapsed_fields_metrics, - } + pub fn flush(&mut self, now: SystemTime, force: bool) -> FlushResult { + self.drain_due_buckets(now, force, StatsBucket::flush) } /// 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) - .0; - buckets.into_iter().map(|(_, bucket)| bucket).collect() + self.drain_due_buckets(now, force, StatsBucket::flush_with_otlp_exact) + .all_buckets() } /// Drain the buckets that are due for flushing, encoding each with `encode`. @@ -421,7 +402,7 @@ impl SpanConcentrator { now: SystemTime, force: bool, encode: impl Fn(StatsBucket, u64) -> T, - ) -> (Vec<(bool, T)>, u64, CollapsedFieldsMetrics) { + ) -> FlushResult { // 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(); @@ -431,9 +412,9 @@ impl SpanConcentrator { align_timestamp(now_timestamp, self.bucket_size) - (self.buffer_len as u64 - 1) * self.bucket_size }; - let mut total_collapsed = 0; - let mut total_collapsed_fields = CollapsedFieldsMetrics::zero(); - let buckets_pb = buckets + let mut collapsed_spans = 0; + let mut collapsed_fields_metrics = CollapsedFieldsMetrics::zero(); + let buckets_pb: Vec<(bool, T)> = buckets .into_iter() .filter_map(|(timestamp, bucket)| { // Always keep `bufferLen` buckets (default is 2: current + previous one). @@ -449,8 +430,8 @@ impl SpanConcentrator { self.buckets.insert(timestamp, bucket); return None; } - total_collapsed += bucket.collapsed_count(); - total_collapsed_fields += bucket.collapsed_fields_metrics(); + collapsed_spans += bucket.collapsed_count(); + collapsed_fields_metrics += bucket.collapsed_fields_metrics(); #[cfg(feature = "stats-obfuscation")] let obfuscated = bucket.obfuscated; #[cfg(not(feature = "stats-obfuscation"))] @@ -458,14 +439,30 @@ impl SpanConcentrator { Some((obfuscated, encode(bucket, self.bucket_size))) }) .collect(); - if total_collapsed > 0 { + if collapsed_spans > 0 { debug!( max_entries_per_bucket = self.cardinality_limits.whole_key_limit, - total_collapsed, + collapsed_spans, "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, total_collapsed_fields) + + let mut obfuscated_buckets = Vec::new(); + let mut unobfuscated_buckets = Vec::new(); + for (obfuscated, bucket) in buckets_pb { + if obfuscated { + obfuscated_buckets.push(bucket); + } else { + unobfuscated_buckets.push(bucket); + } + } + + FlushResult { + obfuscated_buckets, + unobfuscated_buckets, + collapsed_spans, + collapsed_fields_metrics, + } } } From 382b1577411ab61a37f9fdf825a6d4af43d9509d Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Tue, 28 Jul 2026 15:08:27 +0200 Subject: [PATCH 15/15] fix comment --- .../src/span_concentrator/cardinality_limit_telemetry.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/libdd-trace-stats/src/span_concentrator/cardinality_limit_telemetry.rs b/libdd-trace-stats/src/span_concentrator/cardinality_limit_telemetry.rs index dc1bc01c3b..7711af02d7 100644 --- a/libdd-trace-stats/src/span_concentrator/cardinality_limit_telemetry.rs +++ b/libdd-trace-stats/src/span_concentrator/cardinality_limit_telemetry.rs @@ -111,7 +111,8 @@ impl CollapsedFieldsMetrics { CollapsedFieldSet::ADDITIONAL_TAGS => { libdd_common::tag!("collapsed_spans", "additional_metric_tags") } - // Unreachable: asserted just above that field is one of the 4 possible values + // Should be unreachable, but don't fail in prod if provided with an invalid field + // set _ => continue, }; tags.push(field_tag);