diff --git a/downstreamadapter/sink/blackhole/sink.go b/downstreamadapter/sink/blackhole/sink.go index 6124219ad4..6934b3b73d 100644 --- a/downstreamadapter/sink/blackhole/sink.go +++ b/downstreamadapter/sink/blackhole/sink.go @@ -19,7 +19,7 @@ import ( "github.com/pingcap/log" "github.com/pingcap/ticdc/pkg/common" commonEvent "github.com/pingcap/ticdc/pkg/common/event" - "github.com/pingcap/ticdc/pkg/metrics" + "github.com/pingcap/ticdc/pkg/statistics" "github.com/pingcap/ticdc/utils/chann" "go.uber.org/zap" ) @@ -28,13 +28,13 @@ import ( // Including DDL and DML. type Sink struct { eventCh *chann.UnlimitedChannel[*commonEvent.DMLEvent, any] - statistics *metrics.Statistics + statistics *statistics.Statistics } func New(changefeedID common.ChangeFeedID, keyspaceID uint32) (*Sink, error) { return &Sink{ eventCh: chann.NewUnlimitedChannelDefault[*commonEvent.DMLEvent](), - statistics: metrics.NewStatistics(changefeedID, keyspaceID, "sink"), + statistics: statistics.New(changefeedID, keyspaceID), }, nil } @@ -54,6 +54,7 @@ func (s *Sink) AddDMLEvent(event *commonEvent.DMLEvent) { // ref: https://github.com/pingcap/ticdc/blob/da834db76e0662ff15ef12645d1f37bfa6506d83/tests/integration_tests/lossy_ddl/run.sh#L23 // Use zap.Stringer to call String() method which applies log redaction log.Debug("BlackHoleSink: WriteEvents", zap.Stringer("dml", event)) + s.statistics.TrackDMLEvent(event) s.eventCh.Push(event) } @@ -104,12 +105,7 @@ func (s *Sink) Run(ctx context.Context) error { log.Info("blackhole sink event channel closed") return nil } - err := s.statistics.RecordBatchExecution(func() (int, int64, error) { - return int(event.Len()), event.GetSize(), nil - }) - if err != nil { - return err - } + s.statistics.RecordDMLResult(int(event.Len()), nil) event.PostFlush() } } diff --git a/downstreamadapter/sink/cloudstorage/dml_writers.go b/downstreamadapter/sink/cloudstorage/dml_writers.go index 6f4d8e9eb2..d4acb7da48 100644 --- a/downstreamadapter/sink/cloudstorage/dml_writers.go +++ b/downstreamadapter/sink/cloudstorage/dml_writers.go @@ -23,8 +23,8 @@ import ( "github.com/pingcap/ticdc/pkg/cloudstorage" commonType "github.com/pingcap/ticdc/pkg/common" commonEvent "github.com/pingcap/ticdc/pkg/common/event" - "github.com/pingcap/ticdc/pkg/metrics" "github.com/pingcap/ticdc/pkg/sink/codec/common" + "github.com/pingcap/ticdc/pkg/statistics" "github.com/pingcap/ticdc/utils/chann" "github.com/pingcap/tidb/pkg/objstore/storeapi" "go.uber.org/atomic" @@ -34,7 +34,7 @@ import ( // dmlWriters coordinates encoding and output shard writers. type dmlWriters struct { changefeedID commonType.ChangeFeedID - statistics *metrics.Statistics + statistics *statistics.Statistics // msgCh is the only unbounded queue in the storage sink pipeline. // External callers push tasks into it, addTasks consumes it, and @@ -55,7 +55,7 @@ func newDMLWriters( config *cloudstorage.Config, encoderConfig *common.Config, extension string, - statistics *metrics.Statistics, + statistics *statistics.Statistics, columnSelector *columnselector.ColumnSelectors, ) (*dmlWriters, error) { messageCh := chann.NewUnlimitedChannelDefault[*task]() diff --git a/downstreamadapter/sink/cloudstorage/sink.go b/downstreamadapter/sink/cloudstorage/sink.go index 297c5f1e06..499438bf67 100644 --- a/downstreamadapter/sink/cloudstorage/sink.go +++ b/downstreamadapter/sink/cloudstorage/sink.go @@ -29,6 +29,7 @@ import ( "github.com/pingcap/ticdc/pkg/config" "github.com/pingcap/ticdc/pkg/errors" "github.com/pingcap/ticdc/pkg/metrics" + "github.com/pingcap/ticdc/pkg/statistics" "github.com/pingcap/ticdc/pkg/util" "github.com/pingcap/tidb/pkg/meta/model" "github.com/pingcap/tidb/pkg/objstore/storeapi" @@ -65,7 +66,7 @@ type sink struct { lastSendCheckpointTsTime time.Time cron *cron.Cron - statistics *metrics.Statistics + statistics *statistics.Statistics isNormal *atomic.Bool cleanupJobs []func() /* only for test */ @@ -135,7 +136,7 @@ func New( if err != nil { return nil, err } - statistics := metrics.NewStatistics(changefeedID, keyspaceID, "cloudstorage") + statistics := statistics.New(changefeedID, keyspaceID) defer func() { if err != nil { statistics.Close() @@ -205,6 +206,7 @@ func (s *sink) AddDMLEvent(event *commonEvent.DMLEvent) { zap.String("dispatcher", event.GetDispatcherID().String())) return } + s.statistics.TrackDMLEvent(event) s.dmlWriters.addDMLEvent(event) } diff --git a/downstreamadapter/sink/cloudstorage/writer.go b/downstreamadapter/sink/cloudstorage/writer.go index e31a1387c6..55e7b96066 100644 --- a/downstreamadapter/sink/cloudstorage/writer.go +++ b/downstreamadapter/sink/cloudstorage/writer.go @@ -25,7 +25,7 @@ import ( "github.com/pingcap/ticdc/pkg/cloudstorage" "github.com/pingcap/ticdc/pkg/common" "github.com/pingcap/ticdc/pkg/errors" - pmetrics "github.com/pingcap/ticdc/pkg/metrics" + "github.com/pingcap/ticdc/pkg/statistics" "github.com/pingcap/tidb/pkg/objstore/storeapi" "github.com/prometheus/client_golang/prometheus" "go.uber.org/zap" @@ -45,7 +45,7 @@ type writer struct { // the channel does not need to be closed explicitly. flushCh chan flushTask - statistics *pmetrics.Statistics + statistics *statistics.Statistics filePathGenerator *cloudstorage.FilePathGenerator metricFlushBytes prometheus.Observer @@ -74,7 +74,7 @@ func newWriter( storage storeapi.Storage, config *cloudstorage.Config, extension string, - statistics *pmetrics.Statistics, + statistics *statistics.Statistics, spoolBuffer *spool.Spool, ) *writer { var ( @@ -218,46 +218,13 @@ func (d *writer) discardEntries(entries []*spool.Entry) { } } -func (d *writer) writeDataFile(ctx context.Context, dataFilePath, indexFilePath string, payload *payload) error { - keyspace := d.changeFeedID.Keyspace() - changefeed := d.changeFeedID.Name() - start := time.Now() - - err := d.statistics.RecordBatchExecution(func() (int, int64, error) { - if d.config.FlushConcurrency <= 1 { - err := d.storage.WriteFile(ctx, dataFilePath, payload.data) - if err != nil { - return 0, 0, err - } - return payload.rowsCount, payload.nBytes, nil - } - - writer, err := d.storage.Create(ctx, dataFilePath, &storeapi.WriterOption{ - Concurrency: d.config.FlushConcurrency, - }) - if err != nil { - return 0, 0, err - } - - _, err = writer.Write(ctx, payload.data) - if err != nil { - closeErr := writer.Close(ctx) - if closeErr != nil { - log.Warn("failed to close writer after write failure", - zap.String("keyspace", keyspace), zap.String("changefeed", changefeed), - zap.String("path", dataFilePath), zap.Error(closeErr)) - } - return 0, 0, err - } +func (d *writer) writeDataFile(ctx context.Context, dataFilePath, indexFilePath string, payload *payload) (err error) { + defer func() { + d.statistics.RecordDMLResult(payload.rowsCount, err) + }() - if err = writer.Close(ctx); err != nil { - log.Error("failed to close concurrency writer", - zap.String("keyspace", keyspace), zap.String("changefeed", changefeed), - zap.String("path", dataFilePath), zap.Error(err)) - return 0, 0, err - } - return payload.rowsCount, payload.nBytes, nil - }) + start := time.Now() + err = d.writeData(ctx, dataFilePath, payload.data) if err != nil { return err } @@ -265,8 +232,8 @@ func (d *writer) writeDataFile(ctx context.Context, dataFilePath, indexFilePath err = d.storage.WriteFile(ctx, indexFilePath, []byte(path.Base(dataFilePath)+"\n")) if err != nil { log.Error("failed to write index file to external storage", - zap.String("keyspace", keyspace), - zap.String("changefeed", changefeed), + zap.String("keyspace", d.changeFeedID.Keyspace()), + zap.String("changefeed", d.changeFeedID.Name()), zap.String("path", indexFilePath), zap.Int("shardID", d.shardID), zap.Error(err)) @@ -289,6 +256,40 @@ func (d *writer) writeDataFile(ctx context.Context, dataFilePath, indexFilePath return nil } +func (d *writer) writeData(ctx context.Context, dataFilePath string, data []byte) error { + if d.config.FlushConcurrency <= 1 { + return d.storage.WriteFile(ctx, dataFilePath, data) + } + + writer, err := d.storage.Create(ctx, dataFilePath, &storeapi.WriterOption{ + Concurrency: d.config.FlushConcurrency, + }) + if err != nil { + return err + } + + _, err = writer.Write(ctx, data) + if err != nil { + closeErr := writer.Close(ctx) + if closeErr != nil { + log.Warn("failed to close writer after write failure", + zap.String("keyspace", d.changeFeedID.Keyspace()), + zap.String("changefeed", d.changeFeedID.Name()), + zap.String("path", dataFilePath), zap.Error(closeErr)) + } + return err + } + + if err = writer.Close(ctx); err != nil { + log.Error("failed to close concurrency writer", + zap.String("keyspace", d.changeFeedID.Keyspace()), + zap.String("changefeed", d.changeFeedID.Name()), + zap.String("path", dataFilePath), zap.Error(err)) + return err + } + return nil +} + func (d *writer) enqueueTask(ctx context.Context, t *task) error { return d.bufferManager.enqueueTask(ctx, t) } diff --git a/downstreamadapter/sink/cloudstorage/writer_test.go b/downstreamadapter/sink/cloudstorage/writer_test.go index 7aaf02dd82..4747ae8084 100644 --- a/downstreamadapter/sink/cloudstorage/writer_test.go +++ b/downstreamadapter/sink/cloudstorage/writer_test.go @@ -31,9 +31,9 @@ import ( commonType "github.com/pingcap/ticdc/pkg/common" commonEvent "github.com/pingcap/ticdc/pkg/common/event" "github.com/pingcap/ticdc/pkg/config" - "github.com/pingcap/ticdc/pkg/metrics" "github.com/pingcap/ticdc/pkg/pdutil" "github.com/pingcap/ticdc/pkg/sink/codec/common" + "github.com/pingcap/ticdc/pkg/statistics" "github.com/pingcap/ticdc/pkg/util" "github.com/pingcap/tidb/pkg/meta/model" "github.com/pingcap/tidb/pkg/objstore/objectio" @@ -59,7 +59,7 @@ func testWriter(ctx context.Context, t *testing.T, dir string) *writer { require.NoError(t, err) changefeedID := commonType.NewChangefeedID4Test("test", t.Name()) - statistics := metrics.NewStatistics(changefeedID, commonType.DefaultKeyspaceID, t.Name()) + statistics := statistics.New(changefeedID, commonType.DefaultKeyspaceID) spoolBuffer := newTestSpool(t, changefeedID, cfg) d := newWriter(1, changefeedID, storage, cfg, ".json", statistics, spoolBuffer) @@ -479,7 +479,7 @@ func TestWriterStoresPendingMessagesInSpoolBeforeFlush(t *testing.T) { cfg.FlushInterval = time.Hour changefeedID := commonType.NewChangefeedID4Test("test", "spool-pending") - statistics := metrics.NewStatistics(changefeedID, commonType.DefaultKeyspaceID, t.Name()) + statistics := statistics.New(changefeedID, commonType.DefaultKeyspaceID) setPDClockForTest(t, pdutil.NewClock4Test()) spoolBuffer := newTestSpool(t, changefeedID, cfg) @@ -648,7 +648,7 @@ func TestWriterIndexWriteError(t *testing.T) { cfg.FlushInterval = time.Hour changefeedID := commonType.NewChangefeedID4Test("test", "writer-error-metric") - statistics := metrics.NewStatistics(changefeedID, commonType.DefaultKeyspaceID, t.Name()) + statistics := statistics.New(changefeedID, commonType.DefaultKeyspaceID) setPDClockForTest(t, pdutil.NewClock4Test()) spoolBuffer := newTestSpool(t, changefeedID, cfg) d := newWriter(1, changefeedID, storage, cfg, ".json", statistics, spoolBuffer) @@ -714,7 +714,7 @@ func TestWriterDataFileCloseError(t *testing.T) { cfg.FlushInterval = time.Hour changefeedID := commonType.NewChangefeedID4Test("test", "writer-close-error") - statistics := metrics.NewStatistics(changefeedID, commonType.DefaultKeyspaceID, t.Name()) + statistics := statistics.New(changefeedID, commonType.DefaultKeyspaceID) setPDClockForTest(t, pdutil.NewClock4Test()) spoolBuffer := newTestSpool(t, changefeedID, cfg) d := newWriter(1, changefeedID, storage, cfg, ".json", statistics, spoolBuffer) diff --git a/downstreamadapter/sink/kafka/helper.go b/downstreamadapter/sink/kafka/helper.go index 37327548d4..7cf1c34f3e 100644 --- a/downstreamadapter/sink/kafka/helper.go +++ b/downstreamadapter/sink/kafka/helper.go @@ -27,6 +27,7 @@ import ( codecCommon "github.com/pingcap/ticdc/pkg/sink/codec/common" "github.com/pingcap/ticdc/pkg/sink/kafka" "github.com/pingcap/ticdc/pkg/sink/kafka/claimcheck" + "github.com/pingcap/ticdc/pkg/statistics" "github.com/pingcap/tidb/br/pkg/utils" ) @@ -58,6 +59,7 @@ func newKafkaSinkComponent( changefeedID common.ChangeFeedID, sinkURI *url.URL, sinkConfig *config.SinkConfig, + stat *statistics.Statistics, ) (components, config.Protocol, error) { var ( comp components @@ -85,7 +87,7 @@ func newKafkaSinkComponent( } options.Topic = topic - comp.factory, err = kafka.NewSaramaFactory(ctx, options, changefeedID) + comp.factory, err = kafka.NewSaramaFactory(ctx, options, changefeedID, stat) if err != nil { return comp, protocol, err } diff --git a/downstreamadapter/sink/kafka/sink.go b/downstreamadapter/sink/kafka/sink.go index 2133e1c6ba..912f30fedb 100644 --- a/downstreamadapter/sink/kafka/sink.go +++ b/downstreamadapter/sink/kafka/sink.go @@ -31,6 +31,7 @@ import ( codecCommon "github.com/pingcap/ticdc/pkg/sink/codec/common" "github.com/pingcap/ticdc/pkg/sink/kafka" "github.com/pingcap/ticdc/pkg/sink/kafka/claimcheck" + "github.com/pingcap/ticdc/pkg/statistics" "github.com/pingcap/ticdc/pkg/util" "github.com/pingcap/ticdc/utils/chann" "go.uber.org/atomic" @@ -51,7 +52,7 @@ type sink struct { metricsCollector kafka.MetricsCollector comp components - statistics *metrics.Statistics + statistics *statistics.Statistics protocol config.Protocol partitionRule helper.DDLDispatchRule @@ -111,7 +112,7 @@ func Verify(ctx context.Context, changefeedID common.ChangeFeedID, uri *url.URL, return err } - factory, err := kafka.NewSaramaFactory(ctx, options, changefeedID) + factory, err := kafka.NewSaramaFactory(ctx, options, changefeedID, nil) if err != nil { return err } @@ -156,11 +157,13 @@ func Verify(ctx context.Context, changefeedID common.ChangeFeedID, uri *url.URL, func New( ctx context.Context, changefeedID common.ChangeFeedID, sinkURI *url.URL, sinkConfig *config.SinkConfig, keyspaceID uint32, ) (*sink, error) { - comp, protocol, err := newKafkaSinkComponent(ctx, changefeedID, sinkURI, sinkConfig) + stat := statistics.New(changefeedID, keyspaceID) + comp, protocol, err := newKafkaSinkComponent(ctx, changefeedID, sinkURI, sinkConfig, stat) if err != nil { + stat.Close() return nil, err } - return newWithComponents(ctx, changefeedID, keyspaceID, protocol, comp) + return newWithComponents(ctx, changefeedID, keyspaceID, protocol, comp, stat) } func newWithComponents( @@ -169,8 +172,8 @@ func newWithComponents( keyspaceID uint32, protocol config.Protocol, comp components, + stat *statistics.Statistics, ) (*sink, error) { - statistics := metrics.NewStatistics(changefeedID, keyspaceID, "sink") var ( err error asyncProducer kafka.AsyncProducer @@ -187,7 +190,7 @@ func newWithComponents( asyncProducer.Close() } comp.close() - statistics.Close() + stat.Close() }() asyncProducer, err = comp.factory.AsyncProducer(ctx) @@ -208,7 +211,7 @@ func newWithComponents( partitionRule: helper.GetDDLDispatchRule(protocol), protocol: protocol, comp: comp, - statistics: statistics, + statistics: stat, checkpointChan: make(chan uint64, 16), eventChan: chann.NewUnlimitedChannelDefault[*commonEvent.DMLEvent](), @@ -244,6 +247,7 @@ func (s *sink) IsNormal() bool { } func (s *sink) AddDMLEvent(event *commonEvent.DMLEvent) { + s.statistics.TrackDMLEvent(event) s.eventChan.Push(event) } @@ -429,17 +433,12 @@ func (s *sink) sendMessages(ctx context.Context) error { } for _, message := range future.Messages { start := time.Now() - if err = s.statistics.RecordBatchExecution(func() (int, int64, error) { - message.SetPartitionKey(future.Key.PartitionKey) - if err = s.dmlProducer.AsyncSend( - ctx, - future.Key.Topic, - future.Key.Partition, - message); err != nil { - return 0, 0, err - } - return message.GetRowsCount(), int64(message.Length()), nil - }); err != nil { + message.SetPartitionKey(future.Key.PartitionKey) + if err = s.dmlProducer.AsyncSend( + ctx, + future.Key.Topic, + future.Key.Partition, + message); err != nil { return err } metricSendMessageDuration.Observe(time.Since(start).Seconds()) diff --git a/downstreamadapter/sink/kafka/sink_test.go b/downstreamadapter/sink/kafka/sink_test.go index 8b205c1649..eb27f84435 100644 --- a/downstreamadapter/sink/kafka/sink_test.go +++ b/downstreamadapter/sink/kafka/sink_test.go @@ -35,6 +35,7 @@ import ( "github.com/pingcap/ticdc/pkg/sink/codec" codecCommon "github.com/pingcap/ticdc/pkg/sink/codec/common" "github.com/pingcap/ticdc/pkg/sink/kafka" + "github.com/pingcap/ticdc/pkg/statistics" "github.com/stretchr/testify/require" "go.uber.org/atomic" ) @@ -228,7 +229,8 @@ func newKafkaSinkForTestWithProducers(ctx context.Context, } }() - s, err := newWithComponents(ctx, changefeedID, common.DefaultKeyspaceID, protocol, comp) + s, err := newWithComponents(ctx, changefeedID, common.DefaultKeyspaceID, protocol, comp, + statistics.New(changefeedID, common.DefaultKeyspaceID)) if err != nil { return nil, err } @@ -265,7 +267,7 @@ func TestKafkaSinkBasicFunctionality(t *testing.T) { job := helper.DDL2Job(createTableSQL) require.NotNil(t, job) - var count atomic.Int64 + var count, dmlFlushCount atomic.Int64 ddlEvent := &commonEvent.DDLEvent{ Query: job.Query, SchemaName: job.SchemaName, @@ -302,7 +304,10 @@ func TestKafkaSinkBasicFunctionality(t *testing.T) { "insert into t values (1, 'test')", "insert into t values (2, 'test2');") dmlEvent.PostTxnFlushed = []func(){ - func() { count.Add(1) }, + func() { + count.Add(1) + dmlFlushCount.Add(1) + }, } dmlEvent.CommitTs = 2 @@ -310,6 +315,7 @@ func TestKafkaSinkBasicFunctionality(t *testing.T) { ctrl := gomock.NewController(t) asyncProducer := kafka.NewMockAsyncProducer(ctrl) syncProducer := kafka.NewMockSyncProducer(ctrl) + ackCh := make(chan func(), 2) asyncProducer.EXPECT().AsyncRunCallback(gomock.Any()).Return(nil).AnyTimes() asyncProducer.EXPECT().AsyncSend(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()). DoAndReturn(func( @@ -318,9 +324,7 @@ func TestKafkaSinkBasicFunctionality(t *testing.T) { _ int32, message *codecCommon.Message, ) error { - if message.Callback != nil { - message.Callback() - } + ackCh <- message.Callback return nil }).Times(2) asyncProducer.EXPECT().Close().AnyTimes() @@ -337,6 +341,19 @@ func TestKafkaSinkBasicFunctionality(t *testing.T) { kafkaSink.AddDMLEvent(dmlEvent) + require.Eventually(t, func() bool { + return len(ackCh) == 2 + }, 5*time.Second, 10*time.Millisecond) + require.Zero(t, dmlFlushCount.Load()) + + (<-ackCh)() + require.Zero(t, dmlFlushCount.Load()) + + (<-ackCh)() + require.Eventually(t, func() bool { + return dmlFlushCount.Load() == 1 + }, 5*time.Second, 10*time.Millisecond) + ddlEvent2.PostFlush() require.Eventually(t, diff --git a/downstreamadapter/sink/mysql/sink.go b/downstreamadapter/sink/mysql/sink.go index 2856d5fce4..c233fca7ef 100644 --- a/downstreamadapter/sink/mysql/sink.go +++ b/downstreamadapter/sink/mysql/sink.go @@ -28,6 +28,7 @@ import ( "github.com/pingcap/ticdc/pkg/errors" "github.com/pingcap/ticdc/pkg/metrics" "github.com/pingcap/ticdc/pkg/sink/mysql" + "github.com/pingcap/ticdc/pkg/statistics" "github.com/pingcap/tidb/pkg/parser/ast" "go.uber.org/atomic" "go.uber.org/zap" @@ -54,7 +55,7 @@ type Sink struct { // Compatibility callers built through NewMySQLSink use one shared pool. dmlDB *sql.DB controlDB *sql.DB - statistics *metrics.Statistics + statistics *statistics.Statistics conflictDetector *causality.ConflictDetector @@ -158,7 +159,7 @@ func newMySQLSinkWithDBs( progressInterval time.Duration, keyspaceID uint32, ) *Sink { - stat := metrics.NewStatistics(changefeedID, keyspaceID, "TxnSink") + stat := statistics.New(changefeedID, keyspaceID) var activeActiveSyncStatsCollector *mysql.ActiveActiveSyncStatsCollector if enableActiveActive && cfg.IsTiDB && cfg.ActiveActiveSyncStatsInterval > 0 { @@ -320,6 +321,7 @@ func (s *Sink) SetTableSchemaStore(tableSchemaStore *commonEvent.TableSchemaStor } func (s *Sink) AddDMLEvent(event *commonEvent.DMLEvent) { + s.statistics.TrackDMLEvent(event) s.conflictDetector.Add(event) } diff --git a/downstreamadapter/sink/pulsar/dml_producer.go b/downstreamadapter/sink/pulsar/dml_producer.go index f311ed7935..e010bb3957 100644 --- a/downstreamadapter/sink/pulsar/dml_producer.go +++ b/downstreamadapter/sink/pulsar/dml_producer.go @@ -27,6 +27,7 @@ import ( "github.com/pingcap/ticdc/pkg/errors" "github.com/pingcap/ticdc/pkg/sink/codec/common" "github.com/pingcap/ticdc/pkg/sink/pulsar" + "github.com/pingcap/ticdc/pkg/statistics" "go.uber.org/zap" ) @@ -52,6 +53,8 @@ type dmlProducers struct { producers *lru.Cache comp component + // statistics is owned and closed by the sink. + statistics *statistics.Statistics // closedMu is used to protect `closed`. // We need to ensure that closed producers are never written to. @@ -104,6 +107,7 @@ func newDMLProducers( p := &dmlProducers{ changefeedID: changefeedID, comp: comp, + statistics: comp.statistics, producers: producers, closed: false, failpointCh: failpointCh, @@ -138,14 +142,16 @@ func (p *dmlProducers) asyncSendMessage( // If producers are closed, we should skip the message and return an error. if p.closed { - return errors.ErrPulsarProducerClosed.GenWithStackByArgs() + return p.handleSendFailure(message, errors.ErrPulsarProducerClosed.GenWithStackByArgs()) } failpoint.Inject("PulsarSinkAsyncSendError", func() { // simulate sending message to input channel successfully but flushing // message to Pulsar meets error log.Info("PulsarSinkAsyncSendError error injected", zap.String("keyspace", p.changefeedID.Keyspace()), zap.String("changefeed", p.changefeedID.ID().String())) - p.failpointCh <- errors.New("pulsar sink injected error") + err := errors.New("pulsar sink injected error") + p.handleSendFailure(message, err) + p.failpointCh <- err failpoint.Return(nil) }) data := &pulsarClient.ProducerMessage{ @@ -155,13 +161,14 @@ func (p *dmlProducers) asyncSendMessage( producer, err := p.getProducerByTopic(topic) if err != nil { - return err + return p.handleSendFailure(message, err) } // if for stress test record , add count to message callback function producer.SendAsync(ctx, data, func(_ pulsarClient.MessageID, m *pulsarClient.ProducerMessage, err error) { + p.handleAsyncSendResult(message, err) // fail if err != nil { e := errors.WrapError(errors.ErrPulsarAsyncSendMessage, err) @@ -184,7 +191,6 @@ func (p *dmlProducers) asyncSendMessage( } } else if message.Callback != nil { // success - message.Callback() pulsar.IncPublishedDMLSuccess(topic, p.changefeedID.String()) } }) @@ -194,6 +200,28 @@ func (p *dmlProducers) asyncSendMessage( return nil } +func (p *dmlProducers) recordDMLResult(rowCount int, err error) { + if p.statistics != nil { + p.statistics.RecordDMLResult(rowCount, err) + } +} + +// handleAsyncSendResult records the result of an asynchronous send and runs +// the message callback on success. It only touches ticdc-owned types so the +// statistics behavior can be unit-tested without any broker machinery. +func (p *dmlProducers) handleAsyncSendResult(message *common.Message, err error) { + p.recordDMLResult(message.GetRowsCount(), err) + if err == nil && message.Callback != nil { + message.Callback() + } +} + +// handleSendFailure records a failed send attempt and returns the error. +func (p *dmlProducers) handleSendFailure(message *common.Message, err error) error { + p.recordDMLResult(message.GetRowsCount(), err) + return err +} + func (p *dmlProducers) close() { // We have to hold the lock to synchronize closing with writing. if p == nil { return diff --git a/downstreamadapter/sink/pulsar/dml_producer_test.go b/downstreamadapter/sink/pulsar/dml_producer_test.go index d2f03a415f..14ed5de095 100644 --- a/downstreamadapter/sink/pulsar/dml_producer_test.go +++ b/downstreamadapter/sink/pulsar/dml_producer_test.go @@ -1,4 +1,4 @@ -// Copyright 2023 PingCAP, Inc. +// Copyright 2026 PingCAP, Inc. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -8,29 +8,117 @@ // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. package pulsar import ( - "context" + "errors" "testing" - "github.com/pingcap/ticdc/pkg/sink/codec/common" - "github.com/pingcap/ticdc/pkg/util" + "github.com/pingcap/ticdc/pkg/common" + codecCommon "github.com/pingcap/ticdc/pkg/sink/codec/common" + "github.com/pingcap/ticdc/pkg/statistics" + "github.com/prometheus/client_golang/prometheus" + dto "github.com/prometheus/client_model/go" "github.com/stretchr/testify/require" ) -func TestPulsarSyncAsyncSendMessage(t *testing.T) { - t.Parallel() - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - - p := newMockDMLProducer() - err := p.asyncSendMessage(ctx, "test", &common.Message{ - Value: []byte("this value for test input data"), - PartitionKey: util.AddressOf("test_key"), - }) +func gatherMetric( + t *testing.T, reg *prometheus.Registry, name string, labelValues ...string, +) *dto.Metric { + t.Helper() + require.Lenf(t, labelValues, len(labelValues)&^1, "labelValues must be key/value pairs") + mfs, err := reg.Gather() require.NoError(t, err) + for _, mf := range mfs { + if mf.GetName() != name { + continue + } + for _, m := range mf.GetMetric() { + matched := true + for i := 0; i < len(labelValues); i += 2 { + found := false + for _, lp := range m.GetLabel() { + if lp.GetName() == labelValues[i] && lp.GetValue() == labelValues[i+1] { + found = true + break + } + } + if !found { + matched = false + break + } + } + if matched { + return m + } + } + } + return nil +} + +func newTestMessage(rows int, callback func()) *codecCommon.Message { + message := &codecCommon.Message{Key: []byte("k"), Value: []byte("v")} + message.SetRowsCount(rows) + message.Callback = callback + return message +} + +func newTestDMLProducers(t *testing.T, changefeed string) (*dmlProducers, *prometheus.Registry) { + t.Helper() + reg := prometheus.NewRegistry() + statistics.InitMetrics(reg) + stat := statistics.New(common.NewChangefeedID4Test("test-keyspace", changefeed), 123) + t.Cleanup(stat.Close) + return &dmlProducers{statistics: stat}, reg +} + +func TestHandleAsyncSendResultSuccess(t *testing.T) { + p, reg := newTestDMLProducers(t, "async-send-success") + + callbackCalled := make(chan struct{}) + message := newTestMessage(2, func() { close(callbackCalled) }) + p.handleAsyncSendResult(message, nil) + + <-callbackCalled + // The row count is observed into the batch histogram on success. + hist := gatherMetric(t, reg, "ticdc_sink_batch_row_count", + "namespace", "test-keyspace", "changefeed", "async-send-success") + require.NotNil(t, hist) + require.Equal(t, uint64(1), hist.GetHistogram().GetSampleCount()) + require.Equal(t, float64(2), hist.GetHistogram().GetSampleSum()) +} + +func TestHandleAsyncSendResultErrorSkipsCallback(t *testing.T) { + p, reg := newTestDMLProducers(t, "async-send-error") + + message := newTestMessage(4, func() { t.Fatal("callback must not run on failure") }) + p.handleAsyncSendResult(message, errors.New("broker boom")) + + errMetric := gatherMetric(t, reg, "ticdc_sink_execution_error", + "namespace", "test-keyspace", "changefeed", "async-send-error", "event_type", "dml") + require.NotNil(t, errMetric) + require.Equal(t, float64(1), errMetric.GetCounter().GetValue()) + // No rows are observed for failed attempts; the histogram series exists + // (created eagerly by New) but has no samples. + hist := gatherMetric(t, reg, "ticdc_sink_batch_row_count", + "namespace", "test-keyspace", "changefeed", "async-send-error") + require.NotNil(t, hist) + require.Equal(t, uint64(0), hist.GetHistogram().GetSampleCount()) +} + +func TestHandleSendFailureRecordsErrorAndReturnsIt(t *testing.T) { + p, reg := newTestDMLProducers(t, "send-failure") + + sentinel := errors.New("producer closed") + message := newTestMessage(4, nil) + require.ErrorIs(t, p.handleSendFailure(message, sentinel), sentinel) + + errMetric := gatherMetric(t, reg, "ticdc_sink_execution_error", + "namespace", "test-keyspace", "changefeed", "send-failure", "event_type", "dml") + require.NotNil(t, errMetric) + require.Equal(t, float64(1), errMetric.GetCounter().GetValue()) } diff --git a/downstreamadapter/sink/pulsar/helper.go b/downstreamadapter/sink/pulsar/helper.go index b2676b1daa..1e5a19816b 100644 --- a/downstreamadapter/sink/pulsar/helper.go +++ b/downstreamadapter/sink/pulsar/helper.go @@ -29,6 +29,7 @@ import ( "github.com/pingcap/ticdc/pkg/sink/codec" codecCommon "github.com/pingcap/ticdc/pkg/sink/codec/common" "github.com/pingcap/ticdc/pkg/sink/pulsar" + "github.com/pingcap/ticdc/pkg/statistics" putil "github.com/pingcap/ticdc/pkg/util" "go.uber.org/zap" ) @@ -41,6 +42,8 @@ type component struct { eventRouter *eventrouter.EventRouter topicManager topicmanager.TopicManager client pulsarClient.Client + // statistics is a construction dependency owned by the sink. + statistics *statistics.Statistics } func (c component) close() { diff --git a/downstreamadapter/sink/pulsar/mock_producer.go b/downstreamadapter/sink/pulsar/mock_producer.go index a71d6a1247..31cf213790 100644 --- a/downstreamadapter/sink/pulsar/mock_producer.go +++ b/downstreamadapter/sink/pulsar/mock_producer.go @@ -30,6 +30,8 @@ var ( type mockProducer struct { mu sync.Mutex events map[string][]*pulsar.ProducerMessage + // ackCh lets tests delay callbacks until they simulate a successful broker ack. + ackCh chan func() } func newMockDDLProducer() ddlProducer { @@ -74,12 +76,18 @@ func (p *mockProducer) GetProducerByTopic(_ string) (producer pulsar.Producer, e func (p *mockProducer) asyncSendMessage(_ context.Context, topic string, message *common.Message, ) error { p.mu.Lock() - defer p.mu.Unlock() data := &pulsar.ProducerMessage{ Payload: message.Value, Key: message.GetPartitionKey(), } p.events[topic] = append(p.events[topic], data) + ackCh := p.ackCh + p.mu.Unlock() + + if ackCh != nil { + ackCh <- message.Callback + return nil + } if message.Callback != nil { message.Callback() } @@ -93,11 +101,15 @@ func (m *mockProducer) run(_ context.Context) error { // Close close all producers func (p *mockProducer) close() { + p.mu.Lock() + defer p.mu.Unlock() p.events = make(map[string][]*pulsar.ProducerMessage) } // GetAllEvents returns the events received by the mock producer. func (p *mockProducer) GetAllEvents() []*pulsar.ProducerMessage { + p.mu.Lock() + defer p.mu.Unlock() var events []*pulsar.ProducerMessage for _, v := range p.events { events = append(events, v...) @@ -107,5 +119,7 @@ func (p *mockProducer) GetAllEvents() []*pulsar.ProducerMessage { // GetEvents returns the event filtered by the key. func (p *mockProducer) GetEvents(topic string) []*pulsar.ProducerMessage { + p.mu.Lock() + defer p.mu.Unlock() return p.events[topic] } diff --git a/downstreamadapter/sink/pulsar/sink.go b/downstreamadapter/sink/pulsar/sink.go index 9895541233..ad2b0dfa98 100644 --- a/downstreamadapter/sink/pulsar/sink.go +++ b/downstreamadapter/sink/pulsar/sink.go @@ -26,6 +26,7 @@ import ( "github.com/pingcap/ticdc/pkg/errors" "github.com/pingcap/ticdc/pkg/metrics" "github.com/pingcap/ticdc/pkg/sink/codec/common" + "github.com/pingcap/ticdc/pkg/statistics" "github.com/pingcap/ticdc/utils/chann" "go.uber.org/atomic" "go.uber.org/zap" @@ -51,7 +52,7 @@ type sink struct { ddlProducer ddlProducer comp component - statistics *metrics.Statistics + statistics *statistics.Statistics protocol config.Protocol partitionRule helper.DDLDispatchRule @@ -132,7 +133,7 @@ func newWithComponent( var ( dmlProducer dmlProducer ddlProducer ddlProducer - statistics *metrics.Statistics + stat *statistics.Statistics ) defer func() { if err != nil { @@ -142,15 +143,16 @@ func newWithComponent( if dmlProducer != nil { dmlProducer.close() } - if statistics != nil { - statistics.Close() + if stat != nil { + stat.Close() } comp.close() } }() failpointCh := make(chan error, 1) - statistics = metrics.NewStatistics(changefeedID, keyspaceID, "pulsar") + stat = statistics.New(changefeedID, keyspaceID) + comp.statistics = stat dmlProducer, err = newDMLProducer(changefeedID, comp, failpointCh) if err != nil { return nil, err @@ -173,7 +175,7 @@ func newWithComponent( protocol: protocol, partitionRule: helper.GetDDLDispatchRule(protocol), comp: comp, - statistics: statistics, + statistics: stat, isNormal: atomic.NewBool(true), ctx: ctx, }, nil @@ -197,6 +199,7 @@ func (s *sink) IsNormal() bool { } func (s *sink) AddDMLEvent(event *commonEvent.DMLEvent) { + s.statistics.TrackDMLEvent(event) s.eventChan.Push(event) } @@ -533,14 +536,10 @@ func (s *sink) sendMessages(ctx context.Context) error { } for _, message := range future.Messages { start := time.Now() - if err = s.statistics.RecordBatchExecution(func() (int, int64, error) { - message.SetPartitionKey(future.Key.PartitionKey) - if err = s.dmlProducer.asyncSendMessage(ctx, future.Key.Topic, message); err != nil { - return 0, 0, err - } - return message.GetRowsCount(), int64(message.Length()), nil - }); err != nil { - return errors.Trace(err) + message.SetPartitionKey(future.Key.PartitionKey) + if err = s.dmlProducer.asyncSendMessage(ctx, future.Key.Topic, message); err != nil { + err = errors.Trace(err) + return err } metricSendMessageDuration.Observe(time.Since(start).Seconds()) } diff --git a/downstreamadapter/sink/pulsar/sink_test.go b/downstreamadapter/sink/pulsar/sink_test.go index 973e4f15e6..2f1c46890b 100644 --- a/downstreamadapter/sink/pulsar/sink_test.go +++ b/downstreamadapter/sink/pulsar/sink_test.go @@ -25,7 +25,7 @@ import ( commonEvent "github.com/pingcap/ticdc/pkg/common/event" "github.com/pingcap/ticdc/pkg/config" cerror "github.com/pingcap/ticdc/pkg/errors" - "github.com/pingcap/ticdc/pkg/metrics" + "github.com/pingcap/ticdc/pkg/statistics" "github.com/pingcap/ticdc/utils/chann" "github.com/stretchr/testify/require" "go.uber.org/atomic" @@ -48,7 +48,7 @@ func newPulsarSinkForTest(t *testing.T) (*sink, error) { comp, protocol, err := newPulsarSinkComponentForTest(ctx, changefeedID, sinkURI, replicaConfig.Sink) require.NoError(t, err) - statistics := metrics.NewStatistics(changefeedID, common.DefaultKeyspaceID, "sink") + statistics := statistics.New(changefeedID, common.DefaultKeyspaceID) pulsarSink := &sink{ changefeedID: changefeedID, dmlProducer: newMockDMLProducer(), @@ -74,7 +74,7 @@ func TestPulsarSinkBasicFunctionality(t *testing.T) { pulsarSink, err := newPulsarSinkForTest(t) require.NoError(t, err) - var count atomic.Int64 + var count, dmlFlushCount atomic.Int64 helper := commonEvent.NewEventTestHelper(t) defer helper.Close() @@ -116,19 +116,35 @@ func TestPulsarSinkBasicFunctionality(t *testing.T) { dmlEvent := helper.DML2Event("test", "t", "insert into t values (1, 'test')", "insert into t values (2, 'test2');") dmlEvent.PostTxnFlushed = []func(){ - func() { count.Add(1) }, + func() { + count.Add(1) + dmlFlushCount.Add(1) + }, } dmlEvent.CommitTs = 2 + producer := pulsarSink.dmlProducer.(*mockProducer) + producer.ackCh = make(chan func(), 2) err = pulsarSink.WriteBlockEvent(ddlEvent) require.NoError(t, err) pulsarSink.AddDMLEvent(dmlEvent) - time.Sleep(1 * time.Second) + require.Eventually(t, func() bool { + return len(producer.ackCh) == 2 + }, 5*time.Second, 10*time.Millisecond) + require.Zero(t, dmlFlushCount.Load()) + + (<-producer.ackCh)() + require.Zero(t, dmlFlushCount.Load()) + + (<-producer.ackCh)() + require.Eventually(t, func() bool { + return dmlFlushCount.Load() == 1 + }, 5*time.Second, 10*time.Millisecond) ddlEvent2.PostFlush() - require.Len(t, pulsarSink.dmlProducer.(*mockProducer).GetAllEvents(), 2) + require.Len(t, producer.GetAllEvents(), 2) require.Len(t, pulsarSink.ddlProducer.(*mockProducer).GetAllEvents(), 1) require.Equal(t, count.Load(), int64(3)) diff --git a/metrics/grafana/ticdc_new_arch.json b/metrics/grafana/ticdc_new_arch.json index 0a3f4bbd37..a7ef6a1d22 100644 --- a/metrics/grafana/ticdc_new_arch.json +++ b/metrics/grafana/ticdc_new_arch.json @@ -580,7 +580,7 @@ "targets": [ { "exemplar": true, - "expr": "sum(rate(ticdc_sink_dml_event_count{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\",namespace=~\"$namespace\", changefeed=~\"$changefeed\", instance=~\"$ticdc_instance\"}[1m])) by (namespace,changefeed, instance)", + "expr": "sum(rate(ticdc_sink_batch_row_count_sum{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\",namespace=~\"$namespace\", changefeed=~\"$changefeed\", instance=~\"$ticdc_instance\"}[1m])) by (namespace,changefeed, instance)", "format": "time_series", "interval": "", "intervalFactor": 1, @@ -630,6 +630,7 @@ "dashLength": 10, "dashes": false, "datasource": "${DS_TEST-CLUSTER}", + "description": "Approximate raw-entry bytes of DML events successfully written to downstream.", "fieldConfig": { "defaults": {}, "overrides": [] @@ -672,19 +673,19 @@ "targets": [ { "exemplar": true, - "expr": "sum(rate(ticdc_sink_write_bytes_total{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", namespace=~\"$namespace\", changefeed=~\"$changefeed\",instance=~\"$ticdc_instance\"}[1m])) by (instance, type, changefeed)", + "expr": "sum(rate(ticdc_sink_write_bytes_total{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", namespace=~\"$namespace\", changefeed=~\"$changefeed\",instance=~\"$ticdc_instance\"}[1m])) by (namespace, changefeed, instance)", "hide": false, "interval": "", - "legendFormat": "{{instance}}-{{changefeed}}-{{type}}", + "legendFormat": "{{namespace}}-{{changefeed}}-{{instance}}", "queryType": "randomWalk", "refId": "A" }, { "exemplar": true, - "expr": "avg_over_time(sum(rate(ticdc_sink_write_bytes_total{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", namespace=~\"$namespace\", changefeed=~\"$changefeed\",instance=~\"$ticdc_instance\"}[1m])) by (instance, type, changefeed)[1m:])", + "expr": "avg_over_time(sum(rate(ticdc_sink_write_bytes_total{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", namespace=~\"$namespace\", changefeed=~\"$changefeed\",instance=~\"$ticdc_instance\"}[1m])) by (namespace, changefeed, instance)[1m:])", "hide": true, "interval": "", - "legendFormat": "{{instance}}-{{changefeed}}-AVG", + "legendFormat": "{{namespace}}-{{changefeed}}-{{instance}}-AVG", "refId": "B" } ], @@ -16929,7 +16930,7 @@ "targets": [ { "exemplar": true, - "expr": "sum(delta(ticdc_sink_execution_error{k8s_cluster=\"$k8s_cluster\",tidb_cluster=\"$tidb_cluster\",namespace=~\"$namespace\",changefeed=~\"$changefeed\",instance=~\"$ticdc_instance\"}[1m])) by (namespace,changefeed,instance,event_type)", + "expr": "sum(increase(ticdc_sink_execution_error{k8s_cluster=\"$k8s_cluster\",tidb_cluster=\"$tidb_cluster\",namespace=~\"$namespace\",changefeed=~\"$changefeed\",instance=~\"$ticdc_instance\"}[1m])) by (namespace,changefeed,instance,event_type)", "interval": "", "legendFormat": "{{namespace}}-{{changefeed}}-{{instance}}-{{event_type}}", "queryType": "randomWalk", @@ -26590,7 +26591,7 @@ "targets": [ { "exemplar": true, - "expr": "sum(delta(ticdc_ddl_execution{k8s_cluster=\"$k8s_cluster\",tidb_cluster=\"$tidb_cluster\",namespace=~\"$namespace\",changefeed=~\"$changefeed\",instance=~\"$ticdc_instance\"}[1m])) by (namespace,changefeed,ddl_type)", + "expr": "sum(increase(ticdc_ddl_execution{k8s_cluster=\"$k8s_cluster\",tidb_cluster=\"$tidb_cluster\",namespace=~\"$namespace\",changefeed=~\"$changefeed\",instance=~\"$ticdc_instance\"}[1m])) by (namespace,changefeed,ddl_type)", "interval": "", "legendFormat": "{{namespace}}-{{changefeed}}-{{ddl_type}}", "queryType": "randomWalk", @@ -28342,5 +28343,5 @@ "timezone": "browser", "title": "${DS_TEST-CLUSTER}-TiCDC-New-Arch", "uid": "YiGL8hBZ0aac", - "version": 41 + "version": 42 } diff --git a/metrics/nextgengrafana/ticdc_new_arch_next_gen.json b/metrics/nextgengrafana/ticdc_new_arch_next_gen.json index f920ac2ccd..6d709b5ad9 100644 --- a/metrics/nextgengrafana/ticdc_new_arch_next_gen.json +++ b/metrics/nextgengrafana/ticdc_new_arch_next_gen.json @@ -580,7 +580,7 @@ "targets": [ { "exemplar": true, - "expr": "sum(rate(ticdc_sink_dml_event_count{k8s_cluster=\"$k8s_cluster\", sharedpool_id=\"$tidb_cluster\",keyspace_name=~\"$keyspace_name\", changefeed=~\"$changefeed\", instance=~\"$ticdc_instance\"}[1m])) by (keyspace_name,changefeed, instance)", + "expr": "sum(rate(ticdc_sink_batch_row_count_sum{k8s_cluster=\"$k8s_cluster\", sharedpool_id=\"$tidb_cluster\",keyspace_name=~\"$keyspace_name\", changefeed=~\"$changefeed\", instance=~\"$ticdc_instance\"}[1m])) by (keyspace_name,changefeed, instance)", "format": "time_series", "interval": "", "intervalFactor": 1, @@ -630,6 +630,7 @@ "dashLength": 10, "dashes": false, "datasource": "${DS_TEST-CLUSTER}", + "description": "Approximate raw-entry bytes of DML events successfully written to downstream.", "fieldConfig": { "defaults": {}, "overrides": [] @@ -672,19 +673,19 @@ "targets": [ { "exemplar": true, - "expr": "sum(rate(ticdc_sink_write_bytes_total{k8s_cluster=\"$k8s_cluster\", sharedpool_id=\"$tidb_cluster\", keyspace_name=~\"$keyspace_name\", changefeed=~\"$changefeed\",instance=~\"$ticdc_instance\"}[1m])) by (instance, type, changefeed)", + "expr": "sum(rate(ticdc_sink_write_bytes_total{k8s_cluster=\"$k8s_cluster\", sharedpool_id=\"$tidb_cluster\", keyspace_name=~\"$keyspace_name\", changefeed=~\"$changefeed\",instance=~\"$ticdc_instance\"}[1m])) by (keyspace_name, changefeed, instance)", "hide": false, "interval": "", - "legendFormat": "{{instance}}-{{changefeed}}-{{type}}", + "legendFormat": "{{keyspace_name}}-{{changefeed}}-{{instance}}", "queryType": "randomWalk", "refId": "A" }, { "exemplar": true, - "expr": "avg_over_time(sum(rate(ticdc_sink_write_bytes_total{k8s_cluster=\"$k8s_cluster\", sharedpool_id=\"$tidb_cluster\", keyspace_name=~\"$keyspace_name\", changefeed=~\"$changefeed\",instance=~\"$ticdc_instance\"}[1m])) by (instance, type, changefeed)[1m:])", + "expr": "avg_over_time(sum(rate(ticdc_sink_write_bytes_total{k8s_cluster=\"$k8s_cluster\", sharedpool_id=\"$tidb_cluster\", keyspace_name=~\"$keyspace_name\", changefeed=~\"$changefeed\",instance=~\"$ticdc_instance\"}[1m])) by (keyspace_name, changefeed, instance)[1m:])", "hide": true, "interval": "", - "legendFormat": "{{instance}}-{{changefeed}}-AVG", + "legendFormat": "{{keyspace_name}}-{{changefeed}}-{{instance}}-AVG", "refId": "B" } ], @@ -16929,7 +16930,7 @@ "targets": [ { "exemplar": true, - "expr": "sum(delta(ticdc_sink_execution_error{k8s_cluster=\"$k8s_cluster\",sharedpool_id=\"$tidb_cluster\",keyspace_name=~\"$keyspace_name\",changefeed=~\"$changefeed\",instance=~\"$ticdc_instance\"}[1m])) by (keyspace_name,changefeed,instance,event_type)", + "expr": "sum(increase(ticdc_sink_execution_error{k8s_cluster=\"$k8s_cluster\",sharedpool_id=\"$tidb_cluster\",keyspace_name=~\"$keyspace_name\",changefeed=~\"$changefeed\",instance=~\"$ticdc_instance\"}[1m])) by (keyspace_name,changefeed,instance,event_type)", "interval": "", "legendFormat": "{{keyspace_name}}-{{changefeed}}-{{instance}}-{{event_type}}", "queryType": "randomWalk", @@ -26590,7 +26591,7 @@ "targets": [ { "exemplar": true, - "expr": "sum(delta(ticdc_ddl_execution{k8s_cluster=\"$k8s_cluster\",sharedpool_id=\"$tidb_cluster\",keyspace_name=~\"$keyspace_name\",changefeed=~\"$changefeed\",instance=~\"$ticdc_instance\"}[1m])) by (keyspace_name,changefeed,ddl_type)", + "expr": "sum(increase(ticdc_ddl_execution{k8s_cluster=\"$k8s_cluster\",sharedpool_id=\"$tidb_cluster\",keyspace_name=~\"$keyspace_name\",changefeed=~\"$changefeed\",instance=~\"$ticdc_instance\"}[1m])) by (keyspace_name,changefeed,ddl_type)", "interval": "", "legendFormat": "{{keyspace_name}}-{{changefeed}}-{{ddl_type}}", "queryType": "randomWalk", @@ -28342,5 +28343,5 @@ "timezone": "browser", "title": "${DS_TEST-CLUSTER}-TiCDC-New-Arch", "uid": "YiGL8hBZ0aac", - "version": 41 + "version": 42 } diff --git a/metrics/nextgengrafana/ticdc_new_arch_with_keyspace_name.json b/metrics/nextgengrafana/ticdc_new_arch_with_keyspace_name.json index f71e45145a..1432b9b26b 100644 --- a/metrics/nextgengrafana/ticdc_new_arch_with_keyspace_name.json +++ b/metrics/nextgengrafana/ticdc_new_arch_with_keyspace_name.json @@ -391,7 +391,7 @@ "targets": [ { "exemplar": true, - "expr": "sum(rate(ticdc_sink_dml_event_count{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\",keyspace_name=~\"$keyspace_name\", changefeed=~\"$changefeed\", instance=~\"$ticdc_instance\"}[1m])) by (keyspace_name,changefeed, instance)", + "expr": "sum(rate(ticdc_sink_batch_row_count_sum{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\",keyspace_name=~\"$keyspace_name\", changefeed=~\"$changefeed\", instance=~\"$ticdc_instance\"}[1m])) by (keyspace_name,changefeed, instance)", "format": "time_series", "interval": "", "intervalFactor": 1, @@ -441,6 +441,7 @@ "dashLength": 10, "dashes": false, "datasource": "${DS_TEST-CLUSTER}", + "description": "Approximate raw-entry bytes of DML events successfully written to downstream.", "fieldConfig": { "defaults": {}, "overrides": [] @@ -483,19 +484,19 @@ "targets": [ { "exemplar": true, - "expr": "sum(rate(ticdc_sink_write_bytes_total{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", keyspace_name=~\"$keyspace_name\", changefeed=~\"$changefeed\",instance=~\"$ticdc_instance\"}[1m])) by (instance, type, changefeed)", + "expr": "sum(rate(ticdc_sink_write_bytes_total{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", keyspace_name=~\"$keyspace_name\", changefeed=~\"$changefeed\",instance=~\"$ticdc_instance\"}[1m])) by (keyspace_name, changefeed, instance)", "hide": false, "interval": "", - "legendFormat": "{{instance}}-{{changefeed}}-{{type}}", + "legendFormat": "{{keyspace_name}}-{{changefeed}}-{{instance}}", "queryType": "randomWalk", "refId": "A" }, { "exemplar": true, - "expr": "avg_over_time(sum(rate(ticdc_sink_write_bytes_total{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", keyspace_name=~\"$keyspace_name\", changefeed=~\"$changefeed\",instance=~\"$ticdc_instance\"}[1m])) by (instance, type, changefeed)[1m:])", + "expr": "avg_over_time(sum(rate(ticdc_sink_write_bytes_total{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", keyspace_name=~\"$keyspace_name\", changefeed=~\"$changefeed\",instance=~\"$ticdc_instance\"}[1m])) by (keyspace_name, changefeed, instance)[1m:])", "hide": true, "interval": "", - "legendFormat": "{{instance}}-{{changefeed}}-AVG", + "legendFormat": "{{keyspace_name}}-{{changefeed}}-{{instance}}-AVG", "refId": "B" } ], @@ -5454,7 +5455,7 @@ "targets": [ { "exemplar": true, - "expr": "sum(delta(ticdc_sink_execution_error{k8s_cluster=\"$k8s_cluster\",tidb_cluster=\"$tidb_cluster\",keyspace_name=~\"$keyspace_name\",changefeed=~\"$changefeed\",instance=~\"$ticdc_instance\"}[1m])) by (keyspace_name,changefeed,instance,event_type)", + "expr": "sum(increase(ticdc_sink_execution_error{k8s_cluster=\"$k8s_cluster\",tidb_cluster=\"$tidb_cluster\",keyspace_name=~\"$keyspace_name\",changefeed=~\"$changefeed\",instance=~\"$ticdc_instance\"}[1m])) by (keyspace_name,changefeed,instance,event_type)", "interval": "", "legendFormat": "{{keyspace_name}}-{{changefeed}}-{{instance}}-{{event_type}}", "queryType": "randomWalk", @@ -11303,7 +11304,7 @@ "targets": [ { "exemplar": true, - "expr": "sum(delta(ticdc_ddl_execution{k8s_cluster=\"$k8s_cluster\",tidb_cluster=\"$tidb_cluster\",keyspace_name=~\"$keyspace_name\",changefeed=~\"$changefeed\",instance=~\"$ticdc_instance\"}[1m])) by (keyspace_name,changefeed,ddl_type)", + "expr": "sum(increase(ticdc_ddl_execution{k8s_cluster=\"$k8s_cluster\",tidb_cluster=\"$tidb_cluster\",keyspace_name=~\"$keyspace_name\",changefeed=~\"$changefeed\",instance=~\"$ticdc_instance\"}[1m])) by (keyspace_name,changefeed,ddl_type)", "interval": "", "legendFormat": "{{keyspace_name}}-{{changefeed}}-{{ddl_type}}", "queryType": "randomWalk", @@ -11749,5 +11750,5 @@ "timezone": "browser", "title": "${DS_TEST-CLUSTER}-TiCDC-New-Arch-KeyspaceName", "uid": "lGT5hED6vqTn", - "version": 41 + "version": 42 } diff --git a/pkg/metrics/ddl.go b/pkg/metrics/ddl.go index 5194e167fe..9b0c61098d 100644 --- a/pkg/metrics/ddl.go +++ b/pkg/metrics/ddl.go @@ -29,25 +29,6 @@ var ( Buckets: prometheus.ExponentialBuckets(0.01, 2, 18), }, []string{getKeyspaceLabel(), "changefeed"}) - // ExecDDLHistogram records the execution time of a DDL. - ExecDDLHistogram = prometheus.NewHistogramVec( - prometheus.HistogramOpts{ - Namespace: "ticdc", - Subsystem: "ddl", - Name: "exec_duration", - Help: "Bucketed histogram of processing time (s) of a ddl.", - Buckets: prometheus.ExponentialBuckets(0.01, 2, 18), - }, []string{getKeyspaceLabel(), "changefeed"}) - - // ExecDDLRunningGauge records the count of running DDL. - ExecDDLRunningGauge = prometheus.NewGaugeVec( - prometheus.GaugeOpts{ - Namespace: "ticdc", - Subsystem: "ddl", - Name: "exec_running", - Help: "Total count of running ddl.", - }, []string{getKeyspaceLabel(), "changefeed"}) - // ExecDDLBlockingGauge records the count of blocking DDL. ExecDDLBlockingGauge = prometheus.NewGaugeVec( prometheus.GaugeOpts{ @@ -56,21 +37,9 @@ var ( Name: "exec_blocking", Help: "Total count of blocking ddl.", }, []string{getKeyspaceLabel(), "changefeed", "mode"}) - - // ExecDDLCounter records the execution count of different DDL types - ExecDDLCounter = prometheus.NewCounterVec( - prometheus.CounterOpts{ - Namespace: "ticdc", - Subsystem: "ddl", - Name: "execution", - Help: "Total execution count of different DDL types.", - }, []string{getKeyspaceLabel(), "changefeed", "ddl_type"}) ) func initDDLMetrics(registry *prometheus.Registry) { registry.MustRegister(HandleDDLHistogram) - registry.MustRegister(ExecDDLHistogram) - registry.MustRegister(ExecDDLRunningGauge) registry.MustRegister(ExecDDLBlockingGauge) - registry.MustRegister(ExecDDLCounter) } diff --git a/pkg/metrics/init.go b/pkg/metrics/init.go index 7583b99155..39e39725cb 100644 --- a/pkg/metrics/init.go +++ b/pkg/metrics/init.go @@ -18,6 +18,7 @@ import ( "github.com/pingcap/ticdc/pkg/common" "github.com/pingcap/ticdc/pkg/config/kerneltype" "github.com/pingcap/ticdc/pkg/sink/kafka" + "github.com/pingcap/ticdc/pkg/statistics" "github.com/pingcap/ticdc/pkg/txnutil/gc" "github.com/prometheus/client_golang/prometheus" ) @@ -31,6 +32,7 @@ func InitMetrics(registry *prometheus.Registry) { initDispatcherMetrics(registry) initMessagingMetrics(registry) initSinkMetrics(registry) + statistics.InitMetrics(registry) initEventStoreMetrics(registry) initSchemaStoreMetrics(registry) initEventServiceMetrics(registry) diff --git a/pkg/metrics/sink.go b/pkg/metrics/sink.go index a45582e6c2..a6c628ac32 100644 --- a/pkg/metrics/sink.go +++ b/pkg/metrics/sink.go @@ -18,78 +18,13 @@ import "github.com/prometheus/client_golang/prometheus" // LargeRowSizeLowBound is set to 2K, only track data event with size not smaller than it. const LargeRowSizeLowBound = 2 * 1024 -// ---------- Metrics used in Statistics. ---------- // -var ( - // ExecBatchHistogram records batch size of a txn. - ExecBatchHistogram = prometheus.NewHistogramVec( - prometheus.HistogramOpts{ - Namespace: "ticdc", - Subsystem: "sink", - Name: "batch_row_count", - Help: "Row count number for a given batch.", - Buckets: prometheus.ExponentialBuckets(1, 2, 18), - }, []string{getKeyspaceLabel(), "changefeed", "type", "keyspace_id"}) // type is for `sinkType` - - // ExecBatchWriteBytesHistogram records bytes written for each batch. - ExecBatchWriteBytesHistogram = prometheus.NewHistogramVec( - prometheus.HistogramOpts{ - Namespace: "ticdc", - Subsystem: "sink", - Name: "batch_write_bytes", - Help: "Bytes number for a given batch.", - Buckets: prometheus.ExponentialBuckets(1024, 2, 18), // 1KB~128MB - }, []string{getKeyspaceLabel(), "changefeed", "type"}) // type is for `sinkType` - - // ExecWriteBytesGauge records the total number of bytes written by sink. - TotalWriteBytesCounter = prometheus.NewCounterVec( - prometheus.CounterOpts{ - Namespace: "ticdc", - Subsystem: "sink", - Name: "write_bytes_total", - Help: "Total number of bytes written by sink", - }, []string{getKeyspaceLabel(), "changefeed", "type"}) // type is for `sinkType` - - EventSizeHistogram = prometheus.NewHistogramVec( - prometheus.HistogramOpts{ - Namespace: "ticdc", - Subsystem: "sink", - Name: "event_size", - Help: "The size of changed events (in bytes).", - Buckets: prometheus.ExponentialBuckets(0.01, 2, 30), // 0~32M - }, []string{getKeyspaceLabel(), "changefeed"}) - - ExecDMLEventCounter = prometheus.NewCounterVec( - prometheus.CounterOpts{ - Namespace: "ticdc", - Subsystem: "sink", - Name: "dml_event_count", - Help: "Total count of DML events.", - }, []string{getKeyspaceLabel(), "changefeed"}) - - ExecDMLEventRowsAffectedCounter = prometheus.NewCounterVec( - prometheus.CounterOpts{ - Namespace: "ticdc", - Subsystem: "sink", - Name: "dml_event_affected_row_count", - Help: "Total count of affected rows.", - }, []string{getKeyspaceLabel(), "changefeed", "count_type", "row_type"}) - - ActiveActiveConflictSkipRowsCounter = prometheus.NewCounterVec( - prometheus.CounterOpts{ - Namespace: "ticdc", - Subsystem: "sink", - Name: "active_active_conflict_skip_rows_total", - Help: "Total number of rows skipped due to last-write-wins conflict resolution in TiDB active-active replication.", - }, []string{getKeyspaceLabel(), "changefeed"}) - // ExecutionErrorCounter is the counter of execution errors. - ExecutionErrorCounter = prometheus.NewCounterVec( - prometheus.CounterOpts{ - Namespace: "ticdc", - Subsystem: "sink", - Name: "execution_error", - Help: "Total count of execution errors.", - }, []string{getKeyspaceLabel(), "changefeed", "event_type"}) -) +var ActiveActiveConflictSkipRowsCounter = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: "ticdc", + Subsystem: "sink", + Name: "active_active_conflict_skip_rows_total", + Help: "Total number of rows skipped due to last-write-wins conflict resolution in TiDB active-active replication.", + }, []string{getKeyspaceLabel(), "changefeed"}) // ---------- Metrics for txn sink and backends. ---------- // var ( @@ -233,15 +168,7 @@ var ( // InitMetrics registers all metrics in this file. func initSinkMetrics(registry *prometheus.Registry) { - // common sink metrics - registry.MustRegister(ExecBatchHistogram) - registry.MustRegister(ExecBatchWriteBytesHistogram) - registry.MustRegister(TotalWriteBytesCounter) - registry.MustRegister(EventSizeHistogram) - registry.MustRegister(ExecDMLEventCounter) - registry.MustRegister(ExecDMLEventRowsAffectedCounter) registry.MustRegister(ActiveActiveConflictSkipRowsCounter) - registry.MustRegister(ExecutionErrorCounter) // txn sink metrics registry.MustRegister(ConflictDetectDuration) diff --git a/pkg/metrics/statistics.go b/pkg/metrics/statistics.go deleted file mode 100644 index f1815a63e5..0000000000 --- a/pkg/metrics/statistics.go +++ /dev/null @@ -1,167 +0,0 @@ -// Copyright 2020 PingCAP, Inc. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// See the License for the specific language governing permissions and -// limitations under the License. - -package metrics - -import ( - "fmt" - "strings" - "sync" - "time" - - "github.com/pingcap/ticdc/pkg/common" - "github.com/prometheus/client_golang/prometheus" -) - -// NewStatistics creates a statistics -func NewStatistics( - changefeed common.ChangeFeedID, - keyspaceID uint32, - sinkType string, -) *Statistics { - statistics := &Statistics{ - sinkType: sinkType, - changefeedID: changefeed, - keyspaceID: FormatKeyspaceID(keyspaceID), - ddlTypes: sync.Map{}, - rowsAffectedMap: sync.Map{}, - } - - keyspace := changefeed.Keyspace() - changefeedID := changefeed.Name() - statistics.metricExecDDLHis = ExecDDLHistogram.WithLabelValues(keyspace, changefeedID) - statistics.metricExecDDLRunningCnt = ExecDDLRunningGauge.WithLabelValues(keyspace, changefeedID) - statistics.metricExecBatchHis = ExecBatchHistogram.WithLabelValues(keyspace, changefeedID, sinkType, statistics.keyspaceID) - statistics.metricExecBatchBytesHis = ExecBatchWriteBytesHistogram.WithLabelValues(keyspace, changefeedID, sinkType) - statistics.metricTotalWriteBytesCnt = TotalWriteBytesCounter.WithLabelValues(keyspace, changefeedID, sinkType) - statistics.metricExecErrCntForDDL = ExecutionErrorCounter.WithLabelValues(keyspace, changefeedID, "ddl") - statistics.metricExecErrCntForDML = ExecutionErrorCounter.WithLabelValues(keyspace, changefeedID, "dml") - statistics.metricExecDMLCnt = ExecDMLEventCounter.WithLabelValues(keyspace, changefeedID) - - return statistics -} - -// Statistics maintains some status and metrics of the Sink -// Note: All methods of Statistics should be thread-safe. -type Statistics struct { - sinkType string - changefeedID common.ChangeFeedID - keyspaceID string - ddlTypes sync.Map - rowsAffectedMap sync.Map - - // metricExecDDLHis records each DDL execution time duration. - metricExecDDLHis prometheus.Observer - // metricExecDDLRunningCnt records the count of running DDL. - metricExecDDLRunningCnt prometheus.Gauge - // metricExecBatchHis records the executed DML batch size. - // this should be only useful for the MySQL Sink, and Kafka Sink with batched protocol, such as open-protocol. - metricExecBatchHis prometheus.Observer - // metricExecBatchBytesHis records the executed batch write bytes. - metricExecBatchBytesHis prometheus.Observer - // metricTotalWriteBytesCnt records the executed DML event size. - metricTotalWriteBytesCnt prometheus.Counter - - // metricExecErrCntForDDL records the error count of the Sink for DDL. - metricExecErrCntForDDL prometheus.Counter - // metricExecErrCntForDML records the error count of the Sink for DML. - metricExecErrCntForDML prometheus.Counter - // metricExecDMLCnt records the executed DML event count of the Sink. - metricExecDMLCnt prometheus.Counter -} - -// RecordBatchExecution stats batch executors which return (batchRowCount, batchWriteBytes, error). -func (b *Statistics) RecordBatchExecution(executor func() (int, int64, error)) error { - batchSize, batchWriteBytes, err := executor() - if err != nil { - b.metricExecErrCntForDML.Inc() - return err - } - b.metricExecBatchHis.Observe(float64(batchSize)) - b.metricExecBatchBytesHis.Observe(float64(batchWriteBytes)) - b.metricExecDMLCnt.Add(float64(batchSize)) - b.metricTotalWriteBytesCnt.Add(float64(batchWriteBytes)) - return nil -} - -// RecordDDLExecution record the time cost of execute ddl -func (b *Statistics) RecordDDLExecution(executor func() (string, error)) error { - b.metricExecDDLRunningCnt.Inc() - defer b.metricExecDDLRunningCnt.Dec() - - var ( - ddlType string - err error - ) - start := time.Now() - if ddlType, err = executor(); err != nil { - b.metricExecErrCntForDDL.Inc() - return err - } - metricExecDDLCounter := ExecDDLCounter.WithLabelValues( - b.changefeedID.Keyspace(), b.changefeedID.Name(), ddlType) - metricExecDDLCounter.Inc() - b.ddlTypes.Store(ddlType, struct{}{}) - b.metricExecDDLHis.Observe(time.Since(start).Seconds()) - return nil -} - -func (b *Statistics) RecordTotalRowsAffected(actualRowsAffected, expectedRowsAffected int64) { - b.getRowsAffected("actual", "total").Add(float64(actualRowsAffected)) - b.getRowsAffected("expected", "total").Add(float64(expectedRowsAffected)) -} - -func (b *Statistics) RecordRowsAffected(rowsAffected int64, rowType common.RowType) { - b.getRowsAffected("actual", rowType.String()).Add(float64(rowsAffected)) - b.getRowsAffected("expected", rowType.String()).Add(1) - b.RecordTotalRowsAffected(rowsAffected, 1) -} - -func (b *Statistics) getRowsAffected(countType, rowType string) prometheus.Counter { - key := fmt.Sprintf("%s-%s", countType, rowType) - counter, loaded := b.rowsAffectedMap.Load(key) - if !loaded { - keyspace := b.changefeedID.Keyspace() - changefeedID := b.changefeedID.Name() - counter := ExecDMLEventRowsAffectedCounter.WithLabelValues(keyspace, changefeedID, countType, rowType) - b.rowsAffectedMap.Store(key, counter) - return counter - } - return counter.(prometheus.Counter) -} - -// Close release some internal resources. -func (b *Statistics) Close() { - keyspace := b.changefeedID.Keyspace() - changefeedID := b.changefeedID.Name() - ExecDDLHistogram.DeleteLabelValues(keyspace, changefeedID) - ExecBatchHistogram.DeleteLabelValues(keyspace, changefeedID, b.sinkType, b.keyspaceID) - ExecBatchWriteBytesHistogram.DeleteLabelValues(keyspace, changefeedID, b.sinkType) - EventSizeHistogram.DeleteLabelValues(keyspace, changefeedID) - ExecutionErrorCounter.DeleteLabelValues(keyspace, changefeedID, "ddl") - ExecutionErrorCounter.DeleteLabelValues(keyspace, changefeedID, "dml") - b.ddlTypes.Range(func(key, value any) bool { - ddlType := key.(string) - ExecDDLCounter.DeleteLabelValues(keyspace, changefeedID, ddlType) - return true - }) - b.rowsAffectedMap.Range(func(key, value any) bool { - countTypeAndRowType := key.(string) - splitTypes := strings.Split(countTypeAndRowType, "-") - countType, rowType := splitTypes[0], splitTypes[1] - ExecDMLEventRowsAffectedCounter.DeleteLabelValues(keyspace, changefeedID, countType, rowType) - return true - }) - TotalWriteBytesCounter.DeleteLabelValues(keyspace, changefeedID, b.sinkType) - ExecDMLEventCounter.DeleteLabelValues(keyspace, changefeedID) -} diff --git a/pkg/metrics/statistics_test.go b/pkg/metrics/statistics_test.go deleted file mode 100644 index 3afbcb2c6d..0000000000 --- a/pkg/metrics/statistics_test.go +++ /dev/null @@ -1,42 +0,0 @@ -// Copyright 2026 PingCAP, Inc. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// See the License for the specific language governing permissions and -// limitations under the License. - -package metrics - -import ( - "testing" - - "github.com/pingcap/ticdc/pkg/common" - "github.com/prometheus/client_golang/prometheus/testutil" - "github.com/stretchr/testify/require" -) - -func TestExecBatchHistogramKeyspaceIDLabel(t *testing.T) { - ExecBatchHistogram.Reset() - t.Cleanup(ExecBatchHistogram.Reset) - - statistics := NewStatistics( - common.NewChangefeedID4Test("test-keyspace", "batch-row-count-keyspace-id"), - 123, - "sink", - ) - require.NoError(t, statistics.RecordBatchExecution(func() (int, int64, error) { - return 2, 10, nil - })) - - require.Equal(t, 1, testutil.CollectAndCount(ExecBatchHistogram)) - requireMetricHasLabel(t, ExecBatchHistogram, "keyspace_id", "123") - - statistics.Close() - require.Equal(t, 0, testutil.CollectAndCount(ExecBatchHistogram)) -} diff --git a/pkg/sink/kafka/sarama_async_producer.go b/pkg/sink/kafka/sarama_async_producer.go index d3e1781c71..fcf8ec2f4c 100644 --- a/pkg/sink/kafka/sarama_async_producer.go +++ b/pkg/sink/kafka/sarama_async_producer.go @@ -22,6 +22,7 @@ import ( "github.com/pingcap/ticdc/pkg/common" "github.com/pingcap/ticdc/pkg/errors" codecCommon "github.com/pingcap/ticdc/pkg/sink/codec/common" + "github.com/pingcap/ticdc/pkg/statistics" "go.uber.org/atomic" "go.uber.org/zap" ) @@ -30,11 +31,13 @@ type saramaAsyncProducer struct { client sarama.Client producer sarama.AsyncProducer changefeedID common.ChangeFeedID + statistics *statistics.Statistics closed *atomic.Bool } type messageMetadata struct { + rowCount int callback func() logInfo *codecCommon.MessageLogInfo } @@ -100,15 +103,13 @@ func (p *saramaAsyncProducer) AsyncRunCallback( return context.Cause(ctx) case ack := <-p.producer.Successes(): if ack != nil { - switch meta := ack.Metadata.(type) { - case *messageMetadata: - if meta != nil && meta.callback != nil { - meta.callback() - } - default: + meta, ok := ack.Metadata.(*messageMetadata) + if !ok { log.Error("kafka producer received unknown message metadata type", zap.Any("metadata", ack.Metadata)) + continue } + p.handleSuccess(meta) } case err := <-p.producer.Errors(): // We should not wrap a nil pointer if the pointer @@ -119,7 +120,7 @@ func (p *saramaAsyncProducer) AsyncRunCallback( if err == nil { return nil } - return p.handleProducerError(err) + return p.handleFailure(extractRowCount(err.Msg), p.handleProducerError(err)) } } } @@ -140,9 +141,10 @@ func (p *saramaAsyncProducer) AsyncSend( ctx context.Context, topic string, partition int32, message *codecCommon.Message, ) error { if p.closed.Load() { - return errors.ErrKafkaSinkClosed.GenWithStackByArgs() + return p.handleFailure(message.GetRowsCount(), errors.ErrKafkaSinkClosed.GenWithStackByArgs()) } meta := &messageMetadata{ + rowCount: message.GetRowsCount(), callback: message.Callback, logInfo: message.LogInfo, } @@ -155,13 +157,54 @@ func (p *saramaAsyncProducer) AsyncSend( } select { case <-ctx.Done(): - return context.Cause(ctx) + return p.handleFailure(message.GetRowsCount(), context.Cause(ctx)) case p.producer.Input() <- msg: } return nil } +// handleSuccess records a successful delivery and runs the message callback. +// It only touches ticdc-owned types so the statistics behavior can be +// unit-tested without any broker machinery. +func (p *saramaAsyncProducer) handleSuccess(meta *messageMetadata) { + if meta == nil { + return + } + p.recordDMLResult(meta.rowCount, nil) + if meta.callback != nil { + meta.callback() + } +} + +// handleFailure records a failed delivery attempt and returns the error. +func (p *saramaAsyncProducer) handleFailure(rowCount int, err error) error { + p.recordDMLResult(rowCount, err) + return err +} + +func (p *saramaAsyncProducer) recordDMLResult(rowCount int, err error) { + if p.statistics != nil { + p.statistics.RecordDMLResult(rowCount, err) + } +} + +func extractRowCount(msg *sarama.ProducerMessage) int { + meta := extractMessageMetadata(msg) + if meta == nil { + return 0 + } + return meta.rowCount +} + func extractLogInfo(msg *sarama.ProducerMessage) *codecCommon.MessageLogInfo { + meta := extractMessageMetadata(msg) + if meta == nil { + return nil + } + return meta.logInfo +} + +func extractMessageMetadata(msg *sarama.ProducerMessage) *messageMetadata { if msg == nil { return nil } @@ -169,5 +212,5 @@ func extractLogInfo(msg *sarama.ProducerMessage) *codecCommon.MessageLogInfo { if !ok || meta == nil { return nil } - return meta.logInfo + return meta } diff --git a/pkg/sink/kafka/sarama_async_producer_test.go b/pkg/sink/kafka/sarama_async_producer_test.go new file mode 100644 index 0000000000..e1704b3d39 --- /dev/null +++ b/pkg/sink/kafka/sarama_async_producer_test.go @@ -0,0 +1,110 @@ +// Copyright 2026 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package kafka + +import ( + "errors" + "testing" + + "github.com/pingcap/ticdc/pkg/common" + "github.com/pingcap/ticdc/pkg/statistics" + "github.com/prometheus/client_golang/prometheus" + dto "github.com/prometheus/client_model/go" + "github.com/stretchr/testify/require" +) + +// gatherMetric returns the metric of the given name whose labels match +// labelValues (alternating key/value pairs), or nil if it does not exist. +func gatherMetric( + t *testing.T, reg *prometheus.Registry, name string, labelValues ...string, +) *dto.Metric { + t.Helper() + require.Lenf(t, labelValues, len(labelValues)&^1, "labelValues must be key/value pairs") + mfs, err := reg.Gather() + require.NoError(t, err) + for _, mf := range mfs { + if mf.GetName() != name { + continue + } + for _, m := range mf.GetMetric() { + matched := true + for i := 0; i < len(labelValues); i += 2 { + found := false + for _, lp := range m.GetLabel() { + if lp.GetName() == labelValues[i] && lp.GetValue() == labelValues[i+1] { + found = true + break + } + } + if !found { + matched = false + break + } + } + if matched { + return m + } + } + } + return nil +} + +func newTestStatistics(t *testing.T, changefeed string) (*saramaAsyncProducer, *prometheus.Registry) { + t.Helper() + reg := prometheus.NewRegistry() + statistics.InitMetrics(reg) + stat := statistics.New(common.NewChangefeedID4Test("test-keyspace", changefeed), 123) + t.Cleanup(stat.Close) + return &saramaAsyncProducer{statistics: stat}, reg +} + +func TestHandleSuccessRecordsRowsAndRunsCallback(t *testing.T) { + p, reg := newTestStatistics(t, "handle-success") + + callbackCalled := make(chan struct{}) + p.handleSuccess(&messageMetadata{rowCount: 3, callback: func() { close(callbackCalled) }}) + + <-callbackCalled + // The row count is observed into the batch histogram on success. + hist := gatherMetric(t, reg, "ticdc_sink_batch_row_count", + "namespace", "test-keyspace", "changefeed", "handle-success") + require.NotNil(t, hist) + require.Equal(t, uint64(1), hist.GetHistogram().GetSampleCount()) + require.Equal(t, float64(3), hist.GetHistogram().GetSampleSum()) +} + +func TestHandleSuccessNilMetaIsNoop(t *testing.T) { + p, _ := newTestStatistics(t, "handle-success-nil") + require.NotPanics(t, func() { p.handleSuccess(nil) }) +} + +func TestHandleFailureRecordsErrorAndReturnsIt(t *testing.T) { + p, reg := newTestStatistics(t, "handle-failure") + + sentinel := errors.New("broker boom") + require.ErrorIs(t, p.handleFailure(5, sentinel), sentinel) + + // The failed message increments the DML error counter and observes no rows. + errMetric := gatherMetric(t, reg, "ticdc_sink_execution_error", + "namespace", "test-keyspace", "changefeed", "handle-failure", "event_type", "dml") + require.NotNil(t, errMetric) + require.Equal(t, float64(1), errMetric.GetCounter().GetValue()) + // No rows are observed for failed attempts; the histogram series exists + // (created eagerly by New) but has no samples. + hist := gatherMetric(t, reg, "ticdc_sink_batch_row_count", + "namespace", "test-keyspace", "changefeed", "handle-failure") + require.NotNil(t, hist) + require.Equal(t, uint64(0), hist.GetHistogram().GetSampleCount()) +} diff --git a/pkg/sink/kafka/sarama_factory.go b/pkg/sink/kafka/sarama_factory.go index 8f73ca70b5..d1c316b83f 100644 --- a/pkg/sink/kafka/sarama_factory.go +++ b/pkg/sink/kafka/sarama_factory.go @@ -21,6 +21,7 @@ import ( "github.com/pingcap/log" "github.com/pingcap/ticdc/pkg/common" "github.com/pingcap/ticdc/pkg/errors" + "github.com/pingcap/ticdc/pkg/statistics" "github.com/rcrowley/go-metrics" "go.uber.org/atomic" "go.uber.org/zap" @@ -30,13 +31,17 @@ type saramaFactory struct { changefeedID common.ChangeFeedID option *options metricRegistry metrics.Registry + statistics *statistics.Statistics } // NewSaramaFactory constructs a Factory with sarama implementation. +// stat is passed to the DML producer for sink statistics. It may be nil only +// for paths that never send DML messages, such as Verify. func NewSaramaFactory( ctx context.Context, o *options, changefeedID common.ChangeFeedID, + stat *statistics.Statistics, ) (Factory, error) { start := time.Now() config, err := newSaramaConfig(ctx, o) @@ -80,6 +85,7 @@ func NewSaramaFactory( changefeedID: changefeedID, option: o, metricRegistry: metrics.NewRegistry(), + statistics: stat, }, nil } @@ -178,6 +184,7 @@ func (f *saramaFactory) AsyncProducer(ctx context.Context) (AsyncProducer, error client: client, producer: p, changefeedID: f.changefeedID, + statistics: f.statistics, closed: atomic.NewBool(false), }, nil } diff --git a/pkg/sink/mysql/affected_rows.go b/pkg/sink/mysql/affected_rows.go new file mode 100644 index 0000000000..da60c1d4fd --- /dev/null +++ b/pkg/sink/mysql/affected_rows.go @@ -0,0 +1,92 @@ +// Copyright 2026 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package mysql + +import ( + "strings" + "sync" + + "github.com/pingcap/ticdc/pkg/common" + "github.com/pingcap/ticdc/pkg/config/kerneltype" + "github.com/prometheus/client_golang/prometheus" +) + +// execDMLEventRowsAffectedCounter records the affected row counts reported by +// the downstream MySQL, which is a MySQL-sink-specific metric. +var execDMLEventRowsAffectedCounter = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: "ticdc", + Subsystem: "sink", + Name: "dml_event_affected_row_count", + Help: "Total count of affected rows.", + }, []string{getKeyspaceLabel(), "changefeed", "count_type", "row_type"}) + +// InitMetrics registers the MySQL sink metrics. +func InitMetrics(registry *prometheus.Registry) { + registry.MustRegister(execDMLEventRowsAffectedCounter) +} + +// affectedRowsRecorder accumulates affected row statistics for one changefeed. +type affectedRowsRecorder struct { + keyspace string + changefeed string + rowsAffectedMap sync.Map +} + +func newAffectedRowsRecorder(changefeedID common.ChangeFeedID) *affectedRowsRecorder { + return &affectedRowsRecorder{ + keyspace: changefeedID.Keyspace(), + changefeed: changefeedID.Name(), + } +} + +func (r *affectedRowsRecorder) recordTotalRowsAffected(actualRowsAffected, expectedRowsAffected int64) { + r.getRowsAffected("actual", "total").Add(float64(actualRowsAffected)) + r.getRowsAffected("expected", "total").Add(float64(expectedRowsAffected)) +} + +func (r *affectedRowsRecorder) recordRowsAffected(rowsAffected int64, rowType common.RowType) { + r.getRowsAffected("actual", rowType.String()).Add(float64(rowsAffected)) + r.getRowsAffected("expected", rowType.String()).Add(1) + r.recordTotalRowsAffected(rowsAffected, 1) +} + +func (r *affectedRowsRecorder) getRowsAffected(countType, rowType string) prometheus.Counter { + key := countType + "-" + rowType + counter, loaded := r.rowsAffectedMap.Load(key) + if !loaded { + counter := execDMLEventRowsAffectedCounter.WithLabelValues(r.keyspace, r.changefeed, countType, rowType) + r.rowsAffectedMap.Store(key, counter) + return counter + } + return counter.(prometheus.Counter) +} + +// close removes the per-changefeed metric series. +func (r *affectedRowsRecorder) close() { + r.rowsAffectedMap.Range(func(key, value any) bool { + countTypeAndRowType := key.(string) + splitTypes := strings.Split(countTypeAndRowType, "-") + execDMLEventRowsAffectedCounter.DeleteLabelValues(r.keyspace, r.changefeed, splitTypes[0], splitTypes[1]) + return true + }) +} + +func getKeyspaceLabel() string { + if kerneltype.IsNextGen() { + return "keyspace_name" + } + return "namespace" +} diff --git a/pkg/sink/mysql/mysql_writer.go b/pkg/sink/mysql/mysql_writer.go index 4c38fadaab..b92bbd76ed 100644 --- a/pkg/sink/mysql/mysql_writer.go +++ b/pkg/sink/mysql/mysql_writer.go @@ -25,7 +25,7 @@ import ( "github.com/pingcap/ticdc/pkg/common" commonEvent "github.com/pingcap/ticdc/pkg/common/event" "github.com/pingcap/ticdc/pkg/errors" - "github.com/pingcap/ticdc/pkg/metrics" + "github.com/pingcap/ticdc/pkg/statistics" "go.uber.org/zap" ) @@ -64,7 +64,10 @@ type Writer struct { // implement stmtCache to improve performance, especially when the downstream is TiDB stmtCache *lru.Cache - statistics *metrics.Statistics + statistics *statistics.Statistics + + // affectedRows records the affected row counts reported by the downstream. + affectedRows *affectedRowsRecorder // activeActiveSyncStatsCollector accumulates conflict statistics from TiDB session // variable @@tidb_cdc_active_active_sync_stats. It is shared across all DML writers @@ -93,7 +96,7 @@ func NewWriter( db *sql.DB, cfg *Config, changefeedID common.ChangeFeedID, - statistics *metrics.Statistics, + statistics *statistics.Statistics, activeActiveSyncStatsCollector *ActiveActiveSyncStatsCollector, ) *Writer { writerCtx, cancel := context.WithCancel(ctx) @@ -109,6 +112,7 @@ func NewWriter( ddlTsTableInit: false, stmtCache: cfg.stmtCache, statistics: statistics, + affectedRows: newAffectedRowsRecorder(changefeedID), maxDDLTsBatch: cfg.MaxTxnRow, dmlSession: *NewDMLSession(dmlConnIdleTimeout), isInErrorCausedSafeMode: false, @@ -239,9 +243,7 @@ func (w *Writer) Flush(events []*commonEvent.DMLEvent) error { } else { w.tryDryRunBlock() - err = w.statistics.RecordBatchExecution(func() (int, int64, error) { - return dmls.rowCount, dmls.approximateSize, nil - }) + w.statistics.RecordDMLResult(dmls.rowCount, nil) } if err != nil { @@ -293,6 +295,9 @@ func (w *Writer) tryDryRunBlock() { } func (w *Writer) Close() { + if w.affectedRows != nil { + w.affectedRows.close() + } if w.stmtCache != nil { w.stmtCache.Purge() } diff --git a/pkg/sink/mysql/mysql_writer_ddl_ts_test.go b/pkg/sink/mysql/mysql_writer_ddl_ts_test.go index f5ec56f7fe..2f63e3bbcc 100644 --- a/pkg/sink/mysql/mysql_writer_ddl_ts_test.go +++ b/pkg/sink/mysql/mysql_writer_ddl_ts_test.go @@ -25,7 +25,7 @@ import ( "github.com/pingcap/ticdc/heartbeatpb" "github.com/pingcap/ticdc/pkg/common" commonEvent "github.com/pingcap/ticdc/pkg/common/event" - "github.com/pingcap/ticdc/pkg/metrics" + "github.com/pingcap/ticdc/pkg/statistics" "github.com/stretchr/testify/require" ) @@ -41,7 +41,7 @@ func newTestMysqlWriterForDDLTs(t *testing.T) (*Writer, *sql.DB, sqlmock.Sqlmock cfg.EnableDDLTs = true cfg.IsTiDB = false // Default to non-TiDB changefeedID := common.NewChangefeedID4Test("test", "test") - statistics := metrics.NewStatistics(changefeedID, common.DefaultKeyspaceID, "mysqlSink") + statistics := statistics.New(changefeedID, common.DefaultKeyspaceID) writer := NewWriter(ctx, 0, db, cfg, changefeedID, statistics, nil) t.Cleanup(writer.Close) @@ -63,7 +63,7 @@ func newTestMysqlWriterForDDLTsTiDB(t *testing.T) (*Writer, *sql.DB, sqlmock.Sql cfg.EnableDDLTs = true cfg.IsTiDB = true // TiDB downstream changefeedID := common.NewChangefeedID4Test("test", "test") - statistics := metrics.NewStatistics(changefeedID, common.DefaultKeyspaceID, "mysqlSink") + statistics := statistics.New(changefeedID, common.DefaultKeyspaceID) writer := NewWriter(ctx, 0, db, cfg, changefeedID, statistics, nil) t.Cleanup(writer.Close) diff --git a/pkg/sink/mysql/mysql_writer_dml_exec.go b/pkg/sink/mysql/mysql_writer_dml_exec.go index 4d0645bae7..fa60705010 100644 --- a/pkg/sink/mysql/mysql_writer_dml_exec.go +++ b/pkg/sink/mysql/mysql_writer_dml_exec.go @@ -47,7 +47,7 @@ func (w *Writer) execDMLWithMaxRetries(dmls *preparedDMLs) error { writeTimeout, _ := time.ParseDuration(w.cfg.WriteTimeout) writeTimeout += networkDriftDuration - tryExec := func() (int, int64, error) { + tryExec := func() error { start := time.Now() defer func() { if time.Since(start) > w.cfg.SlowQuery { @@ -87,9 +87,9 @@ func (w *Writer) execDMLWithMaxRetries(dmls *preparedDMLs) error { return nil }) if err != nil { - return 0, 0, err + return err } - return dmls.rowCount, dmls.approximateSize, nil + return nil } return retry.Do(w.ctx, func() error { failpoint.Inject("MySQLSinkTxnRandomError", func() { @@ -114,7 +114,8 @@ func (w *Writer) execDMLWithMaxRetries(dmls *preparedDMLs) error { failpoint.Return(err) }) - err := w.statistics.RecordBatchExecution(tryExec) + err := tryExec() + w.statistics.RecordDMLResult(dmls.rowCount, err) if err != nil { return errors.Trace(w.logDMLTxnErr(err, time.Now(), w.ChangefeedID.String(), dmls)) } @@ -172,7 +173,7 @@ func (w *Writer) sequenceExecute( if rowsAffected, err := res.RowsAffected(); err != nil { log.Warn("get rows affected rows failed", zap.Error(err)) } else { - w.statistics.RecordRowsAffected(rowsAffected, dmls.rowTypes[i]) + w.affectedRows.recordRowsAffected(rowsAffected, dmls.rowTypes[i]) } cancelFunc() } @@ -214,7 +215,7 @@ func (w *Writer) multiStmtExecute( if rowsAffected, err := res.RowsAffected(); err != nil { log.Warn("get rows affected rows failed", zap.Error(err)) } else { - w.statistics.RecordTotalRowsAffected(rowsAffected, int64(len(dmls.sqls))) + w.affectedRows.recordTotalRowsAffected(rowsAffected, int64(len(dmls.sqls))) } return nil } diff --git a/pkg/sink/mysql/mysql_writer_test.go b/pkg/sink/mysql/mysql_writer_test.go index 00c37f9f85..d931c5cdff 100644 --- a/pkg/sink/mysql/mysql_writer_test.go +++ b/pkg/sink/mysql/mysql_writer_test.go @@ -31,8 +31,8 @@ import ( "github.com/pingcap/ticdc/pkg/config" "github.com/pingcap/ticdc/pkg/config/kerneltype" cerror "github.com/pingcap/ticdc/pkg/errors" - "github.com/pingcap/ticdc/pkg/metrics" "github.com/pingcap/ticdc/pkg/routing" + "github.com/pingcap/ticdc/pkg/statistics" "github.com/pingcap/tidb/br/pkg/version" ticonfig "github.com/pingcap/tidb/pkg/config" "github.com/pingcap/tidb/pkg/dxf/framework/handle" @@ -52,7 +52,7 @@ func newTestMysqlWriter(t *testing.T) (*Writer, *sql.DB, sqlmock.Sqlmock) { cfg.BatchDMLEnable = true cfg.EnableDDLTs = defaultEnableDDLTs changefeedID := common.NewChangefeedID4Test("test", "test") - statistics := metrics.NewStatistics(changefeedID, common.DefaultKeyspaceID, "mysqlSink") + statistics := statistics.New(changefeedID, common.DefaultKeyspaceID) writer := NewWriter(ctx, 0, db, cfg, changefeedID, statistics, nil) t.Cleanup(writer.Close) // assign a no-op stmt cache to bypass actual DB operations in unit tests @@ -75,7 +75,7 @@ func newTestMysqlWriterForTiDB(t *testing.T) (*Writer, *sql.DB, sqlmock.Sqlmock) cfg.ServerInfo = version.ParseServerInfo(defaultRunningAddIndexNewSQLVersion) changefeedID := common.NewChangefeedID4Test("test", "test") - statistics := metrics.NewStatistics(changefeedID, common.DefaultKeyspaceID, "mysqlSink") + statistics := statistics.New(changefeedID, common.DefaultKeyspaceID) writer := NewWriter(ctx, 0, db, cfg, changefeedID, statistics, nil) t.Cleanup(writer.Close) diff --git a/pkg/statistics/metrics.go b/pkg/statistics/metrics.go new file mode 100644 index 0000000000..ab1dd8bad7 --- /dev/null +++ b/pkg/statistics/metrics.go @@ -0,0 +1,89 @@ +// Copyright 2026 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package statistics + +import ( + "github.com/pingcap/ticdc/pkg/config/kerneltype" + "github.com/prometheus/client_golang/prometheus" +) + +var ( + execDDLHistogram = prometheus.NewHistogramVec( + prometheus.HistogramOpts{ + Namespace: "ticdc", + Subsystem: "ddl", + Name: "exec_duration", + Help: "Bucketed histogram of processing time (s) of a ddl.", + Buckets: prometheus.ExponentialBuckets(0.01, 2, 18), + }, []string{getKeyspaceLabel(), "changefeed"}) + + execDDLRunningGauge = prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Namespace: "ticdc", + Subsystem: "ddl", + Name: "exec_running", + Help: "Total count of running ddl.", + }, []string{getKeyspaceLabel(), "changefeed"}) + + execDDLCounter = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: "ticdc", + Subsystem: "ddl", + Name: "execution", + Help: "Total execution count of different DDL types.", + }, []string{getKeyspaceLabel(), "changefeed", "ddl_type"}) + + execBatchHistogram = prometheus.NewHistogramVec( + prometheus.HistogramOpts{ + Namespace: "ticdc", + Subsystem: "sink", + Name: "batch_row_count", + Help: "Row count number for a given batch.", + Buckets: prometheus.ExponentialBuckets(1, 2, 18), + }, []string{getKeyspaceLabel(), "changefeed", "keyspace_id"}) + + totalWriteBytesCounter = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: "ticdc", + Subsystem: "sink", + Name: "write_bytes_total", + Help: "Total approximate raw bytes of DML events successfully written to downstream.", + }, []string{getKeyspaceLabel(), "changefeed"}) + + executionErrorCounter = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: "ticdc", + Subsystem: "sink", + Name: "execution_error", + Help: "Total count of execution errors.", + }, []string{getKeyspaceLabel(), "changefeed", "event_type"}) +) + +// InitMetrics registers the metrics maintained by Statistics. +func InitMetrics(registry *prometheus.Registry) { + registry.MustRegister(execDDLHistogram) + registry.MustRegister(execDDLRunningGauge) + registry.MustRegister(execDDLCounter) + registry.MustRegister(execBatchHistogram) + registry.MustRegister(totalWriteBytesCounter) + registry.MustRegister(executionErrorCounter) +} + +func getKeyspaceLabel() string { + if kerneltype.IsNextGen() { + return "keyspace_name" + } + return "namespace" +} diff --git a/pkg/statistics/statistics.go b/pkg/statistics/statistics.go new file mode 100644 index 0000000000..81b73117fb --- /dev/null +++ b/pkg/statistics/statistics.go @@ -0,0 +1,128 @@ +// Copyright 2020 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package statistics + +import ( + "strconv" + "sync" + "time" + + "github.com/pingcap/ticdc/pkg/common" + commonEvent "github.com/pingcap/ticdc/pkg/common/event" + "github.com/prometheus/client_golang/prometheus" +) + +// New creates a Statistics. +func New(changefeed common.ChangeFeedID, keyspaceID uint32) *Statistics { + statistics := &Statistics{ + changefeedID: changefeed, + keyspaceID: strconv.FormatUint(uint64(keyspaceID), 10), + ddlTypes: sync.Map{}, + } + + keyspace := changefeed.Keyspace() + changefeedID := changefeed.Name() + statistics.metricExecDDLHis = execDDLHistogram.WithLabelValues(keyspace, changefeedID) + statistics.metricExecDDLRunningCnt = execDDLRunningGauge.WithLabelValues(keyspace, changefeedID) + statistics.metricExecBatchHis = execBatchHistogram.WithLabelValues(keyspace, changefeedID, statistics.keyspaceID) + statistics.metricTotalWriteBytesCnt = totalWriteBytesCounter.WithLabelValues(keyspace, changefeedID) + statistics.metricExecErrCntForDDL = executionErrorCounter.WithLabelValues(keyspace, changefeedID, "ddl") + statistics.metricExecErrCntForDML = executionErrorCounter.WithLabelValues(keyspace, changefeedID, "dml") + + return statistics +} + +// Statistics maintains some status and metrics of the Sink +// Note: All methods of Statistics should be thread-safe. +type Statistics struct { + changefeedID common.ChangeFeedID + keyspaceID string + ddlTypes sync.Map + + // metricExecDDLHis records each DDL execution time duration. + metricExecDDLHis prometheus.Observer + // metricExecDDLRunningCnt records the count of running DDL. + metricExecDDLRunningCnt prometheus.Gauge + // metricExecBatchHis records the executed DML batch size. + // this should be only useful for the MySQL Sink, and Kafka Sink with batched protocol, such as open-protocol. + metricExecBatchHis prometheus.Observer + // metricTotalWriteBytesCnt records the executed DML event size. + metricTotalWriteBytesCnt prometheus.Counter + + // metricExecErrCntForDDL records the error count of the Sink for DDL. + metricExecErrCntForDDL prometheus.Counter + // metricExecErrCntForDML records the error count of the Sink for DML. + metricExecErrCntForDML prometheus.Counter +} + +// RecordDMLResult records the result of one downstream DML execution attempt. +// Successful attempts contribute their row count; failed attempts increment the +// DML execution error counter. DML event bytes are tracked separately because +// they are recorded only after the whole transaction is flushed. +func (b *Statistics) RecordDMLResult(rowCount int, err error) { + if err != nil { + b.metricExecErrCntForDML.Inc() + return + } + b.metricExecBatchHis.Observe(float64(rowCount)) +} + +// TrackDMLEvent records the approximate size reported by a DML event after the +// whole transaction has been flushed to downstream. The size is snapshotted +// here so the callback does not retain the event. +func (b *Statistics) TrackDMLEvent(event *commonEvent.DMLEvent) { + writeBytes := event.GetSize() + event.AddPostFlushFunc(func() { + b.metricTotalWriteBytesCnt.Add(float64(writeBytes)) + }) +} + +// RecordDDLExecution record the time cost of execute ddl +func (b *Statistics) RecordDDLExecution(executor func() (string, error)) error { + b.metricExecDDLRunningCnt.Inc() + defer b.metricExecDDLRunningCnt.Dec() + + var ( + ddlType string + err error + ) + start := time.Now() + if ddlType, err = executor(); err != nil { + b.metricExecErrCntForDDL.Inc() + return err + } + metricExecDDLCounter := execDDLCounter.WithLabelValues( + b.changefeedID.Keyspace(), b.changefeedID.Name(), ddlType) + metricExecDDLCounter.Inc() + b.ddlTypes.Store(ddlType, struct{}{}) + b.metricExecDDLHis.Observe(time.Since(start).Seconds()) + return nil +} + +// Close release some internal resources. +func (b *Statistics) Close() { + keyspace := b.changefeedID.Keyspace() + changefeedID := b.changefeedID.Name() + execDDLHistogram.DeleteLabelValues(keyspace, changefeedID) + execDDLRunningGauge.DeleteLabelValues(keyspace, changefeedID) + execBatchHistogram.DeleteLabelValues(keyspace, changefeedID, b.keyspaceID) + executionErrorCounter.DeleteLabelValues(keyspace, changefeedID, "ddl") + executionErrorCounter.DeleteLabelValues(keyspace, changefeedID, "dml") + b.ddlTypes.Range(func(key, value any) bool { + ddlType := key.(string) + execDDLCounter.DeleteLabelValues(keyspace, changefeedID, ddlType) + return true + }) + totalWriteBytesCounter.DeleteLabelValues(keyspace, changefeedID) +} diff --git a/pkg/statistics/statistics_test.go b/pkg/statistics/statistics_test.go new file mode 100644 index 0000000000..09ab841d83 --- /dev/null +++ b/pkg/statistics/statistics_test.go @@ -0,0 +1,82 @@ +// Copyright 2026 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package statistics + +import ( + "errors" + "testing" + + "github.com/pingcap/ticdc/pkg/common" + commonEvent "github.com/pingcap/ticdc/pkg/common/event" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/testutil" + dto "github.com/prometheus/client_model/go" + "github.com/stretchr/testify/require" +) + +func newTestEvent(size int64) *commonEvent.DMLEvent { + event := commonEvent.NewDMLEvent(common.NewDispatcherID(), 1, 1, 2, nil) + event.ApproximateSize = size + return event +} + +func TestRecordDMLResult(t *testing.T) { + stat := New(common.NewChangefeedID4Test("test-keyspace", "record-dml-result"), 123) + defer stat.Close() + + // failed attempts increment the DML error counter and do not observe rows. + stat.RecordDMLResult(10, errors.New("boom")) + require.Equal(t, float64(1), testutil.ToFloat64(stat.metricExecErrCntForDML)) + + // successful attempts observe the row count into the batch histogram. + stat.RecordDMLResult(2, nil) + var m dto.Metric + require.NoError(t, stat.metricExecBatchHis.(prometheus.Metric).Write(&m)) + require.Equal(t, uint64(1), m.Histogram.GetSampleCount()) + require.Equal(t, float64(2), m.Histogram.GetSampleSum()) +} + +func TestTrackDMLEventCountsBytesOnPostFlush(t *testing.T) { + stat := New(common.NewChangefeedID4Test("test-keyspace", "track-dml-event"), 123) + defer stat.Close() + + event := newTestEvent(1024) + stat.TrackDMLEvent(event) + + // Bytes are only counted after the event is flushed. + require.Zero(t, testutil.ToFloat64(stat.metricTotalWriteBytesCnt)) + event.PostFlush() + require.Equal(t, float64(1024), testutil.ToFloat64(stat.metricTotalWriteBytesCnt)) +} + +func TestCloseDeletesMetricSeries(t *testing.T) { + stat := New(common.NewChangefeedID4Test("test-keyspace", "close-deletes"), 123) + + stat.RecordDMLResult(1, nil) + stat.RecordDMLResult(1, errors.New("boom")) + event := newTestEvent(100) + stat.TrackDMLEvent(event) + event.PostFlush() + + require.Equal(t, 1, testutil.CollectAndCount(execBatchHistogram)) + // New() eagerly creates the ddl series, RecordDMLResult adds the dml one. + require.Equal(t, 2, testutil.CollectAndCount(executionErrorCounter)) + require.Equal(t, 1, testutil.CollectAndCount(totalWriteBytesCounter)) + + stat.Close() + require.Equal(t, 0, testutil.CollectAndCount(execBatchHistogram)) + require.Equal(t, 0, testutil.CollectAndCount(executionErrorCounter)) + require.Equal(t, 0, testutil.CollectAndCount(totalWriteBytesCounter)) +} diff --git a/server/metrics.go b/server/metrics.go index 989a2beda0..88f55d514b 100644 --- a/server/metrics.go +++ b/server/metrics.go @@ -16,6 +16,7 @@ package server import ( "github.com/pingcap/ticdc/pkg/common/event" "github.com/pingcap/ticdc/pkg/metrics" + mysql "github.com/pingcap/ticdc/pkg/sink/mysql" "github.com/prometheus/client_golang/prometheus" ) @@ -24,4 +25,5 @@ var registry = prometheus.NewRegistry() func init() { metrics.InitMetrics(registry) event.InitEventMetrics(registry) + mysql.InitMetrics(registry) }