From cb48ef6b61d4c78e2562a44ee41ecd939d5bb2fe Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Wed, 8 Jul 2026 14:00:57 +0200 Subject: [PATCH 01/16] 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 03b9a4da02..0b679a6791 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 47c6bfe935..b7880f9289 100644 --- a/libdd-trace-stats/src/stats_exporter.rs +++ b/libdd-trace-stats/src/stats_exporter.rs @@ -314,6 +314,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; @@ -580,7 +583,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; @@ -633,7 +635,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 6594e13689a0f2633d61db5f71a1ba4a24b7ce36 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Wed, 8 Jul 2026 14:17:59 +0200 Subject: [PATCH 02/16] 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 fca0f658472f0ce3f3a4e00128ad922c14cc3083 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Wed, 8 Jul 2026 14:37:16 +0200 Subject: [PATCH 03/16] 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 5582c934cd..93c490ae94 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 44ead972e1..5b8c8fb28a 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 0b679a6791..46a02b78d4 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 dc37b7af4d7add5a4cf2cab9a0da892ddd0c7fb7 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Wed, 8 Jul 2026 14:54:31 +0200 Subject: [PATCH 04/16] 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 5abfe75c65d4edab6a0271fb2b83db8b31184910 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Wed, 8 Jul 2026 14:56:36 +0200 Subject: [PATCH 05/16] 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 b7880f9289..717e88b654 100644 --- a/libdd-trace-stats/src/stats_exporter.rs +++ b/libdd-trace-stats/src/stats_exporter.rs @@ -314,9 +314,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 2ac1355ad3b4af122080440c7dd46561c8e3f6c0 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Wed, 8 Jul 2026 15:34:19 +0200 Subject: [PATCH 06/16] 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 69c5b91d7f7b4ef1e5d789b5d4cc300ee09103f8 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Wed, 8 Jul 2026 15:34:45 +0200 Subject: [PATCH 07/16] 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 e6fdca72eb433a4d28ccd34077118c4f67aedd23 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Wed, 8 Jul 2026 15:35:51 +0200 Subject: [PATCH 08/16] 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 d82ebfd172074c96211805484f488dc61a2d89d8 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Wed, 8 Jul 2026 15:58:19 +0200 Subject: [PATCH 09/16] 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 40b7a8b36c3b10d3efc9e019eb7cb648b3b156ad Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Wed, 8 Jul 2026 18:15:02 +0200 Subject: [PATCH 10/16] feat(stats): add telemetry for per-field cardinality limits --- datadog-ipc/src/shm_stats.rs | 4 +- .../src/span_concentrator/aggregation.rs | 43 +++- .../src/span_concentrator/mod.rs | 44 +++- libdd-trace-stats/src/stats_exporter.rs | 198 +++++++++++++++--- 4 files changed, 248 insertions(+), 41 deletions(-) diff --git a/datadog-ipc/src/shm_stats.rs b/datadog-ipc/src/shm_stats.rs index 87b55981da..65277902d8 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, + FixedAggregationKey, FlushResult, FlushableConcentrator, StatsBucketCollapseTelemetry, }; use crate::platform::{FileBackedHandle, MappedMem, NamedShmHandle}; @@ -827,7 +827,7 @@ impl FlushableConcentrator for ShmSpanConcentrator { FlushResult { obfuscated_buckets: vec![], unobfuscated_buckets: self.drain_buckets(force), - collapsed_spans: 0, + collapsed_spans: StatsBucketCollapseTelemetry::default(), } } } diff --git a/libdd-trace-stats/src/span_concentrator/aggregation.rs b/libdd-trace-stats/src/span_concentrator/aggregation.rs index 680b41220c..3a5c369708 100644 --- a/libdd-trace-stats/src/span_concentrator/aggregation.rs +++ b/libdd-trace-stats/src/span_concentrator/aggregation.rs @@ -425,6 +425,31 @@ pub struct OtlpStatsBucket { pub exact: Vec, } +#[derive(Debug, Clone, Copy, Default)] +/// Number of spans collapsed into the overflow bucket for each reason they got collapsed +pub struct StatsBucketCollapseTelemetry { + /// Collapsed due to whole-key cardinality limiting. + pub whole_key: u64, + /// Collapsed due to resource key cardinality limiting. + pub resources: u64, + /// Collapsed due to http_endpoint key cardinality limiting. + pub http_endpoint: u64, + /// Collapsed due to peer_tags key cardinality limiting. + pub peer_tags: u64, + /// Collapsed due to additional_tags key cardinality limiting. + pub additional_tags: u64, +} + +impl std::ops::AddAssign for StatsBucketCollapseTelemetry { + fn add_assign(&mut self, rhs: StatsBucketCollapseTelemetry) { + self.whole_key += rhs.whole_key; + self.resources += rhs.resources; + self.http_endpoint += rhs.http_endpoint; + self.peer_tags += rhs.peer_tags; + self.additional_tags += rhs.additional_tags; + } +} + /// A time bucket used for stats aggregation. It stores a map of GroupedStats storing the stats of /// spans aggregated on their AggregationKey. #[derive(Debug, Clone)] @@ -434,13 +459,13 @@ 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, + // HashSet of hashes of values so save memory, no need for total accuracy here 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, + collapsed_counts: StatsBucketCollapseTelemetry, /// Indicates if stats obfuscated in this bucket. This is set once at creation and stays /// constant per bucket #[cfg(feature = "stats-obfuscation")] @@ -460,19 +485,20 @@ impl StatsBucket { data: HashMap::new(), start: start_timestamp, 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(), + collapsed_counts: StatsBucketCollapseTelemetry::default(), } } - /// Return the number of spans collapsed into the overflow bucket. - pub(super) fn collapsed_count(&self) -> u64 { - self.collapsed_count + /// Return the number of spans collapsed into the overflow buckets, for whole-key and for each + /// field + pub(super) fn collapsed_counts(&self) -> StatsBucketCollapseTelemetry { + self.collapsed_counts } /// Insert a value as stats in the group corresponding to the aggregation key. If the key is new @@ -492,7 +518,7 @@ impl StatsBucket { if self.data.len() >= self.cardinality_limits.whole_key_limit && !self.data.contains_key(&key) { - self.collapsed_count += 1; + self.collapsed_counts.whole_key += 1; self.data .entry(OwnedAggregationKey::overflow_key()) .or_default() @@ -519,6 +545,7 @@ impl StatsBucket { && !self.distinct_resources.contains(&resource_name_hash) { key.fixed.resource_name = TRACER_BLOCKED_VALUE; + self.collapsed_counts.resources += 1; } else { self.distinct_resources.insert(resource_name_hash); } @@ -528,6 +555,7 @@ impl StatsBucket { && !self.distinct_http_endpoint.contains(&http_endpoint_hash) { key.fixed.http_endpoint = TRACER_BLOCKED_VALUE; + self.collapsed_counts.http_endpoint += 1; } else { self.distinct_http_endpoint.insert(http_endpoint_hash); } @@ -537,6 +565,7 @@ impl StatsBucket { && !self.distinct_peer_tags.contains(&peer_tags_hash) { key.peer_tags = vec![(TRACER_BLOCKED_VALUE, Cow::Borrowed(""))]; + self.collapsed_counts.peer_tags += 1; } else { self.distinct_peer_tags.insert(peer_tags_hash); } diff --git a/libdd-trace-stats/src/span_concentrator/mod.rs b/libdd-trace-stats/src/span_concentrator/mod.rs index 0b1faee070..8f571ff045 100644 --- a/libdd-trace-stats/src/span_concentrator/mod.rs +++ b/libdd-trace-stats/src/span_concentrator/mod.rs @@ -11,7 +11,10 @@ use aggregation::StatsBucket; mod aggregation; use aggregation::BorrowedAggregationKey; -pub use aggregation::{FixedAggregationKey, OtlpExactCell, OtlpExactGroup, OtlpStatsBucket}; +pub use aggregation::{ + FixedAggregationKey, OtlpExactCell, OtlpExactGroup, OtlpStatsBucket, + StatsBucketCollapseTelemetry, +}; pub mod stat_span; pub use stat_span::StatSpan; @@ -26,8 +29,8 @@ pub struct FlushResult { /// 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, + /// whole-key cardinality limiting across all flushed time buckets. + pub collapsed_spans: StatsBucketCollapseTelemetry, } impl FlushResult { @@ -164,6 +167,8 @@ pub struct SpanConcentrator { span_kinds_stats_computed: Vec, /// keys for supplementary tags that describe peer.service entities peer_tag_keys: Vec, + /// Cardinality collapsed log was already emitted + cardinality_log_emitted: bool, #[cfg(feature = "stats-obfuscation")] obfuscation_config: SharedStatsComputationObfuscationConfig, } @@ -214,6 +219,7 @@ impl SpanConcentrator { peer_tag_keys, #[cfg(feature = "stats-obfuscation")] obfuscation_config: obfuscation_config.unwrap_or_default(), + cardinality_log_emitted: false, } } @@ -343,7 +349,7 @@ impl SpanConcentrator { now: SystemTime, force: bool, encode: impl Fn(StatsBucket, u64) -> T, - ) -> (Vec<(bool, T)>, u64) { + ) -> (Vec<(bool, T)>, StatsBucketCollapseTelemetry) { // 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(); @@ -353,7 +359,7 @@ 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 = StatsBucketCollapseTelemetry::default(); let buckets_pb = buckets .into_iter() .filter_map(|(timestamp, bucket)| { @@ -370,7 +376,7 @@ impl SpanConcentrator { self.buckets.insert(timestamp, bucket); return None; } - total_collapsed += bucket.collapsed_count(); + total_collapsed += bucket.collapsed_counts(); #[cfg(feature = "stats-obfuscation")] let obfuscated = bucket.obfuscated; #[cfg(not(feature = "stats-obfuscation"))] @@ -378,13 +384,35 @@ impl SpanConcentrator { Some((obfuscated, encode(bucket, self.bucket_size))) }) .collect(); - if total_collapsed > 0 { + + if !self.cardinality_log_emitted && total_collapsed.whole_key > 0 { + self.cardinality_log_emitted = true; debug!( max_entries_per_bucket = self.cardinality_limits.whole_key_limit, - total_collapsed, + total_whole_key_collapsed = total_collapsed.whole_key, "Client-side stats values have been collapsed to 'tracer_blocked_value'. This is due to the cardinality exceeding DD_TRACE_STATS_CARDINALITY_LIMIT" ); } + if !self.cardinality_log_emitted + && (total_collapsed.resources > 0 + || total_collapsed.http_endpoint > 0 + || total_collapsed.peer_tags > 0 + || total_collapsed.additional_tags > 0) + { + self.cardinality_log_emitted = true; + debug!( + max_distinct_resource_per_bucket = self.cardinality_limits.resource_limit, + total_resource_collapsed = total_collapsed.resources, + max_distinct_http_endpoint_per_bucket = self.cardinality_limits.http_endpoint_limit, + total_http_endpoint_collapsed = total_collapsed.http_endpoint, + max_distinct_peer_tags_per_bucket = self.cardinality_limits.peer_tags_limit, + total_peer_tags_collapsed = total_collapsed.peer_tags, + max_distinct_additional_tags_per_bucket = self.cardinality_limits.additional_tags_limit, + total_additional_tags_collapsed = total_collapsed.additional_tags, + "Client-side stats field have been collapsed to 'tracer_blocked_value'. This is due to the cardinality exceeding one of the DD_TRACE_STATS_*_CARDINALITY_LIMIT" + ); + } + (buckets_pb, total_collapsed) } } diff --git a/libdd-trace-stats/src/stats_exporter.rs b/libdd-trace-stats/src/stats_exporter.rs index 717e88b654..63b6ba64d7 100644 --- a/libdd-trace-stats/src/stats_exporter.rs +++ b/libdd-trace-stats/src/stats_exporter.rs @@ -78,6 +78,23 @@ impl From for StatsMetadata { } } +#[cfg(feature = "telemetry")] +#[derive(Debug)] +struct StatsExporterTelemetryKeys { + collapsed_whole_key: libdd_telemetry::metrics::ContextKey, + collapsed_resources: libdd_telemetry::metrics::ContextKey, + collapsed_http_endpoints: libdd_telemetry::metrics::ContextKey, + collapsed_peer_tags: libdd_telemetry::metrics::ContextKey, + collapsed_additional_tags: libdd_telemetry::metrics::ContextKey, +} + +#[cfg(feature = "telemetry")] +#[derive(Debug)] +struct StatsExporterTelemetry { + handle: libdd_telemetry::worker::TelemetryWorkerHandle, + keys: StatsExporterTelemetryKeys, +} + /// An exporter that concentrates and sends stats to the agent. /// /// `Cap` is the capabilities bundle (HTTP + sleep). Leaf crates pin it to a @@ -97,10 +114,7 @@ pub struct StatsExporter< supported_obfuscation_version: &'static str, /// Optional telemetry handle and context key. #[cfg(feature = "telemetry")] - telemetry: Option<( - libdd_telemetry::worker::TelemetryWorkerHandle, - libdd_telemetry::metrics::ContextKey, - )>, + telemetry: Option, /// Optional DogStatsD client. #[cfg(feature = "dogstatsd")] dogstatsd: Option>, @@ -131,14 +145,54 @@ impl ) -> Self { #[cfg(feature = "telemetry")] let telemetry = telemetry.map(|handle| { - let key = handle.register_metric_context( + let collapsed_whole_key = handle.register_metric_context( COLLAPSED_SPANS_TELEMETRY_METRIC.to_string(), vec![libdd_common::tag!("collapsed_spans", "whole_key")], libdd_telemetry::data::metrics::MetricType::Count, true, libdd_telemetry::data::metrics::MetricNamespace::Tracers, ); - (handle, key) + let collapsed_resources = handle.register_metric_context( + COLLAPSED_SPANS_TELEMETRY_METRIC.to_string(), + vec![libdd_common::tag!("collapsed_spans", "resource")], + libdd_telemetry::data::metrics::MetricType::Count, + true, + libdd_telemetry::data::metrics::MetricNamespace::Tracers, + ); + let collapsed_http_endpoints = handle.register_metric_context( + COLLAPSED_SPANS_TELEMETRY_METRIC.to_string(), + vec![libdd_common::tag!("collapsed_spans", "http_endpoint")], + libdd_telemetry::data::metrics::MetricType::Count, + true, + libdd_telemetry::data::metrics::MetricNamespace::Tracers, + ); + let collapsed_peer_tags = handle.register_metric_context( + COLLAPSED_SPANS_TELEMETRY_METRIC.to_string(), + vec![libdd_common::tag!("collapsed_spans", "peer_tags")], + libdd_telemetry::data::metrics::MetricType::Count, + true, + libdd_telemetry::data::metrics::MetricNamespace::Tracers, + ); + let collapsed_additional_tags = handle.register_metric_context( + COLLAPSED_SPANS_TELEMETRY_METRIC.to_string(), + vec![libdd_common::tag!( + "collapsed_spans", + "additional_metric_tags" + )], + libdd_telemetry::data::metrics::MetricType::Count, + true, + libdd_telemetry::data::metrics::MetricNamespace::Tracers, + ); + StatsExporterTelemetry { + handle, + keys: StatsExporterTelemetryKeys { + collapsed_whole_key, + collapsed_resources, + collapsed_http_endpoints, + collapsed_peer_tags, + collapsed_additional_tags, + }, + } }); Self { flush_interval, @@ -179,21 +233,105 @@ impl concentrator.flush_buckets(force_flush) }; - if flush.collapsed_spans > 0 { + if flush.collapsed_spans.whole_key > 0 { #[cfg(feature = "telemetry")] - if let Some((handle, key)) = &self.telemetry { - let _ = handle.add_point(flush.collapsed_spans as f64, key, vec![]); + if let Some(telemetry) = &self.telemetry { + let _ = telemetry.handle.add_point( + flush.collapsed_spans.whole_key as f64, + &telemetry.keys.collapsed_whole_key, + vec![], + ); } #[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, + flush.collapsed_spans.whole_key as i64, [libdd_common::tag!("collapsed_spans", "whole_key")].iter(), )]); } } + if flush.collapsed_spans.resources > 0 { + #[cfg(feature = "telemetry")] + if let Some(telemetry) = &self.telemetry { + let _ = telemetry.handle.add_point( + flush.collapsed_spans.resources as f64, + &telemetry.keys.collapsed_resources, + vec![], + ); + } + #[cfg(feature = "dogstatsd")] + if let Some(client) = &self.dogstatsd { + client.send(vec![libdd_dogstatsd_client::DogStatsDAction::Count( + COLLAPSED_SPANS_HEALTH_METRIC, + flush.collapsed_spans.resources as i64, + [libdd_common::tag!("collapsed_spans", "resource")].iter(), + )]); + } + } + + if flush.collapsed_spans.http_endpoint > 0 { + #[cfg(feature = "telemetry")] + if let Some(telemetry) = &self.telemetry { + let _ = telemetry.handle.add_point( + flush.collapsed_spans.http_endpoint as f64, + &telemetry.keys.collapsed_http_endpoints, + vec![], + ); + } + #[cfg(feature = "dogstatsd")] + if let Some(client) = &self.dogstatsd { + client.send(vec![libdd_dogstatsd_client::DogStatsDAction::Count( + COLLAPSED_SPANS_HEALTH_METRIC, + flush.collapsed_spans.http_endpoint as i64, + [libdd_common::tag!("collapsed_spans", "http_endpoint")].iter(), + )]); + } + } + + if flush.collapsed_spans.peer_tags > 0 { + #[cfg(feature = "telemetry")] + if let Some(telemetry) = &self.telemetry { + let _ = telemetry.handle.add_point( + flush.collapsed_spans.peer_tags as f64, + &telemetry.keys.collapsed_peer_tags, + vec![], + ); + } + #[cfg(feature = "dogstatsd")] + if let Some(client) = &self.dogstatsd { + client.send(vec![libdd_dogstatsd_client::DogStatsDAction::Count( + COLLAPSED_SPANS_HEALTH_METRIC, + flush.collapsed_spans.peer_tags as i64, + [libdd_common::tag!("collapsed_spans", "peer_tags")].iter(), + )]); + } + } + + if flush.collapsed_spans.additional_tags > 0 { + #[cfg(feature = "telemetry")] + if let Some(telemetry) = &self.telemetry { + let _ = telemetry.handle.add_point( + flush.collapsed_spans.additional_tags as f64, + &telemetry.keys.collapsed_additional_tags, + vec![], + ); + } + #[cfg(feature = "dogstatsd")] + if let Some(client) = &self.dogstatsd { + client.send(vec![libdd_dogstatsd_client::DogStatsDAction::Count( + COLLAPSED_SPANS_HEALTH_METRIC, + flush.collapsed_spans.additional_tags as i64, + [libdd_common::tag!( + "collapsed_spans", + "additional_metric_tags" + )] + .iter(), + )]); + } + } + // 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; @@ -636,7 +774,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() }), #[cfg(feature = "stats-obfuscation")] @@ -645,26 +784,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() }, @@ -768,12 +907,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" ); } @@ -819,23 +969,23 @@ mod tests { stats_exporter.send(true).await.unwrap(); let stats_exporter_ref = &stats_exporter; - let (handle_ref, _key) = stats_exporter_ref + let telemetry = stats_exporter_ref .telemetry .as_ref() .expect("telemetry must be set"); - let receiver = handle_ref.stats().expect("failed to request stats"); + let receiver = telemetry.handle.stats().expect("failed to request stats"); let stats = receiver.await.expect("failed to receive stats"); // metric_contexts == 1 verifies that exactly one metric name was registered // (i.e. COLLAPSED_SPANS_METRIC and nothing else). // metric_buckets.buckets == 1 verifies that a data point was recorded for it. // However it does not check the value of the data point. assert_eq!( - stats.metric_contexts, 1, - "exactly one metric context (COLLAPSED_SPANS_METRIC) should be registered" + stats.metric_contexts, 5, + "exactly five metric context (COLLAPSED_SPANS_METRIC, and one for each of the 4 supported field collapsed) 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 639c1a728d83fd3e5a3d1399d32c926b86d6dc8a Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Thu, 9 Jul 2026 13:56:03 +0200 Subject: [PATCH 11/16] feat: log each field limit reached separatly --- .../src/span_concentrator/mod.rs | 55 +++++++++++++------ .../src/span_concentrator/tests.rs | 22 ++++++++ 2 files changed, 60 insertions(+), 17 deletions(-) diff --git a/libdd-trace-stats/src/span_concentrator/mod.rs b/libdd-trace-stats/src/span_concentrator/mod.rs index 8f571ff045..7106865b38 100644 --- a/libdd-trace-stats/src/span_concentrator/mod.rs +++ b/libdd-trace-stats/src/span_concentrator/mod.rs @@ -2,6 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 //! This module implements the SpanConcentrator used to aggregate spans into stats use std::collections::HashMap; +use std::sync::OnceLock; use std::time::{self, Duration, SystemTime}; use tracing::{debug, warn}; @@ -19,6 +20,17 @@ pub use aggregation::{ pub mod stat_span; pub use stat_span::StatSpan; +/// Executes a statement one time. Useful when emitting a warning once +#[macro_export] +macro_rules! once { + ($expression:expr) => {{ + static LOGGED: OnceLock<()> = OnceLock::new(); + LOGGED.get_or_init(|| { + $expression; + }); + }}; +} + /// Result of flushing a concentrator. /// /// Obfuscated and un-obfuscated buckets are kept separate because they must be sent in distinct @@ -167,8 +179,6 @@ pub struct SpanConcentrator { span_kinds_stats_computed: Vec, /// keys for supplementary tags that describe peer.service entities peer_tag_keys: Vec, - /// Cardinality collapsed log was already emitted - cardinality_log_emitted: bool, #[cfg(feature = "stats-obfuscation")] obfuscation_config: SharedStatsComputationObfuscationConfig, } @@ -219,7 +229,6 @@ impl SpanConcentrator { peer_tag_keys, #[cfg(feature = "stats-obfuscation")] obfuscation_config: obfuscation_config.unwrap_or_default(), - cardinality_log_emitted: false, } } @@ -385,32 +394,44 @@ impl SpanConcentrator { }) .collect(); - if !self.cardinality_log_emitted && total_collapsed.whole_key > 0 { - self.cardinality_log_emitted = true; - debug!( + if total_collapsed.whole_key > 0 { + once!(debug!( max_entries_per_bucket = self.cardinality_limits.whole_key_limit, total_whole_key_collapsed = total_collapsed.whole_key, "Client-side stats values have been collapsed to 'tracer_blocked_value'. This is due to the cardinality exceeding DD_TRACE_STATS_CARDINALITY_LIMIT" - ); + )); } - if !self.cardinality_log_emitted - && (total_collapsed.resources > 0 - || total_collapsed.http_endpoint > 0 - || total_collapsed.peer_tags > 0 - || total_collapsed.additional_tags > 0) - { - self.cardinality_log_emitted = true; - debug!( + + if total_collapsed.resources > 0 { + once!(debug!( max_distinct_resource_per_bucket = self.cardinality_limits.resource_limit, total_resource_collapsed = total_collapsed.resources, + "Client-side stats field 'resource' has been collapsed to 'tracer_blocked_value'. This is due to the cardinality exceeding DD_TRACE_STATS_RESOURCE_CARDINALITY_LIMIT" + )); + } + + if total_collapsed.http_endpoint > 0 { + once!(debug!( max_distinct_http_endpoint_per_bucket = self.cardinality_limits.http_endpoint_limit, total_http_endpoint_collapsed = total_collapsed.http_endpoint, + "Client-side stats field 'http_endpoint' has been collapsed to 'tracer_blocked_value'. This is due to the cardinality exceeding DD_TRACE_STATS_HTTP_ENDPOINT_CARDINALITY_LIMIT" + )); + } + + if total_collapsed.peer_tags > 0 { + once!(debug!( max_distinct_peer_tags_per_bucket = self.cardinality_limits.peer_tags_limit, total_peer_tags_collapsed = total_collapsed.peer_tags, + "Client-side stats field 'peer_tags' has been collapsed to 'tracer_blocked_value'. This is due to the cardinality exceeding DD_TRACE_STATS_PEER_TAGS_CARDINALITY_LIMIT" + )); + } + + if total_collapsed.additional_tags > 0 { + once!(debug!( max_distinct_additional_tags_per_bucket = self.cardinality_limits.additional_tags_limit, total_additional_tags_collapsed = total_collapsed.additional_tags, - "Client-side stats field have been collapsed to 'tracer_blocked_value'. This is due to the cardinality exceeding one of the DD_TRACE_STATS_*_CARDINALITY_LIMIT" - ); + "Client-side stats field 'additional_tags' has been collapsed to 'tracer_blocked_value'. This is due to the cardinality exceeding DD_TRACE_STATS_ADDITIONAL_TAGS_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 938ef8e8b9..443ac794ea 100644 --- a/libdd-trace-stats/src/span_concentrator/tests.rs +++ b/libdd-trace-stats/src/span_concentrator/tests.rs @@ -2088,3 +2088,25 @@ fn test_overflow_bucket_key_sentinel_values() { assert_eq!(normal.service, "my-service"); assert_eq!(normal.resource, "my-resource"); } + +#[test] +fn test_once_macro() { + let mut count = 0; + for _ in 0..10 { + once!(count += 1); + } + assert_eq!(count, 1, "once! macro executes its body one time only"); +} + +/// Verifies it works inside a called function too (it has to use a static variable) +#[test] +fn test_once_macro_funcall() { + fn incr(target: &mut usize) { + once!(*target += 1) + } + let mut count = 0; + for _ in 0..10 { + incr(&mut count); + } + assert_eq!(count, 1, "once! macro executes its body one time only"); +} From a45213f29e22e573d6e59d8ec1d82ebb9cc5e807 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Thu, 9 Jul 2026 14:05:09 +0200 Subject: [PATCH 12/16] style: cleanup/simplify sending telemetry --- libdd-trace-stats/src/stats_exporter.rs | 131 +++++++++++------------- 1 file changed, 57 insertions(+), 74 deletions(-) diff --git a/libdd-trace-stats/src/stats_exporter.rs b/libdd-trace-stats/src/stats_exporter.rs index 63b6ba64d7..599ecc2a1e 100644 --- a/libdd-trace-stats/src/stats_exporter.rs +++ b/libdd-trace-stats/src/stats_exporter.rs @@ -9,7 +9,7 @@ use std::{ time, }; -use crate::span_concentrator::{FlushableConcentrator, SpanConcentrator}; +use crate::span_concentrator::{FlushResult, FlushableConcentrator, SpanConcentrator}; use async_trait::async_trait; use libdd_capabilities::{HttpClientCapability, MaybeSend, SleepCapability}; use libdd_common::Endpoint; @@ -227,107 +227,90 @@ 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 flush = { + let FlushResult { + obfuscated_buckets, + unobfuscated_buckets, + collapsed_spans, + } = { #[allow(clippy::unwrap_used)] let mut concentrator = self.concentrator.lock().unwrap(); concentrator.flush_buckets(force_flush) }; - if flush.collapsed_spans.whole_key > 0 { - #[cfg(feature = "telemetry")] - if let Some(telemetry) = &self.telemetry { + #[cfg(feature = "telemetry")] + if let Some(telemetry) = &self.telemetry { + if collapsed_spans.whole_key > 0 { let _ = telemetry.handle.add_point( - flush.collapsed_spans.whole_key as f64, + collapsed_spans.whole_key as f64, &telemetry.keys.collapsed_whole_key, vec![], ); } - #[cfg(feature = "dogstatsd")] - if let Some(client) = &self.dogstatsd { - client.send(vec![libdd_dogstatsd_client::DogStatsDAction::Count( - COLLAPSED_SPANS_HEALTH_METRIC, - flush.collapsed_spans.whole_key as i64, - [libdd_common::tag!("collapsed_spans", "whole_key")].iter(), - )]); - } - } - - if flush.collapsed_spans.resources > 0 { - #[cfg(feature = "telemetry")] - if let Some(telemetry) = &self.telemetry { + if collapsed_spans.resources > 0 { let _ = telemetry.handle.add_point( - flush.collapsed_spans.resources as f64, + collapsed_spans.resources as f64, &telemetry.keys.collapsed_resources, vec![], ); } - #[cfg(feature = "dogstatsd")] - if let Some(client) = &self.dogstatsd { - client.send(vec![libdd_dogstatsd_client::DogStatsDAction::Count( - COLLAPSED_SPANS_HEALTH_METRIC, - flush.collapsed_spans.resources as i64, - [libdd_common::tag!("collapsed_spans", "resource")].iter(), - )]); - } - } - - if flush.collapsed_spans.http_endpoint > 0 { - #[cfg(feature = "telemetry")] - if let Some(telemetry) = &self.telemetry { + if collapsed_spans.http_endpoint > 0 { let _ = telemetry.handle.add_point( - flush.collapsed_spans.http_endpoint as f64, + collapsed_spans.http_endpoint as f64, &telemetry.keys.collapsed_http_endpoints, vec![], ); } - #[cfg(feature = "dogstatsd")] - if let Some(client) = &self.dogstatsd { - client.send(vec![libdd_dogstatsd_client::DogStatsDAction::Count( - COLLAPSED_SPANS_HEALTH_METRIC, - flush.collapsed_spans.http_endpoint as i64, - [libdd_common::tag!("collapsed_spans", "http_endpoint")].iter(), - )]); - } - } - - if flush.collapsed_spans.peer_tags > 0 { - #[cfg(feature = "telemetry")] - if let Some(telemetry) = &self.telemetry { + if collapsed_spans.peer_tags > 0 { let _ = telemetry.handle.add_point( - flush.collapsed_spans.peer_tags as f64, + collapsed_spans.peer_tags as f64, &telemetry.keys.collapsed_peer_tags, vec![], ); } - #[cfg(feature = "dogstatsd")] - if let Some(client) = &self.dogstatsd { - client.send(vec![libdd_dogstatsd_client::DogStatsDAction::Count( - COLLAPSED_SPANS_HEALTH_METRIC, - flush.collapsed_spans.peer_tags as i64, - [libdd_common::tag!("collapsed_spans", "peer_tags")].iter(), - )]); - } - } - - if flush.collapsed_spans.additional_tags > 0 { - #[cfg(feature = "telemetry")] - if let Some(telemetry) = &self.telemetry { + if collapsed_spans.additional_tags > 0 { let _ = telemetry.handle.add_point( - flush.collapsed_spans.additional_tags as f64, + collapsed_spans.additional_tags as f64, &telemetry.keys.collapsed_additional_tags, vec![], ); } - #[cfg(feature = "dogstatsd")] - if let Some(client) = &self.dogstatsd { + } + + #[cfg(feature = "dogstatsd")] + if let Some(client) = &self.dogstatsd { + if collapsed_spans.whole_key > 0 { + client.send(vec![libdd_dogstatsd_client::DogStatsDAction::Count( + COLLAPSED_SPANS_HEALTH_METRIC, + collapsed_spans.whole_key as i64, + [libdd_common::tag!("collapsed_spans", "whole_key")].iter(), + )]); + } + if collapsed_spans.resources > 0 { + client.send(vec![libdd_dogstatsd_client::DogStatsDAction::Count( + COLLAPSED_SPANS_HEALTH_METRIC, + collapsed_spans.resources as i64, + [libdd_common::tag!("collapsed_spans", "resources")].iter(), + )]); + } + if collapsed_spans.http_endpoint > 0 { + client.send(vec![libdd_dogstatsd_client::DogStatsDAction::Count( + COLLAPSED_SPANS_HEALTH_METRIC, + collapsed_spans.http_endpoint as i64, + [libdd_common::tag!("collapsed_spans", "http_endpoint")].iter(), + )]); + } + if collapsed_spans.peer_tags > 0 { + client.send(vec![libdd_dogstatsd_client::DogStatsDAction::Count( + COLLAPSED_SPANS_HEALTH_METRIC, + collapsed_spans.peer_tags as i64, + [libdd_common::tag!("collapsed_spans", "peer_tags")].iter(), + )]); + } + if collapsed_spans.additional_tags > 0 { client.send(vec![libdd_dogstatsd_client::DogStatsDAction::Count( COLLAPSED_SPANS_HEALTH_METRIC, - flush.collapsed_spans.additional_tags as i64, - [libdd_common::tag!( - "collapsed_spans", - "additional_metric_tags" - )] - .iter(), + collapsed_spans.additional_tags as i64, + [libdd_common::tag!("collapsed_spans", "additional_tags")].iter(), )]); } } @@ -335,12 +318,12 @@ 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() { - self.send_payload(flush.obfuscated_buckets, true).await?; + if !obfuscated_buckets.is_empty() { + self.send_payload(obfuscated_buckets, true).await?; sent_stats = true; } - if !flush.unobfuscated_buckets.is_empty() { - self.send_payload(flush.unobfuscated_buckets, false).await?; + if !unobfuscated_buckets.is_empty() { + self.send_payload(unobfuscated_buckets, false).await?; sent_stats = true; } Ok(sent_stats) From f17b5b9d47f8e45939010b393df82f46808fb751 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Thu, 9 Jul 2026 14:09:36 +0200 Subject: [PATCH 13/16] fix: additional_tags -> additional_metric_tags --- libdd-trace-stats/src/stats_exporter.rs | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/libdd-trace-stats/src/stats_exporter.rs b/libdd-trace-stats/src/stats_exporter.rs index 599ecc2a1e..9e3d1c8a4e 100644 --- a/libdd-trace-stats/src/stats_exporter.rs +++ b/libdd-trace-stats/src/stats_exporter.rs @@ -3,8 +3,8 @@ use std::{ sync::{ - atomic::{AtomicU64, Ordering}, Arc, Mutex, + atomic::{AtomicU64, Ordering}, }, time, }; @@ -15,7 +15,7 @@ use libdd_capabilities::{HttpClientCapability, MaybeSend, SleepCapability}; 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}; +use libdd_trace_utils::send_with_retry::{RetryStrategy, send_with_retry}; use libdd_trace_utils::trace_utils::TracerHeaderTags; use libdd_trace_utils::tracer_metadata::TracerMetadata; use std::fmt::Debug; @@ -310,7 +310,11 @@ impl client.send(vec![libdd_dogstatsd_client::DogStatsDAction::Count( COLLAPSED_SPANS_HEALTH_METRIC, collapsed_spans.additional_tags as i64, - [libdd_common::tag!("collapsed_spans", "additional_tags")].iter(), + [libdd_common::tag!( + "collapsed_spans", + "additional_metric_tagss" + )] + .iter(), )]); } } @@ -377,9 +381,9 @@ impl #[cfg_attr(not(target_arch = "wasm32"), async_trait)] #[cfg_attr(target_arch = "wasm32", async_trait(?Send))] impl< - Cap: HttpClientCapability + SleepCapability + MaybeSend + Sync + 'static, - Con: FlushableConcentrator + Send + Debug, - > Worker for StatsExporter + Cap: HttpClientCapability + SleepCapability + MaybeSend + Sync + 'static, + Con: FlushableConcentrator + Send + Debug, +> Worker for StatsExporter { async fn trigger(&mut self) { self.capabilities.sleep(self.flush_interval).await; @@ -438,8 +442,8 @@ mod tests { use crate::span_concentrator::CardinalityLimitConfig; #[cfg(feature = "stats-obfuscation")] use crate::span_concentrator::StatsComputationObfuscationConfig; - use httpmock::prelude::*; use httpmock::MockServer; + use httpmock::prelude::*; use libdd_capabilities_impl::NativeCapabilities; use libdd_shared_runtime::{BlockingRuntime, ForkSafeRuntime, SharedRuntime}; use libdd_trace_utils::span::{trace_utils, v04::SpanSlice}; From e887c99104fec27ebec28d133da26cbfeb9363a0 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Thu, 9 Jul 2026 14:24:42 +0200 Subject: [PATCH 14/16] fix: test compile warnings with no features --- 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 9e3d1c8a4e..99bf5b08b6 100644 --- a/libdd-trace-stats/src/stats_exporter.rs +++ b/libdd-trace-stats/src/stats_exporter.rs @@ -230,6 +230,7 @@ impl let FlushResult { obfuscated_buckets, unobfuscated_buckets, + #[cfg_attr(not(any(feature = "telemetry", feature = "dogstatsd")), allow(unused))] collapsed_spans, } = { #[allow(clippy::unwrap_used)] @@ -439,7 +440,6 @@ 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 httpmock::MockServer; @@ -752,7 +752,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( From d540ebb2aed63234dd6a790d327bba159f290b2f Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Thu, 9 Jul 2026 14:24:59 +0200 Subject: [PATCH 15/16] fix: replace OnceLock -> Once --- libdd-trace-stats/src/span_concentrator/mod.rs | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/libdd-trace-stats/src/span_concentrator/mod.rs b/libdd-trace-stats/src/span_concentrator/mod.rs index 7106865b38..61fb573e43 100644 --- a/libdd-trace-stats/src/span_concentrator/mod.rs +++ b/libdd-trace-stats/src/span_concentrator/mod.rs @@ -2,7 +2,6 @@ // SPDX-License-Identifier: Apache-2.0 //! This module implements the SpanConcentrator used to aggregate spans into stats use std::collections::HashMap; -use std::sync::OnceLock; use std::time::{self, Duration, SystemTime}; use tracing::{debug, warn}; @@ -21,11 +20,10 @@ pub mod stat_span; pub use stat_span::StatSpan; /// Executes a statement one time. Useful when emitting a warning once -#[macro_export] macro_rules! once { ($expression:expr) => {{ - static LOGGED: OnceLock<()> = OnceLock::new(); - LOGGED.get_or_init(|| { + static LOGGED: std::sync::Once = std::sync::Once::new(); + LOGGED.call_once(|| { $expression; }); }}; @@ -428,7 +426,8 @@ impl SpanConcentrator { if total_collapsed.additional_tags > 0 { once!(debug!( - max_distinct_additional_tags_per_bucket = self.cardinality_limits.additional_tags_limit, + max_distinct_additional_tags_per_bucket = + self.cardinality_limits.additional_tags_limit, total_additional_tags_collapsed = total_collapsed.additional_tags, "Client-side stats field 'additional_tags' has been collapsed to 'tracer_blocked_value'. This is due to the cardinality exceeding DD_TRACE_STATS_ADDITIONAL_TAGS_CARDINALITY_LIMIT" )); From 397343518a7c0aa159014b076478e07cc6074de7 Mon Sep 17 00:00:00 2001 From: Oscar Le Dauphin Date: Thu, 9 Jul 2026 14:42:27 +0200 Subject: [PATCH 16/16] fix: bad telemetry tag name again --- libdd-trace-stats/src/stats_exporter.rs | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/libdd-trace-stats/src/stats_exporter.rs b/libdd-trace-stats/src/stats_exporter.rs index 99bf5b08b6..19d17e7d12 100644 --- a/libdd-trace-stats/src/stats_exporter.rs +++ b/libdd-trace-stats/src/stats_exporter.rs @@ -3,8 +3,8 @@ use std::{ sync::{ - Arc, Mutex, atomic::{AtomicU64, Ordering}, + Arc, Mutex, }, time, }; @@ -15,7 +15,7 @@ use libdd_capabilities::{HttpClientCapability, MaybeSend, SleepCapability}; use libdd_common::Endpoint; use libdd_shared_runtime::Worker; use libdd_trace_protobuf::pb; -use libdd_trace_utils::send_with_retry::{RetryStrategy, send_with_retry}; +use libdd_trace_utils::send_with_retry::{send_with_retry, RetryStrategy}; use libdd_trace_utils::trace_utils::TracerHeaderTags; use libdd_trace_utils::tracer_metadata::TracerMetadata; use std::fmt::Debug; @@ -290,7 +290,7 @@ impl client.send(vec![libdd_dogstatsd_client::DogStatsDAction::Count( COLLAPSED_SPANS_HEALTH_METRIC, collapsed_spans.resources as i64, - [libdd_common::tag!("collapsed_spans", "resources")].iter(), + [libdd_common::tag!("collapsed_spans", "resource")].iter(), )]); } if collapsed_spans.http_endpoint > 0 { @@ -313,7 +313,7 @@ impl collapsed_spans.additional_tags as i64, [libdd_common::tag!( "collapsed_spans", - "additional_metric_tagss" + "additional_metric_tags" )] .iter(), )]); @@ -382,9 +382,9 @@ impl #[cfg_attr(not(target_arch = "wasm32"), async_trait)] #[cfg_attr(target_arch = "wasm32", async_trait(?Send))] impl< - Cap: HttpClientCapability + SleepCapability + MaybeSend + Sync + 'static, - Con: FlushableConcentrator + Send + Debug, -> Worker for StatsExporter + Cap: HttpClientCapability + SleepCapability + MaybeSend + Sync + 'static, + Con: FlushableConcentrator + Send + Debug, + > Worker for StatsExporter { async fn trigger(&mut self) { self.capabilities.sleep(self.flush_interval).await; @@ -442,8 +442,8 @@ mod tests { use super::*; #[cfg(feature = "stats-obfuscation")] use crate::span_concentrator::StatsComputationObfuscationConfig; - use httpmock::MockServer; use httpmock::prelude::*; + use httpmock::MockServer; use libdd_capabilities_impl::NativeCapabilities; use libdd_shared_runtime::{BlockingRuntime, ForkSafeRuntime, SharedRuntime}; use libdd_trace_utils::span::{trace_utils, v04::SpanSlice};