chore: add over window executor cache metrics (#12092)

Signed-off-by: Richard Chien <[email protected]>

Author: stdrc

docker/dashboards/risingwave-dev-dashboard.json

@@ -1059,6 +1059,10 @@ def section_streaming_actors(outer_panels):
                             "materialize executor cache miss ratio - table {{table_id}} actor {{actor_id}}  {{instance}}",
                         ),
 
+                        panels.target(
+                            f"(sum(rate({metric('stream_over_window_cache_miss_count')}[$__rate_interval])) by (table_id, actor_id) ) / (sum(rate({metric('stream_over_window_cache_lookup_count')}[$__rate_interval])) by (table_id, actor_id))",
+                            "Over window cache miss ratio - table {{table_id}} actor {{actor_id}} ",
+                        ),
                     ],
                 ),
                 panels.timeseries_actor_latency(
@@ -1231,6 +1235,25 @@ def section_streaming_actors(outer_panels):
 
                     ],
                 ),
+
+                panels.timeseries_actor_ops(
+                    "Over Window Executor Cache",
+                    "",
+                    [
+                        panels.target(
+                            f"rate({table_metric('stream_over_window_cached_entry_count')}[$__rate_interval])",
+                            "cached entry count - table {{table_id}} - actor {{actor_id}}   {{instance}}",
+                        ),
+                        panels.target(
+                            f"rate({table_metric('stream_over_window_cache_lookup_count')}[$__rate_interval])",
+                            "cache lookup count - table {{table_id}} - actor {{actor_id}}   {{instance}}",
+                        ),
+                        panels.target(
+                            f"rate({table_metric('stream_over_window_cache_miss_count')}[$__rate_interval])",
+                            "cache miss count - table {{table_id}} - actor {{actor_id}}   {{instance}}",
+                        ),
+                    ]
+                ),
             ],
         )
     ]

grafana/risingwave-dev-dashboard.dashboard.py

@@ -99,6 +99,11 @@ pub struct StreamingMetrics {
     pub arrangement_backfill_snapshot_read_row_count: GenericCounterVec<AtomicU64>,
     pub arrangement_backfill_upstream_output_row_count: GenericCounterVec<AtomicU64>,
 
+    // Over Window
+    pub over_window_cached_entry_count: GenericGaugeVec<AtomicI64>,
+    pub over_window_cache_lookup_count: GenericCounterVec<AtomicU64>,
+    pub over_window_cache_miss_count: GenericCounterVec<AtomicU64>,
+
     /// The duration from receipt of barrier to all actors collection.
     /// And the max of all node `barrier_inflight_latency` is the latency for a barrier
     /// to flow through the graph.
@@ -595,6 +600,30 @@ impl StreamingMetrics {
             )
             .unwrap();
 
+        let over_window_cached_entry_count = register_int_gauge_vec_with_registry!(
+            "stream_over_window_cached_entry_count",
+            "Total entry (partition) count in over window executor cache",
+            &["table_id", "actor_id"],
+            registry
+        )
+        .unwrap();
+
+        let over_window_cache_lookup_count = register_int_counter_vec_with_registry!(
+            "stream_over_window_cache_lookup_count",
+            "Over window executor cache lookup count",
+            &["table_id", "actor_id"],
+            registry
+        )
+        .unwrap();
+
+        let over_window_cache_miss_count = register_int_counter_vec_with_registry!(
+            "stream_over_window_cache_miss_count",
+            "Over window executor cache miss count",
+            &["table_id", "actor_id"],
+            registry
+        )
+        .unwrap();
+
         let opts = histogram_opts!(
             "stream_barrier_inflight_duration_seconds",
             "barrier_inflight_latency",
@@ -777,6 +806,9 @@ impl StreamingMetrics {
             backfill_upstream_output_row_count,
             arrangement_backfill_snapshot_read_row_count,
             arrangement_backfill_upstream_output_row_count,
+            over_window_cached_entry_count,
+            over_window_cache_lookup_count,
+            over_window_cache_miss_count,
             barrier_inflight_latency,
             barrier_sync_latency,
             barrier_manager_progress,

grafana/risingwave-dev-dashboard.json

@@ -15,6 +15,7 @@
 use std::collections::{btree_map, BTreeMap, HashSet};
 use std::marker::PhantomData;
 use std::ops::RangeInclusive;
+use std::sync::Arc;
 
 use futures::StreamExt;
 use futures_async_stream::try_stream;
@@ -41,6 +42,7 @@ use super::over_partition::{
 use crate::cache::{new_unbounded, ManagedLruCache};
 use crate::common::metrics::MetricsInfo;
 use crate::common::StreamChunkBuilder;
+use crate::executor::monitor::StreamingMetrics;
 use crate::executor::over_window::delta_btree_map::PositionType;
 use crate::executor::test_utils::prelude::StateTable;
 use crate::executor::{
@@ -73,6 +75,7 @@ struct ExecutorInner<S: StateStore> {
 
     state_table: StateTable<S>,
     watermark_epoch: AtomicU64Ref,
+    metrics: Arc<StreamingMetrics>,
 
     /// The maximum size of the chunk produced by executor at a time.
     chunk_size: usize,
@@ -84,9 +87,16 @@ struct ExecutionVars<S: StateStore> {
     cached_partitions: ManagedLruCache<OwnedRow, PartitionCache>,
     /// partition key => recently accessed range.
     recently_accessed_ranges: BTreeMap<DefaultOrdered<OwnedRow>, RangeInclusive<StateKey>>,
+    stats: ExecutionStats,
     _phantom: PhantomData<S>,
 }
 
+#[derive(Default)]
+struct ExecutionStats {
+    cache_miss: u64,
+    cache_lookup: u64,
+}
+
 impl<S: StateStore> Executor for OverWindowExecutor<S> {
     fn execute(self: Box<Self>) -> crate::executor::BoxedMessageStream {
         self.executor_inner().boxed()
@@ -146,6 +156,7 @@ pub struct OverWindowExecutorArgs<S: StateStore> {
 
     pub state_table: StateTable<S>,
     pub watermark_epoch: AtomicU64Ref,
+    pub metrics: Arc<StreamingMetrics>,
 
     pub chunk_size: usize,
     pub cache_policy: CachePolicy,
@@ -196,6 +207,7 @@ impl<S: StateStore> OverWindowExecutor<S> {
                 input_schema_len: input_info.schema.len(),
                 state_table: args.state_table,
                 watermark_epoch: args.watermark_epoch,
+                metrics: args.metrics,
                 chunk_size: args.chunk_size,
                 cache_policy,
             },
@@ -348,7 +360,9 @@ impl<S: StateStore> OverWindowExecutor<S> {
 
         // Build final changes partition by partition.
         for (part_key, delta) in deltas {
+            vars.stats.cache_lookup += 1;
             if !vars.cached_partitions.contains(&part_key.0) {
+                vars.stats.cache_miss += 1;
                 vars.cached_partitions
                     .put(part_key.0.clone(), new_empty_partition_cache());
             }
@@ -593,6 +607,7 @@ impl<S: StateStore> OverWindowExecutor<S> {
         let mut vars = ExecutionVars {
             cached_partitions: new_unbounded(this.watermark_epoch.clone(), metrics_info),
             recently_accessed_ranges: Default::default(),
+            stats: Default::default(),
             _phantom: PhantomData::<S>,
         };
 
@@ -622,6 +637,24 @@ impl<S: StateStore> OverWindowExecutor<S> {
                     this.state_table.commit(barrier.epoch).await?;
                     vars.cached_partitions.evict();
 
+                    {
+                        // update metrics
+                        let actor_id_str = this.actor_ctx.id.to_string();
+                        let table_id_str = this.state_table.table_id().to_string();
+                        this.metrics
+                            .over_window_cached_entry_count
+                            .with_label_values(&[&table_id_str, &actor_id_str])
+                            .set(vars.cached_partitions.len() as _);
+                        this.metrics
+                            .over_window_cache_lookup_count
+                            .with_label_values(&[&table_id_str, &actor_id_str])
+                            .inc_by(std::mem::take(&mut vars.stats.cache_lookup));
+                        this.metrics
+                            .over_window_cache_miss_count
+                            .with_label_values(&[&table_id_str, &actor_id_str])
+                            .inc_by(std::mem::take(&mut vars.stats.cache_miss));
+                    }
+
                     if let Some(vnode_bitmap) = barrier.as_update_vnode_bitmap(this.actor_ctx.id) {
                         let (_, cache_may_stale) =
                             this.state_table.update_vnode_bitmap(vnode_bitmap);

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

@@ -73,6 +73,7 @@ impl ExecutorBuilder for OverWindowExecutorBuilder {
             order_key_order_types,
             state_table,
             watermark_epoch: stream.get_watermark_epoch(),
+            metrics: params.executor_stats,
             chunk_size: params.env.config().developer.chunk_size,
             cache_policy: OverWindowCachePolicy::from_protobuf(node.get_cache_policy()?),
         })

src/stream/src/executor/over_window/general.rs

@@ -17,6 +17,7 @@ use risingwave_expr::agg::{AggArgs, AggKind};
 use risingwave_expr::function::window::{
     Frame, FrameBound, FrameExclusion, WindowFuncCall, WindowFuncKind,
 };
+use risingwave_stream::executor::monitor::StreamingMetrics;
 use risingwave_stream::executor::{OverWindowExecutor, OverWindowExecutorArgs};
 
 use crate::prelude::*;
@@ -78,6 +79,7 @@ async fn create_executor<S: StateStore>(
         order_key_order_types,
         state_table,
         watermark_epoch: Arc::new(AtomicU64::new(0)),
+        metrics: Arc::new(StreamingMetrics::unused()),
         chunk_size: 1024,
         cache_policy: OverWindowCachePolicy::Recent,
     });