feat(metrics): expose sink error in grafana and factor error metrics (#…

…13083)
risingwavelabs · Oct 31, 2023 · b11c831 · b11c831
1 parent 8a90ce5
commit b11c831
Show file tree

Hide file tree

Showing 16 changed files with 348 additions and 164 deletions.
diff --git a/docker/dashboards/risingwave-dev-dashboard.json b/docker/dashboards/risingwave-dev-dashboard.json
diff --git a/grafana/risingwave-dev-dashboard.dashboard.py b/grafana/risingwave-dev-dashboard.dashboard.py
@@ -1518,6 +1518,10 @@ def section_streaming_errors(outer_panels):
                             f"sum({metric('user_compute_error_count')}) by (error_type, error_msg, fragment_id, executor_name)",
                             "{{error_type}}: {{error_msg}} ({{executor_name}}: fragment_id={{fragment_id}})",
                         ),
+                        panels.target(
+                            f"sum({metric('user_compute_error')}) by (error_type, error_msg, fragment_id, executor_name)",
+                            "{{error_type}}: {{error_msg}} ({{executor_name}}: fragment_id={{fragment_id}})",
+                        ),
                     ],
                 ),
                 panels.timeseries_count(
@@ -1528,6 +1532,10 @@ def section_streaming_errors(outer_panels):
                             f"sum({metric('user_source_error_count')}) by (error_type, error_msg, fragment_id, table_id, executor_name)",
                             "{{error_type}}: {{error_msg}} ({{executor_name}}: table_id={{table_id}}, fragment_id={{fragment_id}})",
                         ),
+                        panels.target(
+                            f"sum({metric('user_source_error')}) by (error_type, error_msg, fragment_id, table_id, executor_name)",
+                            "{{error_type}}: {{error_msg}} ({{executor_name}}: table_id={{table_id}}, fragment_id={{fragment_id}})",
+                        ),
                     ],
                 ),
                 panels.timeseries_count(
@@ -1538,6 +1546,20 @@ def section_streaming_errors(outer_panels):
                             f"sum({metric('user_source_reader_error_count')}) by (error_type, error_msg, actor_id, source_id, executor_name)",
                             "{{error_type}}: {{error_msg}} ({{executor_name}}: actor_id={{actor_id}}, source_id={{source_id}})",
                         ),
+                        panels.target(
+                            f"sum({metric('user_source_reader_error')}) by (error_type, error_msg, actor_id, source_id, executor_name)",
+                            "{{error_type}}: {{error_msg}} ({{executor_name}}: actor_id={{actor_id}}, source_id={{source_id}})",
+                        ),
+                    ],
+                ),
+                panels.timeseries_count(
+                    "Sink by Connector",
+                    "",
+                    [
+                        panels.target(
+                            f"sum({metric('user_sink_error')}) by (connector_name, executor_id, error_msg)",
+                            "{{connector_name}}: {{error_msg}} ({{executor_id}})",
+                        ),
                     ],
                 ),
             ],

diff --git a/grafana/risingwave-dev-dashboard.json b/grafana/risingwave-dev-dashboard.json
diff --git a/java/com_risingwave_java_binding_Binding.h b/java/com_risingwave_java_binding_Binding.h
diff --git a/...nnector-service/src/main/java/com/risingwave/connector/JniSinkWriterResponseObserver.java b/...nnector-service/src/main/java/com/risingwave/connector/JniSinkWriterResponseObserver.java
@@ -39,6 +39,8 @@ public void onNext(ConnectorServiceProto.SinkWriterStreamResponse response) {
 
     @Override
     public void onError(Throwable throwable) {
+        this.success =
+                Binding.sendSinkWriterErrorToChannel(this.responseTxPtr, throwable.getMessage());
         LOG.error("JniSinkWriterHandler onError: ", throwable);
     }
 

diff --git a/java/java-binding/src/main/java/com/risingwave/java/binding/Binding.java b/java/java-binding/src/main/java/com/risingwave/java/binding/Binding.java
@@ -83,6 +83,8 @@ public class Binding {
 
     public static native boolean sendSinkWriterResponseToChannel(long channelPtr, byte[] msg);
 
+    public static native boolean sendSinkWriterErrorToChannel(long channelPtr, String msg);
+
     public static native byte[] recvSinkCoordinatorRequestFromChannel(long channelPtr);
 
     public static native boolean sendSinkCoordinatorResponseToChannel(long channelPtr, byte[] msg);

diff --git a/src/common/src/metrics.rs b/src/common/src/metrics.rs
@@ -25,9 +25,11 @@ use tracing_subscriber::Layer;
 
 use crate::monitor::GLOBAL_METRICS_REGISTRY;
 
+mod error_metrics;
 mod guarded_metrics;
 mod relabeled_metric;
 
+pub use error_metrics::*;
 pub use guarded_metrics::*;
 pub use relabeled_metric::*;