Kubernetes 抢占调度Preempt机制源码深入剖析-Kubernetes商业环境实战

梦里梦外; 2024-04-17 06:30 60阅读 0赞

专注于大数据及容器云核心技术解密，可提供全栈的大数据+云原生平台咨询方案，请持续关注本套博客。如有任何学术交流，可随时联系。更多内容请关注《数据云技术社区》公众号。

## 1 抢占调度(预选过程+优选失败后) ##

*  scheduler的cmd代码目录结构
 *  scheduler的pkg代码目录结构

### 1.1 调度关系 ###

*  预选调度->优选调度逻辑->节点抢占逻辑
 *  scheduleOne实现1个pod的完整调度工作流，这个过程是顺序执行的，也就是非并发的。也就是说前一个pod的scheduleOne一完成，一个return，下一个pod的scheduleOne立马接着执行！

pkg/scheduler/algorithm/scheduler_interface.go:78
    type ScheduleAlgorithm interface {
        Schedule(*v1.Pod, NodeLister) (selectedMachine string, err error)
        Preempt(*v1.Pod, NodeLister, error) (selectedNode *v1.Node, preemptedPods []*v1.Pod, cleanupNominatedPods []*v1.Pod, err error)
        Predicates() map[string]FitPredicate
        Prioritizers() []PriorityConfig
    }
    
    pkg/scheduler/scheduler.go:276
    // Run begins watching and scheduling. It waits for cache to be synced, then starts a goroutine and returns immediately.
    func (sched *Scheduler) Run() {
        if !sched.config.WaitForCacheSync() {
            return
        }
        go wait.Until(sched.scheduleOne, 0, sched.config.StopEverything)
    }
    
    pkg/scheduler/scheduler.go:513
    func (sched *Scheduler) scheduleOne() {
        pod := sched.config.NextPod()
        suggestedHost, err := sched.schedule(pod)
        if err != nil {
            if fitError, ok := err.(*core.FitError); ok {
                preemptionStartTime := time.Now()
                sched.preempt(pod, fitError)
            }
            return
        }
        assumedPod := pod.DeepCopy()
        allBound, err := sched.assumeVolumes(assumedPod, suggestedHost)
        err = sched.assume(assumedPod, suggestedHost)
        go func() {
            err := sched.bind(assumedPod, &v1.Binding{
                ObjectMeta: metav1.ObjectMeta{Namespace: assumedPod.Namespace, Name: assumedPod.Name, UID: assumedPod.UID},
                Target: v1.ObjectReference{
                    Kind: "Node",
                    Name: suggestedHost,
                },
            })
        }()
    }
    
    pkg/scheduler/scheduler.go:290
    #一般调度过程（预选过程+优选过程）
    func (g *genericScheduler) Schedule(pod *v1.Pod, nodeLister algorithm.NodeLister) (string, error) {
        nodes, err := nodeLister.List()
        trace.Step("Computing predicates")
        filteredNodes, failedPredicateMap, err := g.findNodesThatFit(pod, nodes)
        trace.Step("Prioritizing")
        priorityList, err := PrioritizeNodes(pod, g.cachedNodeInfoMap, metaPrioritiesInterface, g.prioritizers, filteredNodes, g.extenders)
        trace.Step("Selecting host")
        return g.selectHost(priorityList)
    }
    复制代码

### 1.2 抢占调度诞生原因->Pod优先级 ###

*  Pod 有了 priority(优先级) 后才有优先级调度、抢占调度的说法，高优先级的 pod 可以在调度队列中排到前面，优先选择 node；
 *  当高优先级的 pod 找不到合适的 node 时，就会看 node 上低优先级的 pod 驱逐之后是否能够 run 起来，如果可以，那么 node 上的一个或多个低优先级的 pod 会被驱逐，然后高优先级的 pod 得以成功运行1个 node 上。

apiVersion: scheduling.k8s.io/v1
    kind: PriorityClass
    metadata:
      name: high-priority
    value: 1000000
    globalDefault: false
    description: "This priority class should be used for XYZ service pods only."
    
    apiVersion: v1
    kind: Pod
    metadata:
      name: nginx
      labels:
        env: test
    spec:
      containers:
      - name: nginx
        image: nginx
        imagePullPolicy: IfNotPresent
      priorityClassName: high-priority
    复制代码

## 2 抢占调度入口-外层抢占实施 ##

*  判断是否有关闭抢占机制，如果关闭抢占机制则直接返回。
 *  获取调度失败pod的最新对象数据。
 *  执行抢占算法Algorithm.Preempt，返回预调度节点和需要被剔除的pod列表。
 *  将抢占算法返回的node添加到pod的Status.NominatedNodeName中，并删除需要被剔除的pod。
 *  当抢占算法返回的node是nil的时候，清除pod的Status.NominatedNodeName信息。
 *  整个抢占流程的最终结果实际上是更新Pod.Status.NominatedNodeName属性的信息。如果抢占算法返回的节点不为空，则将该node更新到Pod.Status.NominatedNodeName中，否则就将Pod.Status.NominatedNodeName设置为空。

### 2.1 调用preempt ###

suggestedHost, err := sched.schedule(pod)
    if err != nil {
       if fitError, ok := err.(*core.FitError); ok {
          preemptionStartTime := time.Now()
          sched.preempt(pod, fitError)
          metrics.PreemptionAttempts.Inc()
       } else {
          klog.Errorf("error selecting node for pod: %v", err)
          metrics.PodScheduleErrors.Inc()
       }
       return
    }
    复制代码

### 2.2 外层抢占实施相关函数 ###

*  PodPriorityEnabled:特性没有开启就返回 ""
 *  GetUpdatedPod:更新 pod 信息；入参和返回值都是 \*v1.Pod 类型
 *  sched.config.Algorithm.Preempt:核心抢占逻辑过程

func (sched *Scheduler) preempt(preemptor *v1.Pod, scheduleErr error) (string, error) {
        // 特性没有开启就返回 ""
        if !util.PodPriorityEnabled() || sched.config.DisablePreemption {
            return "", nil
        }
        // 更新 pod 信息；入参和返回值都是 *v1.Pod 类型
        preemptor, err := sched.config.PodPreemptor.GetUpdatedPod(preemptor)
    
        // preempt 过程，核心的preempt
        node, victims, nominatedPodsToClear, err := sched.config.Algorithm.Preempt(preemptor, sched.config.NodeLister, scheduleErr)
    
        var nodeName = ""
        if node != nil {
            nodeName = node.Name
            // 更新队列中“任命pod”队列
            sched.config.SchedulingQueue.UpdateNominatedPodForNode(preemptor, nodeName)
    
            // 设置pod的Status.NominatedNodeName
            err = sched.config.PodPreemptor.SetNominatedNodeName(preemptor, nodeName)
            if err != nil {
                // 如果出错就从 queue 中移除
                sched.config.SchedulingQueue.DeleteNominatedPodIfExists(preemptor)
                return "", err
            }
    
            for _, victim := range victims {
                // 将要驱逐的 pod 驱逐
                if err := sched.config.PodPreemptor.DeletePod(victim); err != nil {
                    return "", err
                }
                sched.config.Recorder.Eventf(victim, v1.EventTypeNormal, "Preempted", "by %v/%v on node %v", preemptor.Namespace, preemptor.Name, nodeName)
            }
        }
        // Clearing nominated pods should happen outside of "if node != nil". 
        // 这个清理过程在上面的if外部，我们回头从 Preempt() 的实现去理解
        for _, p := range nominatedPodsToClear {
            rErr := sched.config.PodPreemptor.RemoveNominatedNodeName(p)
            if rErr != nil {
                klog.Errorf("Cannot remove nominated node annotation of pod: %v", rErr)
                // We do not return as this error is not critical.
            }
        }
        return nodeName, err
    }
    
    // 新获取一次 pod 的信息
    func (p *podPreemptor) GetUpdatedPod(pod *v1.Pod) (*v1.Pod, error) {
       return p.Client.CoreV1().Pods(pod.Namespace).Get(pod.Name, metav1.GetOptions{})
    }
    
    // 删除一个 pod
    func (p *podPreemptor) DeletePod(pod *v1.Pod) error {
       return p.Client.CoreV1().Pods(pod.Namespace).Delete(pod.Name, &metav1.DeleteOptions{})
    }
    
    // 设置pod.Status.NominatedNodeName 为指定的 node name
    func (p *podPreemptor) SetNominatedNodeName(pod *v1.Pod, nominatedNodeName string) error {
       podCopy := pod.DeepCopy()
       podCopy.Status.NominatedNodeName = nominatedNodeName
       _, err := p.Client.CoreV1().Pods(pod.Namespace).UpdateStatus(podCopy)
       return err
    }
    
    // 清空 pod.Status.NominatedNodeName
    func (p *podPreemptor) RemoveNominatedNodeName(pod *v1.Pod) error {
       if len(pod.Status.NominatedNodeName) == 0 {
          return nil
       }
       return p.SetNominatedNodeName(pod, "")
    }
    复制代码

## 3 抢占调度入口-内层评估剔除推荐 ##

### 3.1 内层评估剔除推荐步骤 ###

Preempt的主要实现是找到可以调度的节点和上面因抢占而需要被剔除的pod。基本流程如下：

*  根据调度失败的原因对所有节点先进行一批筛选，筛选出潜在的被调度节点列表。
 *  通过selectNodesForPreemption筛选出需要牺牲的pod和其节点。
 *  基于拓展抢占逻辑再次对上述筛选出来的牺牲者做过滤。
 *  基于上述的过滤结果，选择一个最终可能因抢占被调度的节点。
 *  基于上述的候选节点，找出该节点上优先级低于当前被调度pod的牺牲者pod列表。

### 3.2 核心源码分析 ###

*  Preempt 选择一个 node 然后抢占上面的 pods 资源，返回(这个 node 信息,被抢占的 pods 信息,nominated node name 需要被清理的 node 列表,可能有的 error)

(*v1.Node, []*v1.Pod, []*v1.Pod, error)
    复制代码

*  The nominated pod 会阻止其他 pods 使用“指定”的资源，哪怕花费了很多时间来等待其他 pending 的 pod。
 *  podEligibleToPreemptOthers:有低优先级的 pod 处于删除中状态，就返回 false。
 *  nodesWherePreemptionMightHelp:找 predicates 阶段失败但是通过抢占也许能够调度成功的 nodes。
 *  selectNodesForPreemption：引出selectVictimsOnNode，尝试在给定的 node 中寻找最少数量的需要被驱逐的 pods，同时需要保证驱逐了这些 pods 之后，这个 noode 能够满足该需要抢占Pod运行需求。

计算抢占 node 上所有的低优先级 pods 被驱逐之后能否调度“pod”。
    如果可以，那就按照优先级排序，根据 PDB 是否破坏分成两组，一组是影响 PDB 限制的，另外一组是不影响 PDB。
    两组各自按照优先级排序。然后开始逐渐释放影响 PDB 的 group 中的 pod，
    然后逐渐释放不影响 PDB 的 group 中的 pod，
    在这个过程中要保持“pod”能够 fit 这个 node。
    也就是说一旦放过某一个 pod 导致“pod”不 fit 这个 node 了，
    那就说明这个 pod 不能放过，也就是意味着已经找到了最少 pods 集。
    复制代码

*  pickOneNodeForPreemption:要从给定的 nodes 中选择一个 node，这个函数假设给定的 map 中 value 部分是以 priority 降序排列的。这里选择 node 的标准是：1.最少的 PDB violations 2.最少的高优先级 victim 3.优先级总数字最小 4.victim 总数最小 5.直接返回第一个

func (g *genericScheduler) Preempt(pod *v1.Pod, nodeLister algorithm.NodeLister, scheduleErr error) (*v1.Node, []*v1.Pod, []*v1.Pod, error) {
       // 省略几行
       // 判断执行驱逐操作是否合适
       if !podEligibleToPreemptOthers(pod, g.cachedNodeInfoMap) {
          klog.V(5).Infof("Pod %v/%v is not eligible for more preemption.", pod.Namespace, pod.Name)
          return nil, nil, nil, nil
       }
        // 所有的 nodes
       allNodes, err := nodeLister.List()
       if err != nil {
          return nil, nil, nil, err
       }
       if len(allNodes) == 0 {
          return nil, nil, nil, ErrNoNodesAvailable
       }
        // 计算潜在的执行驱逐后能够用于跑 pod 的 nodes
       potentialNodes := nodesWherePreemptionMightHelp(allNodes, fitError.FailedPredicates)
       if len(potentialNodes) == 0 {
          klog.V(3).Infof("Preemption will not help schedule pod %v/%v on any node.", pod.Namespace, pod.Name)
          // In this case, we should clean-up any existing nominated node name of the pod.
          return nil, nil, []*v1.Pod{pod}, nil
       }
        // 列出 pdb 对象
       pdbs, err := g.pdbLister.List(labels.Everything())
       if err != nil {
          return nil, nil, nil, err
       }
        // 计算所有 node 需要驱逐的 pods 有哪些等，后面细讲
       nodeToVictims, err := selectNodesForPreemption(pod, g.cachedNodeInfoMap, potentialNodes, g.predicates,
          g.predicateMetaProducer, g.schedulingQueue, pdbs)
       if err != nil {
          return nil, nil, nil, err
       }
    
       // 拓展调度的逻辑
       nodeToVictims, err = g.processPreemptionWithExtenders(pod, nodeToVictims)
       if err != nil {
          return nil, nil, nil, err
       }
    
        // 选择1个 node 用于 schedule
       candidateNode := pickOneNodeForPreemption(nodeToVictims)
       if candidateNode == nil {
          return nil, nil, nil, err
       }
    
        // 低优先级的被 nominate 到这个 node 的 pod 很可能已经不再 fit 这个 node 了，所以
        // 需要移除这些 pod 的 nomination，更新这些 pod，挪动到 activeQ 中，让调度器
        // 得以寻找另外一个 node 给这些 pod
       nominatedPods := g.getLowerPriorityNominatedPods(pod, candidateNode.Name)
       if nodeInfo, ok := g.cachedNodeInfoMap[candidateNode.Name]; ok {
          return nodeInfo.Node(), nodeToVictims[candidateNode].Pods, nominatedPods, err
       }
    
       return nil, nil, nil, fmt.Errorf(
          "preemption failed: the target node %s has been deleted from scheduler cache",
          candidateNode.Name)
    }
    复制代码

## 4 总结 ##

Kubernetes 抢占调度Preempt 机制过程非常难以通读，作者花了大量时间和经历进行了代码的走读分析，但本文并不是最合适的代码展现形式，还有更大的优化空间。

转载于:https://juejin.im/post/5d5eb6daf265da03c92700db