This is an automated cherry-pick of #2267 (#2275)

Signed-off-by: ti-chi-bot <ti-community-prow-bot@tidb.io>

Co-authored-by: Ling Jin <7138436+3AceShowHand@users.noreply.github.com>

cdc/model/changefeed.go

@@ -52,6 +52,26 @@ const (
 	StateFinished FeedState = "finished"
 )
 
+// ToInt return a int for each `FeedState`, only use this for metrics.
+func (s FeedState) ToInt() int {
+	switch s {
+	case StateNormal:
+		return 0
+	case StateError:
+		return 1
+	case StateFailed:
+		return 2
+	case StateStopped:
+		return 3
+	case StateFinished:
+		return 4
+	case StateRemoved:
+		return 5
+	}
+	// -1 for unknown feed state
+	return -1
+}
+
 const (
 	// errorHistoryGCInterval represents how long we keep error record in changefeed info
 	errorHistoryGCInterval = time.Minute * 10

cdc/owner.go

@@ -52,7 +52,7 @@ type ownership struct {
 	tickTime     time.Duration
 }
 
-func newOwnersip(tickTime time.Duration) ownership {
+func newOwnership(tickTime time.Duration) ownership {
 	minTickTime := 5 * time.Second
 	if tickTime > minTickTime {
 		log.Panic("ownership counter must be incearsed every 5 seconds")
@@ -1110,7 +1110,6 @@ func (o *Owner) handleAdminJob(ctx context.Context) error {
 			if err != nil {
 				return errors.Trace(err)
 			}
-
 			err = o.dispatchJob(ctx, job)
 			if err != nil {
 				return errors.Trace(err)
@@ -1276,7 +1275,7 @@ func (o *Owner) Run(ctx context.Context, tickTime time.Duration) error {
 	defer feedChangeReceiver.Stop()
 	o.watchFeedChange(ctx1)
 
-	ownership := newOwnersip(tickTime)
+	ownership := newOwnership(tickTime)
 loop:
 	for {
 		select {
@@ -1578,7 +1577,7 @@ func (o *Owner) watchCapture(ctx context.Context) error {
 	failpoint.Inject("sleep-before-watch-capture", nil)
 
 	// When an owner just starts, changefeed information is not updated at once.
-	// Supposing a crased capture should be removed now, the owner will miss deleting
+	// Supposing a crashed capture should be removed now, the owner will miss deleting
 	// task status and task position if changefeed information is not loaded.
 	// If the task positions and status decode failed, remove them.
 	if err := o.checkAndCleanTasksInfo(ctx); err != nil {

cdc/owner/metrics.go

@@ -44,6 +44,13 @@ var (
 			Name:      "maintain_table_num",
 			Help:      "number of replicated tables maintained in owner",
 		}, []string{"changefeed", "capture", "type"})
+	changefeedStatusGauge = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: "ticdc",
+			Subsystem: "owner",
+			Name:      "status",
+			Help:      "The status of changefeeds",
+		}, []string{"changefeed"})
 )
 
 const (
@@ -59,4 +66,5 @@ func InitMetrics(registry *prometheus.Registry) {
 	registry.MustRegister(changefeedCheckpointTsLagGauge)
 	registry.MustRegister(ownershipCounter)
 	registry.MustRegister(ownerMaintainTableNumGauge)
+	registry.MustRegister(changefeedStatusGauge)
 }

cdc/owner/owner.go

@@ -235,6 +235,7 @@ func (o *Owner) updateMetrics(state *model.GlobalReactorState) {
 	o.lastTickTime = now
 
 	ownerMaintainTableNumGauge.Reset()
+	changefeedStatusGauge.Reset()
 	for changefeedID, changefeedState := range state.Changefeeds {
 		for captureID, captureInfo := range state.Captures {
 			taskStatus, exist := changefeedState.TaskStatuses[captureID]
@@ -243,6 +244,9 @@ func (o *Owner) updateMetrics(state *model.GlobalReactorState) {
 			}
 			ownerMaintainTableNumGauge.WithLabelValues(changefeedID, captureInfo.AdvertiseAddr, maintainTableTypeTotal).Set(float64(len(taskStatus.Tables)))
 			ownerMaintainTableNumGauge.WithLabelValues(changefeedID, captureInfo.AdvertiseAddr, maintainTableTypeWip).Set(float64(len(taskStatus.Operation)))
+			if changefeedState.Info != nil {
+				changefeedStatusGauge.WithLabelValues(changefeedID).Set(float64(changefeedState.Info.State.ToInt()))
+			}
 		}
 	}
 }