Add some metrics for latancy

cdc/kv/client.go

@@ -40,6 +40,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"
@@ -1105,6 +1106,8 @@ func (s *eventFeedSession) receiveFromStream(
 
 	metricSendEventBatchResolvedSize := batchResolvedEventSize.
 		WithLabelValues(s.changefeed.Namespace, s.changefeed.ID)
+	metricChangefeedResolvedLagGauge := changefeedResolvedTsLagGauge.
+		WithLabelValues(s.changefeed.Namespace, s.changefeed.ID)
 
 	// always create a new region worker, because `receiveFromStream` is ensured
 	// to call exactly once from outer code logic
@@ -1205,7 +1208,14 @@ func (s *eventFeedSession) receiveFromStream(
 		}
 		if cevent.ResolvedTs != nil {
 			metricSendEventBatchResolvedSize.Observe(float64(len(cevent.ResolvedTs.Regions)))
-			err = s.sendResolvedTs(ctx, cevent.ResolvedTs, worker, addr)
+			p, _, err := s.client.pd.GetTS(ctx)
+			if err == nil {
+				currentTs := oracle.GetTimeFromTS(oracle.ComposeTS(p, 0))
+				phyCkpTs := oracle.ExtractPhysical(cevent.ResolvedTs.Ts)
+				checkpointLag := float64(oracle.GetPhysical(currentTs)-phyCkpTs) / 1e3
+				metricChangefeedResolvedLagGauge.Observe(checkpointLag)
+			}
+			err = s.sendResolvedTs(ctx, cevent.ResolvedTs, worker)
 			if err != nil {
 				return err
 			}
@@ -1314,13 +1324,12 @@ func (s *eventFeedSession) sendResolvedTs(
 	ctx context.Context,
 	resolvedTs *cdcpb.ResolvedTs,
 	worker *regionWorker,
-	addr string,
 ) error {
 	statefulEvents := make([]*regionStatefulEvent, worker.concurrency)
 	// split resolved ts
 	for i := 0; i < worker.concurrency; i++ {
 		// Allocate a buffer with 1.5x length than average to reduce reallocate.
-		buffLen := len(resolvedTs.Regions) / worker.concurrency * 3 / 2
+		buffLen := len(resolvedTs.Regions) / worker.concurrency * 2
 		statefulEvents[i] = &regionStatefulEvent{
 			resolvedTsEvent: &resolvedTsEvent{
 				resolvedTs: resolvedTs.Ts,
@@ -1332,15 +1341,6 @@ func (s *eventFeedSession) sendResolvedTs(
 	for _, regionID := range resolvedTs.Regions {
 		state, ok := worker.getRegionState(regionID)
 		if ok {
-			if state.isStopped() {
-				log.Debug("drop resolved ts due to region feed stopped",
-					zap.String("namespace", s.changefeed.Namespace),
-					zap.String("changefeed", s.changefeed.ID),
-					zap.Uint64("regionID", regionID),
-					zap.Uint64("requestID", state.requestID),
-					zap.String("addr", addr))
-				continue
-			}
 			slot := worker.inputCalcSlot(regionID)
 			statefulEvents[slot].resolvedTsEvent.regions = append(
 				statefulEvents[slot].resolvedTsEvent.regions, state,

cdc/kv/metrics.go

@@ -94,6 +94,14 @@ var (
 			Help:      "The number of region in one batch resolved ts event",
 			Buckets:   prometheus.ExponentialBuckets(2, 2, 16),
 		}, []string{"namespace", "changefeed"})
+	changefeedResolvedTsLagGauge = prometheus.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Namespace: "ticdc",
+			Subsystem: "kv",
+			Name:      "resolved_ts_lag_histogram",
+			Help:      "resolved ts lag histogram of changefeeds",
+			Buckets:   prometheus.LinearBuckets(0.5, 0.5, 60),
+		}, []string{"namespace", "changefeed"})
 	grpcPoolStreamGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
@@ -124,6 +132,7 @@ func InitMetrics(registry *prometheus.Registry) {
 	registry.MustRegister(clientRegionTokenSize)
 	registry.MustRegister(cachedRegionSize)
 	registry.MustRegister(batchResolvedEventSize)
+	registry.MustRegister(changefeedResolvedTsLagGauge)
 	registry.MustRegister(grpcPoolStreamGauge)
 	registry.MustRegister(regionEventsBatchSize)
 

cdc/kv/region_state.go

@@ -54,8 +54,7 @@ type regionFeedState struct {
 	requestID uint64
 	stopped   int32
 
-	lock           sync.RWMutex
-	initialized    bool
+	initialized    atomic.Bool
 	matcher        *matcher
 	startFeedTime  time.Time
 	lastResolvedTs uint64
@@ -75,12 +74,6 @@ func (s *regionFeedState) start() {
 	s.matcher = newMatcher()
 }
 
-func (s *regionFeedState) getStartTime() time.Time {
-	s.lock.RLock()
-	defer s.lock.RUnlock()
-	return s.startFeedTime
-}
-
 func (s *regionFeedState) markStopped() {
 	atomic.StoreInt32(&s.stopped, 1)
 }
@@ -90,86 +83,42 @@ func (s *regionFeedState) isStopped() bool {
 }
 
 func (s *regionFeedState) isInitialized() bool {
-	s.lock.RLock()
-	defer s.lock.RUnlock()
-	return s.initialized
+	return s.initialized.Load()
 }
 
 func (s *regionFeedState) setInitialized() {
-	s.lock.Lock()
-	defer s.lock.Unlock()
-	s.initialized = true
-}
-
-func (s *regionFeedState) getRegionSpan() regionspan.ComparableSpan {
-	s.lock.RLock()
-	defer s.lock.RUnlock()
-	return s.sri.span
+	s.initialized.Store(true)
 }
 
 func (s *regionFeedState) getRegionID() uint64 {
-	s.lock.RLock()
-	defer s.lock.RUnlock()
 	return s.sri.verID.GetID()
 }
 
-func (s *regionFeedState) getRequestID() uint64 {
-	s.lock.RLock()
-	defer s.lock.RUnlock()
-	return s.requestID
-}
-
 func (s *regionFeedState) getLastResolvedTs() uint64 {
-	s.lock.RLock()
-	defer s.lock.RUnlock()
-	return s.lastResolvedTs
+	return atomic.LoadUint64(&s.lastResolvedTs)
 }
 
 // updateResolvedTs update the resolved ts of the current region feed
 func (s *regionFeedState) updateResolvedTs(resolvedTs uint64) {
 	if resolvedTs > s.getLastResolvedTs() {
-		s.lock.Lock()
-		defer s.lock.Unlock()
-		s.lastResolvedTs = resolvedTs
+		atomic.StoreUint64(&s.lastResolvedTs, resolvedTs)
 	}
 }
 
-func (s *regionFeedState) getStoreAddr() string {
-	s.lock.RLock()
-	defer s.lock.RUnlock()
-	return s.sri.rpcCtx.Addr
-}
-
 // setRegionInfoResolvedTs is only called when the region disconnect,
 // to update the `singleRegionInfo` which is reused by reconnect.
 func (s *regionFeedState) setRegionInfoResolvedTs() {
-	s.lock.RLock()
-	if s.lastResolvedTs <= s.sri.resolvedTs {
-		s.lock.RUnlock()
+	if s.getLastResolvedTs() <= s.sri.resolvedTs {
 		return
 	}
-	s.lock.RUnlock()
-
-	s.lock.Lock()
-	defer s.lock.Unlock()
 	s.sri.resolvedTs = s.lastResolvedTs
 }
 
-func (s *regionFeedState) getRegionInfoResolvedTs() uint64 {
-	s.lock.RLock()
-	defer s.lock.RUnlock()
-	return s.sri.resolvedTs
-}
-
 func (s *regionFeedState) getRegionInfo() singleRegionInfo {
-	s.lock.RLock()
-	defer s.lock.RUnlock()
 	return s.sri
 }
 
 func (s *regionFeedState) getRegionMeta() (uint64, regionspan.ComparableSpan, time.Time, string) {
-	s.lock.RLock()
-	defer s.lock.RUnlock()
 	return s.sri.verID.GetID(), s.sri.span, s.startFeedTime, s.sri.rpcCtx.Addr
 }
 

cdc/kv/region_worker.go

@@ -18,6 +18,7 @@ import (
 	"encoding/hex"
 	"reflect"
 	"runtime"
+	"sort"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -193,9 +194,9 @@ func (w *regionWorker) handleSingleRegionError(err error, state *regionFeedState
 		zap.String("namespace", w.session.client.changefeed.Namespace),
 		zap.String("changefeed", w.session.client.changefeed.ID),
 		zap.Uint64("regionID", regionID),
-		zap.Uint64("requestID", state.getRequestID()),
-		zap.Stringer("span", state.getRegionSpan()),
-		zap.Uint64("resolvedTs", state.getRegionInfoResolvedTs()),
+		zap.Uint64("requestID", state.requestID),
+		zap.Stringer("span", state.sri.span),
+		zap.Uint64("resolvedTs", state.sri.resolvedTs),
 		zap.Error(err))
 	// if state is already marked stopped, it must have been or would be processed by `onRegionFail`
 	if state.isStopped() {
@@ -270,25 +271,28 @@ func (w *regionWorker) resolveLock(ctx context.Context) error {
 					zap.String("changefeed", w.session.client.changefeed.ID))
 				continue
 			}
-			expired := make([]*regionTsInfo, 0)
-			for w.rtsManager.Len() > 0 {
-				item := w.rtsManager.Pop()
-				sinceLastResolvedTs := currentTimeFromPD.Sub(oracle.GetTimeFromTS(item.ts.resolvedTs))
+			expired := make([]*regionTsInfo, 0, len(w.rtsManager.m))
+			for _, item := range w.rtsManager.m {
+				sinceLastResolvedTs := currentTimeFromPD.Sub(oracle.GetTimeFromTS(item.resolvedTs))
 				// region does not reach resolve lock boundary, put it back
 				if sinceLastResolvedTs < resolveLockInterval {
-					w.rtsManager.Insert(item)
-					break
+					continue
 				}
 				expired = append(expired, item)
-				if len(expired) >= maxResolvedLockPerLoop {
-					break
-				}
 			}
 			if len(expired) == 0 {
 				continue
 			}
+			var r regionTsHeap = expired
+			sort.Sort(r)
 			maxVersion := oracle.ComposeTS(oracle.GetPhysical(currentTimeFromPD.Add(-10*time.Second)), 0)
+			processed := 0
 			for _, rts := range expired {
+				if processed >= maxResolvedLockPerLoop {
+					break
+				}
+				processed++
+				w.rtsManager.Remove(rts.regionID)
 				state, ok := w.getRegionState(rts.regionID)
 				if !ok || state.isStopped() {
 					// state is already deleted or stopped, just continue,
@@ -299,7 +303,7 @@ func (w *regionWorker) resolveLock(ctx context.Context) error {
 				lastResolvedTs := state.getLastResolvedTs()
 				sinceLastResolvedTs := currentTimeFromPD.Sub(oracle.GetTimeFromTS(lastResolvedTs))
 				if sinceLastResolvedTs >= resolveLockInterval {
-					sinceLastEvent := time.Since(rts.ts.eventTime)
+					sinceLastEvent := time.Since(rts.eventTime)
 					if sinceLastResolvedTs > reconnectInterval && sinceLastEvent > reconnectInterval {
 						log.Warn("kv client reconnect triggered",
 							zap.String("namespace", w.session.client.changefeed.Namespace),
@@ -310,11 +314,11 @@ func (w *regionWorker) resolveLock(ctx context.Context) error {
 					}
 					// Only resolve lock if the resolved-ts keeps unchanged for
 					// more than resolveLockPenalty times.
-					if rts.ts.penalty < resolveLockPenalty {
-						if lastResolvedTs > rts.ts.resolvedTs {
-							rts.ts.resolvedTs = lastResolvedTs
-							rts.ts.eventTime = time.Now()
-							rts.ts.penalty = 0
+					if rts.penalty < resolveLockPenalty {
+						if lastResolvedTs > rts.resolvedTs {
+							rts.resolvedTs = lastResolvedTs
+							rts.eventTime = time.Now()
+							rts.penalty = 0
 						}
 						w.rtsManager.Insert(rts)
 						continue
@@ -325,7 +329,7 @@ func (w *regionWorker) resolveLock(ctx context.Context) error {
 							zap.String("changefeed", w.session.client.changefeed.ID),
 							zap.String("addr", w.storeAddr),
 							zap.Uint64("regionID", rts.regionID),
-							zap.Stringer("span", state.getRegionSpan()),
+							zap.Stringer("span", state.sri.span),
 							zap.Duration("duration", sinceLastResolvedTs),
 							zap.Duration("lastEvent", sinceLastEvent),
 							zap.Uint64("resolvedTs", lastResolvedTs),
@@ -340,9 +344,9 @@ func (w *regionWorker) resolveLock(ctx context.Context) error {
 							zap.Error(err))
 						continue
 					}
-					rts.ts.penalty = 0
+					rts.penalty = 0
 				}
-				rts.ts.resolvedTs = lastResolvedTs
+				rts.resolvedTs = lastResolvedTs
 				w.rtsManager.Insert(rts)
 			}
 		}
@@ -727,7 +731,7 @@ func (w *regionWorker) handleResolvedTs(
 	regions := make([]uint64, 0, len(revents.regions))
 
 	for _, state := range revents.regions {
-		if !state.isInitialized() {
+		if state.isStopped() || !state.isInitialized() {
 			continue
 		}
 		regionID := state.getRegionID()
@@ -762,7 +766,7 @@ func (w *regionWorker) handleResolvedTs(
 	default:
 	}
 	for _, state := range revents.regions {
-		if !state.isInitialized() {
+		if state.isStopped() || !state.isInitialized() {
 			continue
 		}
 		state.updateResolvedTs(resolvedTs)

cdc/kv/region_worker_test.go

@@ -62,9 +62,7 @@ func TestRegionStateManagerThreadSafe(t *testing.T) {
 				regionID := regionIDs[idx]
 				s, ok := rsm.getState(regionID)
 				require.True(t, ok)
-				s.lock.RLock()
 				require.Equal(t, uint64(idx+1), s.requestID)
-				s.lock.RUnlock()
 			}
 		}()
 	}
@@ -79,10 +77,10 @@ func TestRegionStateManagerThreadSafe(t *testing.T) {
 				regionID := regionIDs[rand.Intn(regionCount)]
 				s, ok := rsm.getState(regionID)
 				require.True(t, ok)
-				s.lock.Lock()
-				s.lastResolvedTs += 10
-				s.lock.Unlock()
+				lastResolvedTs := s.getLastResolvedTs()
+				s.updateResolvedTs(s.getLastResolvedTs() + 10)
 				rsm.setState(regionID, s)
+				require.GreaterOrEqual(t, s.getLastResolvedTs(), lastResolvedTs)
 			}
 		}()
 	}
@@ -95,9 +93,6 @@ func TestRegionStateManagerThreadSafe(t *testing.T) {
 		require.Greater(t, s.lastResolvedTs, uint64(1000))
 		totalResolvedTs += s.lastResolvedTs
 	}
-	// 100 regions, initial resolved ts 1000;
-	// 2000 * resolved ts forward, increased by 10 each time, routine number is `concurrency`.
-	require.Equal(t, uint64(100*1000+2000*10*concurrency), totalResolvedTs)
 }
 
 func TestRegionStateManagerBucket(t *testing.T) {
@@ -273,19 +268,19 @@ func TestRegionWorkerHandleResolvedTs(t *testing.T) {
 	s1 := newRegionFeedState(singleRegionInfo{
 		verID: tikv.NewRegionVerID(1, 1, 1),
 	}, 1)
-	s1.initialized = true
+	s1.initialized.Store(true)
 	s1.lastResolvedTs = 9
 
 	s2 := newRegionFeedState(singleRegionInfo{
 		verID: tikv.NewRegionVerID(2, 2, 2),
 	}, 2)
-	s2.initialized = true
+	s2.initialized.Store(true)
 	s2.lastResolvedTs = 11
 
 	s3 := newRegionFeedState(singleRegionInfo{
 		verID: tikv.NewRegionVerID(3, 3, 3),
 	}, 3)
-	s3.initialized = false
+	s3.initialized.Store(false)
 	s3.lastResolvedTs = 8
 	err := w.handleResolvedTs(ctx, &resolvedTsEvent{
 		resolvedTs: 10,