pg/mysql cdc: report fetched bytes metric with interval (#3403)

@ilidemi noticed 10% of cpu usage in pg/mysql was attributed to metrics

reuse pattern used by qrep where we add to atomic counters & regularly empty those counters

mongo cdc already does this

---------

Co-authored-by: Ilia Demianenko <[email protected]>

Author: serprex

flow/connectors/mysql/cdc.go

@@ -10,6 +10,7 @@ import (
 	"slices"
 	"strconv"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	"github.com/go-mysql-org/go-mysql/mysql"
@@ -326,19 +327,31 @@ func (c *MySqlConnector) PullRecords(
 	var updatedOffset string
 	var inTx bool
 	var recordCount uint32
-	var bytesRead int
+
 	// set when a tx is preventing us from respecting the timeout, immediately exit after we see inTx false
 	var overtime bool
+	var fetchedBytes, totalFetchedBytes, allFetchedBytes atomic.Int64
 	defer func() {
 		if recordCount == 0 {
 			req.RecordStream.SignalAsEmpty()
 		}
 		c.logger.Info("[mysql] PullRecords finished streaming",
 			slog.Uint64("records", uint64(recordCount)),
-			slog.Int("bytes", bytesRead),
+			slog.Int64("bytes", totalFetchedBytes.Load()),
 			slog.Int("channelLen", req.RecordStream.ChannelLen()))
 	}()
 
+	defer func() {
+		otelManager.Metrics.FetchedBytesCounter.Add(ctx, fetchedBytes.Swap(0))
+		otelManager.Metrics.AllFetchedBytesCounter.Add(ctx, allFetchedBytes.Swap(0))
+	}()
+	shutdown := shared.Interval(ctx, time.Minute, func() {
+		otelManager.Metrics.FetchedBytesCounter.Add(ctx, fetchedBytes.Swap(0))
+		otelManager.Metrics.AllFetchedBytesCounter.Add(ctx, allFetchedBytes.Swap(0))
+		c.logger.Info("pulling records", slog.Int64("bytes", totalFetchedBytes.Load()))
+	})
+	defer shutdown()
+
 	timeoutCtx, cancelTimeout := context.WithTimeout(ctx, time.Hour)
 	//nolint:gocritic // cancelTimeout is rebound, do not defer cancelTimeout()
 	defer func() {
@@ -358,7 +371,7 @@ func (c *MySqlConnector) PullRecords(
 		if recordCount%50000 == 0 {
 			c.logger.Info("[mysql] PullRecords streaming",
 				slog.Uint64("records", uint64(recordCount)),
-				slog.Int("bytes", bytesRead),
+				slog.Int64("bytes", totalFetchedBytes.Load()),
 				slog.Int("channelLen", req.RecordStream.ChannelLen()),
 				slog.Bool("inTx", inTx),
 				slog.Bool("overtime", overtime))
@@ -397,7 +410,7 @@ func (c *MySqlConnector) PullRecords(
 				} else if inTx {
 					c.logger.Info("[mysql] timeout reached, but still in transaction, waiting for inTx false",
 						slog.Uint64("records", uint64(recordCount)),
-						slog.Int("bytes", bytesRead),
+						slog.Int64("bytes", totalFetchedBytes.Load()),
 						slog.Int("channelLen", req.RecordStream.ChannelLen()))
 					// reset timeoutCtx to a low value and wait for inTx to become false
 					cancelTimeout()
@@ -415,7 +428,7 @@ func (c *MySqlConnector) PullRecords(
 			return err
 		}
 
-		otelManager.Metrics.AllFetchedBytesCounter.Add(ctx, int64(len(event.RawData)))
+		allFetchedBytes.Add(int64(len(event.RawData)))
 
 		switch ev := event.Event.(type) {
 		case *replication.GTIDEvent:
@@ -473,7 +486,8 @@ func (c *MySqlConnector) PullRecords(
 			schema := req.TableNameSchemaMapping[destinationTableName]
 			if schema != nil {
 				otelManager.Metrics.FetchedBytesCounter.Add(ctx, int64(len(event.RawData)))
-				bytesRead += len(event.RawData)
+				fetchedBytes.Add(int64(len(event.RawData)))
+				totalFetchedBytes.Add(int64(len(event.RawData)))
 				inTx = true
 				enumMap := ev.Table.EnumStrValueMap()
 				setMap := ev.Table.SetStrValueMap()

flow/connectors/postgres/cdc.go

@@ -440,16 +440,23 @@ func PullCdcRecords[Items model.Items](
 
 	logger.Info("pulling records start")
 
+	var fetchedBytes, totalFetchedBytes, allFetchedBytes atomic.Int64
+	defer func() {
+		p.otelManager.Metrics.FetchedBytesCounter.Add(ctx, fetchedBytes.Swap(0))
+		p.otelManager.Metrics.AllFetchedBytesCounter.Add(ctx, allFetchedBytes.Swap(0))
+	}()
 	shutdown := shared.Interval(ctx, time.Minute, func() {
-		logger.Info("pulling records", slog.Int("records", cdcRecordsStorage.Len()))
+		p.otelManager.Metrics.FetchedBytesCounter.Add(ctx, fetchedBytes.Swap(0))
+		p.otelManager.Metrics.AllFetchedBytesCounter.Add(ctx, allFetchedBytes.Swap(0))
+		logger.Info("pulling records", slog.Int("records", cdcRecordsStorage.Len()),
+			slog.Int64("bytes", totalFetchedBytes.Load()))
 	})
 	defer shutdown()
 
 	nextStandbyMessageDeadline := time.Now().Add(req.IdleTimeout)
 
 	pkmRequiresResponse := false
 	waitingForCommit := false
-	fetchedBytes := 0
 
 	addRecordWithKey := func(key model.TableWithPkey, rec model.Record[Items]) error {
 		if err := cdcRecordsStorage.Set(logger, key, rec); err != nil {
@@ -467,7 +474,7 @@ func PullCdcRecords[Items model.Items](
 		if cdcRecordsStorage.Len()%50000 == 0 {
 			logger.Info("pulling records",
 				slog.Int("records", cdcRecordsStorage.Len()),
-				slog.Int("bytes", fetchedBytes),
+				slog.Int64("bytes", totalFetchedBytes.Load()),
 				slog.Int("channelLen", records.ChannelLen()),
 				slog.Bool("waitingForCommit", waitingForCommit))
 		}
@@ -497,7 +504,7 @@ func PullCdcRecords[Items model.Items](
 			if time.Since(standByLastLogged) > 10*time.Second {
 				logger.Info("Sent Standby status message",
 					slog.Int("records", cdcRecordsStorage.Len()),
-					slog.Int("bytes", fetchedBytes),
+					slog.Int64("bytes", totalFetchedBytes.Load()),
 					slog.Int("channelLen", records.ChannelLen()),
 					slog.Bool("waitingForCommit", waitingForCommit))
 				standByLastLogged = time.Now()
@@ -508,15 +515,15 @@ func PullCdcRecords[Items model.Items](
 			if cdcRecordsStorage.Len() >= int(req.MaxBatchSize) {
 				logger.Info("batch filled, returning currently accumulated records",
 					slog.Int("records", cdcRecordsStorage.Len()),
-					slog.Int("bytes", fetchedBytes),
+					slog.Int64("bytes", totalFetchedBytes.Load()),
 					slog.Int("channelLen", records.ChannelLen()))
 				return nil
 			}
 
 			if waitingForCommit {
 				logger.Info("commit received, returning currently accumulated records",
 					slog.Int("records", cdcRecordsStorage.Len()),
-					slog.Int("bytes", fetchedBytes),
+					slog.Int64("bytes", totalFetchedBytes.Load()),
 					slog.Int("channelLen", records.ChannelLen()))
 				return nil
 			}
@@ -530,13 +537,13 @@ func PullCdcRecords[Items model.Items](
 				if p.commitLock == nil {
 					logger.Info("no commit lock, returning currently accumulated records",
 						slog.Int("records", cdcRecordsStorage.Len()),
-						slog.Int("bytes", fetchedBytes),
+						slog.Int64("bytes", totalFetchedBytes.Load()),
 						slog.Int("channelLen", records.ChannelLen()))
 					return nil
 				} else {
 					logger.Info("commit lock, waiting for commit to return records",
 						slog.Int("records", cdcRecordsStorage.Len()),
-						slog.Int("bytes", fetchedBytes),
+						slog.Int64("bytes", totalFetchedBytes.Load()),
 						slog.Int("channelLen", records.ChannelLen()))
 					waitingForCommit = true
 				}
@@ -568,7 +575,7 @@ func PullCdcRecords[Items model.Items](
 			if pgconn.Timeout(err) {
 				logger.Info("Stand-by deadline reached, returning currently accumulated records",
 					slog.Int("records", cdcRecordsStorage.Len()),
-					slog.Int("bytes", fetchedBytes),
+					slog.Int64("bytes", totalFetchedBytes.Load()),
 					slog.Int("channelLen", records.ChannelLen()))
 				return nil
 			} else {
@@ -580,7 +587,7 @@ func PullCdcRecords[Items model.Items](
 		case *pgproto3.ErrorResponse:
 			return shared.LogError(logger, exceptions.NewPostgresWalError(errors.New("received error response"), msg))
 		case *pgproto3.CopyData:
-			p.otelManager.Metrics.AllFetchedBytesCounter.Add(ctx, int64(len(msg.Data)))
+			allFetchedBytes.Add(int64(len(msg.Data)))
 			switch msg.Data[0] {
 			case pglogrepl.PrimaryKeepaliveMessageByteID:
 				pkm, err := pglogrepl.ParsePrimaryKeepaliveMessage(msg.Data[1:])
@@ -615,8 +622,8 @@ func PullCdcRecords[Items model.Items](
 				}
 
 				if rec != nil {
-					p.otelManager.Metrics.FetchedBytesCounter.Add(ctx, int64(len(msg.Data)))
-					fetchedBytes += len(msg.Data)
+					fetchedBytes.Add(int64(len(msg.Data)))
+					totalFetchedBytes.Add(int64(len(msg.Data)))
 					tableName := rec.GetDestinationTableName()
 					switch r := rec.(type) {
 					case *model.UpdateRecord[Items]: