feat: add source/destination/e2e commit lag metrics#4573
Conversation
- take clock skew into account for all source connectors
| // A keepalive means there is currently no source change event to lag behind. | ||
| // Reset the gauge so a previously backlogged, now-idle pipe does not retain | ||
| // the old source-lag value. | ||
| p.otelManager.Metrics.CommitLagGauge.Record(ctx, 0) |
There was a problem hiding this comment.
PG does reset the timeout countdown on data events, but when it's decoding a large tx or reading a large row from a throttled volume, it can still send these in between. When the source is backed up like this, this metric would be important in surfacing it. Debezium doesn't seem to be doing resets for PG at all, so maybe that's defensible, and I also wasn't able to find a solid one.
There was a problem hiding this comment.
yeah, but just removing this reset would lead to alerts in these cases: "previously backlogged, now-idle pipe".
wdyt about this approach?
if totalRecords == 0 && int64(clientXLogPos) > req.ConsumedOffset.Load() {
err := p.updateConsumedOffset(ctx, logger, req.FlowJobName, req.ConsumedOffset, clientXLogPos)
if err != nil {
return err
}
lastEmptyBatchPkmSentTime = time.Now()
// Caught up with no records for this flow: clear any stale source lag.
p.otelManager.Metrics.CommitLagGauge.Record(ctx, 0)
}
totalRecords == 0 && clientXLogPos > ConsumedOffset means precisely "this whole batch produced no records for our publication, yet our position advanced" — i.e. our pipe is idle-but-progressing.
and another defensive reset on the keep alive path:
if pkm.ServerWALEnd <= clientXLogPos {
p.otelManager.Metrics.CommitLagGauge.Record(ctx, 0)
}
this handles a case when we're fully caught up with ServerWALEnd. This complements previous check for the cases when ALL tables are idle.
There was a problem hiding this comment.
Will need to think a bit more, brain fried atm
There was a problem hiding this comment.
if pkm.ServerWALEnd <= clientXLogPos {
p.otelManager.Metrics.CommitLagGauge.Record(ctx, 0)
}
May be a bit confusing but ServerWALEnd is not WAL head but the current decoder position. It doesn't mean we've caught up
https://github.com/postgres/postgres/blob/8ac700acffc7b17d88414be47b8dff44fb1ea681/src/backend/replication/walsender.c#L3607-L3609
if totalRecords == 0 && int64(clientXLogPos) > req.ConsumedOffset.Load() {
err := p.updateConsumedOffset(ctx, logger, req.FlowJobName, req.ConsumedOffset, clientXLogPos)
if err != nil {
return err
}
lastEmptyBatchPkmSentTime = time.Now()
// Caught up with no records for this flow: clear any stale source lag.
p.otelManager.Metrics.CommitLagGauge.Record(ctx, 0)
}
totalRecords == 0 can happen when PG is terribly terribly slow decoding a large transaction event (low on logical_decoding_work_mem, overloaded CPU, out of EBS iops budget and the event is 1GB). It is possible that we receive a primary keepalive advancing the XLogPos with some events outside of the publication before that, or receive a random row outside of our tables, the large tx data would still be pending and possibly stale in this case.
For this PR let's remove the zeroing out PG part + requestReply to unblock the rest of the work. If we find some creative and working solution later on, it can be PR'd separately.
Co-authored-by: Ilia Demianenko <[email protected]>
❌ 8 Tests Failed:
View the top 3 failed test(s) by shortest run time
To view more test analytics, go to the Test Analytics Dashboard |
❌ Test FailureAnalysis: The mongo connector test package fails to compile because cdc_test.go:391 calls createPipeline with too few arguments (a signature mismatch), which is a deterministic build error, not a flaky test. |
❌ Test FailureAnalysis: This is a deterministic build failure — the mongo connector test fails to compile because a |
❌ Test FailureAnalysis: This is a deterministic Go build failure — cdc_test.go:391 calls createPipeline with too few arguments, so the mongo connector test package fails to compile, which is a real code bug rather than a flaky test. |
❌ Test FailureAnalysis: Not flaky — a deterministic build failure: cdc_test.go:391 calls createPipeline with too few arguments, breaking compilation of the mongo connector test package. |
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
❌ Test FailureAnalysis: A deterministic build failure — cdc_test.go:391 calls createPipeline with too few arguments — not a flaky test. |
…time are written to batch metadata
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
dtunikov
changed the title
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
🔄 Flaky Test DetectedAnalysis: A single isolated mongo-8.0-only failure where MongoDB reported oplog retention as 23h despite CI explicitly configuring --oplogMinRetentionHours 24, while the same test passed on the 6.0/7.0 jobs and the PR (commit-lag metrics) doesn't touch mongo validation — an environmental/startup timing flake, not a real bug. ✅ Automatically retrying the workflow |
Group time.Time fields before the string/slice so the GC pointer-scan region ends at the slice (88 bytes) instead of inside a time.Time (104). Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
| format!( | ||
| "cannot write value {:?} in postgres protocol: unimplemented", | ||
| &value | ||
| value |
There was a problem hiding this comment.
clippy failed on this one, i guess should be fixed on the main branch
Code reviewNo issues found. Checked for bugs and CLAUDE.md compliance. |
ilidemi
left a comment
ilidemi
left a comment
There was a problem hiding this comment.
Looking great. The only piece left is thinking through Postgres carefully
| // A keepalive means there is currently no source change event to lag behind. | ||
| // Reset the gauge so a previously backlogged, now-idle pipe does not retain | ||
| // the old source-lag value. | ||
| p.otelManager.Metrics.CommitLagGauge.Record(ctx, 0) |
There was a problem hiding this comment.
Will need to think a bit more, brain fried atm
❌ Test FailureAnalysis: Real bug: Test_MySQL_DateTime_ClickHouse_Range failed deterministically in all four suite variants with an identical value mismatch (expected clamped "1900-01-01", got unclamped "1000-06-15"), showing out-of-range MySQL dates are not being clamped to ClickHouse's boundary — with no timeout, race, or network signatures. |
❌ Test FailureAnalysis: Test_MySQL_DateTime_ClickHouse_Range fails deterministically across all four ClickHouse suites with an identical date value mismatch (expected "1900-01-01", got "1000-06-15"), indicating a real MySQL→ClickHouse datetime range-clamping bug rather than a flaky failure. |
Code reviewNo issues found. Checked for bugs and CLAUDE.md compliance. |
❌ Test FailureAnalysis: Not flaky — Test_MySQL_DateTime_ClickHouse_Range fails deterministically in all 8 suite variants with an identical assertion mismatch (expected "1900-01-01", got "1000-06-15"), only on the ClickHouse "latest" job, indicating a real date-range handling change/bug rather than an intermittent failure. |
| // A keepalive means there is currently no source change event to lag behind. | ||
| // Reset the gauge so a previously backlogged, now-idle pipe does not retain | ||
| // the old source-lag value. | ||
| p.otelManager.Metrics.CommitLagGauge.Record(ctx, 0) |
There was a problem hiding this comment.
if pkm.ServerWALEnd <= clientXLogPos {
p.otelManager.Metrics.CommitLagGauge.Record(ctx, 0)
}
May be a bit confusing but ServerWALEnd is not WAL head but the current decoder position. It doesn't mean we've caught up
https://github.com/postgres/postgres/blob/8ac700acffc7b17d88414be47b8dff44fb1ea681/src/backend/replication/walsender.c#L3607-L3609
if totalRecords == 0 && int64(clientXLogPos) > req.ConsumedOffset.Load() {
err := p.updateConsumedOffset(ctx, logger, req.FlowJobName, req.ConsumedOffset, clientXLogPos)
if err != nil {
return err
}
lastEmptyBatchPkmSentTime = time.Now()
// Caught up with no records for this flow: clear any stale source lag.
p.otelManager.Metrics.CommitLagGauge.Record(ctx, 0)
}
totalRecords == 0 can happen when PG is terribly terribly slow decoding a large transaction event (low on logical_decoding_work_mem, overloaded CPU, out of EBS iops budget and the event is 1GB). It is possible that we receive a primary keepalive advancing the XLogPos with some events outside of the publication before that, or receive a random row outside of our tables, the large tx data would still be pending and possibly stale in this case.
For this PR let's remove the zeroing out PG part + requestReply to unblock the rest of the work. If we find some creative and working solution later on, it can be PR'd separately.
2 participants
This PR adds new source/destination/e2e commit lag metrics.
Migration for cdc_batches table was added, it adds two new fields - first_row_received_at and first_row_commit_time. These fields allow us to report destination and e2e lag metrics.