feat(sink): add sink throughput( mb/s) metrics

grafana/risingwave-dev-dashboard.dashboard.py

@@ -970,6 +970,26 @@ def section_streaming(outer_panels):
                         ),
                     ],
                 ),
+                panels.timeseries_bytesps(
+                    "Sink Throughput(MB/s)",
+                    "The figure shows the number of bytes written each sink per second. For sinks with 'sink_decouple = true', please refer to the 'Sink Metrics' section",
+                    [
+                        panels.target(
+                            f"(sum(rate({metric('stream_sink_input_size')}[$__rate_interval])) by (sink_id) * on(sink_id) group_left(sink_name) group({metric('sink_info')}) by (sink_id, sink_name)) / (1000*1000)",
+                            "sink {{sink_id}} {{sink_name}}",
+                        ),
+                    ],
+                ),
+                panels.timeseries_bytesps(
+                    "Sink Throughput(MB/s) per Partition *",
+                    "The number of bytes streamed into each sink per second. For sinks with 'sink_decouple = true', please refer to the 'Sink Metrics' section",
+                    [
+                        panels.target(
+                            f"(sum(rate({metric('stream_sink_input_size')}[$__rate_interval])) by (sink_id, actor_id) * on(actor_id) group_left(sink_name) {metric('sink_info')}) / (1000*1000)",
+                            "sink {{sink_id}} {{sink_name}} - actor {{actor_id}}",
+                        ),
+                    ],
+                ),
                 panels.timeseries_rowsps(
                     "Materialized View Throughput(rows/s)",
                     "The figure shows the number of rows written into each materialized view per second.",

risedev.yml

@@ -38,9 +38,9 @@ profile:
       #   database: metadata
 
       # If you want to enable metrics or tracing, uncomment the following lines.
-      # - use: prometheus  # metrics
-      # - use: tempo       # tracing
-      # - use: grafana     # visualization
+      - use: prometheus  # metrics
+      - use: tempo       # tracing
+      - use: grafana     # visualization
 
       - use: meta-node
         # meta-backend: postgres

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

@@ -73,6 +73,7 @@ pub struct StreamingMetrics {
 
     // Sink
     sink_input_row_count: LabelGuardedIntCounterVec<3>,
+    sink_input_size: LabelGuardedIntCounterVec<3>,
     sink_chunk_buffer_size: LabelGuardedIntGaugeVec<3>,
 
     // Exchange (see also `compute::ExchangeServiceMetrics`)
@@ -244,6 +245,14 @@ impl StreamingMetrics {
         )
         .unwrap();
 
+        let sink_input_size = register_guarded_int_counter_vec_with_registry!(
+            "stream_sink_input_size",
+            "Total size of chunks streamed into sink executors",
+            &["sink_id", "actor_id", "fragment_id"],
+            registry
+        )
+        .unwrap();
+
         let materialize_input_row_count = register_guarded_int_counter_vec_with_registry!(
             "stream_mview_input_row_count",
             "Total number of rows streamed into materialize executors",
@@ -1056,6 +1065,7 @@ impl StreamingMetrics {
             source_split_change_count,
             source_backfill_row_count,
             sink_input_row_count,
+            sink_input_size,
             sink_chunk_buffer_size,
             exchange_frag_recv_size,
             merge_barrier_align_duration,
@@ -1229,6 +1239,7 @@ impl StreamingMetrics {
             sink_input_row_count: self
                 .sink_input_row_count
                 .with_guarded_label_values(label_list),
+            sink_input_size: self.sink_input_size.with_guarded_label_values(label_list),
             sink_chunk_buffer_size: self
                 .sink_chunk_buffer_size
                 .with_guarded_label_values(label_list),
@@ -1520,6 +1531,7 @@ pub struct ActorMetrics {
 
 pub struct SinkExecutorMetrics {
     pub sink_input_row_count: LabelGuardedIntCounter<3>,
+    pub sink_input_size: LabelGuardedIntCounter<3>,
     pub sink_chunk_buffer_size: LabelGuardedIntGauge<3>,
 }
 

src/stream/src/executor/sink.rs

@@ -200,6 +200,7 @@ impl<F: LogStoreFactory> SinkExecutor<F> {
         let input = input.inspect_ok(move |msg| {
             if let Message::Chunk(c) = msg {
                 metrics.sink_input_row_count.inc_by(c.capacity() as u64);
+                metrics.sink_input_size.inc_by(c.estimated_size() as u64);
             }
         });