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