diff --git a/libdd-data-pipeline/src/trace_exporter/builder.rs b/libdd-data-pipeline/src/trace_exporter/builder.rs index 06afbd267a..d23a3f4016 100644 --- a/libdd-data-pipeline/src/trace_exporter/builder.rs +++ b/libdd-data-pipeline/src/trace_exporter/builder.rs @@ -763,6 +763,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 eaff519d3d..a90365f53a 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 d42a1d9294..02f2b382c1 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.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, + ); + Some((key.as_str(), TRACER_BLOCKED_VALUE)) + } else { + Some((key.as_str(), v)) + } + } + _ => None, + }) + .collect(); + Self { fixed: FixedAggregationKey { resource_name, @@ -277,6 +322,7 @@ impl<'a> BorrowedAggregationKey<'a> { }, }, peer_tags, + additional_metric_tags, } } } @@ -300,6 +346,7 @@ impl OwnedAggregationKey { is_trace_root: pb::Trilean::NotSet, }, peer_tags: vec![], + additional_metric_tags: vec![], } } } @@ -330,6 +377,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(), } } } @@ -462,9 +517,11 @@ 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 is subject to the + /// `max_entries` limit, which collapses it into the overflow sentinel key. pub(super) fn insert( &mut self, key: BorrowedAggregationKey<'_>, @@ -472,18 +529,31 @@ 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; + // The map can't change size before the entry below is resolved, so this single read + // covers the `max_entries` check in the vacant branch without a further lookup. + let len_before_insert = self.data.len(); + + match self.data.entry_ref(&key) { + // Existing key, merge + hashbrown::hash_map::EntryRef::Occupied(mut e) => { + e.get_mut().insert(duration, is_error, is_top_level); + } + hashbrown::hash_map::EntryRef::Vacant(e) => { + // New key over the max entry limit, collapse into the overflow + // sentinel. + if len_before_insert >= self.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 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. @@ -561,7 +631,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(), } } @@ -585,12 +659,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![], } } } @@ -1103,7 +1179,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, @@ -1116,7 +1192,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), @@ -1144,7 +1221,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, @@ -1172,7 +1249,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, @@ -1193,7 +1270,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 0f233daca2..74da2aada3 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,25 @@ pub use aggregation::{FixedAggregationKey, OtlpExactCell, OtlpExactGroup, OtlpSt pub mod stat_span; pub use stat_span::StatSpan; +const ADDITIONAL_METRIC_TAGS_MAX_KEYS: usize = 4; + +/// 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 +} + /// Result of flushing a concentrator. /// /// Obfuscated and un-obfuscated buckets are kept separate because they must be sent in distinct @@ -136,6 +156,8 @@ 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, #[cfg(feature = "stats-obfuscation")] obfuscation_config: SharedStatsComputationObfuscationConfig, } @@ -145,9 +167,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, @@ -155,6 +178,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, >, @@ -171,6 +195,9 @@ 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, + ), #[cfg(feature = "stats-obfuscation")] obfuscation_config: obfuscation_config.unwrap_or_default(), } @@ -196,6 +223,16 @@ 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 bucket size used for aggregation pub fn get_bucket_size(&self) -> Duration { Duration::from_nanos(self.bucket_size) @@ -236,8 +273,13 @@ 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()), }; target_bucket.insert( agg_key, diff --git a/libdd-trace-stats/src/span_concentrator/tests.rs b/libdd-trace-stats/src/span_concentrator/tests.rs index 81d7e94778..210bd982c1 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, ); @@ -691,6 +696,7 @@ fn test_ignore_partial_spans() { get_span_kinds(), vec![], None, + vec![], #[cfg(feature = "stats-obfuscation")] None, ); @@ -719,6 +725,7 @@ fn test_force_flush() { get_span_kinds(), vec![], None, + vec![], #[cfg(feature = "stats-obfuscation")] None, ); @@ -803,6 +810,7 @@ fn test_peer_tags_aggregation() { get_span_kinds(), vec![], None, + vec![], #[cfg(feature = "stats-obfuscation")] None, ); @@ -812,6 +820,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, ); @@ -1000,6 +1009,7 @@ fn test_peer_tags_quantization_aggregation() { "peer.hostname".to_string(), ], None, + vec![], #[cfg(feature = "stats-obfuscation")] None, ); @@ -1129,6 +1139,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, ); @@ -1352,6 +1363,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, ); @@ -1471,6 +1483,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(); @@ -1573,6 +1611,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(), @@ -1604,6 +1656,7 @@ fn test_flush_with_otlp_exact_per_cell_scalars() { get_span_kinds(), vec![], None, + vec![], #[cfg(feature = "stats-obfuscation")] None, ); @@ -1655,6 +1708,7 @@ fn make_cardinality_concentrator(max_entries: usize) -> SpanConcentrator { get_span_kinds(), vec![], Some(max_entries), + vec![], #[cfg(feature = "stats-obfuscation")] None, ) @@ -1906,3 +1960,320 @@ 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) + .all_buckets() + .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) + .all_buckets() + .first() + .expect("There should be at least one time bucket") + .stats + .clone(), + ); +} + +#[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.all_buckets()[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.all_buckets()[0].stats[0].additional_metric_tags; + 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.all_buckets()[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.all_buckets()[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![ + "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 24aa5d4c1b..2cd173f9c9 100644 --- a/libdd-trace-stats/src/stats_exporter.rs +++ b/libdd-trace-stats/src/stats_exporter.rs @@ -376,6 +376,7 @@ mod tests { vec![], vec![], None, + vec![], #[cfg(feature = "stats-obfuscation")] obfuscation_config, ); @@ -643,6 +644,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" }