feat(stats)!: add telemetry for per-field cardinality limits V2

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<u64>,
-    /// 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<libdd_dogstatsd_client::Client>) {
+        // 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.

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,14 +334,17 @@ impl SpanConcentrator {
             obfuscated_buckets,
             unobfuscated_buckets,
             collapsed_spans,
+            collapsed_fields_metrics,
         }
     }
 
     /// 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<OtlpStatsBucket> {
-        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)
     }
 }
 

libdd-trace-stats/src/stats_exporter.rs

@@ -196,6 +196,11 @@ impl<Cap: HttpClientCapability + SleepCapability, Con: FlushableConcentrator>
             }
         }
 
+        #[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<http::Uri> {
 #[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"
         );
     }
 }