changefeed (ticdc): remove status of a changefeed after it is removed (…

…#9174) (#9184)

close #9173, ref #9177

cdc/owner/changefeed.go

@@ -805,6 +805,10 @@ func (c *changefeed) cleanupMetrics() {
 
 	changefeedBarrierTsGauge.DeleteLabelValues(c.id.Namespace, c.id.ID)
 	c.metricsChangefeedBarrierTsGauge = nil
+
+	if c.isRemoved {
+		changefeedStatusGauge.DeleteLabelValues(c.id.Namespace, c.id.ID)
+	}
 }
 
 // cleanup redo logs if changefeed is removed and redo log is enabled

cdc/processor/processor.go

@@ -912,6 +912,10 @@ func (p *processor) Close() error {
 		zap.String("namespace", p.changefeedID.Namespace),
 		zap.String("changefeed", p.changefeedID.ID))
 
+	// clean up metrics first to avoid some metrics are not cleaned up
+	// when error occurs during closing the processor
+	p.cleanupMetrics()
+
 	p.sinkManager.stop(p.changefeedID)
 	p.sinkManager.r = nil
 	p.sourceManager.stop(p.changefeedID)
@@ -949,7 +953,6 @@ func (p *processor) Close() error {
 	// mark tables share the same cdcContext with its original table, don't need to cancel
 	failpoint.Inject("processorStopDelay", nil)
 
-	p.cleanupMetrics()
 	log.Info("processor closed",
 		zap.String("namespace", p.changefeedID.Namespace),
 		zap.String("changefeed", p.changefeedID.ID))