Skip to main content

libdd_trace_stats/span_concentrator/
aggregation.rs

1// Copyright 2024-Present Datadog, Inc. https://www.datadoghq.com/
2// SPDX-License-Identifier: Apache-2.0
3
4//! This module implement the logic for stats aggregation into time buckets and stats group.
5//! This includes the aggregation key to group spans together and the computation of stats from a
6//! span.
7
8use hashbrown::{HashMap, HashSet};
9use libdd_trace_obfuscation::ip_address::quantize_peer_ip_addresses;
10use libdd_trace_protobuf::pb;
11use libdd_trace_utils::span::SpanText;
12use std::{
13    borrow::{Borrow, Cow},
14    hash::{DefaultHasher, Hash, Hasher as _},
15};
16use tracing::warn;
17
18use crate::span_concentrator::{cardinality_limit_telemetry::CollapsedFieldSet, StatSpan};
19
20use super::{
21    cardinality_limit_telemetry::{self, CollapsedFieldsMetrics},
22    CardinalityLimitConfig,
23};
24
25/// Sentinel value used for cardinality limiting.
26pub const TRACER_BLOCKED_VALUE: &str = "tracer_blocked_value";
27
28const TAG_STATUS_CODE: &str = "http.status_code";
29const ADDITIONAL_METRIC_TAG_VALUE_MAX_LEN: usize = 200;
30const TAG_SYNTHETICS: &str = "synthetics";
31const TAG_SPANKIND: &str = "span.kind";
32const TAG_ORIGIN: &str = "_dd.origin";
33const TAG_SVC_SRC: &str = "_dd.svc_src";
34const GRPC_STATUS_CODE_FIELD: &[&str] = &[
35    "rpc.grpc.status_code",
36    "grpc.code",
37    "rpc.grpc.status.code",
38    "grpc.status.code",
39];
40
41/// Aggregation key fields shared across all concentrator implementations — everything
42/// **except** peer tags.
43///
44/// `T` is the string representation:
45/// * `&'a str`   — borrowed references used in [`BorrowedAggregationKey`]
46/// * `String`    — owned values used in `OwnedAggregationKey`
47/// * `StringRef` — offset+len into a SHM string pool, used in `ShmKeyHeader`
48#[derive(
49    Clone, Default, Hash, Eq, PartialEq, Debug, PartialOrd, serde::Serialize, serde::Deserialize,
50)]
51pub struct FixedAggregationKey<T> {
52    pub resource_name: T,
53    pub service_name: T,
54    pub operation_name: T,
55    pub span_type: T,
56    pub span_kind: T,
57    pub http_method: T,
58    pub http_endpoint: T,
59    pub service_source: T,
60    pub http_status_code: u32,
61    pub grpc_status_code: Option<u8>,
62    pub is_synthetics_request: bool,
63    pub is_trace_root: pb::Trilean,
64}
65
66impl<T> FixedAggregationKey<T> {
67    /// Map all string fields through `f`, preserving scalar fields.
68    pub fn convert<'a, V: 'a, I: ?Sized + 'a, F: Fn(&'a I) -> V>(
69        &'a self,
70        f: F,
71    ) -> FixedAggregationKey<V>
72    where
73        T: Borrow<I>,
74    {
75        FixedAggregationKey {
76            resource_name: f(self.resource_name.borrow()),
77            service_name: f(self.service_name.borrow()),
78            operation_name: f(self.operation_name.borrow()),
79            span_type: f(self.span_type.borrow()),
80            span_kind: f(self.span_kind.borrow()),
81            http_method: f(self.http_method.borrow()),
82            http_endpoint: f(self.http_endpoint.borrow()),
83            service_source: f(self.service_source.borrow()),
84            http_status_code: self.http_status_code,
85            grpc_status_code: self.grpc_status_code,
86            is_synthetics_request: self.is_synthetics_request,
87            is_trace_root: self.is_trace_root,
88        }
89    }
90}
91
92#[derive(Clone, Hash, PartialEq, Eq)]
93/// Represent a stats aggregation key borrowed from span data
94pub(super) struct BorrowedAggregationKey<'a> {
95    fixed: FixedAggregationKey<&'a str>,
96    peer_tags: Vec<(&'a str, Cow<'a, str>)>,
97    additional_metric_tags: Vec<(&'a str, &'a str)>,
98}
99
100impl hashbrown::Equivalent<OwnedAggregationKey> for BorrowedAggregationKey<'_> {
101    #[inline]
102    fn equivalent(&self, other: &OwnedAggregationKey) -> bool {
103        self.fixed == other.fixed.convert(|s| s)
104            && self.peer_tags.len() == other.peer_tags.len()
105            && self
106                .peer_tags
107                .iter()
108                .zip(other.peer_tags.iter())
109                .all(|((k1, v1), (k2, v2))| k1 == k2 && v1 == v2)
110            && self.additional_metric_tags.len() == other.additional_metric_tags.len()
111            && self
112                .additional_metric_tags
113                .iter()
114                .zip(other.additional_metric_tags.iter())
115                .all(|((k1, v1), (k2, v2))| k1 == k2 && v1 == v2)
116    }
117}
118
119#[derive(Debug, Clone, Hash, PartialEq, Eq, PartialOrd, Default)]
120/// Represents a span aggregation key with owned data
121///
122/// To be able to use BorrowedAggregationKey to index into a stats bucket hashmap two
123/// conditions must stay true:
124/// * Hashing an owned key derived from a borrowed key should produce the same hash as hashing the
125///   borrowed key
126/// * Running the Equivalent trait on an owned key derived from a borrowed key should produce true
127pub(super) struct OwnedAggregationKey {
128    fixed: FixedAggregationKey<String>,
129    peer_tags: Vec<(String, String)>,
130    additional_metric_tags: Vec<(String, String)>,
131}
132
133impl From<&BorrowedAggregationKey<'_>> for OwnedAggregationKey {
134    fn from(value: &BorrowedAggregationKey<'_>) -> Self {
135        OwnedAggregationKey {
136            fixed: value.fixed.convert(str::to_owned),
137            peer_tags: value
138                .peer_tags
139                .iter()
140                .map(|(k, v)| (k.to_string(), v.to_string()))
141                .collect(),
142            additional_metric_tags: value
143                .additional_metric_tags
144                .iter()
145                .map(|(k, v)| (k.to_string(), v.to_string()))
146                .collect(),
147        }
148    }
149}
150
151fn float_to_int(f: f64) -> Option<u8> {
152    if f.floor() != f {
153        return None;
154    }
155    if f < 0.0 || (u8::MAX as f64) < f {
156        return None;
157    }
158    Some(f as u8)
159}
160
161fn get_grpc_status_code<'a>(span: &'a impl StatSpan<'a>) -> Option<u8> {
162    for key in GRPC_STATUS_CODE_FIELD {
163        if let Some(val) = span.get_meta(key) {
164            if let Some(code) = grpc_status_str_to_int_value(val) {
165                return Some(code);
166            }
167        }
168    }
169
170    for key in GRPC_STATUS_CODE_FIELD {
171        if let Some(val) = span.get_metrics(key) {
172            if let Some(code) = float_to_int(val) {
173                return Some(code);
174            }
175        }
176    }
177
178    None
179}
180
181fn grpc_status_str_to_int_value(v: &str) -> Option<u8> {
182    if let Ok(status) = v.parse() {
183        return Some(status);
184    }
185    let mut status_uppercase = [0u8; 32];
186    let mut status = v.trim_start_matches("StatusCode.");
187
188    let mut needs_upcasing = false;
189    for b in status.as_bytes() {
190        if !b.is_ascii() {
191            return None;
192        }
193        needs_upcasing |= b.is_ascii_lowercase()
194    }
195    if needs_upcasing {
196        for (c, d) in status.as_bytes().iter().zip(&mut status_uppercase) {
197            *d = c.to_ascii_uppercase();
198        }
199        status = std::str::from_utf8(&status_uppercase[0..status.len().min(status_uppercase.len())])
200            .ok()?
201    }
202
203    match status {
204        "OK" => return Some(0),
205        "CANCELLED" | "CANCELED" => return Some(1),
206        "UNKNOWN" => return Some(2),
207        "INVALID_ARGUMENT" | "INVALIDARGUMENT" => return Some(3),
208        "DEADLINE_EXCEEDED" | "DEADLINEEXCEEDED" => return Some(4),
209        "NOT_FOUND" | "NOTFOUND" => return Some(5),
210        "ALREADY_EXISTS" | "ALREADYEXISTS" => return Some(6),
211        "PERMISSION_DENIED" | "PERMISSIONDENIED" => return Some(7),
212        "UNAUTHENTICATED" => return Some(16),
213        "RESOURCE_EXHAUSTED" | "RESOURCEEXHAUSTED" => return Some(8),
214        "FAILED_PRECONDITION" | "FAILEDPRECONDITION" => return Some(9),
215        "ABORTED" => return Some(10),
216        "OUT_OF_RANGE" | "OUTOFRANGE" => return Some(11),
217        "UNIMPLEMENTED" => return Some(12),
218        "INTERNAL" => return Some(13),
219        "UNAVAILABLE" => return Some(14),
220        "DATA_LOSS" | "DATALOSS" => return Some(15),
221        _ => {}
222    }
223    None
224}
225
226impl<'a> BorrowedAggregationKey<'a> {
227    /// Return an AggregationKey matching the given span.
228    ///
229    /// If `peer_tag_keys` is not empty then the peer tags of the span will be included in the
230    /// key.
231    /// If `additional_metric_tags` is not empty then matching span tags keys are included in the
232    /// key.
233    pub(super) fn from_span<T: StatSpan<'a>>(
234        span: &'a T,
235        peer_tag_keys: &'a [String],
236        additional_metric_tag_keys: &'a [String],
237    ) -> Self {
238        Self::from_obfuscated_span(
239            span.resource(),
240            span,
241            peer_tag_keys,
242            additional_metric_tag_keys,
243        )
244    }
245
246    pub(crate) fn from_obfuscated_span<'b, T>(
247        resource_name: &'a str,
248        span: &'b T,
249        peer_tag_keys: &'b [String],
250        additional_metric_tag_keys: &'b [String],
251    ) -> BorrowedAggregationKey<'a>
252    where
253        T: StatSpan<'b>,
254        // resource_name is a temporary string on the stack the span will outlive it
255        'b: 'a,
256    {
257        let span_kind = span.get_meta(TAG_SPANKIND).unwrap_or_default();
258        let peer_tags = if should_track_peer_tags(span_kind) {
259            // Parse the meta tags of the span and return a list of the peer tags based on the list
260            // of `peer_tag_keys`. IP address values are quantized to reduce cardinality.
261            peer_tag_keys
262                .iter()
263                .filter_map(|key| {
264                    let value = span.get_meta(key.as_str())?;
265                    Some((key.as_str(), quantize_peer_ip_addresses(value)))
266                })
267                .collect()
268        } else if let Some(base_service) = span.get_meta("_dd.base_service") {
269            // Internal spans with a base service override use only _dd.base_service as peer tag
270            vec![("_dd.base_service", Cow::Borrowed(base_service))]
271        } else {
272            vec![]
273        };
274
275        let http_method = span.get_meta("http.method").unwrap_or_default();
276
277        let http_endpoint = span
278            .get_meta("http.endpoint")
279            .or_else(|| span.get_meta("http.route"))
280            .unwrap_or_default();
281
282        let status_code = if let Some(status_code) = span.get_metrics(TAG_STATUS_CODE) {
283            status_code as u32
284        } else if let Some(status_code) = span.get_meta(TAG_STATUS_CODE) {
285            status_code.parse().unwrap_or_default()
286        } else {
287            0
288        };
289
290        let grpc_status_code = get_grpc_status_code(span);
291        let service_source = span.get_meta(TAG_SVC_SRC).unwrap_or_default();
292
293        let additional_metric_tags: Vec<(&'a str, &'a str)> = additional_metric_tag_keys
294            .iter()
295            .filter_map(|key| match span.get_meta(key.as_str()) {
296                Some(v) if !v.is_empty() => {
297                    // Byte length >= char count, so skip the char walk when byte length alone
298                    // is within the max character length, otherwise stop as soon as we pass the max character length.
299                    if v.len() > ADDITIONAL_METRIC_TAG_VALUE_MAX_LEN
300                        && v.chars().nth(ADDITIONAL_METRIC_TAG_VALUE_MAX_LEN).is_some()
301                    {
302                        warn!(
303                            "additional_metric_tags: value for key '{}' exceeds {} characters; substituting tracer_blocked_value",
304                            key, ADDITIONAL_METRIC_TAG_VALUE_MAX_LEN,
305                        );
306                        Some((key.as_str(), TRACER_BLOCKED_VALUE))
307                    } else {
308                        Some((key.as_str(), v))
309                    }
310                }
311                _ => None,
312            })
313            .collect();
314
315        Self {
316            fixed: FixedAggregationKey {
317                resource_name,
318                service_name: span.service(),
319                operation_name: span.name(),
320                span_type: span.r#type(),
321                span_kind,
322                http_method,
323                http_endpoint,
324                service_source,
325                http_status_code: status_code,
326                grpc_status_code,
327                is_synthetics_request: span
328                    .get_meta(TAG_ORIGIN)
329                    .is_some_and(|origin| origin.starts_with(TAG_SYNTHETICS)),
330                is_trace_root: if span.is_trace_root() {
331                    pb::Trilean::True
332                } else {
333                    pb::Trilean::False
334                },
335            },
336            peer_tags,
337            additional_metric_tags,
338        }
339    }
340
341    /// Truncates string fields in accordance with the cardinality limit RFC
342    ///
343    /// This should be called only after obfuscation
344    #[cfg_attr(not(feature = "stats-obfuscation"), allow(unused))]
345    pub(crate) fn truncate(&mut self, big_resource: bool) {
346        let resource_length_limit = if big_resource { 15_000 } else { 5000 };
347        self.fixed.resource_name = slice_up_to(self.fixed.resource_name, resource_length_limit);
348        self.fixed.service_name = slice_up_to(self.fixed.service_name, 100);
349        self.fixed.operation_name = slice_up_to(self.fixed.operation_name, 100);
350        self.fixed.span_type = slice_up_to(self.fixed.span_type, 100);
351    }
352}
353
354/// Truncate `s` to at most `max_len` bytes
355fn slice_up_to(s: &str, max_len: usize) -> &str {
356    if max_len >= s.len() {
357        return s;
358    }
359    // TODO: use `floor_char_boundary` once our MSRV is 1.91 or higher
360    // (https://doc.rust-lang.org/std/primitive.str.html#method.floor_char_boundary)
361    let mut idx = max_len;
362    while !s.is_char_boundary(idx) {
363        idx -= 1;
364    }
365    &s[..idx]
366}
367
368impl OwnedAggregationKey {
369    /// Return the overflow sentinel key.
370    pub(super) fn overflow_key() -> Self {
371        OwnedAggregationKey {
372            fixed: FixedAggregationKey {
373                resource_name: TRACER_BLOCKED_VALUE.to_owned(),
374                service_name: TRACER_BLOCKED_VALUE.to_owned(),
375                operation_name: TRACER_BLOCKED_VALUE.to_owned(),
376                span_type: TRACER_BLOCKED_VALUE.to_owned(),
377                span_kind: TRACER_BLOCKED_VALUE.to_owned(),
378                http_method: TRACER_BLOCKED_VALUE.to_owned(),
379                http_endpoint: TRACER_BLOCKED_VALUE.to_owned(),
380                service_source: TRACER_BLOCKED_VALUE.to_owned(),
381                http_status_code: 0,
382                grpc_status_code: None,
383                is_synthetics_request: false,
384                is_trace_root: pb::Trilean::NotSet,
385            },
386            peer_tags: vec![(TRACER_BLOCKED_VALUE.to_owned(), "".to_owned())],
387            additional_metric_tags: vec![(TRACER_BLOCKED_VALUE.to_owned(), "".to_owned())],
388        }
389    }
390}
391
392impl From<pb::ClientGroupedStats> for OwnedAggregationKey {
393    fn from(value: pb::ClientGroupedStats) -> Self {
394        Self {
395            fixed: FixedAggregationKey {
396                resource_name: value.resource,
397                service_name: value.service,
398                operation_name: value.name,
399                span_type: value.r#type,
400                span_kind: value.span_kind,
401                http_method: value.http_method,
402                http_endpoint: value.http_endpoint,
403                service_source: value.service_source,
404                http_status_code: value.http_status_code,
405                grpc_status_code: value.grpc_status_code.parse().ok(),
406                is_synthetics_request: value.synthetics,
407                is_trace_root: pb::Trilean::try_from(value.is_trace_root)
408                    .unwrap_or(pb::Trilean::NotSet),
409            },
410            peer_tags: value
411                .peer_tags
412                .into_iter()
413                .filter_map(|t| {
414                    let (key, value) = t.split_once(':')?;
415                    Some((key.to_string(), value.to_string()))
416                })
417                .collect(),
418            additional_metric_tags: value
419                .additional_metric_tags
420                .into_iter()
421                .filter_map(|t| {
422                    let (key, value) = t.split_once(':')?;
423                    Some((key.to_string(), value.to_string()))
424                })
425                .collect(),
426        }
427    }
428}
429
430/// Return true if we care about peer tags on the span
431fn should_track_peer_tags<T>(span_kind: T) -> bool
432where
433    T: SpanText,
434{
435    matches!(
436        span_kind.borrow().to_lowercase().as_str(),
437        "client" | "producer" | "consumer"
438    )
439}
440
441/// The stats computed from a group of span with the same AggregationKey
442#[derive(Debug, Default, Clone)]
443pub(super) struct GroupedStats {
444    hits: u64,
445    errors: u64,
446    duration: u64,
447    top_level_hits: u64,
448    ok_summary: libdd_ddsketch::DDSketch,
449    error_summary: libdd_ddsketch::DDSketch,
450    // Exact per-cell (ok/error) scalars used by the OTLP trace-metrics path. These are tracked
451    // separately from `duration` so the /v0.6/stats agent payload is byte-for-byte unchanged.
452    ok_duration: u64,
453    ok_min: u64,
454    ok_max: u64,
455    error_duration: u64,
456    error_min: u64,
457    error_max: u64,
458}
459
460impl GroupedStats {
461    /// Update the stats of a GroupedStats by inserting a span.
462    fn insert(&mut self, duration: i64, is_error: bool, is_top_level: bool) {
463        self.hits += 1;
464        self.duration += duration as u64;
465        let d = duration as u64;
466        if is_error {
467            self.errors += 1;
468            let _ = self.error_summary.add(duration as f64);
469            self.error_duration += d;
470            self.error_min = if self.errors == 1 {
471                d
472            } else {
473                self.error_min.min(d)
474            };
475            self.error_max = self.error_max.max(d);
476        } else {
477            let _ = self.ok_summary.add(duration as f64);
478            self.ok_duration += d;
479            let ok_count = self.hits - self.errors;
480            self.ok_min = if ok_count == 1 { d } else { self.ok_min.min(d) };
481            self.ok_max = self.ok_max.max(d);
482        }
483        if is_top_level {
484            self.top_level_hits += 1;
485        }
486    }
487}
488
489/// Exact per-cell (ok or error) scalars for one aggregation group, surfaced to the OTLP
490/// trace-metrics path. `count` is exact; `duration_ns`/`min_ns`/`max_ns` are exact when
491/// `count > 0` and meaningless otherwise (the OTLP mapper suppresses empty cells).
492#[derive(Debug, Clone, Copy, Default)]
493pub struct OtlpExactCell {
494    pub count: u64,
495    pub duration_ns: u64,
496    pub min_ns: u64,
497    pub max_ns: u64,
498}
499
500/// Exact OK/ERROR cells for one aggregation group, in the same order as the `stats` vector
501/// of the accompanying [`pb::ClientStatsBucket`].
502#[derive(Debug, Clone, Default)]
503pub struct OtlpExactGroup {
504    pub ok: OtlpExactCell,
505    pub error: OtlpExactCell,
506}
507
508/// A bucket flushed for the OTLP trace-metrics path. `exact[i]` is the exact-scalar sidecar
509/// for `bucket.stats[i]`; the protobuf bucket itself is identical to what the agent path uses.
510#[derive(Debug, Clone)]
511pub struct OtlpStatsBucket {
512    pub bucket: pb::ClientStatsBucket,
513    pub exact: Vec<OtlpExactGroup>,
514}
515
516/// A time bucket used for stats aggregation. It stores a map of GroupedStats storing the stats of
517/// spans aggregated on their AggregationKey.
518#[derive(Debug, Clone)]
519pub(super) struct StatsBucket {
520    data: HashMap<OwnedAggregationKey, GroupedStats>,
521    start: u64,
522    /// Maximum number of distinct aggregation keys this bucket will hold before collapsing new
523    /// ones into the overflow sentinel key.
524    cardinality_limits: CardinalityLimitConfig,
525    // HashSet of hashes of field values so we save memory
526    // This is not 100% accurate but the probability of getting collision is close to 0
527    // In the very rare case we get a collision, we would get one extra bucket which is totally
528    // fine
529    distinct_resources: HashSet<u64>,
530    distinct_http_endpoints: HashSet<u64>,
531    distinct_peer_tags: HashSet<u64>,
532    distinct_additional_tags: HashSet<u64>,
533    /// Number of spans collapsed into the overflow bucket due to whole-key cardinality limiting.
534    collapsed_count: u64,
535    collapsed_fields_metrics: CollapsedFieldsMetrics,
536    /// Indicates if stats obfuscated in this bucket. This is set once at creation and stays
537    /// constant per bucket
538    #[cfg(feature = "stats-obfuscation")]
539    pub(super) obfuscated: bool,
540}
541
542impl StatsBucket {
543    /// Return a new StatsBucket starting at `start_timestamp`.
544    ///
545    /// `cardinality_limits` are the values for whole-key and per-field cardinality limits
546    pub(super) fn new(
547        start_timestamp: u64,
548        cardinality_limits: CardinalityLimitConfig,
549        #[cfg(feature = "stats-obfuscation")] obfuscation_enabled: bool,
550    ) -> Self {
551        Self {
552            data: HashMap::new(),
553            start: start_timestamp,
554            cardinality_limits,
555            collapsed_count: 0,
556            #[cfg(feature = "stats-obfuscation")]
557            obfuscated: obfuscation_enabled,
558            distinct_resources: HashSet::new(),
559            distinct_http_endpoints: HashSet::new(),
560            distinct_peer_tags: HashSet::new(),
561            distinct_additional_tags: HashSet::new(),
562            collapsed_fields_metrics: cardinality_limit_telemetry::CollapsedFieldsMetrics::zero(),
563        }
564    }
565
566    /// Returns metrics on spans field collapse with reasons.
567    pub fn collapsed_fields_metrics(&self) -> cardinality_limit_telemetry::CollapsedFieldsMetrics {
568        self.collapsed_fields_metrics
569    }
570
571    /// Return the number of spans collapsed into the overflow bucket.
572    pub(super) fn collapsed_count(&self) -> u64 {
573        self.collapsed_count
574    }
575
576    /// Insert a value as stats in the group corresponding to the aggregation key, if it does not
577    /// exist it creates it.
578    ///
579    /// Keys that already exist in this bucket always merge normally. A new key is subject to the
580    /// `max_entries` limit, which collapses it into the overflow sentinel key.
581    pub(super) fn insert(
582        &mut self,
583        mut key: BorrowedAggregationKey<'_>,
584        duration: i64,
585        is_error: bool,
586        is_top_level: bool,
587    ) {
588        // Per field cardinality limiting
589        self.collapse_key_fields_cardinality(&mut key);
590
591        // The map can't change size before the entry below is resolved, so this single read
592        // covers the `max_entries` check in the vacant branch without a further lookup.
593        let len_before_insert = self.data.len();
594
595        match self.data.entry_ref(&key) {
596            // Existing key, merge
597            hashbrown::hash_map::EntryRef::Occupied(mut e) => {
598                e.get_mut().insert(duration, is_error, is_top_level);
599            }
600            hashbrown::hash_map::EntryRef::Vacant(e) => {
601                // New key over the max entry limit, collapse into the overflow
602                // sentinel.
603                if len_before_insert >= self.cardinality_limits.whole_key_limit {
604                    self.collapsed_count += 1;
605                    self.data
606                        .entry(OwnedAggregationKey::overflow_key())
607                        .or_default()
608                        .insert(duration, is_error, is_top_level);
609                    return;
610                }
611                // Within the max entry limit, admit key as a new distinct entry.
612                e.insert(GroupedStats::default())
613                    .insert(duration, is_error, is_top_level);
614            }
615        }
616    }
617
618    /// Collapse an aggregation key fields following the bucket's `CardinalityLimitConfig`
619    fn collapse_key_fields_cardinality(&mut self, key: &mut BorrowedAggregationKey<'_>) {
620        use hashbrown::hash_set::Entry;
621        fn hash(input: &impl Hash) -> u64 {
622            let mut hasher = DefaultHasher::new();
623            input.hash(&mut hasher);
624            hasher.finish()
625        }
626
627        let mut collapsed_fields = CollapsedFieldSet::empty();
628
629        let resource_hash = hash(&key.fixed.resource_name);
630        let resources_count = self.distinct_resources.len();
631        if let Entry::Vacant(slot) = self.distinct_resources.entry(resource_hash) {
632            if resources_count >= self.cardinality_limits.resource_limit {
633                key.fixed.resource_name = TRACER_BLOCKED_VALUE;
634                collapsed_fields.add(CollapsedFieldSet::RESOURCE_NAME);
635            } else {
636                slot.insert();
637            }
638        }
639
640        let http_endpoint_hash = hash(&key.fixed.http_endpoint);
641        let http_endpoints_count = self.distinct_http_endpoints.len();
642        if let Entry::Vacant(slot) = self.distinct_http_endpoints.entry(http_endpoint_hash) {
643            if http_endpoints_count >= self.cardinality_limits.http_endpoint_limit {
644                key.fixed.http_endpoint = TRACER_BLOCKED_VALUE;
645                collapsed_fields.add(CollapsedFieldSet::HTTP_ENDPOINT);
646            } else {
647                slot.insert();
648            }
649        }
650
651        let peer_tags_hash = hash(&key.peer_tags);
652        let peer_tags_count = self.distinct_peer_tags.len();
653        if let Entry::Vacant(slot) = self.distinct_peer_tags.entry(peer_tags_hash) {
654            if peer_tags_count >= self.cardinality_limits.peer_tags_limit {
655                key.peer_tags = vec![(TRACER_BLOCKED_VALUE, Cow::Borrowed(""))];
656                collapsed_fields.add(CollapsedFieldSet::PEER_TAGS);
657            } else {
658                slot.insert();
659            }
660        }
661
662        let additional_tags_hash = hash(&key.additional_metric_tags);
663        let additional_tags_count = self.distinct_additional_tags.len();
664        if let Entry::Vacant(slot) = self.distinct_additional_tags.entry(additional_tags_hash) {
665            if additional_tags_count >= self.cardinality_limits.additional_tags_limit {
666                key.additional_metric_tags = vec![(TRACER_BLOCKED_VALUE, "")];
667                collapsed_fields.add(CollapsedFieldSet::ADDITIONAL_TAGS);
668            } else {
669                slot.insert();
670            }
671        }
672        self.collapsed_fields_metrics.increment(collapsed_fields);
673    }
674
675    /// Consume the bucket and return a ClientStatsBucket containing the bucket stats.
676    /// `bucket_duration` is the size of buckets for the concentrator containing the bucket.
677    pub(super) fn flush(self, bucket_duration: u64) -> pb::ClientStatsBucket {
678        self.flush_with_otlp_exact(bucket_duration).bucket
679    }
680
681    /// Like [`Self::flush`], but additionally produces exact per-cell scalars for the OTLP
682    /// trace-metrics path. The `bucket` field is identical to what [`Self::flush`] returns.
683    pub(super) fn flush_with_otlp_exact(self, bucket_duration: u64) -> OtlpStatsBucket {
684        let mut stats = Vec::with_capacity(self.data.len());
685        let mut exact = Vec::with_capacity(self.data.len());
686        for (k, g) in self.data {
687            exact.push(OtlpExactGroup {
688                ok: OtlpExactCell {
689                    count: g.hits.saturating_sub(g.errors),
690                    duration_ns: g.ok_duration,
691                    min_ns: g.ok_min,
692                    max_ns: g.ok_max,
693                },
694                error: OtlpExactCell {
695                    count: g.errors,
696                    duration_ns: g.error_duration,
697                    min_ns: g.error_min,
698                    max_ns: g.error_max,
699                },
700            });
701            stats.push(encode_grouped_stats(k, g));
702        }
703        OtlpStatsBucket {
704            bucket: pb::ClientStatsBucket {
705                start: self.start,
706                duration: bucket_duration,
707                stats,
708                agent_time_shift: 0,
709            },
710            exact,
711        }
712    }
713}
714
715/// Create a ClientGroupedStats struct based on the given AggregationKey and GroupedStats
716fn encode_grouped_stats(key: OwnedAggregationKey, group: GroupedStats) -> pb::ClientGroupedStats {
717    let f = key.fixed;
718    pb::ClientGroupedStats {
719        service: f.service_name,
720        name: f.operation_name,
721        resource: f.resource_name,
722        http_status_code: f.http_status_code,
723        r#type: f.span_type,
724        db_type: String::new(), // db_type is not used yet (see proto definition)
725
726        hits: group.hits,
727        errors: group.errors,
728        duration: group.duration,
729
730        ok_summary: group.ok_summary.encode_to_vec(),
731        error_summary: group.error_summary.encode_to_vec(),
732        synthetics: f.is_synthetics_request,
733        top_level_hits: group.top_level_hits,
734        span_kind: f.span_kind,
735
736        peer_tags: key
737            .peer_tags
738            .into_iter()
739            .map(|(k, v)| {
740                if v.is_empty() {
741                    k.to_string()
742                } else {
743                    format!("{k}:{v}")
744                }
745            })
746            .collect(),
747        is_trace_root: f.is_trace_root.into(),
748        http_method: f.http_method,
749        http_endpoint: f.http_endpoint,
750        grpc_status_code: f
751            .grpc_status_code
752            .map(|c| c.to_string())
753            .unwrap_or_default(),
754        service_source: f.service_source,
755        span_derived_primary_tags: vec![],
756        additional_metric_tags: key
757            .additional_metric_tags
758            .into_iter()
759            .map(|(k, v)| format!("{k}:{v}"))
760            .collect(),
761    }
762}
763
764#[cfg(test)]
765mod tests {
766    use libdd_trace_utils::span::v04::{SpanBytes, SpanSlice};
767
768    use super::*;
769    use libdd_trace_protobuf::pb;
770    use std::hash::Hash;
771
772    fn get_hash(v: &impl Hash) -> u64 {
773        use std::hash::Hasher;
774        let mut hasher = std::hash::DefaultHasher::new();
775        v.hash(&mut hasher);
776        hasher.finish()
777    }
778
779    impl FixedAggregationKey<String> {
780        fn into_key(self) -> OwnedAggregationKey {
781            OwnedAggregationKey {
782                fixed: self,
783                peer_tags: vec![],
784                additional_metric_tags: vec![],
785            }
786        }
787        fn into_key_with_peers(self, peer_tags: Vec<(String, String)>) -> OwnedAggregationKey {
788            OwnedAggregationKey {
789                fixed: self,
790                peer_tags,
791                additional_metric_tags: vec![],
792            }
793        }
794    }
795
796    #[test]
797    fn test_aggregation_key_from_span() {
798        let test_cases: Vec<(SpanBytes, OwnedAggregationKey)> = vec![
799            // Root span
800            (
801                SpanBytes {
802                    service: "service".into(),
803                    name: "op".into(),
804                    resource: "res".into(),
805                    span_id: 1,
806                    parent_id: 0,
807                    ..Default::default()
808                },
809                FixedAggregationKey {
810                    service_name: "service".into(),
811                    operation_name: "op".into(),
812                    resource_name: "res".into(),
813                    is_trace_root: pb::Trilean::True,
814                    ..Default::default()
815                }
816                .into_key(),
817            ),
818            // Span with span kind
819            (
820                SpanBytes {
821                    service: "service".into(),
822                    name: "op".into(),
823                    resource: "res".into(),
824                    span_id: 1,
825                    parent_id: 0,
826                    meta: vec![("span.kind".into(), "client".into())].into(),
827                    ..Default::default()
828                },
829                FixedAggregationKey {
830                    service_name: "service".into(),
831                    operation_name: "op".into(),
832                    resource_name: "res".into(),
833                    span_kind: "client".into(),
834                    is_trace_root: pb::Trilean::True,
835                    ..Default::default()
836                }
837                .into_key(),
838            ),
839            // Span with peer tags but peertags aggregation disabled
840            (
841                SpanBytes {
842                    service: "service".into(),
843                    name: "op".into(),
844                    resource: "res".into(),
845                    span_id: 1,
846                    parent_id: 0,
847                    meta: vec![
848                        ("span.kind".into(), "client".into()),
849                        ("aws.s3.bucket".into(), "bucket-a".into()),
850                    ]
851                    .into(),
852                    ..Default::default()
853                },
854                FixedAggregationKey {
855                    service_name: "service".into(),
856                    operation_name: "op".into(),
857                    resource_name: "res".into(),
858                    span_kind: "client".into(),
859                    is_trace_root: pb::Trilean::True,
860                    ..Default::default()
861                }
862                .into_key(),
863            ),
864            // Span with multiple peer tags but peertags aggregation disabled
865            (
866                SpanBytes {
867                    service: "service".into(),
868                    name: "op".into(),
869                    resource: "res".into(),
870                    span_id: 1,
871                    parent_id: 0,
872                    meta: vec![
873                        ("span.kind".into(), "producer".into()),
874                        ("aws.s3.bucket".into(), "bucket-a".into()),
875                        ("db.instance".into(), "dynamo.test.us1".into()),
876                        ("db.system".into(), "dynamodb".into()),
877                    ]
878                    .into(),
879                    ..Default::default()
880                },
881                FixedAggregationKey {
882                    service_name: "service".into(),
883                    operation_name: "op".into(),
884                    resource_name: "res".into(),
885                    span_kind: "producer".into(),
886                    is_trace_root: pb::Trilean::True,
887                    ..Default::default()
888                }
889                .into_key(),
890            ),
891            // Span with multiple peer tags but peertags aggregation disabled and span kind is
892            // server
893            (
894                SpanBytes {
895                    service: "service".into(),
896                    name: "op".into(),
897                    resource: "res".into(),
898                    span_id: 1,
899                    parent_id: 0,
900                    meta: vec![
901                        ("span.kind".into(), "server".into()),
902                        ("aws.s3.bucket".into(), "bucket-a".into()),
903                        ("db.instance".into(), "dynamo.test.us1".into()),
904                        ("db.system".into(), "dynamodb".into()),
905                    ]
906                    .into(),
907                    ..Default::default()
908                },
909                FixedAggregationKey {
910                    service_name: "service".into(),
911                    operation_name: "op".into(),
912                    resource_name: "res".into(),
913                    span_kind: "server".into(),
914                    is_trace_root: pb::Trilean::True,
915                    ..Default::default()
916                }
917                .into_key(),
918            ),
919            // Span from synthetics
920            (
921                SpanBytes {
922                    service: "service".into(),
923                    name: "op".into(),
924                    resource: "res".into(),
925                    span_id: 1,
926                    parent_id: 0,
927                    meta: vec![("_dd.origin".into(), "synthetics-browser".into())].into(),
928                    ..Default::default()
929                },
930                FixedAggregationKey {
931                    service_name: "service".into(),
932                    operation_name: "op".into(),
933                    resource_name: "res".into(),
934                    is_synthetics_request: true,
935                    is_trace_root: pb::Trilean::True,
936                    ..Default::default()
937                }
938                .into_key(),
939            ),
940            // Span with status code in meta
941            (
942                SpanBytes {
943                    service: "service".into(),
944                    name: "op".into(),
945                    resource: "res".into(),
946                    span_id: 1,
947                    parent_id: 0,
948                    meta: vec![("http.status_code".into(), "418".into())].into(),
949                    ..Default::default()
950                },
951                FixedAggregationKey {
952                    service_name: "service".into(),
953                    operation_name: "op".into(),
954                    resource_name: "res".into(),
955                    is_synthetics_request: false,
956                    is_trace_root: pb::Trilean::True,
957                    http_status_code: 418,
958                    ..Default::default()
959                }
960                .into_key(),
961            ),
962            // Span with invalid status code in meta
963            (
964                SpanBytes {
965                    service: "service".into(),
966                    name: "op".into(),
967                    resource: "res".into(),
968                    span_id: 1,
969                    parent_id: 0,
970                    meta: vec![("http.status_code".into(), "x".into())].into(),
971                    ..Default::default()
972                },
973                FixedAggregationKey {
974                    service_name: "service".into(),
975                    operation_name: "op".into(),
976                    resource_name: "res".into(),
977                    is_synthetics_request: false,
978                    is_trace_root: pb::Trilean::True,
979                    ..Default::default()
980                }
981                .into_key(),
982            ),
983            // Span with status code in metrics
984            (
985                SpanBytes {
986                    service: "service".into(),
987                    name: "op".into(),
988                    resource: "res".into(),
989                    span_id: 1,
990                    parent_id: 0,
991                    metrics: vec![("http.status_code".into(), 418.0)].into(),
992                    ..Default::default()
993                },
994                FixedAggregationKey {
995                    service_name: "service".into(),
996                    operation_name: "op".into(),
997                    resource_name: "res".into(),
998                    is_synthetics_request: false,
999                    is_trace_root: pb::Trilean::True,
1000                    http_status_code: 418,
1001                    ..Default::default()
1002                }
1003                .into_key(),
1004            ),
1005            // Span with http.method and http.route
1006            (
1007                SpanBytes {
1008                    service: "service".into(),
1009                    name: "op".into(),
1010                    resource: "GET /api/v1/users".into(),
1011                    span_id: 1,
1012                    parent_id: 0,
1013                    meta: vec![
1014                        ("http.method".into(), "GET".into()),
1015                        ("http.route".into(), "/api/v1/users".into()),
1016                    ]
1017                    .into(),
1018                    ..Default::default()
1019                },
1020                FixedAggregationKey {
1021                    service_name: "service".into(),
1022                    operation_name: "op".into(),
1023                    resource_name: "GET /api/v1/users".into(),
1024                    http_method: "GET".into(),
1025                    http_endpoint: "/api/v1/users".into(),
1026                    is_synthetics_request: false,
1027                    is_trace_root: pb::Trilean::True,
1028                    ..Default::default()
1029                }
1030                .into_key(),
1031            ),
1032            // Span with http.method and http.endpoint (http.endpoint takes precedence)
1033            (
1034                SpanBytes {
1035                    service: "service".into(),
1036                    name: "op".into(),
1037                    resource: "POST /users/create".into(),
1038                    span_id: 1,
1039                    parent_id: 0,
1040                    meta: vec![
1041                        ("http.method".into(), "POST".into()),
1042                        ("http.route".into(), "/users/create".into()),
1043                        ("http.endpoint".into(), "/users/create2".into()),
1044                    ]
1045                    .into(),
1046                    ..Default::default()
1047                },
1048                FixedAggregationKey {
1049                    service_name: "service".into(),
1050                    operation_name: "op".into(),
1051                    resource_name: "POST /users/create".into(),
1052                    http_method: "POST".into(),
1053                    http_endpoint: "/users/create2".into(),
1054                    is_synthetics_request: false,
1055                    is_trace_root: pb::Trilean::True,
1056                    ..Default::default()
1057                }
1058                .into_key(),
1059            ),
1060            // Span with grpc status from meta as named string
1061            (
1062                SpanBytes {
1063                    meta: vec![("rpc.grpc.status_code".into(), "OK".into())].into(),
1064                    ..Default::default()
1065                },
1066                FixedAggregationKey {
1067                    grpc_status_code: Some(0),
1068                    is_trace_root: pb::Trilean::True,
1069                    ..Default::default()
1070                }
1071                .into_key(),
1072            ),
1073            // grpc.method.name is carried in GroupedStats (for OTLP), not in the aggregation key.
1074            (
1075                SpanBytes {
1076                    meta: vec![("grpc.method.name".into(), "/pkg.Svc/Method".into())].into(),
1077                    ..Default::default()
1078                },
1079                FixedAggregationKey {
1080                    is_trace_root: pb::Trilean::True,
1081                    ..Default::default()
1082                }
1083                .into_key(),
1084            ),
1085            // Span with grpc status from meta as numeric string
1086            (
1087                SpanBytes {
1088                    meta: vec![("rpc.grpc.status_code".into(), "14".into())].into(),
1089                    ..Default::default()
1090                },
1091                FixedAggregationKey {
1092                    grpc_status_code: Some(14),
1093                    is_trace_root: pb::Trilean::True,
1094                    ..Default::default()
1095                }
1096                .into_key(),
1097            ),
1098            // Span with grpc status from meta with StatusCode. prefix
1099            (
1100                SpanBytes {
1101                    meta: vec![("grpc.code".into(), "StatusCode.UNAVAILABLE".into())].into(),
1102                    ..Default::default()
1103                },
1104                FixedAggregationKey {
1105                    grpc_status_code: Some(14),
1106                    is_trace_root: pb::Trilean::True,
1107                    ..Default::default()
1108                }
1109                .into_key(),
1110            ),
1111            // Span with grpc status from metrics takes precedence over meta
1112            (
1113                SpanBytes {
1114                    meta: vec![("rpc.grpc.status_code".into(), "PERMISSION_DENIED".into())].into(),
1115                    metrics: vec![("rpc.grpc.status_code".into(), 2.0)].into(),
1116                    ..Default::default()
1117                },
1118                FixedAggregationKey {
1119                    grpc_status_code: Some(7),
1120                    is_trace_root: pb::Trilean::True,
1121                    ..Default::default()
1122                }
1123                .into_key(),
1124            ),
1125            // Span with grpc status from metrics via secondary key
1126            (
1127                SpanBytes {
1128                    metrics: vec![("grpc.code".into(), 3.0)].into(),
1129                    ..Default::default()
1130                },
1131                FixedAggregationKey {
1132                    grpc_status_code: Some(3),
1133                    is_trace_root: pb::Trilean::True,
1134                    ..Default::default()
1135                }
1136                .into_key(),
1137            ),
1138            // Span with invalid grpc status string
1139            (
1140                SpanBytes {
1141                    meta: vec![("rpc.grpc.status_code".into(), "NOPE".into())].into(),
1142                    ..Default::default()
1143                },
1144                FixedAggregationKey {
1145                    is_trace_root: pb::Trilean::True,
1146                    ..Default::default()
1147                }
1148                .into_key(),
1149            ),
1150            // Span with service source set by integration
1151            (
1152                SpanBytes {
1153                    service: "my-service".into(),
1154                    name: "op".into(),
1155                    resource: "res".into(),
1156                    span_id: 1,
1157                    parent_id: 0,
1158                    meta: vec![("_dd.svc_src".into(), "redis".into())].into(),
1159                    ..Default::default()
1160                },
1161                FixedAggregationKey {
1162                    service_name: "my-service".into(),
1163                    operation_name: "op".into(),
1164                    resource_name: "res".into(),
1165                    is_trace_root: pb::Trilean::True,
1166                    service_source: "redis".into(),
1167                    ..Default::default()
1168                }
1169                .into_key(),
1170            ),
1171            // Span with service source set by configuration option
1172            (
1173                SpanBytes {
1174                    service: "my-service".into(),
1175                    name: "op".into(),
1176                    resource: "res".into(),
1177                    span_id: 1,
1178                    parent_id: 0,
1179                    meta: vec![("_dd.svc_src".into(), "opt.split_by_tag".into())].into(),
1180                    ..Default::default()
1181                },
1182                FixedAggregationKey {
1183                    service_name: "my-service".into(),
1184                    operation_name: "op".into(),
1185                    resource_name: "res".into(),
1186                    is_trace_root: pb::Trilean::True,
1187                    service_source: "opt.split_by_tag".into(),
1188                    ..Default::default()
1189                }
1190                .into_key(),
1191            ),
1192            // Span without service source (default service name)
1193            (
1194                SpanBytes {
1195                    service: "my-service".into(),
1196                    name: "op".into(),
1197                    resource: "res".into(),
1198                    span_id: 1,
1199                    parent_id: 0,
1200                    ..Default::default()
1201                },
1202                FixedAggregationKey {
1203                    service_name: "my-service".into(),
1204                    operation_name: "op".into(),
1205                    resource_name: "res".into(),
1206                    is_trace_root: pb::Trilean::True,
1207                    service_source: "".into(),
1208                    ..Default::default()
1209                }
1210                .into_key(),
1211            ),
1212        ];
1213
1214        let test_peer_tags = vec![
1215            "aws.s3.bucket".to_string(),
1216            "db.instance".to_string(),
1217            "db.system".to_string(),
1218        ];
1219
1220        let test_cases_with_peer_tags: Vec<(SpanSlice, OwnedAggregationKey)> = vec![
1221            // Span with peer tags with peertags aggregation enabled
1222            (
1223                SpanSlice {
1224                    service: "service",
1225                    name: "op",
1226                    resource: "res",
1227                    span_id: 1,
1228                    parent_id: 0,
1229                    meta: vec![("span.kind", "client"), ("aws.s3.bucket", "bucket-a")].into(),
1230                    ..Default::default()
1231                },
1232                FixedAggregationKey {
1233                    service_name: "service".into(),
1234                    operation_name: "op".into(),
1235                    resource_name: "res".into(),
1236                    span_kind: "client".into(),
1237                    is_trace_root: pb::Trilean::True,
1238                    ..Default::default()
1239                }
1240                .into_key_with_peers(vec![("aws.s3.bucket".into(), "bucket-a".into())]),
1241            ),
1242            // Span with multiple peer tags with peertags aggregation enabled
1243            (
1244                SpanSlice {
1245                    service: "service",
1246                    name: "op",
1247                    resource: "res",
1248                    span_id: 1,
1249                    parent_id: 0,
1250                    meta: vec![
1251                        ("span.kind", "producer"),
1252                        ("aws.s3.bucket", "bucket-a"),
1253                        ("db.instance", "dynamo.test.us1"),
1254                        ("db.system", "dynamodb"),
1255                    ]
1256                    .into(),
1257                    ..Default::default()
1258                },
1259                FixedAggregationKey {
1260                    service_name: "service".into(),
1261                    operation_name: "op".into(),
1262                    resource_name: "res".into(),
1263                    span_kind: "producer".into(),
1264                    is_trace_root: pb::Trilean::True,
1265                    ..Default::default()
1266                }
1267                .into_key_with_peers(vec![
1268                    ("aws.s3.bucket".into(), "bucket-a".into()),
1269                    ("db.instance".into(), "dynamo.test.us1".into()),
1270                    ("db.system".into(), "dynamodb".into()),
1271                ]),
1272            ),
1273            // Span with multiple peer tags with peertags aggregation enabled and span kind is
1274            // server
1275            (
1276                SpanSlice {
1277                    service: "service",
1278                    name: "op",
1279                    resource: "res",
1280                    span_id: 1,
1281                    parent_id: 0,
1282                    meta: vec![
1283                        ("span.kind", "server"),
1284                        ("aws.s3.bucket", "bucket-a"),
1285                        ("db.instance", "dynamo.test.us1"),
1286                        ("db.system", "dynamodb"),
1287                    ]
1288                    .into(),
1289                    ..Default::default()
1290                },
1291                FixedAggregationKey {
1292                    service_name: "service".into(),
1293                    operation_name: "op".into(),
1294                    resource_name: "res".into(),
1295                    span_kind: "server".into(),
1296                    is_trace_root: pb::Trilean::True,
1297                    ..Default::default()
1298                }
1299                .into_key(),
1300            ),
1301        ];
1302
1303        for (span, expected_key) in test_cases {
1304            let borrowed_key = BorrowedAggregationKey::from_span(&span, &[], &[]);
1305            assert_eq!(
1306                OwnedAggregationKey::from(&borrowed_key),
1307                expected_key,
1308                "for span {span:?}"
1309            );
1310            assert_eq!(
1311                get_hash(&borrowed_key),
1312                get_hash(&OwnedAggregationKey::from(&borrowed_key))
1313            );
1314        }
1315
1316        for (span, expected_key) in test_cases_with_peer_tags {
1317            let borrowed_key =
1318                BorrowedAggregationKey::from_span(&span, test_peer_tags.as_slice(), &[]);
1319            assert_eq!(OwnedAggregationKey::from(&borrowed_key), expected_key);
1320            assert_eq!(
1321                get_hash(&borrowed_key),
1322                get_hash(&OwnedAggregationKey::from(&borrowed_key))
1323            );
1324        }
1325    }
1326
1327    #[test]
1328    fn test_peer_tag_ip_quantization_in_aggregation_key() {
1329        let peer_tag_keys = vec!["peer.hostname".to_string(), "db.instance".to_string()];
1330
1331        // IPv4 address peer tag gets replaced with blocked-ip-address
1332        let span_ipv4 = SpanSlice {
1333            service: "service",
1334            name: "op",
1335            resource: "res",
1336            span_id: 1,
1337            parent_id: 0,
1338            meta: vec![
1339                ("span.kind", "client"),
1340                ("peer.hostname", "10.1.2.3"),
1341                ("db.instance", "my-db"),
1342            ]
1343            .into(),
1344            ..Default::default()
1345        };
1346        let key = BorrowedAggregationKey::from_span(&span_ipv4, &peer_tag_keys, &[]);
1347        let owned = OwnedAggregationKey::from(&key);
1348        assert_eq!(
1349            owned.peer_tags,
1350            vec![
1351                (
1352                    "peer.hostname".to_string(),
1353                    "blocked-ip-address".to_string()
1354                ),
1355                ("db.instance".to_string(), "my-db".to_string()),
1356            ]
1357        );
1358
1359        // IPv6 address peer tag gets replaced with blocked-ip-address
1360        let span_ipv6 = SpanSlice {
1361            service: "service",
1362            name: "op",
1363            resource: "res",
1364            span_id: 1,
1365            parent_id: 0,
1366            meta: vec![
1367                ("span.kind", "client"),
1368                ("peer.hostname", "2001:db8:3333:4444:CCCC:DDDD:EEEE:FFFF"),
1369            ]
1370            .into(),
1371            ..Default::default()
1372        };
1373        let ipv6_keys = vec!["peer.hostname".to_string()];
1374        let key = BorrowedAggregationKey::from_span(&span_ipv6, &ipv6_keys, &[]);
1375        let owned = OwnedAggregationKey::from(&key);
1376        assert_eq!(
1377            owned.peer_tags,
1378            vec![(
1379                "peer.hostname".to_string(),
1380                "blocked-ip-address".to_string()
1381            )]
1382        );
1383
1384        // Non-IP peer tags pass through unchanged
1385        let span_non_ip = SpanSlice {
1386            service: "service",
1387            name: "op",
1388            resource: "res",
1389            span_id: 1,
1390            parent_id: 0,
1391            meta: vec![("span.kind", "client"), ("db.instance", "dynamo.test.us1")].into(),
1392            ..Default::default()
1393        };
1394        let non_ip_keys = vec!["db.instance".to_string()];
1395        let key = BorrowedAggregationKey::from_span(&span_non_ip, &non_ip_keys, &[]);
1396        let owned = OwnedAggregationKey::from(&key);
1397        assert_eq!(
1398            owned.peer_tags,
1399            vec![("db.instance".to_string(), "dynamo.test.us1".to_string())]
1400        );
1401    }
1402
1403    #[test]
1404    fn test_grpc_status_str_to_int_value() {
1405        // Numeric strings parse directly
1406        assert_eq!(grpc_status_str_to_int_value("0"), Some(0));
1407        assert_eq!(grpc_status_str_to_int_value("14"), Some(14));
1408        assert_eq!(grpc_status_str_to_int_value("255"), Some(255));
1409        assert_eq!(grpc_status_str_to_int_value("256"), None);
1410        assert_eq!(grpc_status_str_to_int_value("-1"), None);
1411
1412        // Named status codes (uppercase)
1413        assert_eq!(grpc_status_str_to_int_value("OK"), Some(0));
1414        assert_eq!(grpc_status_str_to_int_value("CANCELLED"), Some(1));
1415        assert_eq!(grpc_status_str_to_int_value("UNKNOWN"), Some(2));
1416        assert_eq!(grpc_status_str_to_int_value("INVALID_ARGUMENT"), Some(3));
1417        assert_eq!(grpc_status_str_to_int_value("DEADLINE_EXCEEDED"), Some(4));
1418        assert_eq!(grpc_status_str_to_int_value("NOT_FOUND"), Some(5));
1419        assert_eq!(grpc_status_str_to_int_value("ALREADY_EXISTS"), Some(6));
1420        assert_eq!(grpc_status_str_to_int_value("PERMISSION_DENIED"), Some(7));
1421        assert_eq!(grpc_status_str_to_int_value("UNAUTHENTICATED"), Some(16));
1422        assert_eq!(grpc_status_str_to_int_value("RESOURCE_EXHAUSTED"), Some(8));
1423        assert_eq!(grpc_status_str_to_int_value("FAILED_PRECONDITION"), Some(9));
1424        assert_eq!(grpc_status_str_to_int_value("ABORTED"), Some(10));
1425        assert_eq!(grpc_status_str_to_int_value("OUT_OF_RANGE"), Some(11));
1426        assert_eq!(grpc_status_str_to_int_value("UNIMPLEMENTED"), Some(12));
1427        assert_eq!(grpc_status_str_to_int_value("INTERNAL"), Some(13));
1428        assert_eq!(grpc_status_str_to_int_value("UNAVAILABLE"), Some(14));
1429        assert_eq!(grpc_status_str_to_int_value("DATA_LOSS"), Some(15));
1430
1431        // Case-insensitive matching
1432        assert_eq!(grpc_status_str_to_int_value("ok"), Some(0));
1433        assert_eq!(grpc_status_str_to_int_value("Cancelled"), Some(1));
1434        assert_eq!(grpc_status_str_to_int_value("not_found"), Some(5));
1435
1436        // StatusCode. prefix is stripped
1437        assert_eq!(grpc_status_str_to_int_value("StatusCode.OK"), Some(0));
1438        assert_eq!(
1439            grpc_status_str_to_int_value("StatusCode.UNAVAILABLE"),
1440            Some(14)
1441        );
1442        assert_eq!(
1443            grpc_status_str_to_int_value("StatusCode.not_found"),
1444            Some(5)
1445        );
1446
1447        // Alternate spellings
1448        assert_eq!(grpc_status_str_to_int_value("CANCELED"), Some(1));
1449
1450        // Unknown / empty strings
1451        assert_eq!(grpc_status_str_to_int_value("NOPE"), None);
1452        assert_eq!(grpc_status_str_to_int_value(""), None);
1453        assert_eq!(
1454            grpc_status_str_to_int_value("this_is_a_kinda_long_string_that_needs_upcasing"),
1455            None
1456        );
1457
1458        // Non ascii
1459        assert_eq!(grpc_status_str_to_int_value("🤣"), None);
1460    }
1461}