*: support for lag calculation based on PD TSO (#892)

Signed-off-by: dongmen <414110582@qq.com>

coordinator/coordinator.go

@@ -28,12 +28,12 @@ import (
 	"github.com/pingcap/ticdc/pkg/messaging"
 	"github.com/pingcap/ticdc/pkg/metrics"
 	"github.com/pingcap/ticdc/pkg/node"
+	"github.com/pingcap/ticdc/pkg/pdutil"
 	"github.com/pingcap/ticdc/pkg/server"
 	"github.com/pingcap/ticdc/server/watcher"
 	"github.com/pingcap/ticdc/utils/chann"
 	"github.com/pingcap/ticdc/utils/threadpool"
 	"github.com/pingcap/tiflow/cdc/model"
-	"github.com/pingcap/tiflow/pkg/pdutil"
 	"github.com/pingcap/tiflow/pkg/txnutil/gc"
 	"github.com/tikv/client-go/v2/oracle"
 	pd "github.com/tikv/pd/client"

coordinator/coordinator_test.go

@@ -37,10 +37,10 @@ import (
 	"github.com/pingcap/ticdc/pkg/messaging"
 	"github.com/pingcap/ticdc/pkg/messaging/proto"
 	"github.com/pingcap/ticdc/pkg/node"
+	"github.com/pingcap/ticdc/pkg/pdutil"
 	"github.com/pingcap/ticdc/server/watcher"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/pkg/orchestrator"
-	"github.com/pingcap/tiflow/pkg/pdutil"
 	"github.com/stretchr/testify/require"
 	pd "github.com/tikv/pd/client"
 	"go.uber.org/zap"

downstreamadapter/dispatchermanager/event_dispatcher_manager.go

@@ -33,6 +33,7 @@ import (
 	"github.com/pingcap/ticdc/pkg/config"
 	"github.com/pingcap/ticdc/pkg/metrics"
 	"github.com/pingcap/ticdc/pkg/node"
+	"github.com/pingcap/ticdc/pkg/pdutil"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/tikv/client-go/v2/oracle"
 	"go.uber.org/zap"
@@ -56,6 +57,8 @@ type EventDispatcherManager struct {
 	changefeedID common.ChangeFeedID
 	maintainerID node.ID
 
+	pdClock pdutil.Clock
+
 	config       *config.ChangefeedConfig
 	filterConfig *eventpb.FilterConfig
 	// only not nil when enable sync point
@@ -122,10 +125,12 @@ func NewEventDispatcherManager(
 ) (*EventDispatcherManager, uint64, error) {
 	ctx, cancel := context.WithCancel(context.Background())
 	var wg sync.WaitGroup
+	pdClock := appcontext.GetService[pdutil.Clock](appcontext.DefaultPDClock)
 	manager := &EventDispatcherManager{
 		dispatcherMap:                          newDispatcherMap(),
 		changefeedID:                           changefeedID,
 		maintainerID:                           maintainerID,
+		pdClock:                                pdClock,
 		statusesChan:                           make(chan TableSpanStatusWithSeq, 8192),
 		blockStatusesChan:                      make(chan *heartbeatpb.TableSpanBlockStatus, 1024*1024),
 		errCh:                                  make(chan error, 1),
@@ -654,8 +659,9 @@ func (e *EventDispatcherManager) aggregateDispatcherHeartbeats(needCompleteStatu
 	phyCheckpointTs := oracle.ExtractPhysical(message.Watermark.CheckpointTs)
 	phyResolvedTs := oracle.ExtractPhysical(message.Watermark.ResolvedTs)
 
-	e.metricCheckpointTsLag.Set(float64(oracle.GetPhysical(time.Now())-phyCheckpointTs) / 1e3)
-	e.metricResolvedTsLag.Set(float64(oracle.GetPhysical(time.Now())-phyResolvedTs) / 1e3)
+	pdTime := e.pdClock.CurrentTime()
+	e.metricCheckpointTsLag.Set(float64(oracle.GetPhysical(pdTime)-phyCheckpointTs) / 1e3)
+	e.metricResolvedTsLag.Set(float64(oracle.GetPhysical(pdTime)-phyResolvedTs) / 1e3)
 	return &message
 }
 

downstreamadapter/eventcollector/event_collector.go

@@ -34,7 +34,6 @@ import (
 	"github.com/pingcap/ticdc/utils/dynstream"
 	"github.com/pingcap/tiflow/pkg/chann"
 	"github.com/prometheus/client_golang/prometheus"
-	"github.com/tikv/client-go/v2/oracle"
 	"go.uber.org/zap"
 )
 
@@ -428,30 +427,10 @@ func (c *EventCollector) updateMetrics(ctx context.Context) {
 			metricsDSPendingQueueLen.Set(float64(dsMetrics.PendingQueueLen))
 			metricsDSUsedMemoryUsage.Set(float64(dsMetrics.MemoryControl.UsedMemory))
 			metricsDSMaxMemoryUsage.Set(float64(dsMetrics.MemoryControl.MaxMemory))
-			c.updateResolvedTsMetric()
 		}
 	}
 }
 
-func (c *EventCollector) updateResolvedTsMetric() {
-	var minResolvedTs uint64
-	c.dispatcherMap.Range(func(key, value interface{}) bool {
-		if stat, ok := value.(*dispatcherStat); ok {
-			d := stat.target
-			if minResolvedTs == 0 || d.GetResolvedTs() < minResolvedTs {
-				minResolvedTs = d.GetResolvedTs()
-			}
-		}
-		return true
-	})
-
-	if minResolvedTs > 0 {
-		phyResolvedTs := oracle.ExtractPhysical(minResolvedTs)
-		lagMs := float64(oracle.GetPhysical(time.Now())-phyResolvedTs) / 1e3
-		metrics.EventCollectorResolvedTsLagGauge.Set(lagMs)
-	}
-}
-
 // dispatcherStat is a helper struct to manage the state of a dispatcher.
 type dispatcherStat struct {
 	dispatcherID common.DispatcherID

downstreamadapter/sink/helper/topicmanager/kafka_topic_manager.go

@@ -23,8 +23,8 @@ import (
 	"github.com/pingcap/log"
 	"github.com/pingcap/ticdc/pkg/common"
 	cerror "github.com/pingcap/ticdc/pkg/errors"
+	"github.com/pingcap/ticdc/pkg/retry"
 	"github.com/pingcap/ticdc/pkg/sink/kafka"
-	"github.com/pingcap/tiflow/pkg/retry"
 	"go.uber.org/zap"
 )
 

logservice/eventstore/event_store.go

@@ -35,8 +35,8 @@ import (
 	"github.com/pingcap/ticdc/pkg/messaging"
 	"github.com/pingcap/ticdc/pkg/metrics"
 	"github.com/pingcap/ticdc/pkg/node"
+	"github.com/pingcap/ticdc/pkg/pdutil"
 	"github.com/pingcap/ticdc/utils/chann"
-	"github.com/pingcap/tiflow/pkg/pdutil"
 	"github.com/pingcap/tiflow/pkg/util"
 	"github.com/tikv/client-go/v2/oracle"
 	"go.uber.org/zap"
@@ -696,23 +696,23 @@ func (e *eventStore) updateMetrics(ctx context.Context) error {
 }
 
 func (e *eventStore) updateMetricsOnce() {
-	currentTime := e.pdClock.CurrentTime()
-	currentPhyTs := oracle.GetPhysical(currentTime)
+	pdTime := e.pdClock.CurrentTime()
+	pdPhyTs := oracle.GetPhysical(pdTime)
 	minResolvedTs := uint64(0)
 	e.dispatcherMeta.RLock()
 	for _, subscriptionStat := range e.dispatcherMeta.subscriptionStats {
 		// resolved ts lag
 		resolvedTs := subscriptionStat.resolvedTs.Load()
 		resolvedPhyTs := oracle.ExtractPhysical(resolvedTs)
-		resolvedLag := float64(currentPhyTs-resolvedPhyTs) / 1e3
+		resolvedLag := float64(pdPhyTs-resolvedPhyTs) / 1e3
 		metrics.EventStoreDispatcherResolvedTsLagHist.Observe(float64(resolvedLag))
 		if minResolvedTs == 0 || resolvedTs < minResolvedTs {
 			minResolvedTs = resolvedTs
 		}
 		// checkpoint ts lag
 		checkpointTs := subscriptionStat.checkpointTs.Load()
 		watermarkPhyTs := oracle.ExtractPhysical(checkpointTs)
-		watermarkLag := float64(currentPhyTs-watermarkPhyTs) / 1e3
+		watermarkLag := float64(pdPhyTs-watermarkPhyTs) / 1e3
 		metrics.EventStoreDispatcherWatermarkLagHist.Observe(float64(watermarkLag))
 	}
 	e.dispatcherMeta.RUnlock()
@@ -721,7 +721,7 @@ func (e *eventStore) updateMetricsOnce() {
 		return
 	}
 	minResolvedPhyTs := oracle.ExtractPhysical(minResolvedTs)
-	eventStoreResolvedTsLag := float64(currentPhyTs-minResolvedPhyTs) / 1e3
+	eventStoreResolvedTsLag := float64(pdPhyTs-minResolvedPhyTs) / 1e3
 	metrics.EventStoreResolvedTsLagGauge.Set(eventStoreResolvedTsLag)
 }
 

logservice/logpuller/subscription_client.go

@@ -29,8 +29,8 @@ import (
 	appcontext "github.com/pingcap/ticdc/pkg/common/context"
 	cerror "github.com/pingcap/ticdc/pkg/errors"
 	"github.com/pingcap/ticdc/pkg/metrics"
+	"github.com/pingcap/ticdc/pkg/pdutil"
 	"github.com/pingcap/ticdc/utils/dynstream"
-	"github.com/pingcap/tiflow/pkg/pdutil"
 	"github.com/pingcap/tiflow/pkg/security"
 	"github.com/pingcap/tiflow/pkg/spanz"
 	"github.com/pingcap/tiflow/pkg/util"
@@ -867,8 +867,9 @@ func (s *SubscriptionClient) GetResolvedTsLag() float64 {
 	if pullerMinResolvedTs == 0 {
 		return 0
 	}
+	pdTime := s.pdClock.CurrentTime()
 	phyResolvedTs := oracle.ExtractPhysical(pullerMinResolvedTs)
-	lag := float64(oracle.GetPhysical(time.Now())-phyResolvedTs) / 1e3
+	lag := float64(oracle.GetPhysical(pdTime)-phyResolvedTs) / 1e3
 	return lag
 }
 

logservice/logpuller/subscription_client_test.go

@@ -25,8 +25,8 @@ import (
 	"github.com/pingcap/ticdc/logservice/logpuller/regionlock"
 	"github.com/pingcap/ticdc/logservice/txnutil"
 	"github.com/pingcap/ticdc/pkg/common"
+	"github.com/pingcap/ticdc/pkg/pdutil"
 	"github.com/pingcap/tidb/pkg/store/mockstore/mockcopr"
-	"github.com/pingcap/tiflow/pkg/pdutil"
 	"github.com/pingcap/tiflow/pkg/security"
 	"github.com/stretchr/testify/require"
 	"github.com/tikv/client-go/v2/oracle"

logservice/schemastore/schema_store.go

@@ -25,8 +25,8 @@ import (
 	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
 	"github.com/pingcap/ticdc/pkg/filter"
 	"github.com/pingcap/ticdc/pkg/metrics"
+	"github.com/pingcap/ticdc/pkg/pdutil"
 	"github.com/pingcap/tidb/pkg/kv"
-	"github.com/pingcap/tiflow/pkg/pdutil"
 	"github.com/tikv/client-go/v2/oracle"
 	pd "github.com/tikv/pd/client"
 	"go.uber.org/zap"
@@ -160,9 +160,9 @@ func (s *schemaStore) updateResolvedTsPeriodically(ctx context.Context) error {
 	tryUpdateResolvedTs := func() {
 		pendingTs := s.pendingResolvedTs.Load()
 		defer func() {
-			currentPhyTs := oracle.GetPhysical(s.pdClock.CurrentTime())
+			pdPhyTs := oracle.GetPhysical(s.pdClock.CurrentTime())
 			resolvedPhyTs := oracle.ExtractPhysical(pendingTs)
-			resolvedLag := float64(currentPhyTs-resolvedPhyTs) / 1e3
+			resolvedLag := float64(pdPhyTs-resolvedPhyTs) / 1e3
 			metrics.SchemaStoreResolvedTsLagGauge.Set(float64(resolvedLag))
 		}()