fix(metrics): source throughput should be of each source executor (#1…

…0059)
risingwavelabs · May 30, 2023 · 6cb4387 · 6cb4387
1 parent 2b04ed5
commit 6cb4387
Show file tree

Hide file tree

Showing 11 changed files with 17 additions and 40 deletions.
diff --git a/docker/dashboards/risingwave-dev-dashboard.json b/docker/dashboards/risingwave-dev-dashboard.json
diff --git a/docker/dashboards/risingwave-user-dashboard.json b/docker/dashboards/risingwave-user-dashboard.json
diff --git a/grafana/common.py b/grafana/common.py
@@ -100,6 +100,7 @@ class Panels:
         "fillOpacity": 10,
         "interval": "1s",
         "maxDataPoints": 1000,
+        "legendDisplayMode": "table",
     }
 
     def __init__(self, datasource):
@@ -141,7 +142,6 @@ def timeseries(self, title, description, targets):
             description=description,
             targets=targets,
             gridPos=gridPos,
-            legendDisplayMode="table",
             **self.common_options,
         )
 
@@ -158,7 +158,6 @@ def timeseries_count(self,
             targets=targets,
             gridPos=gridPos,
             legendCalcs=legendCols,
-            legendDisplayMode="table",
             **self.common_options,
         )
 
@@ -177,7 +176,6 @@ def timeseries_percentage(self,
             gridPos=gridPos,
             unit="percentunit",
             legendCalcs=legendCols,
-            legendDisplayMode="table",
             **self.common_options,
         )
 
@@ -195,7 +193,6 @@ def timeseries_latency(self,
             gridPos=gridPos,
             unit="s",
             legendCalcs=legendCols,
-            legendDisplayMode="table",
             **self.common_options,
         )
 
@@ -213,7 +210,6 @@ def timeseries_actor_latency(self,
             gridPos=gridPos,
             unit="s",
             legendCalcs=legendCols,
-            legendDisplayMode="table",
             **self.common_options,
         )
 
@@ -231,7 +227,6 @@ def timeseries_actor_latency_small(self,
             gridPos=gridPos,
             unit="s",
             legendCalcs=legendCols,
-            legendDisplayMode="table",
             **self.common_options,
         )
 
@@ -249,7 +244,6 @@ def timeseries_query_per_sec(self,
             gridPos=gridPos,
             unit="Qps",
             legendCalcs=legendCols,
-            legendDisplayMode="table",
             **self.common_options,
         )
 
@@ -267,7 +261,6 @@ def timeseries_bytes_per_sec(self,
             gridPos=gridPos,
             unit="Bps",
             legendCalcs=legendCols,
-            legendDisplayMode="table",
             **self.common_options,
         )
 
@@ -285,7 +278,6 @@ def timeseries_bytes(self,
             gridPos=gridPos,
             unit="bytes",
             legendCalcs=legendCols,
-            legendDisplayMode="table",
             **self.common_options,
         )
 
@@ -299,7 +291,6 @@ def timeseries_row(self, title, description, targets, legendCols=["mean"]):
             gridPos=gridPos,
             unit="row",
             legendCalcs=legendCols,
-            legendDisplayMode="table",
             **self.common_options,
         )
 
@@ -312,7 +303,6 @@ def timeseries_ms(self, title, description, targets, legendCols=["mean"]):
             targets=targets,
             gridPos=gridPos,
             legendCalcs=legendCols,
-            legendDisplayMode="table",
             **self.common_options,
         )
 
@@ -330,7 +320,6 @@ def timeseries_kilobytes(self,
             gridPos=gridPos,
             unit="kbytes",
             legendCalcs=legendCols,
-            legendDisplayMode="table",
             **self.common_options,
         )
 
@@ -348,7 +337,6 @@ def timeseries_dollar(self,
             gridPos=gridPos,
             unit="$",
             legendCalcs=legendCols,
-            legendDisplayMode="table",
             **self.common_options,
         )
 
@@ -362,7 +350,6 @@ def timeseries_ops(self, title, description, targets, legendCols=["mean"]):
             gridPos=gridPos,
             unit="ops",
             legendCalcs=legendCols,
-            legendDisplayMode="table",
             **self.common_options,
         )
 
@@ -380,7 +367,6 @@ def timeseries_actor_ops(self,
             gridPos=gridPos,
             unit="ops",
             legendCalcs=legendCols,
-            legendDisplayMode="table",
             **self.common_options,
         )
 
@@ -398,7 +384,6 @@ def timeseries_actor_ops_small(self,
             gridPos=gridPos,
             unit="ops",
             legendCalcs=legendCols,
-            legendDisplayMode="table",
             **self.common_options,
         )
 
@@ -416,7 +401,6 @@ def timeseries_rowsps(self,
             gridPos=gridPos,
             unit="rows/s",
             legendCalcs=legendCols,
-            legendDisplayMode="table",
             **self.common_options,
         )
 
@@ -434,7 +418,6 @@ def timeseries_bytesps(self,
             gridPos=gridPos,
             unit="MB/s",
             legendCalcs=legendCols,
-            legendDisplayMode="table",
             **self.common_options,
         )
 
@@ -447,7 +430,6 @@ def timeseries_actor_rowsps(self, title, description, targets):
             targets=targets,
             gridPos=gridPos,
             unit="rows/s",
-            legendDisplayMode="table",
             **self.common_options,
         )
 
@@ -460,7 +442,6 @@ def timeseries_memory(self, title, description, targets):
             targets=targets,
             gridPos=gridPos,
             unit="bytes",
-            legendDisplayMode="table",
             **self.common_options,
         )
 
@@ -473,7 +454,6 @@ def timeseries_cpu(self, title, description, targets):
             targets=targets,
             gridPos=gridPos,
             unit="percentunit",
-            legendDisplayMode="table",
             **self.common_options,
         )
 
@@ -486,7 +466,6 @@ def timeseries_latency_small(self, title, description, targets):
             targets=targets,
             gridPos=gridPos,
             unit="s",
-            legendDisplayMode="table",
             **self.common_options,
         )
 
@@ -498,7 +477,6 @@ def timeseries_id(self, title, description, targets):
             description=description,
             targets=targets,
             gridPos=gridPos,
-            legendDisplayMode="table",
             **self.common_options,
         )
 

diff --git a/grafana/risingwave-dev-dashboard.dashboard.py b/grafana/risingwave-dev-dashboard.dashboard.py
@@ -605,7 +605,7 @@ def section_streaming(panels):
             [
                 panels.target(
                     f"rate({metric('stream_source_output_rows_counts')}[$__rate_interval])",
-                    "source={{source_name}} {{source_id}} @ {{instance}}",
+                    "source={{source_name}} actor={{actor_id}} @ {{instance}}",
                 ),
             ],
         ),

diff --git a/grafana/risingwave-dev-dashboard.json b/grafana/risingwave-dev-dashboard.json
diff --git a/grafana/risingwave-user-dashboard.dashboard.py b/grafana/risingwave-user-dashboard.dashboard.py
@@ -591,7 +591,7 @@ def section_streaming(outer_panels):
                     [
                         panels.target(
                             f"rate({metric('stream_source_output_rows_counts')}[$__rate_interval])",
-                            "source={{source_name}} {{source_id}} @ {{instance}}",
+                            "source={{source_name}} actor={{actor_id}} @ {{instance}}",
                         ),
                     ],
                 ),

diff --git a/grafana/risingwave-user-dashboard.json b/grafana/risingwave-user-dashboard.json
diff --git a/src/stream/src/executor/monitor/streaming_stats.rs b/src/stream/src/executor/monitor/streaming_stats.rs
@@ -132,7 +132,7 @@ impl StreamingMetrics {
         let source_output_row_count = register_int_counter_vec_with_registry!(
             "stream_source_output_rows_counts",
             "Total number of rows that have been output from source",
-            &["source_id", "source_name"],
+            &["source_id", "source_name", "actor_id"],
             registry
         )
         .unwrap();

diff --git a/src/stream/src/executor/source/executor_core.rs b/src/stream/src/executor/source/executor_core.rs
@@ -29,8 +29,6 @@ pub struct StreamSourceCore<S: StateStore> {
 
     pub(crate) column_ids: Vec<ColumnId>,
 
-    pub(crate) source_identify: String,
-
     /// `source_desc_builder` will be taken (`mem::take`) on execution. A `SourceDesc` (currently
     /// named `SourceDescV2`) will be constructed and used for execution.
     pub(crate) source_desc_builder: Option<SourceDescBuilder>,
@@ -61,7 +59,6 @@ where
             source_id,
             source_name,
             column_ids,
-            source_identify: "Table_".to_string() + &source_id.table_id().to_string(),
             source_desc_builder: Some(source_desc_builder),
             stream_source_splits: HashMap::new(),
             split_state_store,

diff --git a/src/stream/src/executor/source/fs_source_executor.rs b/src/stream/src/executor/source/fs_source_executor.rs
@@ -383,7 +383,7 @@ impl<S: StateStore> FsSourceExecutor<S> {
                             .source_row_per_barrier
                             .with_label_values(&[
                                 self.ctx.id.to_string().as_str(),
-                                self.stream_source_core.source_identify.as_ref(),
+                                self.stream_source_core.source_id.to_string().as_ref(),
                             ])
                             .inc_by(metric_row_per_barrier);
                         metric_row_per_barrier = 0;
@@ -426,8 +426,9 @@ impl<S: StateStore> FsSourceExecutor<S> {
                     self.metrics
                         .source_output_row_count
                         .with_label_values(&[
-                            self.stream_source_core.source_identify.as_str(),
+                            self.stream_source_core.source_id.to_string().as_ref(),
                             self.stream_source_core.source_name.as_ref(),
+                            self.ctx.id.to_string().as_str(),
                         ])
                         .inc_by(chunk.cardinality() as u64);
                     yield Message::Chunk(chunk);

diff --git a/src/stream/src/executor/source/source_executor.rs b/src/stream/src/executor/source/source_executor.rs
@@ -432,7 +432,8 @@ impl<S: StateStore> SourceExecutor<S> {
                                 self.stream_source_core
                                     .as_ref()
                                     .unwrap()
-                                    .source_identify
+                                    .source_id
+                                    .to_string()
                                     .as_ref(),
                             ])
                             .inc_by(metric_row_per_barrier);
@@ -494,13 +495,15 @@ impl<S: StateStore> SourceExecutor<S> {
                             self.stream_source_core
                                 .as_ref()
                                 .unwrap()
-                                .source_identify
-                                .as_str(),
+                                .source_id
+                                .to_string()
+                                .as_ref(),
                             self.stream_source_core
                                 .as_ref()
                                 .unwrap()
                                 .source_name
                                 .as_ref(),
+                            self.ctx.id.to_string().as_str(),
                         ])
                         .inc_by(chunk.cardinality() as u64);
                     yield Message::Chunk(chunk);
@@ -630,7 +633,6 @@ mod tests {
         let core = StreamSourceCore::<MemoryStateStore> {
             source_id: table_id,
             column_ids,
-            source_identify: "Table_".to_string() + &table_id.table_id().to_string(),
             source_desc_builder: Some(source_desc_builder),
             stream_source_splits: HashMap::new(),
             split_state_store,
@@ -718,7 +720,6 @@ mod tests {
         let core = StreamSourceCore::<MemoryStateStore> {
             source_id: table_id,
             column_ids: column_ids.clone(),
-            source_identify: "Table_".to_string() + &table_id.table_id().to_string(),
             source_desc_builder: Some(source_desc_builder),
             stream_source_splits: HashMap::new(),
             split_state_store,