Go语言kube-scheduler之pod调度怎么实现
来源:亿速云
时间:2023-05-06 07:24:40 289浏览 收藏
一分耕耘,一分收获!既然都打开这篇《Go语言kube-scheduler之pod调度怎么实现》,就坚持看下去,学下去吧!本文主要会给大家讲到go语言、kube-scheduler、pod等等知识点,如果大家对本文有好的建议或者看到有不足之处,非常欢迎大家积极提出!在后续文章我会继续更新Golang相关的内容,希望对大家都有所帮助!
这篇文章主要介绍“Go语言kube-scheduler之pod调度怎么实现”的相关知识,小编通过实际案例向大家展示操作过程,操作方法简单快捷,实用性强,希望这篇“Go语言kube-scheduler之pod调度怎么实现”文章能帮助大家解决问题。
感知 Pod
要能够获取到 Pod 的前提是:kube-scheduler 能感知到有 Pod 需要被调度,得知有 Pod 需要被调度后还需要有地方存放被调度的 Pod 的信息。为了感知有 Pod 需要被调度,kube-scheduler 启动时通过 Informer watch Pod 的变化,它把待调度的 Pod 分了两种情况,代码如下
// pkg/scheduler/eventhandlers.go func addAllEventHandlers(...) { //已经调度过的 Pod 则加到本地缓存,并判断是加入到调度队列还是加入到backoff队列 informerFactory.Core().V1().Pods().Informer().AddEventHandler( cache.FilteringResourceEventHandler{ FilterFunc: func(obj interface{}) bool { switch t := obj.(type) { case *v1.Pod: return assignedPod(t) case cache.DeletedFinalStateUnknown: if _, ok := t.Obj.(*v1.Pod); ok { // The carried object may be stale, so we don't use it to check if // it's assigned or not. Attempting to cleanup anyways. return true } utilruntime.HandleError(fmt.Errorf("unable to convert object %T to *v1.Pod in %T", obj, sched)) return false default: utilruntime.HandleError(fmt.Errorf("unable to handle object in %T: %T", sched, obj)) return false } }, Handler: cache.ResourceEventHandlerFuncs{ AddFunc: sched.addPodToCache, UpdateFunc: sched.updatePodInCache, DeleteFunc: sched.deletePodFromCache, }, }, ) // 没有调度过的Pod,放到调度队列 informerFactory.Core().V1().Pods().Informer().AddEventHandler( cache.FilteringResourceEventHandler{ FilterFunc: func(obj interface{}) bool { switch t := obj.(type) { case *v1.Pod: return !assignedPod(t) && responsibleForPod(t, sched.Profiles) case cache.DeletedFinalStateUnknown: if pod, ok := t.Obj.(*v1.Pod); ok { // The carried object may be stale, so we don't use it to check if // it's assigned or not. return responsibleForPod(pod, sched.Profiles) } utilruntime.HandleError(fmt.Errorf("unable to convert object %T to *v1.Pod in %T", obj, sched)) return false default: utilruntime.HandleError(fmt.Errorf("unable to handle object in %T: %T", sched, obj)) return false } }, Handler: cache.ResourceEventHandlerFuncs{ AddFunc: sched.addPodToSchedulingQueue, UpdateFunc: sched.updatePodInSchedulingQueue, DeleteFunc: sched.deletePodFromSchedulingQueue, }, }, ) ...... }
已经调度过的 Pod 区分是不是调度过的 Pod 是通过:len(pod.Spec.NodeName) != 0 来判断的,因为调度过的 Pod 这个字段总是会被赋予被选中的 Node 名字。但是,既然是调度过的 Pod 下面的代码中为什么还要区分:sched.addPodToCache 和 sched.updatePodInCache 呢?原因在于我们可以在创建 Pod 的时候人为给它分配一个 Node(即给 pod.Spec.NodeName 赋值),这样 kube-scheduler 在监听到该 Pod 后,判断这个 Pod 的该字段不为空就会认为这个 Pod 已经调度过了,但是这个字段不为空并不是 kube-scheduler 调度的结果,而是人为赋值的,那么 kube-scheduler 的 cache(可以参考上一篇 cache 相关的内容)中没有这个 Pod 的信息,所以就需要将 Pod 信息加入到 cache 中。至于在监听到 Pod 后 sched.addPodToCache 和 sched.updatePodInCache 哪个会被调用,这是 Informer 决定的,它会根据监听到变化的 Pod 和 Informer 的本地缓存做对比,要是缓存中没有这个 Pod,那么就调用 add 函数,否则就调用 update 函数。
加入或更新缓存后,还需要做一件事:去 unschedulablePods(调度失败的Pod) 中获取 Pod,这些 Pod 的亲和性和刚刚加入的这个 Pod 匹配,然后根据下面的规则判断是把 Pod 放入 backoffQ 还是放入 activeQ
根据这个 Pod 尝试被调度的次数计算这个 Pod 下次调度应该等待的时间,计算规则为指数级增长,即按照1s,2s,4s,8s这样的时间进行等待,但是这个等待时间也不会无限增加,会受到 podMaxBackoffDuration(默认10s) 的限制,这个参数表示是一个 Pod 处于 backoff 的最大时间,如果等待的时间如果超过了 podMaxBackoffDuration,那么就只等待 podMaxBackoffDuration 就会再次被调度;
当前时间 - 上次调度的时间 > 根据(1)获取到的应该等待的时间,那么就把Pod放到activeQ里面,将会被调度,否则Pod被放入 backoff 队列里等待。
从上面我们可以看到,一个 Pod 的变化会触发此前调度失败的 Pod 重新判断是否可以被调度
没有调度过的 Pod
len(pod.Spec.NodeName) = 0,那么这个 Pod 没有被调度过或者是此前调度过但是调度失败的(用户修改了 Pod 的配置导致 Pod 发生变化,又被 kube-scheduler 感知到了),如果是没有调度过的 Pod 那么直接加入到 activeQ,如果是调度失败的 Pod 则根据上述规则判断是加入 backoffQ 还是 activeQ。加入到 activeQ 会马上被取走,然后开始调度。
那么那些因为调度失败而被放入 unscheduleable 的 Pod 还有其他机会(上面说的有新 Pod 创建是一个机会)重新被调度么?答案是有的,否则他们就“被饿死了”,有两种途径:1. 定期强制将 unscheduleable 的 Pod 放入 backoffQ 或 activeQ,定期将 backoffQ 等待超时的 Pod 放入 ac activeQ;2. 集群内其他相关资源变化时,判断 unscheduleable 中的 Pod 是不是要放入 backoffQ 或 activeQ,其实这跟有 Pod 发生变化的情况是一样的。
第一种情况
在 kube-scheduler启动的时候中会起两个协程,他们会定期把 backoffQ 和 unscheduleable 里面的 Pod拿到activeQ里面去
func (p *PriorityQueue) Run() { go wait.Until(p.flushBackoffQCompleted, 1.0*time.Second, p.stop) go wait.Until(p.flushUnschedulablePodsLeftover, 30*time.Second, p.stop) }
flushUnschedulablePodsLeftover
func (p *PriorityQueue) flushUnschedulablePodsLeftover() { p.lock.Lock() defer p.lock.Unlock() var podsToMove []*framework.QueuedPodInfo currentTime := p.clock.Now() for _, pInfo := range p.unschedulablePods.podInfoMap { lastScheduleTime := pInfo.Timestamp if currentTime.Sub(lastScheduleTime) > p.podMaxInUnschedulablePodsDuration { podsToMove = append(podsToMove, pInfo) } } if len(podsToMove) > 0 { p.movePodsToActiveOrBackoffQueue(podsToMove, UnschedulableTimeout) } }
func (p *PriorityQueue) movePodsToActiveOrBackoffQueue(podInfoList []*framework.QueuedPodInfo, event framework.ClusterEvent) { activated := false for _, pInfo := range podInfoList { // If the event doesn't help making the Pod schedulable, continue. // Note: we don't run the check if pInfo.UnschedulablePlugins is nil, which denotes // either there is some abnormal error, or scheduling the pod failed by plugins other than PreFilter, Filter and Permit. // In that case, it's desired to move it anyways. if len(pInfo.UnschedulablePlugins) != 0 && !p.podMatchesEvent(pInfo, event) { continue } pod := pInfo.Pod if p.isPodBackingoff(pInfo) { if err := p.podBackoffQ.Add(pInfo); err != nil { klog.ErrorS(err, "Error adding pod to the backoff queue", "pod", klog.KObj(pod)) } else { metrics.SchedulerQueueIncomingPods.WithLabelValues("backoff", event.Label).Inc() p.unschedulablePods.delete(pod) } } else { if err := p.activeQ.Add(pInfo); err != nil { klog.ErrorS(err, "Error adding pod to the scheduling queue", "pod", klog.KObj(pod)) } else { metrics.SchedulerQueueIncomingPods.WithLabelValues("active", event.Label).Inc() p.unschedulablePods.delete(pod) } } } p.moveRequestCycle = p.schedulingCycle if activated { p.cond.Broadcast() } }x
将在 unscheduleable 里面停留时长超过 podMaxInUnschedulablePodsDuration(默认是5min)的pod放入到 ActiveQ 或 BackoffQueue,具体是放到哪个队列里面,还是根据我们上文说的那个实际计算规则来。这么做的原因就是给那些“问题少年”一次重新做人的机会,也不能一犯错误(调度失败)就彻底打入死牢了。
flushBackoffQCompleted
去 backoffQ 获取等待结束的 Pod,放入 activeQ
func (p *PriorityQueue) flushBackoffQCompleted() { p.lock.Lock() defer p.lock.Unlock() activated := false for { rawPodInfo := p.podBackoffQ.Peek() if rawPodInfo == nil { break } pod := rawPodInfo.(*framework.QueuedPodInfo).Pod boTime := p.getBackoffTime(rawPodInfo.(*framework.QueuedPodInfo)) if boTime.After(p.clock.Now()) { break } _, err := p.podBackoffQ.Pop() if err != nil { klog.ErrorS(err, "Unable to pop pod from backoff queue despite backoff completion", "pod", klog.KObj(pod)) break } p.activeQ.Add(rawPodInfo) metrics.SchedulerQueueIncomingPods.WithLabelValues("active", BackoffComplete).Inc() activated = true } if activated { p.cond.Broadcast() } }
第二种情况
集群内资源发生变化
有新节点加入集群
节点配置或状态发生变化
已经存在的 Pod 发生变化
集群内有Pod被删除
informerFactory.Core().V1().Nodes().Informer().AddEventHandler( cache.ResourceEventHandlerFuncs{ AddFunc: sched.addNodeToCache, UpdateFunc: sched.updateNodeInCache, DeleteFunc: sched.deleteNodeFromCache, }, )
新加入节点
func (sched *Scheduler) addNodeToCache(obj interface{}) { node, ok := obj.(*v1.Node) if !ok { klog.ErrorS(nil, "Cannot convert to *v1.Node", "obj", obj) return } nodeInfo := sched.Cache.AddNode(node) klog.V(3).InfoS("Add event for node", "node", klog.KObj(node)) sched.SchedulingQueue.MoveAllToActiveOrBackoffQueue(queue.NodeAdd, preCheckForNode(nodeInfo)) }
func preCheckForNode(nodeInfo *framework.NodeInfo) queue.PreEnqueueCheck { // Note: the following checks doesn't take preemption into considerations, in very rare // cases (e.g., node resizing), "pod" may still fail a check but preemption helps. We deliberately // chose to ignore those cases as unschedulable pods will be re-queued eventually. return func(pod *v1.Pod) bool { admissionResults := AdmissionCheck(pod, nodeInfo, false) if len(admissionResults) != 0 { return false } _, isUntolerated := corev1helpers.FindMatchingUntoleratedTaint(nodeInfo.Node().Spec.Taints, pod.Spec.Tolerations, func(t *v1.Taint) bool { return t.Effect == v1.TaintEffectNoSchedule }) return !isUntolerated } }
可以看到,当有节点加入集群的时候,会把 unscheduleable 里面的Pod 依次拿出来做下面的判断:
Pod 对 节点的亲和性
Pod 中 Nodename不为空 那么判断新加入节点的Name判断pod Nodename是否相等
判断 Pod 中容器对端口的要求是否和新加入节点已经被使用的端口冲突
Pod 是否容忍了Node的Pod
只有上述4个条件都满足,那么新加入节点这个事件才会触发这个未被调度的Pod加入到 backoffQ 或者 activeQ,至于是加入哪个queue,上面已经分析过了
节点更新
func (sched *Scheduler) updateNodeInCache(oldObj, newObj interface{}) { oldNode, ok := oldObj.(*v1.Node) if !ok { klog.ErrorS(nil, "Cannot convert oldObj to *v1.Node", "oldObj", oldObj) return } newNode, ok := newObj.(*v1.Node) if !ok { klog.ErrorS(nil, "Cannot convert newObj to *v1.Node", "newObj", newObj) return } nodeInfo := sched.Cache.UpdateNode(oldNode, newNode) // Only requeue unschedulable pods if the node became more schedulable. if event := nodeSchedulingPropertiesChange(newNode, oldNode); event != nil { sched.SchedulingQueue.MoveAllToActiveOrBackoffQueue(*event, preCheckForNode(nodeInfo)) } }
func nodeSchedulingPropertiesChange(newNode *v1.Node, oldNode *v1.Node) *framework.ClusterEvent { if nodeSpecUnschedulableChanged(newNode, oldNode) { return &queue.NodeSpecUnschedulableChange } if nodeAllocatableChanged(newNode, oldNode) { return &queue.NodeAllocatableChange } if nodeLabelsChanged(newNode, oldNode) { return &queue.NodeLabelChange } if nodeTaintsChanged(newNode, oldNode) { return &queue.NodeTaintChange } if nodeConditionsChanged(newNode, oldNode) { return &queue.NodeConditionChange } return nil }
首先是判断节点是何种配置发生了变化,有如下情况
节点可调度情况发生变化
节点可分配资源发生变化
节点标签发生变化
节点污点发生变化
节点状态发生变化
如果某个 Pod 调度失败的原因可以匹配到上面其中一个原因,那么节点更新这个事件才会触发这个未被调度的Pod加入到 backoffQ 或者 activeQ
informerFactory.Core().V1().Pods().Informer().AddEventHandler( cache.FilteringResourceEventHandler{ FilterFunc: func(obj interface{}) bool { switch t := obj.(type) { case *v1.Pod: return assignedPod(t) case cache.DeletedFinalStateUnknown: if _, ok := t.Obj.(*v1.Pod); ok { // The carried object may be stale, so we don't use it to check if // it's assigned or not. Attempting to cleanup anyways. return true } utilruntime.HandleError(fmt.Errorf("unable to convert object %T to *v1.Pod in %T", obj, sched)) return false default: utilruntime.HandleError(fmt.Errorf("unable to handle object in %T: %T", sched, obj)) return false } }, Handler: cache.ResourceEventHandlerFuncs{ AddFunc: sched.addPodToCache, UpdateFunc: sched.updatePodInCache, DeleteFunc: sched.deletePodFromCache, }, }, )
已经存在的 Pod 发生变化
func (sched *Scheduler) addPodToCache(obj interface{}) { pod, ok := obj.(*v1.Pod) if !ok { klog.ErrorS(nil, "Cannot convert to *v1.Pod", "obj", obj) return } klog.V(3).InfoS("Add event for scheduled pod", "pod", klog.KObj(pod)) if err := sched.Cache.AddPod(pod); err != nil { klog.ErrorS(err, "Scheduler cache AddPod failed", "pod", klog.KObj(pod)) } sched.SchedulingQueue.AssignedPodAdded(pod) }
func (p *PriorityQueue) AssignedPodAdded(pod *v1.Pod) { p.lock.Lock() p.movePodsToActiveOrBackoffQueue(p.getUnschedulablePodsWithMatchingAffinityTerm(pod), AssignedPodAdd) p.lock.Unlock() }
func (p *PriorityQueue) getUnschedulablePodsWithMatchingAffinityTerm(pod *v1.Pod) []*framework.QueuedPodInfo { var nsLabels labels.Set nsLabels = interpodaffinity.GetNamespaceLabelsSnapshot(pod.Namespace, p.nsLister) var podsToMove []*framework.QueuedPodInfo for _, pInfo := range p.unschedulablePods.podInfoMap { for _, term := range pInfo.RequiredAffinityTerms { if term.Matches(pod, nsLabels) { podsToMove = append(podsToMove, pInfo) break } } } return podsToMove }
可以看到,已经存在的Pod发生变化后,会把这个Pod亲和性配置依次和 unscheduleable 里面的Pod匹配,如果能够匹配上,那么节点更新这个事件才会触发这个未被调度的Pod加入到 backoffQ 或者 activeQ。
集群内有Pod删除
func (sched *Scheduler) deletePodFromCache(obj interface{}) { var pod *v1.Pod switch t := obj.(type) { case *v1.Pod: pod = t case cache.DeletedFinalStateUnknown: var ok bool pod, ok = t.Obj.(*v1.Pod) if !ok { klog.ErrorS(nil, "Cannot convert to *v1.Pod", "obj", t.Obj) return } default: klog.ErrorS(nil, "Cannot convert to *v1.Pod", "obj", t) return } klog.V(3).InfoS("Delete event for scheduled pod", "pod", klog.KObj(pod)) if err := sched.Cache.RemovePod(pod); err != nil { klog.ErrorS(err, "Scheduler cache RemovePod failed", "pod", klog.KObj(pod)) } sched.SchedulingQueue.MoveAllToActiveOrBackoffQueue(queue.AssignedPodDelete, nil) }
可以看到,Pod删除时间不像其他时间需要做额外的判断,这个preCheck函数是空的,所以所有 unscheduleable 里面的Pod都会被放到 activeQ 或 backoffQ 中。
从上面的情况,我们可以看到,集群内有事件发生变化,是可以加速调度失败的Pod被重新调度的进程的。常规的是,调度失败的 Pod 需要等5min 然后才会被重新加入 backoffQ 或 activeQ。backoffQ里面的Pod也需要等一段时间才会重新调度。这也就是为什么,当你修改节点配置的时候,能看到Pod马上重新被调度的原因
上面就是一个Pod调度失败后,重新触发调度的情况了。
取出 Pod
Scheduler 中有个成员 NextPod 会从 activeQ 队列中尝试获取一个待调度的 Pod,该函数在 SchedulePod 中被调用,如下:
// 启动 Scheduler func (sched *Scheduler) Run(ctx context.Context) { sched.SchedulingQueue.Run() go wait.UntilWithContext(ctx, sched.scheduleOne, 0)NextPod 它被赋予如下函数:
// pkg/scheduler/internal/queue/scheduling_queue.go func MakeNextPodFunc(queue SchedulingQueue) func() *framework.QueuedPodInfo { return func() *framework.QueuedPodInfo { podInfo, err := queue.Pop() if err == nil { klog.V(4).InfoS("About to try and schedule pod", "pod", klog.KObj(podInfo.Pod)) for plugin := range podInfo.UnschedulablePlugins { metrics.UnschedulableReason(plugin, podInfo.Pod.Spec.SchedulerName).Dec() } return podInfo } klog.ErrorS(err, "Error while retrieving next pod from scheduling queue") return nil } }Pop 会一直阻塞,直到 activeQ 长度大于0,然后取出一个 Pod 返回
// pkg/scheduler/internal/queue/scheduling_queue.go func (p *PriorityQueue) Pop() (*framework.QueuedPodInfo, error) { p.lock.Lock() defer p.lock.Unlock() for p.activeQ.Len() == 0 { // When the queue is empty, invocation of Pop() is blocked until new item is enqueued. // When Close() is called, the p.closed is set and the condition is broadcast, // which causes this loop to continue and return from the Pop(). 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.QueuedPodInfo) pInfo.Attempts++ p.schedulingCycle++ return pInfo, nil }调度 Pod
func (sched *Scheduler) scheduleOne(ctx context.Context) { // 取出 Pod podInfo := sched.NextPod() ... // 根据 Pod 的调度名字,获取之前初始化好的调度框架(framework) fwk, err := sched.frameworkForPod(pod) ... // 开始执行插件,包括 filter, socre 两个扩展点内的所有插件,获取一个最合适 Pod 的节点 scheduleResult, err := sched.SchedulePod(schedulingCycleCtx, fwk, state, pod) // 如果获取节点失败,则开始运行 postFilter 开始抢占一个 Pod if err != nil { result, status := fwk.RunPostFilterPlugins(ctx, state, pod, fitError.Diagnosis.NodeToStatusMap) } .... // 将 Pod 放入 assumedPod 存储,即假设 Pod 已经调度成功 err = sched.assume(assumedPod, scheduleResult.SuggestedHost) // 运行 Reserve 插件 fwk.RunReservePluginsReserve(schedulingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost) ... // 运行 Permit 插件 fwk.RunPermitPlugins(schedulingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost) ... // 启动一个协程,开始绑定,主流程到了这里就结束了,然后开始新的一轮调度; go func() { // 执行 preBind 插件 fwk.RunPreBindPlugins(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost) ... // 执行绑定插件,会调用 kube-apiserver 写入etcd 调度结果,就是给 Pod 赋予 Nodename err := sched.bind(bindingCycleCtx, fwk, assumedPod, scheduleResult.SuggestedHost, state) ... // 执行 postBind fwk.RunPostBindPlugins(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost) }
执行 filter 类型扩展点(包括preFilter,filter,postFilter)插件,选出所有符合 Pod 的 Node,如果无法找到符合的 Node, 则把 Pod 加入 unscheduleable 中,此次调度结束;
执行 score 扩展点插件,给所有 Node 打分;
拿出得分最高的 Node;
assume Pod。这一步就是乐观假设 Pod 已经调度成功,更新缓存中 Node 和 PodStats 信息,到了这里scheduling cycle就已经结束了,然后会开启新的一轮调度。至于真正的绑定,则会新起一个协程。
执行 reserve 插件;
启动协程绑定 Pod 到 Node上。实际上就是修改 Pod.spec.nodeName: 选定的node名字,然后调用 kube-apiserver 接口写入 etcd。如果绑定失败了,那么移除缓存中此前加入的信息,然后把 Pod 放入activeQ 中,后续重新调度。执行 postBinding,该步没有实现的插件没所以没有做任何事。
关于“Go语言kube-scheduler之pod调度怎么实现”的内容就介绍到这里了,感谢大家的阅读。如果想了解更多行业相关的知识,可以关注golang学习网行业资讯频道,小编每天都会为大家更新不同的知识点。
今天带大家了解了go语言、kube-scheduler、pod的相关知识,希望对你有所帮助;关于Golang的技术知识我们会一点点深入介绍,欢迎大家关注golang学习网公众号,一起学习编程~
-
185 收藏
-
460 收藏
-
430 收藏
-
450 收藏
-
320 收藏
-
233 收藏
-
322 收藏
-
181 收藏
-
316 收藏
-
244 收藏
-
300 收藏
-
- 前端进阶之JavaScript设计模式
- 设计模式是开发人员在软件开发过程中面临一般问题时的解决方案,代表了最佳的实践。本课程的主打内容包括JS常见设计模式以及具体应用场景,打造一站式知识长龙服务,适合有JS基础的同学学习。
- 立即学习 542次学习
-
- GO语言核心编程课程
- 本课程采用真实案例,全面具体可落地,从理论到实践,一步一步将GO核心编程技术、编程思想、底层实现融会贯通,使学习者贴近时代脉搏,做IT互联网时代的弄潮儿。
- 立即学习 507次学习
-
- 简单聊聊mysql8与网络通信
- 如有问题加微信:Le-studyg;在课程中,我们将首先介绍MySQL8的新特性,包括性能优化、安全增强、新数据类型等,帮助学生快速熟悉MySQL8的最新功能。接着,我们将深入解析MySQL的网络通信机制,包括协议、连接管理、数据传输等,让
- 立即学习 497次学习
-
- JavaScript正则表达式基础与实战
- 在任何一门编程语言中,正则表达式,都是一项重要的知识,它提供了高效的字符串匹配与捕获机制,可以极大的简化程序设计。
- 立即学习 487次学习
-
- 从零制作响应式网站—Grid布局
- 本系列教程将展示从零制作一个假想的网络科技公司官网,分为导航,轮播,关于我们,成功案例,服务流程,团队介绍,数据部分,公司动态,底部信息等内容区块。网站整体采用CSSGrid布局,支持响应式,有流畅过渡和展现动画。
- 立即学习 484次学习