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_FACTOR, REWIND_BASE_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            assert_eq!(2, REWIND_BACKOFF_FACTOR);
1106            let bucket_count = (REWIND_MAX_DELAY.as_secs_f64().log2()
1107                - REWIND_BASE_DELAY.as_secs_f64().log2())
1108            .ceil() as usize;
1109            let buckets = exponential_buckets(
1110                REWIND_BASE_DELAY.as_secs_f64(),
1111                REWIND_BACKOFF_FACTOR as _,
1112                bucket_count,
1113            )
1114            .unwrap();
1115            histogram_opts!(
1116                "kv_log_store_rewind_delay",
1117                "Kv log store rewind delay",
1118                buckets,
1119            )
1120        };
1121
1122        let kv_log_store_rewind_delay = register_guarded_histogram_vec_with_registry!(
1123            kv_log_store_rewind_delay_opts,
1124            &["actor_id", "connector", "sink_id", "sink_name"],
1125            registry
1126        )
1127        .unwrap();
1128
1129        let kv_log_store_buffer_unconsumed_item_count =
1130            register_guarded_int_gauge_vec_with_registry!(
1131                "kv_log_store_buffer_unconsumed_item_count",
1132                "Number of Unconsumed Item in buffer",
1133                &["actor_id", "connector", "sink_id", "sink_name"],
1134                registry
1135            )
1136            .unwrap();
1137
1138        let kv_log_store_buffer_unconsumed_row_count =
1139            register_guarded_int_gauge_vec_with_registry!(
1140                "kv_log_store_buffer_unconsumed_row_count",
1141                "Number of Unconsumed Row in buffer",
1142                &["actor_id", "connector", "sink_id", "sink_name"],
1143                registry
1144            )
1145            .unwrap();
1146
1147        let kv_log_store_buffer_unconsumed_epoch_count =
1148            register_guarded_int_gauge_vec_with_registry!(
1149                "kv_log_store_buffer_unconsumed_epoch_count",
1150                "Number of Unconsumed Epoch in buffer",
1151                &["actor_id", "connector", "sink_id", "sink_name"],
1152                registry
1153            )
1154            .unwrap();
1155
1156        let kv_log_store_buffer_unconsumed_min_epoch =
1157            register_guarded_int_gauge_vec_with_registry!(
1158                "kv_log_store_buffer_unconsumed_min_epoch",
1159                "Number of Unconsumed Epoch in buffer",
1160                &["actor_id", "connector", "sink_id", "sink_name"],
1161                registry
1162            )
1163            .unwrap();
1164
1165        let crossdb_last_consumed_min_epoch = register_guarded_int_gauge_vec_with_registry!(
1166            "crossdb_last_consumed_min_epoch",
1167            "Last consumed min epoch for cross-database changelog stream scan",
1168            &["table_id", "actor_id", "fragment_id"],
1169            registry
1170        )
1171        .unwrap();
1172
1173        let kv_log_store_buffer_memory_bytes =
1174            register_guarded_int_gauge_vec_with_registry!(
1175                "kv_log_store_buffer_memory_bytes",
1176                "Estimated heap bytes used by kv log store buffer (unconsumed + consumed but not truncated)",
1177                &["actor_id", "connector", "sink_id", "sink_name"],
1178                registry
1179            )
1180            .unwrap();
1181
1182        let lru_runtime_loop_count = register_int_counter_with_registry!(
1183            "lru_runtime_loop_count",
1184            "The counts of the eviction loop in LRU manager per second",
1185            registry
1186        )
1187        .unwrap();
1188
1189        let lru_latest_sequence = register_int_gauge_with_registry!(
1190            "lru_latest_sequence",
1191            "Current LRU global sequence",
1192            registry,
1193        )
1194        .unwrap();
1195
1196        let lru_watermark_sequence = register_int_gauge_with_registry!(
1197            "lru_watermark_sequence",
1198            "Current LRU watermark sequence",
1199            registry,
1200        )
1201        .unwrap();
1202
1203        let lru_eviction_policy = register_int_gauge_with_registry!(
1204            "lru_eviction_policy",
1205            "Current LRU eviction policy",
1206            registry,
1207        )
1208        .unwrap();
1209
1210        let jemalloc_allocated_bytes = register_int_gauge_with_registry!(
1211            "jemalloc_allocated_bytes",
1212            "The allocated memory jemalloc, got from jemalloc_ctl",
1213            registry
1214        )
1215        .unwrap();
1216
1217        let jemalloc_active_bytes = register_int_gauge_with_registry!(
1218            "jemalloc_active_bytes",
1219            "The active memory jemalloc, got from jemalloc_ctl",
1220            registry
1221        )
1222        .unwrap();
1223
1224        let jemalloc_resident_bytes = register_int_gauge_with_registry!(
1225            "jemalloc_resident_bytes",
1226            "The active memory jemalloc, got from jemalloc_ctl",
1227            registry
1228        )
1229        .unwrap();
1230
1231        let jemalloc_metadata_bytes = register_int_gauge_with_registry!(
1232            "jemalloc_metadata_bytes",
1233            "The active memory jemalloc, got from jemalloc_ctl",
1234            registry
1235        )
1236        .unwrap();
1237
1238        let jvm_allocated_bytes = register_int_gauge_with_registry!(
1239            "jvm_allocated_bytes",
1240            "The allocated jvm memory",
1241            registry
1242        )
1243        .unwrap();
1244
1245        let jvm_active_bytes = register_int_gauge_with_registry!(
1246            "jvm_active_bytes",
1247            "The active jvm memory",
1248            registry
1249        )
1250        .unwrap();
1251
1252        let materialize_cache_hit_count = register_guarded_int_counter_vec_with_registry!(
1253            "stream_materialize_cache_hit_count",
1254            "Materialize executor cache hit count",
1255            &["actor_id", "table_id", "fragment_id"],
1256            registry
1257        )
1258        .unwrap()
1259        .relabel_debug_1(level);
1260
1261        let materialize_data_exist_count = register_guarded_int_counter_vec_with_registry!(
1262            "stream_materialize_data_exist_count",
1263            "Materialize executor data exist count",
1264            &["actor_id", "table_id", "fragment_id"],
1265            registry
1266        )
1267        .unwrap()
1268        .relabel_debug_1(level);
1269
1270        let materialize_cache_total_count = register_guarded_int_counter_vec_with_registry!(
1271            "stream_materialize_cache_total_count",
1272            "Materialize executor cache total operation",
1273            &["actor_id", "table_id", "fragment_id"],
1274            registry
1275        )
1276        .unwrap()
1277        .relabel_debug_1(level);
1278
1279        let stream_memory_usage = register_guarded_int_gauge_vec_with_registry!(
1280            "stream_memory_usage",
1281            "Memory usage for stream executors",
1282            &["actor_id", "table_id", "desc"],
1283            registry
1284        )
1285        .unwrap()
1286        .relabel_debug_1(level);
1287
1288        let gap_fill_generated_rows_count = register_guarded_int_counter_vec_with_registry!(
1289            "gap_fill_generated_rows_count",
1290            "Total number of rows generated by gap fill executor",
1291            &["actor_id", "fragment_id"],
1292            registry
1293        )
1294        .unwrap()
1295        .relabel_debug_1(level);
1296
1297        let state_table_iter_count = register_guarded_int_counter_vec_with_registry!(
1298            "state_table_iter_count",
1299            "Total number of state table iter operations",
1300            &["actor_id", "fragment_id", "table_id"],
1301            registry
1302        )
1303        .unwrap()
1304        .relabel_debug_1(level);
1305
1306        let state_table_get_count = register_guarded_int_counter_vec_with_registry!(
1307            "state_table_get_count",
1308            "Total number of state table get operations",
1309            &["actor_id", "fragment_id", "table_id"],
1310            registry
1311        )
1312        .unwrap()
1313        .relabel_debug_1(level);
1314
1315        let state_table_iter_vnode_pruned_count = register_guarded_int_counter_vec_with_registry!(
1316            "state_table_iter_vnode_pruned_count",
1317            "Total number of state table iter operations pruned by vnode statistics",
1318            &["actor_id", "fragment_id", "table_id"],
1319            registry
1320        )
1321        .unwrap()
1322        .relabel_debug_1(level);
1323
1324        let state_table_get_vnode_pruned_count = register_guarded_int_counter_vec_with_registry!(
1325            "state_table_get_vnode_pruned_count",
1326            "Total number of state table get operations pruned by vnode statistics",
1327            &["actor_id", "fragment_id", "table_id"],
1328            registry
1329        )
1330        .unwrap()
1331        .relabel_debug_1(level);
1332
1333        Self {
1334            level,
1335            executor_row_count,
1336            mem_stream_node_output_row_count: stream_node_output_row_count,
1337            mem_stream_node_output_blocking_duration_ns: stream_node_output_blocking_duration_ns,
1338            actor_scheduled_duration,
1339            actor_scheduled_cnt,
1340            actor_poll_duration,
1341            actor_poll_cnt,
1342            actor_idle_duration,
1343            actor_idle_cnt,
1344            actor_count,
1345            actor_in_record_cnt,
1346            actor_out_record_cnt,
1347            fragment_channel_buffered_bytes,
1348            actor_current_epoch,
1349            project_expr_inflight_window_size,
1350            source_output_row_count,
1351            source_split_change_count,
1352            source_backfill_row_count,
1353            sink_input_row_count,
1354            sink_input_bytes,
1355            sink_chunk_buffer_size,
1356            exchange_frag_recv_size,
1357            merge_barrier_align_duration,
1358            actor_output_buffer_blocking_duration_ns,
1359            actor_input_buffer_blocking_duration_ns,
1360            join_lookup_miss_count,
1361            join_lookup_total_count,
1362            join_insert_cache_miss_count,
1363            join_actor_input_waiting_duration_ns,
1364            join_match_duration_ns,
1365            join_cached_entry_count,
1366            join_matched_join_keys,
1367            barrier_align_duration,
1368            agg_lookup_miss_count,
1369            agg_total_lookup_count,
1370            agg_cached_entry_count,
1371            agg_chunk_lookup_miss_count,
1372            agg_chunk_total_lookup_count,
1373            agg_dirty_groups_count,
1374            agg_dirty_groups_heap_size,
1375            agg_distinct_cache_miss_count,
1376            agg_distinct_total_cache_count,
1377            agg_distinct_cached_entry_count,
1378            agg_state_cache_lookup_count,
1379            agg_state_cache_miss_count,
1380            group_top_n_cache_miss_count,
1381            group_top_n_total_query_cache_count,
1382            group_top_n_cached_entry_count,
1383            group_top_n_appendonly_cache_miss_count,
1384            group_top_n_appendonly_total_query_cache_count,
1385            group_top_n_appendonly_cached_entry_count,
1386            lookup_cache_miss_count,
1387            lookup_total_query_cache_count,
1388            lookup_cached_entry_count,
1389            temporal_join_cache_miss_count,
1390            temporal_join_total_query_cache_count,
1391            temporal_join_cached_entry_count,
1392            backfill_snapshot_read_row_count,
1393            backfill_upstream_output_row_count,
1394            cdc_backfill_snapshot_read_row_count,
1395            cdc_backfill_upstream_output_row_count,
1396            snapshot_backfill_consume_row_count,
1397            over_window_cached_entry_count,
1398            over_window_cache_lookup_count,
1399            over_window_cache_miss_count,
1400            over_window_range_cache_entry_count,
1401            over_window_range_cache_lookup_count,
1402            over_window_range_cache_left_miss_count,
1403            over_window_range_cache_right_miss_count,
1404            over_window_accessed_entry_count,
1405            over_window_compute_count,
1406            over_window_same_output_count,
1407            barrier_inflight_latency,
1408            barrier_sync_latency,
1409            barrier_batch_size,
1410            barrier_manager_progress,
1411            kv_log_store_storage_write_count,
1412            kv_log_store_storage_write_size,
1413            kv_log_store_rewind_count,
1414            kv_log_store_rewind_delay,
1415            kv_log_store_storage_read_count,
1416            kv_log_store_storage_read_size,
1417            kv_log_store_buffer_unconsumed_item_count,
1418            kv_log_store_buffer_unconsumed_row_count,
1419            kv_log_store_buffer_unconsumed_epoch_count,
1420            kv_log_store_buffer_unconsumed_min_epoch,
1421            kv_log_store_buffer_memory_bytes,
1422            crossdb_last_consumed_min_epoch,
1423            sync_kv_log_store_read_count,
1424            sync_kv_log_store_read_size,
1425            sync_kv_log_store_write_pause_duration_ns,
1426            sync_kv_log_store_state,
1427            sync_kv_log_store_wait_next_poll_ns,
1428            sync_kv_log_store_storage_write_count,
1429            sync_kv_log_store_storage_write_size,
1430            sync_kv_log_store_buffer_unconsumed_item_count,
1431            sync_kv_log_store_buffer_unconsumed_row_count,
1432            sync_kv_log_store_buffer_unconsumed_epoch_count,
1433            sync_kv_log_store_buffer_unconsumed_min_epoch,
1434            sync_kv_log_store_buffer_memory_bytes,
1435            lru_runtime_loop_count,
1436            lru_latest_sequence,
1437            lru_watermark_sequence,
1438            lru_eviction_policy,
1439            jemalloc_allocated_bytes,
1440            jemalloc_active_bytes,
1441            jemalloc_resident_bytes,
1442            jemalloc_metadata_bytes,
1443            jvm_allocated_bytes,
1444            jvm_active_bytes,
1445            stream_memory_usage,
1446            materialize_cache_hit_count,
1447            materialize_data_exist_count,
1448            materialize_cache_total_count,
1449            materialize_input_row_count,
1450            materialize_current_epoch,
1451            pg_cdc_state_table_lsn,
1452            pg_cdc_jni_commit_offset_lsn,
1453            mysql_cdc_state_binlog_file_seq,
1454            mysql_cdc_state_binlog_position,
1455            sqlserver_cdc_state_change_lsn,
1456            sqlserver_cdc_state_commit_lsn,
1457            sqlserver_cdc_jni_commit_offset_lsn,
1458            gap_fill_generated_rows_count,
1459            state_table_iter_count,
1460            state_table_get_count,
1461            state_table_iter_vnode_pruned_count,
1462            state_table_get_vnode_pruned_count,
1463        }
1464    }
1465
1466    /// Create a new `StreamingMetrics` instance used in tests or other places.
1467    pub fn unused() -> Self {
1468        global_streaming_metrics(MetricLevel::Disabled)
1469    }
1470
1471    pub fn new_actor_metrics(&self, actor_id: ActorId, fragment_id: FragmentId) -> ActorMetrics {
1472        let label_list: &[&str; 2] = &[&actor_id.to_string(), &fragment_id.to_string()];
1473        let actor_scheduled_duration = self
1474            .actor_scheduled_duration
1475            .with_guarded_label_values(label_list);
1476        let actor_scheduled_cnt = self
1477            .actor_scheduled_cnt
1478            .with_guarded_label_values(label_list);
1479        let actor_poll_duration = self
1480            .actor_poll_duration
1481            .with_guarded_label_values(label_list);
1482        let actor_poll_cnt = self.actor_poll_cnt.with_guarded_label_values(label_list);
1483        let actor_idle_duration = self
1484            .actor_idle_duration
1485            .with_guarded_label_values(label_list);
1486        let actor_idle_cnt = self.actor_idle_cnt.with_guarded_label_values(label_list);
1487        ActorMetrics {
1488            actor_scheduled_duration,
1489            actor_scheduled_cnt,
1490            actor_poll_duration,
1491            actor_poll_cnt,
1492            actor_idle_duration,
1493            actor_idle_cnt,
1494        }
1495    }
1496
1497    pub(crate) fn new_actor_input_metrics(
1498        &self,
1499        actor_id: ActorId,
1500        fragment_id: FragmentId,
1501        upstream_fragment_id: FragmentId,
1502    ) -> ActorInputMetrics {
1503        let actor_id_str = actor_id.to_string();
1504        let fragment_id_str = fragment_id.to_string();
1505        let upstream_fragment_id_str = upstream_fragment_id.to_string();
1506        ActorInputMetrics {
1507            actor_in_record_cnt: self.actor_in_record_cnt.with_guarded_label_values(&[
1508                &actor_id_str,
1509                &fragment_id_str,
1510                &upstream_fragment_id_str,
1511            ]),
1512            actor_input_buffer_blocking_duration_ns: self
1513                .actor_input_buffer_blocking_duration_ns
1514                .with_guarded_label_values(&[
1515                    &actor_id_str,
1516                    &fragment_id_str,
1517                    &upstream_fragment_id_str,
1518                ]),
1519        }
1520    }
1521
1522    pub fn new_sink_exec_metrics(
1523        &self,
1524        id: SinkId,
1525        actor_id: ActorId,
1526        fragment_id: FragmentId,
1527    ) -> SinkExecutorMetrics {
1528        let label_list: &[&str; 3] = &[
1529            &id.to_string(),
1530            &actor_id.to_string(),
1531            &fragment_id.to_string(),
1532        ];
1533        SinkExecutorMetrics {
1534            sink_input_row_count: self
1535                .sink_input_row_count
1536                .with_guarded_label_values(label_list),
1537            sink_input_bytes: self.sink_input_bytes.with_guarded_label_values(label_list),
1538            sink_chunk_buffer_size: self
1539                .sink_chunk_buffer_size
1540                .with_guarded_label_values(label_list),
1541        }
1542    }
1543
1544    pub fn new_group_top_n_metrics(
1545        &self,
1546        table_id: TableId,
1547        actor_id: ActorId,
1548        fragment_id: FragmentId,
1549    ) -> GroupTopNMetrics {
1550        let label_list: &[&str; 3] = &[
1551            &table_id.to_string(),
1552            &actor_id.to_string(),
1553            &fragment_id.to_string(),
1554        ];
1555
1556        GroupTopNMetrics {
1557            group_top_n_cache_miss_count: self
1558                .group_top_n_cache_miss_count
1559                .with_guarded_label_values(label_list),
1560            group_top_n_total_query_cache_count: self
1561                .group_top_n_total_query_cache_count
1562                .with_guarded_label_values(label_list),
1563            group_top_n_cached_entry_count: self
1564                .group_top_n_cached_entry_count
1565                .with_guarded_label_values(label_list),
1566        }
1567    }
1568
1569    pub fn new_append_only_group_top_n_metrics(
1570        &self,
1571        table_id: TableId,
1572        actor_id: ActorId,
1573        fragment_id: FragmentId,
1574    ) -> GroupTopNMetrics {
1575        let label_list: &[&str; 3] = &[
1576            &table_id.to_string(),
1577            &actor_id.to_string(),
1578            &fragment_id.to_string(),
1579        ];
1580
1581        GroupTopNMetrics {
1582            group_top_n_cache_miss_count: self
1583                .group_top_n_appendonly_cache_miss_count
1584                .with_guarded_label_values(label_list),
1585            group_top_n_total_query_cache_count: self
1586                .group_top_n_appendonly_total_query_cache_count
1587                .with_guarded_label_values(label_list),
1588            group_top_n_cached_entry_count: self
1589                .group_top_n_appendonly_cached_entry_count
1590                .with_guarded_label_values(label_list),
1591        }
1592    }
1593
1594    pub fn new_lookup_executor_metrics(
1595        &self,
1596        table_id: TableId,
1597        actor_id: ActorId,
1598        fragment_id: FragmentId,
1599    ) -> LookupExecutorMetrics {
1600        let label_list: &[&str; 3] = &[
1601            &table_id.to_string(),
1602            &actor_id.to_string(),
1603            &fragment_id.to_string(),
1604        ];
1605
1606        LookupExecutorMetrics {
1607            lookup_cache_miss_count: self
1608                .lookup_cache_miss_count
1609                .with_guarded_label_values(label_list),
1610            lookup_total_query_cache_count: self
1611                .lookup_total_query_cache_count
1612                .with_guarded_label_values(label_list),
1613            lookup_cached_entry_count: self
1614                .lookup_cached_entry_count
1615                .with_guarded_label_values(label_list),
1616        }
1617    }
1618
1619    pub fn new_hash_agg_metrics(
1620        &self,
1621        table_id: TableId,
1622        actor_id: ActorId,
1623        fragment_id: FragmentId,
1624    ) -> HashAggMetrics {
1625        let label_list: &[&str; 3] = &[
1626            &table_id.to_string(),
1627            &actor_id.to_string(),
1628            &fragment_id.to_string(),
1629        ];
1630        HashAggMetrics {
1631            agg_lookup_miss_count: self
1632                .agg_lookup_miss_count
1633                .with_guarded_label_values(label_list),
1634            agg_total_lookup_count: self
1635                .agg_total_lookup_count
1636                .with_guarded_label_values(label_list),
1637            agg_cached_entry_count: self
1638                .agg_cached_entry_count
1639                .with_guarded_label_values(label_list),
1640            agg_chunk_lookup_miss_count: self
1641                .agg_chunk_lookup_miss_count
1642                .with_guarded_label_values(label_list),
1643            agg_chunk_total_lookup_count: self
1644                .agg_chunk_total_lookup_count
1645                .with_guarded_label_values(label_list),
1646            agg_dirty_groups_count: self
1647                .agg_dirty_groups_count
1648                .with_guarded_label_values(label_list),
1649            agg_dirty_groups_heap_size: self
1650                .agg_dirty_groups_heap_size
1651                .with_guarded_label_values(label_list),
1652            agg_state_cache_lookup_count: self
1653                .agg_state_cache_lookup_count
1654                .with_guarded_label_values(label_list),
1655            agg_state_cache_miss_count: self
1656                .agg_state_cache_miss_count
1657                .with_guarded_label_values(label_list),
1658        }
1659    }
1660
1661    pub fn new_agg_distinct_dedup_metrics(
1662        &self,
1663        table_id: TableId,
1664        actor_id: ActorId,
1665        fragment_id: FragmentId,
1666    ) -> AggDistinctDedupMetrics {
1667        let label_list: &[&str; 3] = &[
1668            &table_id.to_string(),
1669            &actor_id.to_string(),
1670            &fragment_id.to_string(),
1671        ];
1672        AggDistinctDedupMetrics {
1673            agg_distinct_cache_miss_count: self
1674                .agg_distinct_cache_miss_count
1675                .with_guarded_label_values(label_list),
1676            agg_distinct_total_cache_count: self
1677                .agg_distinct_total_cache_count
1678                .with_guarded_label_values(label_list),
1679            agg_distinct_cached_entry_count: self
1680                .agg_distinct_cached_entry_count
1681                .with_guarded_label_values(label_list),
1682        }
1683    }
1684
1685    pub fn new_temporal_join_metrics(
1686        &self,
1687        table_id: TableId,
1688        actor_id: ActorId,
1689        fragment_id: FragmentId,
1690    ) -> TemporalJoinMetrics {
1691        let label_list: &[&str; 3] = &[
1692            &table_id.to_string(),
1693            &actor_id.to_string(),
1694            &fragment_id.to_string(),
1695        ];
1696        TemporalJoinMetrics {
1697            temporal_join_cache_miss_count: self
1698                .temporal_join_cache_miss_count
1699                .with_guarded_label_values(label_list),
1700            temporal_join_total_query_cache_count: self
1701                .temporal_join_total_query_cache_count
1702                .with_guarded_label_values(label_list),
1703            temporal_join_cached_entry_count: self
1704                .temporal_join_cached_entry_count
1705                .with_guarded_label_values(label_list),
1706        }
1707    }
1708
1709    pub fn new_backfill_metrics(&self, table_id: TableId, actor_id: ActorId) -> BackfillMetrics {
1710        let label_list: &[&str; 2] = &[&table_id.to_string(), &actor_id.to_string()];
1711        BackfillMetrics {
1712            backfill_snapshot_read_row_count: self
1713                .backfill_snapshot_read_row_count
1714                .with_guarded_label_values(label_list),
1715            backfill_upstream_output_row_count: self
1716                .backfill_upstream_output_row_count
1717                .with_guarded_label_values(label_list),
1718        }
1719    }
1720
1721    pub fn new_cdc_backfill_metrics(
1722        &self,
1723        table_id: TableId,
1724        actor_id: ActorId,
1725    ) -> CdcBackfillMetrics {
1726        let label_list: &[&str; 2] = &[&table_id.to_string(), &actor_id.to_string()];
1727        CdcBackfillMetrics {
1728            cdc_backfill_snapshot_read_row_count: self
1729                .cdc_backfill_snapshot_read_row_count
1730                .with_guarded_label_values(label_list),
1731            cdc_backfill_upstream_output_row_count: self
1732                .cdc_backfill_upstream_output_row_count
1733                .with_guarded_label_values(label_list),
1734        }
1735    }
1736
1737    pub fn new_over_window_metrics(
1738        &self,
1739        table_id: TableId,
1740        actor_id: ActorId,
1741        fragment_id: FragmentId,
1742    ) -> OverWindowMetrics {
1743        let label_list: &[&str; 3] = &[
1744            &table_id.to_string(),
1745            &actor_id.to_string(),
1746            &fragment_id.to_string(),
1747        ];
1748        OverWindowMetrics {
1749            over_window_cached_entry_count: self
1750                .over_window_cached_entry_count
1751                .with_guarded_label_values(label_list),
1752            over_window_cache_lookup_count: self
1753                .over_window_cache_lookup_count
1754                .with_guarded_label_values(label_list),
1755            over_window_cache_miss_count: self
1756                .over_window_cache_miss_count
1757                .with_guarded_label_values(label_list),
1758            over_window_range_cache_entry_count: self
1759                .over_window_range_cache_entry_count
1760                .with_guarded_label_values(label_list),
1761            over_window_range_cache_lookup_count: self
1762                .over_window_range_cache_lookup_count
1763                .with_guarded_label_values(label_list),
1764            over_window_range_cache_left_miss_count: self
1765                .over_window_range_cache_left_miss_count
1766                .with_guarded_label_values(label_list),
1767            over_window_range_cache_right_miss_count: self
1768                .over_window_range_cache_right_miss_count
1769                .with_guarded_label_values(label_list),
1770            over_window_accessed_entry_count: self
1771                .over_window_accessed_entry_count
1772                .with_guarded_label_values(label_list),
1773            over_window_compute_count: self
1774                .over_window_compute_count
1775                .with_guarded_label_values(label_list),
1776            over_window_same_output_count: self
1777                .over_window_same_output_count
1778                .with_guarded_label_values(label_list),
1779        }
1780    }
1781
1782    pub fn new_materialize_cache_metrics(
1783        &self,
1784        table_id: TableId,
1785        actor_id: ActorId,
1786        fragment_id: FragmentId,
1787    ) -> MaterializeCacheMetrics {
1788        let label_list: &[&str; 3] = &[
1789            &actor_id.to_string(),
1790            &table_id.to_string(),
1791            &fragment_id.to_string(),
1792        ];
1793        MaterializeCacheMetrics {
1794            materialize_cache_hit_count: self
1795                .materialize_cache_hit_count
1796                .with_guarded_label_values(label_list),
1797            materialize_data_exist_count: self
1798                .materialize_data_exist_count
1799                .with_guarded_label_values(label_list),
1800            materialize_cache_total_count: self
1801                .materialize_cache_total_count
1802                .with_guarded_label_values(label_list),
1803        }
1804    }
1805
1806    pub fn new_materialize_metrics(
1807        &self,
1808        table_id: TableId,
1809        actor_id: ActorId,
1810        fragment_id: FragmentId,
1811    ) -> MaterializeMetrics {
1812        let label_list: &[&str; 3] = &[
1813            &actor_id.to_string(),
1814            &table_id.to_string(),
1815            &fragment_id.to_string(),
1816        ];
1817        MaterializeMetrics {
1818            materialize_input_row_count: self
1819                .materialize_input_row_count
1820                .with_guarded_label_values(label_list),
1821            materialize_current_epoch: self
1822                .materialize_current_epoch
1823                .with_guarded_label_values(label_list),
1824        }
1825    }
1826
1827    pub fn new_state_table_metrics(
1828        &self,
1829        table_id: TableId,
1830        actor_id: ActorId,
1831        fragment_id: FragmentId,
1832    ) -> StateTableMetrics {
1833        let label_list: &[&str; 3] = &[
1834            &actor_id.to_string(),
1835            &fragment_id.to_string(),
1836            &table_id.to_string(),
1837        ];
1838        StateTableMetrics {
1839            iter_count: self
1840                .state_table_iter_count
1841                .with_guarded_label_values(label_list),
1842            get_count: self
1843                .state_table_get_count
1844                .with_guarded_label_values(label_list),
1845            iter_vnode_pruned_count: self
1846                .state_table_iter_vnode_pruned_count
1847                .with_guarded_label_values(label_list),
1848            get_vnode_pruned_count: self
1849                .state_table_get_vnode_pruned_count
1850                .with_guarded_label_values(label_list),
1851        }
1852    }
1853}
1854
1855pub(crate) struct ActorInputMetrics {
1856    pub(crate) actor_in_record_cnt: LabelGuardedIntCounter,
1857    pub(crate) actor_input_buffer_blocking_duration_ns: LabelGuardedIntCounter,
1858}
1859
1860/// Tokio metrics for actors
1861pub struct ActorMetrics {
1862    pub actor_scheduled_duration: LabelGuardedIntCounter,
1863    pub actor_scheduled_cnt: LabelGuardedIntCounter,
1864    pub actor_poll_duration: LabelGuardedIntCounter,
1865    pub actor_poll_cnt: LabelGuardedIntCounter,
1866    pub actor_idle_duration: LabelGuardedIntCounter,
1867    pub actor_idle_cnt: LabelGuardedIntCounter,
1868}
1869
1870pub struct SinkExecutorMetrics {
1871    pub sink_input_row_count: LabelGuardedIntCounter,
1872    pub sink_input_bytes: LabelGuardedIntCounter,
1873    pub sink_chunk_buffer_size: LabelGuardedIntGauge,
1874}
1875
1876pub struct MaterializeCacheMetrics {
1877    pub materialize_cache_hit_count: LabelGuardedIntCounter,
1878    pub materialize_data_exist_count: LabelGuardedIntCounter,
1879    pub materialize_cache_total_count: LabelGuardedIntCounter,
1880}
1881
1882pub struct MaterializeMetrics {
1883    pub materialize_input_row_count: LabelGuardedIntCounter,
1884    pub materialize_current_epoch: LabelGuardedIntGauge,
1885}
1886
1887pub struct GroupTopNMetrics {
1888    pub group_top_n_cache_miss_count: LabelGuardedIntCounter,
1889    pub group_top_n_total_query_cache_count: LabelGuardedIntCounter,
1890    pub group_top_n_cached_entry_count: LabelGuardedIntGauge,
1891}
1892
1893pub struct LookupExecutorMetrics {
1894    pub lookup_cache_miss_count: LabelGuardedIntCounter,
1895    pub lookup_total_query_cache_count: LabelGuardedIntCounter,
1896    pub lookup_cached_entry_count: LabelGuardedIntGauge,
1897}
1898
1899pub struct HashAggMetrics {
1900    pub agg_lookup_miss_count: LabelGuardedIntCounter,
1901    pub agg_total_lookup_count: LabelGuardedIntCounter,
1902    pub agg_cached_entry_count: LabelGuardedIntGauge,
1903    pub agg_chunk_lookup_miss_count: LabelGuardedIntCounter,
1904    pub agg_chunk_total_lookup_count: LabelGuardedIntCounter,
1905    pub agg_dirty_groups_count: LabelGuardedIntGauge,
1906    pub agg_dirty_groups_heap_size: LabelGuardedIntGauge,
1907    pub agg_state_cache_lookup_count: LabelGuardedIntCounter,
1908    pub agg_state_cache_miss_count: LabelGuardedIntCounter,
1909}
1910
1911pub struct AggDistinctDedupMetrics {
1912    pub agg_distinct_cache_miss_count: LabelGuardedIntCounter,
1913    pub agg_distinct_total_cache_count: LabelGuardedIntCounter,
1914    pub agg_distinct_cached_entry_count: LabelGuardedIntGauge,
1915}
1916
1917pub struct TemporalJoinMetrics {
1918    pub temporal_join_cache_miss_count: LabelGuardedIntCounter,
1919    pub temporal_join_total_query_cache_count: LabelGuardedIntCounter,
1920    pub temporal_join_cached_entry_count: LabelGuardedIntGauge,
1921}
1922
1923pub struct BackfillMetrics {
1924    pub backfill_snapshot_read_row_count: LabelGuardedIntCounter,
1925    pub backfill_upstream_output_row_count: LabelGuardedIntCounter,
1926}
1927
1928#[derive(Clone)]
1929pub struct CdcBackfillMetrics {
1930    pub cdc_backfill_snapshot_read_row_count: LabelGuardedIntCounter,
1931    pub cdc_backfill_upstream_output_row_count: LabelGuardedIntCounter,
1932}
1933
1934pub struct OverWindowMetrics {
1935    pub over_window_cached_entry_count: LabelGuardedIntGauge,
1936    pub over_window_cache_lookup_count: LabelGuardedIntCounter,
1937    pub over_window_cache_miss_count: LabelGuardedIntCounter,
1938    pub over_window_range_cache_entry_count: LabelGuardedIntGauge,
1939    pub over_window_range_cache_lookup_count: LabelGuardedIntCounter,
1940    pub over_window_range_cache_left_miss_count: LabelGuardedIntCounter,
1941    pub over_window_range_cache_right_miss_count: LabelGuardedIntCounter,
1942    pub over_window_accessed_entry_count: LabelGuardedIntCounter,
1943    pub over_window_compute_count: LabelGuardedIntCounter,
1944    pub over_window_same_output_count: LabelGuardedIntCounter,
1945}
1946
1947#[derive(Clone)]
1948pub struct StateTableMetrics {
1949    pub iter_count: LabelGuardedIntCounter,
1950    pub get_count: LabelGuardedIntCounter,
1951    pub iter_vnode_pruned_count: LabelGuardedIntCounter,
1952    pub get_vnode_pruned_count: LabelGuardedIntCounter,
1953}