Merge branch 'release-5.0' into clean-changefeed-metric

Makefile

@@ -78,14 +78,14 @@ build: cdc
 
 build-failpoint:
 	$(FAILPOINT_ENABLE)
-	$(GOBUILD) -ldflags '$(LDFLAGS)' -o bin/cdc ./main.go
+	$(GOBUILD) -ldflags '$(LDFLAGS)' -o bin/cdc ./cmd/cdc/main.go
 	$(FAILPOINT_DISABLE)
 
 cdc:
-	$(GOBUILD) -ldflags '$(LDFLAGS)' -o bin/cdc ./main.go
+	$(GOBUILD) -ldflags '$(LDFLAGS)' -o bin/cdc ./cmd/cdc/main.go
 
 kafka_consumer:
-	$(GOBUILD) -ldflags '$(LDFLAGS)' -o bin/cdc_kafka_consumer ./kafka_consumer/main.go
+	$(GOBUILD) -ldflags '$(LDFLAGS)' -o bin/cdc_kafka_consumer ./cmd/kafka-consumer/main.go
 
 install:
 	go install ./...
@@ -126,9 +126,9 @@ integration_test_build: check_failpoint_ctl
 	$(FAILPOINT_ENABLE)
 	$(GOTEST) -ldflags '$(LDFLAGS)' -c -cover -covermode=atomic \
 		-coverpkg=github.com/pingcap/ticdc/... \
-		-o bin/cdc.test github.com/pingcap/ticdc \
+		-o bin/cdc.test github.com/pingcap/ticdc/cmd/cdc \
 	|| { $(FAILPOINT_DISABLE); exit 1; }
-	$(GOBUILD) -ldflags '$(LDFLAGS)' -o bin/cdc ./main.go \
+	$(GOBUILD) -ldflags '$(LDFLAGS)' -o bin/cdc ./cmd/cdc/main.go \
 	|| { $(FAILPOINT_DISABLE); exit 1; }
 	$(FAILPOINT_DISABLE)
 
@@ -193,6 +193,8 @@ data-flow-diagram: docs/data-flow.dot
 clean:
 	go clean -i ./...
 	rm -rf *.out
+	rm -f bin/cdc
+	rm -f bin/cdc_kafka_consumer
 
 tools/bin/gofumports: tools/check/go.mod
 	cd tools/check; test -e ../bin/gofumports || \

cdc/capture.go

@@ -169,6 +169,7 @@ func (c *Capture) Run(ctx context.Context) (err error) {
 			// 1/3 of session ttl that `session.Done` can't be triggered even
 			// the lease is already revoked.
 			if cerror.ErrEtcdSessionDone.Equal(err) {
+				log.Warn("session is disconnected", zap.Error(err))
 				return cerror.ErrCaptureSuicide.GenWithStackByArgs()
 			}
 			lease, inErr := c.etcdClient.Client.TimeToLive(ctx, c.session.Lease())

cdc/capture/capture.go

@@ -267,6 +267,7 @@ func (c *Capture) runEtcdWorker(ctx cdcContext.Context, reactor orchestrator.Rea
 		switch {
 		case cerror.ErrEtcdSessionDone.Equal(err),
 			cerror.ErrLeaseExpired.Equal(err):
+			log.Warn("session is disconnected", zap.Error(err))
 			return cerror.ErrCaptureSuicide.GenWithStackByArgs()
 		}
 		lease, inErr := ctx.GlobalVars().EtcdClient.Client.TimeToLive(ctx, c.session.Lease())

cdc/changefeed.go

@@ -249,7 +249,7 @@ func (c *changeFeed) updatePartition(tblInfo *timodel.TableInfo, startTs uint64)
 	newPartitionIDs := make([]int64, 0, len(pi.Definitions))
 	for _, partition := range pi.Definitions {
 		pid := partition.ID
-		_, ok := c.orphanTables[pid]
+		_, ok := oldIDs[pid]
 		if !ok {
 			// new partition.
 			c.orphanTables[pid] = startTs

cdc/changefeed_test.go

@@ -19,6 +19,7 @@ import (
 	"time"
 
 	"github.com/pingcap/check"
+	timodel "github.com/pingcap/parser/model"
 	"github.com/pingcap/ticdc/cdc/kv"
 	"github.com/pingcap/ticdc/cdc/model"
 	"github.com/pingcap/ticdc/pkg/etcd"
@@ -128,3 +129,39 @@ func (s *changefeedSuite) TestHandleMoveTableJobs(c *check.C) {
 	c.Assert(changefeed.orphanTables, check.HasKey, model.TableID(1))
 	c.Assert(changefeed.moveTableJobs, check.HasLen, 0)
 }
+
+func (s *changefeedSuite) TestUpdatePartition(c *check.C) {
+	defer testleak.AfterTest(c)()
+	defer s.TearDownTest(c)
+
+	cf := changeFeed{
+		partitions: map[model.TableID][]int64{
+			51: {53, 55, 57},
+		},
+		orphanTables:  make(map[model.TableID]model.Ts),
+		toCleanTables: make(map[model.TableID]model.Ts),
+	}
+	tblInfo := &timodel.TableInfo{
+		ID: 51,
+		Partition: &timodel.PartitionInfo{
+			Enable: true,
+			Definitions: []timodel.PartitionDefinition{
+				{ID: 57}, {ID: 59}, {ID: 61},
+			},
+		},
+	}
+	startTs := uint64(100)
+
+	cf.updatePartition(tblInfo, startTs)
+	c.Assert(cf.orphanTables, check.DeepEquals, map[model.TableID]model.Ts{
+		59: startTs,
+		61: startTs,
+	})
+	c.Assert(cf.toCleanTables, check.DeepEquals, map[model.TableID]model.Ts{
+		53: startTs,
+		55: startTs,
+	})
+	c.Assert(cf.partitions, check.DeepEquals, map[model.TableID][]int64{
+		51: {57, 59, 61},
+	})
+}

cdc/kv/etcd_test.go

@@ -225,6 +225,7 @@ func (s *etcdSuite) TestOpChangeFeedDetail(c *check.C) {
 	ctx := context.Background()
 	detail := &model.ChangeFeedInfo{
 		SinkURI: "root@tcp(127.0.0.1:3306)/mysql",
+		SortDir: "/old-version/sorter",
 	}
 	cfID := "test-op-cf"
 
@@ -236,6 +237,7 @@ func (s *etcdSuite) TestOpChangeFeedDetail(c *check.C) {
 	d, err := s.client.GetChangeFeedInfo(ctx, cfID)
 	c.Assert(err, check.IsNil)
 	c.Assert(d.SinkURI, check.Equals, detail.SinkURI)
+	c.Assert(d.SortDir, check.Equals, detail.SortDir)
 
 	err = s.client.LeaseGuardDeleteChangeFeedInfo(ctx, cfID, sess.Lease())
 	c.Assert(err, check.IsNil)

cdc/model/changefeed.go

@@ -52,6 +52,26 @@ const (
 	StateFinished FeedState = "finished"
 )
 
+// ToInt return a int for each `FeedState`, only use this for metrics.
+func (s FeedState) ToInt() int {
+	switch s {
+	case StateNormal:
+		return 0
+	case StateError:
+		return 1
+	case StateFailed:
+		return 2
+	case StateStopped:
+		return 3
+	case StateFinished:
+		return 4
+	case StateRemoved:
+		return 5
+	}
+	// -1 for unknown feed state
+	return -1
+}
+
 const (
 	// errorHistoryGCInterval represents how long we keep error record in changefeed info
 	errorHistoryGCInterval = time.Minute * 10
@@ -78,7 +98,9 @@ type ChangeFeedInfo struct {
 	AdminJobType AdminJobType `json:"admin-job-type"`
 	Engine       SortEngine   `json:"sort-engine"`
 	// SortDir is deprecated
-	SortDir string `json:"-"`
+	// it cannot be set by user in changefeed level, any assignment to it should be ignored.
+	// but can be fetched for backward compatibility
+	SortDir string `json:"sort-dir"`
 
 	Config   *config.ReplicaConfig `json:"config"`
 	State    FeedState             `json:"state"`

cdc/model/changefeed_test.go

@@ -120,6 +120,7 @@ func (s *configSuite) TestFillV1(c *check.C) {
 		},
 		StartTs: 417136892416622595,
 		Engine:  "memory",
+		SortDir: ".",
 		Config: &config.ReplicaConfig{
 			CaseSensitive: true,
 			Filter: &config.FilterConfig{

cdc/owner.go

@@ -52,7 +52,7 @@ type ownership struct {
 	tickTime     time.Duration
 }
 
-func newOwnersip(tickTime time.Duration) ownership {
+func newOwnership(tickTime time.Duration) ownership {
 	minTickTime := 5 * time.Second
 	if tickTime > minTickTime {
 		log.Panic("ownership counter must be incearsed every 5 seconds")
@@ -1112,7 +1112,6 @@ func (o *Owner) handleAdminJob(ctx context.Context) error {
 			if err != nil {
 				return errors.Trace(err)
 			}
-
 			err = o.dispatchJob(ctx, job)
 			if err != nil {
 				return errors.Trace(err)
@@ -1278,7 +1277,7 @@ func (o *Owner) Run(ctx context.Context, tickTime time.Duration) error {
 	defer feedChangeReceiver.Stop()
 	o.watchFeedChange(ctx1)
 
-	ownership := newOwnersip(tickTime)
+	ownership := newOwnership(tickTime)
 loop:
 	for {
 		select {
@@ -1580,7 +1579,7 @@ func (o *Owner) watchCapture(ctx context.Context) error {
 	failpoint.Inject("sleep-before-watch-capture", nil)
 
 	// When an owner just starts, changefeed information is not updated at once.
-	// Supposing a crased capture should be removed now, the owner will miss deleting
+	// Supposing a crashed capture should be removed now, the owner will miss deleting
 	// task status and task position if changefeed information is not loaded.
 	// If the task positions and status decode failed, remove them.
 	if err := o.checkAndCleanTasksInfo(ctx); err != nil {

cdc/owner/metrics.go

@@ -44,6 +44,13 @@ var (
 			Name:      "maintain_table_num",
 			Help:      "number of replicated tables maintained in owner",
 		}, []string{"changefeed", "capture", "type"})
+	changefeedStatusGauge = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: "ticdc",
+			Subsystem: "owner",
+			Name:      "status",
+			Help:      "The status of changefeeds",
+		}, []string{"changefeed"})
 )
 
 const (
@@ -59,4 +66,5 @@ func InitMetrics(registry *prometheus.Registry) {
 	registry.MustRegister(changefeedCheckpointTsLagGauge)
 	registry.MustRegister(ownershipCounter)
 	registry.MustRegister(ownerMaintainTableNumGauge)
+	registry.MustRegister(changefeedStatusGauge)
 }

cdc/owner/owner.go

@@ -235,6 +235,7 @@ func (o *Owner) updateMetrics(state *model.GlobalReactorState) {
 	o.lastTickTime = now
 
 	ownerMaintainTableNumGauge.Reset()
+	changefeedStatusGauge.Reset()
 	for changefeedID, changefeedState := range state.Changefeeds {
 		for captureID, captureInfo := range state.Captures {
 			taskStatus, exist := changefeedState.TaskStatuses[captureID]
@@ -243,6 +244,9 @@ func (o *Owner) updateMetrics(state *model.GlobalReactorState) {
 			}
 			ownerMaintainTableNumGauge.WithLabelValues(changefeedID, captureInfo.AdvertiseAddr, maintainTableTypeTotal).Set(float64(len(taskStatus.Tables)))
 			ownerMaintainTableNumGauge.WithLabelValues(changefeedID, captureInfo.AdvertiseAddr, maintainTableTypeWip).Set(float64(len(taskStatus.Operation)))
+			if changefeedState.Info != nil {
+				changefeedStatusGauge.WithLabelValues(changefeedID).Set(float64(changefeedState.Info.State.ToInt()))
+			}
 		}
 	}
 }

cdc/puller/sorter/backend_pool.go

@@ -158,7 +158,7 @@ func newBackEndPool(dir string, captureAddr string) (*backEndPool, error) {
 				if err != nil {
 					log.Warn("Cannot remove temporary file for sorting", zap.String("file", backEnd.fileName), zap.Error(err))
 				} else {
-					log.Info("Temporary file removed", zap.String("file", backEnd.fileName))
+					log.Debug("Temporary file removed", zap.String("file", backEnd.fileName))
 					freedCount += 1
 				}
 				if freedCount >= 16 {
@@ -371,7 +371,7 @@ func (p *backEndPool) cleanUpStaleFiles() error {
 	}
 
 	for _, toRemoveFilePath := range files {
-		log.Info("Removing stale sorter temporary file", zap.String("file", toRemoveFilePath))
+		log.Debug("Removing stale sorter temporary file", zap.String("file", toRemoveFilePath))
 		err := os.Remove(toRemoveFilePath)
 		if err != nil {
 			// In production, we do not want an error here to interfere with normal operation,

cdc/puller/sorter/merger.go

@@ -55,7 +55,7 @@ func runMerger(ctx context.Context, numSorters int, in <-chan *flushTask, out ch
 	pendingSet := &sync.Map{}
 
 	defer func() {
-		log.Info("Unified Sorter: merger exiting, cleaning up resources")
+		log.Debug("Unified Sorter: merger exiting, cleaning up resources")
 		// cancel pending async IO operations.
 		onExit()
 		cleanUpTask := func(task *flushTask) {

cdc/server.go

@@ -374,10 +374,8 @@ func (s *Server) initDataDir(ctx context.Context) error {
 		return errors.Trace(err)
 	}
 
-	if diskInfo.Avail < dataDirThreshold {
-		log.Warn(fmt.Sprintf("%s is set as data-dir (%dGB available), ticdc recommend disk for data-dir "+
-			"at least have %dGB available space", conf.DataDir, diskInfo.Avail, dataDirThreshold))
-	}
+	log.Info(fmt.Sprintf("%s is set as data-dir (%dGB available), ticdc recommend disk for data-dir "+
+		"at least have %dGB available space", conf.DataDir, diskInfo.Avail, dataDirThreshold))
 
 	return nil
 }
@@ -440,14 +438,25 @@ func (s *Server) setUpDataDir(ctx context.Context) error {
 // at the moment, only consider available disk space
 func findBestDataDir(candidates []string) (result string, ok bool) {
 	var low uint64 = 0
-	for _, dir := range candidates {
+
+	checker := func(dir string) (*util.DiskInfo, error) {
+		if err := os.MkdirAll(dir, 0o755); err != nil {
+			return nil, err
+		}
 		if err := util.IsDirReadWritable(dir); err != nil {
-			log.Warn("try to get disk info failed", zap.String("dir", dir), zap.Error(err))
-			continue
+			return nil, err
 		}
 		info, err := util.GetDiskInfo(dir)
 		if err != nil {
-			log.Warn("try to get disk info failed", zap.String("dir", dir), zap.Error(err))
+			return nil, err
+		}
+		return info, err
+	}
+
+	for _, dir := range candidates {
+		info, err := checker(dir)
+		if err != nil {
+			log.Warn("check the availability of dir", zap.String("dir", dir), zap.Error(err))
 			continue
 		}
 		if info.Avail > low {

errors.toml

@@ -636,6 +636,11 @@ error = '''
 pulsar send message failed
 '''
 
+["CDC:ErrReachMaxTry"]
+error = '''
+reach maximum try: %d
+'''
+
 ["CDC:ErrReactorFinished"]
 error = '''
 the reactor has done its job and should no longer be executed