owner(cdc): backport stale metrics fix to 5.4 (pingcap#5260)

ref pingcap#4774

cdc/owner/owner.go

@@ -251,6 +251,7 @@ func (o *Owner) WriteDebugInfo(w io.Writer) {
 // AsyncStop stops the owner asynchronously
 func (o *Owner) AsyncStop() {
 	atomic.StoreInt32(&o.closed, 1)
+	o.cleanStaleMetrics()
 }
 
 func (o *Owner) cleanUpChangefeed(state *orchestrator.ChangefeedReactorState) {
@@ -280,6 +281,7 @@ func (o *Owner) cleanUpChangefeed(state *orchestrator.ChangefeedReactorState) {
 // Bootstrap checks if the state contains incompatible or incorrect information and tries to fix it.
 func (o *Owner) Bootstrap(state *orchestrator.GlobalReactorState) {
 	log.Info("Start bootstrapping")
+	o.cleanStaleMetrics()
 	fixChangefeedInfos(state)
 }
 
@@ -298,16 +300,25 @@ func fixChangefeedInfos(state *orchestrator.GlobalReactorState) {
 	}
 }
 
+func (o *Owner) cleanStaleMetrics() {
+	// The gauge metrics of the Owner should be reset
+	// each time a new owner is launched, in case the previous owner
+	// has crashed and has not cleaned up the stale metrics values.
+	changefeedCheckpointTsGauge.Reset()
+	changefeedCheckpointTsLagGauge.Reset()
+	changefeedResolvedTsGauge.Reset()
+	changefeedResolvedTsLagGauge.Reset()
+	ownerMaintainTableNumGauge.Reset()
+	changefeedStatusGauge.Reset()
+}
+
 func (o *Owner) updateMetrics(state *orchestrator.GlobalReactorState) {
 	// Keep the value of prometheus expression `rate(counter)` = 1
 	// Please also change alert rule in ticdc.rules.yml when change the expression value.
 	now := time.Now()
 	ownershipCounter.Add(float64(now.Sub(o.lastTickTime)) / float64(time.Second))
 	o.lastTickTime = now
 
-	ownerMaintainTableNumGauge.Reset()
-	changefeedStatusGauge.Reset()
-
 	conf := config.GetGlobalServerConfig()
 
 	// TODO refactor this piece of code when the new scheduler is stabilized,