kv-client(cdc): add more logs to help debug slow regions (#9981)

close #9980

cdc/kv/client.go

@@ -19,6 +19,7 @@ import (
 	"io"
 	"math/rand"
 	"strconv"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -42,6 +43,7 @@ import (
 	"github.com/pingcap/tiflow/pkg/version"
 	"github.com/prometheus/client_golang/prometheus"
 	tidbkv "github.com/tikv/client-go/v2/kv"
+	"github.com/tikv/client-go/v2/oracle"
 	"github.com/tikv/client-go/v2/tikv"
 	pd "github.com/tikv/pd/client"
 	"go.uber.org/zap"
@@ -423,13 +425,11 @@ func (s *eventFeedSession) eventFeed(ctx context.Context) error {
 
 	g, ctx := errgroup.WithContext(ctx)
 
-	g.Go(func() error {
-		return s.dispatchRequest(ctx)
-	})
+	g.Go(func() error { return s.dispatchRequest(ctx) })
 
-	g.Go(func() error {
-		return s.requestRegionToStore(ctx, g)
-	})
+	g.Go(func() error { return s.requestRegionToStore(ctx, g) })
+
+	g.Go(func() error { return s.logSlowRegions(ctx) })
 
 	g.Go(func() error {
 		for {
@@ -689,13 +689,13 @@ func (s *eventFeedSession) requestRegionToStore(
 		state := newRegionFeedState(sri, requestID)
 		pendingRegions.setByRequestID(requestID, state)
 
-		log.Debug("start new request",
+		log.Info("start new request",
 			zap.String("namespace", s.changefeed.Namespace),
 			zap.String("changefeed", s.changefeed.ID),
 			zap.Int64("tableID", s.tableID),
 			zap.String("tableName", s.tableName),
-			zap.String("addr", storeAddr),
-			zap.Any("request", req))
+			zap.Uint64("regionID", sri.verID.GetID()),
+			zap.String("addr", storeAddr))
 
 		err = stream.client.Send(req)
 
@@ -889,6 +889,13 @@ func (s *eventFeedSession) handleError(ctx context.Context, errInfo regionErrorI
 	switch eerr := errors.Cause(err).(type) {
 	case *eventError:
 		innerErr := eerr.err
+		log.Info("cdc region error",
+			zap.String("namespace", s.changefeed.Namespace),
+			zap.String("changefeed", s.changefeed.ID),
+			zap.Int64("tableID", s.tableID),
+			zap.String("tableName", s.tableName),
+			zap.Stringer("error", innerErr))
+
 		if notLeader := innerErr.GetNotLeader(); notLeader != nil {
 			metricFeedNotLeaderCounter.Inc()
 			s.client.regionCache.UpdateLeader(errInfo.verID, notLeader.GetLeader(), errInfo.rpcCtx.AccessIdx)
@@ -1216,6 +1223,17 @@ func (s *eventFeedSession) sendRegionChangeEvents(
 			continue
 		}
 
+		switch x := event.Event.(type) {
+		case *cdcpb.Event_Error:
+			log.Info("event feed receives a region error",
+				zap.String("namespace", s.changefeed.Namespace),
+				zap.String("changefeed", s.changefeed.ID),
+				zap.Int64("tableID", s.tableID),
+				zap.String("tableName", s.tableName),
+				zap.Uint64("regionID", event.RegionId),
+				zap.Any("error", x.Error))
+		}
+
 		slot := worker.inputCalcSlot(event.RegionId)
 		statefulEvents[slot] = append(statefulEvents[slot], &regionStatefulEvent{
 			changeEvent: event,
@@ -1308,6 +1326,51 @@ func (s *eventFeedSession) getStreamCancel(storeAddr string) (cancel context.Can
 	return
 }
 
+func (s *eventFeedSession) logSlowRegions(ctx context.Context) error {
+	ticker := time.NewTicker(10 * time.Second)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-ticker.C:
+		}
+
+		currTime := s.client.pdClock.CurrentTime()
+		attr := s.rangeLock.CollectLockedRangeAttrs(nil)
+		if attr.SlowestRegion.Initialized {
+			ckptTime := oracle.GetTimeFromTS(attr.SlowestRegion.CheckpointTs)
+			if currTime.Sub(ckptTime) > 2*resolveLockMinInterval {
+				log.Info("event feed finds a slow region",
+					zap.String("namespace", s.changefeed.Namespace),
+					zap.String("changefeed", s.changefeed.ID),
+					zap.Int64("tableID", s.tableID),
+					zap.String("tableName", s.tableName),
+					zap.Any("slowRegion", attr.SlowestRegion))
+			}
+		} else if currTime.Sub(attr.SlowestRegion.Created) > 10*time.Minute {
+			log.Info("event feed initializes a region too slow",
+				zap.String("namespace", s.changefeed.Namespace),
+				zap.String("changefeed", s.changefeed.ID),
+				zap.Int64("tableID", s.tableID),
+				zap.String("tableName", s.tableName),
+				zap.Any("slowRegion", attr.SlowestRegion))
+		}
+		if len(attr.Holes) > 0 {
+			holes := make([]string, 0, len(attr.Holes))
+			for _, hole := range attr.Holes {
+				holes = append(holes, fmt.Sprintf("[%s,%s)", hole.StartKey, hole.EndKey))
+			}
+			log.Info("event feed holes exist",
+				zap.String("namespace", s.changefeed.Namespace),
+				zap.String("changefeed", s.changefeed.ID),
+				zap.Int64("tableID", s.tableID),
+				zap.String("tableName", s.tableName),
+				zap.String("holes", strings.Join(holes, ", ")))
+		}
+	}
+}
+
 func assembleRowEvent(regionID uint64, entry *cdcpb.Event_Row) (model.RegionFeedEvent, error) {
 	var opType model.OpType
 	switch entry.GetOpType() {

cdc/kv/region_worker.go

@@ -204,16 +204,18 @@ func (w *regionWorker) checkShouldExit() error {
 
 func (w *regionWorker) handleSingleRegionError(err error, state *regionFeedState) error {
 	regionID := state.getRegionID()
+	isStale := state.isStale()
 	log.Info("single region event feed disconnected",
 		zap.String("namespace", w.session.client.changefeed.Namespace),
 		zap.String("changefeed", w.session.client.changefeed.ID),
 		zap.Uint64("regionID", regionID),
 		zap.Uint64("requestID", state.requestID),
 		zap.Stringer("span", &state.sri.span),
 		zap.Uint64("resolvedTs", state.sri.resolvedTs()),
+		zap.Bool("isStale", isStale),
 		zap.Error(err))
 	// if state is already marked stopped, it must have been or would be processed by `onRegionFail`
-	if state.isStale() {
+	if isStale {
 		return w.checkShouldExit()
 	}
 	// We need to ensure when the error is handled, `isStale` must be set. So set it before sending the error.

pkg/txnutil/lock_resolver.go

@@ -50,16 +50,32 @@ func NewLockerResolver(
 
 const scanLockLimit = 1024
 
-func (r *resolver) Resolve(ctx context.Context, regionID uint64, maxVersion uint64) error {
-	// TODO test whether this function will kill active transaction
+func (r *resolver) Resolve(ctx context.Context, regionID uint64, maxVersion uint64) (err error) {
+	var lockCount int = 0
+
+	log.Info("resolve lock starts",
+		zap.Uint64("regionID", regionID),
+		zap.Uint64("maxVersion", maxVersion),
+		zap.String("namespace", r.changefeed.Namespace),
+		zap.String("changefeed", r.changefeed.ID))
 
+	defer func() {
+		log.Info("resolve lock finishes",
+			zap.Uint64("regionID", regionID),
+			zap.Int("lockCount", lockCount),
+			zap.Uint64("maxVersion", maxVersion),
+			zap.String("namespace", r.changefeed.Namespace),
+			zap.String("changefeed", r.changefeed.ID),
+			zap.Error(err))
+	}()
+
+	// TODO test whether this function will kill active transaction
 	req := tikvrpc.NewRequest(tikvrpc.CmdScanLock, &kvrpcpb.ScanLockRequest{
 		MaxVersion: maxVersion,
 		Limit:      scanLockLimit,
 	})
 
 	bo := tikv.NewGcResolveLockMaxBackoffer(ctx)
-	var lockCount int
 	var loc *tikv.KeyLocation
 	var key []byte
 	flushRegion := func() error {
@@ -128,11 +144,5 @@ func (r *resolver) Resolve(ctx context.Context, regionID uint64, maxVersion uint
 		}
 		bo = tikv.NewGcResolveLockMaxBackoffer(ctx)
 	}
-	log.Info("resolve lock successfully",
-		zap.Uint64("regionID", regionID),
-		zap.Int("lockCount", lockCount),
-		zap.Uint64("maxVersion", maxVersion),
-		zap.String("namespace", r.changefeed.Namespace),
-		zap.String("changefeed", r.changefeed.ID))
 	return nil
 }