scheduler: consider leader score when evict leader

pkg/schedule/checker/rule_checker.go

@@ -312,7 +312,7 @@ func (c *RuleChecker) fixLooseMatchPeer(region *core.RegionInfo, fit *placement.
 	if region.GetLeader().GetId() != peer.GetId() && rf.Rule.Role == placement.Leader {
 		ruleCheckerFixLeaderRoleCounter.Inc()
 		if c.allowLeader(fit, peer) {
-			return operator.CreateTransferLeaderOperator("fix-leader-role", c.cluster, region, peer.GetStoreId(), []uint64{}, 0)
+			return operator.CreateTransferLeaderOperator("fix-leader-role", c.cluster, region, peer.GetStoreId(), 0)
 		}
 		ruleCheckerNotAllowLeaderCounter.Inc()
 		return nil, errs.ErrPeerCannotBeLeader
@@ -321,7 +321,7 @@ func (c *RuleChecker) fixLooseMatchPeer(region *core.RegionInfo, fit *placement.
 		ruleCheckerFixFollowerRoleCounter.Inc()
 		for _, p := range region.GetPeers() {
 			if c.allowLeader(fit, p) {
-				return operator.CreateTransferLeaderOperator("fix-follower-role", c.cluster, region, p.GetStoreId(), []uint64{}, 0)
+				return operator.CreateTransferLeaderOperator("fix-follower-role", c.cluster, region, p.GetStoreId(), 0)
 			}
 		}
 		ruleCheckerNoNewLeaderCounter.Inc()

pkg/schedule/config/config_provider.go

@@ -67,7 +67,6 @@ type SchedulerConfigProvider interface {
 	IsTraceRegionFlow() bool
 
 	GetTolerantSizeRatio() float64
-	GetLeaderSchedulePolicy() constant.SchedulePolicy
 
 	IsDebugMetricsEnabled() bool
 	IsDiagnosticAllowed() bool
@@ -114,6 +113,7 @@ type SharedConfigProvider interface {
 	IsCrossTableMergeEnabled() bool
 	IsOneWayMergeEnabled() bool
 	GetMergeScheduleLimit() uint64
+	GetLeaderSchedulePolicy() constant.SchedulePolicy
 	GetRegionScoreFormulaVersion() string
 	GetSchedulerMaxWaitingOperator() uint64
 	GetStoreLimitByType(uint64, storelimit.Type) float64

pkg/schedule/filter/comparer.go

@@ -27,6 +27,7 @@ type StoreComparer func(a, b *core.StoreInfo) int
 // score.
 func RegionScoreComparer(conf config.SharedConfigProvider) StoreComparer {
 	return func(a, b *core.StoreInfo) int {
+		// TODO: we should use the real time delta data to calculate the score.
 		sa := a.RegionScore(conf.GetRegionScoreFormulaVersion(), conf.GetHighSpaceRatio(), conf.GetLowSpaceRatio(), 0)
 		sb := b.RegionScore(conf.GetRegionScoreFormulaVersion(), conf.GetHighSpaceRatio(), conf.GetLowSpaceRatio(), 0)
 		switch {
@@ -40,6 +41,25 @@ func RegionScoreComparer(conf config.SharedConfigProvider) StoreComparer {
 	}
 }
 
+// LeaderScoreComparer creates a StoreComparer to sort store by leader
+// score.
+func LeaderScoreComparer(conf config.SharedConfigProvider) StoreComparer {
+	leaderSchedulePolicy := conf.GetLeaderSchedulePolicy()
+	return func(a, b *core.StoreInfo) int {
+		// TODO: we should use the real time delta data to calculate the score.
+		sa := a.LeaderScore(leaderSchedulePolicy, 0)
+		sb := b.LeaderScore(leaderSchedulePolicy, 0)
+		switch {
+		case sa > sb:
+			return 1
+		case sa < sb:
+			return -1
+		default:
+			return 0
+		}
+	}
+}
+
 // IsolationComparer creates a StoreComparer to sort store by isolation score.
 func IsolationComparer(locationLabels []string, regionStores []*core.StoreInfo) StoreComparer {
 	return func(a, b *core.StoreInfo) int {

pkg/schedule/filter/comparer_test.go

@@ -0,0 +1,62 @@
+// Copyright 2025 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,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package filter
+
+import (
+	"math/rand"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/pingcap/kvproto/pkg/metapb"
+
+	"github.com/tikv/pd/pkg/core"
+	"github.com/tikv/pd/pkg/mock/mockconfig"
+)
+
+func TestRegionCompare(t *testing.T) {
+	re := require.New(t)
+	ids := []uint64{1, 2, 3, 4, 5}
+	stores := make([]*core.StoreInfo, 0, len(ids))
+	for _, id := range ids {
+		stores = append(stores, core.NewStoreInfo(
+			&metapb.Store{Id: id},
+			core.SetRegionSize(int64(6-id)*1000),
+		))
+	}
+	cs := NewCandidates(rand.New(rand.NewSource(time.Now().UnixNano())), stores)
+	cfg := mockconfig.NewTestOptions()
+	re.Equal(uint64(1), cs.PickFirst().GetID())
+	cs.Sort(RegionScoreComparer(cfg))
+	re.Equal(uint64(5), cs.PickFirst().GetID())
+}
+
+func TestLeaderCompare(t *testing.T) {
+	re := require.New(t)
+	ids := []uint64{1, 2, 3, 4, 5}
+	stores := make([]*core.StoreInfo, 0, len(ids))
+	for _, id := range ids {
+		stores = append(stores, core.NewStoreInfo(
+			&metapb.Store{Id: id},
+			core.SetLeaderCount(int(6-id)*1000),
+		))
+	}
+	cs := NewCandidates(rand.New(rand.NewSource(time.Now().UnixNano())), stores)
+	cfg := mockconfig.NewTestOptions()
+	re.Equal(uint64(1), cs.PickFirst().GetID())
+	cs.Sort(LeaderScoreComparer(cfg))
+	re.Equal(uint64(5), cs.PickFirst().GetID())
+}

pkg/schedule/handler/handler.go

@@ -420,7 +420,7 @@ func (h *Handler) AddTransferLeaderOperator(regionID uint64, storeID uint64) err
 		return errors.Errorf("region has no voter in store %v", storeID)
 	}
 
-	op, err := operator.CreateTransferLeaderOperator("admin-transfer-leader", c, region, newLeader.GetStoreId(), []uint64{}, operator.OpAdmin)
+	op, err := operator.CreateTransferLeaderOperator("admin-transfer-leader", c, region, newLeader.GetStoreId(), operator.OpAdmin)
 	if err != nil {
 		log.Debug("fail to create transfer leader operator", errs.ZapError(err))
 		return err

pkg/schedule/operator/builder.go

@@ -297,25 +297,6 @@ func (b *Builder) SetLeader(storeID uint64) *Builder {
 	return b
 }
 
-// SetLeaders records all valid target leaders in Builder.
-func (b *Builder) SetLeaders(storeIDs []uint64) *Builder {
-	if b.err != nil {
-		return b
-	}
-	sort.Slice(storeIDs, func(i, j int) bool { return storeIDs[i] < storeIDs[j] })
-	for _, storeID := range storeIDs {
-		peer := b.targetPeers[storeID]
-		if peer == nil || core.IsLearner(peer) || b.unhealthyPeers[storeID] != nil {
-			continue
-		}
-		b.targetLeaderStoreIDs = append(b.targetLeaderStoreIDs, storeID)
-	}
-	// Don't need to check if there's valid target, because `targetLeaderStoreIDs`
-	// can be empty if this is not a multi-target evict leader operation. Besides,
-	// `targetLeaderStoreID` must be valid and there must be at least one valid target.
-	return b
-}
-
 // SetPeers resets the target peer list.
 //
 // If peer's ID is 0, the builder will allocate a new ID later. If current

pkg/schedule/operator/create_operator.go

@@ -80,10 +80,9 @@ func CreateRemovePeerOperator(desc string, ci sche.SharedCluster, kind OpKind, r
 }
 
 // CreateTransferLeaderOperator creates an operator that transfers the leader from a source store to a target store.
-func CreateTransferLeaderOperator(desc string, ci sche.SharedCluster, region *core.RegionInfo, targetStoreID uint64, targetStoreIDs []uint64, kind OpKind) (*Operator, error) {
+func CreateTransferLeaderOperator(desc string, ci sche.SharedCluster, region *core.RegionInfo, targetStoreID uint64, kind OpKind) (*Operator, error) {
 	return NewBuilder(desc, ci, region, SkipOriginJointStateCheck).
 		SetLeader(targetStoreID).
-		SetLeaders(targetStoreIDs).
 		Build(kind)
 }
 

pkg/schedule/operator/create_operator_test.go

@@ -425,7 +425,7 @@ func (suite *createOperatorTestSuite) TestCreateTransferLeaderOperator() {
 	}
 	for _, testCase := range testCases {
 		region := core.NewRegionInfo(&metapb.Region{Id: 1, Peers: testCase.originPeers}, testCase.originPeers[0])
-		op, err := CreateTransferLeaderOperator("test", suite.cluster, region, testCase.targetLeaderStoreID, []uint64{}, 0)
+		op, err := CreateTransferLeaderOperator("test", suite.cluster, region, testCase.targetLeaderStoreID, 0)
 
 		if testCase.isErr {
 			re.Error(err)

pkg/schedule/schedulers/balance_leader.go

@@ -537,7 +537,7 @@ func (s *balanceLeaderScheduler) createOperator(solver *solver, collector *plan.
 	}
 	solver.Step++
 	defer func() { solver.Step-- }()
-	op, err := operator.CreateTransferLeaderOperator(s.GetName(), solver, solver.Region, solver.targetStoreID(), []uint64{}, operator.OpLeader)
+	op, err := operator.CreateTransferLeaderOperator(s.GetName(), solver, solver.Region, solver.targetStoreID(), operator.OpLeader)
 	if err != nil {
 		log.Debug("fail to create balance leader operator", errs.ZapError(err))
 		if collector != nil {

pkg/schedule/schedulers/evict_leader.go

@@ -363,19 +363,12 @@
 		filters = append(filters, &filter.StoreStateFilter{ActionScope: name, TransferLeader: true, OperatorLevel: constant.Urgent})
 		candidates := filter.NewCandidates(r, cluster.GetFollowerStores(region)).
 			FilterTarget(cluster.GetSchedulerConfig(), nil, nil, filters...)
-		// Compatible with old TiKV transfer leader logic.
-		target := candidates.RandomPick()
-		targets := candidates.PickAll()
-		// `targets` MUST contains `target`, so only needs to check if `target` is nil here.
-		if target == nil {
+
+		if len(candidates.Stores) == 0 {
 			evictLeaderNoTargetStoreCounter.Inc()
 			continue
 		}
-		targetIDs := make([]uint64, 0, len(targets))
-		for _, t := range targets {
-			targetIDs = append(targetIDs, t.GetID())
-		}
-		op, err := operator.CreateTransferLeaderOperator(name, cluster, region, target.GetID(), targetIDs, operator.OpLeader)
+		op, err := createOperatorWithSort(name, cluster, candidates, region)
 		if err != nil {
 			log.Debug("fail to create evict leader operator", errs.ZapError(err))
 			continue
@@ -387,6 +380,22 @@
 	return ops
 }
 
+func createOperatorWithSort(name string, cluster sche.SchedulerCluster, candidates *filter.StoreCandidates, region *core.RegionInfo) (*operator.Operator, error) {
+	// we will pick low leader score store firstly.
+	candidates.Sort(filter.LeaderScoreComparer(cluster.GetSharedConfig()))
+	var (
+		op  *operator.Operator
+		err error
+	)
+	for _, target := range candidates.Stores {
+		op, err = operator.CreateTransferLeaderOperator(name, cluster, region, target.GetID(), operator.OpLeader)
+		if op != nil && err == nil {
+			return op, err
+		}
+	}
+	return op, err
+}
+
 type evictLeaderHandler struct {
 	rd     *render.Render
 	config *evictLeaderSchedulerConfig

pkg/schedule/schedulers/grant_hot_region.go

@@ -316,7 +316,7 @@
 	dstStore := &metapb.Peer{StoreId: destStoreIDs[i]}
 
 	if isLeader {
-		op, err = operator.CreateTransferLeaderOperator(s.GetName()+"-leader", cluster, srcRegion, dstStore.StoreId, []uint64{}, operator.OpLeader)
+		op, err = operator.CreateTransferLeaderOperator(s.GetName()+"-leader", cluster, srcRegion, dstStore.StoreId, operator.OpLeader)
 	} else {
 		op, err = operator.CreateMovePeerOperator(s.GetName()+"-move", cluster, srcRegion, operator.OpRegion|operator.OpLeader, srcStore.GetID(), dstStore)
 	}

pkg/schedule/schedulers/hot_region.go

@@ -1475,7 +1475,6 @@ func (bs *balanceSolver) createOperator(region *core.RegionInfo, srcStoreID, dst
 			bs,
 			region,
 			dstStoreID,
-			[]uint64{},
 			operator.OpHotRegion)
 	} else {
 		srcPeer := region.GetStorePeer(srcStoreID) // checked in `filterHotPeers`

pkg/schedule/schedulers/hot_region_test.go

@@ -150,7 +150,7 @@ func checkGCPendingOpInfos(re *require.Assertions, enablePlacementRules bool) {
 		case movePeer:
 			op, err = operator.CreateMovePeerOperator("move-peer-test", tc, region, operator.OpAdmin, 2, &metapb.Peer{Id: region.GetID()*10000 + 1, StoreId: 4})
 		case transferLeader:
-			op, err = operator.CreateTransferLeaderOperator("transfer-leader-test", tc, region, 2, []uint64{}, operator.OpAdmin)
+			op, err = operator.CreateTransferLeaderOperator("transfer-leader-test", tc, region, 2, operator.OpAdmin)
 		}
 		re.NoError(err)
 		re.NotNil(op)

pkg/schedule/schedulers/label.go

@@ -102,7 +102,7 @@ func (s *labelScheduler) Schedule(cluster sche.SchedulerCluster, _ bool) ([]*ope
 				continue
 			}
 
-			op, err := operator.CreateTransferLeaderOperator("label-reject-leader", cluster, region, target.GetID(), []uint64{}, operator.OpLeader)
+			op, err := operator.CreateTransferLeaderOperator("label-reject-leader", cluster, region, target.GetID(), operator.OpLeader)
 			if err != nil {
 				log.Debug("fail to create transfer label reject leader operator", errs.ZapError(err))
 				return nil, nil

pkg/schedule/schedulers/shuffle_leader.go

@@ -89,7 +89,7 @@ func (s *shuffleLeaderScheduler) Schedule(cluster sche.SchedulerCluster, _ bool)
 		shuffleLeaderNoFollowerCounter.Inc()
 		return nil, nil
 	}
-	op, err := operator.CreateTransferLeaderOperator(s.GetName(), cluster, region, targetStore.GetID(), []uint64{}, operator.OpAdmin)
+	op, err := operator.CreateTransferLeaderOperator(s.GetName(), cluster, region, targetStore.GetID(), operator.OpAdmin)
 	if err != nil {
 		log.Debug("fail to create shuffle leader operator", errs.ZapError(err))
 		return nil, nil

pkg/schedule/schedulers/transfer_witness_leader.go

@@ -99,19 +99,22 @@
 	filters = append(filters, filter.NewExcludedFilter(name, nil, unhealthyPeerStores),
 		&filter.StoreStateFilter{ActionScope: name, TransferLeader: true, OperatorLevel: constant.Urgent})
 	candidates := filter.NewCandidates(r, cluster.GetFollowerStores(region)).FilterTarget(cluster.GetSchedulerConfig(), nil, nil, filters...)
-	// Compatible with old TiKV transfer leader logic.
-	target := candidates.RandomPick()
-	targets := candidates.PickAll()
-	// `targets` MUST contains `target`, so only needs to check if `target` is nil here.
-	if target == nil {
+	if len(candidates.Stores) == 0 {
 		transferWitnessLeaderNoTargetStoreCounter.Inc()
 		return nil, errors.New("no target store to schedule")
 	}
-	targetIDs := make([]uint64, 0, len(targets))
-	for _, t := range targets {
-		targetIDs = append(targetIDs, t.GetID())
+	// TODO: also add sort such as evict leader
+	var (
+		op  *operator.Operator
+		err error
+	)
+	for _, target := range candidates.Stores {
+		op, err = operator.CreateTransferLeaderOperator(name, cluster, region, target.GetID(), operator.OpLeader)
+		if op != nil && err == nil {
+			return op, err
+		}
 	}
-	return operator.CreateTransferLeaderOperator(name, cluster, region, target.GetID(), targetIDs, operator.OpWitnessLeader)
+	return op, err
 }
 
 // RecvRegionInfo receives a checked region from coordinator

pkg/utils/operatorutil/operator_check.go

@@ -53,7 +53,7 @@ func CheckMultiTargetTransferLeader(re *require.Assertions, op *operator.Operato
 	re.Equal(1, op.Len())
 	expectedOps := make([]any, 0, len(targetIDs))
 	for _, targetID := range targetIDs {
-		expectedOps = append(expectedOps, operator.TransferLeader{FromStore: sourceID, ToStore: targetID, ToStores: targetIDs})
+		expectedOps = append(expectedOps, operator.TransferLeader{FromStore: sourceID, ToStore: targetID})
 	}
 	re.Contains(expectedOps, op.Step(0))
 	kind |= operator.OpLeader

plugin/scheduler_example/evict_leader.go

@@ -230,7 +230,7 @@ func (s *evictLeaderScheduler) Schedule(cluster sche.SchedulerCluster, _ bool) (
 		if target == nil {
 			continue
 		}
-		op, err := operator.CreateTransferLeaderOperator(s.GetName(), cluster, region, target.GetID(), []uint64{}, operator.OpLeader)
+		op, err := operator.CreateTransferLeaderOperator(s.GetName(), cluster, region, target.GetID(), operator.OpLeader)
 		if err != nil {
 			log.Debug("fail to create evict leader operator", errs.ZapError(err))
 			continue