Skip to main content

risingwave_stream/executor/monitor/
streaming_stats.rs

1// Copyright 2022 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::sync::OnceLock;
16
17use prometheus::{
18    Histogram, IntCounter, IntGauge, Registry, exponential_buckets, histogram_opts,
19    register_histogram_with_registry, register_int_counter_with_registry,
20    register_int_gauge_with_registry,
21};
22use risingwave_common::catalog::TableId;
23use risingwave_common::config::MetricLevel;
24use risingwave_common::metrics::{
25    LabelGuardedHistogramVec, LabelGuardedIntCounter, LabelGuardedIntCounterVec,
26    LabelGuardedIntGauge, LabelGuardedIntGaugeVec, MetricVecRelabelExt,
27    RelabeledGuardedHistogramVec, RelabeledGuardedIntCounterVec, RelabeledGuardedIntGaugeVec,
28};
29use risingwave_common::monitor::GLOBAL_METRICS_REGISTRY;
30use risingwave_common::monitor::in_mem::CountMap;
31use risingwave_common::{
32    register_guarded_histogram_vec_with_registry, register_guarded_int_counter_vec_with_registry,
33    register_guarded_int_gauge_vec_with_registry,
34};
35use risingwave_connector::sink::catalog::SinkId;
36use risingwave_pb::id::ExecutorId;
37
38use crate::common::log_store_impl::kv_log_store::{
39    REWIND_BACKOFF_MULTIPLIER, REWIND_INITIAL_DELAY, REWIND_MAX_DELAY,
40};
41use crate::executor::prelude::ActorId;
42use crate::task::FragmentId;
43
44#[derive(Clone)]
45pub struct StreamingMetrics {
46    pub level: MetricLevel,
47
48    // Executor metrics (disabled by default)
49    pub executor_row_count: RelabeledGuardedIntCounterVec,
50
51    // Profiling Metrics:
52    // Aggregated per operator rather than per actor.
53    // These are purely in-memory, never collected by prometheus.
54    pub mem_stream_node_output_row_count: CountMap<ExecutorId>,
55    pub mem_stream_node_output_blocking_duration_ns: CountMap<ExecutorId>,
56
57    // Streaming actor metrics from tokio (disabled by default)
58    actor_scheduled_duration: RelabeledGuardedIntCounterVec,
59    actor_scheduled_cnt: RelabeledGuardedIntCounterVec,
60    actor_poll_duration: RelabeledGuardedIntCounterVec,
61    actor_poll_cnt: RelabeledGuardedIntCounterVec,
62    actor_idle_duration: RelabeledGuardedIntCounterVec,
63    actor_idle_cnt: RelabeledGuardedIntCounterVec,
64
65    // Streaming actor
66    pub actor_count: LabelGuardedIntGaugeVec,
67    pub actor_in_record_cnt: RelabeledGuardedIntCounterVec,
68    pub actor_out_record_cnt: RelabeledGuardedIntCounterVec,
69    pub fragment_channel_buffered_bytes: LabelGuardedIntGaugeVec,
70    pub actor_current_epoch: RelabeledGuardedIntGaugeVec,
71    pub project_expr_inflight_window_size: LabelGuardedIntGaugeVec,
72
73    // Source
74    pub source_output_row_count: LabelGuardedIntCounterVec,
75    pub source_split_change_count: LabelGuardedIntCounterVec,
76    pub source_backfill_row_count: LabelGuardedIntCounterVec,
77
78    // Sink
79    sink_input_row_count: LabelGuardedIntCounterVec,
80    sink_input_bytes: LabelGuardedIntCounterVec,
81    sink_chunk_buffer_size: LabelGuardedIntGaugeVec,
82
83    // Exchange (see also `compute::ExchangeServiceMetrics`)
84    pub exchange_frag_recv_size: LabelGuardedIntCounterVec,
85
86    // Streaming Merge (We breakout this metric from `barrier_align_duration` because
87    // the alignment happens on different levels)
88    pub merge_barrier_align_duration: RelabeledGuardedIntCounterVec,
89
90    // Backpressure
91    pub actor_output_buffer_blocking_duration_ns: RelabeledGuardedIntCounterVec,
92    actor_input_buffer_blocking_duration_ns: RelabeledGuardedIntCounterVec,
93
94    // Streaming Join
95    pub join_lookup_miss_count: LabelGuardedIntCounterVec,
96    pub join_lookup_total_count: LabelGuardedIntCounterVec,
97    pub join_insert_cache_miss_count: LabelGuardedIntCounterVec,
98    pub join_actor_input_waiting_duration_ns: LabelGuardedIntCounterVec,
99    pub join_match_duration_ns: LabelGuardedIntCounterVec,
100    pub join_cached_entry_count: LabelGuardedIntGaugeVec,
101    pub join_matched_join_keys: RelabeledGuardedHistogramVec,
102
103    // Streaming Join, Streaming Dynamic Filter and Streaming Union
104    pub barrier_align_duration: RelabeledGuardedIntCounterVec,
105
106    // Streaming Aggregation
107    agg_lookup_miss_count: LabelGuardedIntCounterVec,
108    agg_total_lookup_count: LabelGuardedIntCounterVec,
109    agg_cached_entry_count: LabelGuardedIntGaugeVec,
110    agg_chunk_lookup_miss_count: LabelGuardedIntCounterVec,
111    agg_chunk_total_lookup_count: LabelGuardedIntCounterVec,
112    agg_dirty_groups_count: LabelGuardedIntGaugeVec,
113    agg_dirty_groups_heap_size: LabelGuardedIntGaugeVec,
114    agg_distinct_cache_miss_count: LabelGuardedIntCounterVec,
115    agg_distinct_total_cache_count: LabelGuardedIntCounterVec,
116    agg_distinct_cached_entry_count: LabelGuardedIntGaugeVec,
117    agg_state_cache_lookup_count: LabelGuardedIntCounterVec,
118    agg_state_cache_miss_count: LabelGuardedIntCounterVec,
119
120    // Streaming TopN
121    group_top_n_cache_miss_count: LabelGuardedIntCounterVec,
122    group_top_n_total_query_cache_count: LabelGuardedIntCounterVec,
123    group_top_n_cached_entry_count: LabelGuardedIntGaugeVec,
124    // TODO(rc): why not just use the above three?
125    group_top_n_appendonly_cache_miss_count: LabelGuardedIntCounterVec,
126    group_top_n_appendonly_total_query_cache_count: LabelGuardedIntCounterVec,
127    group_top_n_appendonly_cached_entry_count: LabelGuardedIntGaugeVec,
128
129    // Lookup executor
130    lookup_cache_miss_count: LabelGuardedIntCounterVec,
131    lookup_total_query_cache_count: LabelGuardedIntCounterVec,
132    lookup_cached_entry_count: LabelGuardedIntGaugeVec,
133
134    // temporal join
135    temporal_join_cache_miss_count: LabelGuardedIntCounterVec,
136    temporal_join_total_query_cache_count: LabelGuardedIntCounterVec,
137    temporal_join_cached_entry_count: LabelGuardedIntGaugeVec,
138
139    // Backfill
140    backfill_snapshot_read_row_count: LabelGuardedIntCounterVec,
141    backfill_upstream_output_row_count: LabelGuardedIntCounterVec,
142
143    // CDC Backfill
144    cdc_backfill_snapshot_read_row_count: LabelGuardedIntCounterVec,
145    cdc_backfill_upstream_output_row_count: LabelGuardedIntCounterVec,
146
147    // Snapshot Backfill
148    pub(crate) snapshot_backfill_consume_row_count: LabelGuardedIntCounterVec,
149
150    // Over Window
151    over_window_cached_entry_count: LabelGuardedIntGaugeVec,
152    over_window_cache_lookup_count: LabelGuardedIntCounterVec,
153    over_window_cache_miss_count: LabelGuardedIntCounterVec,
154    over_window_range_cache_entry_count: LabelGuardedIntGaugeVec,
155    over_window_range_cache_lookup_count: LabelGuardedIntCounterVec,
156    over_window_range_cache_left_miss_count: LabelGuardedIntCounterVec,
157    over_window_range_cache_right_miss_count: LabelGuardedIntCounterVec,
158    over_window_accessed_entry_count: LabelGuardedIntCounterVec,
159    over_window_compute_count: LabelGuardedIntCounterVec,
160    over_window_same_output_count: LabelGuardedIntCounterVec,
161
162    /// The duration from receipt of barrier to all actors collection.
163    /// The max of all nodes' `barrier_inflight_latency` for a partial graph is the latency for a
164    /// barrier to flow through that partial graph.
165    pub barrier_inflight_latency: LabelGuardedHistogramVec,
166    /// The duration of sync to storage.
167    pub barrier_sync_latency: LabelGuardedHistogramVec,
168    pub barrier_batch_size: Histogram,
169    /// The progress made by the earliest in-flight barriers in the local barrier manager.
170    pub barrier_manager_progress: LabelGuardedIntCounterVec,
171
172    pub kv_log_store_storage_write_count: LabelGuardedIntCounterVec,
173    pub kv_log_store_storage_write_size: LabelGuardedIntCounterVec,
174    pub kv_log_store_rewind_count: LabelGuardedIntCounterVec,
175    pub kv_log_store_rewind_delay: LabelGuardedHistogramVec,
176    pub kv_log_store_storage_read_count: LabelGuardedIntCounterVec,
177    pub kv_log_store_storage_read_size: LabelGuardedIntCounterVec,
178    pub kv_log_store_buffer_unconsumed_item_count: LabelGuardedIntGaugeVec,
179    pub kv_log_store_buffer_unconsumed_row_count: LabelGuardedIntGaugeVec,
180    pub kv_log_store_buffer_unconsumed_epoch_count: LabelGuardedIntGaugeVec,
181    pub kv_log_store_buffer_unconsumed_min_epoch: LabelGuardedIntGaugeVec,
182    pub kv_log_store_buffer_memory_bytes: LabelGuardedIntGaugeVec,
183
184    pub crossdb_last_consumed_min_epoch: LabelGuardedIntGaugeVec,
185
186    pub sync_kv_log_store_read_count: LabelGuardedIntCounterVec,
187    pub sync_kv_log_store_read_size: LabelGuardedIntCounterVec,
188    pub sync_kv_log_store_write_pause_duration_ns: LabelGuardedIntCounterVec,
189    pub sync_kv_log_store_state: LabelGuardedIntCounterVec,
190    pub sync_kv_log_store_wait_next_poll_ns: LabelGuardedIntCounterVec,
191    pub sync_kv_log_store_storage_write_count: LabelGuardedIntCounterVec,
192    pub sync_kv_log_store_storage_write_size: LabelGuardedIntCounterVec,
193    pub sync_kv_log_store_buffer_unconsumed_item_count: LabelGuardedIntGaugeVec,
194    pub sync_kv_log_store_buffer_unconsumed_row_count: LabelGuardedIntGaugeVec,
195    pub sync_kv_log_store_buffer_unconsumed_epoch_count: LabelGuardedIntGaugeVec,
196    pub sync_kv_log_store_buffer_unconsumed_min_epoch: LabelGuardedIntGaugeVec,
197    pub sync_kv_log_store_buffer_memory_bytes: LabelGuardedIntGaugeVec,
198
199    // Memory management
200    pub lru_runtime_loop_count: IntCounter,
201    pub lru_latest_sequence: IntGauge,
202    pub lru_watermark_sequence: IntGauge,
203    pub lru_eviction_policy: IntGauge,
204    pub jemalloc_allocated_bytes: IntGauge,
205    pub jemalloc_active_bytes: IntGauge,
206    pub jemalloc_resident_bytes: IntGauge,
207    pub jemalloc_metadata_bytes: IntGauge,
208    pub jvm_allocated_bytes: IntGauge,
209    pub jvm_active_bytes: IntGauge,
210    pub stream_memory_usage: RelabeledGuardedIntGaugeVec,
211
212    // Materialized view
213    materialize_cache_hit_count: RelabeledGuardedIntCounterVec,
214    materialize_data_exist_count: RelabeledGuardedIntCounterVec,
215    materialize_cache_total_count: RelabeledGuardedIntCounterVec,
216    materialize_input_row_count: RelabeledGuardedIntCounterVec,
217    pub materialize_current_epoch: RelabeledGuardedIntGaugeVec,
218
219    // PostgreSQL CDC LSN monitoring
220    pub pg_cdc_state_table_lsn: LabelGuardedIntGaugeVec,
221    pub pg_cdc_jni_commit_offset_lsn: LabelGuardedIntGaugeVec,
222
223    // MySQL CDC binlog monitoring
224    pub mysql_cdc_state_binlog_file_seq: LabelGuardedIntGaugeVec,
225    pub mysql_cdc_state_binlog_position: LabelGuardedIntGaugeVec,
226
227    // SQL Server CDC LSN monitoring
228    pub sqlserver_cdc_state_change_lsn: LabelGuardedIntGaugeVec,
229    pub sqlserver_cdc_state_commit_lsn: LabelGuardedIntGaugeVec,
230    pub sqlserver_cdc_jni_commit_offset_lsn: LabelGuardedIntGaugeVec,
231
232    // Gap Fill
233    pub gap_fill_generated_rows_count: RelabeledGuardedIntCounterVec,
234
235    // State Table
236    pub state_table_iter_count: RelabeledGuardedIntCounterVec,
237    pub state_table_get_count: RelabeledGuardedIntCounterVec,
238    pub state_table_iter_vnode_pruned_count: RelabeledGuardedIntCounterVec,
239    pub state_table_get_vnode_pruned_count: RelabeledGuardedIntCounterVec,
240}
241
242pub static GLOBAL_STREAMING_METRICS: OnceLock<StreamingMetrics> = OnceLock::new();
243
244fn latency_buckets(max: f64, count: usize) -> Vec<f64> {
245    const MIN: f64 = 0.1;
246
247    assert!(count > 1);
248    let factor = (max / MIN).powf(1.0 / (count - 1) as f64);
249    let mut buckets = exponential_buckets(MIN, factor, count).unwrap();
250    *buckets.last_mut().unwrap() = max;
251    buckets
252}
253
254pub fn global_streaming_metrics(metric_level: MetricLevel) -> StreamingMetrics {
255    GLOBAL_STREAMING_METRICS
256        .get_or_init(|| StreamingMetrics::new(&GLOBAL_METRICS_REGISTRY, metric_level))
257        .clone()
258}
259
260impl StreamingMetrics {
261    pub fn new(registry: &Registry, level: MetricLevel) -> Self {
262        let executor_row_count = register_guarded_int_counter_vec_with_registry!(
263            "stream_executor_row_count",
264            "Total number of rows that have been output from each executor",
265            &["actor_id", "fragment_id", "executor_identity"],
266            registry
267        )
268        .unwrap()
269        .relabel_debug_1(level);
270
271        let stream_node_output_row_count = CountMap::new();
272        let stream_node_output_blocking_duration_ns = CountMap::new();
273
274        let source_output_row_count = register_guarded_int_counter_vec_with_registry!(
275            "stream_source_output_rows_counts",
276            "Total number of rows that have been output from source",
277            &["source_id", "source_name", "actor_id", "fragment_id"],
278            registry
279        )
280        .unwrap();
281
282        let source_split_change_count = register_guarded_int_counter_vec_with_registry!(
283            "stream_source_split_change_event_count",
284            "Total number of split change events that have been operated by source",
285            &["source_id", "source_name", "actor_id", "fragment_id"],
286            registry
287        )
288        .unwrap();
289
290        let source_backfill_row_count = register_guarded_int_counter_vec_with_registry!(
291            "stream_source_backfill_rows_counts",
292            "Total number of rows that have been backfilled for source",
293            &["source_id", "source_name", "actor_id", "fragment_id"],
294            registry
295        )
296        .unwrap();
297
298        let sink_input_row_count = register_guarded_int_counter_vec_with_registry!(
299            "stream_sink_input_row_count",
300            "Total number of rows streamed into sink executors",
301            &["sink_id", "actor_id", "fragment_id"],
302            registry
303        )
304        .unwrap();
305
306        let sink_input_bytes = register_guarded_int_counter_vec_with_registry!(
307            "stream_sink_input_bytes",
308            "Total size of chunks streamed into sink executors",
309            &["sink_id", "actor_id", "fragment_id"],
310            registry
311        )
312        .unwrap();
313
314        let materialize_input_row_count = register_guarded_int_counter_vec_with_registry!(
315            "stream_mview_input_row_count",
316            "Total number of rows streamed into materialize executors",
317            &["actor_id", "table_id", "fragment_id"],
318            registry
319        )
320        .unwrap()
321        .relabel_debug_1(level);
322
323        let materialize_current_epoch = register_guarded_int_gauge_vec_with_registry!(
324            "stream_mview_current_epoch",
325            "The current epoch of the materialized executor",
326            &["actor_id", "table_id", "fragment_id"],
327            registry
328        )
329        .unwrap()
330        .relabel_debug_1(level);
331
332        let pg_cdc_state_table_lsn = register_guarded_int_gauge_vec_with_registry!(
333            "stream_pg_cdc_state_table_lsn",
334            "Current LSN value stored in PostgreSQL CDC state table",
335            &["source_id"],
336            registry,
337        )
338        .unwrap();
339
340        let pg_cdc_jni_commit_offset_lsn = register_guarded_int_gauge_vec_with_registry!(
341            "stream_pg_cdc_jni_commit_offset_lsn",
342            "LSN value when JNI commit offset is called for PostgreSQL CDC",
343            &["source_id"],
344            registry,
345        )
346        .unwrap();
347
348        let mysql_cdc_state_binlog_file_seq = register_guarded_int_gauge_vec_with_registry!(
349            "stream_mysql_cdc_state_binlog_file_seq",
350            "Current binlog file sequence number stored in MySQL CDC state table",
351            &["source_id"],
352            registry,
353        )
354        .unwrap();
355
356        let mysql_cdc_state_binlog_position = register_guarded_int_gauge_vec_with_registry!(
357            "stream_mysql_cdc_state_binlog_position",
358            "Current binlog position stored in MySQL CDC state table",
359            &["source_id"],
360            registry,
361        )
362        .unwrap();
363
364        let sqlserver_cdc_state_change_lsn = register_guarded_int_gauge_vec_with_registry!(
365            "stream_sqlserver_cdc_state_change_lsn",
366            "Current change_lsn value stored in SQL Server CDC state table",
367            &["source_id"],
368            registry,
369        )
370        .unwrap();
371
372        let sqlserver_cdc_state_commit_lsn = register_guarded_int_gauge_vec_with_registry!(
373            "stream_sqlserver_cdc_state_commit_lsn",
374            "Current commit_lsn value stored in SQL Server CDC state table",
375            &["source_id"],
376            registry,
377        )
378        .unwrap();
379
380        let sqlserver_cdc_jni_commit_offset_lsn = register_guarded_int_gauge_vec_with_registry!(
381            "stream_sqlserver_cdc_jni_commit_offset_lsn",
382            "LSN value when JNI commit offset is called for SQL Server CDC",
383            &["source_id"],
384            registry,
385        )
386        .unwrap();
387
388        let sink_chunk_buffer_size = register_guarded_int_gauge_vec_with_registry!(
389            "stream_sink_chunk_buffer_size",
390            "Total size of chunks buffered in a barrier",
391            &["sink_id", "actor_id", "fragment_id"],
392            registry
393        )
394        .unwrap();
395        let actor_output_buffer_blocking_duration_ns =
396            register_guarded_int_counter_vec_with_registry!(
397                "stream_actor_output_buffer_blocking_duration_ns",
398                "Total blocking duration (ns) of output buffer",
399                &["actor_id", "fragment_id", "downstream_fragment_id"],
400                registry
401            )
402            .unwrap()
403            // mask the first label `actor_id` if the level is less verbose than `Debug`
404            .relabel_debug_1(level);
405
406        let actor_input_buffer_blocking_duration_ns =
407            register_guarded_int_counter_vec_with_registry!(
408                "stream_actor_input_buffer_blocking_duration_ns",
409                "Total blocking duration (ns) of input buffer",
410                &["actor_id", "fragment_id", "upstream_fragment_id"],
411                registry
412            )
413            .unwrap()
414            // mask the first label `actor_id` if the level is less verbose than `Debug`
415            .relabel_debug_1(level);
416
417        let fragment_channel_buffered_bytes = register_guarded_int_gauge_vec_with_registry!(
418            "stream_fragment_channel_buffered_bytes",
419            "Estimated buffered bytes for actor channels by fragment",
420            &["fragment_id"],
421            registry
422        )
423        .unwrap();
424
425        let exchange_frag_recv_size = register_guarded_int_counter_vec_with_registry!(
426            "stream_exchange_frag_recv_size",
427            "Total size of messages that have been received from upstream Fragment",
428            &["up_fragment_id", "down_fragment_id"],
429            registry
430        )
431        .unwrap();
432
433        let actor_poll_duration = register_guarded_int_counter_vec_with_registry!(
434            "stream_actor_poll_duration",
435            "tokio's metrics",
436            &["actor_id", "fragment_id"],
437            registry
438        )
439        .unwrap()
440        .relabel_debug_1(level);
441
442        let actor_poll_cnt = register_guarded_int_counter_vec_with_registry!(
443            "stream_actor_poll_cnt",
444            "tokio's metrics",
445            &["actor_id", "fragment_id"],
446            registry
447        )
448        .unwrap()
449        .relabel_debug_1(level);
450
451        let actor_scheduled_duration = register_guarded_int_counter_vec_with_registry!(
452            "stream_actor_scheduled_duration",
453            "tokio's metrics",
454            &["actor_id", "fragment_id"],
455            registry
456        )
457        .unwrap()
458        .relabel_debug_1(level);
459
460        let actor_scheduled_cnt = register_guarded_int_counter_vec_with_registry!(
461            "stream_actor_scheduled_cnt",
462            "tokio's metrics",
463            &["actor_id", "fragment_id"],
464            registry
465        )
466        .unwrap()
467        .relabel_debug_1(level);
468
469        let actor_idle_duration = register_guarded_int_counter_vec_with_registry!(
470            "stream_actor_idle_duration",
471            "tokio's metrics",
472            &["actor_id", "fragment_id"],
473            registry
474        )
475        .unwrap()
476        .relabel_debug_1(level);
477
478        let actor_idle_cnt = register_guarded_int_counter_vec_with_registry!(
479            "stream_actor_idle_cnt",
480            "tokio's metrics",
481            &["actor_id", "fragment_id"],
482            registry
483        )
484        .unwrap()
485        .relabel_debug_1(level);
486
487        let actor_in_record_cnt = register_guarded_int_counter_vec_with_registry!(
488            "stream_actor_in_record_cnt",
489            "Total number of rows actor received",
490            &["actor_id", "fragment_id", "upstream_fragment_id"],
491            registry
492        )
493        .unwrap()
494        .relabel_debug_1(level);
495
496        let actor_out_record_cnt = register_guarded_int_counter_vec_with_registry!(
497            "stream_actor_out_record_cnt",
498            "Total number of rows actor sent",
499            &["actor_id", "fragment_id"],
500            registry
501        )
502        .unwrap()
503        .relabel_debug_1(level);
504
505        let actor_current_epoch = register_guarded_int_gauge_vec_with_registry!(
506            "stream_actor_current_epoch",
507            "Current epoch of actor",
508            &["actor_id", "fragment_id"],
509            registry
510        )
511        .unwrap()
512        .relabel_debug_1(level);
513
514        let project_expr_inflight_window_size = register_guarded_int_gauge_vec_with_registry!(
515            "stream_project_expr_inflight_window_size",
516            "Number of messages waiting in ProjectExecutor's ordered projection window",
517            &["actor_id", "fragment_id"],
518            registry
519        )
520        .unwrap();
521
522        let actor_count = register_guarded_int_gauge_vec_with_registry!(
523            "stream_actor_count",
524            "Total number of actors (parallelism)",
525            &["fragment_id"],
526            registry
527        )
528        .unwrap();
529
530        let merge_barrier_align_duration = register_guarded_int_counter_vec_with_registry!(
531            "stream_merge_barrier_align_duration_ns",
532            "Total merge barrier alignment duration (ns)",
533            &["actor_id", "fragment_id"],
534            registry
535        )
536        .unwrap()
537        .relabel_debug_1(level);
538
539        let join_lookup_miss_count = register_guarded_int_counter_vec_with_registry!(
540            "stream_join_lookup_miss_count",
541            "Join executor lookup miss duration",
542            &["side", "join_table_id", "actor_id", "fragment_id"],
543            registry
544        )
545        .unwrap();
546
547        let join_lookup_total_count = register_guarded_int_counter_vec_with_registry!(
548            "stream_join_lookup_total_count",
549            "Join executor lookup total operation",
550            &["side", "join_table_id", "actor_id", "fragment_id"],
551            registry
552        )
553        .unwrap();
554
555        let join_insert_cache_miss_count = register_guarded_int_counter_vec_with_registry!(
556            "stream_join_insert_cache_miss_count",
557            "Join executor cache miss when insert operation",
558            &["side", "join_table_id", "actor_id", "fragment_id"],
559            registry
560        )
561        .unwrap();
562
563        let join_actor_input_waiting_duration_ns = register_guarded_int_counter_vec_with_registry!(
564            "stream_join_actor_input_waiting_duration_ns",
565            "Total waiting duration (ns) of input buffer of join actor",
566            &["actor_id", "fragment_id"],
567            registry
568        )
569        .unwrap();
570
571        let join_match_duration_ns = register_guarded_int_counter_vec_with_registry!(
572            "stream_join_match_duration_ns",
573            "Matching duration for each side",
574            &["actor_id", "fragment_id", "side"],
575            registry
576        )
577        .unwrap();
578
579        let barrier_align_duration = register_guarded_int_counter_vec_with_registry!(
580            "stream_barrier_align_duration_ns",
581            "Duration of join align barrier",
582            &["actor_id", "fragment_id", "wait_side", "executor"],
583            registry
584        )
585        .unwrap()
586        .relabel_debug_1(level);
587
588        let join_cached_entry_count = register_guarded_int_gauge_vec_with_registry!(
589            "stream_join_cached_entry_count",
590            "Number of cached entries in streaming join operators",
591            &["actor_id", "fragment_id", "side"],
592            registry
593        )
594        .unwrap();
595
596        let join_matched_join_keys_opts = histogram_opts!(
597            "stream_join_matched_join_keys",
598            "The number of keys matched in the opposite side",
599            exponential_buckets(16.0, 2.0, 28).unwrap() // max 2^31
600        );
601
602        let join_matched_join_keys = register_guarded_histogram_vec_with_registry!(
603            join_matched_join_keys_opts,
604            &["actor_id", "fragment_id", "table_id"],
605            registry
606        )
607        .unwrap()
608        .relabel_debug_1(level);
609
610        let agg_lookup_miss_count = register_guarded_int_counter_vec_with_registry!(
611            "stream_agg_lookup_miss_count",
612            "Aggregation executor lookup miss duration",
613            &["table_id", "actor_id", "fragment_id"],
614            registry
615        )
616        .unwrap();
617
618        let agg_total_lookup_count = register_guarded_int_counter_vec_with_registry!(
619            "stream_agg_lookup_total_count",
620            "Aggregation executor lookup total operation",
621            &["table_id", "actor_id", "fragment_id"],
622            registry
623        )
624        .unwrap();
625
626        let agg_distinct_cache_miss_count = register_guarded_int_counter_vec_with_registry!(
627            "stream_agg_distinct_cache_miss_count",
628            "Aggregation executor dinsinct miss duration",
629            &["table_id", "actor_id", "fragment_id"],
630            registry
631        )
632        .unwrap();
633
634        let agg_distinct_total_cache_count = register_guarded_int_counter_vec_with_registry!(
635            "stream_agg_distinct_total_cache_count",
636            "Aggregation executor distinct total operation",
637            &["table_id", "actor_id", "fragment_id"],
638            registry
639        )
640        .unwrap();
641
642        let agg_distinct_cached_entry_count = register_guarded_int_gauge_vec_with_registry!(
643            "stream_agg_distinct_cached_entry_count",
644            "Total entry counts in distinct aggregation executor cache",
645            &["table_id", "actor_id", "fragment_id"],
646            registry
647        )
648        .unwrap();
649
650        let agg_dirty_groups_count = register_guarded_int_gauge_vec_with_registry!(
651            "stream_agg_dirty_groups_count",
652            "Total dirty group counts in aggregation executor",
653            &["table_id", "actor_id", "fragment_id"],
654            registry
655        )
656        .unwrap();
657
658        let agg_dirty_groups_heap_size = register_guarded_int_gauge_vec_with_registry!(
659            "stream_agg_dirty_groups_heap_size",
660            "Total dirty group heap size in aggregation executor",
661            &["table_id", "actor_id", "fragment_id"],
662            registry
663        )
664        .unwrap();
665
666        let agg_state_cache_lookup_count = register_guarded_int_counter_vec_with_registry!(
667            "stream_agg_state_cache_lookup_count",
668            "Aggregation executor state cache lookup count",
669            &["table_id", "actor_id", "fragment_id"],
670            registry
671        )
672        .unwrap();
673
674        let agg_state_cache_miss_count = register_guarded_int_counter_vec_with_registry!(
675            "stream_agg_state_cache_miss_count",
676            "Aggregation executor state cache miss count",
677            &["table_id", "actor_id", "fragment_id"],
678            registry
679        )
680        .unwrap();
681
682        let group_top_n_cache_miss_count = register_guarded_int_counter_vec_with_registry!(
683            "stream_group_top_n_cache_miss_count",
684            "Group top n executor cache miss count",
685            &["table_id", "actor_id", "fragment_id"],
686            registry
687        )
688        .unwrap();
689
690        let group_top_n_total_query_cache_count = register_guarded_int_counter_vec_with_registry!(
691            "stream_group_top_n_total_query_cache_count",
692            "Group top n executor query cache total count",
693            &["table_id", "actor_id", "fragment_id"],
694            registry
695        )
696        .unwrap();
697
698        let group_top_n_cached_entry_count = register_guarded_int_gauge_vec_with_registry!(
699            "stream_group_top_n_cached_entry_count",
700            "Total entry counts in group top n executor cache",
701            &["table_id", "actor_id", "fragment_id"],
702            registry
703        )
704        .unwrap();
705
706        let group_top_n_appendonly_cache_miss_count =
707            register_guarded_int_counter_vec_with_registry!(
708                "stream_group_top_n_appendonly_cache_miss_count",
709                "Group top n appendonly executor cache miss count",
710                &["table_id", "actor_id", "fragment_id"],
711                registry
712            )
713            .unwrap();
714
715        let group_top_n_appendonly_total_query_cache_count =
716            register_guarded_int_counter_vec_with_registry!(
717                "stream_group_top_n_appendonly_total_query_cache_count",
718                "Group top n appendonly executor total cache count",
719                &["table_id", "actor_id", "fragment_id"],
720                registry
721            )
722            .unwrap();
723
724        let group_top_n_appendonly_cached_entry_count =
725            register_guarded_int_gauge_vec_with_registry!(
726                "stream_group_top_n_appendonly_cached_entry_count",
727                "Total entry counts in group top n appendonly executor cache",
728                &["table_id", "actor_id", "fragment_id"],
729                registry
730            )
731            .unwrap();
732
733        let lookup_cache_miss_count = register_guarded_int_counter_vec_with_registry!(
734            "stream_lookup_cache_miss_count",
735            "Lookup executor cache miss count",
736            &["table_id", "actor_id", "fragment_id"],
737            registry
738        )
739        .unwrap();
740
741        let lookup_total_query_cache_count = register_guarded_int_counter_vec_with_registry!(
742            "stream_lookup_total_query_cache_count",
743            "Lookup executor query cache total count",
744            &["table_id", "actor_id", "fragment_id"],
745            registry
746        )
747        .unwrap();
748
749        let lookup_cached_entry_count = register_guarded_int_gauge_vec_with_registry!(
750            "stream_lookup_cached_entry_count",
751            "Total entry counts in lookup executor cache",
752            &["table_id", "actor_id", "fragment_id"],
753            registry
754        )
755        .unwrap();
756
757        let temporal_join_cache_miss_count = register_guarded_int_counter_vec_with_registry!(
758            "stream_temporal_join_cache_miss_count",
759            "Temporal join executor cache miss count",
760            &["table_id", "actor_id", "fragment_id"],
761            registry
762        )
763        .unwrap();
764
765        let temporal_join_total_query_cache_count =
766            register_guarded_int_counter_vec_with_registry!(
767                "stream_temporal_join_total_query_cache_count",
768                "Temporal join executor query cache total count",
769                &["table_id", "actor_id", "fragment_id"],
770                registry
771            )
772            .unwrap();
773
774        let temporal_join_cached_entry_count = register_guarded_int_gauge_vec_with_registry!(
775            "stream_temporal_join_cached_entry_count",
776            "Total entry count in temporal join executor cache",
777            &["table_id", "actor_id", "fragment_id"],
778            registry
779        )
780        .unwrap();
781
782        let agg_cached_entry_count = register_guarded_int_gauge_vec_with_registry!(
783            "stream_agg_cached_entry_count",
784            "Number of cached keys in streaming aggregation operators",
785            &["table_id", "actor_id", "fragment_id"],
786            registry
787        )
788        .unwrap();
789
790        let agg_chunk_lookup_miss_count = register_guarded_int_counter_vec_with_registry!(
791            "stream_agg_chunk_lookup_miss_count",
792            "Aggregation executor chunk-level lookup miss duration",
793            &["table_id", "actor_id", "fragment_id"],
794            registry
795        )
796        .unwrap();
797
798        let agg_chunk_total_lookup_count = register_guarded_int_counter_vec_with_registry!(
799            "stream_agg_chunk_lookup_total_count",
800            "Aggregation executor chunk-level lookup total operation",
801            &["table_id", "actor_id", "fragment_id"],
802            registry
803        )
804        .unwrap();
805
806        let backfill_snapshot_read_row_count = register_guarded_int_counter_vec_with_registry!(
807            "stream_backfill_snapshot_read_row_count",
808            "Total number of rows that have been read from the backfill snapshot",
809            &["table_id", "actor_id"],
810            registry
811        )
812        .unwrap();
813
814        let backfill_upstream_output_row_count = register_guarded_int_counter_vec_with_registry!(
815            "stream_backfill_upstream_output_row_count",
816            "Total number of rows that have been output from the backfill upstream",
817            &["table_id", "actor_id"],
818            registry
819        )
820        .unwrap();
821
822        let cdc_backfill_snapshot_read_row_count = register_guarded_int_counter_vec_with_registry!(
823            "stream_cdc_backfill_snapshot_read_row_count",
824            "Total number of rows that have been read from the cdc_backfill snapshot",
825            &["table_id", "actor_id"],
826            registry
827        )
828        .unwrap();
829
830        let cdc_backfill_upstream_output_row_count =
831            register_guarded_int_counter_vec_with_registry!(
832                "stream_cdc_backfill_upstream_output_row_count",
833                "Total number of rows that have been output from the cdc_backfill upstream",
834                &["table_id", "actor_id"],
835                registry
836            )
837            .unwrap();
838
839        let snapshot_backfill_consume_row_count = register_guarded_int_counter_vec_with_registry!(
840            "stream_snapshot_backfill_consume_snapshot_row_count",
841            "Total number of rows that have been output from snapshot backfill",
842            &["table_id", "actor_id", "stage"],
843            registry
844        )
845        .unwrap();
846
847        let over_window_cached_entry_count = register_guarded_int_gauge_vec_with_registry!(
848            "stream_over_window_cached_entry_count",
849            "Total entry (partition) count in over window executor cache",
850            &["table_id", "actor_id", "fragment_id"],
851            registry
852        )
853        .unwrap();
854
855        let over_window_cache_lookup_count = register_guarded_int_counter_vec_with_registry!(
856            "stream_over_window_cache_lookup_count",
857            "Over window executor cache lookup count",
858            &["table_id", "actor_id", "fragment_id"],
859            registry
860        )
861        .unwrap();
862
863        let over_window_cache_miss_count = register_guarded_int_counter_vec_with_registry!(
864            "stream_over_window_cache_miss_count",
865            "Over window executor cache miss count",
866            &["table_id", "actor_id", "fragment_id"],
867            registry
868        )
869        .unwrap();
870
871        let over_window_range_cache_entry_count = register_guarded_int_gauge_vec_with_registry!(
872            "stream_over_window_range_cache_entry_count",
873            "Over window partition range cache entry count",
874            &["table_id", "actor_id", "fragment_id"],
875            registry,
876        )
877        .unwrap();
878
879        let over_window_range_cache_lookup_count = register_guarded_int_counter_vec_with_registry!(
880            "stream_over_window_range_cache_lookup_count",
881            "Over window partition range cache lookup count",
882            &["table_id", "actor_id", "fragment_id"],
883            registry
884        )
885        .unwrap();
886
887        let over_window_range_cache_left_miss_count =
888            register_guarded_int_counter_vec_with_registry!(
889                "stream_over_window_range_cache_left_miss_count",
890                "Over window partition range cache left miss count",
891                &["table_id", "actor_id", "fragment_id"],
892                registry
893            )
894            .unwrap();
895
896        let over_window_range_cache_right_miss_count =
897            register_guarded_int_counter_vec_with_registry!(
898                "stream_over_window_range_cache_right_miss_count",
899                "Over window partition range cache right miss count",
900                &["table_id", "actor_id", "fragment_id"],
901                registry
902            )
903            .unwrap();
904
905        let over_window_accessed_entry_count = register_guarded_int_counter_vec_with_registry!(
906            "stream_over_window_accessed_entry_count",
907            "Over window accessed entry count",
908            &["table_id", "actor_id", "fragment_id"],
909            registry
910        )
911        .unwrap();
912
913        let over_window_compute_count = register_guarded_int_counter_vec_with_registry!(
914            "stream_over_window_compute_count",
915            "Over window compute count",
916            &["table_id", "actor_id", "fragment_id"],
917            registry
918        )
919        .unwrap();
920
921        let over_window_same_output_count = register_guarded_int_counter_vec_with_registry!(
922            "stream_over_window_same_output_count",
923            "Over window same output count",
924            &["table_id", "actor_id", "fragment_id"],
925            registry
926        )
927        .unwrap();
928
929        let barrier_inflight_latency = register_guarded_histogram_vec_with_registry!(
930            "stream_barrier_inflight_duration_seconds",
931            "barrier_inflight_latency",
932            &["partial_graph"],
933            latency_buckets(600.0, 20),
934            registry
935        )
936        .unwrap();
937
938        let barrier_sync_latency = register_guarded_histogram_vec_with_registry!(
939            "stream_barrier_sync_storage_duration_seconds",
940            "barrier_sync_latency",
941            &["partial_graph"],
942            latency_buckets(600.0, 20),
943            registry
944        )
945        .unwrap();
946
947        let opts = histogram_opts!(
948            "stream_barrier_batch_size",
949            "barrier_batch_size",
950            exponential_buckets(1.0, 2.0, 8).unwrap()
951        );
952        let barrier_batch_size = register_histogram_with_registry!(opts, registry).unwrap();
953
954        let barrier_manager_progress = register_guarded_int_counter_vec_with_registry!(
955            "stream_barrier_manager_progress",
956            "The number of actors that have processed the earliest in-flight barriers",
957            &["partial_graph"],
958            registry
959        )
960        .unwrap();
961
962        let sync_kv_log_store_wait_next_poll_ns = register_guarded_int_counter_vec_with_registry!(
963            "sync_kv_log_store_wait_next_poll_ns",
964            "Total duration (ns) of waiting for next poll",
965            &["actor_id", "target", "fragment_id", "relation"],
966            registry
967        )
968        .unwrap();
969
970        let sync_kv_log_store_read_count = register_guarded_int_counter_vec_with_registry!(
971            "sync_kv_log_store_read_count",
972            "read row count throughput of sync_kv log store",
973            &["type", "actor_id", "target", "fragment_id", "relation"],
974            registry
975        )
976        .unwrap();
977
978        let sync_kv_log_store_read_size = register_guarded_int_counter_vec_with_registry!(
979            "sync_kv_log_store_read_size",
980            "read size throughput of sync_kv log store",
981            &["type", "actor_id", "target", "fragment_id", "relation"],
982            registry
983        )
984        .unwrap();
985
986        let sync_kv_log_store_write_pause_duration_ns =
987            register_guarded_int_counter_vec_with_registry!(
988                "sync_kv_log_store_write_pause_duration_ns",
989                "Duration (ns) of sync_kv log store write pause",
990                &["actor_id", "target", "fragment_id", "relation"],
991                registry
992            )
993            .unwrap();
994
995        let sync_kv_log_store_state = register_guarded_int_counter_vec_with_registry!(
996            "sync_kv_log_store_state",
997            "clean/unclean state transition for sync_kv log store",
998            &["state", "actor_id", "target", "fragment_id", "relation"],
999            registry
1000        )
1001        .unwrap();
1002
1003        let sync_kv_log_store_storage_write_count =
1004            register_guarded_int_counter_vec_with_registry!(
1005                "sync_kv_log_store_storage_write_count",
1006                "Write row count throughput of sync_kv log store",
1007                &["actor_id", "target", "fragment_id", "relation"],
1008                registry
1009            )
1010            .unwrap();
1011
1012        let sync_kv_log_store_storage_write_size = register_guarded_int_counter_vec_with_registry!(
1013            "sync_kv_log_store_storage_write_size",
1014            "Write size throughput of sync_kv log store",
1015            &["actor_id", "target", "fragment_id", "relation"],
1016            registry
1017        )
1018        .unwrap();
1019
1020        let sync_kv_log_store_buffer_unconsumed_item_count =
1021            register_guarded_int_gauge_vec_with_registry!(
1022                "sync_kv_log_store_buffer_unconsumed_item_count",
1023                "Number of Unconsumed Item in buffer",
1024                &["actor_id", "target", "fragment_id", "relation"],
1025                registry
1026            )
1027            .unwrap();
1028
1029        let sync_kv_log_store_buffer_unconsumed_row_count =
1030            register_guarded_int_gauge_vec_with_registry!(
1031                "sync_kv_log_store_buffer_unconsumed_row_count",
1032                "Number of Unconsumed Row in buffer",
1033                &["actor_id", "target", "fragment_id", "relation"],
1034                registry
1035            )
1036            .unwrap();
1037
1038        let sync_kv_log_store_buffer_unconsumed_epoch_count =
1039            register_guarded_int_gauge_vec_with_registry!(
1040                "sync_kv_log_store_buffer_unconsumed_epoch_count",
1041                "Number of Unconsumed Epoch in buffer",
1042                &["actor_id", "target", "fragment_id", "relation"],
1043                registry
1044            )
1045            .unwrap();
1046
1047        let sync_kv_log_store_buffer_unconsumed_min_epoch =
1048            register_guarded_int_gauge_vec_with_registry!(
1049                "sync_kv_log_store_buffer_unconsumed_min_epoch",
1050                "Number of Unconsumed Epoch in buffer",
1051                &["actor_id", "target", "fragment_id", "relation"],
1052                registry
1053            )
1054            .unwrap();
1055        let sync_kv_log_store_buffer_memory_bytes =
1056            register_guarded_int_gauge_vec_with_registry!(
1057                "sync_kv_log_store_buffer_memory_bytes",
1058                "Estimated heap bytes used by synced kv log store buffer (unconsumed + consumed but not truncated)",
1059                &["actor_id", "target", "fragment_id", "relation"],
1060                registry
1061            )
1062            .unwrap();
1063
1064        let kv_log_store_storage_write_count = register_guarded_int_counter_vec_with_registry!(
1065            "kv_log_store_storage_write_count",
1066            "Write row count throughput of kv log store",
1067            &["actor_id", "connector", "sink_id", "sink_name"],
1068            registry
1069        )
1070        .unwrap();
1071
1072        let kv_log_store_storage_write_size = register_guarded_int_counter_vec_with_registry!(
1073            "kv_log_store_storage_write_size",
1074            "Write size throughput of kv log store",
1075            &["actor_id", "connector", "sink_id", "sink_name"],
1076            registry
1077        )
1078        .unwrap();
1079
1080        let kv_log_store_storage_read_count = register_guarded_int_counter_vec_with_registry!(
1081            "kv_log_store_storage_read_count",
1082            "Write row count throughput of kv log store",
1083            &["actor_id", "connector", "sink_id", "sink_name", "read_type"],
1084            registry
1085        )
1086        .unwrap();
1087
1088        let kv_log_store_storage_read_size = register_guarded_int_counter_vec_with_registry!(
1089            "kv_log_store_storage_read_size",
1090            "Write size throughput of kv log store",
1091            &["actor_id", "connector", "sink_id", "sink_name", "read_type"],
1092            registry
1093        )
1094        .unwrap();
1095
1096        let kv_log_store_rewind_count = register_guarded_int_counter_vec_with_registry!(
1097            "kv_log_store_rewind_count",
1098            "Kv log store rewind rate",
1099            &["actor_id", "connector", "sink_id", "sink_name"],
1100            registry
1101        )
1102        .unwrap();
1103
1104        let kv_log_store_rewind_delay_opts = {
1105            let first_delay_secs = REWIND_INITIAL_DELAY.as_secs_f64();
1106            let base = REWIND_BACKOFF_MULTIPLIER as f64;
1107            let bucket_count = (REWIND_MAX_DELAY.as_secs_f64() / first_delay_secs)
1108                .log(base)
1109                .ceil() as usize;
1110            let buckets = exponential_buckets(first_delay_secs, base, bucket_count).unwrap();
1111            histogram_opts!(
1112                "kv_log_store_rewind_delay",
1113                "Kv log store rewind delay",
1114                buckets,
1115            )
1116        };
1117
1118        let kv_log_store_rewind_delay = register_guarded_histogram_vec_with_registry!(
1119            kv_log_store_rewind_delay_opts,
1120            &["actor_id", "connector", "sink_id", "sink_name"],
1121            registry
1122        )
1123        .unwrap();
1124
1125        let kv_log_store_buffer_unconsumed_item_count =
1126            register_guarded_int_gauge_vec_with_registry!(
1127                "kv_log_store_buffer_unconsumed_item_count",
1128                "Number of Unconsumed Item in buffer",
1129                &["actor_id", "connector", "sink_id", "sink_name"],
1130                registry
1131            )
1132            .unwrap();
1133
1134        let kv_log_store_buffer_unconsumed_row_count =
1135            register_guarded_int_gauge_vec_with_registry!(
1136                "kv_log_store_buffer_unconsumed_row_count",
1137                "Number of Unconsumed Row in buffer",
1138                &["actor_id", "connector", "sink_id", "sink_name"],
1139                registry
1140            )
1141            .unwrap();
1142
1143        let kv_log_store_buffer_unconsumed_epoch_count =
1144            register_guarded_int_gauge_vec_with_registry!(
1145                "kv_log_store_buffer_unconsumed_epoch_count",
1146                "Number of Unconsumed Epoch in buffer",
1147                &["actor_id", "connector", "sink_id", "sink_name"],
1148                registry
1149            )
1150            .unwrap();
1151
1152        let kv_log_store_buffer_unconsumed_min_epoch =
1153            register_guarded_int_gauge_vec_with_registry!(
1154                "kv_log_store_buffer_unconsumed_min_epoch",
1155                "Number of Unconsumed Epoch in buffer",
1156                &["actor_id", "connector", "sink_id", "sink_name"],
1157                registry
1158            )
1159            .unwrap();
1160
1161        let crossdb_last_consumed_min_epoch = register_guarded_int_gauge_vec_with_registry!(
1162            "crossdb_last_consumed_min_epoch",
1163            "Last consumed min epoch for cross-database changelog stream scan",
1164            &["table_id", "actor_id", "fragment_id"],
1165            registry
1166        )
1167        .unwrap();
1168
1169        let kv_log_store_buffer_memory_bytes =
1170            register_guarded_int_gauge_vec_with_registry!(
1171                "kv_log_store_buffer_memory_bytes",
1172                "Estimated heap bytes used by kv log store buffer (unconsumed + consumed but not truncated)",
1173                &["actor_id", "connector", "sink_id", "sink_name"],
1174                registry
1175            )
1176            .unwrap();
1177
1178        let lru_runtime_loop_count = register_int_counter_with_registry!(
1179            "lru_runtime_loop_count",
1180            "The counts of the eviction loop in LRU manager per second",
1181            registry
1182        )
1183        .unwrap();
1184
1185        let lru_latest_sequence = register_int_gauge_with_registry!(
1186            "lru_latest_sequence",
1187            "Current LRU global sequence",
1188            registry,
1189        )
1190        .unwrap();
1191
1192        let lru_watermark_sequence = register_int_gauge_with_registry!(
1193            "lru_watermark_sequence",
1194            "Current LRU watermark sequence",
1195            registry,
1196        )
1197        .unwrap();
1198
1199        let lru_eviction_policy = register_int_gauge_with_registry!(
1200            "lru_eviction_policy",
1201            "Current LRU eviction policy",
1202            registry,
1203        )
1204        .unwrap();
1205
1206        let jemalloc_allocated_bytes = register_int_gauge_with_registry!(
1207            "jemalloc_allocated_bytes",
1208            "The allocated memory jemalloc, got from jemalloc_ctl",
1209            registry
1210        )
1211        .unwrap();
1212
1213        let jemalloc_active_bytes = register_int_gauge_with_registry!(
1214            "jemalloc_active_bytes",
1215            "The active memory jemalloc, got from jemalloc_ctl",
1216            registry
1217        )
1218        .unwrap();
1219
1220        let jemalloc_resident_bytes = register_int_gauge_with_registry!(
1221            "jemalloc_resident_bytes",
1222            "The active memory jemalloc, got from jemalloc_ctl",
1223            registry
1224        )
1225        .unwrap();
1226
1227        let jemalloc_metadata_bytes = register_int_gauge_with_registry!(
1228            "jemalloc_metadata_bytes",
1229            "The active memory jemalloc, got from jemalloc_ctl",
1230            registry
1231        )
1232        .unwrap();
1233
1234        let jvm_allocated_bytes = register_int_gauge_with_registry!(
1235            "jvm_allocated_bytes",
1236            "The allocated jvm memory",
1237            registry
1238        )
1239        .unwrap();
1240
1241        let jvm_active_bytes = register_int_gauge_with_registry!(
1242            "jvm_active_bytes",
1243            "The active jvm memory",
1244            registry
1245        )
1246        .unwrap();
1247
1248        let materialize_cache_hit_count = register_guarded_int_counter_vec_with_registry!(
1249            "stream_materialize_cache_hit_count",
1250            "Materialize executor cache hit count",
1251            &["actor_id", "table_id", "fragment_id"],
1252            registry
1253        )
1254        .unwrap()
1255        .relabel_debug_1(level);
1256
1257        let materialize_data_exist_count = register_guarded_int_counter_vec_with_registry!(
1258            "stream_materialize_data_exist_count",
1259            "Materialize executor data exist count",
1260            &["actor_id", "table_id", "fragment_id"],
1261            registry
1262        )
1263        .unwrap()
1264        .relabel_debug_1(level);
1265
1266        let materialize_cache_total_count = register_guarded_int_counter_vec_with_registry!(
1267            "stream_materialize_cache_total_count",
1268            "Materialize executor cache total operation",
1269            &["actor_id", "table_id", "fragment_id"],
1270            registry
1271        )
1272        .unwrap()
1273        .relabel_debug_1(level);
1274
1275        let stream_memory_usage = register_guarded_int_gauge_vec_with_registry!(
1276            "stream_memory_usage",
1277            "Memory usage for stream executors",
1278            &["actor_id", "table_id", "desc"],
1279            registry
1280        )
1281        .unwrap()
1282        .relabel_debug_1(level);
1283
1284        let gap_fill_generated_rows_count = register_guarded_int_counter_vec_with_registry!(
1285            "gap_fill_generated_rows_count",
1286            "Total number of rows generated by gap fill executor",
1287            &["actor_id", "fragment_id"],
1288            registry
1289        )
1290        .unwrap()
1291        .relabel_debug_1(level);
1292
1293        let state_table_iter_count = register_guarded_int_counter_vec_with_registry!(
1294            "state_table_iter_count",
1295            "Total number of state table iter operations",
1296            &["actor_id", "fragment_id", "table_id"],
1297            registry
1298        )
1299        .unwrap()
1300        .relabel_debug_1(level);
1301
1302        let state_table_get_count = register_guarded_int_counter_vec_with_registry!(
1303            "state_table_get_count",
1304            "Total number of state table get operations",
1305            &["actor_id", "fragment_id", "table_id"],
1306            registry
1307        )
1308        .unwrap()
1309        .relabel_debug_1(level);
1310
1311        let state_table_iter_vnode_pruned_count = register_guarded_int_counter_vec_with_registry!(
1312            "state_table_iter_vnode_pruned_count",
1313            "Total number of state table iter operations pruned by vnode statistics",
1314            &["actor_id", "fragment_id", "table_id"],
1315            registry
1316        )
1317        .unwrap()
1318        .relabel_debug_1(level);
1319
1320        let state_table_get_vnode_pruned_count = register_guarded_int_counter_vec_with_registry!(
1321            "state_table_get_vnode_pruned_count",
1322            "Total number of state table get operations pruned by vnode statistics",
1323            &["actor_id", "fragment_id", "table_id"],
1324            registry
1325        )
1326        .unwrap()
1327        .relabel_debug_1(level);
1328
1329        Self {
1330            level,
1331            executor_row_count,
1332            mem_stream_node_output_row_count: stream_node_output_row_count,
1333            mem_stream_node_output_blocking_duration_ns: stream_node_output_blocking_duration_ns,
1334            actor_scheduled_duration,
1335            actor_scheduled_cnt,
1336            actor_poll_duration,
1337            actor_poll_cnt,
1338            actor_idle_duration,
1339            actor_idle_cnt,
1340            actor_count,
1341            actor_in_record_cnt,
1342            actor_out_record_cnt,
1343            fragment_channel_buffered_bytes,
1344            actor_current_epoch,
1345            project_expr_inflight_window_size,
1346            source_output_row_count,
1347            source_split_change_count,
1348            source_backfill_row_count,
1349            sink_input_row_count,
1350            sink_input_bytes,
1351            sink_chunk_buffer_size,
1352            exchange_frag_recv_size,
1353            merge_barrier_align_duration,
1354            actor_output_buffer_blocking_duration_ns,
1355            actor_input_buffer_blocking_duration_ns,
1356            join_lookup_miss_count,
1357            join_lookup_total_count,
1358            join_insert_cache_miss_count,
1359            join_actor_input_waiting_duration_ns,
1360            join_match_duration_ns,
1361            join_cached_entry_count,
1362            join_matched_join_keys,
1363            barrier_align_duration,
1364            agg_lookup_miss_count,
1365            agg_total_lookup_count,
1366            agg_cached_entry_count,
1367            agg_chunk_lookup_miss_count,
1368            agg_chunk_total_lookup_count,
1369            agg_dirty_groups_count,
1370            agg_dirty_groups_heap_size,
1371            agg_distinct_cache_miss_count,
1372            agg_distinct_total_cache_count,
1373            agg_distinct_cached_entry_count,
1374            agg_state_cache_lookup_count,
1375            agg_state_cache_miss_count,
1376            group_top_n_cache_miss_count,
1377            group_top_n_total_query_cache_count,
1378            group_top_n_cached_entry_count,
1379            group_top_n_appendonly_cache_miss_count,
1380            group_top_n_appendonly_total_query_cache_count,
1381            group_top_n_appendonly_cached_entry_count,
1382            lookup_cache_miss_count,
1383            lookup_total_query_cache_count,
1384            lookup_cached_entry_count,
1385            temporal_join_cache_miss_count,
1386            temporal_join_total_query_cache_count,
1387            temporal_join_cached_entry_count,
1388            backfill_snapshot_read_row_count,
1389            backfill_upstream_output_row_count,
1390            cdc_backfill_snapshot_read_row_count,
1391            cdc_backfill_upstream_output_row_count,
1392            snapshot_backfill_consume_row_count,
1393            over_window_cached_entry_count,
1394            over_window_cache_lookup_count,
1395            over_window_cache_miss_count,
1396            over_window_range_cache_entry_count,
1397            over_window_range_cache_lookup_count,
1398            over_window_range_cache_left_miss_count,
1399            over_window_range_cache_right_miss_count,
1400            over_window_accessed_entry_count,
1401            over_window_compute_count,
1402            over_window_same_output_count,
1403            barrier_inflight_latency,
1404            barrier_sync_latency,
1405            barrier_batch_size,
1406            barrier_manager_progress,
1407            kv_log_store_storage_write_count,
1408            kv_log_store_storage_write_size,
1409            kv_log_store_rewind_count,
1410            kv_log_store_rewind_delay,
1411            kv_log_store_storage_read_count,
1412            kv_log_store_storage_read_size,
1413            kv_log_store_buffer_unconsumed_item_count,
1414            kv_log_store_buffer_unconsumed_row_count,
1415            kv_log_store_buffer_unconsumed_epoch_count,
1416            kv_log_store_buffer_unconsumed_min_epoch,
1417            kv_log_store_buffer_memory_bytes,
1418            crossdb_last_consumed_min_epoch,
1419            sync_kv_log_store_read_count,
1420            sync_kv_log_store_read_size,
1421            sync_kv_log_store_write_pause_duration_ns,
1422            sync_kv_log_store_state,
1423            sync_kv_log_store_wait_next_poll_ns,
1424            sync_kv_log_store_storage_write_count,
1425            sync_kv_log_store_storage_write_size,
1426            sync_kv_log_store_buffer_unconsumed_item_count,
1427            sync_kv_log_store_buffer_unconsumed_row_count,
1428            sync_kv_log_store_buffer_unconsumed_epoch_count,
1429            sync_kv_log_store_buffer_unconsumed_min_epoch,
1430            sync_kv_log_store_buffer_memory_bytes,
1431            lru_runtime_loop_count,
1432            lru_latest_sequence,
1433            lru_watermark_sequence,
1434            lru_eviction_policy,
1435            jemalloc_allocated_bytes,
1436            jemalloc_active_bytes,
1437            jemalloc_resident_bytes,
1438            jemalloc_metadata_bytes,
1439            jvm_allocated_bytes,
1440            jvm_active_bytes,
1441            stream_memory_usage,
1442            materialize_cache_hit_count,
1443            materialize_data_exist_count,
1444            materialize_cache_total_count,
1445            materialize_input_row_count,
1446            materialize_current_epoch,
1447            pg_cdc_state_table_lsn,
1448            pg_cdc_jni_commit_offset_lsn,
1449            mysql_cdc_state_binlog_file_seq,
1450            mysql_cdc_state_binlog_position,
1451            sqlserver_cdc_state_change_lsn,
1452            sqlserver_cdc_state_commit_lsn,
1453            sqlserver_cdc_jni_commit_offset_lsn,
1454            gap_fill_generated_rows_count,
1455            state_table_iter_count,
1456            state_table_get_count,
1457            state_table_iter_vnode_pruned_count,
1458            state_table_get_vnode_pruned_count,
1459        }
1460    }
1461
1462    /// Create a new `StreamingMetrics` instance used in tests or other places.
1463    pub fn unused() -> Self {
1464        global_streaming_metrics(MetricLevel::Disabled)
1465    }
1466
1467    pub fn new_actor_metrics(&self, actor_id: ActorId, fragment_id: FragmentId) -> ActorMetrics {
1468        let label_list: &[&str; 2] = &[&actor_id.to_string(), &fragment_id.to_string()];
1469        let actor_scheduled_duration = self
1470            .actor_scheduled_duration
1471            .with_guarded_label_values(label_list);
1472        let actor_scheduled_cnt = self
1473            .actor_scheduled_cnt
1474            .with_guarded_label_values(label_list);
1475        let actor_poll_duration = self
1476            .actor_poll_duration
1477            .with_guarded_label_values(label_list);
1478        let actor_poll_cnt = self.actor_poll_cnt.with_guarded_label_values(label_list);
1479        let actor_idle_duration = self
1480            .actor_idle_duration
1481            .with_guarded_label_values(label_list);
1482        let actor_idle_cnt = self.actor_idle_cnt.with_guarded_label_values(label_list);
1483        ActorMetrics {
1484            actor_scheduled_duration,
1485            actor_scheduled_cnt,
1486            actor_poll_duration,
1487            actor_poll_cnt,
1488            actor_idle_duration,
1489            actor_idle_cnt,
1490        }
1491    }
1492
1493    pub(crate) fn new_actor_input_metrics(
1494        &self,
1495        actor_id: ActorId,
1496        fragment_id: FragmentId,
1497        upstream_fragment_id: FragmentId,
1498    ) -> ActorInputMetrics {
1499        let actor_id_str = actor_id.to_string();
1500        let fragment_id_str = fragment_id.to_string();
1501        let upstream_fragment_id_str = upstream_fragment_id.to_string();
1502        ActorInputMetrics {
1503            actor_in_record_cnt: self.actor_in_record_cnt.with_guarded_label_values(&[
1504                &actor_id_str,
1505                &fragment_id_str,
1506                &upstream_fragment_id_str,
1507            ]),
1508            actor_input_buffer_blocking_duration_ns: self
1509                .actor_input_buffer_blocking_duration_ns
1510                .with_guarded_label_values(&[
1511                    &actor_id_str,
1512                    &fragment_id_str,
1513                    &upstream_fragment_id_str,
1514                ]),
1515        }
1516    }
1517
1518    pub fn new_sink_exec_metrics(
1519        &self,
1520        id: SinkId,
1521        actor_id: ActorId,
1522        fragment_id: FragmentId,
1523    ) -> SinkExecutorMetrics {
1524        let label_list: &[&str; 3] = &[
1525            &id.to_string(),
1526            &actor_id.to_string(),
1527            &fragment_id.to_string(),
1528        ];
1529        SinkExecutorMetrics {
1530            sink_input_row_count: self
1531                .sink_input_row_count
1532                .with_guarded_label_values(label_list),
1533            sink_input_bytes: self.sink_input_bytes.with_guarded_label_values(label_list),
1534            sink_chunk_buffer_size: self
1535                .sink_chunk_buffer_size
1536                .with_guarded_label_values(label_list),
1537        }
1538    }
1539
1540    pub fn new_group_top_n_metrics(
1541        &self,
1542        table_id: TableId,
1543        actor_id: ActorId,
1544        fragment_id: FragmentId,
1545    ) -> GroupTopNMetrics {
1546        let label_list: &[&str; 3] = &[
1547            &table_id.to_string(),
1548            &actor_id.to_string(),
1549            &fragment_id.to_string(),
1550        ];
1551
1552        GroupTopNMetrics {
1553            group_top_n_cache_miss_count: self
1554                .group_top_n_cache_miss_count
1555                .with_guarded_label_values(label_list),
1556            group_top_n_total_query_cache_count: self
1557                .group_top_n_total_query_cache_count
1558                .with_guarded_label_values(label_list),
1559            group_top_n_cached_entry_count: self
1560                .group_top_n_cached_entry_count
1561                .with_guarded_label_values(label_list),
1562        }
1563    }
1564
1565    pub fn new_append_only_group_top_n_metrics(
1566        &self,
1567        table_id: TableId,
1568        actor_id: ActorId,
1569        fragment_id: FragmentId,
1570    ) -> GroupTopNMetrics {
1571        let label_list: &[&str; 3] = &[
1572            &table_id.to_string(),
1573            &actor_id.to_string(),
1574            &fragment_id.to_string(),
1575        ];
1576
1577        GroupTopNMetrics {
1578            group_top_n_cache_miss_count: self
1579                .group_top_n_appendonly_cache_miss_count
1580                .with_guarded_label_values(label_list),
1581            group_top_n_total_query_cache_count: self
1582                .group_top_n_appendonly_total_query_cache_count
1583                .with_guarded_label_values(label_list),
1584            group_top_n_cached_entry_count: self
1585                .group_top_n_appendonly_cached_entry_count
1586                .with_guarded_label_values(label_list),
1587        }
1588    }
1589
1590    pub fn new_lookup_executor_metrics(
1591        &self,
1592        table_id: TableId,
1593        actor_id: ActorId,
1594        fragment_id: FragmentId,
1595    ) -> LookupExecutorMetrics {
1596        let label_list: &[&str; 3] = &[
1597            &table_id.to_string(),
1598            &actor_id.to_string(),
1599            &fragment_id.to_string(),
1600        ];
1601
1602        LookupExecutorMetrics {
1603            lookup_cache_miss_count: self
1604                .lookup_cache_miss_count
1605                .with_guarded_label_values(label_list),
1606            lookup_total_query_cache_count: self
1607                .lookup_total_query_cache_count
1608                .with_guarded_label_values(label_list),
1609            lookup_cached_entry_count: self
1610                .lookup_cached_entry_count
1611                .with_guarded_label_values(label_list),
1612        }
1613    }
1614
1615    pub fn new_hash_agg_metrics(
1616        &self,
1617        table_id: TableId,
1618        actor_id: ActorId,
1619        fragment_id: FragmentId,
1620    ) -> HashAggMetrics {
1621        let label_list: &[&str; 3] = &[
1622            &table_id.to_string(),
1623            &actor_id.to_string(),
1624            &fragment_id.to_string(),
1625        ];
1626        HashAggMetrics {
1627            agg_lookup_miss_count: self
1628                .agg_lookup_miss_count
1629                .with_guarded_label_values(label_list),
1630            agg_total_lookup_count: self
1631                .agg_total_lookup_count
1632                .with_guarded_label_values(label_list),
1633            agg_cached_entry_count: self
1634                .agg_cached_entry_count
1635                .with_guarded_label_values(label_list),
1636            agg_chunk_lookup_miss_count: self
1637                .agg_chunk_lookup_miss_count
1638                .with_guarded_label_values(label_list),
1639            agg_chunk_total_lookup_count: self
1640                .agg_chunk_total_lookup_count
1641                .with_guarded_label_values(label_list),
1642            agg_dirty_groups_count: self
1643                .agg_dirty_groups_count
1644                .with_guarded_label_values(label_list),
1645            agg_dirty_groups_heap_size: self
1646                .agg_dirty_groups_heap_size
1647                .with_guarded_label_values(label_list),
1648            agg_state_cache_lookup_count: self
1649                .agg_state_cache_lookup_count
1650                .with_guarded_label_values(label_list),
1651            agg_state_cache_miss_count: self
1652                .agg_state_cache_miss_count
1653                .with_guarded_label_values(label_list),
1654        }
1655    }
1656
1657    pub fn new_agg_distinct_dedup_metrics(
1658        &self,
1659        table_id: TableId,
1660        actor_id: ActorId,
1661        fragment_id: FragmentId,
1662    ) -> AggDistinctDedupMetrics {
1663        let label_list: &[&str; 3] = &[
1664            &table_id.to_string(),
1665            &actor_id.to_string(),
1666            &fragment_id.to_string(),
1667        ];
1668        AggDistinctDedupMetrics {
1669            agg_distinct_cache_miss_count: self
1670                .agg_distinct_cache_miss_count
1671                .with_guarded_label_values(label_list),
1672            agg_distinct_total_cache_count: self
1673                .agg_distinct_total_cache_count
1674                .with_guarded_label_values(label_list),
1675            agg_distinct_cached_entry_count: self
1676                .agg_distinct_cached_entry_count
1677                .with_guarded_label_values(label_list),
1678        }
1679    }
1680
1681    pub fn new_temporal_join_metrics(
1682        &self,
1683        table_id: TableId,
1684        actor_id: ActorId,
1685        fragment_id: FragmentId,
1686    ) -> TemporalJoinMetrics {
1687        let label_list: &[&str; 3] = &[
1688            &table_id.to_string(),
1689            &actor_id.to_string(),
1690            &fragment_id.to_string(),
1691        ];
1692        TemporalJoinMetrics {
1693            temporal_join_cache_miss_count: self
1694                .temporal_join_cache_miss_count
1695                .with_guarded_label_values(label_list),
1696            temporal_join_total_query_cache_count: self
1697                .temporal_join_total_query_cache_count
1698                .with_guarded_label_values(label_list),
1699            temporal_join_cached_entry_count: self
1700                .temporal_join_cached_entry_count
1701                .with_guarded_label_values(label_list),
1702        }
1703    }
1704
1705    pub fn new_backfill_metrics(&self, table_id: TableId, actor_id: ActorId) -> BackfillMetrics {
1706        let label_list: &[&str; 2] = &[&table_id.to_string(), &actor_id.to_string()];
1707        BackfillMetrics {
1708            backfill_snapshot_read_row_count: self
1709                .backfill_snapshot_read_row_count
1710                .with_guarded_label_values(label_list),
1711            backfill_upstream_output_row_count: self
1712                .backfill_upstream_output_row_count
1713                .with_guarded_label_values(label_list),
1714        }
1715    }
1716
1717    pub fn new_cdc_backfill_metrics(
1718        &self,
1719        table_id: TableId,
1720        actor_id: ActorId,
1721    ) -> CdcBackfillMetrics {
1722        let label_list: &[&str; 2] = &[&table_id.to_string(), &actor_id.to_string()];
1723        CdcBackfillMetrics {
1724            cdc_backfill_snapshot_read_row_count: self
1725                .cdc_backfill_snapshot_read_row_count
1726                .with_guarded_label_values(label_list),
1727            cdc_backfill_upstream_output_row_count: self
1728                .cdc_backfill_upstream_output_row_count
1729                .with_guarded_label_values(label_list),
1730        }
1731    }
1732
1733    pub fn new_over_window_metrics(
1734        &self,
1735        table_id: TableId,
1736        actor_id: ActorId,
1737        fragment_id: FragmentId,
1738    ) -> OverWindowMetrics {
1739        let label_list: &[&str; 3] = &[
1740            &table_id.to_string(),
1741            &actor_id.to_string(),
1742            &fragment_id.to_string(),
1743        ];
1744        OverWindowMetrics {
1745            over_window_cached_entry_count: self
1746                .over_window_cached_entry_count
1747                .with_guarded_label_values(label_list),
1748            over_window_cache_lookup_count: self
1749                .over_window_cache_lookup_count
1750                .with_guarded_label_values(label_list),
1751            over_window_cache_miss_count: self
1752                .over_window_cache_miss_count
1753                .with_guarded_label_values(label_list),
1754            over_window_range_cache_entry_count: self
1755                .over_window_range_cache_entry_count
1756                .with_guarded_label_values(label_list),
1757            over_window_range_cache_lookup_count: self
1758                .over_window_range_cache_lookup_count
1759                .with_guarded_label_values(label_list),
1760            over_window_range_cache_left_miss_count: self
1761                .over_window_range_cache_left_miss_count
1762                .with_guarded_label_values(label_list),
1763            over_window_range_cache_right_miss_count: self
1764                .over_window_range_cache_right_miss_count
1765                .with_guarded_label_values(label_list),
1766            over_window_accessed_entry_count: self
1767                .over_window_accessed_entry_count
1768                .with_guarded_label_values(label_list),
1769            over_window_compute_count: self
1770                .over_window_compute_count
1771                .with_guarded_label_values(label_list),
1772            over_window_same_output_count: self
1773                .over_window_same_output_count
1774                .with_guarded_label_values(label_list),
1775        }
1776    }
1777
1778    pub fn new_materialize_cache_metrics(
1779        &self,
1780        table_id: TableId,
1781        actor_id: ActorId,
1782        fragment_id: FragmentId,
1783    ) -> MaterializeCacheMetrics {
1784        let label_list: &[&str; 3] = &[
1785            &actor_id.to_string(),
1786            &table_id.to_string(),
1787            &fragment_id.to_string(),
1788        ];
1789        MaterializeCacheMetrics {
1790            materialize_cache_hit_count: self
1791                .materialize_cache_hit_count
1792                .with_guarded_label_values(label_list),
1793            materialize_data_exist_count: self
1794                .materialize_data_exist_count
1795                .with_guarded_label_values(label_list),
1796            materialize_cache_total_count: self
1797                .materialize_cache_total_count
1798                .with_guarded_label_values(label_list),
1799        }
1800    }
1801
1802    pub fn new_materialize_metrics(
1803        &self,
1804        table_id: TableId,
1805        actor_id: ActorId,
1806        fragment_id: FragmentId,
1807    ) -> MaterializeMetrics {
1808        let label_list: &[&str; 3] = &[
1809            &actor_id.to_string(),
1810            &table_id.to_string(),
1811            &fragment_id.to_string(),
1812        ];
1813        MaterializeMetrics {
1814            materialize_input_row_count: self
1815                .materialize_input_row_count
1816                .with_guarded_label_values(label_list),
1817            materialize_current_epoch: self
1818                .materialize_current_epoch
1819                .with_guarded_label_values(label_list),
1820        }
1821    }
1822
1823    pub fn new_state_table_metrics(
1824        &self,
1825        table_id: TableId,
1826        actor_id: ActorId,
1827        fragment_id: FragmentId,
1828    ) -> StateTableMetrics {
1829        let label_list: &[&str; 3] = &[
1830            &actor_id.to_string(),
1831            &fragment_id.to_string(),
1832            &table_id.to_string(),
1833        ];
1834        StateTableMetrics {
1835            iter_count: self
1836                .state_table_iter_count
1837                .with_guarded_label_values(label_list),
1838            get_count: self
1839                .state_table_get_count
1840                .with_guarded_label_values(label_list),
1841            iter_vnode_pruned_count: self
1842                .state_table_iter_vnode_pruned_count
1843                .with_guarded_label_values(label_list),
1844            get_vnode_pruned_count: self
1845                .state_table_get_vnode_pruned_count
1846                .with_guarded_label_values(label_list),
1847        }
1848    }
1849}
1850
1851pub(crate) struct ActorInputMetrics {
1852    pub(crate) actor_in_record_cnt: LabelGuardedIntCounter,
1853    pub(crate) actor_input_buffer_blocking_duration_ns: LabelGuardedIntCounter,
1854}
1855
1856/// Tokio metrics for actors
1857pub struct ActorMetrics {
1858    pub actor_scheduled_duration: LabelGuardedIntCounter,
1859    pub actor_scheduled_cnt: LabelGuardedIntCounter,
1860    pub actor_poll_duration: LabelGuardedIntCounter,
1861    pub actor_poll_cnt: LabelGuardedIntCounter,
1862    pub actor_idle_duration: LabelGuardedIntCounter,
1863    pub actor_idle_cnt: LabelGuardedIntCounter,
1864}
1865
1866pub struct SinkExecutorMetrics {
1867    pub sink_input_row_count: LabelGuardedIntCounter,
1868    pub sink_input_bytes: LabelGuardedIntCounter,
1869    pub sink_chunk_buffer_size: LabelGuardedIntGauge,
1870}
1871
1872pub struct MaterializeCacheMetrics {
1873    pub materialize_cache_hit_count: LabelGuardedIntCounter,
1874    pub materialize_data_exist_count: LabelGuardedIntCounter,
1875    pub materialize_cache_total_count: LabelGuardedIntCounter,
1876}
1877
1878pub struct MaterializeMetrics {
1879    pub materialize_input_row_count: LabelGuardedIntCounter,
1880    pub materialize_current_epoch: LabelGuardedIntGauge,
1881}
1882
1883pub struct GroupTopNMetrics {
1884    pub group_top_n_cache_miss_count: LabelGuardedIntCounter,
1885    pub group_top_n_total_query_cache_count: LabelGuardedIntCounter,
1886    pub group_top_n_cached_entry_count: LabelGuardedIntGauge,
1887}
1888
1889pub struct LookupExecutorMetrics {
1890    pub lookup_cache_miss_count: LabelGuardedIntCounter,
1891    pub lookup_total_query_cache_count: LabelGuardedIntCounter,
1892    pub lookup_cached_entry_count: LabelGuardedIntGauge,
1893}
1894
1895pub struct HashAggMetrics {
1896    pub agg_lookup_miss_count: LabelGuardedIntCounter,
1897    pub agg_total_lookup_count: LabelGuardedIntCounter,
1898    pub agg_cached_entry_count: LabelGuardedIntGauge,
1899    pub agg_chunk_lookup_miss_count: LabelGuardedIntCounter,
1900    pub agg_chunk_total_lookup_count: LabelGuardedIntCounter,
1901    pub agg_dirty_groups_count: LabelGuardedIntGauge,
1902    pub agg_dirty_groups_heap_size: LabelGuardedIntGauge,
1903    pub agg_state_cache_lookup_count: LabelGuardedIntCounter,
1904    pub agg_state_cache_miss_count: LabelGuardedIntCounter,
1905}
1906
1907pub struct AggDistinctDedupMetrics {
1908    pub agg_distinct_cache_miss_count: LabelGuardedIntCounter,
1909    pub agg_distinct_total_cache_count: LabelGuardedIntCounter,
1910    pub agg_distinct_cached_entry_count: LabelGuardedIntGauge,
1911}
1912
1913pub struct TemporalJoinMetrics {
1914    pub temporal_join_cache_miss_count: LabelGuardedIntCounter,
1915    pub temporal_join_total_query_cache_count: LabelGuardedIntCounter,
1916    pub temporal_join_cached_entry_count: LabelGuardedIntGauge,
1917}
1918
1919pub struct BackfillMetrics {
1920    pub backfill_snapshot_read_row_count: LabelGuardedIntCounter,
1921    pub backfill_upstream_output_row_count: LabelGuardedIntCounter,
1922}
1923
1924#[derive(Clone)]
1925pub struct CdcBackfillMetrics {
1926    pub cdc_backfill_snapshot_read_row_count: LabelGuardedIntCounter,
1927    pub cdc_backfill_upstream_output_row_count: LabelGuardedIntCounter,
1928}
1929
1930pub struct OverWindowMetrics {
1931    pub over_window_cached_entry_count: LabelGuardedIntGauge,
1932    pub over_window_cache_lookup_count: LabelGuardedIntCounter,
1933    pub over_window_cache_miss_count: LabelGuardedIntCounter,
1934    pub over_window_range_cache_entry_count: LabelGuardedIntGauge,
1935    pub over_window_range_cache_lookup_count: LabelGuardedIntCounter,
1936    pub over_window_range_cache_left_miss_count: LabelGuardedIntCounter,
1937    pub over_window_range_cache_right_miss_count: LabelGuardedIntCounter,
1938    pub over_window_accessed_entry_count: LabelGuardedIntCounter,
1939    pub over_window_compute_count: LabelGuardedIntCounter,
1940    pub over_window_same_output_count: LabelGuardedIntCounter,
1941}
1942
1943#[derive(Clone)]
1944pub struct StateTableMetrics {
1945    pub iter_count: LabelGuardedIntCounter,
1946    pub get_count: LabelGuardedIntCounter,
1947    pub iter_vnode_pruned_count: LabelGuardedIntCounter,
1948    pub get_vnode_pruned_count: LabelGuardedIntCounter,
1949}