diff --git a/lib/saluki-components/src/common/otlp/attributes/mod.rs b/lib/saluki-components/src/common/otlp/attributes/mod.rs index ceae9e298c..fb34e16018 100644 --- a/lib/saluki-components/src/common/otlp/attributes/mod.rs +++ b/lib/saluki-components/src/common/otlp/attributes/mod.rs @@ -231,7 +231,6 @@ fn raw_tag_value(value: &Value) -> Option { } } -#[allow(dead_code)] pub(super) fn origin_id_from_attributes(attributes: &[otlp_common::KeyValue]) -> Option { let mut pod_uid = None; @@ -330,6 +329,46 @@ mod tests { tags.into_iter().any(|t| t.as_str() == tag) } + #[test] + fn origin_id_uses_container_id() { + let attributes = vec![attr(CONTAINER_ID, Value::StringValue("container-123".into()))]; + + assert_eq!( + origin_id_from_attributes(&attributes).as_deref(), + Some("container_id://container-123") + ); + } + + #[test] + fn origin_id_uses_pod_uid_when_container_id_is_absent() { + let attributes = vec![attr(K8S_POD_UID, Value::StringValue("pod-123".into()))]; + + assert_eq!( + origin_id_from_attributes(&attributes).as_deref(), + Some("kubernetes_pod_uid://pod-123") + ); + } + + #[test] + fn origin_id_prefers_container_id_over_pod_uid() { + let attributes = vec![ + attr(K8S_POD_UID, Value::StringValue("pod-123".into())), + attr(CONTAINER_ID, Value::StringValue("container-123".into())), + ]; + + assert_eq!( + origin_id_from_attributes(&attributes).as_deref(), + Some("container_id://container-123") + ); + } + + #[test] + fn origin_id_is_absent_without_supported_attributes() { + let attributes = vec![attr("service.name", Value::StringValue("api".into()))]; + + assert_eq!(origin_id_from_attributes(&attributes), None); + } + #[test] fn mapped_mode_emits_only_recognized_mappings() { let attributes = vec![ diff --git a/lib/saluki-components/src/common/otlp/attributes/translator.rs b/lib/saluki-components/src/common/otlp/attributes/translator.rs index d48df7129a..78c15c2319 100644 --- a/lib/saluki-components/src/common/otlp/attributes/translator.rs +++ b/lib/saluki-components/src/common/otlp/attributes/translator.rs @@ -16,7 +16,6 @@ impl AttributeTranslator { tags_from_attributes(attributes, mode) } - #[allow(dead_code)] pub fn origin_id_from_attributes(&self, attributes: &[otlp_common::KeyValue]) -> Option { origin_id_from_attributes(attributes) } diff --git a/lib/saluki-components/src/sources/otlp/metrics/cache.rs b/lib/saluki-components/src/sources/otlp/metrics/cache.rs index e73d1f07de..9034501f13 100644 --- a/lib/saluki-components/src/sources/otlp/metrics/cache.rs +++ b/lib/saluki-components/src/sources/otlp/metrics/cache.rs @@ -5,6 +5,12 @@ use saluki_common::cache::{Cache, CacheBuilder}; use super::config::OtlpMetricsTranslatorConfig; use super::dimensions::Dimensions; +type CacheKey = saluki_context::ContextKey; + +fn cache_key(dimensions: &Dimensions) -> CacheKey { + dimensions.context_key() +} + /// The state we store for each unique time series. #[derive(Clone, Debug)] pub struct NumberCounter { @@ -23,8 +29,8 @@ pub struct Extrema { /// A cache for storing previous data points to calculate deltas for cumulative metrics. pub struct PointsCache { - number_points: Cache, - extrema_points: Cache, + number_points: Cache, + extrema_points: Cache, } impl PointsCache { @@ -84,7 +90,7 @@ impl PointsCache { } fn put_and_get_diff(&mut self, dims: &Dimensions, start_timestamp: u64, timestamp: u64, value: f64) -> (f64, bool) { - let key = dims.get_cache_key(); + let key = cache_key(dims); let mut dx = 0.0; let mut ok = false; @@ -118,7 +124,7 @@ impl PointsCache { let mut first_point = true; let drop_point = false; - let key = dims.get_cache_key(); + let key = cache_key(dims); if let Some(prev_counter) = self.number_points.get(&key) { if prev_counter.timestamp >= timestamp { @@ -150,7 +156,7 @@ impl PointsCache { fn put_and_check_extrema( &mut self, dims: &Dimensions, start_timestamp: u64, timestamp: u64, current_extrema: f64, is_min: bool, ) -> bool { - let key = dims.get_cache_key(); + let key = cache_key(dims); let mut from_last_window = false; if let Some(prev_extrema) = self.extrema_points.get(&key) { @@ -246,6 +252,43 @@ mod tests { dx } + fn dimensions_with_origin_id(origin_id: &str) -> Dimensions { + Dimensions { + name: "http.request.duration".to_string(), + tags: Default::default(), + host: Some("host-a".into()), + origin_id: Some(origin_id.to_string()), + } + } + + fn assert_origins_use_independent_differential_cache_entries(first_origin_id: &str, second_origin_id: &str) { + let first = dimensions_with_origin_id(first_origin_id); + let second = dimensions_with_origin_id(second_origin_id); + let mut cache = PointsCache::for_tests(); + + assert_ne!(first.get_cache_key(), second.get_cache_key()); + assert_eq!(cache.monotonic_diff(&first, 100, 100, 10.0), (0.0, true, false)); + assert_eq!(cache.monotonic_diff(&second, 100, 100, 10.0), (0.0, true, false)); + assert_eq!(cache.monotonic_diff(&first, 100, 200, 15.0), (5.0, false, false)); + assert_eq!(cache.monotonic_diff(&second, 100, 200, 12.0), (2.0, false, false)); + } + + #[test] + fn container_origins_use_independent_differential_cache_entries() { + assert_origins_use_independent_differential_cache_entries( + "container_id://container-a", + "container_id://container-b", + ); + } + + #[test] + fn kubernetes_pod_origins_use_independent_differential_cache_entries() { + assert_origins_use_independent_differential_cache_entries( + "kubernetes_pod_uid://pod-a", + "kubernetes_pod_uid://pod-b", + ); + } + fn unknown_start_points() -> Vec { vec![ Point { diff --git a/lib/saluki-components/src/sources/otlp/metrics/dimensions.rs b/lib/saluki-components/src/sources/otlp/metrics/dimensions.rs index 298159efec..9fd9754ae5 100644 --- a/lib/saluki-components/src/sources/otlp/metrics/dimensions.rs +++ b/lib/saluki-components/src/sources/otlp/metrics/dimensions.rs @@ -1,7 +1,11 @@ //! OTLP metric dimensions. use otlp_protos::opentelemetry::proto::common::v1 as otlp_common; -use saluki_context::tags::{SharedTagSet, Tag, TagSet}; +use saluki_context::{ + hash_context_with_host, + tags::{SharedTagSet, Tag, TagSet}, + ContextKey, +}; use stringtheory::MetaString; use super::internal::utils::format_key_value_tag; @@ -12,7 +16,6 @@ pub struct Dimensions { pub name: String, pub tags: SharedTagSet, pub host: Option, - #[allow(dead_code)] pub origin_id: Option, } @@ -95,6 +98,7 @@ impl Dimensions { } /// Creates a canonical cache key for the dimension set. + #[cfg(test)] pub fn get_cache_key(&self) -> String { let mut dimensions: Vec = self.tags.into_iter().map(|t| t.to_string()).collect(); @@ -113,6 +117,16 @@ impl Dimensions { // Join with a null character separator. dimensions.join("\0") } + + /// Creates a compact context key for the dimension set. + /// + /// The host remains a distinct scalar dimension, while the origin ID is hashed as an origin tag. + pub fn context_key(&self) -> ContextKey { + let (context_key, _) = + hash_context_with_host(&self.name, self.host.as_deref(), &self.tags, self.origin_id.as_deref()); + + context_key + } } #[cfg(test)] @@ -131,6 +145,25 @@ mod tests { } } + #[test] + fn with_suffix_and_add_tags_preserve_origin_id() { + let dimensions = Dimensions { + name: "http.request.duration".to_string(), + tags: TagSet::default().into_shared(), + host: None, + origin_id: Some("container_id://container-123".to_string()), + }; + + assert_eq!( + dimensions.with_suffix("count").origin_id.as_deref(), + Some("container_id://container-123") + ); + assert_eq!( + dimensions.add_tags(["env:prod".to_string()]).origin_id.as_deref(), + Some("container_id://container-123") + ); + } + #[test] fn add_tags_preserves_order_deduplication_and_base() { let base = dims_with_tags("http.request.duration", &["env:prod", "service:web", "lower_bound:1.0"]); @@ -146,6 +179,22 @@ mod tests { assert_eq!(base_tags, vec!["env:prod", "service:web", "lower_bound:1.0"]); } + #[test] + fn cache_key_is_independent_of_tag_order() { + let first = dims_with_tags("http.request.duration", &["service:web", "env:prod"]); + let second = dims_with_tags("http.request.duration", &["env:prod", "service:web"]); + + assert_eq!(first.get_cache_key(), second.get_cache_key()); + } + + #[test] + fn context_key_is_independent_of_tag_order() { + let first = dims_with_tags("http.request.duration", &["service:web", "env:prod"]); + let second = dims_with_tags("http.request.duration", &["env:prod", "service:web"]); + + assert_eq!(first.context_key(), second.context_key()); + } + #[test] fn add_tags_preserves_canonical_cache_key() { let base = dims_with_tags("http.request.duration", &["service:web", "env:prod"]); diff --git a/lib/saluki-context/src/hash.rs b/lib/saluki-context/src/hash.rs index 99237e8fca..1ed02b9f0e 100644 --- a/lib/saluki-context/src/hash.rs +++ b/lib/saluki-context/src/hash.rs @@ -43,11 +43,24 @@ where /// and metric name are order-dependent scalar context fields, not tags. /// /// If a tag is seen more than once, it will be ignored and not included in the overall hash. This function requires the -/// caller to provide the hash set used for tracking duplicates, and is more efficient than [`hash_context`] which +/// caller to provide the hash set used for tracking duplicates, and is more efficient than `hash_context` which /// allocates a new hash set each time. /// /// Returns a hash that uniquely identifies the combination of name, host, tags, and origin of the value. An unset host /// and an explicitly empty host are distinct contexts. +pub fn hash_context_with_host( + name: &str, host: Option<&str>, tags: I, origin_tags: I2, +) -> (ContextKey, TagSetKey) +where + I: IntoIterator, + T: AsRef, + I2: IntoIterator, + T2: AsRef, +{ + let mut seen = PrehashedHashSet::default(); + hash_context_with_host_and_seen(name, host, tags, origin_tags, &mut seen) +} + pub(super) fn hash_context_with_host_and_seen( name: &str, host: Option<&str>, tags: I, origin_tags: I2, seen: &mut PrehashedHashSet, ) -> (ContextKey, TagSetKey) @@ -108,6 +121,9 @@ where (context_key, tagset_key) } +/// A compact hash key that identifies a metric context. +/// +/// The key is process-local and collision-prone by nature because it stores a 64-bit hash. #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] pub struct ContextKey { hash: u64, diff --git a/lib/saluki-context/src/lib.rs b/lib/saluki-context/src/lib.rs index b7032e3d45..f5ced32eea 100644 --- a/lib/saluki-context/src/lib.rs +++ b/lib/saluki-context/src/lib.rs @@ -6,6 +6,7 @@ mod context; pub use self::context::{Context, TagSetMutView, TagSetMutViewState}; mod hash; +pub use self::hash::{hash_context_with_host, ContextKey}; pub mod origin;