*: scatter the leader distribution in the specified range #1037
Conversation
|
No need to consider any store state filters or location labels? |
| loopEnd := false | ||
| for { | ||
| collect := cluster.ScanRegions(startKey, scanLimit) | ||
| if len(collect) == 0 || loopEnd { |
There was a problem hiding this comment.
Better to check loopEnd before scanRegions.
| break | ||
| } | ||
| for _, r := range collect { | ||
| if bytes.Compare(r.StartKey, l.endKey) < 0 { |
There was a problem hiding this comment.
Need to consider when l.endKey is empty.
|
|
||
| if source == target { | ||
| return nil | ||
| } |
There was a problem hiding this comment.
The usage is to create the scheduler and delete it after some time?
If the scheduler is long-term existence, consider shouldBalance to make sure balance will stabilize.
server/schedule/mockcluster.go
Outdated
| @@ -304,6 +313,7 @@ func (mc *MockCluster) ApplyOperator(op *Operator) { | |||
| StoreId: s.ToStore, | |||
| } | |||
| region.Peers = append(region.Peers, peer) | |||
| region.Voters = region.Peers | |||
There was a problem hiding this comment.
there is a method region.AddPeer() for test use.
| } | ||
| startKey = r.EndKey | ||
| } | ||
|
|
There was a problem hiding this comment.
please delete redundant line
| } | ||
| return nil | ||
| } | ||
| steps = append(steps, schedule.AddPeer{ToStore: target.GetId(), PeerID: peer.GetId()}) |
There was a problem hiding this comment.
I prefer appending all steps in one append().
| source = s | ||
| } | ||
| } | ||
| for _, s := range stores { |
There was a problem hiding this comment.
please extract these logics to a new selector.
| op := schedule.NewOperator("scatter-range-leader", sourceRegion.GetId(), schedule.OpRange|schedule.OpLeader|schedule.OpBalance, steps...) | ||
| return []*schedule.Operator{op} | ||
| } | ||
| func (l *scatterRangeLeaderScheduler) shouldBalance(sourceID, targetID uint64, sourceRegion *core.RegionInfo, regions *core.RegionsInfo) bool { |
There was a problem hiding this comment.
please add blank line between functions.
| } | ||
| return nil | ||
| } | ||
| steps = append(steps, schedule.AddPeer{ToStore: target.GetId(), PeerID: peer.GetId()}) |
There was a problem hiding this comment.
I prefer appending all steps in one append().
| break | ||
| } | ||
| } | ||
| if _, ok := followers[target.GetId()]; ok { |
There was a problem hiding this comment.
these can be removed to if body of l.shouldBalance.
| } | ||
| if _, ok := followers[target.GetId()]; ok { | ||
| step := schedule.TransferLeader{FromStore: source.GetId(), ToStore: target.GetId()} | ||
| op := schedule.NewOperator("scatter-range-leader", sourceRegion.GetId(), schedule.OpRange|schedule.OpLeader, step) |
| } | ||
| peer, err := cluster.AllocPeer(target.GetId()) | ||
| if err != nil { | ||
| if _, ok := followers[target.GetId()]; ok { |
There was a problem hiding this comment.
if target is in followers, it can't reach here.
server/schedule/basic_cluster.go
Outdated
| } | ||
|
|
||
| return m | ||
| return influence | ||
| } | ||
|
|
||
| // OpInfluence is a map of StoreInfluence. |
server/schedule/basic_cluster.go
Outdated
| } | ||
| return storeInfluence | ||
| } | ||
|
|
||
| // GetRegionInfluence gets regionInfluence of specific region. | ||
| func (m OpInfluence) GetRegionInfluence() map[uint64]*Operator { |
There was a problem hiding this comment.
it should be GetRegionsInfluence
server/schedule/basic_cluster.go
Outdated
| @@ -203,3 +217,80 @@ func (bc *BasicCluster) CheckWriteStatus(region *core.RegionInfo) (bool, *core.R | |||
| func (bc *BasicCluster) CheckReadStatus(region *core.RegionInfo) (bool, *core.RegionStat) { | |||
| return bc.HotCache.CheckRead(region, bc.Stores) | |||
| } | |||
|
|
|||
| // RangeCluster isolates the cluster by range. | |||
| type RangeCluster struct { | |||
There was a problem hiding this comment.
I prefer removing it into a new file.
server/schedulers/scatter_range.go
Outdated
| endKey []byte | ||
| balanceLeader schedule.Scheduler | ||
| balanceRegion schedule.Scheduler | ||
| operators []*schedule.Operator |
| ops := l.balanceLeader.Schedule(c, schedule.NewOpInfluence(influence, cluster)) | ||
| if len(ops) > 0 { | ||
| ops[0].SetDesc(fmt.Sprintf("scatter-range-leader-%s", l.rangeName)) | ||
| ops[0].AttachKind(schedule.OpRange) |
| ops = l.balanceRegion.Schedule(c, schedule.NewOpInfluence(influence, cluster)) | ||
| if len(ops) > 0 { | ||
| ops[0].SetDesc(fmt.Sprintf("scatter-range-region-%s", l.rangeName)) | ||
| ops[0].AttachKind(schedule.OpRange) |
server/schedulers/scatter_range.go
Outdated
| ops[0].AttachKind(schedule.OpRange) | ||
| return ops | ||
| } | ||
| return nil |
server/schedule/operator_kind.go
Outdated
| @@ -32,6 +32,7 @@ const ( | |||
| OpReplica // Initiated by replica checkers. | |||
| OpBalance // Initiated by balancers. | |||
| OpMerge // Initiated by merge checkers. | |||
| OpRange | |||
server/schedule/operator.go
Outdated
| @@ -275,6 +275,16 @@ func (o *Operator) Desc() string { | |||
| return o.desc | |||
| } | |||
|
|
|||
| // SetDesc set the description for the operator. | |||
server/schedule/operator.go
Outdated
| o.desc = desc | ||
| } | ||
|
|
||
| // AttachKind attach a operator kind for the operator. |
|
LGTM. |
|
/run-all-tests |
|
/run-all-tests |
|
/run-all-tests |
add a scheduler for balance leader distribution in a range.