coordinator: clean stale owner checkpoint metrics (#5491)

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
 }

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)}

coordinator/controller.go

@@ -222,6 +222,8 @@ func NewController(
 func (c *Controller) collectMetrics(ctx context.Context) error {
 	ticker := time.NewTicker(5 * time.Second)
 	defer ticker.Stop()
+	defer metrics.ResetOwnerChangefeedMetrics()
+
 	errorMetricLabels := make(map[common.ChangeFeedID]changefeedErrorMetricLabels)
 	for {
 		select {
@@ -242,17 +244,16 @@ 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
 				}
 
-				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]
@@ -285,6 +286,27 @@ func (c *Controller) collectMetrics(ctx context.Context) error {
 	}
 }
 
+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 {

coordinator/controller_test.go

@@ -30,12 +30,45 @@ 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"
+	"github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/stretchr/testify/require"
+	"github.com/tikv/client-go/v2/oracle"
 	"go.uber.org/atomic"
 )
 
+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()

pkg/metrics/changefeed.go

@@ -111,6 +111,18 @@ 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()
+}
+
 func initChangefeedMetrics(registry *prometheus.Registry) {
 	registry.MustRegister(MaintainerCheckpointTsGauge)
 	registry.MustRegister(MaintainerCheckpointTsLagGauge)

pkg/metrics/changefeed_test.go

@@ -0,0 +1,46 @@
+// 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)
+
+	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))
+
+	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))
+}