From ff29d885bb830f90a75af36b87f927d30a6ba73d Mon Sep 17 00:00:00 2001 From: dongmen <20351731+asddongmen@users.noreply.github.com> Date: Wed, 24 Jun 2026 13:06:46 +0800 Subject: [PATCH 1/3] This is an automated cherry-pick of #5491 Signed-off-by: ti-chi-bot --- coordinator/changefeed/changefeed_db.go | 2 +- coordinator/changefeed/changefeed_db_test.go | 21 + coordinator/controller.go | 62 +- coordinator/controller_test.go | 559 +++++++++++++++++++ pkg/metrics/changefeed.go | 13 + pkg/metrics/changefeed_test.go | 49 ++ 6 files changed, 703 insertions(+), 3 deletions(-) create mode 100644 pkg/metrics/changefeed_test.go diff --git a/coordinator/changefeed/changefeed_db.go b/coordinator/changefeed/changefeed_db.go index 209b972e26..45a419893a 100644 --- a/coordinator/changefeed/changefeed_db.go +++ b/coordinator/changefeed/changefeed_db.go @@ -141,7 +141,7 @@ func (db *ChangefeedDB) StopByChangefeedID(cfID common.ChangeFeedID, remove bool } metrics.ChangefeedStatusGauge.DeleteLabelValues(cfID.Keyspace(), cfID.Name()) - metrics.ChangefeedCheckpointTsLagGauge.DeleteLabelValues(cfID.Keyspace(), cfID.Name()) + metrics.DeleteChangefeedCheckpointMetrics(cfID.Keyspace(), cfID.Name()) return nodeID } diff --git a/coordinator/changefeed/changefeed_db_test.go b/coordinator/changefeed/changefeed_db_test.go index 450478e455..85689d64e5 100644 --- a/coordinator/changefeed/changefeed_db_test.go +++ b/coordinator/changefeed/changefeed_db_test.go @@ -21,7 +21,9 @@ import ( "github.com/pingcap/ticdc/pkg/common" "github.com/pingcap/ticdc/pkg/config" "github.com/pingcap/ticdc/pkg/errors" + "github.com/pingcap/ticdc/pkg/metrics" "github.com/pingcap/ticdc/pkg/node" + "github.com/prometheus/client_golang/prometheus/testutil" "github.com/stretchr/testify/require" "go.uber.org/atomic" ) @@ -133,6 +135,25 @@ func TestRemoveChangefeed(t *testing.T) { require.False(t, ok) } +func TestStopByChangefeedIDDeletesCheckpointMetrics(t *testing.T) { + metrics.ResetOwnerChangefeedMetrics() + t.Cleanup(metrics.ResetOwnerChangefeedMetrics) + + db := NewChangefeedDB(1216) + cf := &Changefeed{ID: common.NewChangeFeedIDWithName("test-metrics", common.DefaultKeyspaceName)} + db.AddReplicatingMaintainer(cf, "node1") + + metrics.ChangefeedCheckpointTsGauge.WithLabelValues(cf.ID.Keyspace(), cf.ID.Name()).Set(100) + metrics.ChangefeedCheckpointTsLagGauge.WithLabelValues(cf.ID.Keyspace(), cf.ID.Name()).Set(10) + require.Equal(t, 1, testutil.CollectAndCount(metrics.ChangefeedCheckpointTsGauge)) + require.Equal(t, 1, testutil.CollectAndCount(metrics.ChangefeedCheckpointTsLagGauge)) + + require.Equal(t, node.ID("node1"), db.StopByChangefeedID(cf.ID, true)) + + require.Equal(t, 0, testutil.CollectAndCount(metrics.ChangefeedCheckpointTsGauge)) + require.Equal(t, 0, testutil.CollectAndCount(metrics.ChangefeedCheckpointTsLagGauge)) +} + func TestGetByID(t *testing.T) { db := NewChangefeedDB(1216) cf := &Changefeed{ID: common.NewChangeFeedIDWithName("test", common.DefaultKeyspaceName)} diff --git a/coordinator/controller.go b/coordinator/controller.go index fd5217b881..3d8f9b3ff0 100644 --- a/coordinator/controller.go +++ b/coordinator/controller.go @@ -186,6 +186,15 @@ func NewController( func (c *Controller) collectMetrics(ctx context.Context) error { ticker := time.NewTicker(5 * time.Second) defer ticker.Stop() +<<<<<<< HEAD +======= + defer metrics.ResetOwnerChangefeedMetrics() + + // changefeedDownstreamTypeCache is used to cleanup the previous downstream type + // label value when a changefeed's sink-uri is updated. + changefeedDownstreamTypeCache := make(map[common.ChangeFeedDisplayName]string) + errorMetricLabels := make(map[common.ChangeFeedID]changefeedErrorMetricLabels) +>>>>>>> 5d5121dfa (coordinator: clean stale owner checkpoint metrics (#5491)) for { select { case <-ctx.Done(): @@ -203,21 +212,70 @@ func (c *Controller) collectMetrics(ctx context.Context) error { name := info.ChangefeedID.Name() metrics.ChangefeedStatusGauge.WithLabelValues(keyspace, name).Set(float64(info.State.ToInt())) - // don't update checkpoint ts and checkpoint ts lag for stopped changefeed - if info.State == config.StateStopped { + if !updateChangefeedCheckpointMetrics( + keyspace, + name, + info.State, + cf.GetLastSavedCheckPointTs(), + c.pdClock.CurrentTime(), + ) { return } +<<<<<<< HEAD pdPhysicalTime := oracle.GetPhysical(c.pdClock.CurrentTime()) phyCkpTs := oracle.ExtractPhysical(cf.GetLastSavedCheckPointTs()) lag := float64(pdPhysicalTime-phyCkpTs) / 1e3 metrics.ChangefeedCheckpointTsGauge.WithLabelValues(keyspace, name).Set(float64(phyCkpTs)) metrics.ChangefeedCheckpointTsLagGauge.WithLabelValues(keyspace, name).Set(lag) +======= + // sync changefeed error metrics + currentChangefeeds[cf.ID] = struct{}{} + oldLabels, exists := errorMetricLabels[cf.ID] + newLabels, hasError := getChangefeedErrorMetricLabels(cf.GetInfo()) + // If the error state has not changed, do nothing. + if exists && hasError && oldLabels == newLabels { + return + } + // If there was an old metric, delete it, as the state has changed. + if exists { + metrics.ChangefeedErrorInfoGauge.DeleteLabelValues(oldLabels.labelValues()...) + } + if hasError { + // An error exists (either new or changed). Set the new metric and update cache. + metrics.ChangefeedErrorInfoGauge.WithLabelValues(newLabels.labelValues()...).Set(1) + errorMetricLabels[cf.ID] = newLabels + } else { + // The error has disappeared, remove from cache. + delete(errorMetricLabels, cf.ID) + } +>>>>>>> 5d5121dfa (coordinator: clean stale owner checkpoint metrics (#5491)) }) } } } +func updateChangefeedCheckpointMetrics( + keyspace string, + name string, + state config.FeedState, + checkpointTs uint64, + pdTime time.Time, +) bool { + switch state { + case config.StateStopped, config.StateFinished, config.StateRemoved: + metrics.DeleteChangefeedCheckpointMetrics(keyspace, name) + return false + } + + pdPhysicalTime := oracle.GetPhysical(pdTime) + phyCkpTs := oracle.ExtractPhysical(checkpointTs) + lag := float64(pdPhysicalTime-phyCkpTs) / 1e3 + metrics.ChangefeedCheckpointTsGauge.WithLabelValues(keyspace, name).Set(float64(phyCkpTs)) + metrics.ChangefeedCheckpointTsLagGauge.WithLabelValues(keyspace, name).Set(lag) + return true +} + // HandleEvent implements the event-driven process mode func (c *Controller) HandleEvent(ctx context.Context, event *Event) { if event == nil { diff --git a/coordinator/controller_test.go b/coordinator/controller_test.go index 00171e78c3..0e43f7f339 100644 --- a/coordinator/controller_test.go +++ b/coordinator/controller_test.go @@ -28,12 +28,571 @@ import ( "github.com/pingcap/ticdc/pkg/config" "github.com/pingcap/ticdc/pkg/errors" "github.com/pingcap/ticdc/pkg/messaging" + "github.com/pingcap/ticdc/pkg/metrics" "github.com/pingcap/ticdc/pkg/node" "github.com/pingcap/ticdc/server/watcher" +<<<<<<< HEAD +======= + "github.com/pingcap/ticdc/utils/threadpool" + "github.com/prometheus/client_golang/prometheus/testutil" +>>>>>>> 5d5121dfa (coordinator: clean stale owner checkpoint metrics (#5491)) "github.com/stretchr/testify/require" + "github.com/tikv/client-go/v2/oracle" "go.uber.org/atomic" ) +<<<<<<< HEAD +======= +type noopScheduler struct{} + +func (noopScheduler) Execute() time.Time { + return time.Now().Add(time.Hour) +} + +func (noopScheduler) Name() string { + return pkgscheduler.BasicScheduler +} + +func TestUpdateChangefeedCheckpointMetricsDeletesFinishedLabels(t *testing.T) { + metrics.ResetOwnerChangefeedMetrics() + t.Cleanup(metrics.ResetOwnerChangefeedMetrics) + + keyspace := common.DefaultKeyspaceName + name := "finished-metrics" + pdTime := time.UnixMilli(2000) + checkpointTs := oracle.ComposeTS(1000, 0) + + require.True(t, updateChangefeedCheckpointMetrics( + keyspace, + name, + config.StateNormal, + checkpointTs, + pdTime, + )) + require.Equal(t, 1, testutil.CollectAndCount(metrics.ChangefeedCheckpointTsGauge)) + require.Equal(t, 1, testutil.CollectAndCount(metrics.ChangefeedCheckpointTsLagGauge)) + + require.False(t, updateChangefeedCheckpointMetrics( + keyspace, + name, + config.StateFinished, + checkpointTs, + pdTime, + )) + require.Equal(t, 0, testutil.CollectAndCount(metrics.ChangefeedCheckpointTsGauge)) + require.Equal(t, 0, testutil.CollectAndCount(metrics.ChangefeedCheckpointTsLagGauge)) +} + +func TestOnPeriodTaskAdvanceLiveness(t *testing.T) { + newController := func(t *testing.T) (*Controller, chan *messaging.TargetMessage, *changefeed.ChangefeedDB, node.ID) { + t.Helper() + + ctrl := gomock.NewController(t) + t.Cleanup(ctrl.Finish) + + backend := mock_changefeed.NewMockBackend(ctrl) + changefeedDB := changefeed.NewChangefeedDB(1216) + self := node.NewInfo("localhost:8300", "") + target := node.NewInfo("localhost:8301", "") + nodeManager := watcher.NewNodeManager(nil, nil) + nodeManager.GetAliveNodes()[self.ID] = self + nodeManager.GetAliveNodes()[target.ID] = target + + mc := messaging.NewMockMessageCenter() + appcontext.SetService(appcontext.MessageCenter, mc) + appcontext.SetService(watcher.NodeManagerName, nodeManager) + + return &Controller{ + changefeedDB: changefeedDB, + operatorController: operator.NewOperatorController( + self, changefeedDB, backend, nil, 10, + ), + nodeManager: nodeManager, + initialized: atomic.NewBool(true), + drainController: drain.NewController(mc), + pdClient: newDrainTestPDClient(), + bootstrapper: bootstrap.NewBootstrapper[heartbeatpb.CoordinatorBootstrapResponse]( + "test", + func(node.ID, string) *messaging.TargetMessage { return nil }, + ), + messageCenter: mc, + }, mc.GetMessageChannel(), changefeedDB, target.ID + } + + t.Run("skip stopping before bootstrap completion", func(t *testing.T) { + controller, messageCh, _, targetNodeID := newController(t) + controller.initialized.Store(false) + controller.drainController.ObserveHeartbeat(targetNodeID, &heartbeatpb.NodeHeartbeat{ + NodeEpoch: 1, + Liveness: heartbeatpb.NodeLiveness_DRAINING, + }) + + controller.onPeriodTask() + + select { + case msg := <-messageCh: + t.Fatalf("unexpected liveness command sent before bootstrap completion: %v", msg.Type) + default: + } + }) + + t.Run("skip stopping without active drain session", func(t *testing.T) { + controller, messageCh, _, targetNodeID := newController(t) + controller.drainController.ObserveHeartbeat(targetNodeID, &heartbeatpb.NodeHeartbeat{ + NodeEpoch: 1, + Liveness: heartbeatpb.NodeLiveness_DRAINING, + }) + + controller.onPeriodTask() + + select { + case msg := <-messageCh: + t.Fatalf("unexpected liveness command sent without active session: %v", msg.Type) + default: + } + }) + + t.Run("skip stopping when active drain session is not ready", func(t *testing.T) { + controller, messageCh, changefeedDB, targetNodeID := newController(t) + cfID := common.NewChangeFeedIDWithName("test", common.DefaultKeyspaceName) + cf := changefeed.NewChangefeed(cfID, &config.ChangeFeedInfo{ + ChangefeedID: cfID, + Config: config.GetDefaultReplicaConfig(), + SinkURI: "mysql://127.0.0.1:3306", + State: config.StateNormal, + }, 1, true) + changefeedDB.AddReplicatingMaintainer(cf, targetNodeID) + + _, err := controller.ensureDispatcherDrainTarget(context.Background(), targetNodeID) + require.NoError(t, err) + + controller.drainController.ObserveHeartbeat(targetNodeID, &heartbeatpb.NodeHeartbeat{ + NodeEpoch: 1, + Liveness: heartbeatpb.NodeLiveness_DRAINING, + }) + controller.onPeriodTask() + + for { + select { + case msg := <-messageCh: + if msg.Type == messaging.TypeSetNodeLivenessRequest { + t.Fatalf("unexpected liveness command sent before readiness: %v", msg.Type) + } + default: + return + } + } + }) + + t.Run("send stopping only for active drain session target", func(t *testing.T) { + controller, messageCh, _, targetNodeID := newController(t) + _, err := controller.ensureDispatcherDrainTarget(context.Background(), targetNodeID) + require.NoError(t, err) + + controller.drainSessionMu.Lock() + controller.drainSession.dirty = false + controller.drainSession.lastSent = time.Now() + controller.drainSessionMu.Unlock() + + controller.drainController.ObserveHeartbeat(targetNodeID, &heartbeatpb.NodeHeartbeat{ + NodeEpoch: 1, + Liveness: heartbeatpb.NodeLiveness_DRAINING, + }) + + controller.onPeriodTask() + + foundStop := false + for { + select { + case msg := <-messageCh: + if msg.Type != messaging.TypeSetNodeLivenessRequest { + continue + } + require.Equal(t, targetNodeID, msg.To) + req := msg.Message[0].(*heartbeatpb.SetNodeLivenessRequest) + if req.Target == heartbeatpb.NodeLiveness_STOPPING { + require.Equal(t, uint64(1), req.NodeEpoch) + foundStop = true + } + default: + require.True(t, foundStop) + return + } + } + }) +} + +func TestMaintainerHeartbeatAdmissionRequiresInitializedSender(t *testing.T) { + mc := messaging.NewMockMessageCenter() + appcontext.SetService(appcontext.MessageCenter, mc) + + nodeManager := watcher.NewNodeManager(nil, nil) + appcontext.SetService(watcher.NodeManagerName, nodeManager) + + owner := node.ID("owner") + late := node.ID("late") + nodeManager.GetAliveNodes()[owner] = &node.Info{ID: owner} + nodeManager.GetAliveNodes()[late] = &node.Info{ID: late} + + db := changefeed.NewChangefeedDB(1) + cfID := common.NewChangeFeedIDWithName("cf", common.DefaultKeyspaceName) + cf := changefeed.NewChangefeed(cfID, &config.ChangeFeedInfo{ + ChangefeedID: cfID, + Config: config.GetDefaultReplicaConfig(), + SinkURI: "blackhole://", + State: config.StateNormal, + }, 100, false) + db.AddReplicatingMaintainer(cf, late) + + controller := &Controller{ + initialized: atomic.NewBool(true), + changefeedDB: db, + operatorController: operator.NewOperatorController( + &node.Info{ID: node.ID("coordinator")}, + db, + nil, + nil, + 10, + ), + bootstrapper: bootstrap.NewBootstrapper[heartbeatpb.CoordinatorBootstrapResponse]( + "test", + func(id node.ID, _ string) *messaging.TargetMessage { + return messaging.NewSingleTargetMessage( + id, + messaging.MaintainerManagerTopic, + &heartbeatpb.CoordinatorBootstrapRequest{}, + ) + }, + ), + } + + controller.bootstrapper.HandleNodesChange(nodeManager.GetAliveNodes()) + controller.bootstrapper.HandleBootstrapResponse(owner, &heartbeatpb.CoordinatorBootstrapResponse{}) + require.False(t, controller.bootstrapper.AllNodesReady()) + + ignored := &heartbeatpb.MaintainerHeartbeat{ + Statuses: []*heartbeatpb.MaintainerStatus{{ + ChangefeedID: cfID.ToPB(), + CheckpointTs: 200, + State: heartbeatpb.ComponentState_Working, + BootstrapDone: true, + }}, + } + controller.onMessage(context.Background(), &messaging.TargetMessage{ + From: late, + Topic: messaging.CoordinatorTopic, + Type: messaging.TypeMaintainerHeartbeatRequest, + Message: []messaging.IOTypeT{ignored}, + }) + require.Equal(t, uint64(100), cf.GetStatus().CheckpointTs) + + controller.bootstrapper.HandleBootstrapResponse(late, &heartbeatpb.CoordinatorBootstrapResponse{}) + accepted := &heartbeatpb.MaintainerHeartbeat{ + Statuses: []*heartbeatpb.MaintainerStatus{{ + ChangefeedID: cfID.ToPB(), + CheckpointTs: 200, + State: heartbeatpb.ComponentState_Working, + BootstrapDone: true, + }}, + } + controller.onMessage(context.Background(), &messaging.TargetMessage{ + From: late, + Topic: messaging.CoordinatorTopic, + Type: messaging.TypeMaintainerHeartbeatRequest, + Message: []messaging.IOTypeT{accepted}, + }) + require.Equal(t, uint64(200), cf.GetStatus().CheckpointTs) +} + +func TestMaintainerHeartbeatAdmissionDropsStaleMaintainerEpoch(t *testing.T) { + appcontext.SetService(appcontext.MessageCenter, messaging.NewMockMessageCenter()) + appcontext.SetService(watcher.NodeManagerName, watcher.NewNodeManager(nil, nil)) + + db := changefeed.NewChangefeedDB(1) + cfID := common.NewChangeFeedIDWithName("cf", common.DefaultKeyspaceName) + owner := node.ID("owner") + cf := changefeed.NewChangefeed(cfID, &config.ChangeFeedInfo{ + ChangefeedID: cfID, + Config: config.GetDefaultReplicaConfig(), + SinkURI: "blackhole://", + State: config.StateNormal, + Epoch: 2, + }, 100, false) + db.AddReplicatingMaintainer(cf, owner) + + controller := &Controller{ + changefeedDB: db, + operatorController: operator.NewOperatorController( + &node.Info{ID: node.ID("coordinator")}, + db, + nil, + nil, + 10, + ), + } + + stale := &heartbeatpb.MaintainerStatus{ + ChangefeedID: cfID.ToPB(), + CheckpointTs: 200, + State: heartbeatpb.ComponentState_Working, + BootstrapDone: true, + MaintainerEpoch: 1, + } + require.Nil(t, controller.handleSingleMaintainerStatus(owner, stale, cfID)) + require.Equal(t, uint64(100), cf.GetStatus().CheckpointTs) + + current := &heartbeatpb.MaintainerStatus{ + ChangefeedID: cfID.ToPB(), + CheckpointTs: 200, + State: heartbeatpb.ComponentState_Working, + BootstrapDone: true, + MaintainerEpoch: 2, + } + require.NotNil(t, controller.handleSingleMaintainerStatus(owner, current, cfID)) + require.Equal(t, uint64(200), cf.GetStatus().CheckpointTs) +} + +func TestHandleNonExistentChangefeedRemovesWithReportedEpoch(t *testing.T) { + mc := messaging.NewMockMessageCenter() + db := changefeed.NewChangefeedDB(1) + controller := &Controller{ + changefeedDB: db, + operatorController: operator.NewOperatorController( + &node.Info{ID: node.ID("coordinator")}, + db, + nil, + nil, + 10, + ), + messageCenter: mc, + } + cfID := common.NewChangeFeedIDWithName("cf", common.DefaultKeyspaceName) + + controller.handleNonExistentChangefeed(cfID, node.ID("owner"), &heartbeatpb.MaintainerStatus{ + ChangefeedID: cfID.ToPB(), + State: heartbeatpb.ComponentState_Working, + MaintainerEpoch: 7, + }) + + msg := <-mc.GetMessageChannel() + req := msg.Message[0].(*heartbeatpb.RemoveMaintainerRequest) + require.Equal(t, uint64(7), req.MaintainerEpoch) + require.True(t, req.Cascade) + require.True(t, req.Removed) +} + +func TestFinishBootstrapStopsStaleEpochMaintainerWithReportedEpoch(t *testing.T) { + testCases := []struct { + name string + progress config.Progress + expectRemoved bool + expectInDB bool + expectAbsent bool + expectStopped bool + }{ + { + name: "running", + progress: config.ProgressNone, + expectInDB: true, + expectAbsent: true, + }, + { + name: "removing", + progress: config.ProgressRemoving, + expectRemoved: true, + }, + { + name: "stopping", + progress: config.ProgressStopping, + expectInDB: true, + expectStopped: true, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + backend := mock_changefeed.NewMockBackend(ctrl) + mc := messaging.NewMockMessageCenter() + appcontext.SetService(appcontext.MessageCenter, mc) + appcontext.SetService(appcontext.SchemaStore, eventservice.NewMockSchemaStore()) + + nodeManager := watcher.NewNodeManager(nil, nil) + appcontext.SetService(watcher.NodeManagerName, nodeManager) + oldNode := node.ID("old-owner") + nodeManager.GetAliveNodes()[oldNode] = &node.Info{ID: oldNode} + + db := changefeed.NewChangefeedDB(1) + cfID := common.NewChangeFeedIDWithName(tc.name, common.DefaultKeyspaceName) + info := &config.ChangeFeedInfo{ + ChangefeedID: cfID, + Config: config.GetDefaultReplicaConfig(), + SinkURI: "blackhole://", + State: config.StateNormal, + Epoch: 2, + } + db.Init(map[common.ChangeFeedID]*changefeed.Changefeed{ + cfID: changefeed.NewChangefeed(cfID, info, 100, false), + }) + backend.EXPECT().GetAllChangefeeds(gomock.Any()).Return(map[common.ChangeFeedID]*changefeed.ChangefeedMetaWrapper{ + cfID: { + Info: info, + Status: &config.ChangeFeedStatus{CheckpointTs: 100, Progress: tc.progress}, + }, + }, nil).Times(1) + + self := &node.Info{ID: node.ID("coordinator")} + controller := &Controller{ + selfNode: self, + initialized: atomic.NewBool(false), + backend: backend, + changefeedDB: db, + operatorController: operator.NewOperatorController( + self, + db, + backend, + nil, + 10, + ), + nodeManager: nodeManager, + taskScheduler: threadpool.NewThreadPool(1), + scheduler: pkgscheduler.NewController(map[string]pkgscheduler.Scheduler{ + pkgscheduler.BasicScheduler: noopScheduler{}, + }), + messageCenter: mc, + } + t.Cleanup(controller.taskScheduler.Stop) + + controller.finishBootstrap(context.Background(), map[common.ChangeFeedID][]remoteMaintainer{ + cfID: {{ + nodeID: oldNode, + status: &heartbeatpb.MaintainerStatus{ + ChangefeedID: cfID.ToPB(), + State: heartbeatpb.ComponentState_Working, + CheckpointTs: 200, + BootstrapDone: true, + MaintainerEpoch: 1, + }, + }}, + }) + + if tc.expectInDB { + require.NotNil(t, db.GetByID(cfID)) + } else { + require.Nil(t, db.GetByID(cfID)) + } + if tc.expectAbsent { + require.Equal(t, 1, db.GetAbsentSize()) + } + if tc.expectStopped { + require.Equal(t, 1, db.GetStoppedSize()) + } + + op := controller.operatorController.GetOperator(cfID) + require.NotNil(t, op) + require.False(t, op.IsFinished()) + reqMsg := op.Schedule() + require.Equal(t, oldNode, reqMsg.To) + req := reqMsg.Message[0].(*heartbeatpb.RemoveMaintainerRequest) + require.Equal(t, uint64(1), req.MaintainerEpoch) + require.Equal(t, tc.expectRemoved, req.Removed) + }) + } +} + +func TestHandleBootstrapResponsesKeepsCurrentEpochAndStopsStaleDuplicate(t *testing.T) { + ctrl := gomock.NewController(t) + backend := mock_changefeed.NewMockBackend(ctrl) + mc := messaging.NewMockMessageCenter() + oldNode := node.ID("old-owner") + currentNode := node.ID("current-owner") + nodeManager := watcher.NewNodeManager(nil, nil) + nodeManager.GetAliveNodes()[oldNode] = &node.Info{ID: oldNode} + nodeManager.GetAliveNodes()[currentNode] = &node.Info{ID: currentNode} + appcontext.SetService(appcontext.MessageCenter, mc) + appcontext.SetService(appcontext.SchemaStore, eventservice.NewMockSchemaStore()) + appcontext.SetService(watcher.NodeManagerName, nodeManager) + + cfID := common.NewChangeFeedIDWithName("duplicate-epoch", common.DefaultKeyspaceName) + info := &config.ChangeFeedInfo{ + ChangefeedID: cfID, + Config: config.GetDefaultReplicaConfig(), + SinkURI: "blackhole://", + State: config.StateNormal, + Epoch: 2, + } + backend.EXPECT().GetAllChangefeeds(gomock.Any()).Return(map[common.ChangeFeedID]*changefeed.ChangefeedMetaWrapper{ + cfID: { + Info: info, + Status: &config.ChangeFeedStatus{CheckpointTs: 100}, + }, + }, nil).Times(1) + + db := changefeed.NewChangefeedDB(1) + self := &node.Info{ID: node.ID("coordinator")} + controller := &Controller{ + selfNode: self, + initialized: atomic.NewBool(false), + backend: backend, + changefeedDB: db, + operatorController: operator.NewOperatorController( + self, + db, + backend, + nil, + 10, + ), + nodeManager: nodeManager, + taskScheduler: threadpool.NewThreadPool(1), + scheduler: pkgscheduler.NewController(map[string]pkgscheduler.Scheduler{ + pkgscheduler.BasicScheduler: noopScheduler{}, + }), + messageCenter: mc, + bootstrapper: bootstrap.NewBootstrapper[heartbeatpb.CoordinatorBootstrapResponse]( + "test", + func(node.ID, string) *messaging.TargetMessage { return nil }, + ), + } + t.Cleanup(controller.taskScheduler.Stop) + + require.NotPanics(t, func() { + controller.handleBootstrapResponses(context.Background(), map[node.ID]*heartbeatpb.CoordinatorBootstrapResponse{ + oldNode: { + Statuses: []*heartbeatpb.MaintainerStatus{{ + ChangefeedID: cfID.ToPB(), + State: heartbeatpb.ComponentState_Working, + CheckpointTs: 150, + BootstrapDone: true, + MaintainerEpoch: 1, + }}, + }, + currentNode: { + Statuses: []*heartbeatpb.MaintainerStatus{{ + ChangefeedID: cfID.ToPB(), + State: heartbeatpb.ComponentState_Working, + CheckpointTs: 200, + BootstrapDone: true, + MaintainerEpoch: 2, + }}, + }, + }) + }) + + cf := db.GetByID(cfID) + require.NotNil(t, cf) + require.Equal(t, currentNode, cf.GetNodeID()) + require.Equal(t, uint64(200), cf.GetStatus().CheckpointTs) + + op := controller.operatorController.GetOperator(cfID) + require.NotNil(t, op) + reqMsg := op.Schedule() + require.Equal(t, oldNode, reqMsg.To) + req := reqMsg.Message[0].(*heartbeatpb.RemoveMaintainerRequest) + require.Equal(t, uint64(1), req.MaintainerEpoch) + require.False(t, req.Removed) +} + +>>>>>>> 5d5121dfa (coordinator: clean stale owner checkpoint metrics (#5491)) func TestResumeChangefeed(t *testing.T) { ctrl := gomock.NewController(t) backend := mock_changefeed.NewMockBackend(ctrl) diff --git a/pkg/metrics/changefeed.go b/pkg/metrics/changefeed.go index 03e9514a9c..224f6eedee 100644 --- a/pkg/metrics/changefeed.go +++ b/pkg/metrics/changefeed.go @@ -91,6 +91,19 @@ var ( }, []string{getKeyspaceLabel(), "changefeed"}) ) +func DeleteChangefeedCheckpointMetrics(keyspace, changefeed string) { + ChangefeedCheckpointTsGauge.DeleteLabelValues(keyspace, changefeed) + ChangefeedCheckpointTsLagGauge.DeleteLabelValues(keyspace, changefeed) +} + +func ResetOwnerChangefeedMetrics() { + ChangefeedStatusGauge.Reset() + ChangefeedErrorInfoGauge.Reset() + ChangefeedCheckpointTsGauge.Reset() + ChangefeedCheckpointTsLagGauge.Reset() + ChangefeedDownstreamInfoGauge.Reset() +} + func initChangefeedMetrics(registry *prometheus.Registry) { registry.MustRegister(MaintainerCheckpointTsGauge) registry.MustRegister(MaintainerCheckpointTsLagGauge) diff --git a/pkg/metrics/changefeed_test.go b/pkg/metrics/changefeed_test.go new file mode 100644 index 0000000000..6d1fade2ff --- /dev/null +++ b/pkg/metrics/changefeed_test.go @@ -0,0 +1,49 @@ +// Copyright 2026 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package metrics + +import ( + "testing" + + "github.com/prometheus/client_golang/prometheus/testutil" + "github.com/stretchr/testify/require" +) + +func TestResetOwnerChangefeedMetrics(t *testing.T) { + ResetOwnerChangefeedMetrics() + t.Cleanup(ResetOwnerChangefeedMetrics) + + keyspace := "default" + changefeed := "reset-owner-changefeed-metrics" + + ChangefeedStatusGauge.WithLabelValues(keyspace, changefeed).Set(1) + ChangefeedErrorInfoGauge.WithLabelValues(keyspace, changefeed, "failed", "1000", "CDC:ErrTest", "test").Set(1) + ChangefeedCheckpointTsGauge.WithLabelValues(keyspace, changefeed).Set(100) + ChangefeedCheckpointTsLagGauge.WithLabelValues(keyspace, changefeed).Set(10) + ChangefeedDownstreamInfoGauge.WithLabelValues(keyspace, changefeed, "mysql/tidb").Set(1) + + require.Equal(t, 1, testutil.CollectAndCount(ChangefeedStatusGauge)) + require.Equal(t, 1, testutil.CollectAndCount(ChangefeedErrorInfoGauge)) + require.Equal(t, 1, testutil.CollectAndCount(ChangefeedCheckpointTsGauge)) + require.Equal(t, 1, testutil.CollectAndCount(ChangefeedCheckpointTsLagGauge)) + require.Equal(t, 1, testutil.CollectAndCount(ChangefeedDownstreamInfoGauge)) + + ResetOwnerChangefeedMetrics() + + require.Equal(t, 0, testutil.CollectAndCount(ChangefeedStatusGauge)) + require.Equal(t, 0, testutil.CollectAndCount(ChangefeedErrorInfoGauge)) + require.Equal(t, 0, testutil.CollectAndCount(ChangefeedCheckpointTsGauge)) + require.Equal(t, 0, testutil.CollectAndCount(ChangefeedCheckpointTsLagGauge)) + require.Equal(t, 0, testutil.CollectAndCount(ChangefeedDownstreamInfoGauge)) +} From e3ca29c0261b1b5b272393182c492a8c1306af96 Mon Sep 17 00:00:00 2001 From: dongmen <414110582@qq.com> Date: Tue, 28 Jul 2026 15:10:29 +0800 Subject: [PATCH 2/3] coordinator: resolve v8.5.7 cherry-pick conflicts --- coordinator/controller.go | 37 --- coordinator/controller_test.go | 526 --------------------------------- pkg/metrics/changefeed.go | 2 - pkg/metrics/changefeed_test.go | 6 - 4 files changed, 571 deletions(-) diff --git a/coordinator/controller.go b/coordinator/controller.go index 3d8f9b3ff0..a2724c3936 100644 --- a/coordinator/controller.go +++ b/coordinator/controller.go @@ -186,15 +186,7 @@ func NewController( func (c *Controller) collectMetrics(ctx context.Context) error { ticker := time.NewTicker(5 * time.Second) defer ticker.Stop() -<<<<<<< HEAD -======= defer metrics.ResetOwnerChangefeedMetrics() - - // changefeedDownstreamTypeCache is used to cleanup the previous downstream type - // label value when a changefeed's sink-uri is updated. - changefeedDownstreamTypeCache := make(map[common.ChangeFeedDisplayName]string) - errorMetricLabels := make(map[common.ChangeFeedID]changefeedErrorMetricLabels) ->>>>>>> 5d5121dfa (coordinator: clean stale owner checkpoint metrics (#5491)) for { select { case <-ctx.Done(): @@ -221,35 +213,6 @@ func (c *Controller) collectMetrics(ctx context.Context) error { ) { return } - -<<<<<<< HEAD - pdPhysicalTime := oracle.GetPhysical(c.pdClock.CurrentTime()) - phyCkpTs := oracle.ExtractPhysical(cf.GetLastSavedCheckPointTs()) - lag := float64(pdPhysicalTime-phyCkpTs) / 1e3 - metrics.ChangefeedCheckpointTsGauge.WithLabelValues(keyspace, name).Set(float64(phyCkpTs)) - metrics.ChangefeedCheckpointTsLagGauge.WithLabelValues(keyspace, name).Set(lag) -======= - // sync changefeed error metrics - currentChangefeeds[cf.ID] = struct{}{} - oldLabels, exists := errorMetricLabels[cf.ID] - newLabels, hasError := getChangefeedErrorMetricLabels(cf.GetInfo()) - // If the error state has not changed, do nothing. - if exists && hasError && oldLabels == newLabels { - return - } - // If there was an old metric, delete it, as the state has changed. - if exists { - metrics.ChangefeedErrorInfoGauge.DeleteLabelValues(oldLabels.labelValues()...) - } - if hasError { - // An error exists (either new or changed). Set the new metric and update cache. - metrics.ChangefeedErrorInfoGauge.WithLabelValues(newLabels.labelValues()...).Set(1) - errorMetricLabels[cf.ID] = newLabels - } else { - // The error has disappeared, remove from cache. - delete(errorMetricLabels, cf.ID) - } ->>>>>>> 5d5121dfa (coordinator: clean stale owner checkpoint metrics (#5491)) }) } } diff --git a/coordinator/controller_test.go b/coordinator/controller_test.go index 0e43f7f339..2c0a255888 100644 --- a/coordinator/controller_test.go +++ b/coordinator/controller_test.go @@ -31,28 +31,12 @@ import ( "github.com/pingcap/ticdc/pkg/metrics" "github.com/pingcap/ticdc/pkg/node" "github.com/pingcap/ticdc/server/watcher" -<<<<<<< HEAD -======= - "github.com/pingcap/ticdc/utils/threadpool" "github.com/prometheus/client_golang/prometheus/testutil" ->>>>>>> 5d5121dfa (coordinator: clean stale owner checkpoint metrics (#5491)) "github.com/stretchr/testify/require" "github.com/tikv/client-go/v2/oracle" "go.uber.org/atomic" ) -<<<<<<< HEAD -======= -type noopScheduler struct{} - -func (noopScheduler) Execute() time.Time { - return time.Now().Add(time.Hour) -} - -func (noopScheduler) Name() string { - return pkgscheduler.BasicScheduler -} - func TestUpdateChangefeedCheckpointMetricsDeletesFinishedLabels(t *testing.T) { metrics.ResetOwnerChangefeedMetrics() t.Cleanup(metrics.ResetOwnerChangefeedMetrics) @@ -83,516 +67,6 @@ func TestUpdateChangefeedCheckpointMetricsDeletesFinishedLabels(t *testing.T) { require.Equal(t, 0, testutil.CollectAndCount(metrics.ChangefeedCheckpointTsLagGauge)) } -func TestOnPeriodTaskAdvanceLiveness(t *testing.T) { - newController := func(t *testing.T) (*Controller, chan *messaging.TargetMessage, *changefeed.ChangefeedDB, node.ID) { - t.Helper() - - ctrl := gomock.NewController(t) - t.Cleanup(ctrl.Finish) - - backend := mock_changefeed.NewMockBackend(ctrl) - changefeedDB := changefeed.NewChangefeedDB(1216) - self := node.NewInfo("localhost:8300", "") - target := node.NewInfo("localhost:8301", "") - nodeManager := watcher.NewNodeManager(nil, nil) - nodeManager.GetAliveNodes()[self.ID] = self - nodeManager.GetAliveNodes()[target.ID] = target - - mc := messaging.NewMockMessageCenter() - appcontext.SetService(appcontext.MessageCenter, mc) - appcontext.SetService(watcher.NodeManagerName, nodeManager) - - return &Controller{ - changefeedDB: changefeedDB, - operatorController: operator.NewOperatorController( - self, changefeedDB, backend, nil, 10, - ), - nodeManager: nodeManager, - initialized: atomic.NewBool(true), - drainController: drain.NewController(mc), - pdClient: newDrainTestPDClient(), - bootstrapper: bootstrap.NewBootstrapper[heartbeatpb.CoordinatorBootstrapResponse]( - "test", - func(node.ID, string) *messaging.TargetMessage { return nil }, - ), - messageCenter: mc, - }, mc.GetMessageChannel(), changefeedDB, target.ID - } - - t.Run("skip stopping before bootstrap completion", func(t *testing.T) { - controller, messageCh, _, targetNodeID := newController(t) - controller.initialized.Store(false) - controller.drainController.ObserveHeartbeat(targetNodeID, &heartbeatpb.NodeHeartbeat{ - NodeEpoch: 1, - Liveness: heartbeatpb.NodeLiveness_DRAINING, - }) - - controller.onPeriodTask() - - select { - case msg := <-messageCh: - t.Fatalf("unexpected liveness command sent before bootstrap completion: %v", msg.Type) - default: - } - }) - - t.Run("skip stopping without active drain session", func(t *testing.T) { - controller, messageCh, _, targetNodeID := newController(t) - controller.drainController.ObserveHeartbeat(targetNodeID, &heartbeatpb.NodeHeartbeat{ - NodeEpoch: 1, - Liveness: heartbeatpb.NodeLiveness_DRAINING, - }) - - controller.onPeriodTask() - - select { - case msg := <-messageCh: - t.Fatalf("unexpected liveness command sent without active session: %v", msg.Type) - default: - } - }) - - t.Run("skip stopping when active drain session is not ready", func(t *testing.T) { - controller, messageCh, changefeedDB, targetNodeID := newController(t) - cfID := common.NewChangeFeedIDWithName("test", common.DefaultKeyspaceName) - cf := changefeed.NewChangefeed(cfID, &config.ChangeFeedInfo{ - ChangefeedID: cfID, - Config: config.GetDefaultReplicaConfig(), - SinkURI: "mysql://127.0.0.1:3306", - State: config.StateNormal, - }, 1, true) - changefeedDB.AddReplicatingMaintainer(cf, targetNodeID) - - _, err := controller.ensureDispatcherDrainTarget(context.Background(), targetNodeID) - require.NoError(t, err) - - controller.drainController.ObserveHeartbeat(targetNodeID, &heartbeatpb.NodeHeartbeat{ - NodeEpoch: 1, - Liveness: heartbeatpb.NodeLiveness_DRAINING, - }) - controller.onPeriodTask() - - for { - select { - case msg := <-messageCh: - if msg.Type == messaging.TypeSetNodeLivenessRequest { - t.Fatalf("unexpected liveness command sent before readiness: %v", msg.Type) - } - default: - return - } - } - }) - - t.Run("send stopping only for active drain session target", func(t *testing.T) { - controller, messageCh, _, targetNodeID := newController(t) - _, err := controller.ensureDispatcherDrainTarget(context.Background(), targetNodeID) - require.NoError(t, err) - - controller.drainSessionMu.Lock() - controller.drainSession.dirty = false - controller.drainSession.lastSent = time.Now() - controller.drainSessionMu.Unlock() - - controller.drainController.ObserveHeartbeat(targetNodeID, &heartbeatpb.NodeHeartbeat{ - NodeEpoch: 1, - Liveness: heartbeatpb.NodeLiveness_DRAINING, - }) - - controller.onPeriodTask() - - foundStop := false - for { - select { - case msg := <-messageCh: - if msg.Type != messaging.TypeSetNodeLivenessRequest { - continue - } - require.Equal(t, targetNodeID, msg.To) - req := msg.Message[0].(*heartbeatpb.SetNodeLivenessRequest) - if req.Target == heartbeatpb.NodeLiveness_STOPPING { - require.Equal(t, uint64(1), req.NodeEpoch) - foundStop = true - } - default: - require.True(t, foundStop) - return - } - } - }) -} - -func TestMaintainerHeartbeatAdmissionRequiresInitializedSender(t *testing.T) { - mc := messaging.NewMockMessageCenter() - appcontext.SetService(appcontext.MessageCenter, mc) - - nodeManager := watcher.NewNodeManager(nil, nil) - appcontext.SetService(watcher.NodeManagerName, nodeManager) - - owner := node.ID("owner") - late := node.ID("late") - nodeManager.GetAliveNodes()[owner] = &node.Info{ID: owner} - nodeManager.GetAliveNodes()[late] = &node.Info{ID: late} - - db := changefeed.NewChangefeedDB(1) - cfID := common.NewChangeFeedIDWithName("cf", common.DefaultKeyspaceName) - cf := changefeed.NewChangefeed(cfID, &config.ChangeFeedInfo{ - ChangefeedID: cfID, - Config: config.GetDefaultReplicaConfig(), - SinkURI: "blackhole://", - State: config.StateNormal, - }, 100, false) - db.AddReplicatingMaintainer(cf, late) - - controller := &Controller{ - initialized: atomic.NewBool(true), - changefeedDB: db, - operatorController: operator.NewOperatorController( - &node.Info{ID: node.ID("coordinator")}, - db, - nil, - nil, - 10, - ), - bootstrapper: bootstrap.NewBootstrapper[heartbeatpb.CoordinatorBootstrapResponse]( - "test", - func(id node.ID, _ string) *messaging.TargetMessage { - return messaging.NewSingleTargetMessage( - id, - messaging.MaintainerManagerTopic, - &heartbeatpb.CoordinatorBootstrapRequest{}, - ) - }, - ), - } - - controller.bootstrapper.HandleNodesChange(nodeManager.GetAliveNodes()) - controller.bootstrapper.HandleBootstrapResponse(owner, &heartbeatpb.CoordinatorBootstrapResponse{}) - require.False(t, controller.bootstrapper.AllNodesReady()) - - ignored := &heartbeatpb.MaintainerHeartbeat{ - Statuses: []*heartbeatpb.MaintainerStatus{{ - ChangefeedID: cfID.ToPB(), - CheckpointTs: 200, - State: heartbeatpb.ComponentState_Working, - BootstrapDone: true, - }}, - } - controller.onMessage(context.Background(), &messaging.TargetMessage{ - From: late, - Topic: messaging.CoordinatorTopic, - Type: messaging.TypeMaintainerHeartbeatRequest, - Message: []messaging.IOTypeT{ignored}, - }) - require.Equal(t, uint64(100), cf.GetStatus().CheckpointTs) - - controller.bootstrapper.HandleBootstrapResponse(late, &heartbeatpb.CoordinatorBootstrapResponse{}) - accepted := &heartbeatpb.MaintainerHeartbeat{ - Statuses: []*heartbeatpb.MaintainerStatus{{ - ChangefeedID: cfID.ToPB(), - CheckpointTs: 200, - State: heartbeatpb.ComponentState_Working, - BootstrapDone: true, - }}, - } - controller.onMessage(context.Background(), &messaging.TargetMessage{ - From: late, - Topic: messaging.CoordinatorTopic, - Type: messaging.TypeMaintainerHeartbeatRequest, - Message: []messaging.IOTypeT{accepted}, - }) - require.Equal(t, uint64(200), cf.GetStatus().CheckpointTs) -} - -func TestMaintainerHeartbeatAdmissionDropsStaleMaintainerEpoch(t *testing.T) { - appcontext.SetService(appcontext.MessageCenter, messaging.NewMockMessageCenter()) - appcontext.SetService(watcher.NodeManagerName, watcher.NewNodeManager(nil, nil)) - - db := changefeed.NewChangefeedDB(1) - cfID := common.NewChangeFeedIDWithName("cf", common.DefaultKeyspaceName) - owner := node.ID("owner") - cf := changefeed.NewChangefeed(cfID, &config.ChangeFeedInfo{ - ChangefeedID: cfID, - Config: config.GetDefaultReplicaConfig(), - SinkURI: "blackhole://", - State: config.StateNormal, - Epoch: 2, - }, 100, false) - db.AddReplicatingMaintainer(cf, owner) - - controller := &Controller{ - changefeedDB: db, - operatorController: operator.NewOperatorController( - &node.Info{ID: node.ID("coordinator")}, - db, - nil, - nil, - 10, - ), - } - - stale := &heartbeatpb.MaintainerStatus{ - ChangefeedID: cfID.ToPB(), - CheckpointTs: 200, - State: heartbeatpb.ComponentState_Working, - BootstrapDone: true, - MaintainerEpoch: 1, - } - require.Nil(t, controller.handleSingleMaintainerStatus(owner, stale, cfID)) - require.Equal(t, uint64(100), cf.GetStatus().CheckpointTs) - - current := &heartbeatpb.MaintainerStatus{ - ChangefeedID: cfID.ToPB(), - CheckpointTs: 200, - State: heartbeatpb.ComponentState_Working, - BootstrapDone: true, - MaintainerEpoch: 2, - } - require.NotNil(t, controller.handleSingleMaintainerStatus(owner, current, cfID)) - require.Equal(t, uint64(200), cf.GetStatus().CheckpointTs) -} - -func TestHandleNonExistentChangefeedRemovesWithReportedEpoch(t *testing.T) { - mc := messaging.NewMockMessageCenter() - db := changefeed.NewChangefeedDB(1) - controller := &Controller{ - changefeedDB: db, - operatorController: operator.NewOperatorController( - &node.Info{ID: node.ID("coordinator")}, - db, - nil, - nil, - 10, - ), - messageCenter: mc, - } - cfID := common.NewChangeFeedIDWithName("cf", common.DefaultKeyspaceName) - - controller.handleNonExistentChangefeed(cfID, node.ID("owner"), &heartbeatpb.MaintainerStatus{ - ChangefeedID: cfID.ToPB(), - State: heartbeatpb.ComponentState_Working, - MaintainerEpoch: 7, - }) - - msg := <-mc.GetMessageChannel() - req := msg.Message[0].(*heartbeatpb.RemoveMaintainerRequest) - require.Equal(t, uint64(7), req.MaintainerEpoch) - require.True(t, req.Cascade) - require.True(t, req.Removed) -} - -func TestFinishBootstrapStopsStaleEpochMaintainerWithReportedEpoch(t *testing.T) { - testCases := []struct { - name string - progress config.Progress - expectRemoved bool - expectInDB bool - expectAbsent bool - expectStopped bool - }{ - { - name: "running", - progress: config.ProgressNone, - expectInDB: true, - expectAbsent: true, - }, - { - name: "removing", - progress: config.ProgressRemoving, - expectRemoved: true, - }, - { - name: "stopping", - progress: config.ProgressStopping, - expectInDB: true, - expectStopped: true, - }, - } - - for _, tc := range testCases { - t.Run(tc.name, func(t *testing.T) { - ctrl := gomock.NewController(t) - backend := mock_changefeed.NewMockBackend(ctrl) - mc := messaging.NewMockMessageCenter() - appcontext.SetService(appcontext.MessageCenter, mc) - appcontext.SetService(appcontext.SchemaStore, eventservice.NewMockSchemaStore()) - - nodeManager := watcher.NewNodeManager(nil, nil) - appcontext.SetService(watcher.NodeManagerName, nodeManager) - oldNode := node.ID("old-owner") - nodeManager.GetAliveNodes()[oldNode] = &node.Info{ID: oldNode} - - db := changefeed.NewChangefeedDB(1) - cfID := common.NewChangeFeedIDWithName(tc.name, common.DefaultKeyspaceName) - info := &config.ChangeFeedInfo{ - ChangefeedID: cfID, - Config: config.GetDefaultReplicaConfig(), - SinkURI: "blackhole://", - State: config.StateNormal, - Epoch: 2, - } - db.Init(map[common.ChangeFeedID]*changefeed.Changefeed{ - cfID: changefeed.NewChangefeed(cfID, info, 100, false), - }) - backend.EXPECT().GetAllChangefeeds(gomock.Any()).Return(map[common.ChangeFeedID]*changefeed.ChangefeedMetaWrapper{ - cfID: { - Info: info, - Status: &config.ChangeFeedStatus{CheckpointTs: 100, Progress: tc.progress}, - }, - }, nil).Times(1) - - self := &node.Info{ID: node.ID("coordinator")} - controller := &Controller{ - selfNode: self, - initialized: atomic.NewBool(false), - backend: backend, - changefeedDB: db, - operatorController: operator.NewOperatorController( - self, - db, - backend, - nil, - 10, - ), - nodeManager: nodeManager, - taskScheduler: threadpool.NewThreadPool(1), - scheduler: pkgscheduler.NewController(map[string]pkgscheduler.Scheduler{ - pkgscheduler.BasicScheduler: noopScheduler{}, - }), - messageCenter: mc, - } - t.Cleanup(controller.taskScheduler.Stop) - - controller.finishBootstrap(context.Background(), map[common.ChangeFeedID][]remoteMaintainer{ - cfID: {{ - nodeID: oldNode, - status: &heartbeatpb.MaintainerStatus{ - ChangefeedID: cfID.ToPB(), - State: heartbeatpb.ComponentState_Working, - CheckpointTs: 200, - BootstrapDone: true, - MaintainerEpoch: 1, - }, - }}, - }) - - if tc.expectInDB { - require.NotNil(t, db.GetByID(cfID)) - } else { - require.Nil(t, db.GetByID(cfID)) - } - if tc.expectAbsent { - require.Equal(t, 1, db.GetAbsentSize()) - } - if tc.expectStopped { - require.Equal(t, 1, db.GetStoppedSize()) - } - - op := controller.operatorController.GetOperator(cfID) - require.NotNil(t, op) - require.False(t, op.IsFinished()) - reqMsg := op.Schedule() - require.Equal(t, oldNode, reqMsg.To) - req := reqMsg.Message[0].(*heartbeatpb.RemoveMaintainerRequest) - require.Equal(t, uint64(1), req.MaintainerEpoch) - require.Equal(t, tc.expectRemoved, req.Removed) - }) - } -} - -func TestHandleBootstrapResponsesKeepsCurrentEpochAndStopsStaleDuplicate(t *testing.T) { - ctrl := gomock.NewController(t) - backend := mock_changefeed.NewMockBackend(ctrl) - mc := messaging.NewMockMessageCenter() - oldNode := node.ID("old-owner") - currentNode := node.ID("current-owner") - nodeManager := watcher.NewNodeManager(nil, nil) - nodeManager.GetAliveNodes()[oldNode] = &node.Info{ID: oldNode} - nodeManager.GetAliveNodes()[currentNode] = &node.Info{ID: currentNode} - appcontext.SetService(appcontext.MessageCenter, mc) - appcontext.SetService(appcontext.SchemaStore, eventservice.NewMockSchemaStore()) - appcontext.SetService(watcher.NodeManagerName, nodeManager) - - cfID := common.NewChangeFeedIDWithName("duplicate-epoch", common.DefaultKeyspaceName) - info := &config.ChangeFeedInfo{ - ChangefeedID: cfID, - Config: config.GetDefaultReplicaConfig(), - SinkURI: "blackhole://", - State: config.StateNormal, - Epoch: 2, - } - backend.EXPECT().GetAllChangefeeds(gomock.Any()).Return(map[common.ChangeFeedID]*changefeed.ChangefeedMetaWrapper{ - cfID: { - Info: info, - Status: &config.ChangeFeedStatus{CheckpointTs: 100}, - }, - }, nil).Times(1) - - db := changefeed.NewChangefeedDB(1) - self := &node.Info{ID: node.ID("coordinator")} - controller := &Controller{ - selfNode: self, - initialized: atomic.NewBool(false), - backend: backend, - changefeedDB: db, - operatorController: operator.NewOperatorController( - self, - db, - backend, - nil, - 10, - ), - nodeManager: nodeManager, - taskScheduler: threadpool.NewThreadPool(1), - scheduler: pkgscheduler.NewController(map[string]pkgscheduler.Scheduler{ - pkgscheduler.BasicScheduler: noopScheduler{}, - }), - messageCenter: mc, - bootstrapper: bootstrap.NewBootstrapper[heartbeatpb.CoordinatorBootstrapResponse]( - "test", - func(node.ID, string) *messaging.TargetMessage { return nil }, - ), - } - t.Cleanup(controller.taskScheduler.Stop) - - require.NotPanics(t, func() { - controller.handleBootstrapResponses(context.Background(), map[node.ID]*heartbeatpb.CoordinatorBootstrapResponse{ - oldNode: { - Statuses: []*heartbeatpb.MaintainerStatus{{ - ChangefeedID: cfID.ToPB(), - State: heartbeatpb.ComponentState_Working, - CheckpointTs: 150, - BootstrapDone: true, - MaintainerEpoch: 1, - }}, - }, - currentNode: { - Statuses: []*heartbeatpb.MaintainerStatus{{ - ChangefeedID: cfID.ToPB(), - State: heartbeatpb.ComponentState_Working, - CheckpointTs: 200, - BootstrapDone: true, - MaintainerEpoch: 2, - }}, - }, - }) - }) - - cf := db.GetByID(cfID) - require.NotNil(t, cf) - require.Equal(t, currentNode, cf.GetNodeID()) - require.Equal(t, uint64(200), cf.GetStatus().CheckpointTs) - - op := controller.operatorController.GetOperator(cfID) - require.NotNil(t, op) - reqMsg := op.Schedule() - require.Equal(t, oldNode, reqMsg.To) - req := reqMsg.Message[0].(*heartbeatpb.RemoveMaintainerRequest) - require.Equal(t, uint64(1), req.MaintainerEpoch) - require.False(t, req.Removed) -} - ->>>>>>> 5d5121dfa (coordinator: clean stale owner checkpoint metrics (#5491)) func TestResumeChangefeed(t *testing.T) { ctrl := gomock.NewController(t) backend := mock_changefeed.NewMockBackend(ctrl) diff --git a/pkg/metrics/changefeed.go b/pkg/metrics/changefeed.go index 224f6eedee..ac2d8addba 100644 --- a/pkg/metrics/changefeed.go +++ b/pkg/metrics/changefeed.go @@ -98,10 +98,8 @@ func DeleteChangefeedCheckpointMetrics(keyspace, changefeed string) { func ResetOwnerChangefeedMetrics() { ChangefeedStatusGauge.Reset() - ChangefeedErrorInfoGauge.Reset() ChangefeedCheckpointTsGauge.Reset() ChangefeedCheckpointTsLagGauge.Reset() - ChangefeedDownstreamInfoGauge.Reset() } func initChangefeedMetrics(registry *prometheus.Registry) { diff --git a/pkg/metrics/changefeed_test.go b/pkg/metrics/changefeed_test.go index 6d1fade2ff..400e096eb9 100644 --- a/pkg/metrics/changefeed_test.go +++ b/pkg/metrics/changefeed_test.go @@ -28,22 +28,16 @@ func TestResetOwnerChangefeedMetrics(t *testing.T) { changefeed := "reset-owner-changefeed-metrics" ChangefeedStatusGauge.WithLabelValues(keyspace, changefeed).Set(1) - ChangefeedErrorInfoGauge.WithLabelValues(keyspace, changefeed, "failed", "1000", "CDC:ErrTest", "test").Set(1) ChangefeedCheckpointTsGauge.WithLabelValues(keyspace, changefeed).Set(100) ChangefeedCheckpointTsLagGauge.WithLabelValues(keyspace, changefeed).Set(10) - ChangefeedDownstreamInfoGauge.WithLabelValues(keyspace, changefeed, "mysql/tidb").Set(1) require.Equal(t, 1, testutil.CollectAndCount(ChangefeedStatusGauge)) - require.Equal(t, 1, testutil.CollectAndCount(ChangefeedErrorInfoGauge)) require.Equal(t, 1, testutil.CollectAndCount(ChangefeedCheckpointTsGauge)) require.Equal(t, 1, testutil.CollectAndCount(ChangefeedCheckpointTsLagGauge)) - require.Equal(t, 1, testutil.CollectAndCount(ChangefeedDownstreamInfoGauge)) ResetOwnerChangefeedMetrics() require.Equal(t, 0, testutil.CollectAndCount(ChangefeedStatusGauge)) - require.Equal(t, 0, testutil.CollectAndCount(ChangefeedErrorInfoGauge)) require.Equal(t, 0, testutil.CollectAndCount(ChangefeedCheckpointTsGauge)) require.Equal(t, 0, testutil.CollectAndCount(ChangefeedCheckpointTsLagGauge)) - require.Equal(t, 0, testutil.CollectAndCount(ChangefeedDownstreamInfoGauge)) } From 51ce701fc20ea98e604f1cdc7241fff4694ad7a0 Mon Sep 17 00:00:00 2001 From: Ti Chi Robot Date: Sat, 25 Jul 2026 16:44:17 +0800 Subject: [PATCH 3/3] common: update TiDB TableInfo shared schema guard (#5652) (#5741) close pingcap/ticdc#5740 --- pkg/common/table_info_shared_schema_guard_test.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/pkg/common/table_info_shared_schema_guard_test.go b/pkg/common/table_info_shared_schema_guard_test.go index d1f44af88e..705bd35df4 100644 --- a/pkg/common/table_info_shared_schema_guard_test.go +++ b/pkg/common/table_info_shared_schema_guard_test.go @@ -83,7 +83,9 @@ func TestLatestTiDBTableInfoSharedSchemaGuard(t *testing.T) { "Partition", "Compression", "View", "Sequence", "Lock", "Version", "TiFlashReplica", "IsColumnar", "TempTableType", "TableCacheStatusType", "PlacementPolicyRef", "StatsOptions", "ExchangePartitionInfo", "TTLInfo", "IsActiveActive", "SoftdeleteInfo", "Affinity", - "Revision", "DBID", "Mode", + "Revision", "DBID", + // These table-level storage settings do not affect the shared column schema. + "EngineAttribute", "StorageClassTier", "StorageClassTransitions", "Mode", }, }, {