libdd_trace_stats/span_concentrator/

aggregation.rs

// Copyright 2024-Present Datadog, Inc. https://www.datadoghq.com/
// SPDX-License-Identifier: Apache-2.0

//! This module implement the logic for stats aggregation into time buckets and stats group.
//! This includes the aggregation key to group spans together and the computation of stats from a
//! span.

use hashbrown::{HashMap, HashSet};
use libdd_trace_obfuscation::ip_address::quantize_peer_ip_addresses;
use libdd_trace_protobuf::pb;
use std::{
    borrow::{Borrow, Cow},
    hash::{DefaultHasher, Hash, Hasher as _},
};
use tracing::warn;

use crate::span_concentrator::{cardinality_limit_telemetry::CollapsedFieldSet, StatSpan};

use super::{
    cardinality_limit_telemetry::{self, CollapsedFieldsMetrics},
    CardinalityLimitConfig,
};

/// Sentinel value used for cardinality limiting.
pub const TRACER_BLOCKED_VALUE: &str = "tracer_blocked_value";

const TAG_STATUS_CODE: &str = "http.status_code";
const TAG_METHOD: &str = "http.method";
// OpenTelemetry HTTP semantic-convention names for the two attributes above. A tracer running in
// OTel semantics mode (DD_TRACE_OTEL_SEMANTICS_ENABLED) emits these instead of the Datadog names,
// so reading only the Datadog names silently drops the status and method stats dimensions: the
// bucket reports status 0 and an empty method while the span itself is correct.
// http.endpoint and http.route need no equivalent: http.route is already the OTel name, and
// http.endpoint is Datadog-only and retained in that mode.
const TAG_STATUS_CODE_OTEL: &str = "http.response.status_code";
const TAG_METHOD_OTEL: &str = "http.request.method";
const ADDITIONAL_METRIC_TAG_VALUE_MAX_LEN: usize = 200;
const TAG_SYNTHETICS: &str = "synthetics";
const TAG_SPANKIND: &str = "span.kind";
const TAG_ORIGIN: &str = "_dd.origin";
const TAG_SVC_SRC: &str = "_dd.svc_src";
const GRPC_STATUS_CODE_FIELD: &[&str] = &[
    "rpc.grpc.status_code",
    "grpc.code",
    "rpc.grpc.status.code",
    "grpc.status.code",
];

/// Aggregation key fields shared across all concentrator implementations — everything
/// **except** peer tags.
///
/// `T` is the string representation:
/// * `&'a str`   — borrowed references used in [`BorrowedAggregationKey`]
/// * `String`    — owned values used in `OwnedAggregationKey`
/// * `StringRef` — offset+len into a SHM string pool, used in `ShmKeyHeader`
#[derive(
    Clone, Default, Hash, Eq, PartialEq, Debug, PartialOrd, serde::Serialize, serde::Deserialize,
)]
pub struct FixedAggregationKey<T> {
    pub resource_name: T,
    pub service_name: T,
    pub operation_name: T,
    pub span_type: T,
    pub span_kind: T,
    pub http_method: T,
    pub http_endpoint: T,
    pub service_source: T,
    pub http_status_code: u32,
    pub grpc_status_code: Option<u8>,
    pub is_synthetics_request: bool,
    pub is_trace_root: pb::Trilean,
}

impl<T> FixedAggregationKey<T> {
    /// Map all string fields through `f`, preserving scalar fields.
    pub fn convert<'a, V: 'a, I: ?Sized + 'a, F: Fn(&'a I) -> V>(
        &'a self,
        f: F,
    ) -> FixedAggregationKey<V>
    where
        T: Borrow<I>,
    {
        FixedAggregationKey {
            resource_name: f(self.resource_name.borrow()),
            service_name: f(self.service_name.borrow()),
            operation_name: f(self.operation_name.borrow()),
            span_type: f(self.span_type.borrow()),
            span_kind: f(self.span_kind.borrow()),
            http_method: f(self.http_method.borrow()),
            http_endpoint: f(self.http_endpoint.borrow()),
            service_source: f(self.service_source.borrow()),
            http_status_code: self.http_status_code,
            grpc_status_code: self.grpc_status_code,
            is_synthetics_request: self.is_synthetics_request,
            is_trace_root: self.is_trace_root,
        }
    }
}

#[derive(Clone, Hash, PartialEq, Eq)]
/// Represent a stats aggregation key borrowed from span data
pub(super) struct BorrowedAggregationKey<'a> {
    fixed: FixedAggregationKey<&'a str>,
    peer_tags: Vec<(&'a str, Cow<'a, str>)>,
    additional_metric_tags: Vec<(&'a str, &'a str)>,
}

impl hashbrown::Equivalent<OwnedAggregationKey> for BorrowedAggregationKey<'_> {
    #[inline]
    fn equivalent(&self, other: &OwnedAggregationKey) -> bool {
        self.fixed == other.fixed.convert(|s| s)
            && self.peer_tags.len() == other.peer_tags.len()
            && self
                .peer_tags
                .iter()
                .zip(other.peer_tags.iter())
                .all(|((k1, v1), (k2, v2))| k1 == k2 && v1 == v2)
            && self.additional_metric_tags.len() == other.additional_metric_tags.len()
            && self
                .additional_metric_tags
                .iter()
                .zip(other.additional_metric_tags.iter())
                .all(|((k1, v1), (k2, v2))| k1 == k2 && v1 == v2)
    }
}

#[derive(Debug, Clone, Hash, PartialEq, Eq, PartialOrd, Default)]
/// Represents a span aggregation key with owned data
///
/// To be able to use BorrowedAggregationKey to index into a stats bucket hashmap two
/// conditions must stay true:
/// * Hashing an owned key derived from a borrowed key should produce the same hash as hashing the
///   borrowed key
/// * Running the Equivalent trait on an owned key derived from a borrowed key should produce true
pub(super) struct OwnedAggregationKey {
    fixed: FixedAggregationKey<String>,
    peer_tags: Vec<(String, String)>,
    additional_metric_tags: Vec<(String, String)>,
}

impl From<&BorrowedAggregationKey<'_>> for OwnedAggregationKey {
    fn from(value: &BorrowedAggregationKey<'_>) -> Self {
        OwnedAggregationKey {
            fixed: value.fixed.convert(str::to_owned),
            peer_tags: value
                .peer_tags
                .iter()
                .map(|(k, v)| (k.to_string(), v.to_string()))
                .collect(),
            additional_metric_tags: value
                .additional_metric_tags
                .iter()
                .map(|(k, v)| (k.to_string(), v.to_string()))
                .collect(),
        }
    }
}

fn float_to_int(f: f64) -> Option<u8> {
    if f.floor() != f {
        return None;
    }
    if f < 0.0 || (u8::MAX as f64) < f {
        return None;
    }
    Some(f as u8)
}

fn float_to_u32(f: f64) -> Option<u32> {
    if !f.is_finite() || f.floor() != f || f < 0.0 || (u32::MAX as f64) < f {
        return None;
    }
    // The checks above guarantee the finite integral value is representable without loss.
    Some(f as u32)
}

fn is_valid_http_status_code(code: &u32) -> bool {
    (100..600).contains(code)
}

/// The HTTP method, under either naming convention, or "" when the span carries neither.
///
/// The Datadog name is canonical in the Agent semantic registry and is preferred when both names
/// are present.
fn get_http_method<'a>(span: &'a impl StatSpan<'a>) -> &'a str {
    for key in [TAG_METHOD, TAG_METHOD_OTEL] {
        if let Some(value) = span.get_meta(key) {
            if !value.is_empty() {
                return value;
            }
        }
    }

    ""
}

/// The HTTP status code, under either naming convention, or 0 when the span carries neither.
///
/// `metrics` is checked before `meta` for each name because a numeric attribute is routed there
/// by the tracer's own setter, and OTel types `http.response.status_code` as an int. The Datadog
/// key retains its existing conversion and precedence behavior. Invalid OTel values are ignored.
fn get_http_status_code<'a>(span: &'a impl StatSpan<'a>) -> u32 {
    for key in [TAG_STATUS_CODE, TAG_STATUS_CODE_OTEL] {
        if let Some(value) = span.get_metrics(key) {
            if key == TAG_STATUS_CODE {
                // Preserve the existing Datadog-key conversion behavior.
                return value as u32;
            }
            if let Some(code) = float_to_u32(value).filter(is_valid_http_status_code) {
                return code;
            }
        }

        if let Some(value) = span.get_meta(key) {
            if key == TAG_STATUS_CODE {
                // Preserve the existing Datadog-key parse-or-zero behavior.
                return value.parse().unwrap_or_default();
            }
            if let Some(code) = value.parse().ok().filter(is_valid_http_status_code) {
                return code;
            }
        }
    }

    0
}

fn get_grpc_status_code<'a>(span: &'a impl StatSpan<'a>) -> Option<u8> {
    for key in GRPC_STATUS_CODE_FIELD {
        if let Some(val) = span.get_meta(key) {
            if let Some(code) = grpc_status_str_to_int_value(val) {
                return Some(code);
            }
        }
    }

    for key in GRPC_STATUS_CODE_FIELD {
        if let Some(val) = span.get_metrics(key) {
            if let Some(code) = float_to_int(val) {
                return Some(code);
            }
        }
    }

    None
}

fn grpc_status_str_to_int_value(v: &str) -> Option<u8> {
    if let Ok(status) = v.parse() {
        return Some(status);
    }
    let mut status_uppercase = [0u8; 32];
    let mut status = v.trim_start_matches("StatusCode.");

    let mut needs_upcasing = false;
    for b in status.as_bytes() {
        if !b.is_ascii() {
            return None;
        }
        needs_upcasing |= b.is_ascii_lowercase()
    }
    if needs_upcasing {
        for (c, d) in status.as_bytes().iter().zip(&mut status_uppercase) {
            *d = c.to_ascii_uppercase();
        }
        status = std::str::from_utf8(&status_uppercase[0..status.len().min(status_uppercase.len())])
            .ok()?
    }

    match status {
        "OK" => return Some(0),
        "CANCELLED" | "CANCELED" => return Some(1),
        "UNKNOWN" => return Some(2),
        "INVALID_ARGUMENT" | "INVALIDARGUMENT" => return Some(3),
        "DEADLINE_EXCEEDED" | "DEADLINEEXCEEDED" => return Some(4),
        "NOT_FOUND" | "NOTFOUND" => return Some(5),
        "ALREADY_EXISTS" | "ALREADYEXISTS" => return Some(6),
        "PERMISSION_DENIED" | "PERMISSIONDENIED" => return Some(7),
        "UNAUTHENTICATED" => return Some(16),
        "RESOURCE_EXHAUSTED" | "RESOURCEEXHAUSTED" => return Some(8),
        "FAILED_PRECONDITION" | "FAILEDPRECONDITION" => return Some(9),
        "ABORTED" => return Some(10),
        "OUT_OF_RANGE" | "OUTOFRANGE" => return Some(11),
        "UNIMPLEMENTED" => return Some(12),
        "INTERNAL" => return Some(13),
        "UNAVAILABLE" => return Some(14),
        "DATA_LOSS" | "DATALOSS" => return Some(15),
        _ => {}
    }
    None
}

impl<'a> BorrowedAggregationKey<'a> {
    /// Return an AggregationKey matching the given span.
    ///
    /// If `peer_tag_keys` is not empty then the peer tags of the span will be included in the
    /// key.
    /// If `additional_metric_tags` is not empty then matching span tags keys are included in the
    /// key.
    pub(super) fn from_span<T: StatSpan<'a>>(
        span: &'a T,
        peer_tag_keys: &'a [String],
        additional_metric_tag_keys: &'a [String],
    ) -> Self {
        Self::from_obfuscated_span(
            span.resource(),
            span,
            peer_tag_keys,
            additional_metric_tag_keys,
        )
    }

    pub(crate) fn from_obfuscated_span<'b, T>(
        resource_name: &'a str,
        span: &'b T,
        peer_tag_keys: &'b [String],
        additional_metric_tag_keys: &'b [String],
    ) -> BorrowedAggregationKey<'a>
    where
        T: StatSpan<'b>,
        // resource_name is a temporary string on the stack the span will outlive it
        'b: '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`. IP address values are quantized to reduce cardinality.
            peer_tag_keys
                .iter()
                .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", Cow::Borrowed(base_service))]
        } else {
            vec![]
        };

        // The Datadog name wins when both are present. An empty value counts as absent, so a
        // tracer that writes the key unconditionally cannot shadow a real value under the other
        // name, which is the mistake get_grpc_status_code above already avoids.
        let http_method = get_http_method(span);

        let http_endpoint = span
            .get_meta("http.route")
            .filter(|value| !value.is_empty())
            .or_else(|| {
                span.get_meta("http.endpoint")
                    .filter(|value| !value.is_empty())
            })
            .unwrap_or_default();

        let status_code = get_http_status_code(span);

        let grpc_status_code = get_grpc_status_code(span);
        let service_source = span.get_meta(TAG_SVC_SRC).unwrap_or_default();

        let additional_metric_tags: Vec<(&'a str, &'a str)> = additional_metric_tag_keys
            .iter()
            .filter_map(|key| match span.get_meta(key.as_str()) {
                Some(v) if !v.is_empty() => {
                    // Byte length >= char count, so skip the char walk when byte length alone
                    // is within the max character length, otherwise stop as soon as we pass the max character length.
                    if v.len() > ADDITIONAL_METRIC_TAG_VALUE_MAX_LEN
                        && v.chars().nth(ADDITIONAL_METRIC_TAG_VALUE_MAX_LEN).is_some()
                    {
                        warn!(
                            "additional_metric_tags: value for key '{}' exceeds {} characters; substituting tracer_blocked_value",
                            key, ADDITIONAL_METRIC_TAG_VALUE_MAX_LEN,
                        );
                        Some((key.as_str(), TRACER_BLOCKED_VALUE))
                    } else {
                        Some((key.as_str(), v))
                    }
                }
                _ => None,
            })
            .collect();

        Self {
            fixed: FixedAggregationKey {
                resource_name,
                service_name: span.service(),
                operation_name: span.name(),
                span_type: span.r#type(),
                span_kind,
                http_method,
                http_endpoint,
                service_source,
                http_status_code: status_code,
                grpc_status_code,
                is_synthetics_request: span
                    .get_meta(TAG_ORIGIN)
                    .is_some_and(|origin| origin.starts_with(TAG_SYNTHETICS)),
                is_trace_root: if span.is_trace_root() {
                    pb::Trilean::True
                } else {
                    pb::Trilean::False
                },
            },
            peer_tags,
            additional_metric_tags,
        }
    }

    /// Truncates string fields in accordance with the cardinality limit RFC
    ///
    /// This should be called only after obfuscation
    #[cfg_attr(not(feature = "stats-obfuscation"), allow(unused))]
    pub(crate) fn truncate(&mut self, big_resource: bool) {
        let resource_length_limit = if big_resource { 15_000 } else { 5000 };
        self.fixed.resource_name = slice_up_to(self.fixed.resource_name, resource_length_limit);
        self.fixed.service_name = slice_up_to(self.fixed.service_name, 100);
        self.fixed.operation_name = slice_up_to(self.fixed.operation_name, 100);
        self.fixed.span_type = slice_up_to(self.fixed.span_type, 100);
    }
}

/// Truncate `s` to at most `max_len` bytes
fn slice_up_to(s: &str, max_len: usize) -> &str {
    if max_len >= s.len() {
        return s;
    }
    // TODO: use `floor_char_boundary` once our MSRV is 1.91 or higher
    // (https://doc.rust-lang.org/std/primitive.str.html#method.floor_char_boundary)
    let mut idx = max_len;
    while !s.is_char_boundary(idx) {
        idx -= 1;
    }
    &s[..idx]
}

impl OwnedAggregationKey {
    /// Return the overflow sentinel key.
    pub(super) fn overflow_key() -> Self {
        OwnedAggregationKey {
            fixed: FixedAggregationKey {
                resource_name: TRACER_BLOCKED_VALUE.to_owned(),
                service_name: TRACER_BLOCKED_VALUE.to_owned(),
                operation_name: TRACER_BLOCKED_VALUE.to_owned(),
                span_type: TRACER_BLOCKED_VALUE.to_owned(),
                span_kind: TRACER_BLOCKED_VALUE.to_owned(),
                http_method: TRACER_BLOCKED_VALUE.to_owned(),
                http_endpoint: TRACER_BLOCKED_VALUE.to_owned(),
                service_source: TRACER_BLOCKED_VALUE.to_owned(),
                http_status_code: 0,
                grpc_status_code: None,
                is_synthetics_request: false,
                is_trace_root: pb::Trilean::NotSet,
            },
            peer_tags: vec![(TRACER_BLOCKED_VALUE.to_owned(), "".to_owned())],
            additional_metric_tags: vec![(TRACER_BLOCKED_VALUE.to_owned(), "".to_owned())],
        }
    }
}

impl From<pb::ClientGroupedStats> for OwnedAggregationKey {
    fn from(value: pb::ClientGroupedStats) -> Self {
        Self {
            fixed: FixedAggregationKey {
                resource_name: value.resource,
                service_name: value.service,
                operation_name: value.name,
                span_type: value.r#type,
                span_kind: value.span_kind,
                http_method: value.http_method,
                http_endpoint: value.http_endpoint,
                service_source: value.service_source,
                http_status_code: value.http_status_code,
                grpc_status_code: value.grpc_status_code.parse().ok(),
                is_synthetics_request: value.synthetics,
                is_trace_root: pb::Trilean::try_from(value.is_trace_root)
                    .unwrap_or(pb::Trilean::NotSet),
            },
            peer_tags: value
                .peer_tags
                .into_iter()
                .filter_map(|t| {
                    let (key, value) = t.split_once(':')?;
                    Some((key.to_string(), value.to_string()))
                })
                .collect(),
            additional_metric_tags: value
                .additional_metric_tags
                .into_iter()
                .filter_map(|t| {
                    let (key, value) = t.split_once(':')?;
                    Some((key.to_string(), value.to_string()))
                })
                .collect(),
        }
    }
}

/// Return true if we care about peer tags on the span
fn should_track_peer_tags(span_kind: &str) -> bool {
    matches!(
        span_kind.to_lowercase().as_str(),
        "client" | "producer" | "consumer"
    )
}

/// The stats computed from a group of span with the same AggregationKey
#[derive(Debug, Default, Clone)]
pub(super) struct GroupedStats {
    hits: u64,
    errors: u64,
    duration: u64,
    top_level_hits: u64,
    ok_summary: libdd_ddsketch::DDSketch,
    error_summary: libdd_ddsketch::DDSketch,
    // Exact per-cell (ok/error) scalars used by the OTLP trace-metrics path. These are tracked
    // separately from `duration` so the /v0.6/stats agent payload is byte-for-byte unchanged.
    ok_duration: u64,
    ok_min: u64,
    ok_max: u64,
    error_duration: u64,
    error_min: u64,
    error_max: u64,
}

impl GroupedStats {
    /// Update the stats of a GroupedStats by inserting a span.
    fn insert(&mut self, duration: i64, is_error: bool, is_top_level: bool) {
        self.hits += 1;
        self.duration += duration as u64;
        let d = duration as u64;
        if is_error {
            self.errors += 1;
            let _ = self.error_summary.add(duration as f64);
            self.error_duration += d;
            self.error_min = if self.errors == 1 {
                d
            } else {
                self.error_min.min(d)
            };
            self.error_max = self.error_max.max(d);
        } else {
            let _ = self.ok_summary.add(duration as f64);
            self.ok_duration += d;
            let ok_count = self.hits - self.errors;
            self.ok_min = if ok_count == 1 { d } else { self.ok_min.min(d) };
            self.ok_max = self.ok_max.max(d);
        }
        if is_top_level {
            self.top_level_hits += 1;
        }
    }
}

/// Exact per-cell (ok or error) scalars for one aggregation group, surfaced to the OTLP
/// trace-metrics path. `count` is exact; `duration_ns`/`min_ns`/`max_ns` are exact when
/// `count > 0` and meaningless otherwise (the OTLP mapper suppresses empty cells).
#[derive(Debug, Clone, Copy, Default)]
pub struct OtlpExactCell {
    pub count: u64,
    pub duration_ns: u64,
    pub min_ns: u64,
    pub max_ns: u64,
}

/// Exact OK/ERROR cells for one aggregation group, in the same order as the `stats` vector
/// of the accompanying [`pb::ClientStatsBucket`].
#[derive(Debug, Clone, Default)]
pub struct OtlpExactGroup {
    pub ok: OtlpExactCell,
    pub error: OtlpExactCell,
}

/// A bucket flushed for the OTLP trace-metrics path. `exact[i]` is the exact-scalar sidecar
/// for `bucket.stats[i]`; the protobuf bucket itself is identical to what the agent path uses.
#[derive(Debug, Clone)]
pub struct OtlpStatsBucket {
    pub bucket: pb::ClientStatsBucket,
    pub exact: Vec<OtlpExactGroup>,
}

/// A time bucket used for stats aggregation. It stores a map of GroupedStats storing the stats of
/// spans aggregated on their AggregationKey.
#[derive(Debug, Clone)]
pub(super) struct StatsBucket {
    data: HashMap<OwnedAggregationKey, GroupedStats>,
    start: u64,
    /// Maximum number of distinct aggregation keys this bucket will hold before collapsing new
    /// ones into the overflow sentinel key.
    cardinality_limits: CardinalityLimitConfig,
    // HashSet of hashes of field values so we save memory
    // This is not 100% accurate but the probability of getting collision is close to 0
    // In the very rare case we get a collision, we would get one extra bucket which is totally
    // fine
    distinct_resources: HashSet<u64>,
    distinct_http_endpoints: HashSet<u64>,
    distinct_peer_tags: HashSet<u64>,
    distinct_additional_tags: HashSet<u64>,
    /// 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,
}

impl StatsBucket {
    /// Return a new StatsBucket starting at `start_timestamp`.
    ///
    /// `cardinality_limits` are the values for whole-key and per-field cardinality limits
    pub(super) fn new(
        start_timestamp: u64,
        cardinality_limits: CardinalityLimitConfig,
        #[cfg(feature = "stats-obfuscation")] obfuscation_enabled: bool,
    ) -> Self {
        Self {
            data: HashMap::new(),
            start: start_timestamp,
            cardinality_limits,
            collapsed_count: 0,
            #[cfg(feature = "stats-obfuscation")]
            obfuscated: obfuscation_enabled,
            distinct_resources: HashSet::new(),
            distinct_http_endpoints: HashSet::new(),
            distinct_peer_tags: HashSet::new(),
            distinct_additional_tags: HashSet::new(),
            collapsed_fields_metrics: cardinality_limit_telemetry::CollapsedFieldsMetrics::zero(),
        }
    }

    /// Returns metrics on spans field collapse with reasons.
    pub fn collapsed_fields_metrics(&self) -> cardinality_limit_telemetry::CollapsedFieldsMetrics {
        self.collapsed_fields_metrics
    }

    /// Return the number of spans collapsed into the overflow bucket.
    pub(super) fn collapsed_count(&self) -> u64 {
        self.collapsed_count
    }

    /// Insert a value as stats in the group corresponding to the aggregation key, if it does not
    /// exist it creates it.
    ///
    /// Keys that already exist in this bucket always merge normally. A new key is subject to the
    /// `max_entries` limit, which collapses it into the overflow sentinel key.
    pub(super) fn insert(
        &mut self,
        mut key: BorrowedAggregationKey<'_>,
        duration: i64,
        is_error: bool,
        is_top_level: bool,
    ) {
        // Per field cardinality limiting
        self.collapse_key_fields_cardinality(&mut key);

        // The map can't change size before the entry below is resolved, so this single read
        // covers the `max_entries` check in the vacant branch without a further lookup.
        let len_before_insert = self.data.len();

        match self.data.entry_ref(&key) {
            // Existing key, merge
            hashbrown::hash_map::EntryRef::Occupied(mut e) => {
                e.get_mut().insert(duration, is_error, is_top_level);
            }
            hashbrown::hash_map::EntryRef::Vacant(e) => {
                // New key over the max entry limit, collapse into the overflow
                // sentinel.
                if len_before_insert >= self.cardinality_limits.whole_key_limit {
                    self.collapsed_count += 1;
                    self.data
                        .entry(OwnedAggregationKey::overflow_key())
                        .or_default()
                        .insert(duration, is_error, is_top_level);
                    return;
                }
                // Within the max entry limit, admit key as a new distinct entry.
                e.insert(GroupedStats::default())
                    .insert(duration, is_error, is_top_level);
            }
        }
    }

    /// Collapse an aggregation key fields following the bucket's `CardinalityLimitConfig`
    fn collapse_key_fields_cardinality(&mut self, key: &mut BorrowedAggregationKey<'_>) {
        use hashbrown::hash_set::Entry;
        fn hash(input: &impl Hash) -> u64 {
            let mut hasher = DefaultHasher::new();
            input.hash(&mut hasher);
            hasher.finish()
        }

        let mut collapsed_fields = CollapsedFieldSet::empty();

        let resource_hash = hash(&key.fixed.resource_name);
        let resources_count = self.distinct_resources.len();
        if let Entry::Vacant(slot) = self.distinct_resources.entry(resource_hash) {
            if resources_count >= self.cardinality_limits.resource_limit {
                key.fixed.resource_name = TRACER_BLOCKED_VALUE;
                collapsed_fields.add(CollapsedFieldSet::RESOURCE_NAME);
            } else {
                slot.insert();
            }
        }

        let http_endpoint_hash = hash(&key.fixed.http_endpoint);
        let http_endpoints_count = self.distinct_http_endpoints.len();
        if let Entry::Vacant(slot) = self.distinct_http_endpoints.entry(http_endpoint_hash) {
            if http_endpoints_count >= self.cardinality_limits.http_endpoint_limit {
                key.fixed.http_endpoint = TRACER_BLOCKED_VALUE;
                collapsed_fields.add(CollapsedFieldSet::HTTP_ENDPOINT);
            } else {
                slot.insert();
            }
        }

        let peer_tags_hash = hash(&key.peer_tags);
        let peer_tags_count = self.distinct_peer_tags.len();
        if let Entry::Vacant(slot) = self.distinct_peer_tags.entry(peer_tags_hash) {
            if peer_tags_count >= self.cardinality_limits.peer_tags_limit {
                key.peer_tags = vec![(TRACER_BLOCKED_VALUE, Cow::Borrowed(""))];
                collapsed_fields.add(CollapsedFieldSet::PEER_TAGS);
            } else {
                slot.insert();
            }
        }

        let additional_tags_hash = hash(&key.additional_metric_tags);
        let additional_tags_count = self.distinct_additional_tags.len();
        if let Entry::Vacant(slot) = self.distinct_additional_tags.entry(additional_tags_hash) {
            if additional_tags_count >= self.cardinality_limits.additional_tags_limit {
                key.additional_metric_tags = vec![(TRACER_BLOCKED_VALUE, "")];
                collapsed_fields.add(CollapsedFieldSet::ADDITIONAL_TAGS);
            } else {
                slot.insert();
            }
        }
        self.collapsed_fields_metrics.increment(collapsed_fields);
    }

    /// Consume the bucket and return a ClientStatsBucket containing the bucket stats.
    /// `bucket_duration` is the size of buckets for the concentrator containing the bucket.
    pub(super) fn flush(self, bucket_duration: u64) -> pb::ClientStatsBucket {
        self.flush_with_otlp_exact(bucket_duration).bucket
    }

    /// Like [`Self::flush`], but additionally produces exact per-cell scalars for the OTLP
    /// trace-metrics path. The `bucket` field is identical to what [`Self::flush`] returns.
    pub(super) fn flush_with_otlp_exact(self, bucket_duration: u64) -> OtlpStatsBucket {
        let mut stats = Vec::with_capacity(self.data.len());
        let mut exact = Vec::with_capacity(self.data.len());
        for (k, g) in self.data {
            exact.push(OtlpExactGroup {
                ok: OtlpExactCell {
                    count: g.hits.saturating_sub(g.errors),
                    duration_ns: g.ok_duration,
                    min_ns: g.ok_min,
                    max_ns: g.ok_max,
                },
                error: OtlpExactCell {
                    count: g.errors,
                    duration_ns: g.error_duration,
                    min_ns: g.error_min,
                    max_ns: g.error_max,
                },
            });
            stats.push(encode_grouped_stats(k, g));
        }
        OtlpStatsBucket {
            bucket: pb::ClientStatsBucket {
                start: self.start,
                duration: bucket_duration,
                stats,
                agent_time_shift: 0,
            },
            exact,
        }
    }
}

/// Create a ClientGroupedStats struct based on the given AggregationKey and GroupedStats
fn encode_grouped_stats(key: OwnedAggregationKey, group: GroupedStats) -> pb::ClientGroupedStats {
    let f = key.fixed;
    pb::ClientGroupedStats {
        service: f.service_name,
        name: f.operation_name,
        resource: f.resource_name,
        http_status_code: f.http_status_code,
        r#type: f.span_type,
        db_type: String::new(), // db_type is not used yet (see proto definition)

        hits: group.hits,
        errors: group.errors,
        duration: group.duration,

        ok_summary: group.ok_summary.encode_to_vec(),
        error_summary: group.error_summary.encode_to_vec(),
        synthetics: f.is_synthetics_request,
        top_level_hits: group.top_level_hits,
        span_kind: f.span_kind,

        peer_tags: key
            .peer_tags
            .into_iter()
            .map(|(k, v)| {
                if v.is_empty() {
                    k.to_string()
                } else {
                    format!("{k}:{v}")
                }
            })
            .collect(),
        is_trace_root: f.is_trace_root.into(),
        http_method: f.http_method,
        http_endpoint: f.http_endpoint,
        grpc_status_code: f
            .grpc_status_code
            .map(|c| c.to_string())
            .unwrap_or_default(),
        service_source: f.service_source,
        span_derived_primary_tags: vec![],
        additional_metric_tags: key
            .additional_metric_tags
            .into_iter()
            .map(|(k, v)| format!("{k}:{v}"))
            .collect(),
    }
}

#[cfg(test)]
mod tests {
    use libdd_trace_utils::span::v04::{SpanBytes, SpanSlice};

    use super::*;
    use libdd_trace_protobuf::pb;
    use std::hash::Hash;

    fn get_hash(v: &impl Hash) -> u64 {
        use std::hash::Hasher;
        let mut hasher = std::hash::DefaultHasher::new();
        v.hash(&mut hasher);
        hasher.finish()
    }

    impl FixedAggregationKey<String> {
        fn into_key(self) -> OwnedAggregationKey {
            OwnedAggregationKey {
                fixed: self,
                peer_tags: vec![],
                additional_metric_tags: vec![],
            }
        }
        fn into_key_with_peers(self, peer_tags: Vec<(String, String)>) -> OwnedAggregationKey {
            OwnedAggregationKey {
                fixed: self,
                peer_tags,
                additional_metric_tags: vec![],
            }
        }
    }

    #[test]
    fn test_float_to_u32_rejects_malformed_values() {
        assert_eq!(float_to_u32(0.0), Some(0));
        assert_eq!(float_to_u32(500.0), Some(500));
        assert_eq!(float_to_u32(u32::MAX as f64), Some(u32::MAX));
        assert_eq!(float_to_u32(f64::NAN), None);
        assert_eq!(float_to_u32(f64::INFINITY), None);
        assert_eq!(float_to_u32(-1.0), None);
        assert_eq!(float_to_u32(1.5), None);
        assert_eq!(float_to_u32((u32::MAX as f64) + 1.0), None);
    }

    #[test]
    fn test_aggregation_key_from_span() {
        let test_cases: Vec<(SpanBytes, OwnedAggregationKey)> = vec![
            // Root span
            (
                SpanBytes {
                    service: "service".into(),
                    name: "op".into(),
                    resource: "res".into(),
                    span_id: 1,
                    parent_id: 0,
                    ..Default::default()
                },
                FixedAggregationKey {
                    service_name: "service".into(),
                    operation_name: "op".into(),
                    resource_name: "res".into(),
                    is_trace_root: pb::Trilean::True,
                    ..Default::default()
                }
                .into_key(),
            ),
            // Datadog-key numeric conversion retains its existing cast behavior.
            (
                SpanBytes {
                    service: "service".into(),
                    name: "op".into(),
                    resource: "res".into(),
                    span_id: 1,
                    parent_id: 0,
                    meta: vec![("http.response.status_code".into(), "418".into())].into(),
                    metrics: vec![("http.status_code".into(), f64::NAN)].into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    service_name: "service".into(),
                    operation_name: "op".into(),
                    resource_name: "res".into(),
                    is_synthetics_request: false,
                    is_trace_root: pb::Trilean::True,
                    http_status_code: 0,
                    ..Default::default()
                }
                .into_key(),
            ),
            // The Datadog key keeps its existing u32 behavior and wins over the OTel key.
            (
                SpanBytes {
                    service: "service".into(),
                    name: "op".into(),
                    resource: "res".into(),
                    span_id: 1,
                    parent_id: 0,
                    meta: vec![("http.response.status_code".into(), "418".into())].into(),
                    metrics: vec![("http.status_code".into(), 600.0)].into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    service_name: "service".into(),
                    operation_name: "op".into(),
                    resource_name: "res".into(),
                    is_synthetics_request: false,
                    is_trace_root: pb::Trilean::True,
                    http_status_code: 600,
                    ..Default::default()
                }
                .into_key(),
            ),
            // Datadog-key numeric conversion keeps the existing Rust cast behavior.
            (
                SpanBytes {
                    service: "service".into(),
                    name: "op".into(),
                    resource: "res".into(),
                    span_id: 1,
                    parent_id: 0,
                    meta: vec![("http.response.status_code".into(), "418".into())].into(),
                    metrics: vec![("http.status_code".into(), 1.5)].into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    service_name: "service".into(),
                    operation_name: "op".into(),
                    resource_name: "res".into(),
                    is_synthetics_request: false,
                    is_trace_root: pb::Trilean::True,
                    http_status_code: 1,
                    ..Default::default()
                }
                .into_key(),
            ),
            // The Datadog key also keeps its existing u32 behavior when stored as metadata.
            (
                SpanBytes {
                    service: "service".into(),
                    name: "op".into(),
                    resource: "res".into(),
                    span_id: 1,
                    parent_id: 0,
                    meta: vec![("http.status_code".into(), "99".into())].into(),
                    metrics: vec![("http.response.status_code".into(), 418.0)].into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    service_name: "service".into(),
                    operation_name: "op".into(),
                    resource_name: "res".into(),
                    is_synthetics_request: false,
                    is_trace_root: pb::Trilean::True,
                    http_status_code: 99,
                    ..Default::default()
                }
                .into_key(),
            ),
            // An out-of-range OTel status is not a valid HTTP response status.
            (
                SpanBytes {
                    service: "service".into(),
                    name: "op".into(),
                    resource: "res".into(),
                    span_id: 1,
                    parent_id: 0,
                    metrics: vec![("http.response.status_code".into(), 600.0)].into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    service_name: "service".into(),
                    operation_name: "op".into(),
                    resource_name: "res".into(),
                    is_synthetics_request: false,
                    is_trace_root: pb::Trilean::True,
                    http_status_code: 0,
                    ..Default::default()
                }
                .into_key(),
            ),
            // Span with span kind
            (
                SpanBytes {
                    service: "service".into(),
                    name: "op".into(),
                    resource: "res".into(),
                    span_id: 1,
                    parent_id: 0,
                    meta: vec![("span.kind".into(), "client".into())].into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    service_name: "service".into(),
                    operation_name: "op".into(),
                    resource_name: "res".into(),
                    span_kind: "client".into(),
                    is_trace_root: pb::Trilean::True,
                    ..Default::default()
                }
                .into_key(),
            ),
            // Span with peer tags but peertags aggregation disabled
            (
                SpanBytes {
                    service: "service".into(),
                    name: "op".into(),
                    resource: "res".into(),
                    span_id: 1,
                    parent_id: 0,
                    meta: vec![
                        ("span.kind".into(), "client".into()),
                        ("aws.s3.bucket".into(), "bucket-a".into()),
                    ]
                    .into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    service_name: "service".into(),
                    operation_name: "op".into(),
                    resource_name: "res".into(),
                    span_kind: "client".into(),
                    is_trace_root: pb::Trilean::True,
                    ..Default::default()
                }
                .into_key(),
            ),
            // Span with multiple peer tags but peertags aggregation disabled
            (
                SpanBytes {
                    service: "service".into(),
                    name: "op".into(),
                    resource: "res".into(),
                    span_id: 1,
                    parent_id: 0,
                    meta: vec![
                        ("span.kind".into(), "producer".into()),
                        ("aws.s3.bucket".into(), "bucket-a".into()),
                        ("db.instance".into(), "dynamo.test.us1".into()),
                        ("db.system".into(), "dynamodb".into()),
                    ]
                    .into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    service_name: "service".into(),
                    operation_name: "op".into(),
                    resource_name: "res".into(),
                    span_kind: "producer".into(),
                    is_trace_root: pb::Trilean::True,
                    ..Default::default()
                }
                .into_key(),
            ),
            // Span with multiple peer tags but peertags aggregation disabled and span kind is
            // server
            (
                SpanBytes {
                    service: "service".into(),
                    name: "op".into(),
                    resource: "res".into(),
                    span_id: 1,
                    parent_id: 0,
                    meta: vec![
                        ("span.kind".into(), "server".into()),
                        ("aws.s3.bucket".into(), "bucket-a".into()),
                        ("db.instance".into(), "dynamo.test.us1".into()),
                        ("db.system".into(), "dynamodb".into()),
                    ]
                    .into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    service_name: "service".into(),
                    operation_name: "op".into(),
                    resource_name: "res".into(),
                    span_kind: "server".into(),
                    is_trace_root: pb::Trilean::True,
                    ..Default::default()
                }
                .into_key(),
            ),
            // Span from synthetics
            (
                SpanBytes {
                    service: "service".into(),
                    name: "op".into(),
                    resource: "res".into(),
                    span_id: 1,
                    parent_id: 0,
                    meta: vec![("_dd.origin".into(), "synthetics-browser".into())].into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    service_name: "service".into(),
                    operation_name: "op".into(),
                    resource_name: "res".into(),
                    is_synthetics_request: true,
                    is_trace_root: pb::Trilean::True,
                    ..Default::default()
                }
                .into_key(),
            ),
            // Span with status code in meta
            (
                SpanBytes {
                    service: "service".into(),
                    name: "op".into(),
                    resource: "res".into(),
                    span_id: 1,
                    parent_id: 0,
                    meta: vec![("http.status_code".into(), "418".into())].into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    service_name: "service".into(),
                    operation_name: "op".into(),
                    resource_name: "res".into(),
                    is_synthetics_request: false,
                    is_trace_root: pb::Trilean::True,
                    http_status_code: 418,
                    ..Default::default()
                }
                .into_key(),
            ),
            // Span with invalid status code in meta
            (
                SpanBytes {
                    service: "service".into(),
                    name: "op".into(),
                    resource: "res".into(),
                    span_id: 1,
                    parent_id: 0,
                    meta: vec![("http.status_code".into(), "x".into())].into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    service_name: "service".into(),
                    operation_name: "op".into(),
                    resource_name: "res".into(),
                    is_synthetics_request: false,
                    is_trace_root: pb::Trilean::True,
                    ..Default::default()
                }
                .into_key(),
            ),
            // Span with status code in metrics
            (
                SpanBytes {
                    service: "service".into(),
                    name: "op".into(),
                    resource: "res".into(),
                    span_id: 1,
                    parent_id: 0,
                    metrics: vec![("http.status_code".into(), 418.0)].into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    service_name: "service".into(),
                    operation_name: "op".into(),
                    resource_name: "res".into(),
                    is_synthetics_request: false,
                    is_trace_root: pb::Trilean::True,
                    http_status_code: 418,
                    ..Default::default()
                }
                .into_key(),
            ),
            // Span in OTel semantics mode: status code in metrics under the OTel name, which is
            // where the int-typed OTel attribute lands
            (
                SpanBytes {
                    service: "service".into(),
                    name: "op".into(),
                    resource: "res".into(),
                    span_id: 1,
                    parent_id: 0,
                    metrics: vec![("http.response.status_code".into(), 418.0)].into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    service_name: "service".into(),
                    operation_name: "op".into(),
                    resource_name: "res".into(),
                    is_synthetics_request: false,
                    is_trace_root: pb::Trilean::True,
                    http_status_code: 418,
                    ..Default::default()
                }
                .into_key(),
            ),
            // Span in OTel semantics mode: status code in meta under the OTel name, which is
            // where it lands whenever the tracer stringifies its attributes
            (
                SpanBytes {
                    service: "service".into(),
                    name: "op".into(),
                    resource: "res".into(),
                    span_id: 1,
                    parent_id: 0,
                    meta: vec![("http.response.status_code".into(), "418".into())].into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    service_name: "service".into(),
                    operation_name: "op".into(),
                    resource_name: "res".into(),
                    is_synthetics_request: false,
                    is_trace_root: pb::Trilean::True,
                    http_status_code: 418,
                    ..Default::default()
                }
                .into_key(),
            ),
            // Datadog name in meta, OTel name in metrics: the Datadog name still wins, so the
            // metrics-before-meta order inside one name never leaks across names
            (
                SpanBytes {
                    service: "service".into(),
                    name: "op".into(),
                    resource: "res".into(),
                    span_id: 1,
                    parent_id: 0,
                    meta: vec![("http.status_code".into(), "500".into())].into(),
                    metrics: vec![("http.response.status_code".into(), 418.0)].into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    service_name: "service".into(),
                    operation_name: "op".into(),
                    resource_name: "res".into(),
                    is_synthetics_request: false,
                    is_trace_root: pb::Trilean::True,
                    http_status_code: 500,
                    ..Default::default()
                }
                .into_key(),
            ),
            // An unparseable Datadog status retains its existing parse-or-zero behavior.
            (
                SpanBytes {
                    service: "service".into(),
                    name: "op".into(),
                    resource: "res".into(),
                    span_id: 1,
                    parent_id: 0,
                    meta: vec![
                        ("http.status_code".into(), "x".into()),
                        ("http.response.status_code".into(), "418".into()),
                    ]
                    .into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    service_name: "service".into(),
                    operation_name: "op".into(),
                    resource_name: "res".into(),
                    is_synthetics_request: false,
                    is_trace_root: pb::Trilean::True,
                    http_status_code: 0,
                    ..Default::default()
                }
                .into_key(),
            ),
            // Datadog metrics retain their existing precedence and cast behavior.
            (
                SpanBytes {
                    service: "service".into(),
                    name: "op".into(),
                    resource: "res".into(),
                    span_id: 1,
                    parent_id: 0,
                    meta: vec![("http.status_code".into(), "500".into())].into(),
                    metrics: vec![("http.status_code".into(), f64::NAN)].into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    service_name: "service".into(),
                    operation_name: "op".into(),
                    resource_name: "res".into(),
                    is_synthetics_request: false,
                    is_trace_root: pb::Trilean::True,
                    http_status_code: 0,
                    ..Default::default()
                }
                .into_key(),
            ),
            // An empty Datadog method counts as absent, so it cannot shadow the OTel name
            (
                SpanBytes {
                    service: "service".into(),
                    name: "op".into(),
                    resource: "res".into(),
                    span_id: 1,
                    parent_id: 0,
                    meta: vec![
                        ("http.method".into(), "".into()),
                        ("http.request.method".into(), "GET".into()),
                    ]
                    .into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    service_name: "service".into(),
                    operation_name: "op".into(),
                    resource_name: "res".into(),
                    is_synthetics_request: false,
                    is_trace_root: pb::Trilean::True,
                    http_method: "GET".into(),
                    ..Default::default()
                }
                .into_key(),
            ),
            // Both names present: the canonical Datadog names win.
            (
                SpanBytes {
                    service: "service".into(),
                    name: "op".into(),
                    resource: "res".into(),
                    span_id: 1,
                    parent_id: 0,
                    meta: vec![
                        ("http.status_code".into(), "500".into()),
                        ("http.response.status_code".into(), "418".into()),
                        ("http.method".into(), "GET".into()),
                        ("http.request.method".into(), "POST".into()),
                    ]
                    .into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    service_name: "service".into(),
                    operation_name: "op".into(),
                    resource_name: "res".into(),
                    is_synthetics_request: false,
                    is_trace_root: pb::Trilean::True,
                    http_status_code: 500,
                    http_method: "GET".into(),
                    ..Default::default()
                }
                .into_key(),
            ),
            // Span in OTel semantics mode: method under the OTel name, route unchanged because
            // http.route is already the OTel name
            (
                SpanBytes {
                    service: "service".into(),
                    name: "op".into(),
                    resource: "GET /api/v1/users".into(),
                    span_id: 1,
                    parent_id: 0,
                    meta: vec![
                        ("http.request.method".into(), "GET".into()),
                        ("http.route".into(), "/api/v1/users".into()),
                    ]
                    .into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    service_name: "service".into(),
                    operation_name: "op".into(),
                    resource_name: "GET /api/v1/users".into(),
                    is_synthetics_request: false,
                    is_trace_root: pb::Trilean::True,
                    http_method: "GET".into(),
                    http_endpoint: "/api/v1/users".into(),
                    ..Default::default()
                }
                .into_key(),
            ),
            // Span with http.method and http.route
            (
                SpanBytes {
                    service: "service".into(),
                    name: "op".into(),
                    resource: "GET /api/v1/users".into(),
                    span_id: 1,
                    parent_id: 0,
                    meta: vec![
                        ("http.method".into(), "GET".into()),
                        ("http.route".into(), "/api/v1/users".into()),
                    ]
                    .into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    service_name: "service".into(),
                    operation_name: "op".into(),
                    resource_name: "GET /api/v1/users".into(),
                    http_method: "GET".into(),
                    http_endpoint: "/api/v1/users".into(),
                    is_synthetics_request: false,
                    is_trace_root: pb::Trilean::True,
                    ..Default::default()
                }
                .into_key(),
            ),
            // The canonical OTel route takes precedence over the Datadog-only endpoint.
            (
                SpanBytes {
                    service: "service".into(),
                    name: "op".into(),
                    resource: "POST /users/create".into(),
                    span_id: 1,
                    parent_id: 0,
                    meta: vec![
                        ("http.method".into(), "POST".into()),
                        ("http.route".into(), "/users/create".into()),
                        ("http.endpoint".into(), "/users/create2".into()),
                    ]
                    .into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    service_name: "service".into(),
                    operation_name: "op".into(),
                    resource_name: "POST /users/create".into(),
                    http_method: "POST".into(),
                    http_endpoint: "/users/create".into(),
                    is_synthetics_request: false,
                    is_trace_root: pb::Trilean::True,
                    ..Default::default()
                }
                .into_key(),
            ),
            // An empty route falls back to the retained Datadog endpoint.
            (
                SpanBytes {
                    service: "service".into(),
                    name: "op".into(),
                    resource: "POST /users/create".into(),
                    span_id: 1,
                    parent_id: 0,
                    meta: vec![
                        ("http.method".into(), "POST".into()),
                        ("http.route".into(), "".into()),
                        ("http.endpoint".into(), "/users/create2".into()),
                    ]
                    .into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    service_name: "service".into(),
                    operation_name: "op".into(),
                    resource_name: "POST /users/create".into(),
                    http_method: "POST".into(),
                    http_endpoint: "/users/create2".into(),
                    is_synthetics_request: false,
                    is_trace_root: pb::Trilean::True,
                    ..Default::default()
                }
                .into_key(),
            ),
            // Span with grpc status from meta as named string
            (
                SpanBytes {
                    meta: vec![("rpc.grpc.status_code".into(), "OK".into())].into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    grpc_status_code: Some(0),
                    is_trace_root: pb::Trilean::True,
                    ..Default::default()
                }
                .into_key(),
            ),
            // grpc.method.name is carried in GroupedStats (for OTLP), not in the aggregation key.
            (
                SpanBytes {
                    meta: vec![("grpc.method.name".into(), "/pkg.Svc/Method".into())].into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    is_trace_root: pb::Trilean::True,
                    ..Default::default()
                }
                .into_key(),
            ),
            // Span with grpc status from meta as numeric string
            (
                SpanBytes {
                    meta: vec![("rpc.grpc.status_code".into(), "14".into())].into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    grpc_status_code: Some(14),
                    is_trace_root: pb::Trilean::True,
                    ..Default::default()
                }
                .into_key(),
            ),
            // Span with grpc status from meta with StatusCode. prefix
            (
                SpanBytes {
                    meta: vec![("grpc.code".into(), "StatusCode.UNAVAILABLE".into())].into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    grpc_status_code: Some(14),
                    is_trace_root: pb::Trilean::True,
                    ..Default::default()
                }
                .into_key(),
            ),
            // Span with grpc status from metrics takes precedence over meta
            (
                SpanBytes {
                    meta: vec![("rpc.grpc.status_code".into(), "PERMISSION_DENIED".into())].into(),
                    metrics: vec![("rpc.grpc.status_code".into(), 2.0)].into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    grpc_status_code: Some(7),
                    is_trace_root: pb::Trilean::True,
                    ..Default::default()
                }
                .into_key(),
            ),
            // Span with grpc status from metrics via secondary key
            (
                SpanBytes {
                    metrics: vec![("grpc.code".into(), 3.0)].into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    grpc_status_code: Some(3),
                    is_trace_root: pb::Trilean::True,
                    ..Default::default()
                }
                .into_key(),
            ),
            // Span with invalid grpc status string
            (
                SpanBytes {
                    meta: vec![("rpc.grpc.status_code".into(), "NOPE".into())].into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    is_trace_root: pb::Trilean::True,
                    ..Default::default()
                }
                .into_key(),
            ),
            // Span with service source set by integration
            (
                SpanBytes {
                    service: "my-service".into(),
                    name: "op".into(),
                    resource: "res".into(),
                    span_id: 1,
                    parent_id: 0,
                    meta: vec![("_dd.svc_src".into(), "redis".into())].into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    service_name: "my-service".into(),
                    operation_name: "op".into(),
                    resource_name: "res".into(),
                    is_trace_root: pb::Trilean::True,
                    service_source: "redis".into(),
                    ..Default::default()
                }
                .into_key(),
            ),
            // Span with service source set by configuration option
            (
                SpanBytes {
                    service: "my-service".into(),
                    name: "op".into(),
                    resource: "res".into(),
                    span_id: 1,
                    parent_id: 0,
                    meta: vec![("_dd.svc_src".into(), "opt.split_by_tag".into())].into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    service_name: "my-service".into(),
                    operation_name: "op".into(),
                    resource_name: "res".into(),
                    is_trace_root: pb::Trilean::True,
                    service_source: "opt.split_by_tag".into(),
                    ..Default::default()
                }
                .into_key(),
            ),
            // Span without service source (default service name)
            (
                SpanBytes {
                    service: "my-service".into(),
                    name: "op".into(),
                    resource: "res".into(),
                    span_id: 1,
                    parent_id: 0,
                    ..Default::default()
                },
                FixedAggregationKey {
                    service_name: "my-service".into(),
                    operation_name: "op".into(),
                    resource_name: "res".into(),
                    is_trace_root: pb::Trilean::True,
                    service_source: "".into(),
                    ..Default::default()
                }
                .into_key(),
            ),
        ];

        let test_peer_tags = vec![
            "aws.s3.bucket".to_string(),
            "db.instance".to_string(),
            "db.system".to_string(),
        ];

        let test_cases_with_peer_tags: Vec<(SpanSlice, OwnedAggregationKey)> = vec![
            // Span with peer tags with peertags aggregation enabled
            (
                SpanSlice {
                    service: Cow::Borrowed("service"),
                    name: Cow::Borrowed("op"),
                    resource: Cow::Borrowed("res"),
                    span_id: 1,
                    parent_id: 0,
                    meta: vec![
                        (Cow::Borrowed("span.kind"), Cow::Borrowed("client")),
                        (Cow::Borrowed("aws.s3.bucket"), Cow::Borrowed("bucket-a")),
                    ]
                    .into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    service_name: "service".into(),
                    operation_name: "op".into(),
                    resource_name: "res".into(),
                    span_kind: "client".into(),
                    is_trace_root: pb::Trilean::True,
                    ..Default::default()
                }
                .into_key_with_peers(vec![("aws.s3.bucket".into(), "bucket-a".into())]),
            ),
            // Span with multiple peer tags with peertags aggregation enabled
            (
                SpanSlice {
                    service: Cow::Borrowed("service"),
                    name: Cow::Borrowed("op"),
                    resource: Cow::Borrowed("res"),
                    span_id: 1,
                    parent_id: 0,
                    meta: vec![
                        (Cow::Borrowed("span.kind"), Cow::Borrowed("producer")),
                        (Cow::Borrowed("aws.s3.bucket"), Cow::Borrowed("bucket-a")),
                        (
                            Cow::Borrowed("db.instance"),
                            Cow::Borrowed("dynamo.test.us1"),
                        ),
                        (Cow::Borrowed("db.system"), Cow::Borrowed("dynamodb")),
                    ]
                    .into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    service_name: "service".into(),
                    operation_name: "op".into(),
                    resource_name: "res".into(),
                    span_kind: "producer".into(),
                    is_trace_root: pb::Trilean::True,
                    ..Default::default()
                }
                .into_key_with_peers(vec![
                    ("aws.s3.bucket".into(), "bucket-a".into()),
                    ("db.instance".into(), "dynamo.test.us1".into()),
                    ("db.system".into(), "dynamodb".into()),
                ]),
            ),
            // Span with multiple peer tags with peertags aggregation enabled and span kind is
            // server
            (
                SpanSlice {
                    service: Cow::Borrowed("service"),
                    name: Cow::Borrowed("op"),
                    resource: Cow::Borrowed("res"),
                    span_id: 1,
                    parent_id: 0,
                    meta: vec![
                        (Cow::Borrowed("span.kind"), Cow::Borrowed("server")),
                        (Cow::Borrowed("aws.s3.bucket"), Cow::Borrowed("bucket-a")),
                        (
                            Cow::Borrowed("db.instance"),
                            Cow::Borrowed("dynamo.test.us1"),
                        ),
                        (Cow::Borrowed("db.system"), Cow::Borrowed("dynamodb")),
                    ]
                    .into(),
                    ..Default::default()
                },
                FixedAggregationKey {
                    service_name: "service".into(),
                    operation_name: "op".into(),
                    resource_name: "res".into(),
                    span_kind: "server".into(),
                    is_trace_root: pb::Trilean::True,
                    ..Default::default()
                }
                .into_key(),
            ),
        ];

        for (span, expected_key) in test_cases {
            let borrowed_key = BorrowedAggregationKey::from_span(&span, &[], &[]);
            assert_eq!(
                OwnedAggregationKey::from(&borrowed_key),
                expected_key,
                "for span {span:?}"
            );
            assert_eq!(
                get_hash(&borrowed_key),
                get_hash(&OwnedAggregationKey::from(&borrowed_key))
            );
        }

        for (span, expected_key) in test_cases_with_peer_tags {
            let borrowed_key =
                BorrowedAggregationKey::from_span(&span, test_peer_tags.as_slice(), &[]);
            assert_eq!(OwnedAggregationKey::from(&borrowed_key), expected_key);
            assert_eq!(
                get_hash(&borrowed_key),
                get_hash(&OwnedAggregationKey::from(&borrowed_key))
            );
        }
    }

    #[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: Cow::Borrowed("service"),
            name: Cow::Borrowed("op"),
            resource: Cow::Borrowed("res"),
            span_id: 1,
            parent_id: 0,
            meta: vec![
                (Cow::Borrowed("span.kind"), Cow::Borrowed("client")),
                (Cow::Borrowed("peer.hostname"), Cow::Borrowed("10.1.2.3")),
                (Cow::Borrowed("db.instance"), Cow::Borrowed("my-db")),
            ]
            .into(),
            ..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: Cow::Borrowed("service"),
            name: Cow::Borrowed("op"),
            resource: Cow::Borrowed("res"),
            span_id: 1,
            parent_id: 0,
            meta: vec![
                (Cow::Borrowed("span.kind"), Cow::Borrowed("client")),
                (
                    Cow::Borrowed("peer.hostname"),
                    Cow::Borrowed("2001:db8:3333:4444:CCCC:DDDD:EEEE:FFFF"),
                ),
            ]
            .into(),
            ..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: Cow::Borrowed("service"),
            name: Cow::Borrowed("op"),
            resource: Cow::Borrowed("res"),
            span_id: 1,
            parent_id: 0,
            meta: vec![
                (Cow::Borrowed("span.kind"), Cow::Borrowed("client")),
                (
                    Cow::Borrowed("db.instance"),
                    Cow::Borrowed("dynamo.test.us1"),
                ),
            ]
            .into(),
            ..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
        assert_eq!(grpc_status_str_to_int_value("0"), Some(0));
        assert_eq!(grpc_status_str_to_int_value("14"), Some(14));
        assert_eq!(grpc_status_str_to_int_value("255"), Some(255));
        assert_eq!(grpc_status_str_to_int_value("256"), None);
        assert_eq!(grpc_status_str_to_int_value("-1"), None);

        // Named status codes (uppercase)
        assert_eq!(grpc_status_str_to_int_value("OK"), Some(0));
        assert_eq!(grpc_status_str_to_int_value("CANCELLED"), Some(1));
        assert_eq!(grpc_status_str_to_int_value("UNKNOWN"), Some(2));
        assert_eq!(grpc_status_str_to_int_value("INVALID_ARGUMENT"), Some(3));
        assert_eq!(grpc_status_str_to_int_value("DEADLINE_EXCEEDED"), Some(4));
        assert_eq!(grpc_status_str_to_int_value("NOT_FOUND"), Some(5));
        assert_eq!(grpc_status_str_to_int_value("ALREADY_EXISTS"), Some(6));
        assert_eq!(grpc_status_str_to_int_value("PERMISSION_DENIED"), Some(7));
        assert_eq!(grpc_status_str_to_int_value("UNAUTHENTICATED"), Some(16));
        assert_eq!(grpc_status_str_to_int_value("RESOURCE_EXHAUSTED"), Some(8));
        assert_eq!(grpc_status_str_to_int_value("FAILED_PRECONDITION"), Some(9));
        assert_eq!(grpc_status_str_to_int_value("ABORTED"), Some(10));
        assert_eq!(grpc_status_str_to_int_value("OUT_OF_RANGE"), Some(11));
        assert_eq!(grpc_status_str_to_int_value("UNIMPLEMENTED"), Some(12));
        assert_eq!(grpc_status_str_to_int_value("INTERNAL"), Some(13));
        assert_eq!(grpc_status_str_to_int_value("UNAVAILABLE"), Some(14));
        assert_eq!(grpc_status_str_to_int_value("DATA_LOSS"), Some(15));

        // Case-insensitive matching
        assert_eq!(grpc_status_str_to_int_value("ok"), Some(0));
        assert_eq!(grpc_status_str_to_int_value("Cancelled"), Some(1));
        assert_eq!(grpc_status_str_to_int_value("not_found"), Some(5));

        // StatusCode. prefix is stripped
        assert_eq!(grpc_status_str_to_int_value("StatusCode.OK"), Some(0));
        assert_eq!(
            grpc_status_str_to_int_value("StatusCode.UNAVAILABLE"),
            Some(14)
        );
        assert_eq!(
            grpc_status_str_to_int_value("StatusCode.not_found"),
            Some(5)
        );

        // Alternate spellings
        assert_eq!(grpc_status_str_to_int_value("CANCELED"), Some(1));

        // Unknown / empty strings
        assert_eq!(grpc_status_str_to_int_value("NOPE"), None);
        assert_eq!(grpc_status_str_to_int_value(""), None);
        assert_eq!(
            grpc_status_str_to_int_value("this_is_a_kinda_long_string_that_needs_upcasing"),
            None
        );

        // Non ascii
        assert_eq!(grpc_status_str_to_int_value("🤣"), None);
    }
}