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-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"] 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 03b9a4da02..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; @@ -49,7 +51,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`]. @@ -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 @@ -142,7 +144,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..3a5c369708 100644 --- a/libdd-trace-stats/src/span_concentrator/aggregation.rs +++ b/libdd-trace-stats/src/span_concentrator/aggregation.rs @@ -5,14 +5,19 @@ //! 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; -use std::borrow::{Borrow, Cow}; +use std::{ + borrow::{Borrow, Cow}, + hash::{DefaultHasher, Hash, Hasher as _}, +}; 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 +304,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())], } } } @@ -420,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)] @@ -428,9 +458,14 @@ 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, - /// Number of spans collapsed into the overflow bucket due to cardinality limiting. - collapsed_count: u64, + 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, + 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")] @@ -440,26 +475,30 @@ 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, - collapsed_count: 0, + cardinality_limits, #[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 @@ -467,25 +506,71 @@ 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) { - self.collapsed_count += 1; + // 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) + { + self.collapsed_counts.whole_key += 1; self.data .entry(OwnedAggregationKey::overflow_key()) .or_default() .insert(duration, is_error, is_top_level); return; } + 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_fields_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(&resource_name_hash) + { + key.fixed.resource_name = TRACER_BLOCKED_VALUE; + self.collapsed_counts.resources += 1; + } else { + 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(&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); + } + + let peer_tags_hash = hash(&key.peer_tags); + if self.distinct_peer_tags.len() >= self.cardinality_limits.peer_tags_limit + && !self.distinct_peer_tags.contains(&peer_tags_hash) + { + key.peer_tags = vec![(TRACER_BLOCKED_VALUE, Cow::Borrowed(""))]; + self.collapsed_counts.peer_tags += 1; + } else { + self.distinct_peer_tags.insert(peer_tags_hash); + } + } + /// 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 +635,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..61fb573e43 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; @@ -11,11 +11,24 @@ 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; +/// Executes a statement one time. Useful when emitting a warning once +macro_rules! once { + ($expression:expr) => {{ + static LOGGED: std::sync::Once = std::sync::Once::new(); + LOGGED.call_once(|| { + $expression; + }); + }}; +} + /// Result of flushing a concentrator. /// /// Obfuscated and un-obfuscated buckets are kept separate because they must be sent in distinct @@ -26,8 +39,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 { @@ -100,6 +113,34 @@ 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, PartialEq, Eq)] +#[repr(C)] +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 +171,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,11 +195,25 @@ 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, >, ) -> 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(), @@ -167,8 +222,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 +272,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, ) @@ -302,7 +356,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(); @@ -312,7 +366,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)| { @@ -329,7 +383,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"))] @@ -337,9 +391,48 @@ impl SpanConcentrator { Some((obfuscated, encode(bucket, self.bucket_size))) }) .collect(); - if total_collapsed > 0 { - debug!(max_entries_per_bucket = self.max_entries_per_bucket, total_collapsed, "Client-side stats values have been collapsed to 'tracer_blocked_value'. This is due to the cardinality exceeding DD_TRACE_STATS_CARDINALITY_LIMIT"); + + 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 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 '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 81d7e94778..443ac794ea 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), + vec!["peer.hostname".to_owned()], + 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,185 @@ 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).all_buckets(); + 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 key + 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" + ); +} + +/// 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() { 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 +1949,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 +1993,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 +2078,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 @@ -1906,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"); +} diff --git a/libdd-trace-stats/src/stats_exporter.rs b/libdd-trace-stats/src/stats_exporter.rs index 47c6bfe935..19d17e7d12 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; @@ -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, @@ -173,36 +227,108 @@ 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, + #[cfg_attr(not(any(feature = "telemetry", feature = "dogstatsd")), allow(unused))] + collapsed_spans, + } = { #[allow(clippy::unwrap_used)] let mut concentrator = self.concentrator.lock().unwrap(); 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(telemetry) = &self.telemetry { + if collapsed_spans.whole_key > 0 { + let _ = telemetry.handle.add_point( + collapsed_spans.whole_key as f64, + &telemetry.keys.collapsed_whole_key, + vec![], + ); } - #[cfg(feature = "dogstatsd")] - if let Some(client) = &self.dogstatsd { + if collapsed_spans.resources > 0 { + let _ = telemetry.handle.add_point( + collapsed_spans.resources as f64, + &telemetry.keys.collapsed_resources, + vec![], + ); + } + if collapsed_spans.http_endpoint > 0 { + let _ = telemetry.handle.add_point( + collapsed_spans.http_endpoint as f64, + &telemetry.keys.collapsed_http_endpoints, + vec![], + ); + } + if collapsed_spans.peer_tags > 0 { + let _ = telemetry.handle.add_point( + collapsed_spans.peer_tags as f64, + &telemetry.keys.collapsed_peer_tags, + vec![], + ); + } + if collapsed_spans.additional_tags > 0 { + let _ = telemetry.handle.add_point( + collapsed_spans.additional_tags as f64, + &telemetry.keys.collapsed_additional_tags, + vec![], + ); + } + } + + #[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, - flush.collapsed_spans as i64, + 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", "resource")].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, + 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; - 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) @@ -314,6 +440,8 @@ pub fn stats_url_from_agent_url(agent_url: &str) -> anyhow::Result { #[cfg(test)] mod tests { use super::*; + #[cfg(feature = "stats-obfuscation")] + use crate::span_concentrator::StatsComputationObfuscationConfig; use httpmock::prelude::*; use httpmock::MockServer; use libdd_capabilities_impl::NativeCapabilities; @@ -580,7 +708,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; @@ -625,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( @@ -633,33 +762,37 @@ mod tests { SystemTime::now(), vec![], vec![], - Some(1), // max 1 distinct key → second span collapses + Some(CardinalityLimitConfig { + 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")] None, ); 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() }, @@ -763,12 +896,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" ); } @@ -814,23 +958,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" ); } }