diff --git a/datadog-ipc/src/shm_stats.rs b/datadog-ipc/src/shm_stats.rs index 87b55981da..72687985a2 100644 --- a/datadog-ipc/src/shm_stats.rs +++ b/datadog-ipc/src/shm_stats.rs @@ -57,7 +57,8 @@ use zwohash::ZwoHasher; use libdd_ddsketch::DDSketch; use libdd_trace_protobuf::pb; use libdd_trace_stats::span_concentrator::{ - FixedAggregationKey, FlushResult, FlushableConcentrator, + cardinality_limit_telemetry::CollapsedFieldsMetrics, FixedAggregationKey, FlushResult, + FlushableConcentrator, }; use crate::platform::{FileBackedHandle, MappedMem, NamedShmHandle}; @@ -822,12 +823,13 @@ impl ShmSpanConcentrator { } impl FlushableConcentrator for ShmSpanConcentrator { - fn flush_buckets(&mut self, force: bool) -> FlushResult { - // The SHM concentrator does not perform client-side obfuscation. + fn flush_buckets(&mut self, force: bool) -> FlushResult { + // The SHM concentrator does not perform client-side obfuscation nor emits telemetry. FlushResult { obfuscated_buckets: vec![], unobfuscated_buckets: self.drain_buckets(force), collapsed_spans: 0, + collapsed_fields_metrics: CollapsedFieldsMetrics::zero(), } } } diff --git a/libdd-trace-stats/src/span_concentrator/aggregation.rs b/libdd-trace-stats/src/span_concentrator/aggregation.rs index 2c6bda079d..934d99bcdc 100644 --- a/libdd-trace-stats/src/span_concentrator/aggregation.rs +++ b/libdd-trace-stats/src/span_concentrator/aggregation.rs @@ -15,9 +15,12 @@ use std::{ }; use tracing::warn; -use crate::span_concentrator::StatSpan; +use crate::span_concentrator::{cardinality_limit_telemetry::CollapsedFieldSet, StatSpan}; -use super::CardinalityLimitConfig; +use super::{ + cardinality_limit_telemetry::{self, CollapsedFieldsMetrics}, + CardinalityLimitConfig, +}; /// Sentinel value used for cardinality limiting. pub const TRACER_BLOCKED_VALUE: &str = "tracer_blocked_value"; @@ -501,8 +504,9 @@ pub(super) struct StatsBucket { distinct_http_endpoints: HashSet, distinct_peer_tags: HashSet, distinct_additional_tags: HashSet, - /// Number of spans collapsed into the overflow bucket due to cardinality limiting. + /// Number of spans collapsed into the overflow bucket due to whole-key cardinality limiting. collapsed_count: u64, + collapsed_fields_metrics: CollapsedFieldsMetrics, /// Indicates if stats obfuscated in this bucket. This is set once at creation and stays /// constant per bucket #[cfg(feature = "stats-obfuscation")] @@ -529,9 +533,15 @@ impl StatsBucket { distinct_http_endpoints: HashSet::new(), distinct_peer_tags: HashSet::new(), distinct_additional_tags: HashSet::new(), + collapsed_fields_metrics: cardinality_limit_telemetry::CollapsedFieldsMetrics::zero(), } } + /// Returns metrics on spans field collapse with reasons. + pub fn collapsed_fields_metrics(&self) -> cardinality_limit_telemetry::CollapsedFieldsMetrics { + self.collapsed_fields_metrics + } + /// Return the number of spans collapsed into the overflow bucket. pub(super) fn collapsed_count(&self) -> u64 { self.collapsed_count @@ -588,11 +598,14 @@ impl StatsBucket { hasher.finish() } + let mut collapsed_fields = CollapsedFieldSet::empty(); + let resource_hash = hash(&key.fixed.resource_name); let resources_count = self.distinct_resources.len(); if let Entry::Vacant(slot) = self.distinct_resources.entry(resource_hash) { if resources_count >= self.cardinality_limits.resource_limit { key.fixed.resource_name = TRACER_BLOCKED_VALUE; + collapsed_fields.add(CollapsedFieldSet::RESOURCE_NAME); } else { slot.insert(); } @@ -603,6 +616,7 @@ impl StatsBucket { if let Entry::Vacant(slot) = self.distinct_http_endpoints.entry(http_endpoint_hash) { if http_endpoints_count >= self.cardinality_limits.http_endpoint_limit { key.fixed.http_endpoint = TRACER_BLOCKED_VALUE; + collapsed_fields.add(CollapsedFieldSet::HTTP_ENDPOINT); } else { slot.insert(); } @@ -613,6 +627,7 @@ impl StatsBucket { if let Entry::Vacant(slot) = self.distinct_peer_tags.entry(peer_tags_hash) { if peer_tags_count >= self.cardinality_limits.peer_tags_limit { key.peer_tags = vec![(TRACER_BLOCKED_VALUE, Cow::Borrowed(""))]; + collapsed_fields.add(CollapsedFieldSet::PEER_TAGS); } else { slot.insert(); } @@ -623,10 +638,12 @@ impl StatsBucket { if let Entry::Vacant(slot) = self.distinct_additional_tags.entry(additional_tags_hash) { if additional_tags_count >= self.cardinality_limits.additional_tags_limit { key.additional_metric_tags = vec![(TRACER_BLOCKED_VALUE, "")]; + collapsed_fields.add(CollapsedFieldSet::ADDITIONAL_TAGS); } else { slot.insert(); } } + self.collapsed_fields_metrics.increment(collapsed_fields); } /// Consume the bucket and return a ClientStatsBucket containing the bucket stats. diff --git a/libdd-trace-stats/src/span_concentrator/cardinality_limit_telemetry.rs b/libdd-trace-stats/src/span_concentrator/cardinality_limit_telemetry.rs new file mode 100644 index 0000000000..7711af02d7 --- /dev/null +++ b/libdd-trace-stats/src/span_concentrator/cardinality_limit_telemetry.rs @@ -0,0 +1,136 @@ +// Copyright 2024-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +//! This module implement the logic for sending telemetry/dogstatsd related to cardinality limits +// Because the cardinality limit RFC requires one point of telemetry with a set of collapsed fields +// and not one point for each collapsed field, this solution was found as an alternative to adding +// telemetry and dogstatsd client in the SpanConcentrator to keep it as "pure computation logic" + +#[cfg(feature = "telemetry")] +use libdd_capabilities::{HttpClientCapability, MaybeSend, SleepCapability}; +use libdd_common::tag::const_assert; + +/// Bitset of collapsed stats key fields +pub struct CollapsedFieldSet(usize); +impl CollapsedFieldSet { + pub const RESOURCE_NAME: usize = 1 << 0; + pub const HTTP_ENDPOINT: usize = 1 << 1; + pub const PEER_TAGS: usize = 1 << 2; + pub const ADDITIONAL_TAGS: usize = 1 << 3; + + const FIELDS: [usize; 4] = [ + Self::RESOURCE_NAME, + Self::HTTP_ENDPOINT, + Self::PEER_TAGS, + Self::ADDITIONAL_TAGS, + ]; + + /// Default value with none of the fields set (no collapsed fields) + pub fn empty() -> CollapsedFieldSet { + Self(0) + } + + /// Enable the bit corresponding to `field` + pub fn add(&mut self, field: usize) { + debug_assert!(Self::FIELDS.contains(&field)); + self.0 |= field; + } +} + +const COLLAPSED_FIELD_METRIC_SIZE: usize = 1 << CollapsedFieldSet::FIELDS.len(); +// Verify the array is of a reasonable size +const_assert!(COLLAPSED_FIELD_METRIC_SIZE <= 16); +/// Counter of combination of collapsed stats key fields +// Note: slot 0 is a counter for non_collapsed spans. It's not used for emitting telemetry +#[derive(Debug, Clone, Default, Copy)] +pub struct CollapsedFieldsMetrics([usize; COLLAPSED_FIELD_METRIC_SIZE]); + +impl CollapsedFieldsMetrics { + /// Default value every combination of collapsed fields set to 0 + pub fn zero() -> Self { + Self::default() + } + + #[cfg(feature = "dogstatsd")] + pub fn emit_dogstatsd(&self, dogstatsd: &libdd_dogstatsd_client::DogStatsDClient) { + // skip the first slot that is used to count span which have no collapsed fields + for (mask, &count) in self.0.iter().enumerate().skip(1) { + if count > 0 { + let tags = Self::fields_mask_to_list(mask); + dogstatsd.send(vec![libdd_dogstatsd_client::DogStatsDAction::Count( + "datadog.tracer.stats.collapsed_spans", + count as i64, + tags.iter(), + )]); + } + } + } + + #[cfg(feature = "telemetry")] + pub fn emit_telemetry< + Cap: HttpClientCapability + SleepCapability + MaybeSend + Sync + 'static, + >( + &self, + handle: &libdd_telemetry::worker::TelemetryWorkerHandle, + context_key: &libdd_telemetry::metrics::ContextKey, + ) { + // skip the first slot that is used to count span which have no collapsed fields + for (mask, &count) in self.0.iter().enumerate().skip(1) { + if count > 0 { + let tags = Self::fields_mask_to_list(mask); + let _ = handle.add_point(count as f64, context_key, tags); + } + } + } + + /// Given a bitmask of collapsed fields, returns the list of tags to attach to + /// telemetry/dogstatsd + #[cfg(any(feature = "telemetry", feature = "dogstatsd"))] + fn fields_mask_to_list(mask: usize) -> Vec { + let mut tags = Vec::new(); + for field_pow in 0..CollapsedFieldSet::FIELDS.len() { + let field_value = 1 << field_pow; + debug_assert!( + CollapsedFieldSet::FIELDS.contains(&field_value), + "{field_value} is an invalid value for a CollapsedFieldSet" + ); + let has_field = (mask & field_value) != 0; + if !has_field { + continue; + } + let field_tag = match field_value { + CollapsedFieldSet::RESOURCE_NAME => { + libdd_common::tag!("collapsed_spans", "resource") + } + CollapsedFieldSet::HTTP_ENDPOINT => { + libdd_common::tag!("collapsed_spans", "http_endpoint") + } + CollapsedFieldSet::PEER_TAGS => { + libdd_common::tag!("collapsed_spans", "peer_tags") + } + CollapsedFieldSet::ADDITIONAL_TAGS => { + libdd_common::tag!("collapsed_spans", "additional_metric_tags") + } + // Should be unreachable, but don't fail in prod if provided with an invalid field + // set + _ => continue, + }; + tags.push(field_tag); + } + debug_assert!(!tags.is_empty()); + tags + } + + /// Increment the telemetry counter corresponding to this collapsed field combination + pub fn increment(&mut self, field_set: CollapsedFieldSet) { + self.0[field_set.0] += 1; + } +} + +impl std::ops::AddAssign for CollapsedFieldsMetrics { + fn add_assign(&mut self, rhs: Self) { + for i in 0..self.0.len() { + self.0[i] += rhs.0[i]; + } + } +} diff --git a/libdd-trace-stats/src/span_concentrator/mod.rs b/libdd-trace-stats/src/span_concentrator/mod.rs index 08f7179b72..d4ad0292e1 100644 --- a/libdd-trace-stats/src/span_concentrator/mod.rs +++ b/libdd-trace-stats/src/span_concentrator/mod.rs @@ -1,6 +1,10 @@ // Copyright 2024-Present Datadog, Inc. https://www.datadoghq.com/ // SPDX-License-Identifier: Apache-2.0 //! This module implements the SpanConcentrator used to aggregate spans into stats +mod aggregation; +pub mod cardinality_limit_telemetry; +pub mod stat_span; + use std::collections::HashMap; use std::time::Duration; use tracing::{debug, warn}; @@ -11,11 +15,10 @@ use libdd_trace_protobuf::pb; use aggregation::StatsBucket; -mod aggregation; use aggregation::BorrowedAggregationKey; pub use aggregation::{FixedAggregationKey, OtlpExactCell, OtlpExactGroup, OtlpStatsBucket}; +use cardinality_limit_telemetry::CollapsedFieldsMetrics; -pub mod stat_span; pub use stat_span::StatSpan; const ADDITIONAL_METRIC_TAGS_MAX_KEYS: usize = 4; @@ -41,20 +44,20 @@ fn normalize_additional_metric_tag_keys(mut keys: Vec) -> Vec { /// /// Obfuscated and un-obfuscated buckets are kept separate because they must be sent in distinct /// stats payloads: only the obfuscated payload carries the `datadog-obfuscation-version` header. -pub struct FlushResult { +pub struct FlushResult { /// Buckets whose resource names were obfuscated client-side. - pub obfuscated_buckets: Vec, + pub obfuscated_buckets: Vec, /// Buckets whose resource names were left as-is. - pub unobfuscated_buckets: Vec, + pub unobfuscated_buckets: Vec, /// Total number of spans that were collapsed into the overflow sentinel bucket due to /// cardinality limiting across all flushed time buckets. pub collapsed_spans: u64, + pub collapsed_fields_metrics: CollapsedFieldsMetrics, } -impl FlushResult { +impl FlushResult { /// All flushed buckets regardless of obfuscation. - #[cfg(test)] - pub fn all_buckets(self) -> Vec { + pub fn all_buckets(self) -> Vec { let mut buckets = self.obfuscated_buckets; buckets.extend(self.unobfuscated_buckets); buckets @@ -68,11 +71,11 @@ impl FlushResult { pub trait FlushableConcentrator { /// Flush time buckets and return them together with flush metadata. If `force` is true, flush /// all buckets. See [`FlushResult`] for the returned data. - fn flush_buckets(&mut self, force: bool) -> FlushResult; + fn flush_buckets(&mut self, force: bool) -> FlushResult; } impl FlushableConcentrator for SpanConcentrator { - fn flush_buckets(&mut self, force: bool) -> FlushResult { + fn flush_buckets(&mut self, force: bool) -> FlushResult { self.flush(SystemTime::now(), force) } } @@ -361,30 +364,16 @@ impl SpanConcentrator { /// all buckets. /// /// Obfuscated and un-obfuscated buckets are returned separately, see [`FlushResult`]. - pub fn flush(&mut self, now: SystemTime, force: bool) -> FlushResult { - let (buckets, collapsed_spans) = self.drain_due_buckets(now, force, StatsBucket::flush); - let mut obfuscated_buckets = Vec::new(); - let mut unobfuscated_buckets = Vec::new(); - for (obfuscated, bucket) in buckets { - if obfuscated { - obfuscated_buckets.push(bucket); - } else { - unobfuscated_buckets.push(bucket); - } - } - FlushResult { - obfuscated_buckets, - unobfuscated_buckets, - collapsed_spans, - } + pub fn flush(&mut self, now: SystemTime, force: bool) -> FlushResult { + self.drain_due_buckets(now, force, StatsBucket::flush) } /// Like [`Self::flush`], but also emits exact per-cell scalars alongside each bucket for the /// OTLP trace-metrics path. The protobuf bucket inside each [`OtlpStatsBucket`] is identical /// to what [`Self::flush`] would produce, so the /v0.6/stats agent path is unaffected. pub fn flush_with_otlp_exact(&mut self, now: SystemTime, force: bool) -> Vec { - let (buckets, _) = self.drain_due_buckets(now, force, StatsBucket::flush_with_otlp_exact); - buckets.into_iter().map(|(_, bucket)| bucket).collect() + self.drain_due_buckets(now, force, StatsBucket::flush_with_otlp_exact) + .all_buckets() } /// Drain the buckets that are due for flushing, encoding each with `encode`. @@ -398,7 +387,7 @@ impl SpanConcentrator { now: SystemTime, force: bool, encode: impl Fn(StatsBucket, u64) -> T, - ) -> (Vec<(bool, T)>, u64) { + ) -> FlushResult { // TODO: Wait for HashMap::extract_if to be stabilized to avoid a full drain let now_timestamp = system_time_to_unix_duration(now).as_nanos() as u64; let buckets: Vec<(u64, StatsBucket)> = self.buckets.drain().collect(); @@ -408,8 +397,9 @@ impl SpanConcentrator { align_timestamp(now_timestamp, self.bucket_size) - (self.buffer_len as u64 - 1) * self.bucket_size }; - let mut total_collapsed = 0; - let buckets_pb = buckets + let mut collapsed_spans = 0; + let mut collapsed_fields_metrics = CollapsedFieldsMetrics::zero(); + let buckets_pb: Vec<(bool, T)> = buckets .into_iter() .filter_map(|(timestamp, bucket)| { // Always keep `bufferLen` buckets (default is 2: current + previous one). @@ -425,7 +415,8 @@ impl SpanConcentrator { self.buckets.insert(timestamp, bucket); return None; } - total_collapsed += bucket.collapsed_count(); + collapsed_spans += bucket.collapsed_count(); + collapsed_fields_metrics += bucket.collapsed_fields_metrics(); #[cfg(feature = "stats-obfuscation")] let obfuscated = bucket.obfuscated; #[cfg(not(feature = "stats-obfuscation"))] @@ -433,14 +424,30 @@ impl SpanConcentrator { Some((obfuscated, encode(bucket, self.bucket_size))) }) .collect(); - if total_collapsed > 0 { + if collapsed_spans > 0 { debug!( max_entries_per_bucket = self.cardinality_limits.whole_key_limit, - total_collapsed, + collapsed_spans, "Client-side stats values have been collapsed to 'tracer_blocked_value'. This is due to the cardinality exceeding DD_TRACE_STATS_CARDINALITY_LIMIT" ); } - (buckets_pb, total_collapsed) + + let mut obfuscated_buckets = Vec::new(); + let mut unobfuscated_buckets = Vec::new(); + for (obfuscated, bucket) in buckets_pb { + if obfuscated { + obfuscated_buckets.push(bucket); + } else { + unobfuscated_buckets.push(bucket); + } + } + + FlushResult { + obfuscated_buckets, + unobfuscated_buckets, + collapsed_spans, + collapsed_fields_metrics, + } } } diff --git a/libdd-trace-stats/src/stats_exporter.rs b/libdd-trace-stats/src/stats_exporter.rs index a7a9efbada..857187c33f 100644 --- a/libdd-trace-stats/src/stats_exporter.rs +++ b/libdd-trace-stats/src/stats_exporter.rs @@ -137,7 +137,7 @@ impl< let telemetry = telemetry.map(|handle| { let key = handle.register_metric_context( COLLAPSED_SPANS_TELEMETRY_METRIC.to_string(), - vec![libdd_common::tag!("collapsed_spans", "whole_key")], + vec![], libdd_telemetry::data::metrics::MetricType::Count, true, libdd_telemetry::data::metrics::MetricNamespace::Tracers, @@ -182,19 +182,28 @@ impl< concentrator.flush_buckets(force_flush) }; - if flush.collapsed_spans > 0 { - #[cfg(feature = "telemetry")] - if let Some((handle, key)) = &self.telemetry { - let _ = handle.add_point(flush.collapsed_spans as f64, key, vec![]); + #[cfg(feature = "telemetry")] + if let Some((handle, key)) = &self.telemetry { + if flush.collapsed_spans > 0 { + let _ = handle.add_point( + flush.collapsed_spans as f64, + key, + vec![libdd_common::tag!("collapsed_spans", "whole_key")], + ); } - #[cfg(feature = "dogstatsd")] - if let Some(client) = &self.dogstatsd { + flush.collapsed_fields_metrics.emit_telemetry(handle, key); + } + + #[cfg(feature = "dogstatsd")] + if let Some(client) = &self.dogstatsd { + if flush.collapsed_spans > 0 { client.send(vec![libdd_dogstatsd_client::DogStatsDAction::Count( COLLAPSED_SPANS_HEALTH_METRIC, flush.collapsed_spans as i64, [libdd_common::tag!("collapsed_spans", "whole_key")].iter(), )]); } + flush.collapsed_fields_metrics.emit_dogstatsd(client); } let futures = FuturesUnordered::new(); @@ -329,7 +338,6 @@ pub fn stats_url_from_agent_url(agent_url: &str) -> anyhow::Result { #[cfg(test)] mod tests { use super::*; - use crate::span_concentrator::CardinalityLimitConfig; use httpmock::prelude::*; use httpmock::MockServer; use libdd_capabilities_impl::NativeCapabilities; @@ -642,18 +650,31 @@ mod tests { /// Build a concentrator with `max_entries_per_bucket = 1` pre-seeded with four distinct spans /// so that three spans are collapsed into the overflow bucket. - fn get_collapsed_concentrator() -> SpanConcentrator { - use libdd_trace_utils::span::{trace_utils, v04::SpanSlice}; + /// + /// per_key_collapsed: enable small per-key limits to also collapse on `resource` and + /// `http_endpoint` + #[cfg(any(feature = "telemetry", feature = "dogstatsd"))] + fn get_collapsed_concentrator(per_key_collapsed: bool) -> SpanConcentrator { + use crate::span_concentrator::CardinalityLimitConfig; + use libdd_trace_utils::span::{ + trace_utils, + v04::{SpanSlice, VecMap}, + }; + let mut cardinality_limit_config = CardinalityLimitConfig { + whole_key_limit: 2, // max 2 distinct key → third distinct span collapses + ..Default::default() + }; + if per_key_collapsed { + cardinality_limit_config.resource_limit = 1; + cardinality_limit_config.http_endpoint_limit = 1; + } let mut concentrator = SpanConcentrator::new( BUCKETS_DURATION, SystemTime::now(), vec![], vec![], - Some(CardinalityLimitConfig { - whole_key_limit: 1, // max 1 distinct key → second span collapses - ..Default::default() - }), + Some(cardinality_limit_config), vec![], #[cfg(feature = "stats-obfuscation")] None, @@ -661,26 +682,32 @@ mod tests { let mut trace = vec![ SpanSlice { - service: "svc", + service: "svc-a", resource: "resource-a", duration: 10, + meta: VecMap::from_iter([("http.endpoint", "/")]), ..Default::default() }, + // only resource get collapsed if per-key limits are enabled SpanSlice { - service: "svc", + service: "svc-a", resource: "resource-b", duration: 20, + meta: VecMap::from_iter([("http.endpoint", "/")]), ..Default::default() }, + // both resource and http_endpoint get collapsed if per-key limits are enabled SpanSlice { - service: "svc", + service: "svc-b", resource: "resource-c", duration: 20, + meta: VecMap::from_iter([("http.endpoint", "/hello.txt")]), ..Default::default() }, + // both resource and http_endpoint get collapsed if per-key limits are enabled SpanSlice { - service: "svc", - resource: "resource-d", + service: "svc-b", + resource: "resource-b", duration: 20, ..Default::default() }, @@ -738,7 +765,8 @@ mod tests { let result = socket.recv(&mut buf); assert!( result.is_err(), - "No DogStatsD datagram expected when collapsed_spans == 0" + "No DogStatsD datagram expected when collapsed_spans == 0. Got {}", + std::str::from_utf8(&buf[..result.unwrap()]).unwrap() ); } @@ -768,7 +796,7 @@ mod tests { let stats_exporter = StatsExporter::::new( BUCKETS_DURATION, - Arc::new(Mutex::new(get_collapsed_concentrator())), + Arc::new(Mutex::new(get_collapsed_concentrator(false))), get_test_metadata(), Endpoint::from_url(stats_url_from_agent_url(&server.url("/")).unwrap()), NativeCapabilities::new_client(), @@ -787,7 +815,78 @@ mod tests { .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:whole_key", + "DogStatsD datagram must match the expected format" + ); + } + + /// Verify that `COLLAPSED_SPANS_METRIC` is emitted to DogStatsD when spans are collapsed by + /// per-key limits. + #[cfg(feature = "dogstatsd")] + #[cfg_attr(miri, ignore)] + #[tokio::test] + async fn test_collapsed_spans_per_key_dogstatsd() { + use std::net; + + let server = MockServer::start_async().await; + server + .mock_async(|_when, then| { + then.status(200).body(""); + }) + .await; + + let socket = net::UdpSocket::bind("127.0.0.1:0").expect("failed to bind UDP socket"); + socket + .set_read_timeout(Some(std::time::Duration::from_millis(500))) + .unwrap(); + let addr = socket.local_addr().unwrap().to_string(); + + let dogstatsd_client = + libdd_dogstatsd_client::DogStatsDClient::new(libdd_common::Endpoint::from_slice(&addr)) + .expect("failed to create dogstatsd client"); + + let stats_exporter = StatsExporter::::new( + BUCKETS_DURATION, + Arc::new(Mutex::new(get_collapsed_concentrator(true))), + get_test_metadata(), + Endpoint::from_url(stats_url_from_agent_url(&server.url("/")).unwrap()), + NativeCapabilities::new_client(), + #[cfg(feature = "stats-obfuscation")] + "1", + #[cfg(feature = "telemetry")] + None, + Some(dogstatsd_client), + ); + + stats_exporter.send(true).await.unwrap(); + + let mut buf = [0u8; 256]; + + // Whole-key metric emitted first + let n = socket + .recv(&mut buf) + .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-key collapse telemetry + let n = socket + .recv(&mut buf) + .expect("expected a DogStatsD datagram"); + let datagram = std::str::from_utf8(&buf[..n]).expect("valid utf-8"); + assert_eq!( + datagram, "datadog.tracer.stats.collapsed_spans:1|c|#collapsed_spans:resource", + "DogStatsD datagram must match the expected format" + ); + let n = socket + .recv(&mut buf) + .expect("expected a DogStatsD datagram"); + let datagram = std::str::from_utf8(&buf[..n]).expect("valid utf-8"); + assert_eq!( + datagram, "datadog.tracer.stats.collapsed_spans:2|c|#collapsed_spans:resource,collapsed_spans:http_endpoint", "DogStatsD datagram must match the expected format" ); } @@ -818,7 +917,7 @@ mod tests { let stats_exporter = StatsExporter::::new( BUCKETS_DURATION, - Arc::new(Mutex::new(get_collapsed_concentrator())), + Arc::new(Mutex::new(get_collapsed_concentrator(true))), get_test_metadata(), Endpoint::from_url(stats_url_from_agent_url(&server.url("/")).unwrap()), NativeCapabilities::new_client(), @@ -848,8 +947,8 @@ mod tests { "exactly one metric context (COLLAPSED_SPANS_METRIC) should be registered" ); assert_eq!( - stats.metric_buckets.buckets, 1, - "exactly one metric bucket expected after one collapsed-spans emission" + stats.metric_buckets.buckets, 3, + "exactly 3 metric bucket expected after one whole-key collapsed-spans, one resource key collapsed-span and one resource key+http_endpoint key emissions" ); } }