feat: use ip quantization when aggregating peer tags for trace stats (#1944)

duncanpharvey · Eldolfin · web-flow · commit 4ae8ebe25245 · 2026-05-06T13:36:04.000Z
# What does this PR do? Quantize IP addresses for peer tags when aggregating trace stats. # Motivation Needed for agent side stats computation in the Serverless Compatibility Layer (used as the agent in Azure Functions) in DataDog/serverless-components#124. # Additional Notes Anything else we should know when reviewing? # How to test the change? Describe here in detail how the change can be validated. Co-authored-by: Eldolfin <oscar.ledauphin@datadoghq.com>
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/libdd-trace-stats/Cargo.toml b/libdd-trace-stats/Cargo.toml
@@ -16,6 +16,7 @@ libdd-common = { version = "4.0.0", path = "../libdd-common", default-features =
 libdd-ddsketch = { version = "1.0.1", path = "../libdd-ddsketch" }
 libdd-shared-runtime = { version = "0.1.0", path = "../libdd-shared-runtime" }
 libdd-trace-protobuf = { version = "3.0.1", path = "../libdd-trace-protobuf" }
+libdd-trace-obfuscation = { version = "2.0.0", path = "../libdd-trace-obfuscation", default-features = false }
 libdd-trace-utils = { version = "3.0.1", path = "../libdd-trace-utils", default-features = false }
 hashbrown = { version = "0.15" }
 http = "1.1"
diff --git a/libdd-trace-stats/src/span_concentrator/aggregation.rs b/libdd-trace-stats/src/span_concentrator/aggregation.rs
@@ -6,9 +6,10 @@
 //! span.
 
 use hashbrown::HashMap;
+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;
+use std::borrow::{Borrow, Cow};
 
 use crate::span_concentrator::StatSpan;
 
@@ -79,7 +80,7 @@ impl<T> FixedAggregationKey<T> {
 /// Represent a stats aggregation key borrowed from span data
 pub(super) struct BorrowedAggregationKey<'a> {
     fixed: FixedAggregationKey<&'a str>,
-    peer_tags: Vec<(&'a str, &'a str)>,
+    peer_tags: Vec<(&'a str, Cow<'a, str>)>,
 }
 
 impl hashbrown::Equivalent<OwnedAggregationKey> for BorrowedAggregationKey<'_> {
@@ -205,14 +206,17 @@ impl<'a> BorrowedAggregationKey<'a> {
         let span_kind = span.get_meta(TAG_SPANKIND).unwrap_or_default();
         let peer_tags = if should_track_peer_tags(span_kind) {
             // Parse the meta tags of the span and return a list of the peer tags based on the list
-            // of `peer_tag_keys`
+            // of `peer_tag_keys`. IP address values are quantized to reduce cardinality.
             peer_tag_keys
                 .iter()
-                .filter_map(|key| Some(((key.as_str()), (span.get_meta(key.as_str())?))))
+                .filter_map(|key| {
+                    let value = span.get_meta(key.as_str())?;
+                    Some((key.as_str(), quantize_peer_ip_addresses(value)))
+                })
                 .collect()
         } else if let Some(base_service) = span.get_meta("_dd.base_service") {
             // Internal spans with a base service override use only _dd.base_service as peer tag
-            vec![("_dd.base_service", base_service)]
+            vec![("_dd.base_service", Cow::Borrowed(base_service))]
         } else {
             vec![]
         };
@@ -961,6 +965,80 @@ mod tests {
         }
     }
 
+    #[test]
+    fn test_peer_tag_ip_quantization_in_aggregation_key() {
+        let peer_tag_keys = vec!["peer.hostname".to_string(), "db.instance".to_string()];
+
+        // IPv4 address peer tag gets replaced with blocked-ip-address
+        let span_ipv4 = SpanSlice {
+            service: "service",
+            name: "op",
+            resource: "res",
+            span_id: 1,
+            parent_id: 0,
+            meta: HashMap::from([
+                ("span.kind", "client"),
+                ("peer.hostname", "10.1.2.3"),
+                ("db.instance", "my-db"),
+            ]),
+            ..Default::default()
+        };
+        let key = BorrowedAggregationKey::from_span(&span_ipv4, &peer_tag_keys);
+        let owned = OwnedAggregationKey::from(&key);
+        assert_eq!(
+            owned.peer_tags,
+            vec![
+                (
+                    "peer.hostname".to_string(),
+                    "blocked-ip-address".to_string()
+                ),
+                ("db.instance".to_string(), "my-db".to_string()),
+            ]
+        );
+
+        // IPv6 address peer tag gets replaced with blocked-ip-address
+        let span_ipv6 = SpanSlice {
+            service: "service",
+            name: "op",
+            resource: "res",
+            span_id: 1,
+            parent_id: 0,
+            meta: HashMap::from([
+                ("span.kind", "client"),
+                ("peer.hostname", "2001:db8:3333:4444:CCCC:DDDD:EEEE:FFFF"),
+            ]),
+            ..Default::default()
+        };
+        let ipv6_keys = vec!["peer.hostname".to_string()];
+        let key = BorrowedAggregationKey::from_span(&span_ipv6, &ipv6_keys);
+        let owned = OwnedAggregationKey::from(&key);
+        assert_eq!(
+            owned.peer_tags,
+            vec![(
+                "peer.hostname".to_string(),
+                "blocked-ip-address".to_string()
+            )]
+        );
+
+        // Non-IP peer tags pass through unchanged
+        let span_non_ip = SpanSlice {
+            service: "service",
+            name: "op",
+            resource: "res",
+            span_id: 1,
+            parent_id: 0,
+            meta: HashMap::from([("span.kind", "client"), ("db.instance", "dynamo.test.us1")]),
+            ..Default::default()
+        };
+        let non_ip_keys = vec!["db.instance".to_string()];
+        let key = BorrowedAggregationKey::from_span(&span_non_ip, &non_ip_keys);
+        let owned = OwnedAggregationKey::from(&key);
+        assert_eq!(
+            owned.peer_tags,
+            vec![("db.instance".to_string(), "dynamo.test.us1".to_string())]
+        );
+    }
+
     #[test]
     fn test_grpc_status_str_to_int_value() {
         // Numeric strings parse directly
diff --git a/libdd-trace-stats/src/span_concentrator/tests.rs b/libdd-trace-stats/src/span_concentrator/tests.rs
@@ -877,6 +877,116 @@ fn test_peer_tags_aggregation() {
     );
 }
 
+/// Test that spans differing only by peer-tag IPs aggregate after IP quantization
+#[test]
+fn test_peer_tags_quantization_aggregation() {
+    let now = SystemTime::now();
+    let mut spans = vec![
+        get_test_span_with_meta(
+            now,
+            2,
+            1,
+            75,
+            5,
+            "A1",
+            "SELECT user_id from users WHERE user_name = ?",
+            0,
+            &[
+                ("span.kind", "client"),
+                ("db.instance", "i-1234"),
+                ("db.system", "postgres"),
+                ("region", "us1"),
+                ("peer.hostname", "10.1.2.3"),
+            ],
+            &[("_dd.measured", 1.0)],
+        ),
+        get_test_span_with_meta(
+            now,
+            3,
+            1,
+            75,
+            5,
+            "A1",
+            "SELECT user_id from users WHERE user_name = ?",
+            0,
+            &[
+                ("span.kind", "client"),
+                ("db.instance", "i-1234"),
+                ("db.system", "postgres"),
+                ("region", "us1"),
+                ("peer.hostname", "10.1.2.4"),
+            ],
+            &[("_dd.measured", 1.0)],
+        ),
+        get_test_span_with_meta(
+            now,
+            4,
+            1,
+            75,
+            5,
+            "A1",
+            "SELECT user_id from users WHERE user_name = ?",
+            0,
+            &[
+                ("span.kind", "client"),
+                ("db.instance", "i-1234"),
+                ("db.system", "postgres"),
+                ("region", "us1"),
+                ("peer.hostname", "2001:db8:3333:4444:CCCC:DDDD:EEEE:FFFF"),
+            ],
+            &[("_dd.measured", 1.0)],
+        ),
+    ];
+    compute_top_level_span(spans.as_mut_slice());
+    let mut concentrator_with_peer_tags = SpanConcentrator::new(
+        Duration::from_nanos(BUCKET_SIZE),
+        now,
+        get_span_kinds(),
+        vec![
+            "db.instance".to_string(),
+            "db.system".to_string(),
+            "peer.hostname".to_string(),
+        ],
+    );
+    for span in &spans {
+        concentrator_with_peer_tags.add_span(span);
+    }
+
+    let flushtime = now
+        + Duration::from_nanos(
+            concentrator_with_peer_tags.bucket_size * concentrator_with_peer_tags.buffer_len as u64,
+        );
+
+    let expected_with_peer_tags = vec![pb::ClientGroupedStats {
+        service: "A1".to_string(),
+        resource: "SELECT user_id from users WHERE user_name = ?".to_string(),
+        r#type: "db".to_string(),
+        name: "query".to_string(),
+        duration: 225,
+        hits: 3,
+        top_level_hits: 3,
+        errors: 0,
+        is_trace_root: pb::Trilean::False.into(),
+        span_kind: "client".to_string(),
+        peer_tags: vec![
+            "db.instance:i-1234".to_string(),
+            "db.system:postgres".to_string(),
+            "peer.hostname:blocked-ip-address".to_string(),
+        ],
+        ..Default::default()
+    }];
+
+    let stats_with_peer_tags = concentrator_with_peer_tags.flush(flushtime, false);
+    assert_counts_equal(
+        expected_with_peer_tags,
+        stats_with_peer_tags
+            .first()
+            .expect("There should be at least one time bucket")
+            .stats
+            .clone(),
+    );
+}
+
 /// Test that internal spans with _dd.base_service use it as their sole peer tag
 #[test]
 fn test_base_service_peer_tag() {
