diff --git a/flow/connectors/mysql/cdc.go b/flow/connectors/mysql/cdc.go index a993a7cc9..f0ccc3911 100644 --- a/flow/connectors/mysql/cdc.go +++ b/flow/connectors/mysql/cdc.go @@ -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 { @@ -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()), @@ -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))) 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) @@ -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 { @@ -1100,6 +1114,7 @@ func (c *MySqlConnector) PullRecords( } lastEventAt = time.Now() + processStart = lastEventAt allFetchedBytes.Add(int64(len(event.RawData))) @@ -1117,6 +1132,7 @@ func (c *MySqlConnector) PullRecords( return err } } + processTime.Add(int64(time.Since(processStart))) } return nil } diff --git a/flow/connectors/postgres/cdc.go b/flow/connectors/postgres/cdc.go index 75d474b00..3dba2b966 100644 --- a/flow/connectors/postgres/cdc.go +++ b/flow/connectors/postgres/cdc.go @@ -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 @@ -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, @@ -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++ @@ -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 { @@ -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)) @@ -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))) } } diff --git a/flow/otel_metrics/otel_manager.go b/flow/otel_metrics/otel_manager.go index 679c150a6..ee6678c2a 100644 --- a/flow/otel_metrics/otel_manager.go +++ b/flow/otel_metrics/otel_manager.go @@ -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" @@ -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 @@ -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"),