源码分析系列文章已经开源到github，地址如下：
- github：
https://github.com/farmer-hutao/k8s-source-code-analysis - gitbook：
https://farmer-hutao.github.io/k8s-source-code-analysis
本文大纲
1. Pod priority
Pod 有了 priority(优先级) 后才有优先级调度、抢占调度的说法，高优先级的 pod 可以在调度队列中排到前面，优先选择 node；另外当高优先级的 pod 找不到合适的 node 时，就会看 node 上低优先级的 pod 驱逐之后是否能够 run 起来，如果可以，那么 node 上的一个或多个低优先级的 pod 会被驱逐，然后高优先级的 pod 得以成功运行1个 node 上。
今天我们分析 pod 抢占相关的代码。开始之前我们看一下和 priority 相关的2个示例配置文件：
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."
Copy
使用上述 PriorityClass
apiVersion: v1
kind: Pod
metadata:
name: nginx
labels:
env: test
spec:
containers:
- name: nginx
image: nginx
imagePullPolicy: IfNotPresent
priorityClassName: high-priority
Copy
这两个文件的内容这里不解释，Pod priority 相关知识点不熟悉的小伙伴请先查阅官方文档，我们下面看调度器中和 preempt 相关的代码逻辑。
2. preempt 入口
在pkg/scheduler/scheduler.go:513 scheduleOne()方法中我们上一次关注的是suggestedHost, err := sched.schedule(pod)这行代码，也就是关注通常情况下调度器如何给一个 pod 匹配一个最合适的 node. 今天我们来看如果这一行代码返回的 err != nil 情况下，如何开始 preempt 过程。
pkg/scheduler/scheduler.go:529
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
}
Copy
当schedule()函数没有返回 host，也就是没有找到合适的 node 的时候，就会出发 preempt 过程。这时候代码逻辑进入sched.preempt(pod, fitError)这一行。我们先看一下这个函数的整体逻辑，然后深入其中涉及的子过程：
pkg/scheduler/scheduler.go:311
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 过程，下文分析
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
}
Copy
3. preempt 实现
上面 preempt() 函数中涉及到了一些值得深入看看的对象，下面我们逐个看一下这些对象的实现。
3.1. SchedulingQueue
SchedulingQueue 表示的是一个存储待调度 pod 的队列
pkg/scheduler/internal/queue/scheduling_queue.go:60
type SchedulingQueue interface {
Add(pod *v1.Pod) error
AddIfNotPresent(pod *v1.Pod) error
AddUnschedulableIfNotPresent(pod *v1.Pod) error
Pop() (*v1.Pod, error)
Update(oldPod, newPod *v1.Pod) error
Delete(pod *v1.Pod) error
MoveAllToActiveQueue()
AssignedPodAdded(pod *v1.Pod)
AssignedPodUpdated(pod *v1.Pod)
NominatedPodsForNode(nodeName string) []*v1.Pod
WaitingPods() []*v1.Pod
Close()
UpdateNominatedPodForNode(pod *v1.Pod, nodeName string)
DeleteNominatedPodIfExists(pod *v1.Pod)
NumUnschedulablePods() int
}
Copy
在 Scheduler 中 SchedulingQueue 接口对应两种实现：
- FIFO 先进先出队列
- PriorityQueue 优先级队列
3.1.1. FIFO
FIFO 结构是对 cache.FIFO 的简单包装，然后实现了 SchedulingQueue 接口。
pkg/scheduler/internal/queue/scheduling_queue.go:97
type FIFO struct {
*cache.FIFO
}
Copy
cache.FIFO定义在vendor/k8s.io/client-go/tools/cache/fifo.go:93，这个先进先出队列的细节先不讨论。
3.1.2. PriorityQueue
PriorityQueue 同样实现了 SchedulingQueue 接口，PriorityQueue 的顶是最高优先级的 pending pod. 这里的PriorityQueue 有2个子 queue，activeQ 放的是等待调度的 pod，unschedulableQ 放的是已经尝试过调度，然后失败了，被标记为 unschedulable 的 pod.
我们看一下 PriorityQueue 结构的定义：
pkg/scheduler/internal/queue/scheduling_queue.go:201
type PriorityQueue struct {
stop <-chan struct{}
clock util.Clock
lock sync.RWMutex
cond sync.Cond
// heap 头节点存的是最高优先级的 pod
activeQ *Heap
// unschedulableQ holds pods that have been tried and determined unschedulable.
unschedulableQ *UnschedulablePodsMap
// 存储已经被指定好要跑在某个 node 的 pod
nominatedPods *nominatedPodMap
// 只要将 pod 从 unschedulableQ 移动到 activeQ，就设置为true；从 activeQ 中 pop 出来 pod的时候设置为 false. 这个字段表明一个 pod 在被调度的过程中是否接收到了队列 move 操作，如果发生了 move 操作，那么这个 pod 就算被认定为 unschedulable，也被放回到 activeQ.
receivedMoveRequest bool
closed bool
}
Copy
PriorityQueue 的方法比较好理解，我们看几个吧：
1、func (p *PriorityQueue) Add(pod *v1.Pod) error //在 active queue 中添加1个pod
pkg/scheduler/internal/queue/scheduling_queue.go:276
func (p *PriorityQueue) Add(pod *v1.Pod) error {
p.lock.Lock()
defer p.lock.Unlock()
// 直接在 activeQ 中添加 pod
err := p.activeQ.Add(pod)
if err != nil {
klog.Errorf("Error adding pod %v/%v to the scheduling queue: %v", pod.Namespace, pod.Name, err)
} else {
// 如果在 unschedulableQ 中找到这个 pod，抛错误日志后移除队列中该 pod
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)
}
// 队列的 nominatedPods 属性中标记该 pod 不指定到任何 node
p.nominatedPods.add(pod, "")
p.cond.Broadcast()
}
return err
}
Copy
2、func (p *PriorityQueue) AddIfNotPresent(pod *v1.Pod) error//如果2个队列中都不存在该 pod，那么就添加到 active queue 中
pkg/scheduler/internal/queue/scheduling_queue.go:295
func (p *PriorityQueue) AddIfNotPresent(pod *v1.Pod) error {
p.lock.Lock()
defer p.lock.Unlock()
//如果队列 unschedulableQ 中有 pod，啥也不做
if p.unschedulableQ.get(pod) != nil {
return nil
}
//如果队列 activeQ 中有 pod，啥也不做
if _, exists, _ := p.activeQ.Get(pod); exists {
return nil
}
// 添加 pod 到 activeQ
err := p.activeQ.Add(pod)
if err != nil {
klog.Errorf("Error adding pod %v/%v to the scheduling queue: %v", pod.Namespace, pod.Name, err)
} else {
p.nominatedPods.add(pod, "")
p.cond.Broadcast()
}
return err
}
Copy
3、func (p *PriorityQueue) flushUnschedulableQLeftover()//刷新 unschedulableQ 中的 pod，如果一个 pod 的呆的时间超过了 durationStayUnschedulableQ，就移动到 activeQ 中
pkg/scheduler/internal/queue/scheduling_queue.go:346
func (p *PriorityQueue) flushUnschedulableQLeftover() {
p.lock.Lock()
defer p.lock.Unlock()
var podsToMove []*v1.Pod
currentTime := p.clock.Now()
// 遍历 unschedulableQ 中的 pod
for _, pod := range p.unschedulableQ.pods {
lastScheduleTime := podTimestamp(pod)
// 这里的默认值是 60s，所以超过 60s 的 pod 将得到进入 activeQ 的机会
if !lastScheduleTime.IsZero() && currentTime.Sub(lastScheduleTime.Time) > unschedulableQTimeInterval {
podsToMove = append(podsToMove, pod)
}
}
if len(podsToMove) > 0 {
// 全部移到 activeQ 中，又有机会被调度了
p.movePodsToActiveQueue(podsToMove)
}
}
Copy
4、func (p *PriorityQueue) Pop() (*v1.Pod, error)//从 activeQ 中 pop 一个 pod
pkg/scheduler/internal/queue/scheduling_queue.go:367
func (p *PriorityQueue) Pop() (*v1.Pod, error) {
p.lock.Lock()
defer p.lock.Unlock()
for len(p.activeQ.data.queue) == 0 {
// 当队列为空的时候会阻塞
if p.closed {
return nil, fmt.Errorf(queueClosed)
}
p.cond.Wait()
}
obj, err := p.activeQ.Pop()
if err != nil {
return nil, err
}
pod := obj.(*v1.Pod)
// 标记 receivedMoveRequest 为 false，表示新的一次调度开始了
p.receivedMoveRequest = false
return pod, err
}
Copy
再看个别 PriorityQueue.nominatedPods 属性相关操作的方法，也就是 preempt() 函数中多次调用到的方法：
5、`func (p PriorityQueue) UpdateNominatedPodForNode(pod v1.Pod, nodeName string)`//pod 抢占的时候，确定一个 node 可以用于跑这个 pod 时，通过调用这个方法将 pod nominated 到 指定的 node 上。
pkg/scheduler/internal/queue/scheduling_queue.go:567
func (p *PriorityQueue) UpdateNominatedPodForNode(pod *v1.Pod, nodeName string) {
p.lock.Lock()
//逻辑在这里面
p.nominatedPods.add(pod, nodeName)
p.lock.Unlock()
}
Copy
先看 nominatedPods 属性的类型，这个类型用于存储 pods 被 nominate 到 nodes 的信息：
pkg/scheduler/internal/queue/scheduling_queue.go:822
type nominatedPodMap struct {
// key 是 node name，value 是 nominated 到这个 node 上的 pods
nominatedPods map[string][]*v1.Pod
// 和上面结构相反，key 是 pod 信息，值是 node 信息
nominatedPodToNode map[ktypes.UID]string
}
Copy
在看一下add()方法的实现：
pkg/scheduler/internal/queue/scheduling_queue.go:832
func (npm *nominatedPodMap) add(p *v1.Pod, nodeName string) {
// 不管有没有，先删一下，防止重了
npm.delete(p)
nnn := nodeName
// 如果传入的 nodeName 是 “”
if len(nnn) == 0 {
// 查询 pod 的 pod.Status.NominatedNodeName
nnn = NominatedNodeName(p)
// 如果 pod.Status.NominatedNodeName 也是 “”,return
if len(nnn) == 0 {
return
}
}
// 逻辑到这里说明要么 nodeName 不为空字符串，要么 nodeName 为空字符串但是 pod 的 pod.Status.NominatedNodeName 不为空字符串，这时候开始下面的赋值
npm.nominatedPodToNode[p.UID] = nnn
for _, np := range npm.nominatedPods[nnn] {
if np.UID == p.UID {
klog.V(4).Infof("Pod %v/%v already exists in the nominated map!", p.Namespace, p.Name)
return
}
}
npm.nominatedPods[nnn] = append(npm.nominatedPods[nnn], p)
}
Copy
3.2. PodPreemptor
PodPreemptor 用来驱逐 pods 和更新 pod annotations.
pkg/scheduler/factory/factory.go:145
type PodPreemptor interface {
GetUpdatedPod(pod *v1.Pod) (*v1.Pod, error)
DeletePod(pod *v1.Pod) error
SetNominatedNodeName(pod *v1.Pod, nominatedNode string) error
RemoveNominatedNodeName(pod *v1.Pod) error
}
Copy
这个 interface 对应的实现类型是：
pkg/scheduler/factory/factory.go:1620
type podPreemptor struct {
Client clientset.Interface
}
Copy
这个类型绑定了4个方法：
pkg/scheduler/factory/factory.go:1624
// 新获取一次 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, "")
}
Copy
3.3. xx.Algorithm.Preempt
3.3.1. 接口定义
我们回到挺久之前讲常规调度过程的时候提过的一个接口：
pkg/scheduler/algorithm/scheduler_interface.go:78
type ScheduleAlgorithm interface {
Schedule(*v1.Pod, NodeLister) (selectedMachine string, err error)
// Preempt 在 pod 调度发生失败的时候尝试抢占低优先级的 pod.
// 返回发生 preemption 的 node, 被 preempt的 pods 列表,
// nominated node name 需要被移除的 pods 列表，一个 error 信息.
Preempt(*v1.Pod, NodeLister, error) (selectedNode *v1.Node, preemptedPods []*v1.Pod, cleanupNominatedPods []*v1.Pod, err error)
Predicates() map[string]FitPredicate
Prioritizers() []PriorityConfig
}
Copy
这个接口上次我们讲到的时候关注了Schedule()、Predicates()和Prioritizers()，这次来看Preempt()是怎么实现的。
3.3.2. 整体流程
Preempt()同样由genericScheduler类型(pkg/scheduler/core/generic_scheduler.go:98)实现，方法前的一大串英文注释先来理解一下：
- Preempt 寻找一个在发生抢占之后能够成功调度“pod”的node.
- Preempt 选择一个 node 然后抢占上面的 pods 资源，返回：
- 这个 node 信息
- 被抢占的 pods 信息
- nominated node name 需要被清理的 node 列表
- 可能有的 error
- Preempt 过程不涉及快照更新（快照的逻辑以后再讲）
- 避免出现这种情况：preempt 发现一个不需要驱逐任何 pods 就能够跑“pod”的 node.
- 当有很多 pending pods 在调度队列中的时候，a nominated pod 会排到队列中相同优先级的 pod 后面.
- The nominated pod 会阻止其他 pods 使用“指定”的资源，哪怕花费了很多时间来等待其他 pending 的 pod.
我们先过整体流程，然后逐个分析子流程调用：
pkg/scheduler/core/generic_scheduler.go:251
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)
}
Copy
上面涉及到一些子过程调用，我们逐个来看～
podEligibleToPreemptOthers()// 如何判断是否适合抢占？nodesWherePreemptionMightHelp()// 怎么寻找能够用于 preempt 的 nodes？selectNodesForPreemption()// 这个过程计算的是什么？pickOneNodeForPreemption()// 怎么从选择最合适被抢占的 node？
3.3.3. podEligibleToPreemptOthers
podEligibleToPreemptOthers做的事情是判断一个 pod 是否应该去抢占其他 pods. 如果这个 pod 已经抢占过其他 pods，那些 pods 还在 graceful termination period 中，那就不应该再次发生抢占。- 如果一个 node 已经被这个 pod nominated，并且这个 node 上有处于 terminating 状态的 pods，那么就不考虑驱逐更多的 pods.
这个函数逻辑很简单，我们直接看源码：
pkg/scheduler/core/generic_scheduler.go:1110
func podEligibleToPreemptOthers(pod *v1.Pod, nodeNameToInfo map[string]*schedulercache.NodeInfo) bool {
nomNodeName := pod.Status.NominatedNodeName
// 如果 pod.Status.NominatedNodeName 不是空字符串
if len(nomNodeName) > 0 {
// 被 nominate 的 node
if nodeInfo, found := nodeNameToInfo[nomNodeName]; found {
for _, p := range nodeInfo.Pods() {
// 有低优先级的 pod 处于删除中状态，就返回 false
if p.DeletionTimestamp != nil && util.GetPodPriority(p) < util.GetPodPriority(pod) {
// There is a terminating pod on the nominated node.
return false
}
}
}
}
return true
}
Copy
3.3.4. nodesWherePreemptionMightHelp
nodesWherePreemptionMightHelp 要做的事情是寻找 predicates 阶段失败但是通过抢占也许能够调度成功的 nodes.
这个函数也不怎么长，看下代码：
pkg/scheduler/core/generic_scheduler.go:1060
func nodesWherePreemptionMightHelp(nodes []*v1.Node, failedPredicatesMap FailedPredicateMap) []*v1.Node {
// 潜力 node， 用于存储返回值的 slice
potentialNodes := []*v1.Node{}
for _, node := range nodes {
// 这个为 true 表示一个 node 驱逐 pod 也不一定能适合当前 pod 运行
unresolvableReasonExist := false
// 一个 node 对应的所有失败的 predicates
failedPredicates, _ := failedPredicatesMap[node.Name]
// 遍历，看是不是再下面指定的这些原因中，如果在，就标记 unresolvableReasonExist = true
for _, failedPredicate := range failedPredicates {
switch failedPredicate {
case
predicates.ErrNodeSelectorNotMatch,
predicates.ErrPodAffinityRulesNotMatch,
predicates.ErrPodNotMatchHostName,
predicates.ErrTaintsTolerationsNotMatch,
predicates.ErrNodeLabelPresenceViolated,
predicates.ErrNodeNotReady,
predicates.ErrNodeNetworkUnavailable,
predicates.ErrNodeUnderDiskPressure,
predicates.ErrNodeUnderPIDPressure,
predicates.ErrNodeUnderMemoryPressure,
predicates.ErrNodeOutOfDisk,
predicates.ErrNodeUnschedulable,
predicates.ErrNodeUnknownCondition,
predicates.ErrVolumeZoneConflict,
predicates.ErrVolumeNodeConflict,
predicates.ErrVolumeBindConflict:
unresolvableReasonExist = true
// 如果找到一个上述失败原因，说明这个 node 已经可以排除了，break 后继续下一个 node 的计算
break
}
}
// false 的时候，也就是这个 node 也许驱逐 pods 后有用，那就添加到 potentialNodes 中
if !unresolvableReasonExist {
klog.V(3).Infof("Node %v is a potential node for preemption.", node.Name)
potentialNodes = append(potentialNodes, node)
}
}
return potentialNodes
}
Copy
3.3.5. selectNodesForPreemption
这个函数会并发计算所有的 nodes 是否通过驱逐实现 pod 抢占。
看这个函数内容之前我们先看一下返回值的类型：
map[*v1.Node]*schedulerapi.Victims 的 key 很好理解，value 是啥呢：
type Victims struct {
Pods []*v1.Pod
NumPDBViolations int
}
Copy
这里的 Pods 是被选中准备要驱逐的；NumPDBViolations 表示的是要破坏多少个 PDB 限制。这里肯定也就是要尽量符合 PDB 要求，能不和 PDB 冲突就不冲突。
然后看一下这个函数的整体过程：
pkg/scheduler/core/generic_scheduler.go:895
func selectNodesForPreemption(pod *v1.Pod,
nodeNameToInfo map[string]*schedulercache.NodeInfo,
potentialNodes []*v1.Node, // 上一个函数计算出来的 nodes
predicates map[string]algorithm.FitPredicate,
metadataProducer algorithm.PredicateMetadataProducer,
queue internalqueue.SchedulingQueue, // 这里其实是前面讲的优先级队列 PriorityQueue
pdbs []*policy.PodDisruptionBudget, // pdb 列表
) (map[*v1.Node]*schedulerapi.Victims, error) {
nodeToVictims := map[*v1.Node]*schedulerapi.Victims{}
var resultLock sync.Mutex
// We can use the same metadata producer for all nodes.
meta := metadataProducer(pod, nodeNameToInfo)
// 这种形式的并发已经不陌生了，前面遇到过几次了
checkNode := func(i int) {
nodeName := potentialNodes[i].Name
var metaCopy algorithm.PredicateMetadata
if meta != nil {
metaCopy = meta.ShallowCopy()
}
// 这里有一个子过程调用，下面单独介绍
pods, numPDBViolations, fits := selectVictimsOnNode(pod, metaCopy, nodeNameToInfo[nodeName], predicates, queue, pdbs)
if fits {
resultLock.Lock()
victims := schedulerapi.Victims{
Pods: pods,
NumPDBViolations: numPDBViolations,
}
// 如果 fit，就添加到 nodeToVictims 中，也就是最后的返回值
nodeToVictims[potentialNodes[i]] = &victims
resultLock.Unlock()
}
}
workqueue.ParallelizeUntil(context.TODO(), 16, len(potentialNodes), checkNode)
return nodeToVictims, nil
}
Copy
上面这个函数的核心逻辑在 selectVictimsOnNode 中，这个函数尝试在给定的 node 中寻找最少数量的需要被驱逐的 pods，同时需要保证驱逐了这些 pods 之后，这个 noode 能够满足“pod”运行需求。
这些被驱逐的 pods 计算同时需要满足一个约束，就是能够删除低优先级的 pod 绝不先删高优先级的 pod.
这个算法首选计算当这个 node 上所有的低优先级 pods 被驱逐之后能否调度“pod”. 如果可以，那就按照优先级排序，根据 PDB 是否破坏分成两组，一组是影响 PDB 限制的，另外一组是不影响 PDB. 两组各自按照优先级排序。然后开始逐渐释放影响 PDB 的 group 中的 pod，然后逐渐释放不影响 PDB 的 group 中的 pod，在这个过程中要保持“pod”能够 fit 这个 node. 也就是说一旦放过某一个 pod 导致“pod”不 fit 这个 node 了，那就说明这个 pod 不能放过，也就是意味着已经找到了最少 pods 集。
看一下具体的实现吧：
FILENAME pkg/scheduler/core/generic_scheduler.go:983
func selectVictimsOnNode(
pod *v1.Pod,
meta algorithm.PredicateMetadata,
nodeInfo *schedulercache.NodeInfo,
fitPredicates map[string]algorithm.FitPredicate,
queue internalqueue.SchedulingQueue,
pdbs []*policy.PodDisruptionBudget,
) ([]*v1.Pod, int, bool) {
if nodeInfo == nil {
return nil, 0, false
}
// 排个序
potentialVictims := util.SortableList{CompFunc: util.HigherPriorityPod}
nodeInfoCopy := nodeInfo.Clone()
// 定义删除 pod 函数
removePod := func(rp *v1.Pod) {
nodeInfoCopy.RemovePod(rp)
if meta != nil {
meta.RemovePod(rp)
}
}
// 定义添加 pod 函数
addPod := func(ap *v1.Pod) {
nodeInfoCopy.AddPod(ap)
if meta != nil {
meta.AddPod(ap, nodeInfoCopy)
}
}
// 删除所有的低优先级 pod 看是不是能够满足调度需求了
podPriority := util.GetPodPriority(pod)
for _, p := range nodeInfoCopy.Pods() {
if util.GetPodPriority(p) < podPriority {
// 删除的意思其实就是添加元素到 potentialVictims.Items
potentialVictims.Items = append(potentialVictims.Items, p)
removePod(p)
}
}
// 排个序
potentialVictims.Sort()
// 如果删除了所有的低优先级 pods 之后还不能跑这个新 pod，那么差不多就可以判断这个 node 不适合 preemption 了，还有一点点需要考虑的是这个“pod”的不 fit 的原因是由于 pod affinity 不满足了。
// 后续可能会增加当前 pod 和低优先级 pod 之间的 优先级检查。
// 这个函数调用其实就是之前讲到过的预选函数的调用逻辑，判断这个 pod 是否合适跑在这个 node 上。
if fits, _, err := podFitsOnNode(pod, meta, nodeInfoCopy, fitPredicates, nil, queue, false, nil); !fits {
if err != nil {
klog.Warningf("Encountered error while selecting victims on node %v: %v", nodeInfo.Node().Name, err)
}
return nil, 0, false
}
var victims []*v1.Pod
numViolatingVictim := 0
// 尝试尽量多地释放这些 pods，也就是说能少杀就少杀；这里先从 PDB violating victims 中释放，再从 PDB non-violating victims 中释放；两个组都是从高优先级的 pod 开始释放。
violatingVictims, nonViolatingVictims := filterPodsWithPDBViolation(potentialVictims.Items, pdbs)
// 释放 pods 的函数，来一个放一个
reprievePod := func(p *v1.Pod) bool {
addPod(p)
fits, _, _ := podFitsOnNode(pod, meta, nodeInfoCopy, fitPredicates, nil, queue, false, nil)
if !fits {
removePod(p)
victims = append(victims, p)
klog.V(5).Infof("Pod %v is a potential preemption victim on node %v.", p.Name, nodeInfo.Node().Name)
}
return fits
}
// 释放 violatingVictims 中元素的同时会记录放了多少个
for _, p := range violatingVictims {
if !reprievePod(p) {
numViolatingVictim++
}
}
// 开始释放 non-violating victims.
for _, p := range nonViolatingVictims {
reprievePod(p)
}
return victims, numViolatingVictim, true
}
Copy
3.3.6. pickOneNodeForPreemption
pickOneNodeForPreemption 要从给定的 nodes 中选择一个 node，这个函数假设给定的 map 中 value 部分是以 priority 降序排列的。这里选择 node 的标准是：
- 最少的 PDB violations
- 最少的高优先级 victim
- 优先级总数字最小
- victim 总数最小
- 直接返回第一个
pkg/scheduler/core/generic_scheduler.go:788
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
// 这个循环要找到 PDBViolatingPods 最少的 node，如果有多个，就全部存在 minNodes1 中
for node, victims := range nodesToVictims {
if len(victims.Pods) == 0 {
// 如果发现一个不需要驱逐 pod 的 node，马上返回
return node
}
numPDBViolatingPods := victims.NumPDBViolations
if numPDBViolatingPods < minNumPDBViolatingPods {
minNumPDBViolatingPods = numPDBViolatingPods
minNodes1 = nil
lenNodes1 = 0
}
if numPDBViolatingPods == minNumPDBViolatingPods {
minNodes1 = append(minNodes1, node)
lenNodes1++
}
}
// 如果只找到1个 PDB violations 最少的 node，那就直接返回这个 node 就 ok 了
if lenNodes1 == 1 {
return minNodes1[0]
}
// 还剩下多个 node，那就寻找 highest priority victim 最小的 node
minHighestPriority := int32(math.MaxInt32)
var minNodes2 = make([]*v1.Node, lenNodes1)
lenNodes2 := 0
// 这个循环要做的事情是看2个 node 上 victims 中最高优先级的 pod 哪个优先级更高
for i := 0; i < lenNodes1; i++ {
node := minNodes1[i]
victims := nodesToVictims[node]
// highestPodPriority is the highest priority among the victims on this node.
highestPodPriority := util.GetPodPriority(victims.Pods[0])
if highestPodPriority < minHighestPriority {
minHighestPriority = highestPodPriority
lenNodes2 = 0
}
if highestPodPriority == minHighestPriority {
minNodes2[lenNodes2] = node
lenNodes2++
}
}
// 发现只有1个，那就直接返回
if lenNodes2 == 1 {
return minNodes2[0]
}
// 这时候还没有抉择出一个 node，那就开始计算优先级总和了，看哪个更低
minSumPriorities := int64(math.MaxInt64)
lenNodes1 = 0
for i := 0; i < lenNodes2; i++ {
var sumPriorities int64
node := minNodes2[i]
for _, pod := range nodesToVictims[node].Pods {
// 这里的累加考虑到了先把优先级搞成正数。不然会出现1个 node 上有1优先级为 -3 的 pod，另外一个 node 上有2个优先级为 -3 的 pod，结果 -3>-6，有2个 pod 的 node 反而被认为总优先级更低！
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]
}
// 还是没有分出胜负，于是开始用 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++
}
}
// 还是没有区分出来1个 node 的话，只能放弃区分了，直接返回第一个结果
if lenNodes2 > 0 {
return minNodes2[0]
}
klog.Errorf("Error in logic of node scoring for preemption. We should never reach here!")
return nil
}
Copy
4. 小结
咋个说呢，此处应该有总结的，抢占过程的逻辑比我想象中的复杂，设计很巧妙，行云流水，大快人心！preemption 可以简单说成再预选->再优选吧；还是不多说了，一天写这么多有点坐不住了，下回再继续聊调度器～
有疑问加站长微信联系（非本文作者）
