From d65a0d2b9589107ead3d0ebeda0c8f254d43736c 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/2] 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 | 44 +++++++++++++---- coordinator/controller_test.go | 50 ++++++++++++++++++++ pkg/metrics/changefeed.go | 13 +++++ pkg/metrics/changefeed_test.go | 49 +++++++++++++++++++ 6 files changed, 170 insertions(+), 9 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 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..37ef1c4376 100644 --- a/coordinator/controller.go +++ b/coordinator/controller.go @@ -222,6 +222,14 @@ 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) +>>>>>>> 5d5121dfa (coordinator: clean stale owner checkpoint metrics (#5491)) errorMetricLabels := make(map[common.ChangeFeedID]changefeedErrorMetricLabels) for { select { @@ -242,17 +250,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 +292,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..9f03143450 100644 --- a/coordinator/controller_test.go +++ b/coordinator/controller_test.go @@ -30,12 +30,62 @@ 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)) +} + +>>>>>>> 5d5121dfa (coordinator: clean stale owner checkpoint metrics (#5491)) 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..ed4c299247 100644 --- a/pkg/metrics/changefeed.go +++ b/pkg/metrics/changefeed.go @@ -111,6 +111,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 55ef4302e6af0dfca068317e3512ea7b610609ee Mon Sep 17 00:00:00 2001 From: dongmen <414110582@qq.com> Date: Tue, 28 Jul 2026 08:03:02 +0800 Subject: [PATCH 2/2] coordinator: resolve release-8.5 cherry-pick conflicts --- coordinator/controller.go | 6 ------ coordinator/controller_test.go | 17 ----------------- pkg/metrics/changefeed.go | 1 - pkg/metrics/changefeed_test.go | 3 --- 4 files changed, 27 deletions(-) diff --git a/coordinator/controller.go b/coordinator/controller.go index 37ef1c4376..f7d51da281 100644 --- a/coordinator/controller.go +++ b/coordinator/controller.go @@ -222,14 +222,8 @@ 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) ->>>>>>> 5d5121dfa (coordinator: clean stale owner checkpoint metrics (#5491)) errorMetricLabels := make(map[common.ChangeFeedID]changefeedErrorMetricLabels) for { select { diff --git a/coordinator/controller_test.go b/coordinator/controller_test.go index 9f03143450..8283b8b84b 100644 --- a/coordinator/controller_test.go +++ b/coordinator/controller_test.go @@ -33,28 +33,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) @@ -85,7 +69,6 @@ func TestUpdateChangefeedCheckpointMetricsDeletesFinishedLabels(t *testing.T) { require.Equal(t, 0, testutil.CollectAndCount(metrics.ChangefeedCheckpointTsLagGauge)) } ->>>>>>> 5d5121dfa (coordinator: clean stale owner checkpoint metrics (#5491)) 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 ed4c299247..73be97af75 100644 --- a/pkg/metrics/changefeed.go +++ b/pkg/metrics/changefeed.go @@ -121,7 +121,6 @@ func ResetOwnerChangefeedMetrics() { 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..38fdf1e758 100644 --- a/pkg/metrics/changefeed_test.go +++ b/pkg/metrics/changefeed_test.go @@ -31,13 +31,11 @@ func TestResetOwnerChangefeedMetrics(t *testing.T) { 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() @@ -45,5 +43,4 @@ func TestResetOwnerChangefeedMetrics(t *testing.T) { 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)) }