feat(connector): structural logging for parsing error

[BugenZhao]

Signed-off-by: Bugen Zhao [email protected]I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

See #12366 for the background.

The error-handling path in the connector parser is kind of complicated, and users often complain about the confusion of the behavior when parsing a message fails. We've had some attempts before (like #10454). In this PR, we further unify the error handling strategy and code path, then provide tracing observability on that.

Specifically,

• To write a record to the chunk builder, the only way is through SourceStreamChunkRowWriter. Based on a similar idea to refactor(connector): remove WriteGuard and fulfill_meta_column from parser #12542, we can also unify the error handling to the do_action method. This means that parser implementors don't have to care about error handling anymore.
• Follows the convention that whoever consumes the error is responsible for logging it. So there'll be redundant error messages in the log.
• Provide tracing spans for the parsing procedure, and log the errors in a structural way.

Take the logs for plain JSON parser as an example:

• If the message itself is an invalid JSON, an error will be produced at the very first step of parsing into a json::Value before any calling to Writer::{insert|delete|update}. Therefore, the error will be thrown and lead to the whole message being skipped.

2023-10-08T12:01:38.612275+08:00 ERROR 
actor{otel.name="Actor 4" actor_id=4 prev_epoch=5210273369423872 curr_epoch=5210273377091584}:..:
executor{otel.name="SourceExecutor 40000273C (actor 4)" actor_id=4}:
source_parser{actor_id=4 source_id=1001}:
parse_one{offset="0"}: 
risingwave_connector::parser: failed to parse message, skipping 
error=Protocol error: ExpectedObjectContent at character 1 ('v')

• If the message is valid, while some (non-primary key) fields are missing or in the wrong types, the error will be logged as a warning and ignored in Writer.

2023-10-08T12:04:15.503359+08:00  WARN 
actor{otel.name="Actor 4" actor_id=4 prev_epoch=5210273369423872 curr_epoch=5210273377091584}:..:
executor{otel.name="SourceExecutor 40000273C (actor 4)" actor_id=4}:
source_parser{actor_id=4 source_id=1001}:
parse_one{offset="2"}: 
risingwave_connector::parser: failed to parse non-pk column, padding with `NULL` 
error=Expected type `Some(Int32)` but got `string` for `bbbb` 
column="v1"

As a result, we may instruct users to filter the logs with the target risingwave_connector::parser and source_ids to inspect the messages that are failed to parse.

Checklist

• I have written necessary rustdoc comments
• I have added necessary unit tests and integration tests
• All checks passed in ./risedev check (or alias, ./risedev c)

Documentation

• My PR needs documentation updates. (Please use the Release note section below to summarize the impact on users)

Release note

If this PR includes changes that directly affect users or other significant modifications relevant to the community, kindly draft a release note to provide a concise summary of these changes. Please prioritize highlighting the impact these changes will have on users.

[fuyufjh]

Generally LGTM!

[fuyufjh]

As a result, we may instruct users to filter the logs with the target risingwave_connector::parser and source_ids to inspect the messages that are failed to parse.

I am wondering that is it possible to filter out all user-facing errors with some Loki query? Or shall we just consider all errors as user-facing errors? (The later one seems not bad 🤔)

[BugenZhao]

Or shall we just consider all errors as user-facing errors?

I think so. 👍

[codecov]

Codecov Report

Merging #12659 (79b5cb4) into main (437d36e) will decrease coverage by 0.08%.
Report is 5 commits behind head on main.
The diff coverage is 61.11%.

@@            Coverage Diff             @@
##             main   #12659      +/-   ##
==========================================
- Coverage   69.40%   69.33%   -0.08%     
==========================================
  Files        1470     1470              
  Lines      241531   241474      -57     
==========================================
- Hits       167637   167426     -211     
- Misses      73894    74048     +154     

Flag	Coverage Δ
rust	69.33% <61.11%> (-0.08%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files	Coverage Δ
...connector/src/parser/debezium/mongo_json_parser.rs	81.10% <100.00%> (ø)
src/connector/src/parser/maxwell/maxwell_parser.rs	55.81% <100.00%> (ø)
src/connector/src/parser/plain_parser.rs	56.41% <100.00%> (ø)
src/connector/src/parser/unified/mod.rs	87.50% <ø> (ø)
src/connector/src/parser/unified/util.rs	88.46% <100.00%> (+18.34%)	⬆️
src/connector/src/parser/upsert_parser.rs	67.85% <100.00%> (+0.38%)	⬆️
src/meta/src/barrier/mod.rs	83.19% <100.00%> (-0.05%)	⬇️
...c/connector/src/parser/debezium/debezium_parser.rs	46.34% <0.00%> (-3.66%)	⬇️
.../connector/src/parser/debezium/simd_json_parser.rs	97.22% <85.71%> (+0.31%)	⬆️
src/connector/src/parser/canal/simd_json_parser.rs	84.54% <50.00%> (-0.20%)	⬇️
... and 3 more

... and 23 files with indirect coverage changes

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more