Skip to content

Commit c8d4d68

Browse files
coordinator: clean stale owner checkpoint metrics (#5491) (#5793)
* This is an automated cherry-pick of #5491 Signed-off-by: ti-chi-bot <ti-community-prow-bot@tidb.io> * coordinator: resolve v8.5.7 cherry-pick conflicts * common: update TiDB TableInfo shared schema guard (#5652) (#5741) close #5740 --------- Signed-off-by: ti-chi-bot <ti-community-prow-bot@tidb.io> Co-authored-by: dongmen <20351731+asddongmen@users.noreply.github.com> Co-authored-by: dongmen <414110582@qq.com>
1 parent e7e1f61 commit c8d4d68

7 files changed

Lines changed: 141 additions & 10 deletions

File tree

coordinator/changefeed/changefeed_db.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -141,7 +141,7 @@ func (db *ChangefeedDB) StopByChangefeedID(cfID common.ChangeFeedID, remove bool
141141
}
142142

143143
metrics.ChangefeedStatusGauge.DeleteLabelValues(cfID.Keyspace(), cfID.Name())
144-
metrics.ChangefeedCheckpointTsLagGauge.DeleteLabelValues(cfID.Keyspace(), cfID.Name())
144+
metrics.DeleteChangefeedCheckpointMetrics(cfID.Keyspace(), cfID.Name())
145145

146146
return nodeID
147147
}

coordinator/changefeed/changefeed_db_test.go

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,9 @@ import (
2121
"github.com/pingcap/ticdc/pkg/common"
2222
"github.com/pingcap/ticdc/pkg/config"
2323
"github.com/pingcap/ticdc/pkg/errors"
24+
"github.com/pingcap/ticdc/pkg/metrics"
2425
"github.com/pingcap/ticdc/pkg/node"
26+
"github.com/prometheus/client_golang/prometheus/testutil"
2527
"github.com/stretchr/testify/require"
2628
"go.uber.org/atomic"
2729
)
@@ -133,6 +135,25 @@ func TestRemoveChangefeed(t *testing.T) {
133135
require.False(t, ok)
134136
}
135137

138+
func TestStopByChangefeedIDDeletesCheckpointMetrics(t *testing.T) {
139+
metrics.ResetOwnerChangefeedMetrics()
140+
t.Cleanup(metrics.ResetOwnerChangefeedMetrics)
141+
142+
db := NewChangefeedDB(1216)
143+
cf := &Changefeed{ID: common.NewChangeFeedIDWithName("test-metrics", common.DefaultKeyspaceName)}
144+
db.AddReplicatingMaintainer(cf, "node1")
145+
146+
metrics.ChangefeedCheckpointTsGauge.WithLabelValues(cf.ID.Keyspace(), cf.ID.Name()).Set(100)
147+
metrics.ChangefeedCheckpointTsLagGauge.WithLabelValues(cf.ID.Keyspace(), cf.ID.Name()).Set(10)
148+
require.Equal(t, 1, testutil.CollectAndCount(metrics.ChangefeedCheckpointTsGauge))
149+
require.Equal(t, 1, testutil.CollectAndCount(metrics.ChangefeedCheckpointTsLagGauge))
150+
151+
require.Equal(t, node.ID("node1"), db.StopByChangefeedID(cf.ID, true))
152+
153+
require.Equal(t, 0, testutil.CollectAndCount(metrics.ChangefeedCheckpointTsGauge))
154+
require.Equal(t, 0, testutil.CollectAndCount(metrics.ChangefeedCheckpointTsLagGauge))
155+
}
156+
136157
func TestGetByID(t *testing.T) {
137158
db := NewChangefeedDB(1216)
138159
cf := &Changefeed{ID: common.NewChangeFeedIDWithName("test", common.DefaultKeyspaceName)}

coordinator/controller.go

Lines changed: 29 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -186,6 +186,7 @@ func NewController(
186186
func (c *Controller) collectMetrics(ctx context.Context) error {
187187
ticker := time.NewTicker(5 * time.Second)
188188
defer ticker.Stop()
189+
defer metrics.ResetOwnerChangefeedMetrics()
189190
for {
190191
select {
191192
case <-ctx.Done():
@@ -203,21 +204,41 @@ func (c *Controller) collectMetrics(ctx context.Context) error {
203204
name := info.ChangefeedID.Name()
204205
metrics.ChangefeedStatusGauge.WithLabelValues(keyspace, name).Set(float64(info.State.ToInt()))
205206

206-
// don't update checkpoint ts and checkpoint ts lag for stopped changefeed
207-
if info.State == config.StateStopped {
207+
if !updateChangefeedCheckpointMetrics(
208+
keyspace,
209+
name,
210+
info.State,
211+
cf.GetLastSavedCheckPointTs(),
212+
c.pdClock.CurrentTime(),
213+
) {
208214
return
209215
}
210-
211-
pdPhysicalTime := oracle.GetPhysical(c.pdClock.CurrentTime())
212-
phyCkpTs := oracle.ExtractPhysical(cf.GetLastSavedCheckPointTs())
213-
lag := float64(pdPhysicalTime-phyCkpTs) / 1e3
214-
metrics.ChangefeedCheckpointTsGauge.WithLabelValues(keyspace, name).Set(float64(phyCkpTs))
215-
metrics.ChangefeedCheckpointTsLagGauge.WithLabelValues(keyspace, name).Set(lag)
216216
})
217217
}
218218
}
219219
}
220220

221+
func updateChangefeedCheckpointMetrics(
222+
keyspace string,
223+
name string,
224+
state config.FeedState,
225+
checkpointTs uint64,
226+
pdTime time.Time,
227+
) bool {
228+
switch state {
229+
case config.StateStopped, config.StateFinished, config.StateRemoved:
230+
metrics.DeleteChangefeedCheckpointMetrics(keyspace, name)
231+
return false
232+
}
233+
234+
pdPhysicalTime := oracle.GetPhysical(pdTime)
235+
phyCkpTs := oracle.ExtractPhysical(checkpointTs)
236+
lag := float64(pdPhysicalTime-phyCkpTs) / 1e3
237+
metrics.ChangefeedCheckpointTsGauge.WithLabelValues(keyspace, name).Set(float64(phyCkpTs))
238+
metrics.ChangefeedCheckpointTsLagGauge.WithLabelValues(keyspace, name).Set(lag)
239+
return true
240+
}
241+
221242
// HandleEvent implements the event-driven process mode
222243
func (c *Controller) HandleEvent(ctx context.Context, event *Event) {
223244
if event == nil {

coordinator/controller_test.go

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,12 +28,45 @@ import (
2828
"github.com/pingcap/ticdc/pkg/config"
2929
"github.com/pingcap/ticdc/pkg/errors"
3030
"github.com/pingcap/ticdc/pkg/messaging"
31+
"github.com/pingcap/ticdc/pkg/metrics"
3132
"github.com/pingcap/ticdc/pkg/node"
3233
"github.com/pingcap/ticdc/server/watcher"
34+
"github.com/prometheus/client_golang/prometheus/testutil"
3335
"github.com/stretchr/testify/require"
36+
"github.com/tikv/client-go/v2/oracle"
3437
"go.uber.org/atomic"
3538
)
3639

40+
func TestUpdateChangefeedCheckpointMetricsDeletesFinishedLabels(t *testing.T) {
41+
metrics.ResetOwnerChangefeedMetrics()
42+
t.Cleanup(metrics.ResetOwnerChangefeedMetrics)
43+
44+
keyspace := common.DefaultKeyspaceName
45+
name := "finished-metrics"
46+
pdTime := time.UnixMilli(2000)
47+
checkpointTs := oracle.ComposeTS(1000, 0)
48+
49+
require.True(t, updateChangefeedCheckpointMetrics(
50+
keyspace,
51+
name,
52+
config.StateNormal,
53+
checkpointTs,
54+
pdTime,
55+
))
56+
require.Equal(t, 1, testutil.CollectAndCount(metrics.ChangefeedCheckpointTsGauge))
57+
require.Equal(t, 1, testutil.CollectAndCount(metrics.ChangefeedCheckpointTsLagGauge))
58+
59+
require.False(t, updateChangefeedCheckpointMetrics(
60+
keyspace,
61+
name,
62+
config.StateFinished,
63+
checkpointTs,
64+
pdTime,
65+
))
66+
require.Equal(t, 0, testutil.CollectAndCount(metrics.ChangefeedCheckpointTsGauge))
67+
require.Equal(t, 0, testutil.CollectAndCount(metrics.ChangefeedCheckpointTsLagGauge))
68+
}
69+
3770
func TestResumeChangefeed(t *testing.T) {
3871
ctrl := gomock.NewController(t)
3972
backend := mock_changefeed.NewMockBackend(ctrl)

pkg/common/table_info_shared_schema_guard_test.go

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -83,7 +83,9 @@ func TestLatestTiDBTableInfoSharedSchemaGuard(t *testing.T) {
8383
"Partition", "Compression", "View", "Sequence", "Lock", "Version", "TiFlashReplica", "IsColumnar",
8484
"TempTableType", "TableCacheStatusType", "PlacementPolicyRef", "StatsOptions",
8585
"ExchangePartitionInfo", "TTLInfo", "IsActiveActive", "SoftdeleteInfo", "Affinity",
86-
"Revision", "DBID", "Mode",
86+
"Revision", "DBID",
87+
// These table-level storage settings do not affect the shared column schema.
88+
"EngineAttribute", "StorageClassTier", "StorageClassTransitions", "Mode",
8789
},
8890
},
8991
{

pkg/metrics/changefeed.go

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -91,6 +91,17 @@ var (
9191
}, []string{getKeyspaceLabel(), "changefeed"})
9292
)
9393

94+
func DeleteChangefeedCheckpointMetrics(keyspace, changefeed string) {
95+
ChangefeedCheckpointTsGauge.DeleteLabelValues(keyspace, changefeed)
96+
ChangefeedCheckpointTsLagGauge.DeleteLabelValues(keyspace, changefeed)
97+
}
98+
99+
func ResetOwnerChangefeedMetrics() {
100+
ChangefeedStatusGauge.Reset()
101+
ChangefeedCheckpointTsGauge.Reset()
102+
ChangefeedCheckpointTsLagGauge.Reset()
103+
}
104+
94105
func initChangefeedMetrics(registry *prometheus.Registry) {
95106
registry.MustRegister(MaintainerCheckpointTsGauge)
96107
registry.MustRegister(MaintainerCheckpointTsLagGauge)

pkg/metrics/changefeed_test.go

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,43 @@
1+
// Copyright 2026 PingCAP, Inc.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// See the License for the specific language governing permissions and
12+
// limitations under the License.
13+
14+
package metrics
15+
16+
import (
17+
"testing"
18+
19+
"github.com/prometheus/client_golang/prometheus/testutil"
20+
"github.com/stretchr/testify/require"
21+
)
22+
23+
func TestResetOwnerChangefeedMetrics(t *testing.T) {
24+
ResetOwnerChangefeedMetrics()
25+
t.Cleanup(ResetOwnerChangefeedMetrics)
26+
27+
keyspace := "default"
28+
changefeed := "reset-owner-changefeed-metrics"
29+
30+
ChangefeedStatusGauge.WithLabelValues(keyspace, changefeed).Set(1)
31+
ChangefeedCheckpointTsGauge.WithLabelValues(keyspace, changefeed).Set(100)
32+
ChangefeedCheckpointTsLagGauge.WithLabelValues(keyspace, changefeed).Set(10)
33+
34+
require.Equal(t, 1, testutil.CollectAndCount(ChangefeedStatusGauge))
35+
require.Equal(t, 1, testutil.CollectAndCount(ChangefeedCheckpointTsGauge))
36+
require.Equal(t, 1, testutil.CollectAndCount(ChangefeedCheckpointTsLagGauge))
37+
38+
ResetOwnerChangefeedMetrics()
39+
40+
require.Equal(t, 0, testutil.CollectAndCount(ChangefeedStatusGauge))
41+
require.Equal(t, 0, testutil.CollectAndCount(ChangefeedCheckpointTsGauge))
42+
require.Equal(t, 0, testutil.CollectAndCount(ChangefeedCheckpointTsLagGauge))
43+
}

0 commit comments

Comments
 (0)