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 25d59cda5b..b7df02de80 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 5f01c837df..f7d51da281 100644 --- a/coordinator/controller.go +++ b/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 { diff --git a/coordinator/controller_test.go b/coordinator/controller_test.go index dace08fe7e..8283b8b84b 100644 --- a/coordinator/controller_test.go +++ b/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() diff --git a/pkg/metrics/changefeed.go b/pkg/metrics/changefeed.go index c8c94ce3ba..73be97af75 100644 --- a/pkg/metrics/changefeed.go +++ b/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) diff --git a/pkg/metrics/changefeed_test.go b/pkg/metrics/changefeed_test.go new file mode 100644 index 0000000000..38fdf1e758 --- /dev/null +++ b/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)) +}