feat(otlp): update OTLP cumulative metric caching and increased testing

lib/saluki-components/src/common/otlp/attributes/mod.rs

@@ -231,7 +231,6 @@ fn raw_tag_value(value: &Value) -> Option<String> {
     }
 }
 
-#[allow(dead_code)]
 pub(super) fn origin_id_from_attributes(attributes: &[otlp_common::KeyValue]) -> Option<String> {
     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![

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<String> {
         origin_id_from_attributes(attributes)
     }

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<String, NumberCounter>,
-    extrema_points: Cache<String, Extrema>,
+    number_points: Cache<CacheKey, NumberCounter>,
+    extrema_points: Cache<CacheKey, Extrema>,
 }
 
 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<Point> {
         vec![
             Point {

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<MetaString>,
-    #[allow(dead_code)]
     pub origin_id: Option<String>,
 }
 
@@ -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<String> = 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"]);

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<I, I2, T, T2>(
+    name: &str, host: Option<&str>, tags: I, origin_tags: I2,
+) -> (ContextKey, TagSetKey)
+where
+    I: IntoIterator<Item = T>,
+    T: AsRef<str>,
+    I2: IntoIterator<Item = T2>,
+    T2: AsRef<str>,
+{
+    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<I, I2, T, T2>(
     name: &str, host: Option<&str>, tags: I, origin_tags: I2, seen: &mut PrehashedHashSet<u64>,
 ) -> (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,

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;