diff --git a/libdd-trace-stats/src/span_concentrator/aggregation.rs b/libdd-trace-stats/src/span_concentrator/aggregation.rs index 73873779fb..fa6a914370 100644 --- a/libdd-trace-stats/src/span_concentrator/aggregation.rs +++ b/libdd-trace-stats/src/span_concentrator/aggregation.rs @@ -6,6 +6,7 @@ //! span. use hashbrown::{hash_set::Entry, HashMap, HashSet}; +use libdd_common::tag::const_assert; use libdd_trace_obfuscation::ip_address::quantize_peer_ip_addresses; use libdd_trace_protobuf::pb; use libdd_trace_utils::span::SpanText; @@ -442,14 +443,100 @@ pub(super) struct StatsBucket { reason = "FIXME(SVLS-8787|github.com/DataDog/libdatadog/pull/2170): implement stats additional tags" )] distinct_additional_tags: HashSet, - /// Number of spans collapsed into the overflow bucket due to cardinality limiting. + /// Number of spans collapsed into the overflow bucket due to whole-key cardinality limiting. collapsed_count: u64, + collapsed_fields_metrics: CollapsedFieldsMetrics, /// Indicates if stats obfuscated in this bucket. This is set once at creation and stays /// constant per bucket #[cfg(feature = "stats-obfuscation")] pub(super) obfuscated: bool, } +#[repr(transparent)] +pub struct CollapsedField; +impl CollapsedField { + pub const RESOURCE_NAME: usize = 1 << 1; + pub const HTTP_ENDPOINT: usize = 1 << 2; + pub const PEER_TAGS: usize = 1 << 3; + #[allow( + unused, + reason = "FIXME(SVLS-8787|github.com/DataDog/libdatadog/pull/2170): implement stats additional tags" + )] + pub const ADDITIONAL_TAGS: usize = 1 << 4; + pub const COUNT: u8 = 5; +} + +const COLLAPSED_FIELD_METRIC_SIZE: usize = 1 << CollapsedField::COUNT; +#[derive(Debug, Clone, Default, Copy)] +// note slot 0 is a counter for non_collapsed spans. useless +pub struct CollapsedFieldsMetrics([usize; COLLAPSED_FIELD_METRIC_SIZE]); + +const_assert!(COLLAPSED_FIELD_METRIC_SIZE <= 32); // Metrics table is of reasonable size + +impl CollapsedFieldsMetrics { + pub(crate) fn zero() -> Self { + Self::default() + } + + #[cfg(feature = "dogstatsd")] + pub fn emit_dogstatsd(&self, dogstatsd: &std::sync::Arc) { + // skip the first slot that is used to count span which have no collapsed fields. + for (mask, &count) in self.0.iter().enumerate().skip(1) { + if count > 0 { + let mut tags = Vec::new(); + for field_pow in 1..CollapsedField::COUNT { + let field_value = 1 << field_pow; + assert!([ + CollapsedField::RESOURCE_NAME, + CollapsedField::HTTP_ENDPOINT, + CollapsedField::PEER_TAGS, + CollapsedField::ADDITIONAL_TAGS + ] + .contains(&field_value)); + let has_field = (mask & field_value) != 0; + if !has_field { + continue; + } + let field_tag = match field_value { + CollapsedField::RESOURCE_NAME => { + libdd_common::tag!("collapsed_spans", "resource") + } + CollapsedField::HTTP_ENDPOINT => { + libdd_common::tag!("collapsed_spans", "http_endpoint") + } + CollapsedField::PEER_TAGS => { + libdd_common::tag!("collapsed_spans", "peer_tags") + } + CollapsedField::ADDITIONAL_TAGS => { + libdd_common::tag!("collapsed_spans", "additional_metric_tags") + } + #[allow( + clippy::unreachable, + reason = "field pow is between 1..CollapsedField::COUNT, so field_value is a valid CollapsedField value. (Asserted just above)" + )] + _ => unreachable!(), + }; + tags.push(field_tag); + } + assert!(!tags.is_empty()); + dogstatsd.send(vec![libdd_dogstatsd_client::DogStatsDAction::Count( + "datadog.tracer.stats.collapsed_spans", + count as i64, + tags.iter(), + )]); + } + } + } +} + +impl std::ops::AddAssign for CollapsedFieldsMetrics { + fn add_assign(&mut self, rhs: Self) { + for i in 0..self.0.len() { + self.0[i] += rhs.0[i]; + } + } +} + impl StatsBucket { /// Return a new StatsBucket starting at `start_timestamp`. /// @@ -470,9 +557,15 @@ impl StatsBucket { distinct_http_endpoints: HashSet::new(), distinct_peer_tags: HashSet::new(), distinct_additional_tags: HashSet::new(), + collapsed_fields_metrics: CollapsedFieldsMetrics::zero(), } } + /// Returns metrics on spans field collapse with reasons. + pub fn collapsed_fields_metrics(&self) -> CollapsedFieldsMetrics { + self.collapsed_fields_metrics + } + /// Return the number of spans collapsed into the overflow bucket. pub(super) fn collapsed_count(&self) -> u64 { self.collapsed_count @@ -517,11 +610,14 @@ impl StatsBucket { hasher.finish() } + let mut collapsed_fields = 0; + let resource_hash = hash(&key.fixed.resource_name); let resources_count = self.distinct_resources.len(); if let Entry::Vacant(slot) = self.distinct_resources.entry(resource_hash) { if resources_count >= self.cardinality_limits.resource_limit { key.fixed.resource_name = TRACER_BLOCKED_VALUE; + collapsed_fields |= CollapsedField::RESOURCE_NAME; } else { slot.insert(); } @@ -532,6 +628,7 @@ impl StatsBucket { if let Entry::Vacant(slot) = self.distinct_http_endpoints.entry(http_endpoint_hash) { if http_endpoints_count >= self.cardinality_limits.http_endpoint_limit { key.fixed.http_endpoint = TRACER_BLOCKED_VALUE; + collapsed_fields |= CollapsedField::HTTP_ENDPOINT; } else { slot.insert(); } @@ -542,10 +639,13 @@ impl StatsBucket { if let Entry::Vacant(slot) = self.distinct_peer_tags.entry(peer_tags_hash) { if peer_tags_count >= self.cardinality_limits.peer_tags_limit { key.peer_tags = vec![(TRACER_BLOCKED_VALUE, Cow::Borrowed(""))]; + collapsed_fields |= CollapsedField::PEER_TAGS; } else { slot.insert(); } } + + self.collapsed_fields_metrics.0[collapsed_fields] += 1; } /// Consume the bucket and return a ClientStatsBucket containing the bucket stats. diff --git a/libdd-trace-stats/src/span_concentrator/mod.rs b/libdd-trace-stats/src/span_concentrator/mod.rs index 6f62b5e53c..162391ac42 100644 --- a/libdd-trace-stats/src/span_concentrator/mod.rs +++ b/libdd-trace-stats/src/span_concentrator/mod.rs @@ -7,7 +7,7 @@ use tracing::{debug, warn}; use libdd_trace_protobuf::pb; -use aggregation::StatsBucket; +use aggregation::{CollapsedFieldsMetrics, StatsBucket}; mod aggregation; use aggregation::BorrowedAggregationKey; @@ -28,6 +28,7 @@ pub struct FlushResult { /// Total number of spans that were collapsed into the overflow sentinel bucket due to /// cardinality limiting across all flushed time buckets. pub collapsed_spans: u64, + pub collapsed_fields_metrics: CollapsedFieldsMetrics, } impl FlushResult { @@ -318,7 +319,8 @@ impl SpanConcentrator { /// /// Obfuscated and un-obfuscated buckets are returned separately, see [`FlushResult`]. pub fn flush(&mut self, now: SystemTime, force: bool) -> FlushResult { - let (buckets, collapsed_spans) = self.drain_due_buckets(now, force, StatsBucket::flush); + let (buckets, collapsed_spans, collapsed_fields_metrics) = + self.drain_due_buckets(now, force, StatsBucket::flush); let mut obfuscated_buckets = Vec::new(); let mut unobfuscated_buckets = Vec::new(); for (obfuscated, bucket) in buckets { @@ -332,6 +334,7 @@ impl SpanConcentrator { obfuscated_buckets, unobfuscated_buckets, collapsed_spans, + collapsed_fields_metrics, } } @@ -339,7 +342,9 @@ impl SpanConcentrator { /// OTLP trace-metrics path. The protobuf bucket inside each [`OtlpStatsBucket`] is identical /// to what [`Self::flush`] would produce, so the /v0.6/stats agent path is unaffected. pub fn flush_with_otlp_exact(&mut self, now: SystemTime, force: bool) -> Vec { - let (buckets, _) = self.drain_due_buckets(now, force, StatsBucket::flush_with_otlp_exact); + let buckets = self + .drain_due_buckets(now, force, StatsBucket::flush_with_otlp_exact) + .0; buckets.into_iter().map(|(_, bucket)| bucket).collect() } @@ -354,7 +359,7 @@ impl SpanConcentrator { now: SystemTime, force: bool, encode: impl Fn(StatsBucket, u64) -> T, - ) -> (Vec<(bool, T)>, u64) { + ) -> (Vec<(bool, T)>, u64, CollapsedFieldsMetrics) { // TODO: Wait for HashMap::extract_if to be stabilized to avoid a full drain let now_timestamp = system_time_to_unix_duration(now).as_nanos() as u64; let buckets: Vec<(u64, StatsBucket)> = self.buckets.drain().collect(); @@ -365,6 +370,7 @@ impl SpanConcentrator { - (self.buffer_len as u64 - 1) * self.bucket_size }; let mut total_collapsed = 0; + let mut total_collapsed_fields = CollapsedFieldsMetrics::zero(); let buckets_pb = buckets .into_iter() .filter_map(|(timestamp, bucket)| { @@ -382,6 +388,7 @@ impl SpanConcentrator { return None; } total_collapsed += bucket.collapsed_count(); + total_collapsed_fields += bucket.collapsed_fields_metrics(); #[cfg(feature = "stats-obfuscation")] let obfuscated = bucket.obfuscated; #[cfg(not(feature = "stats-obfuscation"))] @@ -396,7 +403,7 @@ impl SpanConcentrator { "Client-side stats values have been collapsed to 'tracer_blocked_value'. This is due to the cardinality exceeding DD_TRACE_STATS_CARDINALITY_LIMIT" ); } - (buckets_pb, total_collapsed) + (buckets_pb, total_collapsed, total_collapsed_fields) } } diff --git a/libdd-trace-stats/src/stats_exporter.rs b/libdd-trace-stats/src/stats_exporter.rs index 302edf75cc..eb4754f538 100644 --- a/libdd-trace-stats/src/stats_exporter.rs +++ b/libdd-trace-stats/src/stats_exporter.rs @@ -196,6 +196,11 @@ impl } } + #[cfg(feature = "dogstatsd")] + if let Some(client) = &self.dogstatsd { + flush.collapsed_fields_metrics.emit_dogstatsd(client); + } + let futures = FuturesUnordered::new(); if !flush.obfuscated_buckets.is_empty() { @@ -323,7 +328,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::prelude::*; @@ -636,7 +640,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( @@ -645,7 +651,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")] @@ -654,26 +661,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() }, @@ -732,7 +739,7 @@ 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() ); } @@ -777,12 +784,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" ); } @@ -843,8 +861,8 @@ mod tests { "exactly one metric context (COLLAPSED_SPANS_METRIC) should be registered" ); assert_eq!( - stats.metric_buckets.buckets, 1, - "exactly one metric bucket expected after one collapsed-spans emission" + stats.metric_buckets.buckets, 2, + "exactly one metric bucket expected after collapsing both on whole-key and on the resource field" ); } }