feat(stream): source_error_count reporting to prometheus

grafana/risingwave-dashboard.dashboard.py

@@ -1461,18 +1461,28 @@ def section_streaming_errors(outer_panels):
     panels = outer_panels.sub_panel()
     return [
         outer_panels.row_collapsed(
-            "Streaming Errors",
+            "User Streaming Errors",
             [
                 panels.timeseries_count(
-                    "User Errors by Type",
+                    "Compute Errors by Type",
                     "",
                     [
                         panels.target(
-                            f"sum({metric('user_error_count')}) by (error_type, error_msg, fragment_id, executor_name)",
+                            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.timeseries_count(
+                    "Source Errors by Type",
+                    "",
+                    [
+                        panels.target(
+                            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}})",
+                        ),
+                    ],
+                ),
             ],
         ),
     ]

src/batch/src/executor/source.rs

@@ -161,7 +161,7 @@ impl SourceExecutor {
                 Some(vec![self.split]),
                 self.column_ids,
                 self.metrics,
-                SourceInfo::new(u32::MAX, self.source_id),
+                SourceInfo::new(u32::MAX, self.source_id, u32::MAX),
             )
             .await?;
 

src/connector/src/macros.rs

@@ -182,6 +182,7 @@ macro_rules! impl_common_parser_logic {
                             if let Err(e) = self.parse_inner(content.as_ref(), builder.row_writer())
                                 .await
                             {
+                                self.error_ctx.report_stream_source_error(&e);
                                 tracing::warn!("message parsing failed {}, skipping", e.to_string());
                                 continue;
                             }
@@ -241,6 +242,11 @@ macro_rules! impl_common_split_reader_logic {
                 let source_id = self.source_info.source_id.to_string();
                 let split_id = self.split_id.clone();
                 let metrics = self.metrics.clone();
+                let error_ctx = $crate::source::base::SourceErrorContext::new(
+                    self.source_info.source_id.table_id,
+                    self.source_info.fragment_id,
+                    self.metrics.clone()
+                );
 
                 let data_stream = self.into_data_stream();
 
@@ -265,7 +271,7 @@ macro_rules! impl_common_split_reader_logic {
                     })
                     .boxed();
                 let parser =
-                    $crate::parser::ByteStreamSourceParserImpl::create(parser_config)?;
+                    $crate::parser::ByteStreamSourceParserImpl::create(parser_config, error_ctx)?;
                 #[for_await]
                 for msg_batch in parser.into_stream(data_stream) {
                     yield msg_batch?;

src/connector/src/parser/avro/parser.rs

@@ -31,7 +31,7 @@ use crate::parser::avro::util::avro_field_to_column_desc;
 use crate::parser::schema_registry::{extract_schema_id, Client};
 use crate::parser::util::get_kafka_topic;
 use crate::parser::{SourceStreamChunkRowWriter, WriteGuard};
-use crate::source::SourceColumnDesc;
+use crate::source::{SourceColumnDesc, SourceErrorContext};
 
 impl_common_parser_logic!(AvroParser);
 
@@ -40,6 +40,7 @@ pub struct AvroParser {
     schema: Arc<Schema>,
     schema_resolver: Option<Arc<ConfluentSchemaResolver>>,
     rw_columns: Vec<SourceColumnDesc>,
+    error_ctx: SourceErrorContext,
 }
 
 #[derive(Debug, Clone)]
@@ -104,7 +105,11 @@ impl AvroParserConfig {
 
 // confluent_wire_format, kafka only, subject-name: "${topic-name}-value"
 impl AvroParser {
-    pub fn new(rw_columns: Vec<SourceColumnDesc>, config: AvroParserConfig) -> Result<Self> {
+    pub fn new(
+        rw_columns: Vec<SourceColumnDesc>,
+        config: AvroParserConfig,
+        error_ctx: SourceErrorContext,
+    ) -> Result<Self> {
         let AvroParserConfig {
             schema,
             schema_resolver,
@@ -113,6 +118,7 @@ impl AvroParser {
             schema,
             schema_resolver,
             rw_columns,
+            error_ctx,
         })
     }
 
@@ -196,7 +202,7 @@ mod test {
     };
     use crate::parser::avro::util::unix_epoch_days;
     use crate::parser::SourceStreamChunkBuilder;
-    use crate::source::SourceColumnDesc;
+    use crate::source::{SourceColumnDesc, SourceErrorContext};
 
     fn test_data_path(file_name: &str) -> String {
         let curr_dir = env::current_dir().unwrap().into_os_string();
@@ -254,7 +260,7 @@ mod test {
 
     async fn new_avro_parser_from_local(file_name: &str) -> error::Result<AvroParser> {
         let conf = new_avro_conf_from_local(file_name).await?;
-        AvroParser::new(Vec::default(), conf)
+        AvroParser::new(Vec::default(), conf, SourceErrorContext::for_test())
     }
 
     #[tokio::test]

src/connector/src/parser/canal/simd_json_parser.rs

@@ -28,7 +28,7 @@ use simd_json::{BorrowedValue, StaticNode, ValueAccess};
 use crate::parser::canal::operators::*;
 use crate::parser::util::at_least_one_ok;
 use crate::parser::{SourceStreamChunkRowWriter, WriteGuard};
-use crate::source::SourceColumnDesc;
+use crate::source::{SourceColumnDesc, SourceErrorContext};
 use crate::{ensure_rust_type, ensure_str, impl_common_parser_logic};
 
 const AFTER: &str = "data";
@@ -40,11 +40,15 @@ impl_common_parser_logic!(CanalJsonParser);
 #[derive(Debug)]
 pub struct CanalJsonParser {
     pub(crate) rw_columns: Vec<SourceColumnDesc>,
+    error_ctx: SourceErrorContext,
 }
 
 impl CanalJsonParser {
-    pub fn new(rw_columns: Vec<SourceColumnDesc>) -> Result<Self> {
-        Ok(Self { rw_columns })
+    pub fn new(rw_columns: Vec<SourceColumnDesc>, error_ctx: SourceErrorContext) -> Result<Self> {
+        Ok(Self {
+            rw_columns,
+            error_ctx,
+        })
     }
 
     #[allow(clippy::unused_async)]
@@ -255,7 +259,7 @@ mod tests {
             SourceColumnDesc::simple("win_rate", DataType::Float64, 5.into()),
         ];
 
-        let parser = CanalJsonParser::new(descs.clone()).unwrap();
+        let parser = CanalJsonParser::new(descs.clone(), SourceErrorContext::for_test()).unwrap();
 
         let mut builder = SourceStreamChunkBuilder::with_capacity(descs, 2);
 
@@ -332,7 +336,7 @@ mod tests {
             SourceColumnDesc::simple("v2", DataType::Int32, 1.into()),
         ];
 
-        let parser = CanalJsonParser::new(descs.clone()).unwrap();
+        let parser = CanalJsonParser::new(descs.clone(), SourceErrorContext::for_test()).unwrap();
 
         let mut builder = SourceStreamChunkBuilder::with_capacity(descs, 2);
 

src/connector/src/parser/csv_parser.rs

@@ -28,7 +28,7 @@ use crate::parser::{
     BoxSourceWithStateStream, ByteStreamSourceParser, SourceColumnDesc, SourceStreamChunkBuilder,
     SourceStreamChunkRowWriter, StreamChunkWithState, WriteGuard,
 };
-use crate::source::{BoxSourceStream, SplitId};
+use crate::source::{BoxSourceStream, SourceErrorContext, SplitId};
 
 macro_rules! to_rust_type {
     ($v:ident, $t:ty) => {
@@ -53,10 +53,15 @@ pub struct CsvParser {
     output_cursor: usize,
     ends: Vec<usize>,
     ends_cursor: usize,
+    error_ctx: SourceErrorContext,
 }
 
 impl CsvParser {
-    pub fn new(rw_columns: Vec<SourceColumnDesc>, parser_config: CsvParserConfig) -> Result<Self> {
+    pub fn new(
+        rw_columns: Vec<SourceColumnDesc>,
+        parser_config: CsvParserConfig,
+        error_ctx: SourceErrorContext,
+    ) -> Result<Self> {
         let CsvParserConfig {
             delimiter,
             has_header,
@@ -70,6 +75,7 @@ impl CsvParser {
             output_cursor: 0,
             ends: vec![0],
             ends_cursor: 1,
+            error_ctx,
         })
     }
 
@@ -316,7 +322,7 @@ mod tests {
             delimiter: b',',
             has_header: true,
         };
-        let parser = CsvParser::new(descs, config).unwrap();
+        let parser = CsvParser::new(descs, config, SourceErrorContext::for_test()).unwrap();
         let data = b"
 name,age
 pite,20
@@ -341,7 +347,7 @@ alex,10";
             delimiter: b',',
             has_header: true,
         };
-        let parser = CsvParser::new(descs, config).unwrap();
+        let parser = CsvParser::new(descs, config, SourceErrorContext::for_test()).unwrap();
         let data = b"
 name,age
 pite,20

src/connector/src/parser/debezium/avro_parser.rs

@@ -35,7 +35,7 @@ use crate::parser::schema_registry::{extract_schema_id, Client};
 use crate::parser::schema_resolver::ConfluentSchemaResolver;
 use crate::parser::util::get_kafka_topic;
 use crate::parser::{SourceStreamChunkRowWriter, WriteGuard};
-use crate::source::SourceColumnDesc;
+use crate::source::{SourceColumnDesc, SourceErrorContext};
 
 const BEFORE: &str = "before";
 const AFTER: &str = "after";
@@ -51,6 +51,7 @@ pub struct DebeziumAvroParser {
     inner_schema: Arc<Schema>,
     schema_resolver: Arc<ConfluentSchemaResolver>,
     rw_columns: Vec<SourceColumnDesc>,
+    error_ctx: SourceErrorContext,
 }
 
 #[derive(Debug, Clone)]
@@ -167,6 +168,7 @@ impl DebeziumAvroParser {
     pub fn new(
         rw_columns: Vec<SourceColumnDesc>,
         config: DebeziumAvroParserConfig,
+        error_ctx: SourceErrorContext,
     ) -> Result<Self> {
         let DebeziumAvroParserConfig {
             outer_schema,
@@ -179,6 +181,7 @@ impl DebeziumAvroParser {
             inner_schema,
             schema_resolver,
             rw_columns,
+            error_ctx,
         })
     }
 
@@ -435,7 +438,8 @@ mod tests {
             .map(|c| SourceColumnDesc::from(&c))
             .collect_vec();
 
-        let parser = DebeziumAvroParser::new(columns.clone(), config)?;
+        let parser =
+            DebeziumAvroParser::new(columns.clone(), config, SourceErrorContext::for_test())?;
         let [(op, row)]: [_; 1] = parse_one(parser, columns, DEBEZIUM_AVRO_DATA)
             .await
             .try_into()