kube-scheduler 優先級與搶佔機制源碼分析

2019 年 12 月 20 日
筆記

前面已經分析了 kube-scheduler 的代碼邏輯以及 predicates 與 priorities 算法，本節會繼續講 scheduler 中的一個重要機制，pod 優先級與搶佔機制(Pod Priority and Preemption)，該功能是在 v1.8 中引入的，v1.11 中該功能為 beta 版本且默認啟用了，v1.14 為 stable 版本。

為什麼要有優先級與搶佔機制

正常情況下，當一個 pod 調度失敗後，就會被暫時「擱置」處於 pending 狀態，直到 pod 被更新或者集群狀態發生變化，調度器才會對這個 pod 進行重新調度。但在實際的業務場景中會存在在線與離線業務之分，若在線業務的 pod 因資源不足而調度失敗時，此時就需要離線業務下掉一部分為在線業務提供資源，即在線業務要搶佔離線業務的資源，此時就需要 scheduler 的優先級和搶佔機制了，該機制解決的是 pod 調度失敗時該怎麼辦的問題，若該 pod 的優先級比較高此時並不會被」擱置」，而是會」擠走」某個 node 上的一些低優先級的 pod，這樣就可以保證高優先級的 pod 調度成功。

優先級與搶佔機制源碼分析

kubernetes 版本: v1.16

搶佔發生的原因，一定是一個高優先級的 pod 調度失敗，我們稱這個 pod 為「搶佔者」，稱被搶佔的 pod 為「犧牲者」(victims)。而 kubernetes 調度器實現搶佔算法的一個最重要的設計，就是在調度隊列的實現里，使用了兩個不同的隊列。

第一個隊列叫作 activeQ，凡是在 activeQ 里的 pod，都是下一個調度周期需要調度的對象。所以，當你在 kubernetes 集群里新創建一個 pod 的時候，調度器會將這個 pod 入隊到 activeQ 裏面，調度器不斷從隊列里出隊(pop)一個 pod 進行調度，實際上都是從 activeQ 里出隊的。

第二個隊列叫作 unschedulableQ，專門用來存放調度失敗的 pod，當一個 unschedulableQ 里的 pod 被更新之後，調度器會自動把這個 pod 移動到 activeQ 里，從而給這些調度失敗的 pod 「重新做人」的機會。

當 pod 擁有了優先級之後，高優先級的 pod 就可能會比低優先級的 pod 提前出隊，從而儘早完成調度過程。

k8s.io/kubernetes/pkg/scheduler/internal/queue/scheduling_queue.go

// NewSchedulingQueue initializes a priority queue as a new scheduling queue.  func NewSchedulingQueue(stop <-chan struct{}, fwk framework.Framework) SchedulingQueue {      return NewPriorityQueue(stop, fwk)  }  // NewPriorityQueue creates a PriorityQueue object.  func NewPriorityQueue(stop <-chan struct{}, fwk framework.Framework) *PriorityQueue {      return NewPriorityQueueWithClock(stop, util.RealClock{}, fwk)  }    // NewPriorityQueueWithClock creates a PriorityQueue which uses the passed clock for time.  func NewPriorityQueueWithClock(stop <-chan struct{}, clock util.Clock, fwk framework.Framework) *PriorityQueue {      comp := activeQComp      if fwk != nil {          if queueSortFunc := fwk.QueueSortFunc(); queueSortFunc != nil {              comp = func(podInfo1, podInfo2 interface{}) bool {                  pInfo1 := podInfo1.(*framework.PodInfo)                  pInfo2 := podInfo2.(*framework.PodInfo)                    return queueSortFunc(pInfo1, pInfo2)              }          }      }        pq := &PriorityQueue{          clock:            clock,          stop:             stop,          podBackoff:       NewPodBackoffMap(1*time.Second, 10*time.Second),          activeQ:          util.NewHeapWithRecorder(podInfoKeyFunc, comp, metrics.NewActivePodsRecorder()),          unschedulableQ:   newUnschedulablePodsMap(metrics.NewUnschedulablePodsRecorder()),          nominatedPods:    newNominatedPodMap(),          moveRequestCycle: -1,      }      pq.cond.L = &pq.lock      pq.podBackoffQ = util.NewHeapWithRecorder(podInfoKeyFunc, pq.podsCompareBackoffCompleted, metrics.NewBackoffPodsRecorder())        pq.run()        return pq  }

前面的文章已經說了 scheduleOne() 是執行調度算法的主邏輯，其主要功能有：

調用 sched.schedule()，即執行 predicates 算法和 priorities 算法
若執行失敗，會返回 core.FitError
若開啟了搶佔機制，則執行搶佔機制
……

k8s.io/kubernetes/pkg/scheduler/scheduler.go:516

func (sched *Scheduler) scheduleOne() {      ......      scheduleResult, err := sched.schedule(pod, pluginContext)      // predicates 算法和 priorities 算法執行失敗      if err != nil {          if fitError, ok := err.(*core.FitError); ok {              // 是否開啟搶佔機制              if sched.DisablePreemption {                  .......              } else {              	// 執行搶佔機制                  preemptionStartTime := time.Now()                  sched.preempt(pluginContext, fwk, pod, fitError)                  ......              }              ......          } else {              ......          }          return      }      ......  }

我們主要來看其中的搶佔機制，sched.preempt() 是執行搶佔機制的主邏輯，主要功能有：

從 apiserver 獲取 pod info
調用 sched.Algorithm.Preempt()執行搶佔邏輯，該函數會返回搶佔成功的 node、被搶佔的 pods(victims) 以及需要被移除已提名的 pods
更新 scheduler 緩存，為搶佔者綁定 nodeName，即設定 pod.Status.NominatedNodeName
將 pod info 提交到 apiserver
刪除被搶佔的 pods
刪除被搶佔 pods 的 NominatedNodeName 字段

可以看到當上述搶佔過程發生時，搶佔者並不會立刻被調度到被搶佔的 node 上，調度器只會將搶佔者的 status.nominatedNodeName 字段設置為被搶佔的 node 的名字。然後，搶佔者會重新進入下一個調度周期，在新的調度周期里來決定是不是要運行在被搶佔的節點上，當然，即使在下一個調度周期，調度器也不會保證搶佔者一定會運行在被搶佔的節點上。

這樣設計的一個重要原因是調度器只會通過標準的 DELETE API 來刪除被搶佔的 pod，所以，這些 pod 必然是有一定的「優雅退出」時間（默認是 30s）的。而在這段時間裏，其他的節點也是有可能變成可調度的，或者直接有新的節點被添加到這個集群中來。所以，鑒於優雅退出期間集群的可調度性可能會發生的變化，把搶佔者交給下一個調度周期再處理，是一個非常合理的選擇。而在搶佔者等待被調度的過程中，如果有其他更高優先級的 pod 也要搶佔同一個節點，那麼調度器就會清空原搶佔者的 status.nominatedNodeName 字段，從而允許更高優先級的搶佔者執行搶佔，並且，這也使得原搶佔者本身也有機會去重新搶佔其他節點。以上這些都是設置 nominatedNodeName 字段的主要目的。

k8s.io/kubernetes/pkg/scheduler/scheduler.go:352

func (sched *Scheduler) preempt(pluginContext *framework.PluginContext, fwk framework.Framework, preemptor *v1.Pod, scheduleErr error) (string, error) {      // 獲取 pod info      preemptor, err := sched.PodPreemptor.GetUpdatedPod(preemptor)      if err != nil {          klog.Errorf("Error getting the updated preemptor pod object: %v", err)          return "", err      }        // 執行搶佔算法      node, victims, nominatedPodsToClear, err := sched.Algorithm.Preempt(pluginContext, preemptor, scheduleErr)      if err != nil {          ......      }      var nodeName = ""      if node != nil {          nodeName = node.Name          // 更新 scheduler 緩存，為搶佔者綁定 nodename，即設定 pod.Status.NominatedNodeName          sched.SchedulingQueue.UpdateNominatedPodForNode(preemptor, nodeName)            // 將 pod info 提交到 apiserver          err = sched.PodPreemptor.SetNominatedNodeName(preemptor, nodeName)          if err != nil {              sched.SchedulingQueue.DeleteNominatedPodIfExists(preemptor)              return "", err          }          // 刪除被搶佔的 pods          for _, victim := range victims {              if err := sched.PodPreemptor.DeletePod(victim); err != nil {                  return "", err              }              ......          }      }        // 刪除被搶佔 pods 的 NominatedNodeName 字段      for _, p := range nominatedPodsToClear {          rErr := sched.PodPreemptor.RemoveNominatedNodeName(p)          if rErr != nil {              ......          }      }      return nodeName, err  }

preempt()中會調用 sched.Algorithm.Preempt()來執行實際搶佔的算法，其主要功能有：

判斷 err 是否為 FitError
調用podEligibleToPreemptOthers()確認 pod 是否有搶佔其他 pod 的資格，若 pod 已經搶佔了低優先級的 pod，被搶佔的 pod 處於 terminating 狀態中，則不會繼續進行搶佔
如果確定搶佔可以發生，調度器會把自己緩存的所有節點信息複製一份，然後使用這個副本來模擬搶佔過程
過濾預選失敗的 node 列表，此處會檢查 predicates 失敗的原因，若存在 NodeSelectorNotMatch、PodNotMatchHostName 這些 error 則不能成為搶佔者，如果過濾出的候選 node 為空則返回搶佔者作為 nominatedPodsToClear
獲取 PodDisruptionBudget 對象
從預選失敗的 node 列表中並發計算可以被搶佔的 nodes，得到 nodeToVictims
若聲明了 extenders 則調用 extenders 再次過濾 nodeToVictims
調用 pickOneNodeForPreemption() 從 nodeToVictims 中選出一個節點作為最佳候選人
移除低優先級 pod 的 Nominated，更新這些 pod，移動到 activeQ 隊列中，讓調度器為這些 pod 重新 bind node

k8s.io/kubernetes/pkg/scheduler/core/generic_scheduler.go:320

func (g *genericScheduler) Preempt(pluginContext *framework.PluginContext, pod *v1.Pod, scheduleErr error) (*v1.Node, []*v1.Pod, []*v1.Pod, error) {      fitError, ok := scheduleErr.(*FitError)      if !ok || fitError == nil {          return nil, nil, nil, nil      }      // 判斷 pod 是否支持搶佔，若 pod 已經搶佔了低優先級的 pod，被搶佔的 pod 處於 terminating 狀態中，則不會繼續進行搶佔      if !podEligibleToPreemptOthers(pod, g.nodeInfoSnapshot.NodeInfoMap, g.enableNonPreempting) {          return nil, nil, nil, nil      }      // 從緩存中獲取 node list      allNodes := g.cache.ListNodes()      if len(allNodes) == 0 {          return nil, nil, nil, ErrNoNodesAvailable      }      // 過濾 predicates 算法執行失敗的 node 作為搶佔的候選 node      potentialNodes := nodesWherePreemptionMightHelp(allNodes, fitError)      // 如果過濾出的候選 node 為空則返回搶佔者作為 nominatedPodsToClear      if len(potentialNodes) == 0 {          return nil, nil, []*v1.Pod{pod}, nil      }      // 獲取 PodDisruptionBudget objects      pdbs, err := g.pdbLister.List(labels.Everything())      if err != nil {          return nil, nil, nil, err      }      // 過濾出可以搶佔的 node 列表      nodeToVictims, err := g.selectNodesForPreemption(pluginContext, pod, g.nodeInfoSnapshot.NodeInfoMap, potentialNodes, g.predicates,          g.predicateMetaProducer, g.schedulingQueue, pdbs)      if err != nil {          return nil, nil, nil, err      }        // 若有 extender 則執行      nodeToVictims, err = g.processPreemptionWithExtenders(pod, nodeToVictims)      if err != nil {          return nil, nil, nil, err      }        // 選出最佳的 node      candidateNode := pickOneNodeForPreemption(nodeToVictims)      if candidateNode == nil {          return nil, nil, nil, nil      }        // 移除低優先級 pod 的 Nominated，更新這些 pod，移動到 activeQ 隊列中，讓調度器      // 為這些 pod 重新 bind node      nominatedPods := g.getLowerPriorityNominatedPods(pod, candidateNode.Name)      if nodeInfo, ok := g.nodeInfoSnapshot.NodeInfoMap[candidateNode.Name]; ok {          return nodeInfo.Node(), nodeToVictims[candidateNode].Pods, nominatedPods, nil      }        return nil, nil, nil, fmt.Errorf(          "preemption failed: the target node %s has been deleted from scheduler cache",          candidateNode.Name)  }

該函數中調用了多個函數： nodesWherePreemptionMightHelp()：過濾 predicates 算法執行失敗的 node selectNodesForPreemption()：過濾出可以搶佔的 node 列表 pickOneNodeForPreemption()：選出最佳的 node getLowerPriorityNominatedPods()：移除低優先級 pod 的 Nominated

selectNodesForPreemption() 從 prediacates 算法執行失敗的 node 列表中來尋找可以被搶佔的 node，通過workqueue.ParallelizeUntil()並發執行checkNode()函數檢查 node。

k8s.io/kubernetes/pkg/scheduler/core/generic_scheduler.go:996

func (g *genericScheduler) selectNodesForPreemption(     ......     ) (map[*v1.Node]*schedulerapi.Victims, error) {      nodeToVictims := map[*v1.Node]*schedulerapi.Victims{}      var resultLock sync.Mutex        meta := metadataProducer(pod, nodeNameToInfo)      // checkNode 函數      checkNode := func(i int) {          nodeName := potentialNodes[i].Name          var metaCopy predicates.PredicateMetadata          if meta != nil {              metaCopy = meta.ShallowCopy()          }          // 調用 selectVictimsOnNode 函數進行檢查          pods, numPDBViolations, fits := g.selectVictimsOnNode(pluginContext, pod, metaCopy, nodeNameToInfo[nodeName], fitPredicates, queue, pdbs)          if fits {              resultLock.Lock()              victims := schedulerapi.Victims{                  Pods:             pods,                  NumPDBViolations: numPDBViolations,              }              nodeToVictims[potentialNodes[i]] = &victims              resultLock.Unlock()          }      }      // 啟動 16 個 goroutine 並發執行      workqueue.ParallelizeUntil(context.TODO(), 16, len(potentialNodes), checkNode)      return nodeToVictims, nil  }

其中調用的selectVictimsOnNode()是來獲取每個 node 上 victims pod 的，首先移除所有低優先級的 pod 嘗試搶佔者是否可以調度成功，如果能夠調度成功，然後基於 pod 是否有 PDB 被分為兩組 violatingVictims 和 nonViolatingVictims，再對每一組的 pod 按優先級進行排序。PDB(pod 中斷預算)是 kubernetes 保證副本高可用的一個對象。

然後開始逐一」刪除「 pod 即要刪掉最少的 pod 數來完成這次搶佔即可，先從 violatingVictims(有PDB)的一組中進行」刪除「 pod，並且記錄刪除有 PDB pod 的數量，然後再「刪除」 nonViolatingVictims 組中的 pod，每次」刪除「一個 pod 都要檢查一下搶佔者是否能夠運行在該 node 上即執行一次預選策略，若執行預選策略失敗則該 node 當前不滿足搶佔需要繼續」刪除「 pod 並將該 pod 加入到 victims 中，直到」刪除「足夠多的 pod 可以滿足搶佔，最後返回 victims 以及刪除有 PDB pod 的數量。

k8s.io/kubernetes/pkg/scheduler/core/generic_scheduler.go:1086

func (g *genericScheduler) selectVictimsOnNode(  		......  ) ([]*v1.Pod, int, bool) {      if nodeInfo == nil {          return nil, 0, false      }        potentialVictims := util.SortableList{CompFunc: util.MoreImportantPod}      nodeInfoCopy := nodeInfo.Clone()        removePod := func(rp *v1.Pod) {          nodeInfoCopy.RemovePod(rp)          if meta != nil {              meta.RemovePod(rp, nodeInfoCopy.Node())          }      }      addPod := func(ap *v1.Pod) {          nodeInfoCopy.AddPod(ap)          if meta != nil {              meta.AddPod(ap, nodeInfoCopy)          }      }      // 先刪除所有的低優先級 pod 檢查是否能滿足搶佔 pod 的調度需求      podPriority := util.GetPodPriority(pod)      for _, p := range nodeInfoCopy.Pods() {          if util.GetPodPriority(p) < podPriority {              potentialVictims.Items = append(potentialVictims.Items, p)              removePod(p)          }      }      // 如果刪除所有低優先級的 pod 不符合要求則直接過濾掉該 node      // podFitsOnNode 就是前文講過用來執行預選函數的      if fits, _, _, err := g.podFitsOnNode(pluginContext, pod, meta, nodeInfoCopy, fitPredicates, queue, false); !fits {          if err != nil {  						......          }          return nil, 0, false      }      var victims []*v1.Pod      numViolatingVictim := 0      potentialVictims.Sort()        // 嘗試盡量多地「刪除」這些 pods，先從 PDB violating victims 中「刪除」，再從 PDB non-violating victims 中「刪除」      violatingVictims, nonViolatingVictims := filterPodsWithPDBViolation(potentialVictims.Items, pdbs)        // reprievePod 是「刪除」 pods 的函數      reprievePod := func(p *v1.Pod) bool {          addPod(p)          // 同樣也會調用 podFitsOnNode 再次執行 predicates 算法          fits, _, _, _ := g.podFitsOnNode(pluginContext, pod, meta, nodeInfoCopy, fitPredicates, queue, false)          if !fits {              removePod(p)              // 加入到 victims 中              victims = append(victims, p)          }          return fits      }       // 刪除 violatingVictims 中的 pod，同時也記錄刪除了多少個      for _, p := range violatingVictims {          if !reprievePod(p) {              numViolatingVictim++          }      }      // 刪除 nonViolatingVictims 中的 pod      for _, p := range nonViolatingVictims {          reprievePod(p)      }      return victims, numViolatingVictim, true  }

pickOneNodeForPreemption() 用來選出最佳的 node 作為搶佔者的 node，該函數主要基於 6 個原則：

PDB violations 值最小的 node
挑選具有高優先級較少的 node
對每個 node 上所有 victims 的優先級進項累加，選取最小的
如果多個 node 優先級總和相等，選擇具有最小 victims 數量的 node
如果多個 node 優先級總和相等，選擇具有高優先級且 pod 運行時間最短的
如果依據以上策略仍然選出了多個 node 則直接返回第一個 node

k8s.io/kubernetes/pkg/scheduler/core/generic_scheduler.go:867

func pickOneNodeForPreemption(nodesToVictims map[*v1.Node]*schedulerapi.Victims) *v1.Node {      if len(nodesToVictims) == 0 {          return nil      }      minNumPDBViolatingPods := math.MaxInt32      var minNodes1 []*v1.Node      lenNodes1 := 0      for node, victims := range nodesToVictims {          if len(victims.Pods) == 0 {  	        // 若該 node 沒有 victims 則返回              return node          }          numPDBViolatingPods := victims.NumPDBViolations          if numPDBViolatingPods < minNumPDBViolatingPods {              minNumPDBViolatingPods = numPDBViolatingPods              minNodes1 = nil              lenNodes1 = 0          }          if numPDBViolatingPods == minNumPDBViolatingPods {              minNodes1 = append(minNodes1, node)              lenNodes1++          }      }      if lenNodes1 == 1 {          return minNodes1[0]      }        // 選出 PDB violating pods 數量最少的或者高優先級 victim 數量少的      minHighestPriority := int32(math.MaxInt32)      var minNodes2 = make([]*v1.Node, lenNodes1)      lenNodes2 := 0      for i := 0; i < lenNodes1; i++ {          node := minNodes1[i]          victims := nodesToVictims[node]          highestPodPriority := util.GetPodPriority(victims.Pods[0])          if highestPodPriority < minHighestPriority {              minHighestPriority = highestPodPriority              lenNodes2 = 0          }          if highestPodPriority == minHighestPriority {              minNodes2[lenNodes2] = node              lenNodes2++          }      }      if lenNodes2 == 1 {          return minNodes2[0]      }      // 若多個 node 高優先級的 pod 同樣少，則選出加權得分最小的      minSumPriorities := int64(math.MaxInt64)      lenNodes1 = 0      for i := 0; i < lenNodes2; i++ {          var sumPriorities int64          node := minNodes2[i]          for _, pod := range nodesToVictims[node].Pods {              sumPriorities += int64(util.GetPodPriority(pod)) + int64(math.MaxInt32+1)          }          if sumPriorities < minSumPriorities {              minSumPriorities = sumPriorities              lenNodes1 = 0          }          if sumPriorities == minSumPriorities {              minNodes1[lenNodes1] = node              lenNodes1++          }      }      if lenNodes1 == 1 {          return minNodes1[0]      }      // 若多個 node 高優先級的 pod 數量同等且加權分數相等，則選出 pod 數量最少的      minNumPods := math.MaxInt32      lenNodes2 = 0      for i := 0; i < lenNodes1; i++ {          node := minNodes1[i]          numPods := len(nodesToVictims[node].Pods)          if numPods < minNumPods {              minNumPods = numPods              lenNodes2 = 0          }          if numPods == minNumPods {              minNodes2[lenNodes2] = node              lenNodes2++          }      }      if lenNodes2 == 1 {          return minNodes2[0]      }      // 若多個 node 的 pod 數量相等，則選出高優先級 pod 啟動時間最短的      latestStartTime := util.GetEarliestPodStartTime(nodesToVictims[minNodes2[0]])      if latestStartTime == nil {          return minNodes2[0]      }      nodeToReturn := minNodes2[0]      for i := 1; i < lenNodes2; i++ {          node := minNodes2[i]          earliestStartTimeOnNode := util.GetEarliestPodStartTime(nodesToVictims[node])          if earliestStartTimeOnNode == nil {              klog.Errorf("earliestStartTime is nil for node %s. Should not reach here.", node)              continue          }          if earliestStartTimeOnNode.After(latestStartTime.Time) {              latestStartTime = earliestStartTimeOnNode              nodeToReturn = node          }      }        return nodeToReturn  }

以上就是對搶佔機制代碼的一個通讀。

優先級與搶佔機制的使用

1、創建 PriorityClass 對象：

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."

2、在 deployment、statefulset 或者 pod 中聲明使用已有的 priorityClass 對象即可

在 pod 中使用：

apiVersion: v1  kind: Pod  metadata:    labels:      app: nginx-a    name: nginx-a  spec:    containers:    - image: nginx:1.7.9      imagePullPolicy: IfNotPresent      name: nginx-a      ports:      - containerPort: 80        protocol: TCP      resources:        requests:          memory: "64Mi"          cpu: 5        limits:          memory: "128Mi"          cpu: 5    priorityClassName: high-priority

在 deployment 中使用：

template:    spec:      containers:      - image: nginx        name: nginx-deployment        priorityClassName: high-priority

3、測試過程中可以看到高優先級的 nginx-a 會搶佔 nginx-5754944d6c 的資源：

$ kubectl get pod -o  wide -w  NAME                     READY   STATUS    RESTARTS   AGE   IP           NODE          NOMINATED NODE   READINESS GATES  nginx-5754944d6c-9mnxa   1/1     Running   0          37s   10.244.1.4   test-worker   <none>           <none>  nginx-a                  0/1     Pending   0          0s    <none>       <none>        <none>           <none>  nginx-a                  0/1     Pending   0          0s    <none>       <none>        <none>           <none>  nginx-a                  0/1     Pending   0          0s    <none>       <none>        test-worker      <none>  nginx-5754944d6c-9mnxa   1/1     Terminating   0          45s   10.244.1.4   test-worker   <none>           <none>  nginx-5754944d6c-9mnxa   0/1     Terminating   0          46s   10.244.1.4   test-worker   <none>           <none>  nginx-5754944d6c-9mnxa   0/1     Terminating   0          47s   10.244.1.4   test-worker   <none>           <none>  nginx-5754944d6c-9mnxa   0/1     Terminating   0          47s   10.244.1.4   test-worker   <none>           <none>  nginx-a                  0/1     Pending       0          2s    <none>       test-worker   test-worker      <none>  nginx-a                  0/1     ContainerCreating   0          2s    <none>       test-worker   <none>           <none>  nginx-a                  1/1     Running             0          4s    10.244.1.5   test-worker   <none>           <none>

總結

這篇文章主要講述 kube-scheduler 中的優先級與搶佔機制，可以看到搶佔機制比 predicates 與 priorities 算法都要複雜，其中的許多細節仍然沒有提到，本文只是通讀了大部分代碼，某些代碼的實現需要精讀，限於筆者時間的關係，對於 kube-scheduler 的代碼暫時分享到此處。

參考：

https://kubernetes.io/docs/concepts/configuration/pod-priority-preemption/