scheduler: add source filter in transferLeaderIn

Signed-off-by: lhy1024 <[email protected]>

server/schedulers/balance_leader.go

@@ -377,6 +377,7 @@ func (l *balanceLeaderScheduler) Schedule(cluster schedule.Cluster, dryRun bool)
 	usedRegions := make(map[uint64]struct{})
 
 	result := make([]*operator.Operator, 0, batch)
+	log.Info("start to schedule leader", zap.String("scheduler", l.GetName()), zap.Int("source-count", len(sourceCandidate.stores)), zap.Int("target-count", len(targetCandidate.stores)))
 	for sourceCandidate.hasStore() || targetCandidate.hasStore() {
 		// first choose source
 		if sourceCandidate.hasStore() {
@@ -531,6 +532,13 @@ func (l *balanceLeaderScheduler) transferLeaderIn(solver *solver, collector *pla
 	if leaderFilter := filter.NewPlacementLeaderSafeguard(l.GetName(), opts, solver.GetBasicCluster(), solver.GetRuleManager(), solver.region, solver.source, false /*allowMoveLeader*/); leaderFilter != nil {
 		finalFilters = append(l.filters, leaderFilter)
 	}
+	sources := filter.NewCandidates([]*core.StoreInfo{solver.source}).
+		FilterSource(opts, nil, l.filterCounter, finalFilters...).PickAll()
+	if len(sources) == 0 {
+		log.Debug("store cannot be used as source", zap.String("scheduler", l.GetName()), zap.Uint64("store-id", solver.source.GetID()))
+		schedulerCounter.WithLabelValues(l.GetName(), "store-source-filtered").Inc()
+		return nil
+	}
 	target := filter.NewCandidates([]*core.StoreInfo{solver.target}).
 		FilterTarget(opts, nil, l.filterCounter, finalFilters...).
 		PickFirst()