• 中文
    • English
  • 注册
  • 查看作者
  • 图解kubernetes调度器SchedulingQueue核心源码实现

    SchedulingQueue是kubernetes scheduler中负责进行等待调度pod存储的对,Scheduler通过SchedulingQueue来获取当前系统中等待调度的Pod,本文主要讨论SchedulingQueue的设计与实现的各种实现, 了解探究其内部实现与底层源码,本系列代码基于kubernets1.1.6分析而来,图解主要位于第二部分

    SchedulingQueue设计队列与优先级队列与场景

    其实在大多数的调度场景中,大多都是采用优先级队列来实现,优先满足优先级比较高的任务或者需求,从而减少后续高优先级对低优先级的抢占,scheduler中也是如此

    优先级的选择k8s中调度的单元是Pod,scheduler中根据pod的优先级的高低来进行优先级队列的构建, 这个其实是在kubernets的adminission准入插件中,会为用户创建的pod根据用户的设置,进行优先级字段的计算

    三级队列活动队列活动队列存储当前系统中所有正在等待调度的队列

    不可调度队列当pod的资源在当前集群中不能被满足时,则会被加入到一个不可调度队列中,然后等待稍后再进行尝试

    backoff队列backoff机制是并发编程中常见的一种机制,即如果任务反复执行依旧失败,则会按次增长等待调度时间,降低重试效率,从而避免反复失败浪费调度资源

    针对调度失败的pod会优先存储在backoff队列中,等待后续重试

    阻塞与抢占阻塞设计当队列中不存在等待调度的pod的时候,会阻塞scheduler等待有需要调度的pod的时候再唤醒调度器,获取pod进行调度

    抢占相关nominatedPods存储pod被提议运行的node,主要用于抢占调度流程中使用,本节先不分析

    源码分析数据结构kubernetes中默认的schedulingQueue实现是PriorityQueue,本章就以该数据结构来分析

    type PriorityQueue struct {stop <-chan struct{}clock util.Clock// 存储backoff的pod计时器podBackoff *PodBackoffMap

    }PriorityQueue作为实现SchedulingQueue的实现,其核心数据结构主要包含三个队列:activeQ、podBackoffQ、unscheduleQ内部通过cond来实现Pop操作的阻塞与通知,接下来先分析核心的调度流程,最后再分析util.Heap里面的具体实现

    activeQ存储所有等待调度的Pod的队列,默认是基于堆来实现,其中元素的优先级则通过对比pod的创建时间和pod的优先级来进行排序

    优先级比较函数

    // activeQComp is the function used by the activeQ heap algorithm to sort pods.// It sorts pods based on their priority. When priorities are equal, it uses// PodInfo.timestamp.func activeQComp(podInfo1, podInfo2 interface{}) bool {pInfo1 := podInfo1.(*framework.PodInfo)pInfo2 := podInfo2.(*framework.PodInfo)prio1 := util.GetPodPriority(pInfo1.Pod)prio2 := util.GetPodPriority(pInfo2.Pod)// 首先根据优先级的高低进行比较,然后根据pod的创建时间,越高优先级的Pod越被优先调度// 越早创建的pod越优先return (prio1 > prio2) || (prio1 == prio2 && pInfo1.Timestamp.Before(pInfo2.Timestamp))}podbackOffQpodBackOffQ主要存储那些在多个schedulingCycle中依旧调度失败的情况下,则会通过之前说的backOff机制,延迟等待调度的时间

    podBackOff上面提到podBackOffQ队列中并没有存储pod的backOff的具体信息,比如backoff的计数器,最后一次更新的时间等,podBackOff则类似一个记分板,记录这些信息,供podBackOffQ使用

    // PodBackoffMap is a structure that stores backoff related information for podstype PodBackoffMap struct {// lock for performing actions on this PodBackoffMaplock sync.RWMutex// initial backoff durationinitialDuration time.Duration // 当前值是1秒// maximal backoff durationmaxDuration time.Duration // 当前值是1分钟// map for pod -> number of attempts for this podpodAttempts map[ktypes.NamespacedName]int// map for pod -> lastUpdateTime pod of this podpodLastUpdateTime map[ktypes.NamespacedName]time.Time}unschedulableQ存储已经尝试调度但是当前集群资源不满足的pod的队列

    moveRequestCycle当因为集群资源发生变化会尝试进行unschedulableQ中的pod转移到activeQ,moveRequestCycle就是存储资源变更时的schedulingCycle

    func (p *PriorityQueue) MoveAllToActiveQueue() {// 省略其他代码p.moveRequestCycle = p.schedulingCycle}schedulingCycleschedulingCycle是一个递增的序列每次从activeQ中pop出一个pod都会递增

    func (p *PriorityQueue) Pop() (*v1.Pod, error) {//省略其他p.schedulingCycle++}并发活动队列并发从活动队列中获取pod

    SchedulingQueue提供了一个Pop接口用于从获取当前集群中等待调度的pod,其内部实现主要通过上面cond与activeQ来实现

    当前队列中没有可调度的pod的时候,则通过cond.Wait来进行阻塞,然后在忘activeQ中添加pod的时候通过cond.Broadcast来实现通知

    func (p *PriorityQueue) Pop() (*v1.Pod, error) {p.lock.Lock()defer p.lock.Unlock()for p.activeQ.Len() == 0 {if p.closed {return nil, fmt.Errorf(queueClosed)}//p.cond.Wait()}obj, err := p.activeQ.Pop()if err != nil {return nil, err}pInfo := obj.(*framework.PodInfo)p.schedulingCycle++return pInfo.Pod, err}加入调度pod到活动队列

    当pod加入活动队列中,除了加入activeQ的优先级队列中,还需要从podBackoffQ和unschedulableQ中移除当前的pod,最后进行广播通知阻塞在Pop操作的scheudler进行最新pod的获取

    func (p *PriorityQueue) Add(pod *v1.Pod) error {p.lock.Lock()defer p.lock.Unlock()pInfo := p.newPodInfo(pod)// 加入activeQif err := p.activeQ.Add(pInfo); err != nil {klog.Errorf(“Error adding pod %v/%v to the scheduling queue: %v”, pod.Namespace, pod.Name, err)return err}// 从unschedulableQ删除if p.unschedulableQ.get(pod) != nil {klog.Errorf(“Error: pod %v/%v is already in the unschedulable queue.”, pod.Namespace, pod.Name)p.unschedulableQ.delete(pod)}// Delete pod from backoffQ if it is backing off// 从podBackoffQ删除if err := p.podBackoffQ.Delete(pInfo); err == nil {klog.Errorf(“Error: pod %v/%v is already in the podBackoff queue.”, pod.Namespace, pod.Name)}// 存储pod和被提名的nodep.nominatedPods.add(pod, “”)p.cond.Broadcast()

    }schedulingCycle与moveRequestCycle未调度的队列的及时重试

    导致调度周期schedulingCyclye变更主要因素如下:1.当集群资源发生变化的时候:比如新添加pv、node等资源,那之前在unschedulableQ中因为资源不满足需求的pod就可以进行放入activeQ中或者podBackoffQ中,及时进行调度2.pod被成功调度: 之前由于亲和性不满足被放入到unschedulableQ中的pod,此时也可以进行尝试,而不必等到超时之后,再加入

    这两种情况下会分别触发MoveAllToActiveQueue和movePodsToActiveQueue变更moveRequestCycle使其等于schedulingCycle

    对重试机制的影响当前一个pod失败的时候,有两种选择一是加入podBackoffQ中,二是加入unschedulableQ中,那么针对一个失败的pod如何选择该进入那个队列中呢

    结合上面的moveRequestCycle变更时机,什么时候moveRequestCycle会大于等于podSchedulingCycle呢?答案就是当前集群中进行过集群资源的变更或者pod被成功分配,那这个时候我们如果重试一个失败的调度则可能会成功,因为集群资源变更了可能有新的资源加入

    失败处理逻辑的注入注入调度失败逻辑处理在创建scheduler Config的时候会通过MakeDefaultErrorFunc注入一个失败处理函数, 在scheduler调度的时候会进行调用

    kubernetes/pkg/scheduler/factory/factory.go: MakeDefaultErrorFunc会将没有调度到任何一个node的pod重新放回到优先级队列中

    失败处理的回调当调度pod的失败的时候, scheduler会同时调用sched.Error就是上面注入的失败处理逻辑,来将调度失败未分配node的pod节点重新加入到队里钟

    kubernetes/pkg/scheduler/scheduler.go

    func (sched *Scheduler) recordSchedulingFailure(pod *v1.Pod, err error, reason string, message string) {// 错误回调sched.Error(pod, err)sched.Recorder.Eventf(pod, nil, v1.EventTypeWarning, “FailedScheduling”, “Scheduling”, message)if err := sched.PodConditionUpdater.Update(pod, &v1.PodCondition{Type: v1.PodScheduled,Status: v1.ConditionFalse,Reason: reason,Message: err.Error(),}); err != nil {klog.Errorf(“Error updating the condition of the pod %s/%s: %v”, pod.Namespace, pod.Name, err)}}PodBackoffMapPodBackoffMap主要用于存储pod的最后一次失败的更新时间与实现次数,从而根据这些数据来进行pod的backoffTime的计算

    数据结构设计type PodBackoffMap struct {// lock for performing actions on this PodBackoffMaplock sync.RWMutex// 初始化 backoff durationinitialDuration time.Duration // 当前值是1秒// 最大 backoff durationmaxDuration time.Duration // 当前值是1分钟// 记录pod重试的次数podAttempts map[ktypes.NamespacedName]int// 记录pod的最后一次的更新时间podLastUpdateTime map[ktypes.NamespacedName]time.Time}backoffTime计算算法初始化的时候回设定initialDuration和maxDuration,在当前版本中分别是1s和10s,也就是backoffQ中的pod最长10s就会重新加入activeQ中(需要等待定时任务进行辅助)

    在每次失败回调的时候,都会进行BackoffPod方法来进行计数更新,在后续获取pod的backoffTime的时候,只需要获取次数然后结合initialDuration进行算法计算,结合pod最后一次的更新时间,就会获取pod的backoffTime的终止时间

    backoffDuration计算其实最终的计算很简单就是2的N次幂

    func (pbm *PodBackoffMap) calculateBackoffDuration(nsPod ktypes.NamespacedName) time.Duration {// initialDuration是1sbackoffDuration := pbm.initialDurationif _, found := pbm.podAttempts[nsPod]; found {// podAttempts里面包含pod的尝试失败的次数for i := 1; i < pbm.podAttempts[nsPod]; i++ {backoffDuration = backoffDuration * 2// 最大10sif backoffDuration > pbm.maxDuration {return pbm.maxDuration}}}return backoffDuration}podBackoffQ

    优先级函数podBackoffQ实际上会根据pod的backoffTime来进行优先级排序,所以podBackoffQ的队列头部,就是最近一个要过期的pod

    func (p *PriorityQueue) podsCompareBackoffCompleted(podInfo1, podInfo2 interface{}) bool {pInfo1 := podInfo1.(*framework.PodInfo)pInfo2 := podInfo2.(*framework.PodInfo)bo1, _ := p.podBackoff.GetBackoffTime(nsNameForPod(pInfo1.Pod))bo2, _ := p.podBackoff.GetBackoffTime(nsNameForPod(pInfo2.Pod))return bo1.Before(bo2)}调度失败加入到podBackoffQ如果调度失败,并且moveRequestCycle=podSchedulingCycle的时候就加入podBackfoffQ中

    func (p *PriorityQueue) AddUnschedulableIfNotPresent(pod *v1.Pod, podSchedulingCycle int64) error {// 省略检查性代码// 更新pod的backoff 信息p.backoffPod(pod)

    }从unschedulableQ迁移在前面介绍的当集群资源发生变更的时候,会触发尝试unschedulabelQ中的pod进行转移,如果发现当前pod还未到达backoffTime,就加入到podBackoffQ中

    podBackoffQ定时转移在创建PriorityQueue的时候,会创建两个定时任务其中一个就是讲backoffQ中的pod到期后的转移,每秒钟尝试一次

    func (p
    PriorityQueue) run() {go wait.Until(p.flushBackoffQCompleted, 1.0
    time.Second, p.stop)go wait.Until(p.flushUnschedulableQLeftover, 30*time.Second, p.stop)}因为是一个堆结果,所以只需要获取堆顶的元素,然后确定是否到期,如果到期后则进行pop处来,加入到activeQ中

    func (p *PriorityQueue) flushBackoffQCompleted() {p.lock.Lock()defer p.lock.Unlock()

    }unschedulableQ

    调度失败调度失败后,如果当前集群资源没有发生变更,就加入到unschedulable,原因上面说过

    func (p *PriorityQueue) AddUnschedulableIfNotPresent(pod *v1.Pod, podSchedulingCycle int64) error {// 省略检查性代码// 更新pod的backoff 信息p.backoffPod(pod)

    }定时转移任务定时任务每30秒执行一次

    func (p
    PriorityQueue) run() {go wait.Until(p.flushUnschedulableQLeftover, 30
    time.Second, p.stop)}逻辑其实就非常简单如果当前时间-pod的最后调度时间大于60s,就重新调度,转移到podBackoffQ或者activeQ中

    func (p *PriorityQueue) flushUnschedulableQLeftover() {p.lock.Lock()defer p.lock.Unlock()

    }调度队列总结数据流设计总结

    3.1.1 三队列与后台定时任务从设计上三队列分别存储:活动队列、bakcoff队列、不可调度队列,其中backoff中会根据任务的失败来逐步递增重试时间(最长10s)、unschedulableQ队列则延迟60s

    通过后台定时任务分别将backoffQ队列、unschedulableQ队列来进行重试,加入到activeQ中,从而加快完成pod的失败重试调度

    cycle与优先调度schedulingCycle、moveRequestCycle两个cycle其实本质上也是为了加快失败任务的重试调度,当集群资源发生变化的时候,进行立即重试,那些失败的优先级比较高、亲和性问题的pod都可能会被优先调度

    锁与cond实现线程安全pop内部通过lock保证线程安全,并通过cond来实现阻塞等待,从而实现阻塞scheduler worker的通知

  • 0
  • 0
  • 0
  • 35
  • 请登录之后再进行评论

    登录
  • 任务
  • 实时动态
  • 发布
  • 单栏布局 侧栏位置: