TiDB PD 调度器模块

作者：薛港-移动云

原文来源：https://tidb.net/blog/9cdf0f5f

1 调度器管理模块coordinator\ 初始化过程分析\ 因为调度器管理模块从属于 RaftCluster 模块,所以启动 RaftCluster 模块的时候，我们初始化调度器管理模块，从初始化代码，能够看出来，coordinator模块依赖上层模块raft cluster以及用于发送调作的模块HeartbeatStreams。\ raft cluster：用于存储当前系统最新的region以及store信息，基于这些信息，我们分析，统计\ 当前cluster数据分布现状，以及冷热现状。调度器模块需要这些信息，然后基于一些\ rule,或者为了系统健康或者更好的用户体验的想法，生成一系列调度操作

HeartbeatStreams：生成的调度操作，最终需要发送给tikv节点，用于实施调度操作，从而达\ 到用户或者系统定义健康的期望。这个模块主要用于连接管理，发现调度操作\ // Start starts a cluster.\ func (c *RaftCluster) Start(s Server) error {\ c.coordinator = newCoordinator(c.ctx, cluster, s.GetHBStreams())\ }

// newCoordinator creates a new coordinator.\ func newCoordinator(ctx context.Context, cluster *RaftCluster, hbStreams *hbstream.HeartbeatStreams) *coordinator {\ ctx, cancel := context.WithCancel(ctx)\ opController := schedule.NewOperatorController(ctx, cluster, hbStreams)\ return \&coordinator{\ ctx: ctx,\ cancel: cancel,\ cluster: cluster,\ checkers: schedule.NewCheckerController(ctx, cluster, cluster.ruleManager, opController),\ regionScatterer: schedule.NewRegionScatterer(ctx, cluster),\ regionSplitter: schedule.NewRegionSplitter(cluster, schedule.NewSplitRegionsHandler(cluster, opController)),\ schedulers: make(map[string]*scheduleController),\ opController: opController,\ hbStreams: hbStreams,\ pluginInterface: schedule.NewPluginInterface(),\ }\ }

调度器管理模块，说白了，就是为了生成一系列调度操作，以及控制调度操作生成的速度，每类调度操作对应region的管理，一个region对应一个raft group,所以针对region的管理，就是针对raft group 成员的管理，例如增加/删除member，基于不同目的(负载，数据量)的raft member 位置转移等，调度模块从属关系coordinator–>RaftCluster -->Server\ 调度器coordinator模块和其它模块的关系\ type RaftCluster struct {\ coordinator *coordinator\ }

type Server struct {\ cluster *cluster.RaftCluster\ }

调度器模块定义，从调度器定义里，发现调试器模块包含一系列其它模块\ // coordinator is used to manage all schedulers and checkers to decide //if the region needs to be scheduled.\ type coordinator struct {\ sync.RWMutex

wg sync.WaitGroup ctx context.Context cancel context.CancelFunc cluster *RaftCluster checkers *schedule.CheckerController regionScatterer *schedule.RegionScatterer regionSplitter *schedule.RegionSplitter schedulers map[string]*scheduleController opController *schedule.OperatorController hbStreams *hbstream.HeartbeatStreams pluginInterface *schedule.PluginInterface }

从上面coordinator的定义，也能看出来coordinator 包含一系列子模块，例如子模块分成三类:\ 不同目的的调度操作生成模块：checkers，regionScatterer，regionSplitter\ 限制调度速度的模块：opController\ 用于调度操作发送到tikv的模块：hbStreams

生成调度子模块分析\ 2.1 子模块 CheckerController\ 这个模块主要用于定期检察所有的数据，如果发现数据副本数，或者位置等不符合用户的期望，会生成一系列的调度操作，这些调度操作会不停的针对region实施调度操作，最终达到用户的期望。CheckerController 的初始化以及定义如下：\ check 控制器的初始化过程\ checkers: schedule.NewCheckerController(ctx, cluster, cluster.ruleManager, opController)

// NewCheckerController create a new CheckerController.\ // TODO: isSupportMerge should be removed.\ func NewCheckerController(ctx context.Context, cluster opt.Cluster, ruleManager *placement.RuleManager, opController *OperatorController) *CheckerController {\ regionWaitingList := cache.NewDefaultCache(DefaultCacheSize)\ return \&CheckerController{\ cluster: cluster,\ opts: cluster.GetOpts(),\ opController: opController,\ learnerChecker: checker.NewLearnerChecker(cluster),\ replicaChecker: checker.NewReplicaChecker(cluster, regionWaitingList),\ ruleChecker: checker.NewRuleChecker(cluster, ruleManager, regionWaitingList),\ mergeChecker: checker.NewMergeChecker(ctx, cluster),\ jointStateChecker: checker.NewJointStateChecker(cluster),\ regionWaitingList: regionWaitingList,\ }\ }

Check 控制器的定义，里面包含一系列不同目的的检察器\ // 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\ jointStateChecker *checker.JointStateChecker\ regionWaitingList cache.Cache\ }\ 从检察控制模块CheckerController 定义能够看出来，这个模块包含一系列用于不同目的检察调度模块，如果每类检察发现问题，会生成对应的调度操作。例如replicaChecker 模块，这个模块主要用于检察复制数，如果发现问题，生成调度操作，并通过stream发现到对应的TIKV，最终让region达到用户期望副本数。由于每类检察调度模块，基本框架一致，我们只要分析一个检察控制模块，其它依次推理分析，本文由于篇幅不会全部展开.\ 2.1.1 模块 LearnerChecker\ 初始化，以及模块定义

learnerChecker: checker.NewLearnerChecker(cluster)

// LearnerChecker ensures region has a learner will be promoted.\ type LearnerChecker struct {\ cluster opt.Cluster\ }

// NewLearnerChecker creates a learner checker.\ func NewLearnerChecker(cluster opt.Cluster) *LearnerChecker {\ return \&LearnerChecker{\ cluster: cluster,\ }\ }

2.1.2LearnerChecker生成调度操作operator流程分析\ LearnerChecker调用chek 函数分析特定的region, 如果发现这个region存在learner角色，并且这个learner角色，满足升级到voter角色要求，那么就会创建一个operator，用于升级这个region的learndr角色的副本到voter角色，从而让这个副本能够参与后期选举。\ // Check verifies a region’s role, creating an Operator if need.\ func (l *LearnerChecker) Check(region *core.RegionInfo) *operator.Operator {\ for _, p := range region.GetLearners() {\ op, err := operator.CreatePromoteLearnerOperator(“promote-learner”, l.cluster, region, p)\ continue\ }\ return op\ }\ return nil\ }

重点分析函数CreatePromoteLearnerOperator：\ op, err := operator.CreatePromoteLearnerOperator(“promote-learner”, l.cluster, region, p)

// CreatePromoteLearnerOperator creates an operator that promotes a learner.\ func CreatePromoteLearnerOperator(desc string, cluster opt.Cluster, region *core.RegionInfo, peer *metapb.Peer) (*Operator, error) {\ return NewBuilder(desc, cluster, region).\ PromoteLearner(peer.GetStoreId()).\ Build(0)\ }\ 我们下面重点分析模块Builder模块，这个模块用于针对region生成相应operator\ build 模块定义如下

// Builder is used to create operators. Usage:\ // op, err := NewBuilder(desc, cluster, region).\ // RemovePeer(store1).\ // AddPeer(peer1).\ // SetLeader(store2).\ // Build(kind)\ // The generated Operator will choose the most appropriate execution order\ // according to various constraints.\ type Builder struct {\ // basic info\ desc string\ cluster opt.Cluster\ regionID uint64\ regionEpoch *metapb.RegionEpoch\ rules []*placement.Rule\ expectedRoles map[uint64]placement.PeerRoleType

``` // operation record originPeers peersMap unhealthyPeers peersMap originLeaderStoreID uint64 targetPeers peersMap targetLeaderStoreID uint64 err error

// skip origin check flags
skipOriginJointStateCheck bool

// build flags
allowDemote       bool
useJointConsensus bool
lightWeight       bool
forceTargetLeader bool

// intermediate states
currentPeers                         peersMap
currentLeaderStoreID                 uint64
toAdd, toRemove, toPromote, toDemote peersMap       // pending tasks.
steps                                []OpStep       // generated steps.
peerAddStep                          map[uint64]int // record at which step a peer is created.

// comparison function
stepPlanPreferFuncs []func(stepPlan) int // for buildStepsWithoutJointConsensus

} ```

Build 模块初始化过程如下\ 1.根据参数region 对象，抽取出这个region对应的所有peer成员存放store，处于Pending状态存入的store，当前的leader所在的store，\ // NewBuilder creates a Builder.\ func NewBuilder(desc string, cluster opt.Cluster, region *core.RegionInfo, opts …BuilderOption) *Builder {\ b := \&Builder{\ desc: desc,\ cluster: cluster,\ regionID: region.GetID(),\ regionEpoch: region.GetRegionEpoch(),\ }

``` // options for _, option := range opts { option(b) }

// origin peers
err := b.err
originPeers := newPeersMap()
unhealthyPeers := newPeersMap()

for _, p := range region.GetPeers() {
    originPeers.Set(p)
}

for _, p := range region.GetPendingPeers() {
    unhealthyPeers.Set(p)
}

for _, p := range region.GetDownPeers() {
    unhealthyPeers.Set(p.Peer)
}

// origin leader
originLeaderStoreID := region.GetLeader().GetStoreId()
b.rules = rules
b.originPeers = originPeers
b.unhealthyPeers = unhealthyPeers
b.originLeaderStoreID = originLeaderStoreID
b.targetPeers = originPeers.Copy()
b.allowDemote = supportJointConsensus
return b

} ```

下一步分析build的PromoteLearner方法,这个方法目的：根据你想要升级的learner peer，初始化相关build 成员 ,也就是修改build用于存放目标peer的角色 ，从之前leaner 角色，修改成metapb.PeerRole_Voter\ // CreatePromoteLearnerOperator creates an operator that promotes a learner.\ func CreatePromoteLearnerOperator(desc string, cluster opt.Cluster, region *core.RegionInfo, peer *metapb.Peer) (*Operator, error) {\ return NewBuilder(desc, cluster, region).\ PromoteLearner(peer.GetStoreId()).\ Build(0)\ }

// PromoteLearner records a promote learner operation in Builder.\ func (b *Builder) PromoteLearner(storeID uint64) *Builder {

``` b.targetPeers.Set(&metapb.Peer{ Id: peer.GetId(), StoreId: peer.GetStoreId(), Role: metapb.PeerRole_Voter, }

return b

} ```

下一步高用build的Build(0)方法，这个方法，会根据build 成员的状态，生成相应的调试操作，原理其实也很简单，build记录两部分信息，一部分信息对应原始的region 信息，另一部分对应region期望的region目标状态信息。然后两组信息对比，最终生成调度操作（用于修正region到目标状态）。

// Build creates the Operator.\ func (b *Builder) Build(kind OpKind) (*Operator, error) {\ var brief string\ if brief, b.err = b.prepareBuild(); b.err != nil {\ return nil, b.err\ }

``` kind, b.err = b.buildStepsWithoutJointConsensus(kind) }

return NewOperator(b.desc, brief, b.regionID, b.regionEpoch, kind, b.steps...), nil

} ```

两组信息对比算法如下，我们重点关注leaner到follower角色的转换.\ b.originPeers 保存的region原始状态信息\ b.targetPeers 保存的region期望的目标状态信息

在我们的例子里，如果 通过b.originPeers 发现 原始的peer是leaner， b.targetPeers对应peer的期望状态是voter角色。那么我们会保存一条记录，表示那个store的peer角色由learner升级到voter角色\ // Initialize intermediate states.\ // TODO: simplify the code\ func (b *Builder) prepareBuild() (string, error) {\ b.toPromote = newPeersMap()\ // Diff originPeers and targetPeers to initialize toAdd, toRemove, toPromote, toDemote.\ for _, o := range b.originPeers {\ n := b.targetPeers[o.GetStoreId()]\ if core.IsLearner(o) {\ if !core.IsLearner(n) {\ // learner -> voter\ b.toPromote.Set(n)\ }\ }\ }

``` b.peerAddStep = make(map[uint64]int)

return b.brief(), nil

} ```

每个调度操作，分很多steps,下面这个函数用于生成调度操作的steps

// Some special cases, and stores that do not support using joint consensus.\ func (b *Builder) buildStepsWithoutJointConsensus(kind OpKind) (OpKind, error) {\ for len(b.toPromote) > 0 {\ plan := b.peerPlan()

if plan.promote != nil { b.execPromoteLearner(plan.promote) } } return kind, nil }

调用b.peerPlan生成对应step paln, 最终基于stepplan生成操作步骤，

func (b *Builder) peerPlan() stepPlan {\ if p := b.planPromotePeer(); !p.IsEmpty() {\ return p\ }\ }

func (b *Builder) planPromotePeer() stepPlan {\ for _, i := range b.toPromote.IDs() {\ peer := b.toPromote[i]\ return stepPlan{promote: peer}\ }\ }

type stepPlan struct {\ promote *metapb.Peer\ }\ 基于stepplan生成具体的steps。\ PromoteLearner表示promote调度步骤,stote表示leaner对应的stroe,peerID表示对应peer\ if plan.promote != nil {\ b.execPromoteLearner(plan.promote)\ }

func (b *Builder) execPromoteLearner(peer *metapb.Peer) {\ b.steps = append(b.steps, PromoteLearner{ToStore: peer.GetStoreId(), PeerID: peer.GetId()})\ }

// PromoteLearner is an OpStep that promotes a region learner peer to normal voter.\ type PromoteLearner struct {\ ToStore, PeerID uint64\ }

基于以上步骤，调用build 函数生成opeartor，代码如下:\ // CreatePromoteLearnerOperator creates an operator that promotes a learner.\ func CreatePromoteLearnerOperator(desc string, cluster opt.Cluster, region *core.RegionInfo, peer *metapb.Peer) (*Operator, error) {\ return NewBuilder(desc, cluster, region).\ PromoteLearner(peer.GetStoreId()).\ Build(0)\ }

// Build creates the Operator.\ func (b *Builder) Build(kind OpKind) (*Operator, error) {\ return NewOperator(b.desc, brief, b.regionID, b.regionEpoch, kind, b.steps…), nil\ }

// NewOperator creates a new operator.\ func NewOperator(desc, brief string, regionID uint64, regionEpoch *metapb.RegionEpoch, kind OpKind, steps …OpStep) *Operator {\ return \&Operator{\ desc: desc,\ brief: brief,\ regionID: regionID,\ regionEpoch: regionEpoch,\ kind: kind,\ steps: steps,\ stepsTime: make([]int64, len(steps)),\ status: NewOpStatusTracker(),\ level: level,\ AdditionalInfos: make(map[string]string),\ }\ }

限制调度速度的模块opController\ 模块定义\ opController := schedule.NewOperatorController(ctx, cluster, hbStreams)

// NewOperatorController creates a OperatorController.\ func NewOperatorController(ctx context.Context, cluster opt.Cluster, hbStreams *hbstream.HeartbeatStreams) *OperatorController {\ return \&OperatorController{\ ctx: ctx,\ cluster: cluster,\ operators: make(map[uint64]*operator.Operator),\ hbStreams: hbStreams,\ histories: list.New(),\ counts: make(map[operator.OpKind]uint64),\ opRecords: NewOperatorRecords(ctx),\ storesLimit: make(map[uint64]map[storelimit.Type]*storelimit.StoreLimit),\ wop: NewRandBuckets(),\ wopStatus: NewWaitingOperatorStatus(),\ opNotifierQueue: make(operatorQueue, 0),\ }\ }

// OperatorController is used to limit the speed of scheduling.\ type OperatorController struct {\ sync.RWMutex\ ctx context.Context\ cluster opt.Cluster\ operators map[uint64]*operator.Operator\ hbStreams *hbstream.HeartbeatStreams\ histories *list.List\ counts map[operator.OpKind]uint64\ opRecords *OperatorRecords\ storesLimit map[uint64]map[storelimit.Type]*storelimit.StoreLimit\ wop WaitingOperator\ wopStatus *WaitingOperatorStatus\ opNotifierQueue operatorQueue\ }

这个模块的核心功能，就是控制把opeartor 发送到TIKV的速度，所以我们找出opcontrol发送opeartor到tikv的函数：\ // Dispatch is used to dispatch the operator of a region.\ func (oc *OperatorController) Dispatch(region *core.RegionInfo, source string) {\ // Check existed operator.\ if op := oc.GetOperator(region.GetID()); op != nil {\ // Update operator status:\ // The operator status should be STARTED.\ // Check will call CheckSuccess and CheckTimeout.\ step := op.Check(region)

switch op.Status() { case operator.STARTED: oc.SendScheduleCommand(region, step, source) } } }

oc.SendScheduleCommand(region, step, source),针对指定region对应opeartor的step到tikv节点,前面分析我们的step对应PromoteLearner,也就是上升leaner到voter。生成heatbeatResponse对象，这个对象会通过region的心跳请求回复给这个tikv.从而推动tikv升级region peer 到voter\ // SendScheduleCommand sends a command to the region.\ func (oc *OperatorController) SendScheduleCommand(region *core.RegionInfo, step operator.OpStep, source string) {\ var cmd *pdpb.RegionHeartbeatResponse\ switch st := step.(type) {\ case operator.PromoteLearner:\ cmd = \&pdpb.RegionHeartbeatResponse{\ ChangePeer: \&pdpb.ChangePeer{\ // reuse AddNode type\ ChangeType: eraftpb.ConfChangeType_AddNode,\ Peer: \&metapb.Peer{\ Id: st.PeerID,\ StoreId: st.ToStore,\ Role: metapb.PeerRole_Voter,\ },\ },\ }

oc.hbStreams.SendMsg(region, cmd) }

第三个部分，分析tikv和pd心跳流处理

用于处理心跳信息模块hbStreams\ 心跳回复信息发送,这个函数会把消息发送到hbstreams的chanel 。用于消息的异常发送\ oc.hbStreams.SendMsg(region, cmd)

// SendMsg sends a message to related store.\ func (s *HeartbeatStreams) SendMsg(region *core.RegionInfo, msg *pdpb.RegionHeartbeatResponse) {

select { case s.msgCh <- msg: case <-s.hbStreamCtx.Done(): } }

hbstream模块，会启动一个后台服务线程，用于定期从channel 模块里取出消息，处理发送到相应的tikv\ func (s *HeartbeatStreams) run() {\ for {\ select {

``` case msg := <-s.msgCh: storeID := msg.GetTargetPeer().GetStoreId() storeLabel := strconv.FormatUint(storeID, 10) store := s.storeInformer.GetStore(storeID)

        storeAddress := store.GetAddress()
        if stream, ok := s.streams[storeID]; ok {
            if err := stream.Send(msg); 
        }

```

}

HeartbeatStreams模块通过一个map结构管理和其它TIKV结点的流,这个数据结构的数据更新，通过每次心跳grpc 处理更新\ / RegionHeartbeat implements gRPC PDServer.

func (s *Server) RegionHeartbeat(stream pdpb.PD_RegionHeartbeatServer) error {\ server := \&heartbeatServer{stream: stream}\ s.hbStreams.BindStream(storeID, server)\ }

type heartbeatServer struct {\ stream pdpb.PD_RegionHeartbeatServer\ closed int32\ }

heartbeatServer是pdpb.PD_RegionHeartbeatServer包装器\ func (s *heartbeatServer) Send(m *pdpb.RegionHeartbeatResponse) error {

done := make(chan error, 1) go func() { done <- s.stream.Send(m) }() select { case err := <-done: return errors.WithStack(err) case <-time.After(regionHeartbeatSendTimeout): atomic.StoreInt32(&s.closed, 1) return errors.WithStack(errSendRegionHeartbeatTimeout) } }