server/operator: add operator state

server/coordinator.go

@@ -222,22 +222,26 @@ func (c *coordinator) runScheduler(s *scheduleController) {
 
 func (c *coordinator) addOperator(op Operator) bool {
 	c.Lock()
-	defer c.Unlock()
 
 	regionID := op.GetRegionID()
 
 	if old, ok := c.operators[regionID]; ok {
 		if !isHigherPriorityOperator(op, old) {
+			c.Unlock()
 			return false
 		}
-		c.limiter.removeOperator(old)
+		c.Unlock()
+		old.SetState(OperatorReplaced)
+		c.removeOperator(old)
+		c.Lock()
 		log.Infof("coordinator: add operator %+v with higher priority, remove operator: %+v", op, old)
 	}
 
 	c.histories.add(regionID, op)
 	c.limiter.addOperator(op)
 	c.operators[regionID] = op
 	collectOperatorCounterMetrics(op)
+	c.Unlock()
 	return true
 }
 
@@ -260,6 +264,7 @@ func (c *coordinator) removeOperator(op Operator) {
 	delete(c.operators, regionID)
 
 	c.histories.add(regionID, op)
+	collectOperatorCounterMetrics(op)
 }
 
 func (c *coordinator) getOperator(regionID uint64) Operator {
@@ -391,21 +396,11 @@ func (s *scheduleController) AllowSchedule() bool {
 }
 
 func collectOperatorCounterMetrics(op Operator) {
-	metrics := make(map[string]uint64)
 	regionOp, ok := op.(*regionOperator)
 	if !ok {
 		return
 	}
 	for _, op := range regionOp.Ops {
-		switch o := op.(type) {
-		case *changePeerOperator:
-			metrics[o.Name]++
-		case *transferLeaderOperator:
-			metrics[o.Name]++
-		}
-	}
-
-	for label, value := range metrics {
-		operatorCounter.WithLabelValues(label).Add(float64(value))
+		operatorCounter.WithLabelValues(op.GetName(), op.GetState().String()).Add(1)
 	}
 }

server/coordinator_test.go

@@ -24,16 +24,20 @@ import (
 type testOperator struct {
 	RegionID uint64
 	Kind     ResourceKind
+	State    OperatorState
 }
 
 func newTestOperator(regionID uint64, kind ResourceKind) Operator {
 	region := newRegionInfo(&metapb.Region{Id: regionID}, nil)
-	op := &testOperator{RegionID: regionID, Kind: kind}
+	op := &testOperator{RegionID: regionID, Kind: kind, State: OperatorRunning}
 	return newRegionOperator(region, kind, op)
 }
 
 func (op *testOperator) GetRegionID() uint64           { return op.RegionID }
 func (op *testOperator) GetResourceKind() ResourceKind { return op.Kind }
+func (op *testOperator) GetState() OperatorState       { return op.State }
+func (op *testOperator) SetState(state OperatorState)  { op.State = state }
+func (op *testOperator) GetName() string               { return "test" }
 func (op *testOperator) Do(region *RegionInfo) (*pdpb.RegionHeartbeatResponse, bool) {
 	return nil, false
 }

server/event.go

@@ -157,6 +157,18 @@ func (op *splitOperator) GetResourceKind() ResourceKind {
 	return OtherKind
 }
 
+func (op *splitOperator) GetState() OperatorState {
+	return OperatorFinished
+}
+
+func (op *splitOperator) SetState(state OperatorState) {
+	return
+}
+
+func (op *splitOperator) GetName() string {
+	return op.Name
+}
+
 // Do implements Operator.Do interface.
 func (op *splitOperator) Do(region *RegionInfo) (*pdpb.RegionHeartbeatResponse, bool) {
 	return nil, true

server/metrics.go

@@ -39,7 +39,7 @@ var (
 			Subsystem: "schedule",
 			Name:      "operators_count",
 			Help:      "Counter of schedule operators.",
-		}, []string{"type"})
+		}, []string{"type", "state"})
 
 	clusterStatusGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{