*: fix changefeed checkpoint lag negative value error (#3013) (#3535)

cdc/capture/capture.go

@@ -32,6 +32,7 @@ import (
  cdcContext "github.com/pingcap/ticdc/pkg/context"
  cerror "github.com/pingcap/ticdc/pkg/errors"
  "github.com/pingcap/ticdc/pkg/orchestrator"
+ "github.com/pingcap/ticdc/pkg/pdtime"
  "github.com/pingcap/ticdc/pkg/version"
  tidbkv "github.com/pingcap/tidb/kv"
  pd "github.com/tikv/pd/client"
@@ -54,12 +55,12 @@ type Capture struct {
  session *concurrency.Session
  election *concurrency.Election
 
- pdClient pd.Client
- kvStorage tidbkv.Storage
- etcdClient *kv.CDCEtcdClient
- grpcPool kv.GrpcPool
-
- cancel context.CancelFunc
+ pdClient  pd.Client
+ kvStorage  tidbkv.Storage
+ etcdClient  *kv.CDCEtcdClient
+ grpcPool  kv.GrpcPool
+ TimeAcquirer pdtime.TimeAcquirer
+ cancel  context.CancelFunc
 
  newProcessorManager func() *processor.Manager
  newOwner func(pd.Client) *owner.Owner
@@ -99,6 +100,12 @@ func (c *Capture) reset(ctx context.Context) error {
  }
  c.session = sess
  c.election = concurrency.NewElection(sess, kv.CaptureOwnerKey)
+
+ if c.TimeAcquirer != nil {
+ c.TimeAcquirer.Stop()
+ }
+ c.TimeAcquirer = pdtime.NewTimeAcquirer(c.pdClient)
+
  if c.grpcPool != nil {
  c.grpcPool.Close()
  }
@@ -147,11 +154,12 @@ func (c *Capture) Run(ctx context.Context) error {
 
 func (c *Capture) run(stdCtx context.Context) error {
  ctx := cdcContext.NewContext(stdCtx, &cdcContext.GlobalVars{
- PDClient: c.pdClient,
- KVStorage: c.kvStorage,
- CaptureInfo: c.info,
- EtcdClient: c.etcdClient,
- GrpcPool: c.grpcPool,
+ PDClient: c.pdClient,
+ KVStorage: c.kvStorage,
+ CaptureInfo: c.info,
+ EtcdClient: c.etcdClient,
+ GrpcPool: c.grpcPool,
+ TimeAcquirer: c.TimeAcquirer,
  })
  err := c.register(ctx)
  if err != nil {
@@ -165,7 +173,7 @@ func (c *Capture) run(stdCtx context.Context) error {
  cancel()
  }()
  wg := new(sync.WaitGroup)
- wg.Add(3)
+ wg.Add(4)
  var ownerErr, processorErr error
  go func() {
  defer wg.Done()
@@ -187,6 +195,10 @@ func (c *Capture) run(stdCtx context.Context) error {
  processorErr = c.runEtcdWorker(ctx, c.processorManager, model.NewGlobalState(), processorFlushInterval)
  log.Info("the processor routine has exited", zap.Error(processorErr))
  }()
+ go func() {
+ defer wg.Done()
+ c.TimeAcquirer.Run(ctx)
+ }()
  go func() {
  defer wg.Done()
  c.grpcPool.RecycleConn(ctx)

cdc/owner/changefeed.go

@@ -16,7 +16,6 @@ package owner
 import (
  "context"
  "sync"
- "time"
 
  "github.com/pingcap/errors"
  "github.com/pingcap/failpoint"
@@ -178,7 +177,9 @@ func (c *changefeed) tick(ctx cdcContext.Context, state *model.ChangefeedReactor
  return errors.Trace(err)
  }
  if shouldUpdateState {
- c.updateStatus(barrierTs)
+ pdTime, _ := ctx.GlobalVars().TimeAcquirer.CurrentTimeFromCached()
+ currentTs := oracle.GetPhysical(pdTime)
+ c.updateStatus(currentTs, barrierTs)
  }
  return nil
 }
@@ -438,7 +439,7 @@ func (c *changefeed) asyncExecDDL(ctx cdcContext.Context, job *timodel.Job) (don
  return done, nil
 }
 
-func (c *changefeed) updateStatus(barrierTs model.Ts) {
+func (c *changefeed) updateStatus(currentTs int64, barrierTs model.Ts) {
  resolvedTs := barrierTs
  for _, position := range c.state.TaskPositions {
  if resolvedTs > position.ResolvedTs {
@@ -470,12 +471,10 @@ func (c *changefeed) updateStatus(barrierTs model.Ts) {
  }
  return status, changed, nil
  })
-
  phyTs := oracle.ExtractPhysical(checkpointTs)
+
  c.metricsChangefeedCheckpointTsGauge.Set(float64(phyTs))
- // It is more accurate to get tso from PD, but in most cases since we have
- // deployed NTP service, a little bias is acceptable here.
- c.metricsChangefeedCheckpointTsLagGauge.Set(float64(oracle.GetPhysical(time.Now())-phyTs) / 1e3)
+ c.metricsChangefeedCheckpointTsLagGauge.Set(float64(currentTs-phyTs) / 1e3)
 }
 
 func (c *changefeed) Close() {

cdc/owner/changefeed_test.go

@@ -26,6 +26,7 @@ import (
  "github.com/pingcap/ticdc/pkg/config"
  cdcContext "github.com/pingcap/ticdc/pkg/context"
  "github.com/pingcap/ticdc/pkg/orchestrator"
+ "github.com/pingcap/ticdc/pkg/pdtime"
  "github.com/pingcap/ticdc/pkg/txnutil/gc"
  "github.com/pingcap/ticdc/pkg/util/testleak"
  "github.com/pingcap/ticdc/pkg/version"
@@ -217,6 +218,7 @@ func (s *changefeedSuite) TestExecDDL(c *check.C) {
  AdvertiseAddr: "127.0.0.1:0000",
  Version: version.ReleaseVersion,
  },
+ TimeAcquirer: pdtime.NewTimeAcquirer4Test(),
  })
  ctx = cdcContext.WithChangefeedVars(ctx, &cdcContext.ChangefeedVars{
  ID: "changefeed-id-test",

cdc/processor/manager.go

@@ -70,6 +70,7 @@ func (m *Manager) Tick(stdCtx context.Context, state orchestrator.ReactorState)
  if err := m.handleCommand(); err != nil {
  return state, err
  }
+
  captureID := ctx.GlobalVars().CaptureInfo.ID
  var inactiveChangefeedCount int
  for changefeedID, changefeedState := range globalState.Changefeeds {

cdc/processor/processor.go

@@ -20,7 +20,6 @@ import (
  "math"
  "strconv"
  "sync"
- "time"
 
  "github.com/pingcap/errors"
  "github.com/pingcap/failpoint"
@@ -154,7 +153,8 @@ func (p *processor) tick(ctx cdcContext.Context, state *model.ChangefeedReactorS
  if !p.checkChangefeedNormal() {
  return nil, cerror.ErrAdminStopProcessor.GenWithStackByArgs()
  }
- if skip := p.checkPosition(); skip {
+ // we should skip this tick after create a task position
+ if p.createTaskPosition() {
  return p.changefeed, nil
  }
  if err := p.handleErrorCh(ctx); err != nil {
@@ -169,7 +169,12 @@ func (p *processor) tick(ctx cdcContext.Context, state *model.ChangefeedReactorS
  if err := p.checkTablesNum(ctx); err != nil {
  return nil, errors.Trace(err)
  }
- p.handlePosition()
+
+ // it is no need to check the err here, because we will use
+ // local time when an error return, which is acceptable
+ pdTime, _ := ctx.GlobalVars().TimeAcquirer.CurrentTimeFromCached()
+
+ p.handlePosition(oracle.GetPhysical(pdTime))
  p.pushResolvedTs2Table()
  p.handleWorkload()
  p.doGCSchemaStorage(ctx)
@@ -187,10 +192,10 @@ func (p *processor) checkChangefeedNormal() bool {
  return true
 }
 
-// checkPosition create a new task position, and put it into the etcd state.
-// task position maybe be not exist only when the processor is running first time.
-func (p *processor) checkPosition() (skipThisTick bool) {
- if p.changefeed.TaskPositions[p.captureInfo.ID] != nil {
+// createTaskPosition will create a new task position if a task position does not exist.
+// task position not exist only when the processor is running first in the first tick.
+func (p *processor) createTaskPosition() (skipThisTick bool) {
+ if _, exist := p.changefeed.TaskPositions[p.captureInfo.ID]; exist {
  return false
  }
  if p.initialized {
@@ -552,7 +557,7 @@ func (p *processor) checkTablesNum(ctx cdcContext.Context) error {
 }
 
 // handlePosition calculates the local resolved ts and local checkpoint ts
-func (p *processor) handlePosition() {
+func (p *processor) handlePosition(currentTs int64) {
  minResolvedTs := uint64(math.MaxUint64)
  if p.schemaStorage != nil {
  minResolvedTs = p.schemaStorage.ResolvedTs()
@@ -573,15 +578,11 @@ func (p *processor) handlePosition() {
  }
 
  resolvedPhyTs := oracle.ExtractPhysical(minResolvedTs)
- // It is more accurate to get tso from PD, but in most cases we have
- // deployed NTP service, a little bias is acceptable here.
- p.metricResolvedTsLagGauge.Set(float64(oracle.GetPhysical(time.Now())-resolvedPhyTs) / 1e3)
+ p.metricResolvedTsLagGauge.Set(float64(currentTs-resolvedPhyTs) / 1e3)
  p.metricResolvedTsGauge.Set(float64(resolvedPhyTs))
 
  checkpointPhyTs := oracle.ExtractPhysical(minCheckpointTs)
- // It is more accurate to get tso from PD, but in most cases we have
- // deployed NTP service, a little bias is acceptable here.
- p.metricCheckpointTsLagGauge.Set(float64(oracle.GetPhysical(time.Now())-checkpointPhyTs) / 1e3)
+ p.metricCheckpointTsLagGauge.Set(float64(currentTs-checkpointPhyTs) / 1e3)
  p.metricCheckpointTsGauge.Set(float64(checkpointPhyTs))
 
  // minResolvedTs and minCheckpointTs may less than global resolved ts and global checkpoint ts when a new table added, the startTs of the new table is less than global checkpoint ts.

pkg/context/context.go

@@ -18,6 +18,8 @@ import (
  "log"
  "time"
 
+ "github.com/pingcap/ticdc/pkg/pdtime"
+
  "github.com/pingcap/ticdc/pkg/version"
 
  "github.com/pingcap/ticdc/cdc/kv"
@@ -33,11 +35,12 @@ import (
 // the lifecycle of vars in the GlobalVars should be aligned with the ticdc server process.
 // All field in Vars should be READ-ONLY and THREAD-SAFE
 type GlobalVars struct {
- PDClient pd.Client
- KVStorage tidbkv.Storage
- CaptureInfo *model.CaptureInfo
- EtcdClient *kv.CDCEtcdClient
- GrpcPool kv.GrpcPool
+ PDClient pd.Client
+ KVStorage tidbkv.Storage
+ CaptureInfo *model.CaptureInfo
+ EtcdClient *kv.CDCEtcdClient
+ GrpcPool kv.GrpcPool
+ TimeAcquirer pdtime.TimeAcquirer
 }
 
 // ChangefeedVars contains some vars which can be used anywhere in a pipeline
@@ -184,6 +187,7 @@ func NewBackendContext4Test(withChangefeedVars bool) Context {
  AdvertiseAddr: "127.0.0.1:0000",
  Version: version.ReleaseVersion,
  },
+ TimeAcquirer: pdtime.NewTimeAcquirer4Test(),
  })
  if withChangefeedVars {
  ctx = WithChangefeedVars(ctx, &ChangefeedVars{

pkg/orchestrator/etcd_worker_test.go

@@ -18,6 +18,7 @@ import (
  "encoding/json"
  "regexp"
  "strconv"
+ "strings"
  "sync"
  "testing"
  "time"
@@ -224,7 +225,6 @@ func (s *etcdWorkerSuite) TestEtcdSum(c *check.C) {
  defer func() {
  _ = cli.Unwrap().Close()
  }()
-
  _, err := cli.Put(ctx, testEtcdKeyPrefix+"/sum", "0")
  c.Check(err, check.IsNil)
 
@@ -273,7 +273,9 @@ func (s *etcdWorkerSuite) TestEtcdSum(c *check.C) {
  }
 
  err = errg.Wait()
- if err != nil && (errors.Cause(err) == context.DeadlineExceeded || errors.Cause(err) == context.Canceled) {
+ if err != nil && (errors.Cause(err) == context.DeadlineExceeded ||
+ errors.Cause(err) == context.Canceled ||
+ strings.Contains(err.Error(), "etcdserver: request timeout")) {
  return
  }
  c.Check(err, check.IsNil)