rule_checker: can replace unhealthPeer with orphanPeer (#6831)

Makefile

@@ -148,7 +148,7 @@ static: install-tools
 	@ echo "gofmt ..."
 	@ gofmt -s -l -d $(PACKAGE_DIRECTORIES) 2>&1 | awk '{ print } END { if (NR > 0) { exit 1 } }'
 	@ echo "golangci-lint ..."
-	@ golangci-lint run --verbose $(PACKAGE_DIRECTORIES)
+	@ golangci-lint run --verbose $(PACKAGE_DIRECTORIES) --allow-parallel-runners
 	@ echo "revive ..."
 	@ revive -formatter friendly -config revive.toml $(PACKAGES)
 

server/api/operator_test.go

@@ -217,7 +217,13 @@ func (suite *transferRegionOperatorTestSuite) TestTransferRegionWithPlacementRul
 	regionURL := fmt.Sprintf("%s/operators/%d", suite.urlPrefix, region.GetId())
 	operator := mustReadURL(re, regionURL)
 	suite.Contains(operator, "operator not found")
-
+	convertStepsToStr := func(steps []string) string {
+		stepStrs := make([]string, len(steps))
+		for i := range steps {
+			stepStrs[i] = fmt.Sprintf("%d:{%s}", i, steps[i])
+		}
+		return strings.Join(stepStrs, ", ")
+	}
 	testCases := []struct {
 		name                string
 		placementRuleEnable bool
@@ -231,25 +237,25 @@ func (suite *transferRegionOperatorTestSuite) TestTransferRegionWithPlacementRul
 			placementRuleEnable: false,
 			input:               []byte(`{"name":"transfer-region", "region_id": 1, "to_store_ids": [2, 3]}`),
 			expectedError:       nil,
-			expectSteps: strings.Join([]string{
+			expectSteps: convertStepsToStr([]string{
 				pdoperator.AddLearner{ToStore: 3, PeerID: 1}.String(),
 				pdoperator.PromoteLearner{ToStore: 3, PeerID: 1}.String(),
 				pdoperator.TransferLeader{FromStore: 1, ToStore: 2}.String(),
 				pdoperator.RemovePeer{FromStore: 1, PeerID: 1}.String(),
-			}, ", "),
+			}),
 		},
 		{
 			name:                "placement rule disable with peer role",
 			placementRuleEnable: false,
 			input:               []byte(`{"name":"transfer-region", "region_id": 1, "to_store_ids": [2, 3], "peer_roles":["follower", "leader"]}`),
 			expectedError:       nil,
-			expectSteps: strings.Join([]string{
+			expectSteps: convertStepsToStr([]string{
 				pdoperator.AddLearner{ToStore: 3, PeerID: 2}.String(),
 				pdoperator.PromoteLearner{ToStore: 3, PeerID: 2}.String(),
 				pdoperator.TransferLeader{FromStore: 1, ToStore: 2}.String(),
 				pdoperator.RemovePeer{FromStore: 1, PeerID: 2}.String(),
 				pdoperator.TransferLeader{FromStore: 2, ToStore: 3}.String(),
-			}, ", "),
+			}),
 		},
 		{
 			name:                "default placement rule without peer role",
@@ -262,13 +268,13 @@ func (suite *transferRegionOperatorTestSuite) TestTransferRegionWithPlacementRul
 			name:                "default placement rule with peer role",
 			placementRuleEnable: true,
 			input:               []byte(`{"name":"transfer-region", "region_id": 1, "to_store_ids": [2, 3], "peer_roles":["follower", "leader"]}`),
-			expectSteps: strings.Join([]string{
+			expectSteps: convertStepsToStr([]string{
 				pdoperator.AddLearner{ToStore: 3, PeerID: 3}.String(),
 				pdoperator.PromoteLearner{ToStore: 3, PeerID: 3}.String(),
 				pdoperator.TransferLeader{FromStore: 1, ToStore: 2}.String(),
 				pdoperator.RemovePeer{FromStore: 1, PeerID: 1}.String(),
 				pdoperator.TransferLeader{FromStore: 2, ToStore: 3}.String(),
-			}, ", "),
+			}),
 		},
 		{
 			name:                "default placement rule with invalid input",
@@ -323,12 +329,12 @@ func (suite *transferRegionOperatorTestSuite) TestTransferRegionWithPlacementRul
 			},
 			input:         []byte(`{"name":"transfer-region", "region_id": 1, "to_store_ids": [2, 3], "peer_roles":["follower", "leader"]}`),
 			expectedError: nil,
-			expectSteps: strings.Join([]string{
+			expectSteps: convertStepsToStr([]string{
 				pdoperator.AddLearner{ToStore: 3, PeerID: 5}.String(),
 				pdoperator.PromoteLearner{ToStore: 3, PeerID: 5}.String(),
 				pdoperator.TransferLeader{FromStore: 1, ToStore: 3}.String(),
 				pdoperator.RemovePeer{FromStore: 1, PeerID: 1}.String(),
-			}, ", "),
+			}),
 		},
 		{
 			name:                "customized placement rule with valid peer role2",
@@ -363,12 +369,12 @@ func (suite *transferRegionOperatorTestSuite) TestTransferRegionWithPlacementRul
 			},
 			input:         []byte(`{"name":"transfer-region", "region_id": 1, "to_store_ids": [2, 3], "peer_roles":["leader", "follower"]}`),
 			expectedError: nil,
-			expectSteps: strings.Join([]string{
+			expectSteps: convertStepsToStr([]string{
 				pdoperator.AddLearner{ToStore: 3, PeerID: 6}.String(),
 				pdoperator.PromoteLearner{ToStore: 3, PeerID: 6}.String(),
 				pdoperator.TransferLeader{FromStore: 1, ToStore: 2}.String(),
 				pdoperator.RemovePeer{FromStore: 1, PeerID: 1}.String(),
-			}, ", "),
+			}),
 		},
 	}
 	for _, testCase := range testCases {

server/schedule/checker/rule_checker.go

@@ -390,14 +390,15 @@ func (c *RuleChecker) fixOrphanPeers(region *core.RegionInfo, fit *placement.Reg
 	if len(fit.OrphanPeers) == 0 {
 		return nil, nil
 	}
+	var pinDownPeer *metapb.Peer
 	isUnhealthyPeer := func(id uint64) bool {
-		for _, pendingPeer := range region.GetPendingPeers() {
-			if pendingPeer.GetId() == id {
+		for _, downPeer := range region.GetDownPeers() {
+			if downPeer.Peer.GetId() == id {
 				return true
 			}
 		}
-		for _, downPeer := range region.GetDownPeers() {
-			if downPeer.Peer.GetId() == id {
+		for _, pendingPeer := range region.GetPendingPeers() {
+			if pendingPeer.GetId() == id {
 				return true
 			}
 		}
@@ -414,16 +415,56 @@ loopFits:
 		}
 		for _, p := range rf.Peers {
 			if isUnhealthyPeer(p.GetId()) {
+				// make sure is down peer.
+				if region.GetDownPeer(p.GetId()) != nil {
+					pinDownPeer = p
+				}
 				hasUnhealthyFit = true
 				break loopFits
 			}
 		}
 	}
+
 	// If hasUnhealthyFit is false, it is safe to delete the OrphanPeer.
 	if !hasUnhealthyFit {
 		checkerCounter.WithLabelValues("rule_checker", "remove-orphan-peer").Inc()
 		return operator.CreateRemovePeerOperator("remove-orphan-peer", c.cluster, 0, region, fit.OrphanPeers[0].StoreId)
 	}
+
+	// try to use orphan peers to replace unhealthy down peers.
+	for _, orphanPeer := range fit.OrphanPeers {
+		if pinDownPeer != nil {
+			// make sure the orphan peer is healthy.
+			if isUnhealthyPeer(orphanPeer.GetId()) {
+				continue
+			}
+			// no consider witness in this path.
+			if pinDownPeer.GetIsWitness() || orphanPeer.GetIsWitness() {
+				continue
+			}
+			// down peer's store should be down.
+			if !c.isStoreDownTimeHitMaxDownTime(pinDownPeer.GetStoreId()) {
+				continue
+			}
+			// check if down peer can replace with orphan peer.
+			dstStore := c.cluster.GetStore(orphanPeer.GetStoreId())
+			if fit.Replace(pinDownPeer.GetStoreId(), dstStore) {
+				destRole := pinDownPeer.GetRole()
+				orphanPeerRole := orphanPeer.GetRole()
+				checkerCounter.WithLabelValues("rule_checker", "replace-orphan-peer").Inc()
+				switch {
+				case orphanPeerRole == metapb.PeerRole_Learner && destRole == metapb.PeerRole_Voter:
+					return operator.CreatePromoteLearnerOperatorAndRemovePeer("replace-down-peer-with-orphan-peer", c.cluster, region, orphanPeer, pinDownPeer)
+				case orphanPeerRole == metapb.PeerRole_Voter && destRole == metapb.PeerRole_Learner:
+					return operator.CreateDemoteLearnerOperatorAndRemovePeer("replace-down-peer-with-orphan-peer", c.cluster, region, orphanPeer, pinDownPeer)
+				default:
+					// destRole should not same with orphanPeerRole. if role is same, it fit with orphanPeer should be better than now.
+					// destRole never be leader, so we not consider it.
+				}
+			}
+		}
+	}
+
 	// If hasUnhealthyFit is true, try to remove unhealthy orphan peers only if number of OrphanPeers is >= 2.
 	// Ref https://github.com/tikv/pd/issues/4045
 	if len(fit.OrphanPeers) >= 2 {
@@ -462,6 +503,10 @@ func (c *RuleChecker) isDownPeer(region *core.RegionInfo, peer *metapb.Peer) boo
 
 func (c *RuleChecker) isStoreDownTimeHitMaxDownTime(storeID uint64) bool {
 	store := c.cluster.GetStore(storeID)
+	if store == nil {
+		log.Warn("lost the store, maybe you are recovering the PD cluster", zap.Uint64("store-id", storeID))
+		return false
+	}
 	return store.DownTime() >= c.cluster.GetOpts().GetMaxStoreDownTime()
 }
 

server/schedule/checker/rule_checker_test.go

@@ -361,7 +361,6 @@ func (suite *ruleCheckerTestSuite) TestFixRuleWitness() {
 	op := suite.rc.Check(suite.cluster.GetRegion(1))
 	suite.NotNil(op)
 	suite.Equal("add-rule-peer", op.Desc())
-	fmt.Println(op)
 	suite.Equal(uint64(3), op.Step(0).(operator.AddLearner).ToStore)
 	suite.True(op.Step(0).(operator.AddLearner).IsWitness)
 }
@@ -685,6 +684,132 @@ func (suite *ruleCheckerTestSuite) TestPriorityFixOrphanPeer() {
 	suite.Equal("remove-orphan-peer", op.Desc())
 }
 
+func (suite *ruleCheckerTestSuite) TestPriorityFitHealthWithDifferentRole1() {
+	suite.cluster.SetEnableUseJointConsensus(true)
+	suite.cluster.AddLabelsStore(1, 1, map[string]string{"host": "host1"})
+	suite.cluster.AddLabelsStore(2, 1, map[string]string{"host": "host2"})
+	suite.cluster.AddLabelsStore(3, 1, map[string]string{"host": "host3"})
+	suite.cluster.AddLabelsStore(4, 1, map[string]string{"host": "host4"})
+	suite.cluster.AddRegionWithLearner(1, 1, []uint64{2, 3}, []uint64{4})
+	r1 := suite.cluster.GetRegion(1)
+	suite.cluster.GetStore(3).GetMeta().LastHeartbeat = time.Now().Add(-31 * time.Minute).UnixNano()
+
+	// set peer3 to pending and down
+	r1 = r1.Clone(core.WithPendingPeers([]*metapb.Peer{r1.GetPeer(3)}))
+	r1 = r1.Clone(core.WithDownPeers([]*pdpb.PeerStats{
+		{
+			Peer:        r1.GetStorePeer(3),
+			DownSeconds: 30000,
+		},
+	}))
+	suite.cluster.PutRegion(r1)
+
+	op := suite.rc.Check(suite.cluster.GetRegion(1))
+	suite.Equal(uint64(3), op.Step(0).(operator.ChangePeerV2Enter).DemoteVoters[0].ToStore)
+	suite.Equal(uint64(4), op.Step(0).(operator.ChangePeerV2Enter).PromoteLearners[0].ToStore)
+	suite.Equal(uint64(3), op.Step(1).(operator.ChangePeerV2Leave).DemoteVoters[0].ToStore)
+	suite.Equal(uint64(4), op.Step(1).(operator.ChangePeerV2Leave).PromoteLearners[0].ToStore)
+	suite.Equal("replace-down-peer-with-orphan-peer", op.Desc())
+
+	// set peer3 only pending
+	r1 = r1.Clone(core.WithDownPeers(nil))
+	suite.cluster.PutRegion(r1)
+	op = suite.rc.Check(suite.cluster.GetRegion(1))
+	suite.Nil(op)
+}
+
+func (suite *ruleCheckerTestSuite) TestPriorityFitHealthWithDifferentRole2() {
+	suite.cluster.SetEnableUseJointConsensus(true)
+	suite.cluster.AddLabelsStore(1, 1, map[string]string{"host": "host1"})
+	suite.cluster.AddLabelsStore(2, 1, map[string]string{"host": "host2"})
+	suite.cluster.AddLabelsStore(3, 1, map[string]string{"host": "host3"})
+	suite.cluster.AddLabelsStore(4, 1, map[string]string{"host": "host4"})
+	suite.cluster.AddLabelsStore(5, 1, map[string]string{"host": "host5"})
+	suite.cluster.AddLeaderRegion(1, 1, 2, 3, 4, 5)
+	r1 := suite.cluster.GetRegion(1)
+
+	// set peer3 to pending and down, and peer 3 to learner, and store 3 is down
+	suite.cluster.GetStore(3).GetMeta().LastHeartbeat = time.Now().Add(-31 * time.Minute).UnixNano()
+	r1 = r1.Clone(core.WithLearners([]*metapb.Peer{r1.GetPeer(3)}))
+	r1 = r1.Clone(
+		core.WithPendingPeers([]*metapb.Peer{r1.GetPeer(3)}),
+		core.WithDownPeers([]*pdpb.PeerStats{
+			{
+				Peer:        r1.GetStorePeer(3),
+				DownSeconds: 30000,
+			},
+		}),
+	)
+	suite.cluster.PutRegion(r1)
+
+	// default and test group => 3 voter  + 1 learner
+	err := suite.ruleManager.SetRule(&placement.Rule{
+		GroupID: "test",
+		ID:      "10",
+		Role:    placement.Learner,
+		Count:   1,
+	})
+	suite.NoError(err)
+
+	op := suite.rc.Check(suite.cluster.GetRegion(1))
+	suite.Equal(uint64(5), op.Step(0).(operator.ChangePeerV2Enter).DemoteVoters[0].ToStore)
+	suite.Equal(uint64(3), op.Step(1).(operator.RemovePeer).FromStore)
+	suite.Equal("replace-down-peer-with-orphan-peer", op.Desc())
+}
+
+func (suite *ruleCheckerTestSuite) TestPriorityFitHealthPeersAndTiFlash() {
+	suite.cluster.SetEnableUseJointConsensus(true)
+	suite.cluster.AddLabelsStore(1, 1, map[string]string{"host": "host1"})
+	suite.cluster.AddLabelsStore(2, 1, map[string]string{"host": "host2"})
+	suite.cluster.AddLabelsStore(3, 1, map[string]string{"host": "host3"})
+	suite.cluster.AddLabelsStore(4, 1, map[string]string{"host": "host4", "engine": "tiflash"})
+	suite.cluster.AddRegionWithLearner(1, 1, []uint64{2, 3}, []uint64{4})
+	rule := &placement.Rule{
+		GroupID: "pd",
+		ID:      "test",
+		Role:    placement.Voter,
+		Count:   3,
+	}
+	rule2 := &placement.Rule{
+		GroupID: "pd",
+		ID:      "test2",
+		Role:    placement.Learner,
+		Count:   1,
+		LabelConstraints: []placement.LabelConstraint{
+			{
+				Key:    "engine",
+				Op:     placement.In,
+				Values: []string{"tiflash"},
+			},
+		},
+	}
+	suite.ruleManager.SetRule(rule)
+	suite.ruleManager.SetRule(rule2)
+	suite.ruleManager.DeleteRule("pd", "default")
+
+	r1 := suite.cluster.GetRegion(1)
+	// set peer3 to pending and down
+	r1 = r1.Clone(core.WithPendingPeers([]*metapb.Peer{r1.GetPeer(3)}))
+	r1 = r1.Clone(core.WithDownPeers([]*pdpb.PeerStats{
+		{
+			Peer:        r1.GetStorePeer(3),
+			DownSeconds: 30000,
+		},
+	}))
+	suite.cluster.PutRegion(r1)
+	suite.cluster.GetStore(3).GetMeta().LastHeartbeat = time.Now().Add(-31 * time.Minute).UnixNano()
+
+	op := suite.rc.Check(suite.cluster.GetRegion(1))
+	// should not promote tiflash peer
+	suite.Nil(op)
+
+	// scale a node, can replace the down peer
+	suite.cluster.AddLabelsStore(5, 1, map[string]string{"host": "host5"})
+	op = suite.rc.Check(suite.cluster.GetRegion(1))
+	suite.NotNil(op)
+	suite.Equal("replace-rule-down-peer", op.Desc())
+}
+
 func (suite *ruleCheckerTestSuite) TestIssue3293() {
 	suite.cluster.AddLabelsStore(1, 1, map[string]string{"host": "host1"})
 	suite.cluster.AddLabelsStore(2, 1, map[string]string{"host": "host1"})

server/schedule/operator/builder.go

@@ -403,7 +403,6 @@ func (b *Builder) Build(kind OpKind) (*Operator, error) {
 	if brief, b.err = b.prepareBuild(); b.err != nil {
 		return nil, b.err
 	}
-
 	if b.useJointConsensus {
 		kind, b.err = b.buildStepsWithJointConsensus(kind)
 	} else {
@@ -549,6 +548,10 @@ func (b *Builder) brief() string {
 		return fmt.Sprintf("%s: store %s to %s", op, b.toRemove, b.toAdd)
 	case len(b.toAdd) > 0:
 		return fmt.Sprintf("add peer: store %s", b.toAdd)
+	case len(b.toRemove) > 0 && len(b.toPromote) > 0:
+		return fmt.Sprintf("promote peer: store %s, rm peer: store %s", b.toRemove, b.toPromote)
+	case len(b.toRemove) > 0 && len(b.toDemote) > 0:
+		return fmt.Sprintf("demote peer: store %s, rm peer: store %s", b.toDemote, b.toRemove)
 	case len(b.toRemove) > 0:
 		return fmt.Sprintf("rm peer: store %s", b.toRemove)
 	case len(b.toPromote) > 0:

server/schedule/operator/create_operator.go

@@ -50,6 +50,25 @@ func CreatePromoteLearnerOperator(desc string, ci ClusterInformer, region *core.
 		Build(0)
 }
 
+// CreatePromoteLearnerOperatorAndRemovePeer creates an operator that promotes a learner and removes a peer.
+func CreatePromoteLearnerOperatorAndRemovePeer(desc string, ci ClusterInformer, region *core.RegionInfo, toPromote *metapb.Peer, toRemove *metapb.Peer) (*Operator, error) {
+	return NewBuilder(desc, ci, region).
+		PromoteLearner(toPromote.GetStoreId()).
+		RemovePeer(toRemove.GetStoreId()).
+		Build(0)
+}
+
+// CreateDemoteLearnerOperatorAndRemovePeer creates an operator that demotes a learner and removes a peer.
+func CreateDemoteLearnerOperatorAndRemovePeer(desc string, ci ClusterInformer, region *core.RegionInfo, toDemote *metapb.Peer, toRemove *metapb.Peer) (*Operator, error) {
+	if !ci.GetOpts().IsUseJointConsensus() {
+		return nil, errors.Errorf("cannot build demote learner operator due to disabling using joint state")
+	}
+	return NewBuilder(desc, ci, region).
+		DemoteVoter(toDemote.GetStoreId()).
+		RemovePeer(toRemove.GetStoreId()).
+		Build(0)
+}
+
 // CreateRemovePeerOperator creates an operator that removes a peer from region.
 func CreateRemovePeerOperator(desc string, ci ClusterInformer, kind OpKind, region *core.RegionInfo, storeID uint64) (*Operator, error) {
 	return NewBuilder(desc, ci, region).
@@ -238,7 +257,7 @@ func CreateLeaveJointStateOperator(desc string, ci ClusterInformer, origin *core
 	b := NewBuilder(desc, ci, origin, SkipOriginJointStateCheck, SkipPlacementRulesCheck)
 
 	if b.err == nil && !core.IsInJointState(origin.GetPeers()...) {
-		b.err = errors.Errorf("cannot build leave joint state operator for region which is not in joint state")
+		b.err = errors.Errorf("cannot build leave joint state operator due to disabling using joint state")
 	}
 
 	if b.err != nil {

server/schedule/operator/operator.go

@@ -87,7 +87,7 @@ func (o *Operator) Sync(other *Operator) {
 func (o *Operator) String() string {
 	stepStrs := make([]string, len(o.steps))
 	for i := range o.steps {
-		stepStrs[i] = o.steps[i].String()
+		stepStrs[i] = fmt.Sprintf("%d:{%s}", i, o.steps[i].String())
 	}
 	s := fmt.Sprintf("%s {%s} (kind:%s, region:%v(%v, %v), createAt:%s, startAt:%s, currentStep:%v, size:%d, steps:[%s]，timeout:[%s])",
 		o.desc, o.brief, o.kind, o.regionID, o.regionEpoch.GetVersion(), o.regionEpoch.GetConfVer(), o.GetCreateTime(),