Kubernetes 控制器与调度器源码解剖：从控制循环到自研调度框架

Kubernetes 有两个"看起来不难、写起来全是坑"的地方：控制器的调谐循环，和调度器的插件框架。

前者难在并发——你面对的是一个最终一致的分布式缓存，写操作和缓存可见性之间有延迟，稍不留神就会重复创建 Pod；后者难在契约——十几个扩展点、几十个插件、三种队列状态，要理解"为什么这个 Pod 现在不能被调度"得同时看懂队列、缓存和抢占。

本文的读法是从基础设施往上层走：先看 client-go 那套事件流水线（Reflector → DeltaFIFO → Indexer → Workqueue），再用 Deployment 和 ReplicaSet 两个内置控制器把控制循环串起来；然后拆 kube-scheduler 的调度周期与框架契约；最后落到自研调度器 Sky-Scheduler——看它在 kube-scheduler 的框架之外，为什么选择了另一套 Session / Action / Statement 模型。

代码基线：Kubernetes v1.32.0-alpha.0（commit 1bbe775d5ff，master，2024-09-29）；自研调度器 Sky-Scheduler（约 3 万行 Go，下文源码路径均相对其仓库根）。

前置阅读：Kubernetes Informer 机制详解 讲的是概念层，本文补的是源码层；自研调度器的设计细节（Quota 两级调度、RDMA 拓扑、Spot GPU）见 Sky-Scheduler：面向 AI 训练的拓扑感知与 Quota 两级调度器，本文不重复，只做与 kube-scheduler 的架构对照。

---

一、控制循环：一个反直觉的起点

先纠正一个常见的理解偏差：控制器并不"监听事件"。

事件确实会来（informer 的 AddFunc/UpdateFunc），但回调里做的事情通常只有一件——把对象的 key 塞进一个队列：

// pkg/controller/deployment/deployment_controller.go:400-408
func (dc *DeploymentController) enqueue(deployment *apps.Deployment) {
    key, err := controller.KeyFunc(deployment)
    if err != nil { utilruntime.HandleError(err); return }
    dc.queue.Add(key)
}

真正的业务逻辑在 worker 里，它拿到的是一个 key，然后回缓存里重新读当前状态，再决定做什么：

// pkg/controller/deployment/deployment_controller.go:487-498
func (dc *DeploymentController) processNextWorkItem(ctx context.Context) bool {
    key, quit := dc.queue.Get()
    if quit { return false }
    defer dc.queue.Done(key)
    err := dc.syncHandler(ctx, key.(string))
    dc.handleErr(ctx, err, key)
    return true
}

这个设计有三层含义：

1. 事件只是"提示"，状态才是真相。事件可能丢失、可能重复、可能乱序，但 syncHandler 每次都从缓存读最新状态，所以幂等。
2. 队列做了合并去重。同一个 Deployment 在 1 秒内被改了 100 次，队列里只有 1 个 key，控制循环只跑一次——这是 K8s 能扛住大规模抖动的关键。
3. “调谐”（reconcile）意味着双向。不是"收到创建事件就创建"，而是"把实际状态推向期望状态"，所以它既能处理"该有 Pod 但没有"，也能处理"不该有 Pod 却有了"。

标准控制器的骨架是一条带反馈的闭环——写回 API Server 的操作会产生新的 watch 事件，重新流回 Reflector：

把这张图拆成三段来看，正好对应后面三章：

段	节点	关键词	对应章节
采集	①→②	List 全量 + Watch 增量，用 delta 序列保序	第二章 2.1 / 2.2
分发	③→④→⑤	delta 变回调，回调只做入队，队列去重	第二章 2.3~2.5、第三章
调谐	⑥→回写→①	读缓存决策、写 API Server、事件回流	第四~六章

注意最后那条虚线不是装饰：控制器的每次写操作都会绕一圈变成自己的输入。这条回路决定了控制器必须幂等，也决定了 expectations 机制为什么必要（第五章）。

下面把这条流水线逐段拆开。

---

二、事件是怎么进来的

2.1 Reflector：Run 是退避外壳，ListAndWatch 是内核

Reflector 只干两件事：List 拿全量，Watch 拿增量。

// staging/src/k8s.io/client-go/tools/cache/reflector.go:304-312
func (r *Reflector) Run(stopCh <-chan struct{}) {
    wait.BackoffUntil(func() {
        if err := r.ListAndWatch(stopCh); err != nil {
            r.watchErrorHandler(r, err)
        }
    }, r.backoffManager, true, stopCh)
}

Run 本身不含业务逻辑，它是个退避重试外壳：ListAndWatch 出错返回，就按 backoffManager 睡一会儿再来一遍。注意 v1.32 里已经没有 ListAndWatchWithContext 了（老版本有，后来被移除）。

ListAndWatch 内部（reflector.go:340-370）先尝试 WatchList（一种"用 watch 代替 list"的优化），失败则 fallbackToList=true 退回普通 list()。

List 用哪个 resourceVersion，是个有讲究的决策：

// reflector.go:894-910（relistResourceVersion 的逻辑）
// 1. 上次同步的 RV 已被判定不可用 → ""（空字符串 = quorum read，最贵但最新）
// 2. 首次 List → "0"（走 API Server 的 watch cache，便宜）
// 3. 其他情况 → lastSyncResourceVersion（增量，最便宜）

这个三档设计是性能与正确性的折中："" 会穿透 watch cache 直达 etcd，代价高；"0" 能利用 watch cache 但可能拿到稍旧的数据；用上次的 RV 最省，但要求那个 RV 还在 etcd 的历史窗口里。

历史窗口过期会怎样？API Server 返回 410 Gone，Reflector 这样处理：

// reflector.go:932-938
func isExpiredError(err error) bool {
    return apierrors.IsResourceExpired(err) || apierrors.IsGone(err)
}

一旦命中，就标记 lastSyncResourceVersionUnavailable，下一轮 List 用 "" 强制做一次 quorum read 重新对齐——这就是为什么 informer 偶尔会有一次"突然变慢"的 List。

Watch 断连后的重连策略也值得一提（reflector.go:410-490）：每轮用 lastSyncResourceVersion + 随机 TimeoutSeconds + AllowWatchBookmarks: true 发起 watch。随机超时是为了打散所有客户端的重连时刻，避免 API Server 重启后惊群；Bookmark 事件不带对象只推进 RV，让 Reflector 知道"我还同步着"而不必传输数据。

2.2 DeltaFIFO：队列里存的是"变化"而不是"对象"

这是 client-go 里最容易被误解的一个数据结构。它不存对象，存的是同一对象上的一串变化：

// tools/cache/delta_fifo.go:163-175
const (
    Added   DeltaType = "Added"
    Updated DeltaType = "Updated"
    Deleted DeltaType = "Deleted"
    Replaced DeltaType = "Replaced"  // 仅当 EmitDeltaTypeReplaced=true
    Sync    DeltaType = "Sync"       // 向后兼容的别名
)
// delta_fifo.go:191
type Deltas []Delta  // index 0 最旧，末尾最新

为什么要存"变化序列"而不是"最新对象"？因为下游的 processDeltas 需要按顺序重放才能得到正确的回调语义。如果只存最新对象，Added 后面紧跟 Deleted 的场景就会被压缩成"一个不存在的对象"，回调就发不出来了。

队列本身还做了两层优化：

• 同 key 合并（delta_fifo.go:400 dedupDeltas / :439 queueActionLocked）：连续的同类型 delta 会被合并，避免队列膨胀。
• Delete 的短路（delta_fifo.go:337-356）：如果被删对象既不在 items 也不在 knownObjects 里，直接忽略——这个对象控制器从来没见过，删了也没人在意。

Replace 是另一个关键路径（delta_fifo.go:636-715），对应"List 拿回全量"的场景。它做三件事：对 list 中每项发 Sync/Replaced；对本地有、新 list 没有的 key 发 Deleted（用 DeletedFinalStateUnknown 包一层，因为对象已经拿不到了，只能给个 key）；如果配了 knownObjects，对缓存里有但 list 里没有的也发 Deleted。

2.3 processDeltas：delta 怎么变成回调

注意 processDeltas 不在 delta_fifo.go 里，而在 controller.go：

// tools/cache/controller.go:541-573（节选）
case Sync, Replaced, Added, Updated:
    if old, exists, err := clientState.Get(obj); err == nil && exists {
        clientState.Update(obj); handler.OnUpdate(old, obj)
    } else {
        clientState.Add(obj); handler.OnAdd(obj, isInInitialList)
    }
case Deleted:
    clientState.Delete(obj); handler.OnDelete(obj)

这里藏着**"Add 和 Update 怎么区分"的答案**：不是靠 delta 类型，而是靠本地缓存里有没有这个对象。一个 Added delta 如果发现缓存里已经有了，就会走 OnUpdate。这解释了为什么控制器里 OnAdd 和 OnUpdate 经常写成同一个函数。

那个"本地缓存"就是 Indexer——clientState 是 processDeltas 的第二个形参，而 HandleDeltas 传进去的正是 s.indexer：

// shared_informer.go:638-646
func (s *sharedIndexInformer) HandleDeltas(obj interface{}, isInInitialList bool) error {
    s.blockDeltas.Lock()
    defer s.blockDeltas.Unlock()
    if deltas, ok := obj.(Deltas); ok {
        return processDeltas(s, s.indexer, deltas, isInInitialList)  // clientState = s.indexer
    }
    return errors.New("object given as Process argument is not Deltas")
}

注意形参类型声明的是 Store 而不是 Indexer（controller.go:542-548）——processDeltas 只用到 Get/Add/Update/Delete，不需要索引能力，所以按最小接口声明。而 Indexer 接口本身内嵌了 Store，所以实参可以直接传 Indexer。同一个函数在 shared informer 场景下拿到的就是 Indexer，在裸 controller 场景下则是调用方自己传入的 Store。

2.4 Indexer：不只是缓存，还是索引

Indexer 是带索引的本地缓存（index.go:35-55），默认带一个 namespace 索引（MetaNamespaceIndexFunc）。

索引的价值在于按维度批量取，而不是遍历全部对象：

// pkg/controller/nodelifecycle/node_lifecycle_controller.go:129
nodeNameKeyIndex = "spec.nodeName"
// :391 注册
podIndexer.AddIndexers(cache.Indexers{nodeNameKeyIndex: func(obj interface{}) ([]string, error) {...}})
// :406 使用
pods, err := podIndexer.ByIndex(nodeNameKeyIndex, nodeName)

节点生命周期控制器要频繁回答"这个节点上有哪些 Pod"，如果每次都全量遍历所有 Pod，万节点集群下就是 O(N) 的灾难。注册一个 spec.nodeName 索引后，查询退化成一次 map 查找（thread_safe_store.go:313-327 的 ByIndex）。

同样的手法在 pkg/controller/tainteviction/taint_eviction.go:203 也用了。

2.5 SharedInformer：一份缓存，多路分发

sharedIndexInformer 的 Run 里能看到清晰的启动顺序（shared_informer.go:459-509）：

// :471-475  先建 DeltaFIFO，把 indexer 作为 knownObjects 传进去
fifo := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
    KnownObjects: s.indexer, EmitDeltaTypeReplaced: true,
})
// :486-490  组装 Config，Process 指向 s.HandleDeltas
cfg := &Config{Queue: fifo, Process: s.HandleDeltas, ...}
s.controller = New(cfg)
// :501      先起 processor 的监听 goroutine
go s.processor.run(...)
// :508      再启动 controller（Reflector + processLoop）
s.controller.Run(stopCh)

顺序很重要：processor 必须先跑起来，controller 才能往它里面塞事件，否则前几个事件会丢。另外 processor 用的是独立的 stopCh，保证它严格晚于 controller 停止。

HandleDeltas 拿到 delta 后调 processDeltas(s, s.indexer, deltas, isInInitialList)（:638-646），一路分发到所有注册的 listener。

OnUpdate 里有个细节值得注意：

// shared_informer.go:668
isSync = accessor.GetResourceVersion() == oldAccessor.GetResourceVersion()

resourceVersion 没变，说明这不是真更新，而是 resync 造出来的"假更新"。这种事件只会分发给那些显式要求 resync 的 listener，避免控制器被无意义的周期性事件唤醒。

processorListener 内部还有一个环形缓冲区（shared_informer.go:869-905 的 pendingNotifications buffer.RingGrowing）：如果 handler 处理太慢，事件会先落进 ring buffer 而不是阻塞分发。这是个典型的用内存换吞吐的取舍——代价是慢消费者会吃掉内存，所以 initialBufferSize = 1024 起，能涨但不会无限涨。

HasSynced 有两级语义，这点经常被搞混：

• sharedIndexInformer.HasSynced（:517-525）：委托给 DeltaFIFO，表示至少完成过一次全量同步。
• processorListener.HasSynced（:909-911）：基于 syncTracker，要求源 informer 已同步 且 本 listener 已消费完初始 Add 事件。

区别的意义在于：如果你在 informer 启动之后才注册 handler，AddEventHandlerWithResyncPeriod（:581-636）会持 blockDeltas 锁，把 indexer 里现有的对象合成一批 isInInitialList: true 的 Add 事件喂给这个新 listener。这样新 handler 的 HasSynced 就不会被误判成"已经同步完了"，控制器也就不会在没看到存量对象的情况下开始工作。

WaitForCacheSync 的实现朴素到有点意外（:327-342）：每 100ms 轮询一次所有 InformerSynced（syncedPollPeriod = 100ms）。没有 channel、没有条件变量，就是轮询。

---

三、Workqueue：三种队列合一

控制器的队列不是普通队列，Typed[T] 内部同时维护了三个集合：

// staging/src/k8s.io/client-go/util/workqueue/queue.go:192-216（字段节选）
type Typed[t comparable] struct {
    queue      Queue[t]   // 待处理，有序
    dirty      set[t]     // 需要处理
    processing set[t]     // 正在处理
}

v1.32 的 workqueue 已经全面泛型化（Type = Typed[any] 是兼容别名）。

Add 的语义（:240-264）：

if q.dirty.has(item) { return }      // 已在待处理集合，直接忽略
q.dirty.insert(item)
if q.processing.has(item) { return } // 正在处理 → 不重复入队，但记住"脏了"
q.queue.Push(item)

Get（:278-297）和 Done（:302-315）配对：

// Get
q.processing.insert(item); q.dirty.delete(item)
// Done
q.processing.delete(item)
if q.dirty.has(item) { q.queue.Push(item); q.cond.Signal() }

这三行是精髓。设想一个场景：worker 正在处理 Deployment foo（已在 processing），此时 foo 又被改了两次。第二次 Add 时 dirty 已经有它了，直接返回；Done 时发现 dirty 里还有它，于是重新入队。

效果是：处理期间的所有变更被压缩成一次额外的重跑。既不会漏（脏标记保证了重跑），也不会爆（N 次变更只多跑一次）。如果只用普通队列，这个场景要么丢更新，要么排 N 个任务。

延迟队列与限速

DelayingInterface 支持 AddAfter(item, duration)（delaying_queue.go:240-259），内部用一个按 readyAt 排序的最小堆（waitingLoop，:267+），配合 10s 的 heartbeat 兜底。这个能力被控制器用来做"稍后再试"，比如 ReplicaSet 在 MinReadySeconds 场景下的 queue.AddAfter。

限速器默认是这样组合的：

// util/workqueue/default_rate_limiters.go:50-56
func DefaultTypedControllerRateLimiter[T comparable]() TypedRateLimiter[T] {
    return NewTypedMaxOfRateLimiter(
        NewTypedItemExponentialFailureRateLimiter[T](5*time.Millisecond, 1000*time.Second),
        &TypedBucketRateLimiter[T]{Limiter: rate.NewLimiter(rate.Limit(10), 100)},
    )
}

即取两者中较慢的那个：单条目指数退避 5ms × 2^n（封顶 1000s），整体令牌桶 10 qps / burst 100。前者防止单个坏对象打爆 API Server，后者限制控制器整体的写速率。

Forget 是这套机制的另一半——它清空该 key 的重试计数。控制器在 err == nil 时必须调 Forget，否则这个 key 的退避时间会一直累积下去，下次真出错时第一次重试就要等几分钟。

processLoop：把队列和 informer 缝起来

// tools/cache/controller.go:193-206
func (c *controller) processLoop() {
    for {
        obj, err := c.config.Queue.Pop(PopProcessFunc(c.config.Process))
        if err != nil {
            if err == ErrFIFOClosed { return }
            if c.config.RetryOnError { c.config.Queue.AddIfNotPresent(obj) }
        }
    }
}

注意这里的 Queue 是 DeltaFIFO，不是 workqueue。processLoop 负责把 DeltaFIFO 里的事件喂给 Process（也就是 HandleDeltas → 事件回调 → queue.Add(key)），workqueue 那一层是控制器自己接在事件回调后面的。两条队列、两级缓冲：DeltaFIFO 缓冲"从 API Server 来的事件"，workqueue 缓冲"待调谐的对象"。

---

四、Deployment Controller：一次 syncDeployment 的全过程

4.1 接线

NewDeploymentController（deployment_controller.go:101-159）注册了三个 informer 的事件处理：

资源	注册的事件	行号
Deployment	Add / Update / Delete	:120-131
ReplicaSet	Add / Update / Delete	:132-142
Pod	仅 Delete	:143-147

Pod 只监听 Delete 是个有意的取舍：Pod 的创建和更新对 Deployment 的决策影响不大（Deployment 关心的是 RS 的 Status.AvailableReplicas），但 Pod 被删会影响"可用副本数"，需要及时重算。少监听一种事件，就少一次无谓的调谐。

worker 启动前必须等缓存同步：

// deployment_controller.go:176
if !cache.WaitForNamedCacheSync("deployment", ctx.Done(), dListerSynced, rsListerSynced, podListerSynced) {
    return
}

三个 lister 全同步才开工。否则第一轮调谐会看到"RS 列表是空的"，误判成需要创建新 RS。

4.2 主干分支

syncDeployment（:590-680）是一棵判断树，按优先级从高到低：

判断顺序本身就是优先级声明：删除中 > 暂停 > 回滚 > 扩缩事件 > 正常滚动。越靠前的分支越"紧急且不能被打断"。

两个细节值得留意：

• DeepCopy 是必须的（:604-615）。informer 缓存里的对象是所有控制器共享的，直接改会造成难以排查的脏数据。
• DeletionTimestamp != nil 时只同步状态（:643-645）。正在被删除的 Deployment 不应该再触发扩缩，否则会出现"删到一半又创建 Pod"的诡异现象。

4.3 滚动更新的数学

rolloutRolling（rolling.go:31-66）的策略是先扩容新的，扩不动了再缩旧的：

1. reconcileNewReplicaSet   → 成功就返回（先把新 RS 拉起来）
2. reconcileOldReplicaSets  → 成功就返回（新 RS 满了才缩旧的）
3. cleanupDeployment        → 清理历史 RS
4. syncRolloutStatus

新 RS 该扩到多少，公式在 deployment_util.go:784-809：

maxTotalPods  = replicas + maxSurge
currentPodCount >= maxTotalPods ? 保持当前 newRS.replicas
                                : scaleUpCount = min(maxTotalPods - currentPodCount,
                                                     replicas - newRS.replicas)
                                  返回 newRS.replicas + scaleUpCount

maxSurge 和 maxUnavailable 的解析有个坑，在 ResolveFenceposts（:851-870）：maxSurge 向上取整，maxUnavailable 向下取整——即 10% 的 surge 在 3 副本下是 1（0.3 向上取整），而 10% 的 unavailable 在 3 副本下是 0。而且当两者算出都是 0 时，会强制把 unavailable 设为 1，否则滚动更新永远推进不下去。

旧 RS 的缩容更微妙，reconcileOldReplicaSets（rolling.go:86-152）里有三个连着的公式：

// rolling.go:127-129
minAvailable            := *(deployment.Spec.Replicas) - maxUnavailable
newRSUnavailablePodCount := *(newRS.Spec.Replicas) - newRS.Status.AvailableReplicas
maxScaledDown           := allPodsCount - minAvailable - newRSUnavailablePodCount

第三个式子是关键：新 RS 里那些还没 Ready 的 Pod 要算进"不可用预算"。如果不减掉 newRSUnavailablePodCount，就会在新 Pod 还没起来的时候急着缩旧 Pod，导致瞬时可用副本跌破 minAvailable。这是滚动更新里最经典的一个 off-by-design 错误，K8s 用这个减法规避掉了。

在真正缩容前，cleanupUnhealthyReplicas（:155-189）会先把旧 RS 里 replicas - available 的那部分"不健康副本"清掉——这些 Pod 本来就没在工作，缩掉它们不消耗可用性预算。

4.4 一个容易过时的认知：Deployment 现在没有 expectations

很多资料（包括不少书）会讲 DeploymentController 的 expectations 机制。在 v1.32 里它已经不存在了：

$ grep -n "expectations" pkg/controller/deployment/deployment_controller.go
（无输出）

这套机制在 2016 年 2 月的 commit d0dd4f8a939（“Eliminate use of expectations in Deployment controller.”）里被移除了。现在的 Deployment 依赖 ReplicaSet 自己的 expectations 加上状态收敛来保证正确性。

expectations 本身没有消失，它只是下沉到了真正做 Pod 增删的那一层——也就是下面要讲的 ReplicaSet。

---

五、Expectations：分布式系统里最容易被忽略的一处补丁

5.1 问题是什么

控制器通过 informer 缓存读状态，但写操作到缓存可见之间有延迟。上下对比就能看出 expectations 补的是哪个洞：

这个窗口在集群压力大时可能持续数百毫秒到数秒。解决思路很直接：发起写操作时"记账"，等 informer 观察到相应事件后再允许下一次判断。

5.2 实现

// pkg/controller/controller_utils.go:160-162
type ControllerExpectations struct {
    cache.Store
}
// :186-211
func (r *ControllerExpectations) SatisfiedExpectations(logger, controllerKey string) bool {
    if exp, exists, err := r.GetExpectations(controllerKey); err == nil && exists {
        return exp.Fulfilled() || exp.isExpired()
    }
    return true  // 没有记录 = 满足
}

两个关键点：

其一，超时是必须的。isExpired（:216-218）判断 RealClock.Since(exp.timestamp) > ExpectationsTimeout，而 ExpectationsTimeout = 5 * time.Minute（:70）。

为什么必须有超时？因为 expectations 是乐观的：如果它期待 3 个 Pod 创建事件，但其中一个创建请求失败了（配额不足、webhook 拒绝），事件永远不来，计数永远减不到 0，控制器就会永久卡死。5 分钟超时是"宁可重复一次操作，也不能永远不工作"的兜底。这也解释了为什么控制器逻辑必须幂等——超时兜底意味着重复执行是设计内的行为。

其二，删除比创建麻烦得多。UIDTrackingControllerExpectations（:331-339）额外用了一个 uidStore 记录待删除 Pod 的 UID。原因是优雅删除会产生两个事件：先是 DeletionTimestamp 被设置的 Update，再是真正的 Delete。如果只数 Delete 事件，可能会因为中间那次 Update 而算错账；而记录 UID 后，DeletionObserved（:371-381）可以精确判断"这个 Pod 是不是我期待被删的那个"。

5.3 在 ReplicaSet 里的用法

// pkg/controller/replicaset/replica_set.go:696
rsNeedsSync := rsc.expectations.SatisfiedExpectations(logger, key)
// :721-723
if rsNeedsSync && rs.DeletionTimestamp == nil {
    manageReplicas(ctx, filteredPods, rs)
}

expectations 没满足，就整个跳过 manageReplicas。注意它只挡住了"改变 Pod 数量"这一件事，状态同步（updateReplicaSetStatus）照常执行——这是有意的，状态同步是只读的，不受延迟影响。

记账发生在 manageReplicas 里：

// :587
rsc.expectations.ExpectCreations(logger, rsKey, diff)
// :638
rsc.expectations.ExpectDeletions(logger, rsKey, getPodKeys(podsToDelete))

而销账在 Pod 事件回调里（:405 的 CreationObserved、:536 的 DeletionObserved）。

---

六、ReplicaSet：慢启动与收养

6.1 manageReplicas

manageReplicas（replica_set.go:569-670）算出差额 diff = len(filteredPods) - *rs.Spec.Replicas，然后：

• diff < 0：创建。先把 diff 封顶到 burstReplicas（:579-581），再 slowStartBatch 批量创建。
• diff > 0：删除。同样封顶，用 getPodsToDelete（:825-834）按"关联 RS 在同节点的分布"排序后挑选，尽量打散。

BurstReplicas = 500（:68）——单轮最多动 500 个 Pod。没有这个上限，一个"从 0 扩到 10000"的 RS 会瞬间向 API Server 打出一万个创建请求。

6.2 slowStartBatch：为什么要慢

// replica_set.go:771-795（节选）
for batchSize := min(remaining, initialBatchSize); batchSize > 0; batchSize = min(2*batchSize, remaining) {
    // 并发跑 batchSize 个 fn()
    wg.Wait()
    if len(errCh) > 0 { return successes, <-errCh }  // 任一批有错就停
    remaining -= batchSize
}

SlowStartInitialBatchSize = 1（controller_utils.go:85），所以批次是 1 → 2 → 4 → 8 → … 指数增长。

为什么第一批只发 1 个？因为创建 Pod 很可能失败（节点资源不足、配额限制、admission webhook 拒绝）。如果一上来就并发 500 个全部失败，那就是 500 次无效的 API 调用加 500 条错误日志。慢启动用第一个请求当"探针"，成功才加速；失败立刻停手并返回错误，让控制器退避重试。

这是个很朴素的拥塞控制思想，和 TCP 慢启动同源。

6.3 收养与释放

Pod 和 ReplicaSet 之间靠 OwnerReference 关联。当你在 RS 上改 selector，会发生两件事：新匹配的孤儿 Pod 被收养，不再匹配的 Pod 被释放。统一入口是 ClaimObject：

// pkg/controller/controller_ref_manager.go:69-128（逻辑归纳）
// 1. 有 controllerRef 且 UID 不是自己  → 忽略（别人的）
// 2. 有自己的 controllerRef 且 selector 匹配 → 保留
// 3. 有自己的 controllerRef 但 selector 不匹配 → release
// 4. 没有 controllerRef（孤儿）且匹配且 namespace 相同 → adopt

controller=true 的 OwnerReference 是唯一的（由 ValidateOwnerReferences 保证），所以 metav1.GetControllerOf 返回的那条就是 ControllerRef。这个唯一性假设是整套收养逻辑能成立的前提。

6.4 GC 的另一半

OwnerReference 不只是在控制器里用的元数据，它还是垃圾回收的图。pkg/controller/garbagecollector/garbagecollector.go 的 attemptToDeleteItem（:498）会把 ownerRefs 分成三类：

• solid：owner 存在且正常 → 删除自己即可
• dangling：owner 不存在了 → 自己变成孤儿，按 orphan 或 delete 策略处理
• waitingForDependentsDeletion：owner 正在被删且带 Foreground 策略 → 得等依赖项先删完

Foreground / Background / Orphan 三种级联策略的差异，最终就落在这个分类上。

---

七、kube-scheduler：一个调度周期的全景

从这一节开始进入调度器。先看主干：

// pkg/scheduler/scheduler.go:470
go wait.UntilWithContext(ctx, sched.ScheduleOne, 0)

间隔是 0，也就是"尽快"。这是一个单 goroutine 的串行循环——ScheduleOne 一次处理一个 Pod。

但绑定不在这里做：

// pkg/scheduler/schedule_one.go:65-133（结构归纳）
podInfo := sched.NextPod(logger)              // :67  从队列取
fwk := sched.frameworkForPod(pod)             // :85  按 schedulerName 选 profile
state := framework.NewCycleState()            // :103 一次调度的临时状态
scheduleResult, err := sched.schedulingCycle(...)  // :113 同步
if err != nil { sched.FailureHandler(...) }   // :115 失败处理
go func() { ... sched.bindingCycle(...) }()   // :120-132 异步！

调度周期（scheduling cycle）同步、绑定周期（binding cycle）异步——这是 kube-scheduler 吞吐量的主要来源。调度决策（选节点）必须在单线程里串行做，因为它要读写共享的缓存快照；但"把决策写回 API Server"是 IO 密集型操作，可以并行。

7.1 调度周期做了什么

schedulingCycle（:138-263）的关键顺序是：

schedulePod（选节点）
  → 失败且是 FitError → RunPostFilterPlugins（抢占）
  → assume（乐观写入缓存）
  → RunReservePluginsReserve
  → RunPermitPlugins

注意 assume 在 Reserve 之前（:197 vs :208）。assume 做的事情是把 Pod 的 NodeName 直接设上并写进调度器缓存（schedule_one.go:946-963）：

assumed.Spec.NodeName = host
sched.Cache.AssumePod(assumed)
sched.Cache.DeleteNominatedPodIfExists(...)

为什么要"假设"？ 因为下一个 Pod 的调度决策必须看到这个 Pod 已经占了资源，否则会把它也调度到同一个节点上，造成超卖。但从调度决策到真正 bind 成功有网络往返，如果等 bind 完成再调下一个 Pod，吞吐会低得无法接受。

代价是：假设可能是错的（bind 失败）。所以有兜底：

// pkg/scheduler/backend/cache/cache.go:35
cleanAssumedPeriod = 1 * time.Second
// :730-753 cleanupAssumedPods

每秒扫一遍 assumed 状态超时的 Pod，清理掉。durationToExpireAssumedPod = 0（scheduler.go:57）意味着只要 bind 没在预期时间内完成就回收。用"可能出错然后修正"换吞吐，这是调度器里最核心的一个工程取舍。

7.2 节点打分为什么要采样

节点少的时候，Filter + Score 遍历所有节点没问题。但万节点集群里，每个 Pod 都做 O(N) 的过滤和打分是不可能的。所以有：

// pkg/scheduler/schedule_one.go:676-702
if percentage == 0 {
    percentage = int32(50) - numAllNodes/125
    if percentage < minFeasibleNodesPercentageToFind {  // 5
        percentage = minFeasibleNodesPercentageToFind
    }
}
numNodes = numAllNodes * percentage / 100
if numNodes < minFeasibleNodesToFind {  // 100
    return minFeasibleNodesToFind
}

三个常量的配合（minFeasibleNodesPercentageToFind = 5、minFeasibleNodesToFind = 100、公式里的 50 - N/125）给出的曲线是：

集群规模	采样比例	实际检查节点数
100	50%	100（下限）
1,000	42%	420
5,000	10%	500
10,000	5%（下限）	500

节点越多，采样比例越低，绝对检查数收敛在几百量级。这背后是一个判断：在几千个节点里，随机采样 500 个已经足够找到一个"足够好"的节点，找到"最优"节点的收益抵不上多出来的延迟。这是有界延迟优先于最优解的设计。

还有个细节：nextStartNodeIndex（schedule_one.go:520）会记住上次从哪里开始遍历，下次从那里接着走——避免每次都从第一个节点开始采样，导致后面的节点永远不被考虑。

7.3 并发模型

Filter 和 Score 对节点的遍历是并行的，但并行度不是拍脑袋定的：

// pkg/scheduler/framework/parallelize/parallelism.go:43-52
func (p Parallelizer) chunkSizeFor(numItems int) int {
    return max(1, min(int(math.Sqrt(float64(numItems))), numItems/p.parallelism))
}
// DefaultParallelism = 16（:28）

分块大小取 √n 和 n/16 的较小值。1000 个节点时：√1000 ≈ 31，1000/16 = 62，取 31 → 分 33 块。100 个节点时：√100 = 10，100/16 = 6，取 6 → 分 17 块。

块内串行、块间并行，goroutine 数量随节点数增长但增长很慢（亚线性）。这避免了"1 万个节点起 1 万个 goroutine"的调度开销失控。

7.4 选节点的最后一步：蓄水池采样

// pkg/scheduler/schedule_one.go:873-920（节选）
if ns.TotalScore == sortedNodeScoreList[0].TotalScore {
    cntOfMaxScore++
    if rand.Intn(cntOfMaxScore) == 0 {
        // Replace the candidate with probability of 1/cntOfMaxScore
        selectedIndex = cntOfMaxScore - 1
    }
}

这是标准的蓄水池采样（reservoir sampling），在 O(N) 内从所有最高分节点中等概率挑一个。

为什么不是"取第一个最高分"？因为如果所有节点同分（比如一个完全空闲的同构集群），取第一个会让所有 Pod 都挤到同一个节点上，直到它不再是最高分——这是典型的热点问题。随机化把负载摊开了。

---

八、调度框架的插件契约

8.1 扩展点

pkg/scheduler/framework/interface.go 定义了整套插件契约。按调用顺序：

扩展点	接口定义行	职责
PreEnqueue	:453-457	Pod 入队前的门控（gate）
QueueSort	:465-469	activeQ 的排序规则
PreFilter	:513-532	预计算，可返回 PreFilterResult 缩小候选节点
Filter	:541-558	单节点可行性判定
PostFilter	:562-582	Filter 全失败后调用（抢占入口）
PreScore	:588-596	打分前的预计算
Score	:608-617	单节点打分（0-100）
Reserve	:625-637	预留资源（可回滚）
Permit	:661-670	准入/延迟绑定
PreBind	:641-646	绑定前的最后准备（如真正绑 PV）
Bind	:674-683	执行绑定
PostBind	:650-657	绑定后的收尾

Score 配套一个可选的 ScoreExtensions.NormalizeScore（:599-604）。

8.2 Status 的语义：Unschedulable 还是 UnschedulableAndUnresolvable

Code 常量在 :191-245：

Success = 0; Error = 1; Unschedulable = 2; UnschedulableAndUnresolvable = 3
Wait = 4; Skip = 5; Pending = 6

Unschedulable 和 UnschedulableAndUnresolvable 的区别是整套队列优化机制的基础：

• Unschedulable：当前不满足，但集群变化后可能满足。比如"节点资源不够"——加个节点就好了。
• UnschedulableAndUnresolvable：当前不满足，且集群变化也救不了。比如"这个 PVC 的 StorageClass 根本不存在"。

这个区别决定了 Pod 被塞进哪个队列、以及后续集群事件要不要唤醒它（见 9.1 的 QueueingHint）。插件作者如果随便返回 Unschedulable，会导致大量无意义的重新入队。

8.3 CycleState 与 StateData

CycleState（cycle_state.go:48-57）是单次调度的临时状态，底层是 sync.Map——"一次写、多次读"的模式下读是并发安全的。

插件之间靠它传递数据，约定是"PreFilter 写，Filter/Score 读"：

// cycle_state.go:31-37
type StateData interface {
    Clone() StateData
}

接口只有 Clone()，注释里说明了原因：clone 应该对"不受 PreFilter 的 AddPod/RemovePod 影响的成员"做浅拷贝，以省开销。

Clone 存在的意义在抢占（preemption）里体现得最清楚：

// pkg/scheduler/framework/preemption/preemption.go:548-590（DryRunPreemption）
// 拷贝 Snapshot、State.Clone()，然后在副本上试算

抢占是试算：要模拟"如果驱逐这些 Pod，被抢占的 Pod 能不能调度上去"。这个试算会写 CycleState、会改 NodeInfo。如果不克隆，试算过程就污染了真实的调度状态。

8.4 RunScorePlugins：三趟并行

// pkg/scheduler/framework/runtime/framework.go:1101-1207
// 第一趟：每个插件跑 Score（并行）
// 第二趟：每个插件跑 NormalizeScore（并行）
// 第三趟：按 weight 累加

分三趟是因为 NormalizeScore 需要看到该插件在所有节点上的原始分才能归一化。先全部算完原始分，再统一归一化，最后加权求和。

8.5 Unreserve 必须逆序

// framework.go:1403-1425
// Execute the Unreserve operation of each reserve plugin in the
// *reverse* order in which the Reserve operation was executed.
for i := len(f.reservePlugins) - 1; i >= 0; i-- {

这是个很容易被忽略但很关键的细节。Reserve 是有依赖的：如果插件 A 先预留了资源，插件 B 再基于 A 的结果做预留，那么回滚时必须先回滚 B 再回滚 A，否则会短暂处于不一致状态。和数据库事务的回滚、栈的弹出是同一个道理。

8.6 默认插件与 MultiPoint

pkg/scheduler/apis/config/v1/default_plugins.go 的 getDefaultPlugins（:30-58）定义了默认启用的插件列表。这些插件是用 MultiPoint 形式声明的——一个插件可以在多个扩展点上生效：

// framework.go:511-596 expandMultiPointPlugins
// 把 Plugins.MultiPoint 里声明的插件，按它实现了哪些接口，
// 展开到对应的扩展点 slice 里

所以插件的"启用"和"在哪个扩展点生效"是分开的：你启用 NodeResourcesFit，它自动出现在 PreFilter、Filter、Score 三个位置上，因为它实现了这三个接口。

---

九、三级队列与 Cache

9.1 三种队列状态

调度队列不是一条队列，是三条（pkg/scheduler/backend/queue/scheduling_queue.go）：

                 ┌──────────────┐
集群事件/新 Pod ──►│   activeQ    │──► 出队调度
                 └──────┬───────┘
                        │ 调度失败
                 ┌──────▼───────┐
                 │   backoffQ   │  按 calculateBackoffDuration 退避
                 └──────┬───────┘
                        │
                 ┌──────▼───────────────┐
                 │ unschedulablePods    │  带 UnschedulablePlugins 记录
                 └──────────────────────┘

Run 里有两个定时器（:356-363）：

// 每 1 秒
go wait.UntilWithContext(ctx, sched.flushBackoffQCompleted, 1*time.Second)
// 每 30 秒
go wait.UntilWithContext(ctx, sched.flushUnschedulablePodsLeftover, 30*time.Second)

退避时长是经典的指数退避（:1280-1296），参数默认 DefaultPodInitialBackoffDuration = 1s（:77）、DefaultPodMaxBackoffDuration = 10s（:81）。

unschedulablePods 里的 Pod 有一个"保底时长" DefaultPodMaxInUnschedulablePodsDuration = 5min（:64）——超过这个时间还没被任何事件唤醒，就强制回 activeQ 重试一次。这是防止 Pod 永久饿死的兜底。

关键问题：集群变化时，要不要把所有 unschedulable 的 Pod 都唤醒？

朴素的答案是把所有 Pod 都扔回 activeQ——但这会造成"来一个节点，全集群几千个 Pod 重新调度一遍"的雪崩。v1.32 的答案是 QueueingHint：

// scheduling_queue.go:406-480 isPodWorthRequeuing
// 返回 queueSkip / queueAfterBackoff / queueImmediately

每个插件通过 EnqueueExtensions.EventsToRegister（interface.go:483-497）声明"我在意哪些事件"，队列在收到事件时逐个问插件"这个变化对这个 Pod 有意义吗"。moveRequestCycle（:181，更新在 :1108）记录了本轮事件的版本号，用来识别"这个 Pod 在本轮事件里已经被评估过了"。

9.2 Nominator：被提名的节点

// pkg/scheduler/backend/queue/nominator.go:35-53
nominatedPods     map[string][]podRef   // node → pods
nominatedPodToNode map[types.UID]string // pod UID → node

抢占发生后，被抢占的 Pod 会"提名"一个节点（它抢到的那个）。这个提名记录有两个作用：

1. evaluateNominatedNode（schedule_one.go:547-565）：被提名的 Pod 再次调度时，会优先检查提名节点，跳过 PreFilter 的一部分约束。
2. addNominatedPods（framework.go:1023-1046）：其他 Pod 在这个节点上做 Filter 时，要把"被提名的 Pod"也算进去，避免抢占刚成功就被别的 Pod 抢走位置。

9.3 NodeInfo 的 Generation

// pkg/scheduler/framework/types.go:649-651
func nextGeneration() int64 { return atomic.AddInt64(&generation, 1) }

NodeInfo 每次变更都会自增 Generation（:588-628 结构体里的字段）。快照机制用它来判断"这个节点自上次快照以来变了吗"——没变就复用同一个 NodeInfo 指针，不做克隆（cache.go:186-280 的 UpdateSnapshot）。

调度周期是每秒几十上百次的，每次全量克隆所有 NodeInfo 的开销无法接受。Generation 让增量更新成为可能。

---

十、几个关键插件的源码

10.1 NodeResourcesFit：三种打分公式

resource_allocation.go 里有个统一的 resourceAllocationScorer（:38-45），三种策略只是换打分函数：

// noderesources/least_allocated.go:52-61
return ((capacity - requested) * framework.MaxNodeScore) / capacity
// noderesources/most_allocated.go:54-65
return (requested * framework.MaxNodeScore) / capacity

• LeastAllocated：越空分越高 → 打散，适合通用负载，降低单点故障影响面。
• MostAllocated：越满分越高 → 装箱，适合批处理，把空闲节点腾出来缩容。
• RequestedToCapacityRatio（requested_to_capacity_ratio.go:31-58）：用一条可配置的折线（shape）插值，可以表达"用到 70% 之前不敏感，超过 70% 急剧降分"这类非线性偏好。

三种策略在 fit.go:59-84 的 nodeResourceStrategyTypeMap 里注册。

10.2 PodTopologySpread：skew 怎么算

拓扑打散的核心是 skew：

// podtopologyspread/filtering.go:354
skew = matchNum + selfMatchNum - minMatchNum

含义是：当前拓扑域内已有的匹配 Pod 数（加上本 Pod 自己）减去所有域中的最小值。Filter 阶段要求 skew <= maxSkew。

打分阶段（scoring.go:191-224）用 scoreForCount(cnt, maxSkew, weight)，归一化时：

// scoring.go:263
MaxNodeScore * (maxScore + minScore - s) / maxScore

越少越均匀，分越高。归一化权重是 math.Log(size+2)（:295-297）——域越大，权重越高，因为在大的域里打散更重要。

系统还有一组默认约束（plugin.go:47-58）：Hostname 维度 maxSkew=3、Zone 维度 maxSkew=5，都是 ScheduleAnyway（软约束）。这意味着即使你没配 topologySpreadConstraints，跨主机打散也在默默生效。

10.3 VolumeBinding：为什么必须拆成三个阶段

这是调度框架里最能体现"扩展点设计"的一个插件：

// volumebinding/volume_binding.go
PreFilter  :323-353   // 无 PVC 返回 Skip；immediate 模式未绑定的 PVC 直接判 UnschedulableAndUnresolvable
Filter     :387-416   // 对每个候选节点查 FindPodVolumes
Reserve    :469-487   // AssumePodVolumes —— 假定绑定
PreBind    :494-517   // BindPodVolumes —— 真正调 API
Unreserve  :521-532   // RevertAssumedPodVolumes —— 回滚

为什么不能在一个地方做完？

• PreFilter 阶段快速失败：如果 PVC 的 StorageClass 是 Immediate 且还没绑定，这个 Pod 无论调到哪个节点都跑不起来。早失败比在 Filter 里对每个节点重复判断便宜得多。
• Reserve 阶段做假定绑定：PV 是独占资源。如果两个 Pod 同时调度，都看到"这个 PV 可用"，就会双双选中——但 PV 只能绑一个。所以在 Reserve 阶段就要用 AssumeCache（assume_cache.go:298 的 Assume）把 PV 临时标记为已绑定，让后来的 Pod 看不到它。
• PreBind 才真正落盘：因为直到这一刻，调度决策还可能被抢占或 Permit 插件推翻。真正写 API 必须放在最后。

assume_cache.go 里的 Assume/Restore（:298 / :332）就是这套"乐观占位 + 可回滚"的实现。

10.4 DefaultPreemption：PostFilter 里发生了什么

PostFilter（default_preemption.go:85-105）只在 Filter 全失败时触发，但它做的事情相当重：

1. PodEligibleToPreemptOthers (:239-264)  —— 这个 Pod 有资格抢占别人吗
2. findCandidates (:216-250)              —— 找可能装得下的节点
3. SelectVictimsOnNode (:140-229)         —— 对每个候选节点算要驱逐谁
4. pickOneNodeForPreemption (:431-517)    —— 在候选节点里挑一个
5. prepareCandidate (:349-409)            —— 真正驱逐

第 4 步用了五个有序的评分函数（preemption.go:442-494），按优先级依次比较：

minNumPDBViolating → minHighestPriority → minSumPriorities → minNumPods → latestStartTime

PDB（PodDisruptionBudget）违规数排在第一位，优先于"被驱逐 Pod 的优先级"。这个排序说明了设计意图：尽量不破坏可用性保证，哪怕代价是驱逐更高优先级的 Pod。

filterPodsWithPDBViolation（:290-336）负责算违规数。而整个试算过程跑在 DryRunPreemption（:548-590）里——在克隆的 Snapshot 和 CycleState 上做，不污染真实状态。

---

十一、从 kube-scheduler 到自研：什么时候原生不够

kube-scheduler 的设计目标很明确：把单个 Pod 放到最合适的节点上，延迟有界。这个目标下有几个它天然不擅长的事：

需求	kube-scheduler 的处境
Gang 调度（全有或全无）	单 Pod 贪心，先调度成功的一半会占着资源等另一半，可能死锁
队列配额 / 借还	无队列概念，只有 namespace 级 ResourceQuota（在 API Server 侧，不参与调度决策）
拓扑感知（RDMA/NVLink 域）	PodTopologySpread 只能按 label 打散，不能表达"必须落在同一交换机域"
配额先于节点（两级调度）	只有"Pod → Node"一跳
超卖 / Spot 抢占	Priority + Preemption 可以做，但没有"可超卖资源池"的概念

要满足这些，就得换一套调度模型。

11.1 Sky-Scheduler 的定位：共存而非替换

这一点在代码里写得很直白：

// pkg/scheduler/cache/cache_informers.go:26-38
FilterFunc: func(obj interface{}) bool {
    switch obj := obj.(type) {
    case *v1.Pod:
        if strings.Compare(obj.Spec.SchedulerName, sc.schedulerName) == 0 {
            return true
        }
        return len(obj.Spec.NodeName) > 0
    default:
        return false
    }
},

只收 schedulerName 匹配的 Pod，或者已经绑定了节点的 Pod。前者是它要调度的，后者只用来做资源账本。调度器名由 --scheduler-name 指定（pkg/scheduler/run.go:41）。

所以它和 kube-scheduler 是并存的：只有显式声明了对应 schedulerName 的 Pod 走它，其他 Pod 照常由 kube-scheduler 处理。多调度器共存的机制与坑，前面那篇多调度器共存实践指南已经讲透了。

部署上是三个二进制：

二进制	职责	状态
scheduler	主调度进程，2 副本 + leader election	完整
topology-discovery	探测 RDMA 拓扑，写 NodeTopology CRD	完整（IB 用 Deployment，RoCE 用 DaemonSet）
controller-manager	CRD 控制器	空壳，cmd/controller-manager/main.go 只有 func main() {}

11.2 核心改造：Session / Action / Statement

对比两套框架的抽象，差异一目了然：

// k8s：按扩展点切分的多个接口
type FilterPlugin interface { Filter(ctx, state, pod, node) *Status }
type ScorePlugin  interface { Score(ctx, state, pod, nodeName) (int64, *Status) }
type ReservePlugin interface { Reserve(ctx, state, pod, nodeName) *Status }
// ...

// Sky-Scheduler：只有两个接口（framework/interface.go:27-54）
type Action interface {
    Name() string; Initialize(); Execute(ssn *Session); UnInitialize()
}
type Plugin interface {
    Name() string; OnSessionOpen(ssn *Session); OnSessionClose(ssn *Session)
}

根本差异在"调度单位"和"状态生命周期"：

	kube-scheduler	Sky-Scheduler
调度单位	单个 Pod	Job / Queue / Quota（批）
触发方式	事件驱动（Pod 入队即调度）	周期驱动（wait.Until(runOnce, 1s)）
状态作用域	CycleState，一次调度	Session，整个集群一个周期
插件介入方式	实现固定接口，框架按扩展点调用	Add*Fn 注册回调，Action 遍历时按需调用
变更生效	直接改缓存	Statement 事务累积，最后 Commit / Discard

主循环长这样：

// pkg/scheduler/scheduler.go:102-131（结构归纳）
func (pc *Scheduler) runOnce() {
    // 加锁快照 actions / plugins
    // 重建 conf.EnabledActionMap
    ssn := framework.OpenSession(...)          // :120
    defer framework.CloseSession(ssn)          // :122
    for _, action := range pc.actions {
        action.Execute(ssn)                    // :128 顺序执行
    }
}

OpenSession（framework/framework.go:31-46）做的是：构建所有 map、调 cache.SnapshotForScheduling() 拿快照、过滤无效 Job、然后逐个 plugin.OnSessionOpen(ssn) 让插件往 session 的函数表里注册回调。

注意这里是个"周期性全量重扫"模型：每 1 秒，把整个集群的 Job/Quota/Node 状态快照出来，从头算一遍该把哪些 task 放到哪些节点。这和 kube-scheduler 的"来一个 Pod 调度一个"是完全不同的哲学。

全量重扫的代价是每周期都做 O(集群规模) 的工作；收益是天然的全局视野——gang 调度、配额借还、拓扑成组，这些都需要看到全局才能决策。批调度器选这条路的很多（Volcano、Koordinator 都是），本质上是因为这些约束不是局部的。

11.3 Statement：把调度变成事务

这是整套设计里最值得学的一点。Statement（framework/statement.go，660 行）定义了五种操作：

// statement.go:30-41
const ( Evict = iota; Pipeline; Allocate; AllocateTaskToQuota; AllocateQuota )

每个动作方法做三件事：改内存对象状态 → 触发插件事件回调 → 把操作压进 operations 栈。

以 Allocate 为例（:440-505）：

podVolumes, err := s.ssn.cache.GetPodVolumes(...)   // :441
if err != nil { return err }
if err := s.ssn.cache.AllocateVolumes(...); err != nil {
    defer s.ssn.cache.RevertVolumes(...)            // :450-454 失败即回滚卷
    return err
}
// 改内存：job 状态 → Allocated，node.AddTask，pod.Spec.NodeName = hostname（:484）
s.operations = append(s.operations, operation{name: Allocate, task: task, hostname: hostname})

然后由 Action 在合适的时机决定提交还是丢弃：

// statement.go:618-660  Commit：正序执行真实副作用
for _, op := range s.operations {
    if err := s.evict/pipeline/allocate/...; err != nil {
        discardOperations = append(discardOperations, ...)
    }
}
// statement.go:582-615  Discard：逆序回滚
for i := len(s.operations) - 1; i >= 0; i-- {
    switch s.operations[i].name {
    case Evict: s.unevict(...)
    ...
    }
}

三个设计点：

其一，试算不落盘。 调度过程中所有的状态变更都只发生在内存快照上，直到 Commit 才写 API Server。Pipeline 操作的真实执行甚至是 no-op（:215-216）——它纯粹是"占位预演"，表示"这个 task 打算放这里"，但不产生任何副作用。

其二，失败可整体回滚。 如果 Job 的某个 task 放不下，Discard 会把之前所有 task 的操作逆序撤销，Job 回到干净状态，下个周期重来。这解决了批调度里最烦人的部分成功问题：不会出现"一个 gang 的 3 个 task 调度成功、2 个失败，成功的那 3 个占着资源干等"。

其三，卷绑定被拉进了同一个事务。 statement.go:441-454 在 Allocate 里同步做 GetPodVolumes → AllocateVolumes，失败就 RevertVolumes。对比 kube-scheduler 的 VolumeBinding 插件需要靠 Reserve/Unreserve 两个扩展点来保证可回滚性，这里因为有了事务栈，可以直接同步做——框架能力决定了插件能有多简单。

11.4 Action 流水

Action 的顺序不是硬编码在 Go 里，而是在默认配置 YAML 中（pkg/scheduler/utils.go:14-59）：

actions:
- allocateJob      # Job → Quota：给 task 选配额
- allocateZone     # 拓扑：圈定 zone
- allocateQuota    # Quota → Node：给配额选节点
- bind             # 真正把 pod 绑到节点
- reclaim          # 抢占
- backfill         # spot 回填
- releaseQuota     # 释放未用的配额
plugins:
- name: rescheduling
- name: jobqueue
- ...
- name: topology
- name: storage

reclaim 和 backfill 的分工很有意思，两者服务完全不同的队列：

// actions/reclaim/reclaim.go:51-53
if queue.Reclaimable() { continue }   // 跳过 spot 队列
// actions/backfill/backfill.go:101-103
if queue.Queue.Spec.Type != v1alpha1.QueueTypeSpot { continue }   // 只服务 spot 队列

• reclaim：服务在线/配额队列。走"驱逐低优 task + Pipeline 预占节点"，是抢占路径（reclaim/preempt.go:72 的 stmt.Evict，:84-88 的 stmt.Pipeline）。
• backfill：只服务 Spot 队列，直接在节点上找空隙 stmt.Allocate（backfill.go:235），不驱逐、不改配额绑定。

把"抢占"和"填缝"拆成两个 action 的好处是策略可以分别配置权重，坏处是要多维护一条流水线。

11.5 插件参数的组合：表达式引擎

kube-scheduler 里，多个 Score 插件的分数是加权求和（framework.go:1177），这个组合方式是写死的。

Sky-Scheduler 把它做成了可配置的表达式（pkg/scheduler/expr/）：

// exprs.go:28
OpAnd / OpOr / OpNot / OpSum / OpMin / OpMax
OpIntersect / OpUnion / OpJoint / OpFirstNotPass

session_plugin.go:55-63 的 DefaultExpressions 把插件名拼成表达式串，比如 JobReady 默认是 OpAnd（所有插件都说 ready 才 ready），JobPipelined 默认是 OpMin（取最小值），NOrder 默认是 OpSum（分数求和）。

这是"不改代码就能调策略"的实现方式：想让某个插件在 JobReady 判定里有一票否决权，把 and 换成 firstNotPass 就行。对比 kube-scheduler 需要重新编译插件，这是自研框架能拿到的额外灵活度。

11.6 拓扑插件：这个项目最值得看的部分

pkg/scheduler/plugins/topology/ 和它的 algos/ 子目录，是 Sky-Scheduler 相对 Volcano 的核心增量。

拓扑数据的来源是独立的 discovery 进程：

topology-discovery（每节点/全局）           scheduler
  ├─ RoCE: DaemonSet，lldpcli 查 LLDP        ├─ 读 NodeTopology CRD
  │   得到 nic → 交换机名                     ├─ genRDMATopoKey 生成拓扑 key
  └─ IB:   Deployment（全局视角）             └─ topology 插件按 key 分组
      ibnetdiscover + python 解析
           │
           └──► 写 NodeTopology CRD + node label "zone"

api/topology.go:39-45 生成拓扑 key 的优先级是：annotation → node label zone → 由交换机名生成。nodetopology.go:84-104 的 genRDMAZoneName 取 IB（优先）或 RoCE 的所有 SwitchName，排序后 md5 作为 zone 名——zone 就是"挂在同一组交换机下的节点集合"。

algos/ 里有三种算法实现：

算法	文件	思路
continuity	algos/continuity.go	节点索引连续（delta==1）的切分为一个 zone，fix() 在 :70-103
rdma	algos/namedzone.go	按 help.ZoneName 分组，同 zone 同名才算有效（ValidateZone :186-201）
factory	algos/factory.go	SchedulingHelper 提供 NodeIndex/ZoneName/IsZoneMember/ScoreZone

选择算法的接口是统一的 TopologySchedulingAlgo（algos/base.go:20-28）：Init / PredicateForJob / PredicateForTask / AntiAffinityScore / Allocate / DeAllocate / ValidateZone。

IsZoneMember（factory.go:77-79）的判定条件是：节点 ready、无 quota 占用、无 taint——即"这个节点干净且可用"。AntiAffinityScore 则实现打散：namedzone.go:157-165 用的是 1/zone.Len()，zone 越大分越低，鼓励把任务铺开。

11.7 可观测性：三个 HTTP 服务

自研调度器相比 kube-scheduler 的一个明显优势是诊断能力（pkg/scheduler/service/）：

端点	用途
GET /diagnose/job/:ns/:name	为什么这个 Job 没调度上——人类可读的原因
GET /snapshot/dump	导出调度单元的完整视图（Pod/Job/Node/Quota/Queue/Cluster 六类）
GET /simulate/scheduling?object=quota&target=node	模拟 quota→node 分配，看会不会冲突

/diagnose 尤其实用。kube-scheduler 里"Pod 为什么 Pending"需要看 PodScheduled condition 的 message，再结合事件推断；这里直接给结论。实现思路在 diagnose/diagnose.go:44-55 的 TryAllocate：用 req.LessEqualWithResourcesName(Idle) 逐个候选单位试算，装不下就记录原因。

还有个小设计值得一提：middlewares.go:12-32 的 leaderProxy 在非 leader 时把请求反向代理到 leader。因为只有 leader 在跑调度、才有实时快照，直接返回错误对使用者不友好。

/simulate 用的是加权二分图匹配（simulate/bipartite_graph.go，201 行自研实现）：把 quota→node 分配建模成二分图，GPU 数作权重，已绑定的用 Link(..., true) 强制连边，然后求最大匹配。这比"逐个贪心"更接近理论最优。

11.8 一个值得注意的工程细节：依赖版本错配

看 go.mod 会发现版本并不统一：

k8s.io/api            v0.32.1
k8s.io/client-go      v0.32.1
k8s.io/kubernetes     v1.28.10     ← 落后
k8s.io/component-helpers v0.28.10  ← 落后
k8s.io/apiserver      v0.30.3      ← 落后

原因是 pkg/kubernetes/volume/scheduling/ 这套卷调度代码是从上游 VolumeBinding 插件整体搬运改写的（volume_binding.go:1-15 保留了 Apache 头），它依赖 k8s.io/kubernetes 里 1.28 版本的 scheduler framework API。主流程用 client-go 0.32，搬运来的代码用 1.28 的框架——两套 k8s 依赖并存。

这是"fork 上游代码"这种做法的典型代价：改起来快，但从此背上了一个版本升级的包袱。值不值，取决于这段代码的定制程度——这里它被改造成了 VolumeBinder 接口（cache/default.go:156-208），嵌进 Statement 事务，定制程度已经不低。

---

十二、两种调度模型对照

把 kube-scheduler 和 Sky-Scheduler 放在一起看：

维度	kube-scheduler	Sky-Scheduler
调度单位	Pod	Job（PodGroup）/ Task
触发	事件驱动，Pod 入队即调度	周期驱动，默认 1s 全量重扫
主循环	wait.UntilWithContext(ScheduleOne, 0)，单 goroutine	wait.Until(runOnce, 1s)，单 goroutine
绑定	异步 goroutine（bindingCycle）	异步（cache.Bind 里 go Binder.Bind）
队列	activeQ / backoffQ / unschedulablePods 三级	无 Pod 队列；JobQueue 是业务语义的 FIFO
状态	CycleState（单 Pod，sync.Map）	Session（全集群，一周期一份快照）
事务	无（靠 Assume + Unreserve 回滚）	Statement 事务栈，Commit / Discard
插件	实现固定扩展点接口	Add*Fn 注册回调 + 表达式组合
配额	无（ResourceQuota 在 API Server 侧）	Quota 是一等调度单位，两级调度
抢占	PostFilter + DefaultPreemption	reclaim action + stmt.Evict
拓扑	PodTopologySpread（label 维度）	topology 插件 + RDMA zone + 三种算法
诊断	靠 condition / event 推断	/diagnose、/snapshot、/simulate
部署关系	默认调度器	按 schedulerName 共存

一句话总结差异：kube-scheduler 优化的是"单个 Pod 的调度延迟"，Sky-Scheduler 优化的是"一批任务的整体调度质量"。前者的敌人是节点数（所以有采样、有并行、有 assume），后者的敌人是约束的组合爆炸（所以有事务、有配额层、有拓扑算法）。目标函数不同，架构就必然不同。

---

十三、总结

回到开头说的两个难点。

控制器的难点在并发，而 K8s 的解法可以概括成三句话：

1. 事件只用来提示，状态才是真相——所以回调里只做 queue.Add(key)，逻辑在 worker 里重读状态。
2. 写操作要记账，账要能超时——expectations 保证不会因缓存延迟而重复操作，5 分钟超时保证不会因事件丢失而永久卡死。
3. 批量操作要限速和慢启动——BurstReplicas=500 和 slowStartBatch 的 1→2→4 是同一类保护。

调度器的难点在契约，而 kube-scheduler 的解法是用扩展点把"策略"和"机制"彻底分开：

• 框架只负责什么时候调用谁（PreFilter → Filter → PostFilter → Score → Reserve → Permit → PreBind → Bind），插件只负责单个判断。
• Unschedulable vs UnschedulableAndUnresolvable 的分级、QueueingHint 的按需唤醒、percentageOfNodesToScore 的采样——都是在用有界延迟换近似最优。
• Assume + cleanupAssumedPods 是这个取舍最赤裸的表达：先假定成功，出错再收拾。

而自研调度器的价值，恰恰在于它展示了当目标函数变了，框架该怎么重写：把调度单位从 Pod 提到 Job，把状态从 CycleState 提到 Session，把直接改缓存换成 Statement 事务。kube-scheduler 的框架很优雅，但它的优雅是为"单 Pod、低延迟"这个目标服务的；换一个目标，优雅的地方就变成了束缚。

读源码的收益，大概就在这一层——不是记住某个函数在哪一行，而是理解某个设计在为什么样的约束服务。

---

参考

• Kubernetes v1.32.0-alpha.0（commit 1bbe775d5ff）源码
• Sky-Scheduler 源码（pkg/scheduler、pkg/topology）
• 延伸阅读：Sky-Scheduler：面向 AI 训练的拓扑感知与 Quota 两级调度器、Kubernetes Informer 机制详解、Volcano 调度器、一个集群，多个调度器：Kubernetes 多调度器共存实践指南