Repository: risingwavelabs/risingwave
File: src/stream/src/executor/monitor/streaming_stats.rs (Rust)

Code before:

#[derive(Clone)]
pub struct StreamingMetrics {
    pub level: MetricLevel,

    // Executor metrics (disabled by default)
    pub executor_row_count: RelabeledGuardedIntCounterVec,

    // Profiling Metrics:
    // Aggregated per operator rather than per actor.
    // These are purely in-memory, never collected by prometheus.
    pub mem_stream_node_output_row_count: CountMap,
    pub mem_stream_node_output_blocking_duration_ns: CountMap,

    // Streaming actor metrics from tokio (disabled by default)
    actor_execution_duration: RelabeledGuardedIntCounterVec,
    actor_scheduled_duration: RelabeledGuardedIntCounterVec,
    actor_scheduled_cnt: RelabeledGuardedIntCounterVec,
    actor_fast_poll_duration: RelabeledGuardedIntCounterVec,
    actor_fast_poll_cnt: RelabeledGuardedIntCounterVec,
    actor_slow_poll_duration: RelabeledGuardedIntCounterVec,
    actor_slow_poll_cnt: RelabeledGuardedIntCounterVec,
    actor_poll_duration: RelabeledGuardedIntCounterVec,
    actor_poll_cnt: RelabeledGuardedIntCounterVec,
    actor_idle_duration: RelabeledGuardedIntCounterVec,
    actor_idle_cnt: RelabeledGuardedIntCounterVec,

    // Streaming actor
    pub actor_count: LabelGuardedIntGaugeVec,
    pub actor_in_record_cnt: RelabeledGuardedIntCounterVec,
    pub actor_out_record_cnt: RelabeledGuardedIntCounterVec,
    pub actor_current_epoch: RelabeledGuardedIntGaugeVec,

    // Source
    pub source_output_row_count: LabelGuardedIntCounterVec,
    pub source_split_change_count: LabelGuardedIntCounterVec,
    pub source_backfill_row_count: LabelGuardedIntCounterVec,

    // Sink
    sink_input_row_count: LabelGuardedIntCounterVec,
    sink_input_bytes: LabelGuardedIntCounterVec,
    sink_chunk_buffer_size: LabelGuardedIntGaugeVec,

    // Exchange (see also `compute::ExchangeServiceMetrics`)
    pub exchange_frag_recv_size: LabelGuardedIntCounterVec,

    // Streaming Merge (We break out this metric from `barrier_align_duration` because
    // the alignment happens on different levels)
    pub merge_barrier_align_duration: RelabeledGuardedHistogramVec,

    // Backpressure

Review comment (bug):
We use `RelabeledGuardedIntCounterVec` to aggregate by fragment_id, when the metric level is lower than `debug`, e.g. `info`.

We also change these metrics from recording `Gauge` to `Counter`. This is because these metrics should only increase. Given that we will continually update these metrics, I think it's better to choose a more performant metric type.

Code after:

#[derive(Clone)]
pub struct StreamingMetrics {
    pub level: MetricLevel,

    // Executor metrics (disabled by default)
    pub executor_row_count: RelabeledGuardedIntCounterVec,

    // Profiling Metrics:
    // Aggregated per operator rather than per actor.
    // These are purely in-memory, never collected by prometheus.
    pub mem_stream_node_output_row_count: CountMap,
    pub mem_stream_node_output_blocking_duration_ns: CountMap,

    // Streaming actor metrics from tokio (disabled by default)
    actor_scheduled_duration: RelabeledGuardedIntCounterVec,
    actor_scheduled_cnt: RelabeledGuardedIntCounterVec,
    actor_poll_duration: RelabeledGuardedIntCounterVec,
    actor_poll_cnt: RelabeledGuardedIntCounterVec,
    actor_idle_duration: RelabeledGuardedIntCounterVec,
    actor_idle_cnt: RelabeledGuardedIntCounterVec,

    // Streaming actor
    pub actor_count: LabelGuardedIntGaugeVec,
    pub actor_in_record_cnt: RelabeledGuardedIntCounterVec,
    pub actor_out_record_cnt: RelabeledGuardedIntCounterVec,
    pub actor_current_epoch: RelabeledGuardedIntGaugeVec,

    // Source
    pub source_output_row_count: LabelGuardedIntCounterVec,
    pub source_split_change_count: LabelGuardedIntCounterVec,
    pub source_backfill_row_count: LabelGuardedIntCounterVec,

    // Sink
    sink_input_row_count: LabelGuardedIntCounterVec,
    sink_input_bytes: LabelGuardedIntCounterVec,
    sink_chunk_buffer_size: LabelGuardedIntGaugeVec,

    // Exchange (see also `compute::ExchangeServiceMetrics`)
    pub exchange_frag_recv_size: LabelGuardedIntCounterVec,

    // Streaming Merge (We break out this metric from `barrier_align_duration` because
    // the alignment happens on different levels)
    pub merge_barrier_align_duration: RelabeledGuardedHistogramVec,

    // Backpressure
    pub actor_output_buffer_blocking_duration_ns: RelabeledGuardedIntCounterVec,
    actor_input_buffer_blocking_duration_ns: RelabeledGuardedIntCounterVec,

    // Streaming Join
    pub join_lookup_miss_count: LabelGuardedIntCounterVec,