From 60cc3545d62cebfb45c8b58dbcf6985513523443 Mon Sep 17 00:00:00 2001 From: Duncan Harvey Date: Wed, 24 Jun 2026 12:16:36 -0400 Subject: [PATCH 1/2] implement additional_metric_tags in libdd-trace-stats --- .../src/trace_exporter/builder.rs | 1 + .../src/trace_exporter/stats.rs | 1 + libdd-trace-protobuf/src/pb_test.rs | 3 +- libdd-trace-stats/README.md | 1 + .../benches/span_concentrator_bench.rs | 1 + .../src/span_concentrator/aggregation.rs | 210 ++++- .../src/span_concentrator/mod.rs | 77 +- .../src/span_concentrator/tests.rs | 736 ++++++++++++++++++ libdd-trace-stats/src/stats_exporter.rs | 2 + libdd-trace-utils/src/stats_utils.rs | 1 + 10 files changed, 1005 insertions(+), 28 deletions(-) diff --git a/libdd-data-pipeline/src/trace_exporter/builder.rs b/libdd-data-pipeline/src/trace_exporter/builder.rs index 5582c934cd..5496203551 100644 --- a/libdd-data-pipeline/src/trace_exporter/builder.rs +++ b/libdd-data-pipeline/src/trace_exporter/builder.rs @@ -750,6 +750,7 @@ impl TraceExporterBuilder { span_kinds, self.peer_tags.clone(), None, + vec![], #[cfg(feature = "stats-obfuscation")] None, ))); diff --git a/libdd-data-pipeline/src/trace_exporter/stats.rs b/libdd-data-pipeline/src/trace_exporter/stats.rs index 2fdaa29728..15a9c9f9bf 100644 --- a/libdd-data-pipeline/src/trace_exporter/stats.rs +++ b/libdd-data-pipeline/src/trace_exporter/stats.rs @@ -143,6 +143,7 @@ pub(crate) fn start_stats_computation< span_kinds, peer_tags, ctx.stats_cardinality_limit, + vec![], #[cfg(feature = "stats-obfuscation")] Some(client_side_stats.obfuscation_config.clone()), ))); diff --git a/libdd-trace-protobuf/src/pb_test.rs b/libdd-trace-protobuf/src/pb_test.rs index d474cdd571..27b5d84dda 100644 --- a/libdd-trace-protobuf/src/pb_test.rs +++ b/libdd-trace-protobuf/src/pb_test.rs @@ -72,7 +72,8 @@ mod tests { ], "HTTPMethod": "GET", "HTTPEndpoint": "/test", - "GRPCStatusCode": "0" + "GRPCStatusCode": "0", + "AdditionalMetricTags": [] } ] } diff --git a/libdd-trace-stats/README.md b/libdd-trace-stats/README.md index 789b35f0e0..ab87d112fd 100644 --- a/libdd-trace-stats/README.md +++ b/libdd-trace-stats/README.md @@ -76,6 +76,7 @@ let mut concentrator = SpanConcentrator::new( SystemTime::now(), vec!["client".to_string(), "server".to_string()], // eligible span kinds vec!["peer.service".to_string()], // peer tag keys + vec!["example.key".to_string()], // additional metric tag keys ); // Add spans diff --git a/libdd-trace-stats/benches/span_concentrator_bench.rs b/libdd-trace-stats/benches/span_concentrator_bench.rs index 1e162cb4a2..386ebcb207 100644 --- a/libdd-trace-stats/benches/span_concentrator_bench.rs +++ b/libdd-trace-stats/benches/span_concentrator_bench.rs @@ -45,6 +45,7 @@ pub fn criterion_benchmark(c: &mut Criterion) { vec![], vec!["db_name".into(), "bucket_s3".into()], None, + vec![], #[cfg(feature = "stats-obfuscation")] None, ); diff --git a/libdd-trace-stats/src/span_concentrator/aggregation.rs b/libdd-trace-stats/src/span_concentrator/aggregation.rs index 5d17d54056..e638c87122 100644 --- a/libdd-trace-stats/src/span_concentrator/aggregation.rs +++ b/libdd-trace-stats/src/span_concentrator/aggregation.rs @@ -10,6 +10,7 @@ use libdd_trace_obfuscation::ip_address::quantize_peer_ip_addresses; use libdd_trace_protobuf::pb; use libdd_trace_utils::span::SpanText; use std::borrow::{Borrow, Cow}; +use tracing::warn; use crate::span_concentrator::StatSpan; @@ -17,6 +18,7 @@ use crate::span_concentrator::StatSpan; pub const TRACER_BLOCKED_VALUE: &str = "tracer_blocked_value"; const TAG_STATUS_CODE: &str = "http.status_code"; +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"; @@ -84,6 +86,7 @@ impl FixedAggregationKey { 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 for BorrowedAggregationKey<'_> { @@ -96,6 +99,12 @@ impl hashbrown::Equivalent for BorrowedAggregationKey<'_> { .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) } } @@ -110,6 +119,7 @@ impl hashbrown::Equivalent for BorrowedAggregationKey<'_> { pub(super) struct OwnedAggregationKey { fixed: FixedAggregationKey, peer_tags: Vec<(String, String)>, + additional_metric_tags: Vec<(String, String)>, } impl From<&BorrowedAggregationKey<'_>> for OwnedAggregationKey { @@ -121,6 +131,11 @@ impl From<&BorrowedAggregationKey<'_>> for OwnedAggregationKey { .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(), } } } @@ -203,16 +218,28 @@ fn grpc_status_str_to_int_value(v: &str) -> Option { impl<'a> BorrowedAggregationKey<'a> { /// Return an AggregationKey matching the given span. /// - /// If `peer_tags_keys` is not empty then the peer tags of the span will be included in the + /// If `peer_tag_keys` is not empty then the peer tags of the span will be included in the /// key. - pub(super) fn from_span>(span: &'a T, peer_tag_keys: &'a [String]) -> Self { - Self::from_obfuscated_span(span.resource(), span, peer_tag_keys) + /// If `additional_metric_tags` is not empty then matching span tags keys are included in the + /// key. + pub(super) fn from_span>( + 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>, @@ -255,6 +282,24 @@ impl<'a> BorrowedAggregationKey<'a> { 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() => { + if v.len() > ADDITIONAL_METRIC_TAG_VALUE_MAX_LEN { + 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, @@ -277,6 +322,26 @@ impl<'a> BorrowedAggregationKey<'a> { }, }, peer_tags, + additional_metric_tags, + } + } + + /// Return an owned copy of this key with all additional metric tag values replaced by + /// `TRACER_BLOCKED_VALUE`. Used when the per-bucket additional-metric-tags cardinality limit + /// is exceeded. + pub(super) fn into_masked_owned(self) -> OwnedAggregationKey { + OwnedAggregationKey { + fixed: self.fixed.convert(str::to_owned), + peer_tags: self + .peer_tags + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect(), + additional_metric_tags: self + .additional_metric_tags + .iter() + .map(|(k, _)| (k.to_string(), TRACER_BLOCKED_VALUE.to_string())) + .collect(), } } } @@ -300,6 +365,7 @@ impl OwnedAggregationKey { is_trace_root: pb::Trilean::NotSet, }, peer_tags: vec![], + additional_metric_tags: vec![], } } } @@ -330,6 +396,14 @@ impl From for OwnedAggregationKey { 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(), } } } @@ -431,6 +505,10 @@ pub(super) struct StatsBucket { max_entries: usize, /// Number of spans collapsed into the overflow bucket due to cardinality limiting. collapsed_count: u64, + /// Number of distinct entries with additional metric tags admitted this bucket. + additional_metric_tags_entry_count: usize, + /// Maximum distinct entries with additional metric tags per bucket. + additional_metric_tags_max_entries: usize, } impl StatsBucket { @@ -438,12 +516,22 @@ impl StatsBucket { /// /// `max_entries` is the maximum number of distinct aggregation keys the bucket will hold. /// Once the limit is reached, new distinct keys are collapsed into the overflow sentinel key. - pub(super) fn new(start_timestamp: u64, max_entries: usize) -> Self { + /// `additional_metric_tags_max_entries` is the maximum number of distinct aggregation keys + /// with additional metric tags the bucket will hold. Once the limit is reached, new distinct + /// keys have their additional metric tag values masked to `TRACER_BLOCKED_VALUE` before being + /// subject to the `max_entries` check. + pub(super) fn new( + start_timestamp: u64, + max_entries: usize, + additional_metric_tags_max_entries: usize, + ) -> Self { Self { data: HashMap::new(), start: start_timestamp, max_entries, collapsed_count: 0, + additional_metric_tags_entry_count: 0, + additional_metric_tags_max_entries, } } @@ -452,9 +540,14 @@ impl StatsBucket { self.collapsed_count } - /// Insert a value as stats in the group corresponding to the aggregation key. If the key is new - /// and the `max_entries` limit has not been reached, a new entry is created, else the span is - /// instead merged into the overflow sentinel key. + /// 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 that carries + /// additional metric tags and would exceed the `additional_metric_tags_max_entries` has its + /// additional tag values masked to `TRACER_BLOCKED_VALUE` before insertion. Any new key, + /// masked or otherwise, is then subject to the `max_entries` limit, which collapses it into + /// the overflow sentinel key. pub(super) fn insert( &mut self, key: BorrowedAggregationKey<'_>, @@ -462,18 +555,80 @@ impl StatsBucket { is_error: bool, is_top_level: bool, ) { - if self.data.len() >= self.max_entries && !self.data.contains_key(&key) { - self.collapsed_count += 1; - self.data - .entry(OwnedAggregationKey::overflow_key()) - .or_default() - .insert(duration, is_error, is_top_level); - return; + let has_additional_tags = !key.additional_metric_tags.is_empty(); + // The map can't change size before the entry below is resolved, so this single read + // covers the `max_entries` check in either 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 additional-metric-tags max entry limit, mask its tag values and + // re-resolve under the possibly different masked identity. + if has_additional_tags + && self.additional_metric_tags_entry_count + >= self.additional_metric_tags_max_entries + { + let masked = key.into_masked_owned(); + self.insert_masked(masked, len_before_insert, duration, is_error, is_top_level); + return; + } + // New key over the max entry limit, collapse into the overflow + // sentinel. + if len_before_insert >= self.max_entries { + self.collapsed_count += 1; + self.data + .entry(OwnedAggregationKey::overflow_key()) + .or_default() + .insert(duration, is_error, is_top_level); + return; + } + // Within the max entry and additional-metric-tag limits, admit key as a new + // distinct entry. + if has_additional_tags { + self.additional_metric_tags_entry_count += 1; + } + e.insert(GroupedStats::default()) + .insert(duration, is_error, is_top_level); + } + } + } + + /// Insert an already masked owned key produced when the additional-metric-tags limit was + /// exceeded. The key identity changed from the original, so it needs its own lookup rather than + /// reusing the caller's entry. + fn insert_masked( + &mut self, + key: OwnedAggregationKey, + len_before_insert: usize, + duration: i64, + is_error: bool, + is_top_level: bool, + ) { + match self.data.entry(key) { + // Existing key, merge + hashbrown::hash_map::Entry::Occupied(mut e) => { + e.get_mut().insert(duration, is_error, is_top_level); + } + hashbrown::hash_map::Entry::Vacant(e) => { + // New masked key over the max entry limit, collapse into the + // overflow sentinel. + if len_before_insert >= self.max_entries { + 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 the masked key as a new distinct entry. + e.insert(GroupedStats::default()) + .insert(duration, is_error, is_top_level); + } } - self.data - .entry_ref(&key) - .or_default() - .insert(duration, is_error, is_top_level); } /// Consume the bucket and return a ClientStatsBucket containing the bucket stats. @@ -551,7 +706,11 @@ fn encode_grouped_stats(key: OwnedAggregationKey, group: GroupedStats) -> pb::Cl .unwrap_or_default(), service_source: f.service_source, span_derived_primary_tags: vec![], - additional_metric_tags: vec![], + additional_metric_tags: key + .additional_metric_tags + .into_iter() + .map(|(k, v)| format!("{k}:{v}")) + .collect(), } } @@ -575,12 +734,14 @@ mod tests { 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![], } } } @@ -1093,7 +1254,7 @@ mod tests { ]; for (span, expected_key) in test_cases { - let borrowed_key = BorrowedAggregationKey::from_span(&span, &[]); + let borrowed_key = BorrowedAggregationKey::from_span(&span, &[], &[]); assert_eq!( OwnedAggregationKey::from(&borrowed_key), expected_key, @@ -1106,7 +1267,8 @@ mod tests { } for (span, expected_key) in test_cases_with_peer_tags { - let borrowed_key = BorrowedAggregationKey::from_span(&span, test_peer_tags.as_slice()); + 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), @@ -1134,7 +1296,7 @@ mod tests { .into(), ..Default::default() }; - let key = BorrowedAggregationKey::from_span(&span_ipv4, &peer_tag_keys); + let key = BorrowedAggregationKey::from_span(&span_ipv4, &peer_tag_keys, &[]); let owned = OwnedAggregationKey::from(&key); assert_eq!( owned.peer_tags, @@ -1162,7 +1324,7 @@ mod tests { ..Default::default() }; let ipv6_keys = vec!["peer.hostname".to_string()]; - let key = BorrowedAggregationKey::from_span(&span_ipv6, &ipv6_keys); + let key = BorrowedAggregationKey::from_span(&span_ipv6, &ipv6_keys, &[]); let owned = OwnedAggregationKey::from(&key); assert_eq!( owned.peer_tags, @@ -1183,7 +1345,7 @@ mod tests { ..Default::default() }; let non_ip_keys = vec!["db.instance".to_string()]; - let key = BorrowedAggregationKey::from_span(&span_non_ip, &non_ip_keys); + let key = BorrowedAggregationKey::from_span(&span_non_ip, &non_ip_keys, &[]); let owned = OwnedAggregationKey::from(&key); assert_eq!( owned.peer_tags, diff --git a/libdd-trace-stats/src/span_concentrator/mod.rs b/libdd-trace-stats/src/span_concentrator/mod.rs index f1beb5eab9..2de888c60f 100644 --- a/libdd-trace-stats/src/span_concentrator/mod.rs +++ b/libdd-trace-stats/src/span_concentrator/mod.rs @@ -6,6 +6,7 @@ use std::time::{self, Duration, SystemTime}; use tracing::debug; use libdd_trace_protobuf::pb; +use tracing::warn; use aggregation::StatsBucket; @@ -16,6 +17,26 @@ pub use aggregation::{FixedAggregationKey, OtlpExactCell, OtlpExactGroup, OtlpSt pub mod stat_span; pub use stat_span::StatSpan; +const ADDITIONAL_METRIC_TAGS_MAX_KEYS: usize = 4; +const DEFAULT_ADDITIONAL_METRIC_TAGS_MAX_ENTRIES: usize = 100; + +/// Deduplicate, sort alphabetically, and cap `keys` using [`ADDITIONAL_METRIC_TAGS_MAX_KEYS`]. +/// Excess keys are dropped and logged as a one time warning. +fn normalize_additional_metric_tag_keys(mut keys: Vec) -> Vec { + keys.sort_unstable(); + keys.dedup(); + if keys.len() > ADDITIONAL_METRIC_TAGS_MAX_KEYS { + let dropped = keys.split_off(ADDITIONAL_METRIC_TAGS_MAX_KEYS); + warn!( + "additional_metric_tag_keys: {} additional metric tag keys exceed the cap of {}; dropping: {:?}", + dropped.len() + ADDITIONAL_METRIC_TAGS_MAX_KEYS, + ADDITIONAL_METRIC_TAGS_MAX_KEYS, + dropped, + ); + } + keys +} + /// Concentrators that can provide raw time buckets for export implement this trait. /// /// `StatsExporter` is generic over `C: FlushableConcentrator` so it can work with @@ -112,6 +133,10 @@ pub struct SpanConcentrator { span_kinds_stats_computed: Vec, /// keys for supplementary tags that describe peer.service entities peer_tag_keys: Vec, + /// keys for additional tags on trace stats + additional_metric_tag_keys: Vec, + /// limit on distinct stat entries with additional metric tags per flush bucket + additional_metric_tags_max_entries: usize, #[cfg(feature = "stats-obfuscation")] obfuscation_config: SharedStatsComputationObfuscationConfig, } @@ -121,9 +146,10 @@ impl SpanConcentrator { /// - `bucket_size` is the size of the time buckets /// - `now` the current system time, used to define the oldest bucket /// - `span_kinds_stats_computed` list of span kinds eligible for stats computation - /// - `peer_tags_keys` list of keys considered as peer tags for aggregation + /// - `peer_tag_keys` list of keys considered as peer tags for aggregation /// - `override_max_entries_per_bucket` maximum distinct aggregation keys per time bucket before /// cardinality limiting applies. Pass `None` to use [`DEFAULT_MAX_ENTRIES_PER_BUCKET`]. + /// - `additional_metric_tag_keys` list of keys considered as addtional tags for aggregation /// - `obfuscation_config` optional and updatable config for resource key obfuscation pub fn new( bucket_size: Duration, @@ -131,6 +157,7 @@ impl SpanConcentrator { span_kinds_stats_computed: Vec, peer_tag_keys: Vec, override_max_entries_per_bucket: Option, + additional_metric_tag_keys: Vec, #[cfg(feature = "stats-obfuscation")] obfuscation_config: Option< SharedStatsComputationObfuscationConfig, >, @@ -147,6 +174,10 @@ impl SpanConcentrator { .unwrap_or(DEFAULT_MAX_ENTRIES_PER_BUCKET), span_kinds_stats_computed, peer_tag_keys, + additional_metric_tag_keys: normalize_additional_metric_tag_keys( + additional_metric_tag_keys, + ), + additional_metric_tags_max_entries: DEFAULT_ADDITIONAL_METRIC_TAGS_MAX_ENTRIES, #[cfg(feature = "stats-obfuscation")] obfuscation_config: obfuscation_config.unwrap_or_default(), } @@ -172,6 +203,35 @@ impl SpanConcentrator { self.peer_tag_keys = peer_tags; } + /// Return the list of keys considered as additional_metric_tag_keys for aggregation + pub fn additional_metric_tag_keys(&self) -> &[String] { + &self.additional_metric_tag_keys + } + + /// Set the list of keys considered as additional_metric_tag_keys for aggregation + pub fn set_additional_metric_tag_keys(&mut self, tag_keys: Vec) { + self.additional_metric_tag_keys = normalize_additional_metric_tag_keys(tag_keys); + } + + /// Return the per-bucket limit on distinct stat entries that include additional metric tags + pub fn additional_metric_tags_max_entries(&self) -> usize { + self.additional_metric_tags_max_entries + } + + /// Set the per-bucket limit on distinct stat entries that include additional metric tags. + /// Values less than or equal to 0 are rejected and the existing limit is preserved with a + /// warning. + pub fn set_additional_metric_tags_max_entries(&mut self, limit: usize) { + if limit == 0 { + warn!( + "DD_TRACE_STATS_ADDITIONAL_TAGS_CARDINALITY_LIMIT must be > 0; keeping default of {}", + self.additional_metric_tags_max_entries, + ); + return; + } + self.additional_metric_tags_max_entries = limit; + } + /// Return the bucket size used for aggregation pub fn get_bucket_size(&self) -> Duration { Duration::from_nanos(self.bucket_size) @@ -196,12 +256,23 @@ impl SpanConcentrator { res, span, self.peer_tag_keys.as_slice(), + self.additional_metric_tag_keys.as_slice(), + ), + None => BorrowedAggregationKey::from_span( + span, + self.peer_tag_keys.as_slice(), + self.additional_metric_tag_keys.as_slice(), ), - None => BorrowedAggregationKey::from_span(span, self.peer_tag_keys.as_slice()), }; self.buckets .entry(bucket_timestamp) - .or_insert_with(|| StatsBucket::new(bucket_timestamp, self.max_entries_per_bucket)) + .or_insert_with(|| { + StatsBucket::new( + bucket_timestamp, + self.max_entries_per_bucket, + self.additional_metric_tags_max_entries, + ) + }) .insert( agg_key, span.duration(), diff --git a/libdd-trace-stats/src/span_concentrator/tests.rs b/libdd-trace-stats/src/span_concentrator/tests.rs index 3fd088e09b..9ef3e9c761 100644 --- a/libdd-trace-stats/src/span_concentrator/tests.rs +++ b/libdd-trace-stats/src/span_concentrator/tests.rs @@ -106,6 +106,7 @@ fn test_concentrator_oldest_timestamp_cold() { vec![], vec![], None, + vec![], #[cfg(feature = "stats-obfuscation")] None, ); @@ -163,6 +164,7 @@ fn test_concentrator_oldest_timestamp_hot() { vec![], vec![], None, + vec![], #[cfg(feature = "stats-obfuscation")] None, ); @@ -243,6 +245,7 @@ fn test_concentrator_stats_totals() { vec![], vec![], None, + vec![], #[cfg(feature = "stats-obfuscation")] None, ); @@ -310,6 +313,7 @@ fn test_concentrator_stats_counts() { vec![], vec![], None, + vec![], #[cfg(feature = "stats-obfuscation")] None, ); @@ -608,6 +612,7 @@ fn test_span_should_be_included_in_stats() { get_span_kinds(), vec![], None, + vec![], #[cfg(feature = "stats-obfuscation")] None, ); @@ -689,6 +694,7 @@ fn test_ignore_partial_spans() { get_span_kinds(), vec![], None, + vec![], #[cfg(feature = "stats-obfuscation")] None, ); @@ -715,6 +721,7 @@ fn test_force_flush() { get_span_kinds(), vec![], None, + vec![], #[cfg(feature = "stats-obfuscation")] None, ); @@ -799,6 +806,7 @@ fn test_peer_tags_aggregation() { get_span_kinds(), vec![], None, + vec![], #[cfg(feature = "stats-obfuscation")] None, ); @@ -808,6 +816,7 @@ fn test_peer_tags_aggregation() { get_span_kinds(), vec!["db.instance".to_string(), "db.system".to_string()], None, + vec![], #[cfg(feature = "stats-obfuscation")] None, ); @@ -992,6 +1001,7 @@ fn test_peer_tags_quantization_aggregation() { "peer.hostname".to_string(), ], None, + vec![], #[cfg(feature = "stats-obfuscation")] None, ); @@ -1119,6 +1129,7 @@ fn test_base_service_peer_tag() { get_span_kinds(), vec!["db.instance".to_string(), "db.system".to_string()], None, + vec![], #[cfg(feature = "stats-obfuscation")] None, ); @@ -1342,6 +1353,7 @@ fn test_pb_span() { get_span_kinds(), vec!["db.instance".to_string(), "db.system".to_string()], None, + vec!["custom.primary".to_string()], #[cfg(feature = "stats-obfuscation")] None, ); @@ -1461,6 +1473,32 @@ fn test_pb_span() { span_events: vec![], } }, + // Span with measured flag and additional metric tags + { + let mut meta = std::collections::HashMap::new(); + meta.insert("custom.primary".to_string(), "val".to_string()); + + let mut metrics = std::collections::HashMap::new(); + metrics.insert("_dd.measured".to_string(), 1.0); + + pb::Span { + service: "service1".to_string(), + name: "query".to_string(), + resource: "database_query".to_string(), + trace_id: 1, + span_id: 6, + parent_id: 1, + start: (aligned_now - BUCKET_SIZE + 40) as i64, + duration: 150, + error: 1, + r#type: "db".to_string(), + meta, + metrics, + meta_struct: std::collections::HashMap::new(), + span_links: vec![], + span_events: vec![], + } + }, // Grpc span { let mut meta = std::collections::HashMap::new(); @@ -1563,6 +1601,20 @@ fn test_pb_span() { is_trace_root: pb::Trilean::False.into(), ..Default::default() }, + // Measured span with additional metric tags + pb::ClientGroupedStats { + service: "service1".to_string(), + resource: "database_query".to_string(), + r#type: "db".to_string(), + name: "query".to_string(), + duration: 150, + hits: 1, + top_level_hits: 0, + errors: 1, + is_trace_root: pb::Trilean::False.into(), + additional_metric_tags: vec!["custom.primary:val".to_string()], + ..Default::default() + }, pb::ClientGroupedStats { service: "service1".to_string(), name: "rpc.grpc".to_string(), @@ -1594,6 +1646,7 @@ fn test_flush_with_otlp_exact_per_cell_scalars() { get_span_kinds(), vec![], None, + vec![], #[cfg(feature = "stats-obfuscation")] None, ); @@ -1645,6 +1698,7 @@ fn make_cardinality_concentrator(max_entries: usize) -> SpanConcentrator { get_span_kinds(), vec![], Some(max_entries), + vec![], #[cfg(feature = "stats-obfuscation")] None, ) @@ -1896,3 +1950,685 @@ fn test_overflow_bucket_key_sentinel_values() { assert_eq!(normal.service, "my-service"); assert_eq!(normal.resource, "my-resource"); } + +#[test] +fn test_additional_metric_tags_aggregation() { + let now = SystemTime::now(); + let mut spans = vec![ + get_test_span_with_meta( + now, + 1, + 0, + 100, + 5, + "A1", + "GET /objects", + 0, + &[("custom.primary", "a")], + &[("_dd.measured", 1.0)], + ), + get_test_span_with_meta( + now, + 2, + 0, + 100, + 5, + "A1", + "GET /objects", + 0, + &[("custom.primary", "b")], + &[("_dd.measured", 1.0)], + ), + ]; + compute_top_level_span(spans.as_mut_slice()); + + let mut concentrator_without_additional_metric_tags = SpanConcentrator::new( + Duration::from_nanos(BUCKET_SIZE), + now, + get_span_kinds(), + vec![], + None, + vec![], + #[cfg(feature = "stats-obfuscation")] + None, + ); + let mut concentrator_with_additional_metric_tags = SpanConcentrator::new( + Duration::from_nanos(BUCKET_SIZE), + now, + get_span_kinds(), + vec![], + None, + vec!["custom.primary".to_string()], + #[cfg(feature = "stats-obfuscation")] + None, + ); + for span in &spans { + concentrator_without_additional_metric_tags.add_span(span); + concentrator_with_additional_metric_tags.add_span(span); + } + + let flushtime = now + + Duration::from_nanos( + concentrator_with_additional_metric_tags.bucket_size + * concentrator_with_additional_metric_tags.buffer_len as u64, + ); + + let expected_without_additional_metric_tags = vec![pb::ClientGroupedStats { + service: "A1".to_string(), + resource: "GET /objects".to_string(), + r#type: "db".to_string(), + name: "query".to_string(), + duration: 200, + hits: 2, + top_level_hits: 2, + errors: 0, + is_trace_root: pb::Trilean::True.into(), + ..Default::default() + }]; + + let expected_with_additional_metric_tags = vec![ + pb::ClientGroupedStats { + service: "A1".to_string(), + resource: "GET /objects".to_string(), + r#type: "db".to_string(), + name: "query".to_string(), + additional_metric_tags: vec!["custom.primary:a".to_string()], + duration: 100, + hits: 1, + top_level_hits: 1, + errors: 0, + is_trace_root: pb::Trilean::True.into(), + ..Default::default() + }, + pb::ClientGroupedStats { + service: "A1".to_string(), + resource: "GET /objects".to_string(), + r#type: "db".to_string(), + name: "query".to_string(), + additional_metric_tags: vec!["custom.primary:b".to_string()], + duration: 100, + hits: 1, + top_level_hits: 1, + errors: 0, + is_trace_root: pb::Trilean::True.into(), + ..Default::default() + }, + ]; + + assert_counts_equal( + expected_without_additional_metric_tags, + concentrator_without_additional_metric_tags + .flush(flushtime, false) + .0 + .first() + .expect("There should be at least one time bucket") + .stats + .clone(), + ); + assert_counts_equal( + expected_with_additional_metric_tags, + concentrator_with_additional_metric_tags + .flush(flushtime, false) + .0 + .first() + .expect("There should be at least one time bucket") + .stats + .clone(), + ); +} + +#[test] +fn test_additional_metric_tags_max_entries_masks_overflow_entries() { + // With a cap of 1, the first distinct tag value is admitted; the second gets masked. + let now = SystemTime::now(); + let meta_a = [("region", "us-east-1")]; + let meta_b = [("region", "eu-west-1")]; + let mut spans = vec![ + get_test_span_with_meta( + now, + 1, + 0, + 100, + 5, + "svc", + "GET /a", + 0, + &meta_a, + &[("_dd.measured", 1.0)], + ), + get_test_span_with_meta( + now, + 2, + 0, + 100, + 5, + "svc", + "GET /b", + 0, + &meta_b, + &[("_dd.measured", 1.0)], + ), + ]; + compute_top_level_span(spans.as_mut_slice()); + + let mut concentrator = SpanConcentrator::new( + Duration::from_nanos(BUCKET_SIZE), + now, + get_span_kinds(), + vec![], + None, + vec!["region".to_string()], + #[cfg(feature = "stats-obfuscation")] + None, + ); + concentrator.set_additional_metric_tags_max_entries(1); + for span in &spans { + concentrator.add_span(span); + } + + let flushtime = + now + Duration::from_nanos(concentrator.bucket_size * concentrator.buffer_len as u64); + let (buckets, _) = concentrator.flush(flushtime, false); + let mut tags: Vec<&str> = buckets[0] + .stats + .iter() + .flat_map(|s| s.additional_metric_tags.iter().map(String::as_str)) + .collect(); + tags.sort_unstable(); + assert_eq!( + tags, + vec!["region:tracer_blocked_value", "region:us-east-1"] + ); +} + +#[test] +fn test_additional_metric_tags_max_entries_existing_keys_merge_after_limit() { + // A key admitted before the cap is hit continues merging after the cap is exceeded. + let now = SystemTime::now(); + let meta_a = [("region", "us-east-1")]; + let meta_b = [("region", "eu-west-1")]; + // Three spans: first two fill and exceed a cap of 1, third re-hits the admitted key. + let mut spans = vec![ + get_test_span_with_meta( + now, + 1, + 0, + 100, + 5, + "svc", + "GET /a", + 0, + &meta_a, + &[("_dd.measured", 1.0)], + ), + get_test_span_with_meta( + now, + 2, + 0, + 100, + 5, + "svc", + "GET /b", + 0, + &meta_b, + &[("_dd.measured", 1.0)], + ), + get_test_span_with_meta( + now, + 3, + 0, + 100, + 5, + "svc", + "GET /a", + 0, + &meta_a, + &[("_dd.measured", 1.0)], + ), + ]; + compute_top_level_span(spans.as_mut_slice()); + + let mut concentrator = SpanConcentrator::new( + Duration::from_nanos(BUCKET_SIZE), + now, + get_span_kinds(), + vec![], + None, + vec!["region".to_string()], + #[cfg(feature = "stats-obfuscation")] + None, + ); + concentrator.set_additional_metric_tags_max_entries(1); + for span in &spans { + concentrator.add_span(span); + } + + let flushtime = + now + Duration::from_nanos(concentrator.bucket_size * concentrator.buffer_len as u64); + let (buckets, _) = concentrator.flush(flushtime, false); + let admitted = buckets[0] + .stats + .iter() + .find(|s| s.additional_metric_tags == vec!["region:us-east-1"]) + .expect("admitted entry should exist"); + // spans 1 and 3 both have region:us-east-1 and the same agg key, so hits == 2. + assert_eq!(admitted.hits, 2); +} + +#[test] +fn test_additional_metric_tags_max_entries_resets_on_flush() { + // After flushing, a new bucket starts with a fresh cap budget. + let now = SystemTime::now(); + let meta_a = [("region", "us-east-1")]; + let meta_b = [("region", "eu-west-1")]; + let mut spans = vec![ + get_test_span_with_meta( + now, + 1, + 0, + 100, + 5, + "svc", + "GET /a", + 0, + &meta_a, + &[("_dd.measured", 1.0)], + ), + get_test_span_with_meta( + now, + 2, + 0, + 100, + 5, + "svc", + "GET /b", + 0, + &meta_b, + &[("_dd.measured", 1.0)], + ), + ]; + compute_top_level_span(spans.as_mut_slice()); + + let mut concentrator = SpanConcentrator::new( + Duration::from_nanos(BUCKET_SIZE), + now, + get_span_kinds(), + vec![], + None, + vec!["region".to_string()], + #[cfg(feature = "stats-obfuscation")] + None, + ); + concentrator.set_additional_metric_tags_max_entries(1); + + // First bucket: cap of 1 means only us-east-1 is admitted unmasked. + for span in &spans { + concentrator.add_span(span); + } + let flush1_time = + now + Duration::from_nanos(concentrator.bucket_size * concentrator.buffer_len as u64); + let _first_buckets = concentrator.flush(flush1_time, true); + + // Second bucket: cap resets; eu-west-1 can now be admitted unmasked. + let later = flush1_time + Duration::from_nanos(BUCKET_SIZE); + let meta_b2 = [("region", "eu-west-1")]; + let mut spans2 = vec![get_test_span_with_meta( + later, + 4, + 0, + 100, + 5, + "svc", + "GET /b", + 0, + &meta_b2, + &[("_dd.measured", 1.0)], + )]; + compute_top_level_span(spans2.as_mut_slice()); + concentrator.add_span(&spans2[0]); + + let flush2_time = + later + Duration::from_nanos(concentrator.bucket_size * concentrator.buffer_len as u64); + let (second_buckets, _) = concentrator.flush(flush2_time, true); + let tags: Vec<&str> = second_buckets + .iter() + .flat_map(|b| b.stats.iter()) + .flat_map(|s| s.additional_metric_tags.iter().map(String::as_str)) + .collect(); + assert!( + tags.contains(&"region:eu-west-1"), + "eu-west-1 should be admitted unmasked in a fresh bucket, got: {tags:?}" + ); +} + +#[test] +fn test_additional_metric_tags_new_key_collapses_into_overflow_before_masking() { + // An untagged span already fills the bucket's max_entries. A new tagged span, even though + // it's under the additional_metric_tags_max_entries budget, must go straight to overflow + // instead of being masked. + let now = SystemTime::now(); + let mut spans = vec![ + get_test_span_with_meta( + now, + 1, + 0, + 100, + 5, + "svc", + "GET /a", + 0, + &[], + &[("_dd.measured", 1.0)], + ), + get_test_span_with_meta( + now, + 2, + 0, + 100, + 5, + "svc", + "GET /b", + 0, + &[("region", "us-east-1")], + &[("_dd.measured", 1.0)], + ), + ]; + compute_top_level_span(spans.as_mut_slice()); + + let mut concentrator = SpanConcentrator::new( + Duration::from_nanos(BUCKET_SIZE), + now, + get_span_kinds(), + vec![], + Some(1), + vec!["region".to_string()], + #[cfg(feature = "stats-obfuscation")] + None, + ); + concentrator.set_additional_metric_tags_max_entries(5); + for span in &spans { + concentrator.add_span(span); + } + + let flushtime = + now + Duration::from_nanos(concentrator.bucket_size * concentrator.buffer_len as u64); + let (buckets, _) = concentrator.flush(flushtime, false); + let stats = &buckets[0].stats; + + assert_eq!( + stats.len(), + 2, + "expected the untagged entry plus one overflow group, got {stats:?}" + ); + let overflow = stats + .iter() + .find(|s| s.resource == TRACER_BLOCKED_VALUE) + .expect("overflow group must exist"); + assert_eq!(overflow.hits, 1); + assert!( + overflow.additional_metric_tags.is_empty(), + "the overflow sentinel carries no additional metric tags, masked or otherwise" + ); +} + +#[test] +fn test_additional_metric_tags_masked_entry_collapses_into_overflow() { + // A span exceeds the additional_metric_tags_max_entries budget, so its tag values get + // masked. But the bucket's max_entries is also already full, so that masked key must also + // collapse into the overflow sentinel instead of becoming its own distinct entry. + let now = SystemTime::now(); + let mut spans = vec![ + get_test_span_with_meta( + now, + 1, + 0, + 100, + 5, + "svc", + "GET /a", + 0, + &[("region", "us-east-1")], + &[("_dd.measured", 1.0)], + ), + get_test_span_with_meta( + now, + 2, + 0, + 100, + 5, + "svc", + "GET /b", + 0, + &[("region", "eu-west-1")], + &[("_dd.measured", 1.0)], + ), + ]; + compute_top_level_span(spans.as_mut_slice()); + + let mut concentrator = SpanConcentrator::new( + Duration::from_nanos(BUCKET_SIZE), + now, + get_span_kinds(), + vec![], + Some(1), + vec!["region".to_string()], + #[cfg(feature = "stats-obfuscation")] + None, + ); + concentrator.set_additional_metric_tags_max_entries(1); + for span in &spans { + concentrator.add_span(span); + } + + let flushtime = + now + Duration::from_nanos(concentrator.bucket_size * concentrator.buffer_len as u64); + let (buckets, _) = concentrator.flush(flushtime, false); + let stats = &buckets[0].stats; + + assert_eq!( + stats.len(), + 2, + "expected the admitted entry plus one overflow group, no separate masked group, got {stats:?}" + ); + let overflow = stats + .iter() + .find(|s| s.resource == TRACER_BLOCKED_VALUE) + .expect("overflow group must exist"); + assert_eq!(overflow.hits, 1); + assert!( + overflow.additional_metric_tags.is_empty(), + "the overflow sentinel carries no additional metric tags, masked or otherwise" + ); +} + +#[test] +fn test_additional_metric_tags_multiple_masked_entries_merge() { + // Two distinct source keys that both get masked to the same resource/tag combination must + // merge into a single masked entry rather than producing duplicate masked rows. + let now = SystemTime::now(); + let mut spans = vec![ + get_test_span_with_meta( + now, + 1, + 0, + 100, + 5, + "svc", + "GET /a", + 0, + &[("region", "us-east-1")], + &[("_dd.measured", 1.0)], + ), + get_test_span_with_meta( + now, + 2, + 0, + 100, + 5, + "svc", + "GET /b", + 0, + &[("region", "eu-west-1")], + &[("_dd.measured", 1.0)], + ), + get_test_span_with_meta( + now, + 3, + 0, + 100, + 5, + "svc", + "GET /b", + 0, + &[("region", "ap-south-1")], + &[("_dd.measured", 1.0)], + ), + ]; + compute_top_level_span(spans.as_mut_slice()); + + let mut concentrator = SpanConcentrator::new( + Duration::from_nanos(BUCKET_SIZE), + now, + get_span_kinds(), + vec![], + None, + vec!["region".to_string()], + #[cfg(feature = "stats-obfuscation")] + None, + ); + concentrator.set_additional_metric_tags_max_entries(1); + for span in &spans { + concentrator.add_span(span); + } + + let flushtime = + now + Duration::from_nanos(concentrator.bucket_size * concentrator.buffer_len as u64); + let (buckets, _) = concentrator.flush(flushtime, false); + let stats = &buckets[0].stats; + + assert_eq!( + stats.len(), + 2, + "expected one admitted entry and one merged masked entry, got {stats:?}" + ); + let masked = stats + .iter() + .find(|s| s.additional_metric_tags == vec!["region:tracer_blocked_value".to_string()]) + .expect("masked entry should exist"); + assert_eq!( + masked.hits, 2, + "both GET /b spans should merge into the same masked entry" + ); +} + +#[test] +fn test_additional_metric_tag_value_length_cap_substitutes_blocked_value() { + let now = SystemTime::now(); + let long_value = "x".repeat(201); + let meta = [("region", long_value.as_str())]; + let mut spans = vec![get_test_span_with_meta( + now, + 1, + 0, + 100, + 5, + "svc", + "GET /foo", + 0, + &meta, + &[("_dd.measured", 1.0)], + )]; + compute_top_level_span(spans.as_mut_slice()); + + let mut concentrator = SpanConcentrator::new( + Duration::from_nanos(BUCKET_SIZE), + now, + get_span_kinds(), + vec![], + None, + vec!["region".to_string()], + #[cfg(feature = "stats-obfuscation")] + None, + ); + concentrator.add_span(&spans[0]); + + let flushtime = + now + Duration::from_nanos(concentrator.bucket_size * concentrator.buffer_len as u64); + let buckets = concentrator.flush(flushtime, false); + let tags = &buckets.0[0].stats[0].additional_metric_tags; + assert_eq!(tags, &["region:tracer_blocked_value"]); +} + +#[test] +fn test_additional_metric_tag_value_at_length_cap_passes_through() { + let now = SystemTime::now(); + let ok_value = "x".repeat(200); + let meta = [("region", ok_value.as_str())]; + let mut spans = vec![get_test_span_with_meta( + now, + 1, + 0, + 100, + 5, + "svc", + "GET /foo", + 0, + &meta, + &[("_dd.measured", 1.0)], + )]; + compute_top_level_span(spans.as_mut_slice()); + + let mut concentrator = SpanConcentrator::new( + Duration::from_nanos(BUCKET_SIZE), + now, + get_span_kinds(), + vec![], + None, + vec!["region".to_string()], + #[cfg(feature = "stats-obfuscation")] + None, + ); + concentrator.add_span(&spans[0]); + + let flushtime = + now + Duration::from_nanos(concentrator.bucket_size * concentrator.buffer_len as u64); + let buckets = concentrator.flush(flushtime, false); + let tags = &buckets.0[0].stats[0].additional_metric_tags; + assert_eq!(tags, &[format!("region:{ok_value}")]); +} + +#[test] +fn test_normalize_additional_metric_tag_keys_sort() { + let keys = vec![ + "region".to_string(), + "env".to_string(), + "tenant".to_string(), + ]; + let result = normalize_additional_metric_tag_keys(keys); + assert_eq!(result, vec!["env", "region", "tenant"]); +} + +#[test] +fn test_normalize_additional_metric_tag_keys_dedup() { + let keys = vec![ + "region".to_string(), + "region".to_string(), + "tenant".to_string(), + ]; + let result = normalize_additional_metric_tag_keys(keys); + assert_eq!(result, vec!["region", "tenant"]); +} + +#[test] +fn test_normalize_additional_metric_tag_keys_limit() { + let keys = vec![ + "aaa".to_string(), + "bbb".to_string(), + "ccc".to_string(), + "ddd".to_string(), + "eee".to_string(), + ]; + let result = normalize_additional_metric_tag_keys(keys); + assert_eq!(result, vec!["aaa", "bbb", "ccc", "ddd"]); + assert_eq!(result.len(), 4); +} diff --git a/libdd-trace-stats/src/stats_exporter.rs b/libdd-trace-stats/src/stats_exporter.rs index c771fd5b2b..6c8f17f1d8 100644 --- a/libdd-trace-stats/src/stats_exporter.rs +++ b/libdd-trace-stats/src/stats_exporter.rs @@ -360,6 +360,7 @@ mod tests { vec![], vec![], None, + vec![], #[cfg(feature = "stats-obfuscation")] None, ); @@ -632,6 +633,7 @@ mod tests { vec![], vec![], Some(1), // max 1 distinct key → second span collapses + vec![], #[cfg(feature = "stats-obfuscation")] None, ); diff --git a/libdd-trace-utils/src/stats_utils.rs b/libdd-trace-utils/src/stats_utils.rs index 60324f0ce7..01d21e80c1 100644 --- a/libdd-trace-utils/src/stats_utils.rs +++ b/libdd-trace-utils/src/stats_utils.rs @@ -136,6 +136,7 @@ mod mini_agent_tests { 0 ], "GRPCStatusCode": "0", + "AdditionalMetricTags": [], "HTTPMethod": "GET", "HTTPEndpoint": "/test" } From a7a73a5ffe7dfe1651b618c9556efcfc5ce16e30 Mon Sep 17 00:00:00 2001 From: Duncan Harvey Date: Mon, 6 Jul 2026 16:44:09 -0400 Subject: [PATCH 2/2] use character limit rather than byte limit for ADDITIONAL_METRIC_TAG_VALUE_MAX_LEN --- .../src/span_concentrator/aggregation.rs | 2 +- .../src/span_concentrator/tests.rs | 79 +++++++++++++++++++ 2 files changed, 80 insertions(+), 1 deletion(-) diff --git a/libdd-trace-stats/src/span_concentrator/aggregation.rs b/libdd-trace-stats/src/span_concentrator/aggregation.rs index e638c87122..3dd4b1954e 100644 --- a/libdd-trace-stats/src/span_concentrator/aggregation.rs +++ b/libdd-trace-stats/src/span_concentrator/aggregation.rs @@ -286,7 +286,7 @@ impl<'a> BorrowedAggregationKey<'a> { .iter() .filter_map(|key| match span.get_meta(key.as_str()) { Some(v) if !v.is_empty() => { - if v.len() > ADDITIONAL_METRIC_TAG_VALUE_MAX_LEN { + if v.chars().count() > ADDITIONAL_METRIC_TAG_VALUE_MAX_LEN { warn!( "additional_metric_tags: value for key '{}' exceeds {} characters; substituting tracer_blocked_value", key, ADDITIONAL_METRIC_TAG_VALUE_MAX_LEN, diff --git a/libdd-trace-stats/src/span_concentrator/tests.rs b/libdd-trace-stats/src/span_concentrator/tests.rs index 9ef3e9c761..e43428688f 100644 --- a/libdd-trace-stats/src/span_concentrator/tests.rs +++ b/libdd-trace-stats/src/span_concentrator/tests.rs @@ -2597,6 +2597,85 @@ fn test_additional_metric_tag_value_at_length_cap_passes_through() { assert_eq!(tags, &[format!("region:{ok_value}")]); } +#[test] +fn test_additional_metric_tag_value_multibyte_at_length_cap_passes_through() { + // 101 two-byte characters: 202 bytes but only 101 chars, well under the 200-character cap. + let now = SystemTime::now(); + let ok_value = "é".repeat(101); + let meta = [("region", ok_value.as_str())]; + let mut spans = vec![get_test_span_with_meta( + now, + 1, + 0, + 100, + 5, + "svc", + "GET /foo", + 0, + &meta, + &[("_dd.measured", 1.0)], + )]; + compute_top_level_span(spans.as_mut_slice()); + + let mut concentrator = SpanConcentrator::new( + Duration::from_nanos(BUCKET_SIZE), + now, + get_span_kinds(), + vec![], + None, + vec!["region".to_string()], + #[cfg(feature = "stats-obfuscation")] + None, + ); + concentrator.add_span(&spans[0]); + + let flushtime = + now + Duration::from_nanos(concentrator.bucket_size * concentrator.buffer_len as u64); + let buckets = concentrator.flush(flushtime, false); + let tags = &buckets.0[0].stats[0].additional_metric_tags; + assert_eq!(tags, &[format!("region:{ok_value}")]); +} + +#[test] +fn test_additional_metric_tag_value_multibyte_over_length_cap_substitutes_blocked_value() { + // 201 two-byte characters exceeds the 200-character cap even though earlier bytes-based + // checks would have let a 201-char, sub-200-byte value like this through incorrectly. + let now = SystemTime::now(); + let long_value = "é".repeat(201); + let meta = [("region", long_value.as_str())]; + let mut spans = vec![get_test_span_with_meta( + now, + 1, + 0, + 100, + 5, + "svc", + "GET /foo", + 0, + &meta, + &[("_dd.measured", 1.0)], + )]; + compute_top_level_span(spans.as_mut_slice()); + + let mut concentrator = SpanConcentrator::new( + Duration::from_nanos(BUCKET_SIZE), + now, + get_span_kinds(), + vec![], + None, + vec!["region".to_string()], + #[cfg(feature = "stats-obfuscation")] + None, + ); + concentrator.add_span(&spans[0]); + + let flushtime = + now + Duration::from_nanos(concentrator.bucket_size * concentrator.buffer_len as u64); + let buckets = concentrator.flush(flushtime, false); + let tags = &buckets.0[0].stats[0].additional_metric_tags; + assert_eq!(tags, &["region:tracer_blocked_value"]); +} + #[test] fn test_normalize_additional_metric_tag_keys_sort() { let keys = vec![