sink: make sink statistics consistent and flush-aware

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()
 		}
 	}

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]()

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)
 }
 

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,55 +218,22 @@ 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
 	}
 
 	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)
 }

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)

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
 	}