feat(metrics): add metrics for streaming actor lru cache eviction

Cargo.toml

@@ -149,7 +149,7 @@ arrow-schema-deltalake = { package = "arrow-schema", version = "48.0.1" }
 deltalake = { git = "https://github.com/risingwavelabs/delta-rs", rev = "5c2dccd4640490202ffe98adbd13b09cef8e007b", features = [
   "s3-no-concurrent-write",
 ] }
-lru = { git = "https://github.com/risingwavelabs/lru-rs.git", rev = "95f347b" }
+lru = { git = "https://github.com/risingwavelabs/lru-rs.git", rev = "2682b85" }
 parquet = "50"
 thiserror-ext = "0.0.11"
 tikv-jemalloc-ctl = { git = "https://github.com/risingwavelabs/jemallocator.git", rev = "64a2d9" }

grafana/risingwave-dev-dashboard.dashboard.py

@@ -1067,6 +1067,20 @@ def section_streaming_actors(outer_panels):
                         ),
                     ],
                 ),
+                panels.timeseries_bytes(
+                    "Executor Evicted Cache Memory",
+                    "The operator-level evicted memory statistics collected by each LRU cache",
+                    [
+                        panels.target(
+                            f"sum({metric('stream_memory_evicted')}) by (table_id, epoch)",
+                            "table {{table_id}} epoch: {{epoch}}",
+                        ),
+                        panels.target_hidden(
+                            f"{metric('stream_memory_evicted')}",
+                            "table {{table_id}} actor {{actor_id}} epoch: {{epoch}}",
+                        ),
+                    ],
+                ),
                 panels.timeseries_bytes(
                     "Executor Cache Memory Usage of Materialized Views",
                     "Memory usage aggregated by materialized views",

src/common/estimate_size/src/collections/lru.rs

@@ -40,7 +40,7 @@ impl<K: Hash + Eq + EstimateSize, V: EstimateSize, S: BuildHasher, A: Clone + Al
 
     /// Evict epochs lower than the watermark
     pub fn evict_by_epoch(&mut self, epoch: u64) {
-        while let Some((key, value)) = self.inner.pop_lru_by_epoch(epoch) {
+        while let Some((key, value, _)) = self.inner.pop_lru_by_epoch(epoch) {
             self.kv_heap_size.sub(&key, &value);
         }
     }

src/stream/src/cache/managed_lru.rs

@@ -21,7 +21,7 @@ use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
 
 use lru::{DefaultHasher, LruCache};
-use risingwave_common::metrics::LabelGuardedIntGauge;
+use risingwave_common::metrics::{LabelGuardedIntGauge, LabelGuardedIntGaugeVec};
 use risingwave_common::util::epoch::Epoch;
 use risingwave_common_estimate_size::EstimateSize;
 
@@ -40,10 +40,11 @@ pub struct ManagedLruCache<K, V, S = DefaultHasher, A: Clone + Allocator = Globa
     kv_heap_size: usize,
     /// The metrics of memory usage
     memory_usage_metrics: LabelGuardedIntGauge<3>,
+    /// The metrics of evicted memory by epoch.
+    memory_evicted_metrics: LabelGuardedIntGaugeVec<3>,
     // The metrics of evicted watermark time
     lru_evicted_watermark_time_ms: LabelGuardedIntGauge<3>,
     // Metrics info
-    #[expect(dead_code)]
     metrics_info: MetricsInfo,
     /// The size reported last time
     last_reported_size_bytes: usize,
@@ -73,6 +74,8 @@ impl<K: Hash + Eq + EstimateSize, V: EstimateSize, S: BuildHasher, A: Clone + Al
             ]);
         memory_usage_metrics.set(0.into());
 
+        let memory_evicted_metrics = metrics_info.metrics.stream_memory_evicted.clone();
+
         let lru_evicted_watermark_time_ms = metrics_info
             .metrics
             .lru_evicted_watermark_time_ms
@@ -87,6 +90,7 @@ impl<K: Hash + Eq + EstimateSize, V: EstimateSize, S: BuildHasher, A: Clone + Al
             watermark_epoch,
             kv_heap_size: 0,
             memory_usage_metrics,
+            memory_evicted_metrics,
             lru_evicted_watermark_time_ms,
             metrics_info,
             last_reported_size_bytes: 0,
@@ -106,8 +110,32 @@ impl<K: Hash + Eq + EstimateSize, V: EstimateSize, S: BuildHasher, A: Clone + Al
 
     /// Evict epochs lower than the watermark
     fn evict_by_epoch(&mut self, epoch: u64) {
-        while let Some((key, value)) = self.inner.pop_lru_by_epoch(epoch) {
-            self.kv_heap_size_dec(key.estimated_size() + value.estimated_size());
+        let report = |lru: &Self, epoch: u64, evicted: usize| {
+            lru.memory_evicted_metrics
+                .with_guarded_label_values(&[
+                    &lru.metrics_info.table_id,
+                    &lru.metrics_info.actor_id,
+                    &epoch.to_string(),
+                ])
+                .set(evicted as _);
+        };
+
+        let mut last_epoch = 0; // real epoch must be greater than 0
+        let mut evicted = 0;
+
+        while let Some((key, value, e)) = self.inner.pop_lru_by_epoch(epoch) {
+            let charge = key.estimated_size() + value.estimated_size();
+            self.kv_heap_size_dec(charge);
+            // The popped epoch must be monotonically decreasing.
+            if e != last_epoch {
+                report(self, last_epoch, evicted);
+                last_epoch = e;
+                evicted = 0;
+            }
+            evicted += charge;
+        }
+        if evicted > 0 {
+            report(self, last_epoch, evicted);
         }
         self.report_evicted_watermark_time(epoch);
     }

src/stream/src/executor/monitor/streaming_stats.rs

@@ -186,6 +186,7 @@ pub struct StreamingMetrics {
 
     // Memory
     pub stream_memory_usage: LabelGuardedIntGaugeVec<3>,
+    pub stream_memory_evicted: LabelGuardedIntGaugeVec<3>,
 }
 
 pub static GLOBAL_STREAMING_METRICS: OnceLock<StreamingMetrics> = OnceLock::new();
@@ -998,6 +999,14 @@ impl StreamingMetrics {
         )
         .unwrap();
 
+        let stream_memory_evicted = register_guarded_int_gauge_vec_with_registry!(
+            "stream_memory_evicted",
+            "Memory evicted for stream executors",
+            &["table_id", "actor_id", "epoch"],
+            registry
+        )
+        .unwrap();
+
         let iceberg_write_qps = register_guarded_int_counter_vec_with_registry!(
             "iceberg_write_qps",
             "The qps of iceberg writer",
@@ -1139,6 +1148,7 @@ impl StreamingMetrics {
             materialize_cache_hit_count,
             materialize_cache_total_count,
             stream_memory_usage,
+            stream_memory_evicted,
         }
     }