refactor(metrics): display actor-level metrics based on configuration

Signed-off-by: Bugen Zhao <[email protected]>
risingwavelabs · Aug 30, 2024 · 606bb71 · 606bb71
1 parent 04619f6
commit 606bb71
Show file tree

Hide file tree

Showing 5 changed files with 29 additions and 15 deletions.
diff --git a/docker/dashboards/risingwave-dev-dashboard.json b/docker/dashboards/risingwave-dev-dashboard.json
diff --git a/e2e_test/nexmark/create_sources.slt.part b/e2e_test/nexmark/create_sources.slt.part
@@ -36,7 +36,7 @@ CREATE SOURCE nexmark (
 ) WITH (
     connector = 'nexmark',
     nexmark.split.num = '2',
-    nexmark.min.event.gap.in.ns = '100'
+    nexmark.min.event.gap.in.ns = '100000000'
 );
 
 statement ok

diff --git a/grafana/risingwave-dev-dashboard.dashboard.py b/grafana/risingwave-dev-dashboard.dashboard.py
@@ -1121,7 +1121,8 @@ def section_streaming_cdc(outer_panels):
     ]
 
 
-def section_streaming_actors(outer_panels):
+def section_streaming_actors(outer_panels: Panels):
+    actor_level_filter = "actor_id!=''"
     panels = outer_panels.sub_panel()
     return [
         outer_panels.row_collapsed(
@@ -1159,10 +1160,10 @@ def section_streaming_actors(outer_panels):
                     [
                         panels.target(
                             f"sum(rate({metric('stream_actor_in_record_cnt')}[$__rate_interval])) by (fragment_id, upstream_fragment_id)",
-                            "fragment {{fragment_id}}<-{{upstream_fragment_id}}",
+                            "fragment total {{fragment_id}}<-{{upstream_fragment_id}}",
                         ),
-                        panels.target_hidden(
-                            f"rate({metric('stream_actor_in_record_cnt')}[$__rate_interval])",
+                        panels.target(
+                            f"rate({metric('stream_actor_in_record_cnt', actor_level_filter)}[$__rate_interval])",
                             "actor {{actor_id}}",
                         ),
                     ],
@@ -1173,10 +1174,10 @@ def section_streaming_actors(outer_panels):
                     [
                         panels.target(
                             f"sum(rate({metric('stream_actor_out_record_cnt')}[$__rate_interval])) by (fragment_id)",
-                            "fragment {{fragment_id}}",
+                            "fragment total {{fragment_id}}",
                         ),
-                        panels.target_hidden(
-                            f"rate({metric('stream_actor_out_record_cnt')}[$__rate_interval])",
+                        panels.target(
+                            f"rate({metric('stream_actor_out_record_cnt', actor_level_filter)}[$__rate_interval])",
                             "actor {{actor_id}}",
                         ),
                     ],
@@ -1189,9 +1190,9 @@ def section_streaming_actors(outer_panels):
                             f"sum({metric('stream_memory_usage')}) by (table_id, desc)",
                             "table {{table_id}} desc: {{desc}}",
                         ),
-                        panels.target_hidden(
-                            f"{metric('stream_memory_usage')}",
-                            "table {{table_id}} actor {{actor_id}} desc: {{desc}}",
+                        panels.target(
+                            f"{metric('stream_memory_usage', actor_level_filter)}",
+                            "actor {{actor_id}} table {{table_id}} desc: {{desc}}",
                         ),
                     ],
                 ),

diff --git a/grafana/risingwave-dev-dashboard.json b/grafana/risingwave-dev-dashboard.json
diff --git a/src/stream/src/executor/monitor/streaming_stats.rs b/src/stream/src/executor/monitor/streaming_stats.rs
@@ -65,8 +65,8 @@ pub struct StreamingMetrics {
     pub actor_count: LabelGuardedIntGaugeVec<1>,
     #[expect(dead_code)]
     actor_memory_usage: LabelGuardedIntGaugeVec<2>,
-    actor_in_record_cnt: LabelGuardedIntCounterVec<3>,
-    pub actor_out_record_cnt: LabelGuardedIntCounterVec<2>,
+    actor_in_record_cnt: RelabeledGuardedIntCounterVec<3>,
+    pub actor_out_record_cnt: RelabeledGuardedIntCounterVec<2>,
 
     // Source
     pub source_output_row_count: LabelGuardedIntCounterVec<4>,
@@ -396,6 +396,12 @@ impl StreamingMetrics {
             registry
         )
         .unwrap();
+        let actor_in_record_cnt = RelabeledGuardedIntCounterVec::with_metric_level_relabel_n(
+            MetricLevel::Debug,
+            actor_in_record_cnt,
+            level,
+            1,
+        );
 
         let actor_out_record_cnt = register_guarded_int_counter_vec_with_registry!(
             "stream_actor_out_record_cnt",
@@ -404,6 +410,12 @@ impl StreamingMetrics {
             registry
         )
         .unwrap();
+        let actor_out_record_cnt = RelabeledGuardedIntCounterVec::with_metric_level_relabel_n(
+            MetricLevel::Debug,
+            actor_out_record_cnt,
+            level,
+            1,
+        );
 
         let actor_count = register_guarded_int_gauge_vec_with_registry!(
             "stream_actor_count",
@@ -413,6 +425,7 @@ impl StreamingMetrics {
         )
         .unwrap();
 
+        // dead code
         let actor_memory_usage = register_guarded_int_gauge_vec_with_registry!(
             "actor_memory_usage",
             "Memory usage (bytes)",