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

libdd_trace_stats/span_concentrator/
aggregation.rs
