operator: check store status for running operators (#4223)

server/schedule/operator/step.go

@@ -26,6 +26,7 @@ import (
 	"github.com/tikv/pd/pkg/typeutil"
 	"github.com/tikv/pd/server/core"
 	"github.com/tikv/pd/server/core/storelimit"
+	"github.com/tikv/pd/server/schedule/opt"
 	"go.uber.org/zap"
 )
 
@@ -34,7 +35,7 @@ type OpStep interface {
 	fmt.Stringer
 	ConfVerChanged(region *core.RegionInfo) uint64
 	IsFinish(region *core.RegionInfo) bool
-	CheckSafety(region *core.RegionInfo) error
+	CheckInProgress(cluster opt.Cluster, region *core.RegionInfo) error
 	Influence(opInfluence OpInfluence, region *core.RegionInfo)
 }
 
@@ -57,16 +58,16 @@ func (tl TransferLeader) IsFinish(region *core.RegionInfo) bool {
 	return region.GetLeader().GetStoreId() == tl.ToStore
 }
 
-// CheckSafety checks if the step meets the safety properties.
-func (tl TransferLeader) CheckSafety(region *core.RegionInfo) error {
+// CheckInProgress checks if the step is in the progress of advancing.
+func (tl TransferLeader) CheckInProgress(cluster opt.Cluster, region *core.RegionInfo) error {
 	peer := region.GetStorePeer(tl.ToStore)
 	if peer == nil {
 		return errors.New("peer does not existed")
 	}
 	if core.IsLearner(peer) {
 		return errors.New("peer already is a learner")
 	}
-	return nil
+	return validateStore(cluster, tl.ToStore)
 }
 
 // Influence calculates the store difference that current step makes.
@@ -117,8 +118,11 @@ func (ap AddPeer) Influence(opInfluence OpInfluence, region *core.RegionInfo) {
 	to.AdjustStepCost(storelimit.AddPeer, regionSize)
 }
 
-// CheckSafety checks if the step meets the safety properties.
-func (ap AddPeer) CheckSafety(region *core.RegionInfo) error {
+// CheckInProgress checks if the step is in the progress of advancing.
+func (ap AddPeer) CheckInProgress(cluster opt.Cluster, region *core.RegionInfo) error {
+	if err := validateStore(cluster, ap.ToStore); err != nil {
+		return err
+	}
 	peer := region.GetStorePeer(ap.ToStore)
 	if peer != nil && peer.GetId() != ap.PeerID {
 		return errors.Errorf("peer %d has already existed in store %d, the operator is trying to add peer %d on the same store", peer.GetId(), ap.ToStore, ap.PeerID)
@@ -153,8 +157,11 @@ func (al AddLearner) IsFinish(region *core.RegionInfo) bool {
 	return false
 }
 
-// CheckSafety checks if the step meets the safety properties.
-func (al AddLearner) CheckSafety(region *core.RegionInfo) error {
+// CheckInProgress checks if the step is in the progress of advancing.
+func (al AddLearner) CheckInProgress(cluster opt.Cluster, region *core.RegionInfo) error {
+	if err := validateStore(cluster, al.ToStore); err != nil {
+		return err
+	}
 	peer := region.GetStorePeer(al.ToStore)
 	if peer == nil {
 		return nil
@@ -204,8 +211,8 @@ func (pl PromoteLearner) IsFinish(region *core.RegionInfo) bool {
 	return false
 }
 
-// CheckSafety checks if the step meets the safety properties.
-func (pl PromoteLearner) CheckSafety(region *core.RegionInfo) error {
+// CheckInProgress checks if the step is in the progress of advancing.
+func (pl PromoteLearner) CheckInProgress(cluster opt.Cluster, region *core.RegionInfo) error {
 	peer := region.GetStorePeer(pl.ToStore)
 	if peer.GetId() != pl.PeerID {
 		return errors.New("peer does not exist")
@@ -244,8 +251,8 @@ func (rp RemovePeer) IsFinish(region *core.RegionInfo) bool {
 	return region.GetStorePeer(rp.FromStore) == nil
 }
 
-// CheckSafety checks if the step meets the safety properties.
-func (rp RemovePeer) CheckSafety(region *core.RegionInfo) error {
+// CheckInProgress checks if the step is in the progress of advancing.
+func (rp RemovePeer) CheckInProgress(cluster opt.Cluster, region *core.RegionInfo) error {
 	if rp.FromStore == region.GetLeader().GetStoreId() {
 		return errors.New("cannot remove leader peer")
 	}
@@ -292,8 +299,8 @@ func (mr MergeRegion) IsFinish(region *core.RegionInfo) bool {
 	return false
 }
 
-// CheckSafety checks if the step meets the safety properties.
-func (mr MergeRegion) CheckSafety(region *core.RegionInfo) error {
+// CheckInProgress checks if the step is in the progress of advancing.
+func (mr MergeRegion) CheckInProgress(cluster opt.Cluster, region *core.RegionInfo) error {
 	return nil
 }
 
@@ -342,8 +349,8 @@ func (sr SplitRegion) Influence(opInfluence OpInfluence, region *core.RegionInfo
 	}
 }
 
-// CheckSafety checks if the step meets the safety properties.
-func (sr SplitRegion) CheckSafety(region *core.RegionInfo) error {
+// CheckInProgress checks if the step is in the progress of advancing.
+func (sr SplitRegion) CheckInProgress(cluster opt.Cluster, region *core.RegionInfo) error {
 	return nil
 }
 
@@ -374,8 +381,11 @@ func (ap AddLightPeer) IsFinish(region *core.RegionInfo) bool {
 	return false
 }
 
-// CheckSafety checks if the step meets the safety properties.
-func (ap AddLightPeer) CheckSafety(region *core.RegionInfo) error {
+// CheckInProgress checks if the step meets the safety properties.
+func (ap AddLightPeer) CheckInProgress(cluster opt.Cluster, region *core.RegionInfo) error {
+	if err := validateStore(cluster, ap.ToStore); err != nil {
+		return err
+	}
 	peer := region.GetStorePeer(ap.ToStore)
 	if peer != nil && peer.GetId() != ap.PeerID {
 		return errors.Errorf("peer %d has already existed in store %d, the operator is trying to add peer %d on the same store", peer.GetId(), ap.ToStore, ap.PeerID)
@@ -418,8 +428,11 @@ func (al AddLightLearner) IsFinish(region *core.RegionInfo) bool {
 	return false
 }
 
-// CheckSafety checks if the step meets the safety properties.
-func (al AddLightLearner) CheckSafety(region *core.RegionInfo) error {
+// CheckInProgress checks if the step meets the safety properties.
+func (al AddLightLearner) CheckInProgress(cluster opt.Cluster, region *core.RegionInfo) error {
+	if err := validateStore(cluster, al.ToStore); err != nil {
+		return err
+	}
 	peer := region.GetStorePeer(al.ToStore)
 	if peer == nil {
 		return nil
@@ -467,8 +480,8 @@ func (df DemoteFollower) IsFinish(region *core.RegionInfo) bool {
 	return false
 }
 
-// CheckSafety checks if the step meets the safety properties.
-func (df DemoteFollower) CheckSafety(region *core.RegionInfo) error {
+// CheckInProgress checks if the step is in the progress of advancing.
+func (df DemoteFollower) CheckInProgress(cluster opt.Cluster, region *core.RegionInfo) error {
 	peer := region.GetStorePeer(df.ToStore)
 	if peer.GetId() != df.PeerID {
 		return errors.New("peer does not exist")
@@ -567,8 +580,8 @@ func (cpe ChangePeerV2Enter) IsFinish(region *core.RegionInfo) bool {
 	return true
 }
 
-// CheckSafety checks if the step meets the safety properties.
-func (cpe ChangePeerV2Enter) CheckSafety(region *core.RegionInfo) error {
+// CheckInProgress checks if the step is in the progress of advancing.
+func (cpe ChangePeerV2Enter) CheckInProgress(cluster opt.Cluster, region *core.RegionInfo) error {
 	inJointState, notInJointState := false, false
 	for _, pl := range cpe.PromoteLearners {
 		peer := region.GetStorePeer(pl.ToStore)
@@ -707,8 +720,8 @@ func (cpl ChangePeerV2Leave) IsFinish(region *core.RegionInfo) bool {
 	return true
 }
 
-// CheckSafety checks if the step meets the safety properties.
-func (cpl ChangePeerV2Leave) CheckSafety(region *core.RegionInfo) error {
+// CheckInProgress checks if the step is in the progress of advancing.
+func (cpl ChangePeerV2Leave) CheckInProgress(cluster opt.Cluster, region *core.RegionInfo) error {
 	inJointState, notInJointState, demoteLeader := false, false, false
 	leaderStoreID := region.GetLeader().GetStoreId()
 
@@ -768,3 +781,14 @@ func (cpl ChangePeerV2Leave) CheckSafety(region *core.RegionInfo) error {
 
 // Influence calculates the store difference that current step makes.
 func (cpl ChangePeerV2Leave) Influence(opInfluence OpInfluence, region *core.RegionInfo) {}
+
+func validateStore(cluster opt.Cluster, id uint64) error {
+	store := cluster.GetStore(id)
+	if store == nil {
+		return errors.New("target store does not exist")
+	}
+	if store.DownTime() > cluster.GetOpts().GetMaxStoreDownTime() {
+		return errors.New("target store is down")
+	}
+	return nil
+}

server/schedule/operator/step_test.go

@@ -14,20 +14,162 @@
 package operator
 
 import (
+	"context"
+
 	. "github.com/pingcap/check"
 	"github.com/pingcap/kvproto/pkg/metapb"
+	"github.com/tikv/pd/pkg/mock/mockcluster"
+	"github.com/tikv/pd/server/config"
 	"github.com/tikv/pd/server/core"
 )
 
-type testStepSuite struct{}
+type testStepSuite struct {
+	cluster *mockcluster.Cluster
+}
 
 var _ = Suite(&testStepSuite{})
 
 type testCase struct {
 	Peers          []*metapb.Peer // first is leader
 	ConfVerChanged uint64
 	IsFinish       bool
-	CheckSafety    Checker
+	CheckInProgres Checker
+}
+
+func (s *testStepSuite) SetUpTest(c *C) {
+	s.cluster = mockcluster.NewCluster(context.Background(), config.NewTestOptions())
+	for i := 1; i <= 10; i++ {
+		s.cluster.PutStoreWithLabels(uint64(i))
+	}
+	s.cluster.SetStoreDown(8)
+	s.cluster.SetStoreDown(9)
+	s.cluster.SetStoreDown(10)
+}
+
+func (s *testStepSuite) TestTransferLeader(c *C) {
+	step := TransferLeader{FromStore: 1, ToStore: 2}
+	cases := []testCase{
+		{
+			[]*metapb.Peer{
+				{Id: 1, StoreId: 1, Role: metapb.PeerRole_Voter},
+				{Id: 2, StoreId: 2, Role: metapb.PeerRole_Voter},
+				{Id: 3, StoreId: 3, Role: metapb.PeerRole_Voter},
+			},
+			0,
+			false,
+			IsNil,
+		},
+		{
+			[]*metapb.Peer{
+				{Id: 2, StoreId: 2, Role: metapb.PeerRole_Voter},
+				{Id: 1, StoreId: 1, Role: metapb.PeerRole_Voter},
+				{Id: 3, StoreId: 3, Role: metapb.PeerRole_Voter},
+			},
+			0,
+			true,
+			IsNil,
+		},
+		{
+			[]*metapb.Peer{
+				{Id: 3, StoreId: 3, Role: metapb.PeerRole_Voter},
+				{Id: 1, StoreId: 1, Role: metapb.PeerRole_Voter},
+				{Id: 2, StoreId: 2, Role: metapb.PeerRole_Voter},
+			},
+			0,
+			false,
+			IsNil,
+		},
+	}
+	s.check(c, step, "transfer leader from store 1 to store 2", cases)
+
+	step = TransferLeader{FromStore: 1, ToStore: 9} // 9 is down
+	cases = []testCase{
+		{
+			[]*metapb.Peer{
+				{Id: 1, StoreId: 1, Role: metapb.PeerRole_Voter},
+				{Id: 2, StoreId: 2, Role: metapb.PeerRole_Voter},
+				{Id: 9, StoreId: 9, Role: metapb.PeerRole_Voter},
+			},
+			0,
+			false,
+			NotNil,
+		},
+	}
+	s.check(c, step, "transfer leader from store 1 to store 9", cases)
+}
+
+func (s *testStepSuite) TestAddPeer(c *C) {
+	step := AddPeer{ToStore: 2, PeerID: 2}
+	cases := []testCase{
+		{
+			[]*metapb.Peer{
+				{Id: 1, StoreId: 1, Role: metapb.PeerRole_Voter},
+			},
+			0,
+			false,
+			IsNil,
+		},
+		{
+			[]*metapb.Peer{
+				{Id: 1, StoreId: 1, Role: metapb.PeerRole_Voter},
+				{Id: 2, StoreId: 2, Role: metapb.PeerRole_Voter},
+			},
+			1,
+			true,
+			IsNil,
+		},
+	}
+	s.check(c, step, "add peer 2 on store 2", cases)
+
+	step = AddPeer{ToStore: 9, PeerID: 9}
+	cases = []testCase{
+		{
+			[]*metapb.Peer{
+				{Id: 1, StoreId: 1, Role: metapb.PeerRole_Voter},
+			},
+			0,
+			false,
+			NotNil,
+		},
+	}
+	s.check(c, step, "add peer 9 on store 9", cases)
+}
+
+func (s *testStepSuite) TestAddLearner(c *C) {
+	step := AddLearner{ToStore: 2, PeerID: 2}
+	cases := []testCase{
+		{
+			[]*metapb.Peer{
+				{Id: 1, StoreId: 1, Role: metapb.PeerRole_Voter},
+			},
+			0,
+			false,
+			IsNil,
+		},
+		{
+			[]*metapb.Peer{
+				{Id: 1, StoreId: 1, Role: metapb.PeerRole_Voter},
+				{Id: 2, StoreId: 2, Role: metapb.PeerRole_Learner},
+			},
+			1,
+			true,
+			IsNil,
+		},
+	}
+	s.check(c, step, "add learner peer 2 on store 2", cases)
+
+	step = AddLearner{ToStore: 9, PeerID: 9}
+	cases = []testCase{
+		{
+			[]*metapb.Peer{
+				{Id: 1, StoreId: 1, Role: metapb.PeerRole_Voter},
+			},
+			0,
+			false,
+			NotNil,
+		},
+	}
+	s.check(c, step, "add learner peer 9 on store 9", cases)
 }
 
 func (s *testStepSuite) TestDemoteFollower(c *C) {
@@ -344,6 +486,6 @@ func (s *testStepSuite) check(c *C, step OpStep, desc string, cases []testCase)
 		region := core.NewRegionInfo(&metapb.Region{Id: 1, Peers: tc.Peers}, tc.Peers[0])
 		c.Assert(step.ConfVerChanged(region), Equals, tc.ConfVerChanged)
 		c.Assert(step.IsFinish(region), Equals, tc.IsFinish)
-		c.Assert(step.CheckSafety(region), tc.CheckSafety)
+		c.Assert(step.CheckInProgress(s.cluster, region), tc.CheckInProgres)
 	}
 }

server/schedule/operator_controller.go

@@ -161,7 +161,7 @@ func (oc *OperatorController) Dispatch(region *core.RegionInfo, source string) {
 }
 
 func (oc *OperatorController) checkStaleOperator(op *operator.Operator, step operator.OpStep, region *core.RegionInfo) bool {
-	err := step.CheckSafety(region)
+	err := step.CheckInProgress(oc.cluster, region)
 	if err != nil {
 		if oc.RemoveOperator(op, zap.String("reason", err.Error())) {
 			operatorCounter.WithLabelValues(op.Desc(), "stale").Inc()