Skip to content

Report processing vs receive vs add record time in CDC pull #4749

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ilidemi merged 3 commits into from
Aug 31, 2026
Merged

Report processing vs receive vs add record time in CDC pull #4749

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
219993b
Report processing vs receive vs add record time in CDC pull
ilidemi Aug 31, 2026
2695769
Merge branch 'main' into cdc-receive-process-time-metrics
ilidemi Aug 31, 2026
9695cb4
lint
ilidemi Aug 31, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions flow/connectors/mysql/cdc.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -594,6 +594,8 @@ func (c *MySqlConnector) PullRecords(
// set when a tx is preventing us from respecting the timeout, immediately exit after we see inTx false
var overtime bool
var fetchedBytes, totalFetchedBytes, allFetchedBytes atomic.Int64
var receiveTime, processTime, addRecordTime atomic.Int64
var processStart time.Time
pullStart := time.Now()
defer func() {
if recordCount == 0 {
Expand All @@ -617,10 +619,16 @@ func (c *MySqlConnector) PullRecords(
defer func() {
otelManager.Metrics.FetchedBytesCounter.Add(ctx, fetchedBytes.Swap(0))
otelManager.Metrics.AllFetchedBytesCounter.Add(ctx, allFetchedBytes.Swap(0))
otelManager.Metrics.CDCReceiveTimeCounter.Add(ctx, receiveTime.Swap(0))
otelManager.Metrics.CDCProcessTimeCounter.Add(ctx, processTime.Swap(0))
otelManager.Metrics.CDCAddRecordTimeCounter.Add(ctx, addRecordTime.Swap(0))
}()
shutdown := common.Interval(ctx, time.Minute, func() {
otelManager.Metrics.FetchedBytesCounter.Add(ctx, fetchedBytes.Swap(0))
otelManager.Metrics.AllFetchedBytesCounter.Add(ctx, allFetchedBytes.Swap(0))
otelManager.Metrics.CDCReceiveTimeCounter.Add(ctx, receiveTime.Swap(0))
otelManager.Metrics.CDCProcessTimeCounter.Add(ctx, processTime.Swap(0))
otelManager.Metrics.CDCAddRecordTimeCounter.Add(ctx, addRecordTime.Swap(0))
c.logger.Info("[mysql] pulling records",
slog.Uint64("records", uint64(recordCount)),
slog.Int64("bytes", totalFetchedBytes.Load()),
Expand All @@ -641,9 +649,13 @@ func (c *MySqlConnector) PullRecords(

addRecord := func(ctx context.Context, record model.Record[model.RecordItems]) error {
recordCount += 1
addStart := time.Now()
processTime.Add(int64(addStart.Sub(processStart)))

@dtunikov dtunikov Aug 31, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

hm, not sure I'm following this part
afaiu we want to deduct the time we spent in req.RecordStream.AddRecord from the processTime
isn't it smth like that? i just don't understand why we set processTime back to now() on every addRecord call 🤔

Suggested change
processTime.Add(int64(addStart.Sub(processStart)))
addStart := time.Now()
if err := req.RecordStream.AddRecord(ctx, record); err != nil {
return err
}
addTook := int64(time.Since(addStart))
addRecordTime.Add(addTook)
processTime.Add(-addTook);

@ilidemi ilidemi Aug 31, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

These are kind of two ways of achieving the same thing, just the PR one is without subtractions. If that's less clear, can do -addRecord as well. processTime is set back to now() here so that the next processTime.Add wouldn't count addRecord time

@ilidemi ilidemi Aug 31, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Looked a bit further, and alternative approaches either have gaps or introduce long-distance math between variables:

  • In the suggested version, processTime is undercounted until processTime.Add(int64(time.Since(processStart))) a few lines below and can get reported on timer
  • Alternative is to save the duration of last addRecords and subtract it at processTime.Add time but that's 300 lines of scrolling to match up variable names

In this case it's all contained to the transition points and just stops the old counter and starts the next one

if err := req.RecordStream.AddRecord(ctx, record); err != nil {
return err
}
processStart = time.Now()
addRecordTime.Add(int64(processStart.Sub(addStart)))
if recordCount == 1 {
req.RecordStream.SignalAsNotEmpty()
resetTimeout(req.IdleTimeout)
Expand Down Expand Up @@ -1054,7 +1066,9 @@ func (c *MySqlConnector) PullRecords(
// don't gamble on closed timeoutCtx.Done() being prioritized over event backlog channel
err := timeoutCtx.Err()
if err == nil {
receiveStart := time.Now()
event, err = mystream.GetEvent(timeoutCtx)
receiveTime.Add(int64(time.Since(receiveStart)))
}
if err != nil {
if ctxErr := ctx.Err(); ctxErr != nil {
Expand Down Expand Up @@ -1100,6 +1114,7 @@ func (c *MySqlConnector) PullRecords(
}

lastEventAt = time.Now()
processStart = lastEventAt

allFetchedBytes.Add(int64(len(event.RawData)))

Expand All @@ -1117,6 +1132,7 @@ func (c *MySqlConnector) PullRecords(
return err
}
}
processTime.Add(int64(time.Since(processStart)))
}
return nil
}
Expand Down
26 changes: 24 additions & 2 deletions flow/connectors/postgres/cdc.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -567,6 +567,8 @@ func PullCdcRecords[Items model.Items](
warnedReplIdentTables := make(map[string]struct{})
var totalRecords int64
var fetchedBytes, totalFetchedBytes, allFetchedBytes atomic.Int64
var receiveTime, processTime, addRecordTime atomic.Int64
var processStart time.Time
// clientXLogPos is the last checkpoint id, we need to ack that we have processed
// until clientXLogPos each time we send a standby status update.
var clientXLogPos pglogrepl.LSN
Expand Down Expand Up @@ -615,10 +617,16 @@ func PullCdcRecords[Items model.Items](
defer func() {
p.otelManager.Metrics.FetchedBytesCounter.Add(ctx, fetchedBytes.Swap(0))
p.otelManager.Metrics.AllFetchedBytesCounter.Add(ctx, allFetchedBytes.Swap(0))
p.otelManager.Metrics.CDCReceiveTimeCounter.Add(ctx, receiveTime.Swap(0))
p.otelManager.Metrics.CDCProcessTimeCounter.Add(ctx, processTime.Swap(0))
p.otelManager.Metrics.CDCAddRecordTimeCounter.Add(ctx, addRecordTime.Swap(0))
}()
shutdown := common.Interval(ctx, time.Minute, func() {
p.otelManager.Metrics.FetchedBytesCounter.Add(ctx, fetchedBytes.Swap(0))
p.otelManager.Metrics.AllFetchedBytesCounter.Add(ctx, allFetchedBytes.Swap(0))
p.otelManager.Metrics.CDCReceiveTimeCounter.Add(ctx, receiveTime.Swap(0))
p.otelManager.Metrics.CDCProcessTimeCounter.Add(ctx, processTime.Swap(0))
p.otelManager.Metrics.CDCAddRecordTimeCounter.Add(ctx, addRecordTime.Swap(0))

if lastXLogDataServerWALEnd.Load() > 0 {
p.otelManager.Metrics.ServerWalEndLagGauge.Record(ctx,
Expand All @@ -645,9 +653,13 @@ func PullCdcRecords[Items model.Items](
return err
}
}
addStart := time.Now()
processTime.Add(int64(addStart.Sub(processStart)))
if err := records.AddRecord(ctx, rec); err != nil {
return err
}
processStart = time.Now()
addRecordTime.Add(int64(processStart.Sub(addStart)))

totalRecords++

Expand Down Expand Up @@ -765,11 +777,13 @@ func PullCdcRecords[Items model.Items](
receiveDeadline = nextRecordDeadline
}
receiveCtx, cancel := context.WithDeadline(ctx, receiveDeadline)
receiveStart := time.Now()
rawMsg, err := func() (pgproto3.BackendMessage, error) {
replLock.Lock()
defer replLock.Unlock()
return conn.ReceiveMessage(receiveCtx)
}()
receiveTime.Add(int64(time.Since(receiveStart)))
cancel()

if ctxErr := ctx.Err(); ctxErr != nil {
Expand All @@ -795,6 +809,7 @@ func PullCdcRecords[Items model.Items](
return fmt.Errorf("ReceiveMessage failed: %w", err)
}

processStart = time.Now()
switch msg := rawMsg.(type) {
case *pgproto3.ErrorResponse:
return shared.LogError(logger, exceptions.NewPostgresWalError(errors.New("received error response"), msg))
Expand Down Expand Up @@ -961,13 +976,20 @@ func PullCdcRecords[Items model.Items](
return err
}
}
} else if err := records.AddRecord(ctx, rec); err != nil {
return err
} else {
addStart := time.Now()
processTime.Add(int64(addStart.Sub(processStart)))
if err := records.AddRecord(ctx, rec); err != nil {
return err
}
processStart = time.Now()
addRecordTime.Add(int64(processStart.Sub(addStart)))
}
}
}
}
}
processTime.Add(int64(time.Since(processStart)))
}
}

Expand Down
27 changes: 27 additions & 0 deletions flow/otel_metrics/otel_manager.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,9 @@ const (
AllFetchedBytesCounterName = "all_fetched_bytes"
FetchedBytesCounterName = "fetched_bytes"
FetchedEventSizeHistogramName = "fetched_event_size"
CDCReceiveTimeCounterName = "cdc_receive_time"
CDCProcessTimeCounterName = "cdc_process_time"
CDCAddRecordTimeCounterName = "cdc_add_record_time"
SourceLagGaugeName = "source_lag"
DestinationLagGaugeName = "destination_lag"
E2ELagGaugeName = "e2e_lag"
Expand Down Expand Up @@ -123,6 +126,9 @@ type Metrics struct {
AllFetchedBytesCounter metric.Int64Counter
FetchedBytesCounter metric.Int64Counter
FetchedEventSizeHistogram metric.Int64Histogram
CDCReceiveTimeCounter metric.Int64Counter
CDCProcessTimeCounter metric.Int64Counter
CDCAddRecordTimeCounter metric.Int64Counter
SourceLagGauge metric.Int64Gauge
DestinationLagGauge metric.Int64Gauge
E2ELagGauge metric.Int64Gauge
Expand Down Expand Up @@ -481,6 +487,27 @@ func (om *OtelManager) setupMetrics(ctx context.Context) error {
return err
}

if om.Metrics.CDCReceiveTimeCounter, err = om.GetOrInitInt64Counter(BuildMetricName(CDCReceiveTimeCounterName),
metric.WithUnit("ns"),
metric.WithDescription("Time the CDC pull loop spent in the receive call waiting for the next replication message"),
); err != nil {
return err
}

if om.Metrics.CDCProcessTimeCounter, err = om.GetOrInitInt64Counter(BuildMetricName(CDCProcessTimeCounterName),
metric.WithUnit("ns"),
metric.WithDescription("Time the CDC pull loop spent handling received replication messages, excluding time in AddRecord"),
); err != nil {
return err
}

if om.Metrics.CDCAddRecordTimeCounter, err = om.GetOrInitInt64Counter(BuildMetricName(CDCAddRecordTimeCounterName),
metric.WithUnit("ns"),
metric.WithDescription("Time the CDC pull loop spent in AddRecord passing records to the record stream"),
); err != nil {
return err
}

if om.Metrics.SourceLagGauge, err = om.GetOrInitInt64Gauge(BuildMetricName(SourceLagGaugeName),
metric.WithUnit("ms"),
metric.WithDescription("Lag in milliseconds from a source event's commit timestamp to when PeerDB receives it"),
Expand Down
Loading