enhancement(o11y): emit tag filterlist cache hit/miss/evict counters (#1820)

## Summary

Adds telemetry for cache evictions as well as exposes the hits, misses, and evictions telemetry for tag filter component to Core Agent via Remote Agent Interface.

## Test plan

- [ ] `cargo test -p saluki-common` — cache infrastructure tests pass
- [ ] `cargo test -p agent-data-plane` — remapping output and help text tests pass

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: jesse.szwedko <jesse.szwedko@datadoghq.com>

Author: jszwedko

bin/agent-data-plane/src/state/metrics/mod.rs

@@ -830,6 +830,21 @@ mod tests {
                 ),
                 23.0,
             )),
+            Event::Metric(Metric::counter(
+                Context::from_static_parts("adp.cache_hits_total", &["cache_id:tag_filterlist/context_cache"]),
+                13.0,
+            )),
+            Event::Metric(Metric::counter(
+                Context::from_static_parts("adp.cache_misses_total", &["cache_id:tag_filterlist/context_cache"]),
+                17.0,
+            )),
+            Event::Metric(Metric::counter(
+                Context::from_static_parts(
+                    "adp.cache_items_evicted_total",
+                    &["cache_id:tag_filterlist/context_cache"],
+                ),
+                19.0,
+            )),
         ];
 
         for metric in metrics {
@@ -847,11 +862,15 @@ mod tests {
         assert!(output.contains("tag_filterlist__size 9"));
         assert!(output.contains("tag_filterlist__updates 11"));
         assert!(output.contains("aggregator__filtered_tags 23"));
+        assert!(output.contains("aggregator__filtered_tags_cache_hit 13"));
+        assert!(output.contains("aggregator__filtered_tags_cache_miss 17"));
+        assert!(output.contains("aggregator__filtered_tags_cache_evict 19"));
         assert!(!output.contains("datadog__agent__filterlist__size"));
         assert!(!output.contains("datadog__agent__filterlist__updates"));
         assert!(!output.contains("datadog__agent__dogstatsd__listener_filtered_points"));
         assert!(!output.contains("datadog__agent__aggregator__dogstatsd_filtered_metrics"));
         assert!(!output.contains("component_id="));
+        assert!(!output.contains("cache_id="));
     }
 
     #[test]
@@ -907,5 +926,17 @@ mod tests {
             find("aggregator.filtered_tags"),
             Some("How many tags were filtered from a metric sample")
         );
+        assert_eq!(
+            find("aggregator.filtered_tags_cache_hit"),
+            Some("How many times we hit the cache on filtering tags")
+        );
+        assert_eq!(
+            find("aggregator.filtered_tags_cache_miss"),
+            Some("How many times we missed the cache on filtering tags")
+        );
+        assert_eq!(
+            find("aggregator.filtered_tags_cache_evict"),
+            Some("How many times an entry was evicted from the tag filter cache")
+        );
     }
 }

bin/agent-data-plane/src/state/metrics/rules/dogstatsd.rs

@@ -39,6 +39,24 @@ pub fn get_dogstatsd_remappings() -> Vec<RemapperRule> {
             "aggregator.filtered_tags",
         )
         .with_help_text("How many tags were filtered from a metric sample"),
+        RemapperRule::by_name_and_tags(
+            "adp.cache_hits_total",
+            &["cache_id:tag_filterlist/context_cache"],
+            "aggregator.filtered_tags_cache_hit",
+        )
+        .with_help_text("How many times we hit the cache on filtering tags"),
+        RemapperRule::by_name_and_tags(
+            "adp.cache_misses_total",
+            &["cache_id:tag_filterlist/context_cache"],
+            "aggregator.filtered_tags_cache_miss",
+        )
+        .with_help_text("How many times we missed the cache on filtering tags"),
+        RemapperRule::by_name_and_tags(
+            "adp.cache_items_evicted_total",
+            &["cache_id:tag_filterlist/context_cache"],
+            "aggregator.filtered_tags_cache_evict",
+        )
+        .with_help_text("How many times an entry was evicted from the tag filter cache"),
         RemapperRule::by_name_and_tags(
             "adp.object_pool_acquired",
             &["pool_name:dsd_packet_bufs"],

lib/saluki-common/src/cache/expiry.rs

@@ -6,6 +6,7 @@ use std::{
 
 use crossbeam_queue::ArrayQueue;
 use quick_cache::Lifecycle;
+use saluki_metrics::reexport::metrics::Counter;
 
 use crate::{
     collections::FastHashMap,
@@ -15,6 +16,7 @@ use crate::{
 /// Builder for creating an expiration configuration.
 pub struct ExpirationBuilder<K> {
     time_to_idle: Option<Duration>,
+    items_evicted: Counter,
     _key: PhantomData<K>,
 }
 
@@ -23,9 +25,10 @@ where
     K: Eq + std::hash::Hash,
 {
     /// Creates a new `ExpirationBuilder`.
-    pub fn new() -> Self {
+    pub fn new(items_evicted: Counter) -> Self {
         Self {
             time_to_idle: None,
+            items_evicted,
             _key: PhantomData,
         }
     }
@@ -47,11 +50,14 @@ where
     /// Builds the expiration configuration.
     pub fn build(self) -> (Expiration<K>, ExpiryCapableLifecycle<K>) {
         match self.time_to_idle {
-            None => (Expiration::disabled(), ExpiryCapableLifecycle::disabled()),
+            None => (
+                Expiration::disabled(),
+                ExpiryCapableLifecycle::disabled(self.items_evicted),
+            ),
             Some(time_to_idle) => {
                 let state = Arc::new(State::new(time_to_idle));
                 let expiration = Expiration::from_state(Arc::clone(&state));
-                let lifecycle = ExpiryCapableLifecycle::from_state(state);
+                let lifecycle = ExpiryCapableLifecycle::with_state(state, self.items_evicted);
 
                 (expiration, lifecycle)
             }
@@ -228,15 +234,22 @@ where
 #[derive(Clone)]
 pub(super) struct ExpiryCapableLifecycle<K> {
     state: Option<Arc<State<K>>>,
+    items_evicted: Counter,
 }
 
 impl<K> ExpiryCapableLifecycle<K> {
-    fn disabled() -> Self {
-        Self { state: None }
+    fn disabled(items_evicted: Counter) -> Self {
+        Self {
+            state: None,
+            items_evicted,
+        }
     }
 
-    fn from_state(state: Arc<State<K>>) -> Self {
-        Self { state: Some(state) }
+    fn with_state(state: Arc<State<K>>, items_evicted: Counter) -> Self {
+        Self {
+            state: Some(state),
+            items_evicted,
+        }
     }
 }
 
@@ -251,6 +264,10 @@ where
 
     #[inline]
     fn on_evict(&self, _state: &mut Self::RequestState, key: K, _value: V) {
+        // Note: this fires for all capacity-driven evictions including rejected overweight
+        // inserts (items whose weight exceeds the cache capacity). Callers using custom
+        // weighters may see slight inflation from rejected inserts.
+        self.items_evicted.increment(1);
         if let Some(state) = self.state.as_ref() {
             state.mark_entry_removed(key);
         }

lib/saluki-common/src/cache/mod.rs

@@ -1,6 +1,7 @@
 use std::{marker::PhantomData, num::NonZeroUsize, sync::Arc, time::Duration};
 
 use saluki_error::GenericError;
+use saluki_metrics::reexport::metrics::Counter;
 use saluki_metrics::static_metrics;
 use tokio::time::sleep;
 use tokio_util::sync::{CancellationToken, DropGuard};
@@ -26,6 +27,7 @@ static_metrics! {
         gauge(weight_limit),
         counter(hits_total),
         counter(misses_total),
+        counter(items_evicted_total),
         debug_counter(items_inserted_total),
         debug_counter(items_removed_total),
         debug_counter(items_expired_total),
@@ -224,7 +226,12 @@ where
         telemetry.weight_limit().set(capacity as f64);
 
         // Configure expiration if enabled.
-        let mut expiration_builder = ExpirationBuilder::new();
+        let eviction_counter = if self.telemetry_enabled {
+            telemetry.items_evicted_total().clone()
+        } else {
+            Counter::noop()
+        };
+        let mut expiration_builder = ExpirationBuilder::new(eviction_counter);
         if let Some(time_to_idle) = self.idle_period {
             expiration_builder = expiration_builder.with_time_to_idle(time_to_idle);
         }