Skip to content

Commit 615d0b8

Browse files
serprexserprex
committed
otel spans for cdc.batch: pull/sync/normalize
1 parent e239a2e commit 615d0b8

7 files changed

Lines changed: 147 additions & 36 deletions

File tree

‎flow/activities/flowable_core.go‎

Lines changed: 96 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,9 @@ import (
1212
"github.com/jackc/pgerrcode"
1313
"github.com/jackc/pgx/v5"
1414
"go.opentelemetry.io/otel/attribute"
15+
"go.opentelemetry.io/otel/codes"
1516
"go.opentelemetry.io/otel/metric"
17+
"go.opentelemetry.io/otel/trace"
1618
"go.temporal.io/sdk/activity"
1719
"go.temporal.io/sdk/log"
1820
"golang.org/x/sync/errgroup"
@@ -136,6 +138,11 @@ func syncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDCSyncCon
136138
ctx = context.WithValue(ctx, shared.FlowNameKey, flowName)
137139
logger := internal.LoggerFromCtx(ctx)
138140

141+
ctx, batchSpan := a.OtelManager.Tracer.Start(ctx, "cdc.batch", trace.WithAttributes(
142+
attribute.String(otel_metrics.FlowNameKey, flowName),
143+
))
144+
defer batchSpan.End()
145+
139146
tblNameMapping := make(map[string]model.NameAndExclude, len(options.TableMappings))
140147
for _, v := range options.TableMappings {
141148
tblNameMapping[v.SourceTableIdentifier] = model.NewNameAndExclude(v.DestinationTableIdentifier, v.Exclude)
@@ -193,11 +200,42 @@ func syncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDCSyncCon
193200
return nil, err
194201
}
195202

203+
syncBatchID, err := func() (int64, error) {
204+
// special case pg-pg replication, where batch ID is stored on destination instead of catalog
205+
if _, isSourcePg := any(srcConn).(*connpostgres.PostgresConnector); isSourcePg {
206+
dstPgConn, dstPgClose, err := connectors.GetPostgresConnectorByName(ctx, config.Env, a.CatalogPool, config.DestinationName)
207+
if err != nil {
208+
if !errors.Is(err, errors.ErrUnsupported) {
209+
return 0, fmt.Errorf("failed to get destination connector to get last sync batch ID: %w", err)
210+
}
211+
// else fallthrough to loading from catalog
212+
} else {
213+
defer dstPgClose(ctx)
214+
return dstPgConn.GetLastSyncBatchID(ctx, flowName)
215+
}
216+
}
217+
pgMetadata := connmetadata.NewPostgresMetadataFromCatalog(logger, a.CatalogPool)
218+
return pgMetadata.GetLastSyncBatchID(ctx, flowName)
219+
}()
220+
if err != nil {
221+
batchSpan.RecordError(err)
222+
batchSpan.SetStatus(codes.Error, err.Error())
223+
return nil, a.Alerter.LogFlowError(ctx, flowName, err)
224+
}
225+
syncBatchID += 1
226+
batchSpan.SetAttributes(attribute.Int64(otel_metrics.BatchIdKey, syncBatchID))
227+
196228
startTime := time.Now()
197229
syncState.Store(new("syncing"))
198230
errGroup, errCtx := errgroup.WithContext(ctx)
199231
errGroup.Go(func() error {
200-
return pull(srcConn, errCtx, a.CatalogPool, a.OtelManager, &model.PullRecordsRequest[Items]{
232+
pullCtx, pullSpan := a.OtelManager.Tracer.Start(errCtx, "cdc.pull", trace.WithAttributes(
233+
attribute.String(otel_metrics.FlowNameKey, flowName),
234+
attribute.Int(otel_metrics.TableCountKey, len(options.TableMappings)),
235+
attribute.Int64(otel_metrics.BatchIdKey, syncBatchID),
236+
))
237+
defer pullSpan.End()
238+
err := pull(srcConn, pullCtx, a.CatalogPool, a.OtelManager, &model.PullRecordsRequest[Items]{
201239
FlowJobName: flowName,
202240
SrcTableIDNameMapping: options.SrcTableIdNameMapping,
203241
TableNameMapping: tblNameMapping,
@@ -212,6 +250,11 @@ func syncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDCSyncCon
212250
Env: config.Env,
213251
InternalVersion: config.Version,
214252
})
253+
if err != nil {
254+
pullSpan.RecordError(err)
255+
pullSpan.SetStatus(codes.Error, err.Error())
256+
}
257+
return err
215258
})
216259

217260
hasRecords := !recordBatchSync.WaitAndCheckEmpty()
@@ -247,46 +290,35 @@ func syncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDCSyncCon
247290

248291
var res *model.SyncResponse
249292
errGroup.Go(func() error {
250-
syncBatchID, err := func() (int64, error) {
251-
// special case pg-pg replication, where batch ID is stored on destination instead of catalog
252-
if _, isSourcePg := any(srcConn).(*connpostgres.PostgresConnector); isSourcePg {
253-
dstPgConn, dstPgClose, err := connectors.GetPostgresConnectorByName(ctx, config.Env, a.CatalogPool, config.DestinationName)
254-
if err != nil {
255-
if !errors.Is(err, errors.ErrUnsupported) {
256-
return 0, fmt.Errorf("failed to get destination connector to get last sync batch ID: %w", err)
257-
}
258-
// else fallthrough to loading from catalog
259-
} else {
260-
defer dstPgClose(ctx)
261-
return dstPgConn.GetLastSyncBatchID(errCtx, flowName)
262-
}
263-
}
264-
pgMetadata := connmetadata.NewPostgresMetadataFromCatalog(logger, a.CatalogPool)
265-
return pgMetadata.GetLastSyncBatchID(errCtx, flowName)
266-
}()
267-
if err != nil {
268-
return err
269-
}
270-
syncBatchID += 1
293+
syncCtx, syncSpan := a.OtelManager.Tracer.Start(errCtx, "cdc.sync", trace.WithAttributes(
294+
attribute.String(otel_metrics.FlowNameKey, flowName),
295+
attribute.Int(otel_metrics.TableCountKey, len(options.TableMappings)),
296+
attribute.Int64(otel_metrics.BatchIdKey, syncBatchID),
297+
))
298+
defer syncSpan.End()
271299
syncingBatchID.Store(syncBatchID)
272300
logger.Info("begin pulling records for batch", slog.Int64("syncBatchID", syncBatchID))
273301

274-
if err := monitoring.AddCDCBatchForFlow(errCtx, a.CatalogPool, flowName, monitoring.CDCBatchInfo{
302+
if err := monitoring.AddCDCBatchForFlow(syncCtx, a.CatalogPool, flowName, monitoring.CDCBatchInfo{
275303
BatchID: syncBatchID,
276304
RowsInBatch: 0,
277305
BatchEndlSN: 0,
278306
StartTime: startTime,
279307
}); err != nil {
308+
syncSpan.RecordError(err)
309+
syncSpan.SetStatus(codes.Error, err.Error())
280310
return a.Alerter.LogFlowError(ctx, flowName, err)
281311
}
282312

283-
dstConn, dstClose, err := connectors.GetByNameAs[TSync](ctx, config.Env, a.CatalogPool, config.DestinationName)
313+
dstConn, dstClose, err := connectors.GetByNameAs[TSync](syncCtx, config.Env, a.CatalogPool, config.DestinationName)
284314
if err != nil {
315+
syncSpan.RecordError(err)
316+
syncSpan.SetStatus(codes.Error, err.Error())
285317
return fmt.Errorf("failed to get destination connector: %w", err)
286318
}
287319
defer dstClose(ctx)
288320

289-
res, err = sync(dstConn, errCtx, &model.SyncRecordsRequest[Items]{
321+
res, err = sync(dstConn, syncCtx, &model.SyncRecordsRequest[Items]{
290322
SyncBatchID: syncBatchID,
291323
Records: recordBatchSync,
292324
ConsumedOffset: &consumedOffset,
@@ -300,8 +332,11 @@ func syncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDCSyncCon
300332
Flags: config.Flags,
301333
})
302334
if err != nil {
335+
syncSpan.RecordError(err)
336+
syncSpan.SetStatus(codes.Error, err.Error())
303337
return a.Alerter.LogFlowError(ctx, flowName, fmt.Errorf("failed to push records: %w", err))
304338
}
339+
syncSpan.SetAttributes(attribute.Int64(otel_metrics.RowsInBatchKey, res.NumRecordsSynced))
305340
for _, warning := range res.Warnings {
306341
a.Alerter.LogFlowWarning(ctx, flowName, warning)
307342
}
@@ -317,13 +352,21 @@ func syncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDCSyncCon
317352
if !(isDesync || shared.IsSQLStateError(err, pgerrcode.ObjectInUse)) {
318353
_ = a.Alerter.LogFlowError(ctx, flowName, err)
319354
}
355+
batchSpan.RecordError(err)
356+
batchSpan.SetStatus(codes.Error, err.Error())
320357
return nil, fmt.Errorf("[cdc] failed to pull records: %w", err)
321358
}
322359
syncState.Store(new("bookkeeping"))
323360

324361
syncDuration := time.Since(syncStartTime)
325362
lastCheckpoint := recordBatchSync.GetLastCheckpoint()
326363
logger.Info("batch synced", slog.Any("checkpoint", lastCheckpoint))
364+
batchSpan.SetAttributes(
365+
attribute.Int64(otel_metrics.LastCheckpointIDKey, lastCheckpoint.ID),
366+
attribute.String(otel_metrics.LastCheckpointTextKey, lastCheckpoint.Text),
367+
attribute.Int64(otel_metrics.RowsInBatchKey, res.NumRecordsSynced),
368+
)
369+
327370
if err := srcConn.UpdateReplStateLastOffset(ctx, lastCheckpoint); err != nil {
328371
return nil, a.Alerter.LogFlowError(ctx, flowName, err)
329372
}
@@ -684,17 +727,34 @@ func (a *FlowableActivity) startNormalize(
684727

685728
for {
686729
logger.Info("normalizing batches", slog.Int64("syncBatchID", batchID))
687-
res, err := dstConn.NormalizeRecords(ctx, &model.NormalizeRecordsRequest{
688-
FlowJobName: config.FlowJobName,
689-
Env: config.Env,
690-
TableNameSchemaMapping: tableNameSchemaMapping,
691-
TableMappings: config.TableMappings,
692-
SoftDeleteColName: config.SoftDeleteColName,
693-
SyncedAtColName: config.SyncedAtColName,
694-
SyncBatchID: batchID,
695-
Version: config.Version,
696-
Flags: config.Flags,
697-
})
730+
res, err := func() (model.NormalizeResponse, error) {
731+
normCtx, normSpan := a.OtelManager.Tracer.Start(ctx, "cdc.normalize", trace.WithAttributes(
732+
attribute.String(otel_metrics.FlowNameKey, config.FlowJobName),
733+
attribute.Int64(otel_metrics.BatchIdKey, batchID),
734+
))
735+
defer normSpan.End()
736+
res, err := dstConn.NormalizeRecords(normCtx, &model.NormalizeRecordsRequest{
737+
FlowJobName: config.FlowJobName,
738+
Env: config.Env,
739+
TableNameSchemaMapping: tableNameSchemaMapping,
740+
TableMappings: config.TableMappings,
741+
SoftDeleteColName: config.SoftDeleteColName,
742+
SyncedAtColName: config.SyncedAtColName,
743+
SyncBatchID: batchID,
744+
Version: config.Version,
745+
Flags: config.Flags,
746+
})
747+
if err != nil {
748+
normSpan.RecordError(err)
749+
normSpan.SetStatus(codes.Error, err.Error())
750+
return res, err
751+
}
752+
normSpan.SetAttributes(
753+
attribute.Int64(otel_metrics.StartBatchIDKey, res.StartBatchID),
754+
attribute.Int64(otel_metrics.EndBatchIDKey, res.EndBatchID),
755+
)
756+
return res, nil
757+
}()
698758
if err != nil {
699759
return a.Alerter.LogFlowError(ctx, config.FlowJobName,
700760
exceptions.NewNormalizationError(fmt.Errorf("failed to normalize records: %w", err)))

‎flow/connectors/mongo/cdc.go‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,8 @@ import (
1313
"go.mongodb.org/mongo-driver/v2/bson"
1414
"go.mongodb.org/mongo-driver/v2/mongo"
1515
"go.mongodb.org/mongo-driver/v2/mongo/options"
16+
"go.opentelemetry.io/otel/attribute"
17+
"go.opentelemetry.io/otel/trace"
1618

1719
"github.com/PeerDB-io/peerdb/flow/generated/protos"
1820
"github.com/PeerDB-io/peerdb/flow/model"
@@ -215,6 +217,18 @@ func (c *MongoConnector) PullRecords(
215217
if recordCount == 0 {
216218
req.RecordStream.SignalAsEmpty()
217219
}
220+
span := trace.SpanFromContext(ctx)
221+
span.SetAttributes(
222+
attribute.Int64(otel_metrics.RowsInBatchKey, int64(recordCount)),
223+
attribute.Int64(otel_metrics.BytesPulledKey, cumulativeBytesProcessed.Load()),
224+
)
225+
if rt := changeStream.ResumeToken(); rt != nil {
226+
rtStr := base64.StdEncoding.EncodeToString(rt)
227+
if len(rtStr) > 64 {
228+
rtStr = rtStr[:64]
229+
}
230+
span.SetAttributes(attribute.String(otel_metrics.ResumeTokenKey, rtStr))
231+
}
218232
c.logger.Info("[mongo] PullRecords finished streaming",
219233
slog.Uint64("records", uint64(recordCount)),
220234
slog.Int64("bytes", cumulativeBytesProcessed.Load()),

‎flow/connectors/mysql/cdc.go‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,8 @@ import (
1717
"github.com/pingcap/tidb/pkg/parser"
1818
"github.com/pingcap/tidb/pkg/parser/ast"
1919
_ "github.com/pingcap/tidb/pkg/types/parser_driver"
20+
"go.opentelemetry.io/otel/attribute"
21+
"go.opentelemetry.io/otel/trace"
2022
"google.golang.org/protobuf/proto"
2123

2224
"github.com/PeerDB-io/peerdb/flow/connectors/utils"
@@ -409,6 +411,14 @@ func (c *MySqlConnector) PullRecords(
409411
if recordCount == 0 {
410412
req.RecordStream.SignalAsEmpty()
411413
}
414+
span := trace.SpanFromContext(ctx)
415+
span.SetAttributes(
416+
attribute.Int64(otel_metrics.RowsInBatchKey, int64(recordCount)),
417+
attribute.Int64(otel_metrics.BytesPulledKey, totalFetchedBytes.Load()),
418+
)
419+
if updatedOffset != "" {
420+
span.SetAttributes(attribute.String(otel_metrics.GtidKey, updatedOffset))
421+
}
412422
c.logger.Info("[mysql] PullRecords finished streaming",
413423
slog.Uint64("records", uint64(recordCount)),
414424
slog.Int64("bytes", totalFetchedBytes.Load()),

‎flow/connectors/postgres/cdc.go‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@ import (
2222
"github.com/pgvector/pgvector-go"
2323
"go.opentelemetry.io/otel/attribute"
2424
"go.opentelemetry.io/otel/metric"
25+
"go.opentelemetry.io/otel/trace"
2526
"go.temporal.io/sdk/log"
2627

2728
connmetadata "github.com/PeerDB-io/peerdb/flow/connectors/external_metadata"
@@ -524,6 +525,11 @@ func PullCdcRecords[Items model.Items](
524525
if totalRecords == 0 {
525526
records.SignalAsEmpty()
526527
}
528+
trace.SpanFromContext(ctx).SetAttributes(
529+
attribute.Int64(otel_metrics.RowsInBatchKey, totalRecords),
530+
attribute.Int64(otel_metrics.BytesPulledKey, totalFetchedBytes.Load()),
531+
attribute.Int64(otel_metrics.LastCheckpointIDKey, int64(clientXLogPos)),
532+
)
527533
logger.Info("[finished] PullRecords",
528534
slog.Int64("records", totalRecords),
529535
slog.Int64("bytes", totalFetchedBytes.Load()),

‎flow/otel_metrics/attributes.go‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,15 @@ const (
3636
BackendStateKey = "backendState"
3737
WalStatusKey = "walStatus"
3838
PendingRestartKey = "pendingRestart"
39+
RowsInBatchKey = "rowsInBatch"
40+
BytesPulledKey = "bytesPulled"
41+
TableCountKey = "tableCount"
42+
LastCheckpointIDKey = "lastCheckpoint.ID"
43+
LastCheckpointTextKey = "lastCheckpoint.Text"
44+
StartBatchIDKey = "startBatchID"
45+
EndBatchIDKey = "endBatchID"
46+
ResumeTokenKey = "resumeToken"
47+
GtidKey = "gtid"
3948
)
4049

4150
const (

‎flow/otel_metrics/otel_manager.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@ import (
1616
sdkmetric "go.opentelemetry.io/otel/sdk/metric"
1717
"go.opentelemetry.io/otel/sdk/metric/metricdata"
1818
"go.opentelemetry.io/otel/sdk/resource"
19+
"go.opentelemetry.io/otel/trace"
1920

2021
"github.com/PeerDB-io/peerdb/flow/internal"
2122
)
@@ -168,6 +169,7 @@ type OtelManager struct {
168169
Metrics Metrics
169170
MetricsProvider metric.MeterProvider
170171
Meter metric.Meter
172+
Tracer trace.Tracer
171173
Float64GaugesCache map[string]metric.Float64Gauge
172174
Int64GaugesCache map[string]metric.Int64Gauge
173175
Int64CountersCache map[string]metric.Int64Counter
@@ -184,6 +186,7 @@ func NewOtelManager(ctx context.Context, serviceName string, enabled bool) (*Ote
184186
Enabled: enabled,
185187
MetricsProvider: metricsProvider,
186188
Meter: metricsProvider.Meter("io.peerdb." + serviceName),
189+
Tracer: Tracer(),
187190
Float64GaugesCache: make(map[string]metric.Float64Gauge),
188191
Int64GaugesCache: make(map[string]metric.Int64Gauge),
189192
Int64CountersCache: make(map[string]metric.Int64Counter),

‎flow/otel_metrics/trace_provider.go‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,8 +9,11 @@ import (
99
"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
1010
"go.opentelemetry.io/otel/propagation"
1111
sdktrace "go.opentelemetry.io/otel/sdk/trace"
12+
"go.opentelemetry.io/otel/trace"
1213
)
1314

15+
const TracerName = "io.peerdb.flow"
16+
1417
func SetupTracerProvider(ctx context.Context, serviceName string, enabled bool) (*sdktrace.TracerProvider, error) {
1518
if !enabled {
1619
return nil, nil
@@ -38,3 +41,9 @@ func SetupTracerProvider(ctx context.Context, serviceName string, enabled bool)
3841
slog.InfoContext(ctx, "Tracer provider initialized", slog.String("service", serviceName))
3942
return tp, nil
4043
}
44+
45+
// Tracer returns the shared tracer. Safe whether or not SetupTracerProvider ran:
46+
// global default is a noop tracer
47+
func Tracer() trace.Tracer {
48+
return otel.Tracer(TracerName)
49+
}

0 commit comments

Comments
 (0)