Kubernetes 源码【左扬精讲】—— kube-scheduler（调度专题 · 八）：内部架构与核心组件

本文是 kube-scheduler 调度专题的开篇。kube-scheduler 是 Kubernetes 控制平面的"大脑"——它负责将未调度的 Pod 分配到最合适的节点上。这个看似简单的决策过程，背后其实是一套精密的工程系统：Informer 负责感知集群状态，Scheduling Queue 管理调度任务队列，Cache 维护节点与 Pod 的内存快照，Scheduling Framework 则将调度流水线插件化。

读完本篇，你应该能回答：kube-scheduler 有哪些核心组件？它们各自的源码路径和核心结构体是什么？调度流程是如何将这四个组件串联起来的？

Kubernetes Scheduler Informer Scheduling Queue Cache Scheduling Framework k8s v1.36.1

学习重点提示 — 建议先通读全文，再重点回顾标注内容

重点掌握（必须）

• Scheduler struct 四大成员（pkg/scheduler/scheduler.go:68-109）：Cache、SchedulingQueue、Profiles、nodeInfoSnapshot
• PriorityQueue 三个内部队列（pkg/scheduler/backend/queue/scheduling_queue.go:183-187）：activeQ / backoffQ / unschedulablePods
• cacheImpl 双向链表结构（pkg/scheduler/backend/cache/cache.go:53-84）：nodeInfoListItem 实现 MRU 顺序
• ScheduleOne 调度主循环（pkg/scheduler/schedule_one.go:67-96）：schedulingCycle + bindingCycle 两阶段分离
• frameworkImpl 扩展点插件切片（pkg/scheduler/framework/runtime/framework.go:64-81）：11 类插件接口

次重点（了解即可）

• Cache 的 Pod 状态机（pkg/scheduler/backend/cache/cache.go:34-49）：Assumed / Added / Forgotten 三态
• Informers 注册的 9 类资源（pkg/scheduler/eventhandlers.go:481-706）
• Opportunistic Batching 与 Pod Signer（pkg/scheduler/framework/runtime/framework.go:872）

文章目录

一、架构总览：四大组件的协同关系

kube-scheduler 的源码组织在 pkg/scheduler/ 下，核心文件如下：

文件	组件归属	职责
scheduler.go	编排层	Scheduler struct，组装四大组件，主循环入口
schedule_one.go	编排层	ScheduleOne 主循环，一次调度一个 Pod
eventhandlers.go	Informer	注册所有 Informer 的事件回调
backend/queue/scheduling_queue.go	Scheduling Queue	PriorityQueue，三个内部队列管理待调度 Pod
backend/cache/cache.go	Cache	内存缓存，维护节点信息和 Pod 状态
framework/runtime/framework.go	Scheduling Framework	frameworkImpl，执行扩展点插件流水线

四者的关系用一句话概括：Informer 感知变化，Scheduling Queue 管理任务，Cache 提供快照，Framework 执行调度逻辑。下面这张时序图展示了 kube-scheduler 启动时四组件的组装顺序：

Scheduler.New(ctx)
│
├─ 1. internalcache.New()
│     → cacheImpl{}          // 内存缓存，空壳，等待 Informer 填充
│
├─ 2. internalqueue.NewSchedulingQueue()
│     → PriorityQueue{}      // activeQ / backoffQ / unschedulablePods
│
├─ 3. profile.NewMap()
│     → frameworkImpl{}      // 注册插件，初始化扩展点切片
│
├─ 4. addAllEventHandlers()
│     → 注册 9 类资源的 Informer handlers
│     → handler 回调写入 cache / schedulingQueue
│
└─ Scheduler.Run()
      ├─ SchedulingQueue.Run()       → BackoffQ pump goroutine
      ├─ APIDispatcher.Run()         → 异步 API 调用 goroutine
      └─ go wait.Until(ScheduleOne)  → 调度主循环（goroutine）

二、Component 1：Informer —— 事件驱动的状态感知

2.1 整体架构：为什么需要 Informer？

kube-scheduler 需要感知集群中 Pod、Node、PV、PVC、StorageClass 等资源的变化。如果每次调度都直接请求 apiserver，数千节点的集群会导致 apiserver 过载。Informer 模式（基于 client-go 的 SharedInformer）通过"本地缓存 + 增量推送"解决了这个问题：

• 本地缓存（SharedIndexInformer）：Informer 在内存中维护一份完整资源副本，调度器读本地缓存而不每次查 etcd
• 增量推送（DeltaFIFO）：Informer 监听 apiserver 的 watch 事件，增量更新本地缓存，触发 Handler 回调
• SharedInformerFactory：多个组件（scheduler、controller）共享同一个 Informer 实例，减少 apiserver 连接数

2.2 源码位置与核心结构

scheduler 的 Informer 注册逻辑集中在 pkg/scheduler/eventhandlers.go，入口函数 addAllEventHandlers()（行 481-706）。这个函数在 Scheduler.New()（pkg/scheduler/scheduler.go:463）中被调用。

// pkg/scheduler/eventhandlers.go (行 481-520, k8s v1.36.1)
// addAllEventHandlers — 一次性注册所有 Informer 的事件回调
func addAllEventHandlers(
    sched *Scheduler,
    informerFactory informers.SharedInformerFactory,
    dynInformerFactory dynamicinformer.DynamicSharedInformerFactory,
    resourceClaimCache *assumecache.AssumeCache,
    resourceSliceTracker *resourceslicetracker.Tracker,
    draManager fwk.SharedDRAManager,
    gvkMap map[fwk.EventResource]fwk.ActionType,
) error {

    // ---- Pod Informer ----
    podInformer := informerFactory.Core().V1().Pods()
    _, err := podInformer.Informer().AddEventHandler(
        cache.FilteredResourceEventHandler{
            Handler: cache.ResourceEventHandlerFuncs{
                AddFunc:    func(obj interface{}) { sched.addPod(logger, obj) },
                UpdateFunc: func(oldObj, newObj interface{}) { sched.updatePod(logger, oldObj, newObj) },
                DeleteFunc: func(obj interface{}) { sched.deletePod(logger, obj) },
            },
            // 过滤：只关心 "schedulerName=本调度器" 且 "未绑定节点" 的 Pod
            FilterFunc: func(obj interface{}) bool { ... },
        },
    )

    // ---- Node Informer ----
    _, err = informerFactory.Core().V1().Nodes().Informer().AddEventHandler(
        cache.ResourceEventHandlerFuncs{
            AddFunc:    func(obj interface{}) { sched.addNodeToCache(logger, obj) },
            UpdateFunc: func(oldObj, newObj interface{}) { sched.updateNodeInCache(logger, oldObj, newObj) },
            DeleteFunc: func(obj interface{}) { sched.deleteNodeFromCache(logger, obj) },
        },
    )

    // ---- CSI 资源（PV/PVC/StorageClass/VolumeAttachment） ----
    csiInformers := []struct {
        lister  interface{}  // Lister
        addFn   interface{}  // Add handler
        name    string
    }{ ... }
}

2.3 Handler 分类与分发逻辑

所有 Handler 分为两类：写 Cache 和写 Scheduling Queue。addAllEventHandlers() 的设计哲学是——事件回调只做最小化工作，真正复杂的逻辑（如"节点变化后哪些 Pod 需要重新调度"）由队列模块的 MoveAllToActiveOrBackoffQueue() 处理。

Pod Handler（行 128-208）

// pkg/scheduler/eventhandlers.go (行 128-141, k8s v1.36.1)
// addPod — Pod 创建时，区分"已调度"和"未调度"走不同分支
func (sched *Scheduler) addPod(logger klog.Logger, podObj interface{}) {
    pod, err := util.GetPod(podObj)
    if err != nil { return }

    if len(pod.Spec.NodeName) == 0 {
        // 未调度 Pod → 进入 SchedulingQueue
        sched.addPodToSchedulingQueue(logger, pod)
    } else {
        // 已调度 Pod → 进入 Cache（通知调度器该 Pod 已在某节点运行）
        sched.addAssignedPodToCache(logger, pod)
    }
}

// updatePod — Pod 更新时，判断是否需要重新调度
func (sched *Scheduler) updatePod(logger klog.Logger, oldObj, newObj interface{}) {
    oldPod, _ := util.GetPod(oldObj)
    newPod, _ := util.GetPod(newObj)
    if oldPod == nil || newPod == nil { return }

    if len(newPod.Spec.NodeName) == 0 && len(oldPod.Spec.NodeName) == 0 {
        // 未调度 → 未调度：更新 SchedulingQueue
        sched.updatePodInSchedulingQueue(logger, oldPod, newPod)
    } else if len(newPod.Spec.NodeName) > 0 && len(oldPod.Spec.NodeName) == 0 {
        // 未调度 → 已调度：移出 SchedulingQueue，加入 Cache
        sched.addAssignedPodToCache(logger, newPod)
        sched.deletePodFromSchedulingQueue(logger, oldPod)
    } else if len(oldPod.Spec.NodeName) > 0 && len(newPod.Spec.NodeName) == 0 {
        // 已调度 → 未调度（解绑）：移出 Cache，进入 SchedulingQueue
        sched.addPodToSchedulingQueue(logger, newPod)
        sched.deleteAssignedPodFromCache(logger, oldPod)
    }
    // 已调度 → 已调度：仅更新 Cache 中的 Pod 信息
}

Node Handler（行 53-126）

// pkg/scheduler/eventhandlers.go (行 53-66, k8s v1.36.1)
// addNodeToCache — 节点新增时，更新 Cache 并触发待调度 Pod 重试
func (sched *Scheduler) addNodeToCache(logger klog.Logger, obj interface{}) {
    node, err := util.GetNode(obj)
    if err != nil { return }

    nodeInfo := sched.Cache.AddNode(logger, node)
    if nodeInfo != nil {
        // 节点新增后，可能有之前因该节点信息缺失而调度失败的 Pod
        sched.SchedulingQueue.MoveAllToActiveOrBackoffQueue(logger,
            fwk.ClusterEvent{Resource: fwk.Node, ActionType: fwk.Added},
            nil, nil, nil,
        )
    }
}

// updateNodeInCache — 节点变化时，判断是否需要重排待调度 Pod
func (sched *Scheduler) updateNodeInCache(logger klog.Logger, oldObj, newObj interface{}) {
    node, _ := util.GetNode(newObj)
    oldNode, _ := util.GetNode(oldObj)
    if node == nil { return }

    oldNodeInfo := sched.Cache.UpdateNode(logger, oldNode, node)
    // 若节点变为 Unschedulable，触发相关 Pod 重新入队
    if oldNodeInfo != nil && oldNodeInfo.Node().Spec.Unschedulable != node.Spec.Unschedulable {
        sched.SchedulingQueue.MoveAllToActiveOrBackoffQueue(logger, ...)
    }
}

2.4 注册的 9 类资源

addAllEventHandlers() 共注册了 9 类资源的 Informer Handler（行 498-666），以下是完整列表：

资源类型	Informer	Handler 回调	写入目标
Pod	CoreV1().Pods()	addPod / updatePod / deletePod	Queue 或 Cache
Node	CoreV1().Nodes()	addNodeToCache / updateNodeInCache / deleteNodeFromCache	Cache + Queue
PersistentVolume	CoreV1().PersistentVolumes()	onPvAdd / onPvUpdate / onPvDelete	VolumeBinding Cache
PersistentVolumeClaim	CoreV1().PersistentVolumeClaims()	onPvcAdd / onPvcUpdate / onPvcDelete	VolumeBinding Cache
StorageClass	StorageV1().StorageClasses()	onStorageClassAdd	VolumeBinding Cache
CSINode	StorageV1().CSINodes()	onCSINodeAdd	Cache
CSIStorageCapacity	StorageV1().CSIStorageCapacities()	onCSIStorageCapacityAdd	VolumeBinding Cache
VolumeAttachment	StorageV1().VolumeAttachments()	onVaAdd	VolumeBinding Cache
PodGroup（可选）	schedulingv1alpha1().PodGroups()	onPodGroupAdd	SchedulingQueue

三、Component 2：Scheduling Queue —— 调度任务的缓冲区

3.1 为什么需要调度队列？

Pod 调度不是"来一个调度一个"那么简单。调度失败时，Pod 需要等待条件变化后重试；节点变化时，未调度的 Pod 需要被触发重新评估。Scheduling Queue 就是管理这些"待调度 Pod"的缓冲区——它不是简单的 FIFO 队列，而是一个包含优先级排序、退避重试、条件触发的智能调度缓冲系统。

3.2 核心结构：PriorityQueue

调度队列在 v1.36.1 中的实现是 PriorityQueue（pkg/scheduler/backend/queue/scheduling_queue.go:172），对外暴露 SchedulingQueue 接口（行 96-153）。它内部维护三个子队列：

// pkg/scheduler/backend/queue/scheduling_queue.go (行 60-84, k8s v1.36.1)
// 常量定义
const (
    DefaultPodMaxInUnschedulablePodsDuration time.Duration = 5 * time.Minute  // Pod 在不可调度队列中最多待 5 分钟
    DefaultPodInitialBackoffDuration time.Duration = 1 * time.Second             // 初始退避 1s
    DefaultPodMaxBackoffDuration time.Duration = 10 * time.Second               // 最大退避 10s
})

// pkg/scheduler/backend/queue/scheduling_queue.go (行 172-210, k8s v1.36.1)
// PriorityQueue — 调度队列的核心实现
type PriorityQueue struct {
    *nominator                  // 记录 Pod 的 nominatedNodeName（等待抢占的节点）
    stop  chan struct{}
    clock clock.WithTicker
    lock sync.RWMutex

    // 三个内部队列
    activeQ           activeQueuer    // 堆结构，待调度的 Pod 按优先级排序
    backoffQ          backoffQueuer  // 堆结构，调度失败后进入退避等待
    unschedulablePods *unschedulablePods  // map 结构，调度失败的 Pod

    moveRequestCycle int64
    // PreEnqueue 插件 map：哪个 Pod 满足哪个插件的前置条件
    preEnqueuePluginMap map[string]map[string]fwk.PreEnqueuePlugin
    // QueueingHintMap：插件注册的事件 → Pod 重排策略
    queueingHintMap    QueueingHintMapPerProfile
    // Pod 签名器：Opportunistic Batching 优化
    podSigners map[string]PodSigner
    // ...
}

3.3 三个子队列的行为

队列	底层数据结构	进入条件	离开条件	关键方法
activeQ	堆（heap）	Pod 新建、Backoff 结束、节点变化触发	Pop() 被调用	Add() / Pop()
backoffQ	堆（heap）+ 定时器	调度失败后进入，指数退避	退避计时器到期 → activeQ	moveToBackoffQ()
unschedulablePods	map + 堆（超时清理）	调度失败（不可重试/永久性原因）	5 分钟后超时 → backoffQ / 节点变化触发	AddUnschedulableIfNotPresent()

3.4 Pod 在三个队列之间的流转

// 场景 1: Pod 新建 → activeQ
// pkg/scheduler/backend/queue/scheduling_queue.go (行 739-750)
func (p *PriorityQueue) Add(ctx context.Context, pod *v1.Pod) {
    p.lock.Lock()
    defer p.lock.Unlock()
    if err := p.activeQ.Add(p.podInfo(pod)); err != nil {
        logger := p.logger
        logger.Error(err, "Error adding pod to the active queue", "pod", klog.KObj(pod))
    }
}

// 场景 2: 调度失败 → backoffQ 或 unschedulablePods
// pkg/scheduler/backend/queue/scheduling_queue.go (行 901-965)
func (p *PriorityQueue) AddUnschedulableIfNotPresent(
    logger klog.Logger, podInfo *framework.QueuedPodInfo, podSchedulingCycle int64,
) error {
    // 判断是否值得等待（可重试 vs 永久性失败）
    if p.isPodWorthRequeuing(podInfo, frwk.Unschedulable) {
        // 永久性失败 → unschedulablePods（5 分钟后超时）
        p.unschedulablePods.Add(podInfo.Pod, p.moveRequestCycle)
    } else {
        // 可重试 → backoffQ（指数退避）
        p.podInfo(podInfo.Pod).UnschedulableTime = time.Now()
        if err := p.backoffQ.Add(p.podInfo(podInfo.Pod)); err != nil {
            logger.Error(err, "Error adding pod to the backoff queue", "pod", klog.KObj(podInfo.Pod))
        }
    }
}

// 场景 3: 节点变化 → 相关 Pod 从 unschedulablePods 移出
// pkg/scheduler/backend/queue/scheduling_queue.go (行 1269-1277)
func (p *PriorityQueue) MoveAllToActiveOrBackoffQueue(...) {
    // 通过 QueueingHintMap 判断哪些 Pod 需要重新入队
    for _, pInfo := range p.unschedulablePods.Pods() {
        if p.isPodWorthRequeuing(...) {
            p.moveToBackoffQ(pInfo)
        }
    }
}

四、Component 3：Cache —— 节点与 Pod 的内存快照

4.1 Cache 的定位

Cache 是 kube-scheduler 的"数据库"——它存储了调度所需的所有节点和 Pod 的本地内存副本。与 apiserver 中的 etcd 不同，Cache 是纯内存结构，不做持久化；与 Informer 的本地 Indexer 不同，Cache 存储的是经过调度语义加工的数据（如"节点上还剩多少 CPU"）。

4.2 核心接口与结构体

// pkg/scheduler/backend/cache/interface.go (行 56-127, k8s v1.36.1)
// Cache 接口 — 调度器只依赖这个抽象，不关心具体实现
type Cache interface {
    NodeCount() int
    PodCount() (int, error)
    GetNode(name string) (*framework.NodeInfo, error)

    // Pod 生命周期管理（Assumed = 乐观假设，Added = 确认绑定）
    AssumePod(logger klog.Logger, pod *v1.Pod) error   // 调度成功后"假设"Pod 已在节点
    ForgetPod(logger klog.Logger, pod *v1.Pod) error   // 假设失败，取消假设
    AddPod(logger klog.Logger, pod *v1.Pod) error      // 确认 Pod 已绑定
    UpdatePod(logger klog.Logger, oldPod, newPod *v1.Pod) error
    RemovePod(logger klog.Logger, pod *v1.Pod) error
    IsAssumedPod(pod *v1.Pod) (bool, error)            // 检查是否在 Assumed 状态

    // Node 管理
    AddNode(logger klog.Logger, node *v1.Node) *framework.NodeInfo
    UpdateNode(logger klog.Logger, oldNode, newNode *v1.Node) *framework.NodeInfo
    RemoveNode(logger klog.Logger, node *v1.Node) error

    // 快照：每个调度周期从 Cache 拿一份快照
    UpdateSnapshot(logger klog.Logger, nodeSnapshot *Snapshot) error

    // 调试与 PodGroup
    Dump() *Dump
    BindPod(binding *v1.Binding) (

// pkg/scheduler/backend/cache/cache.go (行 59-88, k8s v1.36.1)
// cacheImpl — Cache 的生产实现
type cacheImpl struct {
    stop   

4.3 Pod 的三态状态机

Cache 中的 Pod 遵循一个三态状态机（pkg/scheduler/backend/cache/cache.go:34-49）：

// pkg/scheduler/backend/cache/cache.go (行 34-49, k8s v1.36.1)
// Pod 在 Cache 中的三种状态及其转换
//
//   Initial  ──────►  Assumed  ──────►  Added
//   (未在 Cache)         (已假设)          (已确认)
//                          │                 │
//                          │                 │
//                          ▼                 ▼
//                      Forgotten        Deleted
//                    (假设被取消)      (从 Cache 移除)
//
// Assumed 状态是"乐观预占"：调度周期开始时，
// 在真正 Bind 之前，把 Pod 的资源算进节点的已分配量。
// 这样可以避免两个 Pod 同时调度到同一节点（虽然最终 Bind 只会有一个成功）。

// AssumePod — 调度成功后，把 Pod "假设"到目标节点
// pkg/scheduler/backend/cache/cache.go (行 397-410)
func (cache *cacheImpl) AssumePod(logger klog.Logger, pod *v1.Pod) error {
    key, err := cache.keyFunc(pod)
    if err != nil { return err }

    cache.mu.Lock()
    defer cache.mu.Unlock()

    if _, ok := cache.podStates[key]; ok {
        return fmt.Errorf("pod %v is not assumed and thus cannot be assumed again", key)
    }

    // 记录到 assumedPods set
    cache.assumedPods.Insert(key)
    // 记录 podState（含 pod 指针）
    cache.podStates[key] = &podState{pod: pod}

    // 把 Pod 的资源"算进"节点的已分配量
    if err := cache.addPod(pod); err != nil {
        cache.assumedPods.Delete(key)
        delete(cache.podStates, key)
        return err
    }
    return nil
}

// ForgetPod — 假设失败或被取消，把 Pod 从 Cache 移除
// pkg/scheduler/backend/cache/cache.go (行 412-434)
func (cache *cacheImpl) ForgetPod(logger klog.Logger, pod *v1.Pod) error {
    key, err := cache.keyFunc(pod)
    if err != nil { return err }

    cache.mu.Lock()
    defer cache.mu.Unlock()

    if !cache.assumedPods.Has(key) {
        return fmt.Errorf("pod %v is not assumed, cannot be forgotten", key)
    }

    // 从 assumedPods 移除，并从节点的已分配量中减去
    if err := cache.removePod(pod); err != nil {
        return err
    }
    cache.assumedPods.Delete(key)
    delete(cache.podStates, key)
    return nil
}

// AddPod — Bind 确认后，把 Pod 加入 Cache（或从 assumed 升级为 added）
// pkg/scheduler/backend/cache/cache.go (行 515-545)
func (cache *cacheImpl) AddPod(logger klog.Logger, pod *v1.Pod) error {
    key, err := cache.keyFunc(pod)
    if err != nil { return err }

    cache.mu.Lock()
    defer cache.mu.Unlock()

    if cache.assumedPods.Has(key) {
        // 从 Assumed → Added：Pod 已绑定，无需额外操作
        cache.assumedPods.Delete(key)
    } else {
        // 从外部进入（已在某节点运行）：需要加入 Cache
        if err := cache.addPod(pod); err != nil { return err }
    }

    cache.podStates[key] = &podState{pod: pod}
    return nil
}

4.4 Snapshot 机制

每个调度周期开始时，调度器从 Cache 提取一份只读快照（nodeInfoSnapshot），Filter 和 Score 都在这个快照上做计算：

// pkg/scheduler/backend/cache/cache.go (行 184-296, k8s v1.36.1)
// UpdateSnapshot — 从 Cache 提取节点列表，构建只读快照
func (cache *cacheImpl) UpdateSnapshot(logger klog.Logger, nodeSnapshot *internalcache.Snapshot) error {
    cache.mu.RLock()
    defer cache.mu.RUnlock()

    // 按双向链表的 MRU 顺序遍历节点
    nodes := make([]*framework.NodeInfo, 0, len(cache.nodes))
    for item := cache.headNode; item != nil; item = item.next {
        nodes = append(nodes, item.info)
    }

    // 反转（因为链表是从 headNode → tailNode，但调度希望从尾部开始）
    for i, j := 0, len(nodes)-1; i < j; i, j = i+1, j-1 {
        nodes[i], nodes[j] = nodes[j], nodes[i]
    }

    // 写入快照
    nodeSnapshot.NodeInfoMap = make(map[string]*framework.NodeInfo, len(nodes))
    for _, nodeInfo := range nodes {
        nodeSnapshot.NodeInfoMap[nodeInfo.Node().Name] = nodeInfo
    }
    nodeSnapshot.Nodes = nodes
    return nil
}

4.5 双向链表的 MRU 设计

Cache 用双向链表管理节点（nodeInfoListItem，pkg/scheduler/backend/cache/cache.go:53-60），目的是维护MRU（Most Recently Used）顺序。节点被访问（Filter 阶段遍历）时会被移到链表头部，下次遍历优先使用——这是一种近似轮询（approximate round-robin）的负载均衡策略。

// pkg/scheduler/backend/cache/cache.go (行 114-163, k8s v1.36.1)
// moveNodeInfoToHead — 节点被访问后移到链表头部（MRU）
func (cache *cacheImpl) moveNodeInfoToHead(logger klog.Logger, n *nodeInfoListItem) {
    if n == nil { return }
    if n == cache.headNode { return }  // 已在头部，无需移动

    // 从原位置摘下
    n.prev.next = n.next
    if n.next != nil { n.next.prev = n.prev }

    // 插入头部
    n.prev = nil
    n.next = cache.headNode
    cache.headNode.prev = n
    cache.headNode = n
}

// removeNodeInfoFromList — 节点删除时从链表摘下
func (cache *cacheImpl) removeNodeInfoFromList(n *nodeInfoListItem) {
    if n == nil { return }
    if n.prev != nil { n.prev.next = n.next }
    if n.next != nil { n.next.prev = n.prev }
    if n == cache.headNode { cache.headNode = n.next }
}

五、Component 4：Scheduling Framework —— 插件化的调度流水线

5.1 为什么需要插件化？

Scheduler 面对的调度规则五花八门：节点资源、节点亲和、Pod 间亲和、污点容忍、拓扑打散、卷绑定……每一条都是一个独立的业务规则。如果全量硬编码进调度主循环，schedule_one.go 会膨胀到几万行；新增一条规则就要改核心代码。Scheduling Framework 就是把调度规则全部抽象成扩展点接口，每条规则单独实现成插件。

5.2 扩展点全景

Scheduling Framework 定义了 13 个扩展点（对应 13 个 Go 接口），按执行顺序排列：

扩展点	Go 接口	阶段	作用
PreEnqueue	PreEnqueuePlugin	入队前	判断 Pod 是否满足前置条件（如 SchedulingGates 已全部移除）
QueueSort	QueueSortPlugin	队列排序	决定 activeQ 中 Pod 的优先级（由 PrioritySort 实现）
PreFilter	PreFilterPlugin	过滤前预处理	预处理（如算好资源请求），结果写入 CycleState
Filter	FilterPlugin	过滤	逐个 Node 检查硬约束（资源、亲和、污点等）
PostFilter	PostFilterPlugin	过滤后	过滤失败时执行（典型：DefaultPreemption 抢占）
PreScore	PreScorePlugin	打分前预处理	预计算 Pod 特征向量，供 Score 阶段使用
Score	ScorePlugin	打分	为每个 Feasible Node 打 0~100 分
Reserve	ReservePlugin	预约	Bind 前"占座"（典型：VolumeBinding 占用 PV）
Permit	PermitPlugin	批准	异步等待外部信号（GangScheduling 等齐兄弟 Pod）
PreBind	PreBindPlugin	绑定前	真正创建 PV 绑定等预处理
Bind	BindPlugin	绑定	向 apiserver 提交 Binding 对象（由 DefaultBinder 完成）
PostBind	PostBindPlugin	绑定后	Bind 成功后清理（释放资源）
Sign	SignPlugin	签名	Opportunistic Batching：给 Pod 生成签名，复用调度结果

5.3 框架运行时：frameworkImpl

// pkg/scheduler/framework/runtime/framework.go (行 58-112, k8s v1.36.1)
// frameworkImpl — 框架运行时的核心结构体
type frameworkImpl struct {
    // ---- 13 类插件切片 ----
    registry                  runtime.Registry
    snapshotSharedLister      fwk.SharedLister  // NodeInfo 快照（从 Cache.UpdateSnapshot 传入）

    // 队列相关
    waitingPods               *waitingPodsMap     // 等待 Permit 批准的 Pod
    podsInPreBind             *podsInPreBindMap  // 正在 PreBind 的 Pod
    queueSortPlugins         []fwk.QueueSortPlugin

    // 调度流水线
    preEnqueuePlugins        []fwk.PreEnqueuePlugin
    enqueueExtensions        []fwk.EnqueueExtensions
    preFilterPlugins         []fwk.PreFilterPlugin
    filterPlugins            []fwk.FilterPlugin
    postFilterPlugins        []fwk.PostFilterPlugin
    preScorePlugins          []fwk.PreScorePlugin
    scorePlugins             []fwk.ScorePlugin
    reservePlugins           []fwk.ReservePlugin
    preBindPlugins           []fwk.PreBindPlugin
    bindPlugins              []fwk.BindPlugin
    postBindPlugins          []fwk.PostBindPlugin
    permitPlugins            []fwk.PermitPlugin

    // 插件权重（Score 时加权）
    scorePluginWeight         map[string]int

    // 签名与批处理
    batchablePlugins          []fwk.SignPlugin
    batch                    *OpportunisticBatch

    // DRA / CSI / PodGroup
    sharedDRAManager         fwk.SharedDRAManager
    sharedCSIManager         fwk.CSIManager
    podGroupManager          fwk.PodGroupManager

    // 并行化器
    parallelizer             fwk.Parallelizer

    // 辅助工具
    clientSet                clientset.Interface
    informerFactory          informers.SharedInformerFactory
    apiDispatcher            *apidispatcher.APIDispatcher
    logger                   klog.Logger
    metricsRecorder          *metrics.MetricAsyncRecorder
    profileName              string
}

5.4 插件注册中心：NewInTreeRegistry

// pkg/scheduler/framework/plugins/registry.go (行 50-79, k8s v1.36.1)
// NewInTreeRegistry — 23 个内置插件的注册中心
func NewInTreeRegistry() runtime.Registry {
    fts := plfeature.NewSchedulerFeaturesFromGates(feature.DefaultFeatureGate)
    registry := runtime.Registry{
        // ---- Filter/PreFilter 插件 ----
        noderesources.Name:               runtime.FactoryAdapter(fts, noderesources.NewFit),
        tainttoleration.Name:            runtime.FactoryAdapter(fts, tainttoleration.New),
        nodeaffinity.Name:               runtime.FactoryAdapter(fts, nodeaffinity.New),
        nodename.Name:                   runtime.FactoryAdapter(fts, nodename.New),
        nodeports.Name:                  runtime.FactoryAdapter(fts, nodeports.New),
        interpodaffinity.Name:           runtime.FactoryAdapter(fts, interpodaffinity.New),
        podtopologyspread.Name:          runtime.FactoryAdapter(fts, podtopologyspread.New),
        nodeunschedulable.Name:          runtime.FactoryAdapter(fts, nodeunschedulable.New),
        nodedeclaredfeatures.Name:       runtime.FactoryAdapter(fts, nodedeclaredfeatures.New),
        nodevolumelimits.CSIName:        runtime.FactoryAdapter(fts, nodevolumelimits.NewCSI),

        // ---- 资源打分插件 ----
        noderesources.BalancedAllocationName: runtime.FactoryAdapter(fts, noderesources.NewBalancedAllocation),
        imagelocality.Name:                   runtime.FactoryAdapter(fts, imagelocality.New),

        // ---- 卷相关插件 ----
        volumebinding.Name:               runtime.FactoryAdapter(fts, volumebinding.New),
        volumerestrictions.Name:         runtime.FactoryAdapter(fts, volumerestrictions.New),
        volumezone.Name:                 runtime.FactoryAdapter(fts, volumezone.New),

        // ---- 抢占 & 队列 ----
        defaultpreemption.Name:          runtime.FactoryAdapter(fts, defaultpreemption.New),
        queuesort.Name:                 queuesort.New,           // 无需 FeatureGate，直接注册

        // ---- Bind & PreBind ----
        defaultbinder.Name:              defaultbinder.New,       // 无需 FeatureGate，直接注册

        // ---- 高级特性 ----
        dynamicresources.Name:           runtime.FactoryAdapter(fts, dynamicresources.New),
        gangscheduling.Name:             runtime.FactoryAdapter(fts, gangscheduling.New),
        schedulinggates.Name:            runtime.FactoryAdapter(fts, schedulinggates.New),
        topologyaware.Name:              runtime.FactoryAdapter(fts, topologyaware.New),
        podgrouppodscount.Name:          runtime.FactoryAdapter(fts, podgrouppodscount.New),
    }
    return registry
}

六、调度主循环：四组件如何串联

6.1 ScheduleOne 的两阶段设计

ScheduleOne()（pkg/scheduler/schedule_one.go:67）是 kube-scheduler 的主循环入口，每次处理一个 Pod。它分为同步的 schedulingCycle 和异步的 bindingCycle 两阶段：

// pkg/scheduler/schedule_one.go (行 67-148, k8s v1.36.1)
// ScheduleOne — 一次调度一个 Pod 的主循环
func (sched *Scheduler) ScheduleOne(ctx context.Context) {
    // ① 从 SchedulingQueue 拿出下一个待调度的 Pod（阻塞）
    podInfo, err := sched.NextPod(logger)
    if err != nil { return }
    if podInfo == nil || podInfo.Pod == nil { return }

    if sched.genericWorkloadEnabled && podInfo.Pod.Spec.SchedulingGroup != nil {
        // PodGroup 调度：批量调度一族相关的 Pod
        podGroupInfo, err := sched.podGroupInfoForPod(ctx, podInfo)
        sched.scheduleOnePodGroup(ctx, podGroupInfo)
    } else {
        // 普通 Pod 调度
        sched.scheduleOnePod(ctx, podInfo)
    }
}

// pkg/scheduler/schedule_one.go (行 99-148, k8s v1.36.1)
// scheduleOnePod — 单个 Pod 的调度流程
func (sched *Scheduler) scheduleOnePod(ctx context.Context, podInfo *fwk.QueuedPodInfo) {
    // 获取当前调度周期的 framework（根据 Pod 的 schedulerName 选择 Profile）
    fwk, state := sched.getFrameworkAndState(ctx, podInfo)
    if fwk == nil { return }

    // 同步阶段：schedulingCycle
    scheduleResult, facStatus := sched.schedulingCycle(ctx, fwk, state, podInfo, logger)

    if facStatus != nil {
        // 调度失败：处理抢占或回到队列
        sched.handleSchedulingFailure(ctx, fwk, podInfo, facStatus, ...)

        // 如果触发了抢占，抢占成功后会调用 nominatedNodeName，
        // 调度器会等待该节点的 Pod 被驱逐后重试
        return
    }

    // 异步阶段：bindingCycle（在 goroutine 中执行）
    go func() {
        defer func() {
            // bindingCycle 结束后通知队列
            if !podInfo.DeferAfter(fwk) {
                sched.SchedulingQueue.Done(podInfo)
            }
        }()
        sched.bindingCycle(ctx, fwk, state, podInfo, scheduleResult, logger)
    }()
}

6.2 schedulingCycle：同步调度阶段

// pkg/scheduler/schedule_one.go (行 175-198, k8s v1.36.1)
// schedulingCycle — 同步调度阶段
func (sched *Scheduler) schedulingCycle(
    ctx context.Context, fwk fwk.Framework, state *fwk.CycleState,
    podInfo *fwk.QueuedPodInfo, logger klog.Logger,
) (ScheduleResult, *fwk.Status) {

    // ① 从 Cache 提取节点快照（所有 Filter/Score 在这个快照上运行）
    if err := sched.Cache.UpdateSnapshot(logger, sched.nodeInfoSnapshot); err != nil {
        return ScheduleResult{}, fwk.AsStatus(err)
    }

    // ② 调用 Framework 执行调度算法
    scheduleResult, schedStatus := sched.SchedulePod(ctx, fwk, state, podInfo)
    if schedStatus != nil {
        return ScheduleResult{}, schedStatus
    }

    // ③ 假设 Pod 已调度 + 执行 Reserve 插件（"占座"）
    assumeFinish, err := sched.assumeAndReserve(ctx, fwk, state, podInfo, scheduleResult.SuggestedHost)
    if err != nil {
        return ScheduleResult{}, fwk.AsStatus(err)
    }

    // ④ 执行 Permit 插件（等待批准，可能返回 Wait）
    permitStatus := sched.prepareForBindingCycle(ctx, fwk, state, podInfo)
    return scheduleResult, permitStatus
}

// pkg/scheduler/schedule_one.go (行 570-718, k8s v1.36.1)
// SchedulePod — 找节点 + 打分，核心调度算法
func (sched *Scheduler) SchedulePod(
    ctx context.Context, fwk fwk.Framework, state fwk.CycleState,
    podInfo *fwk.QueuedPodInfo,
) (ScheduleResult, *fwk.Status) {

    // ① PreFilter + Filter：找出所有满足硬约束的节点
    feasibleNodes, filteredNodesStatuses, fitErr := sched.findNodesThatFitPod(ctx, fwk, state, pod)
    if fitErr != nil {
        return ScheduleResult{}, fwk.AsStatus(fitErr)
    }

    // ② 如果没有可行节点 → 运行 PostFilter（DefaultPreemption 抢占）
    if len(feasibleNodes) == 0 {
        status := fwk.RunPostFilterPlugins(ctx, state, pod, filteredNodesStatuses)
        if !status.IsSuccess() && status.Requeue() {
            // 抢占成功，等待 nominatedNode
        }
        return ScheduleResult{}, status
    }

    // ③ PreScore + Score：为所有可行节点打分
    priorityList, scoreErr := sched.prioritizeNodes(ctx, fwk, state, pod, feasibleNodes)
    if scoreErr != nil {
        return ScheduleResult{}, fwk.AsStatus(scoreErr)
    }

    // ④ 选择得分最高的节点
    host, err := selectHost(logger, priorityList)
    return ScheduleResult{
        SuggestedHost: host,
        EvaluatedNodes: len(feasibleNodes),
        FeasibleNodes: len(feasibleNodes),
    }, nil
}

// pkg/scheduler/schedule_one.go (行 628-718)
// findNodesThatFitPod — Filter 阶段的实现
func (sched *Scheduler) findNodesThatFitPod(
    ctx context.Context, fwk fwk.Framework, state fwk.CycleState, pod *v1.Pod,
) ([]*v1.Node, *fwk.NodeToStatus, error) {

    // ① PreFilter 插件（预处理）
    _, s := fwk.RunPreFilterPlugins(ctx, state, pod)
    if !s.IsSuccess() { return nil, nil, s.AsError() }

    // ② Filter 插件（并行遍历所有节点）
    nodes, statuses := fwk.RunFilterPluginsWithNominatedPods(ctx, state, pod, sched.nodeInfoSnapshot)
    return nodes, statuses, nil
}

// pkg/scheduler/schedule_one.go (行 938-1054)
// prioritizeNodes — Score 阶段的实现
func (sched *Scheduler) prioritizeNodes(
    ctx context.Context, fwk fwk.Framework, state fwk.CycleState,
    pod *v1.Pod, feasibleNodes []*v1.Node,
) (framework.NodeScoreList, error) {

    // ① PreScore 插件（预处理）
    fwk.RunPreScorePlugins(ctx, state, pod)

    // ② Score 插件（并行打分）
    scores, s := fwk.RunScorePlugins(ctx, state, pod)
    if !s.IsSuccess() { return nil, s.AsError() }

    // ③ Extender 扩展（如果有外部调度器）
    if len(sched.Extenders) > 0 {
        // 调用外部 Extender
    }

    // ④ 加权求和，返回排序后的节点列表
    result := fwk.ScoreNormalizedNodes(ctx, pod, scores)
    return result, nil
}

6.3 bindingCycle：异步绑定阶段

// pkg/scheduler/schedule_one.go (行 397-503, k8s v1.36.1)
// bindingCycle — 异步绑定阶段（在 goroutine 中运行）
func (sched *Scheduler) bindingCycle(
    ctx context.Context, fwk fwk.Framework, state *fwk.CycleState,
    podInfo *fwk.QueuedPodInfo, scheduleResult ScheduleResult, logger klog.Logger,
) {
    // ① PreBind PreFlight：检查前提条件（如 Pod 是否还在）
    if !fwk.RunPreBindPreFlights(ctx, podInfo.Pod) {
        sched.handleBindingCycleError(ctx, fwk, podInfo, scheduleResult, ...)
        return
    }

    // ② 等待 Permit 批准（如果 Permit 返回了 Wait）
    waitStatus := fwk.WaitOnPermit(ctx, podInfo.Pod)
    if !waitStatus.IsSuccess() {
        sched.handleBindingCycleError(ctx, fwk, podInfo, scheduleResult, ...)
        return
    }

    // ③ PreBind 插件（真正创建 PV 绑定等）
    if !fwk.RunPreBindPlugins(ctx, state, podInfo.Pod, scheduleResult.SuggestedHost).IsSuccess() {
        sched.handleBindingCycleError(ctx, fwk, podInfo, scheduleResult, ...)
        return
    }

    // ④ Bind 插件（向 apiserver 提交 Binding 对象）
    bindStatus := fwk.RunBindPlugins(ctx, state, podInfo.Pod, scheduleResult.SuggestedHost)
    if !bindStatus.IsSuccess() {
        sched.handleBindingCycleError(ctx, fwk, podInfo, scheduleResult, ...)
        return
    }

    // ⑤ PostBind 插件（清理）
    fwk.RunPostBindPlugins(ctx, podInfo.Pod, scheduleResult.SuggestedHost, scheduleResult.EvaluatedNodes)

    // ⑥ 通知 SchedulingQueue：该 Pod 调度完成
    sched.SchedulingQueue.Done(podInfo)
}

6.4 四组件的完整时序图

┌──────────────────────────────────────────────────────────────────────────────────┐
│                           ScheduleOne(ctx)                                        │
│                     pkg/scheduler/schedule_one.go:67                               │
│                                                                                   │
│  ┌──────────────────────────────────────────────────────────────────────────┐   │
│  │  阶段一：schedulingCycle（同步）                                           │   │
│  │                                                                              │   │
│  │  ① Cache.UpdateSnapshot()                                                  │   │
│  │     └─ 从 Cache 的双向链表提取 NodeInfo 快照                                │   │
│  │                                                                           │   │
│  │  ② Framework.RunPreFilterPlugins()                                         │   │
│  │     └─ PreFilter：预处理（NodeResourcesFit 算资源）                         │   │
│  │                                                                           │   │
│  │  ③ Framework.RunFilterPluginsWithNominatedPods()                            │   │
│  │     └─ Filter：并行遍历所有节点，Filter 插件链过滤                          │   │
│  │        └─ NodeResourcesFit / NodeAffinity / TaintToleration / ...         │   │
│  │                                                                           │   │
│  │  ④ [无可行节点] → RunPostFilterPlugins()                                   │   │
│  │     └─ DefaultPreemption：尝试抢占低优先级 Pod                              │   │
│  │                                                                           │   │
│  │  ⑤ Framework.RunScorePlugins()                                            │   │
│  │     └─ Score：并行对可行节点打分                                            │   │
│  │        └─ NodeResourcesFit / InterPodAffinity / PodTopologySpread / ...   │   │
│  │                                                                           │   │
│  │  ⑥ assumeAndReserve()                                                      │   │
│  │     ├─ Cache.AssumePod()     → 把 Pod "假设" 到目标节点                    │   │
│  │     └─ RunReservePluginsReserve() → Reserve：VolumeBinding 占 PV           │   │
│  │                                                                           │   │
│  │  ⑦ RunPermitPlugins()                                                      │   │
│  │     └─ Permit：GangScheduling 等齐（可能返回 Wait）                        │   │
│  └──────────────────────────────────────────────────────────────────────────┘   │
│                                    │                                              │
│                                    │ permitStatus（Wait 或 Success）              │
│                                    ▼                                              │
│  ┌──────────────────────────────────────────────────────────────────────────┐   │
│  │  阶段二：bindingCycle（异步 goroutine）                                     │   │
│  │                                                                              │   │
│  │  ① RunPreBindPreFlights()                                                  │   │
│  │     └─ 检查 Pod 是否仍然存在、未被删除                                      │   │
│  │                                                                           │   │
│  │  ② WaitOnPermit()                                                         │   │
│  │     └─ 等待 Permit 批准（GangScheduling 等齐）                             │   │
│  │                                                                           │   │
│  │  ③ RunPreBindPlugins()                                                     │   │
│  │     └─ PreBind：VolumeBinding 真正创建 PV 绑定                             │   │
│  │                                                                           │   │
│  │  ④ RunBindPlugins()                                                        │   │
│  │     └─ Bind：DefaultBinder 提交 Binding 到 apiserver                       │   │
│  │        └─ apiserver ← Binding 对象                                         │   │
│  │                                                                           │   │
│  │  ⑤ RunPostBindPlugins()                                                   │   │
│  │     └─ PostBind：清理                                                      │   │
│  │                                                                           │   │
│  │  ⑥ SchedulingQueue.Done()                                                │   │
│  │     └─ 从 activeQ 标记该 Pod 已完成                                        │   │
│  └──────────────────────────────────────────────────────────────────────────┘   │
│                                                                                  │
│  ════════════════════════════════════════════════════════════════════════════════ │
│                                                                                  │
│  ┌──────────────────────────────────────────────────────────────────────────┐   │
│  │  Informer 事件回调（与调度主循环并发运行）                                  │   │
│  │                                                                              │   │
│  │  Node Updated → Cache.UpdateNode() + MoveAllToActiveOrBackoffQueue()       │   │
│  │     └─ 触发 unschedulableQ 中的相关 Pod 重新入队                           │   │
│  │                                                                              │   │
│  │  Pod Deleted (scheduled) → Cache.RemovePod()                               │   │
│  │     └─ 释放该节点资源                                                      │   │
│  │                                                                              │   │
│  │  Pod Created → addPodToSchedulingQueue() → activeQ.Add()                  │   │
│  │     └─ Pod 进入队列，等待 NextPod() 被调用                                  │   │
│  └──────────────────────────────────────────────────────────────────────────┘   │
└──────────────────────────────────────────────────────────────────────────────────┘

七、对比与关联：四组件的数据流

7.1 数据流总图

┌──────────────┐    事件回调    ┌──────────────────┐    AssumePod     ┌───────────┐
│   Informer   │──────────────►│  SchedulingQueue │                │   Cache    │
│  (apiserver) │               │  (activeQ 等)   │◄─────────────│  (NodeInfo │
│              │◄──────────────│                 │   AddPod       │   快照)   │
│  Watch 事件  │  NextPod()   │                 │─────────────►│            │
└──────────────┘               └────────┬─────────┘  UpdateSnapshot│           │
                                        │                         │           │
                              Pop()      │                         │           │
                                        ▼                         │           │
┌──────────────┐               ┌──────────────────┐               │           │
│   kube-apiserver              │  ScheduleOne()   │               │           │
│              │◄──────────────│                  │               │           │
│  Binding     │   Bind 提交   │  Filter + Score  │◄──────────────┘           │
│  写入        │               │  (Framework)    │  NodeInfoSnapshot          │
└──────────────┘               └──────────────────┘                            │
                                 │                                                   │
                                 │ Pod 解绑                                          │
                                 ▼                                                   │
                        ┌──────────────────┐                                       │
                        │  回到 SchedulingQueue                                     │
                        │  (backoffQ 或 unschedulablePods)                          │
                        └───────────────────────────────────────────────────────────┘

7.2 四组件职责矩阵

组件	源码路径	核心结构体	数据类型	线程安全	与 apiserver 的关系
Informer	eventhandlers.go	SharedInformerFactory	Raw API Objects	线程安全	Watch + List
SchedulingQueue	backend/queue/	PriorityQueue	QueuedPodInfo	需要锁（lock）	无直接交互
Cache	backend/cache/	cacheImpl	NodeInfo（含聚合资源）	需要锁（mu）	Bind 时写入 Binding
Scheduling Framework	framework/runtime/	frameworkImpl	CycleState	无状态（只读快照）	无直接交互

八、Roadmap：调度专题后续预告

本篇覆盖了 kube-scheduler 的内部架构骨架——四大组件的定位、数据结构和协作方式。后续篇章会逐一深入每个组件的细节：

• 专题二（已发布）：Scheduling Framework 6 大内置插件精讲——NodeResourcesFit / NodeAffinity / TaintToleration / InterPodAffinity / PodTopologySpread / VolumeBinding 的源码行号 + 接口断言
• 专题三：Scheduling Framework 扩展点源码精讲——framework/runtime/framework.go 的 13 个 RunXxxPlugins 方法，CycleState 如何在扩展点之间传值
• 专题四：DefaultPreemption 抢占机制——当 Filter 阶段没有可行节点时，调度器如何通过抢占低优先级 Pod 来创造空间
• 专题五：SchedulingQueue 深度解析——QueueingHint 机制、PriorityQueue 的三个子队列如何协同、Backoff 算法的指数退避策略
• 专题六：Out-of-Tree 插件实战——用 RegisterPluginBuilder 注册自定义调度插件

调度专题的目标读者是资深运维开发：能读 Go 源码、有集群运维经验、对 k8s 整体架构已有认知。下一篇我会从 Scheduling Framework 的扩展点源码切入，把框架运行时的"骨架"补全。

---

本文参考与源码链接：
  • scheduler.go · Scheduler struct
  • eventhandlers.go · Informer 注册
  • scheduling_queue.go · PriorityQueue
  • cache.go · cacheImpl
  • framework.go · frameworkImpl
  • registry.go · NewInTreeRegistry
  • schedule_one.go · ScheduleOne 入口