schedule: add joint state checker

server/schedule/checker/joint_state_checker.go

@@ -0,0 +1,55 @@
+// Copyright 2020 TiKV Project Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package checker
+
+import (
+	"github.com/pingcap/log"
+	"github.com/tikv/pd/server/core"
+	"github.com/tikv/pd/server/schedule/operator"
+	"github.com/tikv/pd/server/schedule/opt"
+	"go.uber.org/zap"
+)
+
+// JointStateChecker ensures region is in joint state will leave.
+type JointStateChecker struct {
+	cluster opt.Cluster
+}
+
+// NewJointStateChecker creates a joint state checker.
+func NewJointStateChecker(cluster opt.Cluster) *JointStateChecker {
+	return &JointStateChecker{
+		cluster: cluster,
+	}
+}
+
+// Check verifies a region's role, creating an Operator if need.
+func (c *JointStateChecker) Check(region *core.RegionInfo) *operator.Operator {
+	checkerCounter.WithLabelValues("joint_state_checker", "check").Inc()
+	if !core.IsInJointState(region.GetPeers()...) {
+		return nil
+	}
+	op, err := operator.CreateLeaveJointStateOperator("leave-joint-state", c.cluster, region)
+	if err != nil {
+		checkerCounter.WithLabelValues("joint_state_checker", "create-operator-fail").Inc()
+		log.Debug("fail to create leave joint state operator", zap.Error(err))
+		return nil
+	} else if op != nil {
+		checkerCounter.WithLabelValues("joint_state_checker", "new-operator").Inc()
+		if op.Len() > 1 {
+			checkerCounter.WithLabelValues("joint_state_checker", "transfer-leader").Inc()
+		}
+		op.SetPriorityLevel(core.HighPriority)
+	}
+	return op
+}

server/schedule/checker/joint_state_checker_test.go

@@ -0,0 +1,155 @@
+// Copyright 2020 TiKV Project Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package checker
+
+import (
+	. "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"
+	"github.com/tikv/pd/server/schedule/operator"
+)
+
+var _ = Suite(&testJointStateCheckerSuite{})
+
+type testJointStateCheckerSuite struct {
+	cluster *mockcluster.Cluster
+	jsc     *JointStateChecker
+}
+
+func (s *testJointStateCheckerSuite) SetUpTest(c *C) {
+	s.cluster = mockcluster.NewCluster(config.NewTestOptions())
+	s.jsc = NewJointStateChecker(s.cluster)
+	for id := uint64(1); id <= 10; id++ {
+		s.cluster.PutStoreWithLabels(id)
+	}
+}
+
+func (s *testJointStateCheckerSuite) TestLeaveJointState(c *C) {
+	jsc := s.jsc
+	type testCase struct {
+		Peers   []*metapb.Peer // first is leader
+		OpSteps []operator.OpStep
+	}
+	cases := []testCase{
+		{
+			[]*metapb.Peer{
+				{Id: 101, StoreId: 1, Role: metapb.PeerRole_Voter},
+				{Id: 102, StoreId: 2, Role: metapb.PeerRole_DemotingVoter},
+				{Id: 103, StoreId: 3, Role: metapb.PeerRole_IncomingVoter},
+			},
+			[]operator.OpStep{
+				operator.ChangePeerV2Leave{
+					PromoteLearners: []operator.PromoteLearner{{ToStore: 3}},
+					DemoteVoters:    []operator.DemoteVoter{{ToStore: 2}},
+				},
+			},
+		},
+		{
+			[]*metapb.Peer{
+				{Id: 101, StoreId: 1, Role: metapb.PeerRole_Voter},
+				{Id: 102, StoreId: 2, Role: metapb.PeerRole_Voter},
+				{Id: 103, StoreId: 3, Role: metapb.PeerRole_IncomingVoter},
+			},
+			[]operator.OpStep{
+				operator.ChangePeerV2Leave{
+					PromoteLearners: []operator.PromoteLearner{{ToStore: 3}},
+					DemoteVoters:    []operator.DemoteVoter{},
+				},
+			},
+		},
+		{
+			[]*metapb.Peer{
+				{Id: 101, StoreId: 1, Role: metapb.PeerRole_Voter},
+				{Id: 102, StoreId: 2, Role: metapb.PeerRole_DemotingVoter},
+				{Id: 103, StoreId: 3, Role: metapb.PeerRole_Voter},
+			},
+			[]operator.OpStep{
+				operator.ChangePeerV2Leave{
+					PromoteLearners: []operator.PromoteLearner{},
+					DemoteVoters:    []operator.DemoteVoter{{ToStore: 2}},
+				},
+			},
+		},
+		{
+			[]*metapb.Peer{
+				{Id: 101, StoreId: 1, Role: metapb.PeerRole_DemotingVoter},
+				{Id: 102, StoreId: 2, Role: metapb.PeerRole_Voter},
+				{Id: 103, StoreId: 3, Role: metapb.PeerRole_Voter},
+			},
+			[]operator.OpStep{
+				operator.TransferLeader{FromStore: 1, ToStore: 2},
+				operator.ChangePeerV2Leave{
+					PromoteLearners: []operator.PromoteLearner{},
+					DemoteVoters:    []operator.DemoteVoter{{ToStore: 1}},
+				},
+			},
+		},
+		{
+			[]*metapb.Peer{
+				{Id: 101, StoreId: 1, Role: metapb.PeerRole_Voter},
+				{Id: 102, StoreId: 2, Role: metapb.PeerRole_Voter},
+				{Id: 103, StoreId: 3, Role: metapb.PeerRole_Voter},
+			},
+			nil,
+		},
+		{
+			[]*metapb.Peer{
+				{Id: 101, StoreId: 1, Role: metapb.PeerRole_Voter},
+				{Id: 102, StoreId: 2, Role: metapb.PeerRole_Voter},
+				{Id: 103, StoreId: 3, Role: metapb.PeerRole_Learner},
+			},
+			nil,
+		},
+	}
+
+	for _, tc := range cases {
+		region := core.NewRegionInfo(&metapb.Region{Id: 1, Peers: tc.Peers}, tc.Peers[0])
+		op := jsc.Check(region)
+		s.checkSteps(c, op, tc.OpSteps)
+	}
+}
+
+func (s *testJointStateCheckerSuite) checkSteps(c *C, op *operator.Operator, steps []operator.OpStep) {
+	if len(steps) == 0 {
+		c.Assert(op, IsNil)
+		return
+	}
+
+	c.Assert(op, NotNil)
+	c.Assert(op.Desc(), Equals, "leave-joint-state")
+
+	c.Assert(op.Len(), Equals, len(steps))
+	for i := range steps {
+		switch obtain := op.Step(i).(type) {
+		case operator.ChangePeerV2Leave:
+			expect := steps[i].(operator.ChangePeerV2Leave)
+			c.Assert(len(obtain.PromoteLearners), Equals, len(expect.PromoteLearners))
+			c.Assert(len(obtain.DemoteVoters), Equals, len(expect.DemoteVoters))
+			for j, p := range expect.PromoteLearners {
+				c.Assert(expect.PromoteLearners[j].ToStore, Equals, p.ToStore)
+			}
+			for j, d := range expect.DemoteVoters {
+				c.Assert(obtain.DemoteVoters[j].ToStore, Equals, d.ToStore)
+			}
+		case operator.TransferLeader:
+			expect := steps[i].(operator.TransferLeader)
+			c.Assert(obtain.FromStore, Equals, expect.FromStore)
+			c.Assert(obtain.ToStore, Equals, expect.ToStore)
+		default:
+			c.Fatal("unknown operator step type")
+		}
+	}
+}

server/schedule/checker_controller.go

@@ -26,26 +26,28 @@ import (
 
 // CheckerController is used to manage all checkers.
 type CheckerController struct {
-	cluster        opt.Cluster
-	opts           *config.PersistOptions
-	opController   *OperatorController
-	learnerChecker *checker.LearnerChecker
-	replicaChecker *checker.ReplicaChecker
-	ruleChecker    *checker.RuleChecker
-	mergeChecker   *checker.MergeChecker
+	cluster           opt.Cluster
+	opts              *config.PersistOptions
+	opController      *OperatorController
+	learnerChecker    *checker.LearnerChecker
+	replicaChecker    *checker.ReplicaChecker
+	ruleChecker       *checker.RuleChecker
+	mergeChecker      *checker.MergeChecker
+	jointStateChecker *checker.JointStateChecker
 }
 
 // NewCheckerController create a new CheckerController.
 // TODO: isSupportMerge should be removed.
 func NewCheckerController(ctx context.Context, cluster opt.Cluster, ruleManager *placement.RuleManager, opController *OperatorController) *CheckerController {
 	return &CheckerController{
-		cluster:        cluster,
-		opts:           cluster.GetOpts(),
-		opController:   opController,
-		learnerChecker: checker.NewLearnerChecker(cluster),
-		replicaChecker: checker.NewReplicaChecker(cluster),
-		ruleChecker:    checker.NewRuleChecker(cluster, ruleManager),
-		mergeChecker:   checker.NewMergeChecker(ctx, cluster),
+		cluster:           cluster,
+		opts:              cluster.GetOpts(),
+		opController:      opController,
+		learnerChecker:    checker.NewLearnerChecker(cluster),
+		replicaChecker:    checker.NewReplicaChecker(cluster),
+		ruleChecker:       checker.NewRuleChecker(cluster, ruleManager),
+		mergeChecker:      checker.NewMergeChecker(ctx, cluster),
+		jointStateChecker: checker.NewJointStateChecker(cluster),
 	}
 }
 
@@ -55,6 +57,11 @@ func (c *CheckerController) CheckRegion(region *core.RegionInfo) (bool, []*opera
 	// Don't check isRaftLearnerEnabled cause it maybe disable learner feature but there are still some learners to promote.
 	opController := c.opController
 	checkerIsBusy := true
+
+	if op := c.jointStateChecker.Check(region); op != nil {
+		return false, []*operator.Operator{op}
+	}
+
 	if c.opts.IsPlacementRulesEnabled() {
 		if opController.OperatorCount(operator.OpReplica) < c.opts.GetReplicaScheduleLimit() {
 			checkerIsBusy = false