cherry-pick refactor(metrics): use counter instead of histogram for barrier align duration

grafana/risingwave-dev-dashboard.dashboard.py

@@ -1210,21 +1210,17 @@ def section_streaming_actors(outer_panels):
                         ),
                     ],
                 ),
-                panels.timeseries_actor_latency(
-                    "Executor Barrier Align",
+                panels.timeseries_percentage(
+                    "Executor Barrier Align Per Second",
                     "",
                     [
-                        *quantile(
-                            lambda quantile, legend: panels.target(
-                                f"histogram_quantile({quantile}, sum(rate({metric('stream_barrier_align_duration_bucket')}[$__rate_interval])) by (le, executor, fragment_id, wait_side, {COMPONENT_LABEL}))",
-                                f"p{legend} - executor {{{{executor}}}} fragment {{{{fragment_id}}}} {{{{wait_side}}}} - {{{{{COMPONENT_LABEL}}}}}",
-                            ),
-                            [90, 99, 999, "max"],
-                        ),
                         panels.target(
-                            f"sum by(le, executor, fragment_id, wait_side, job)(rate({metric('stream_barrier_align_duration_sum')}[$__rate_interval])) / sum by(le,executor,fragment_id,wait_side,{COMPONENT_LABEL}) (rate({metric('stream_barrier_align_duration_count')}[$__rate_interval])) > 0",
-                            "avg - executor {{executor}} fragment {{fragment_id}} {{wait_side}} - {{%s}}"
-                            % COMPONENT_LABEL,
+                            f"avg(rate({metric('stream_barrier_align_duration_ns')}[$__rate_interval]) / 1000000000) by (fragment_id,wait_side, executor)",
+                            "fragment {{fragment_id}} {{wait_side}} {{executor}}",
+                        ),
+                        panels.target_hidden(
+                            f"rate({metric('stream_barrier_align_duration_ns')}[$__rate_interval]) / 1000000000",
+                            "actor {{actor_id}} fragment {{fragment_id}} {{wait_side}} {{executor}}",
                         ),
                     ],
                 ),

src/stream/src/executor/barrier_align.rs

@@ -121,7 +121,7 @@ pub async fn barrier_align(
                         Message::Barrier(barrier) => {
                             yield AlignedMessage::Barrier(barrier);
                             right_barrier_align_duration
-                                .observe(start_time.elapsed().as_secs_f64());
+                                .inc_by(start_time.elapsed().as_nanos() as u64);
                             break;
                         }
                     }
@@ -144,7 +144,8 @@ pub async fn barrier_align(
                         Message::Chunk(chunk) => yield AlignedMessage::Left(chunk),
                         Message::Barrier(barrier) => {
                             yield AlignedMessage::Barrier(barrier);
-                            left_barrier_align_duration.observe(start_time.elapsed().as_secs_f64());
+                            left_barrier_align_duration
+                                .inc_by(start_time.elapsed().as_nanos() as u64);
                             break;
                         }
                     }

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

@@ -24,7 +24,7 @@ use risingwave_common::config::MetricLevel;
 use risingwave_common::metrics::{
     LabelGuardedGauge, LabelGuardedGaugeVec, LabelGuardedHistogramVec, LabelGuardedIntCounter,
     LabelGuardedIntCounterVec, LabelGuardedIntGauge, LabelGuardedIntGaugeVec,
-    RelabeledGuardedHistogramVec,
+    RelabeledGuardedHistogramVec, RelabeledGuardedIntCounterVec,
 };
 use risingwave_common::monitor::GLOBAL_METRICS_REGISTRY;
 use risingwave_common::{
@@ -96,7 +96,7 @@ pub struct StreamingMetrics {
     pub join_matched_join_keys: RelabeledGuardedHistogramVec<3>,
 
     // Streaming Join, Streaming Dynamic Filter and Streaming Union
-    pub barrier_align_duration: RelabeledGuardedHistogramVec<4>,
+    pub barrier_align_duration: RelabeledGuardedIntCounterVec<4>,
 
     // Streaming Aggregation
     agg_lookup_miss_count: LabelGuardedIntCounterVec<3>,
@@ -465,19 +465,15 @@ impl StreamingMetrics {
         )
         .unwrap();
 
-        let opts = histogram_opts!(
-            "stream_barrier_align_duration",
+        let barrier_align_duration = register_guarded_int_counter_vec_with_registry!(
+            "stream_barrier_align_duration_ns",
             "Duration of join align barrier",
-            exponential_buckets(0.0001, 2.0, 21).unwrap() // max 104s
-        );
-        let barrier_align_duration = register_guarded_histogram_vec_with_registry!(
-            opts,
             &["actor_id", "fragment_id", "wait_side", "executor"],
             registry
         )
         .unwrap();
 
-        let barrier_align_duration = RelabeledGuardedHistogramVec::with_metric_level_relabel_n(
+        let barrier_align_duration = RelabeledGuardedIntCounterVec::with_metric_level_relabel_n(
             MetricLevel::Debug,
             barrier_align_duration,
             level,

src/stream/src/executor/union.rs

@@ -164,7 +164,7 @@ async fn merge(
             None => {
                 assert!(active.is_terminated());
                 let barrier = current_barrier.take().unwrap();
-                barrier_align.observe(start_time.elapsed().as_secs_f64());
+                barrier_align.inc_by(start_time.elapsed().as_nanos() as u64);
 
                 let upstreams = std::mem::take(&mut blocked);
                 active.extend(upstreams.into_iter().map(|upstream| upstream.into_future()));