【kubernetes v1.21】（四）kube-scheduler 超深度源码分析
·
kube-scheduler 超深度源码分析
基于 Kubernetes 源码
cmd/kube-scheduler/、pkg/scheduler/、staging/src/k8s.io/kube-scheduler/进行逐行级专业分析
一、模块定位
1.1 业务职责
kube-scheduler 是 Kubernetes 控制平面的核心组件之一，负责将未调度的 Pod 绑定到最合适的 Node。它是 Kubernetes 默认调度器的参考实现，其核心职责可概括为：
| 职责 | 说明 |
|---|---|
| Pod 调度 | 监听未调度 Pod，为其选择最优 Node |
| 资源匹配 | 基于 CPU/Memory/存储等资源约束过滤节点 |
| 策略执行 | 通过插件框架执行 Filter（过滤）、Score（打分）、Bind（绑定）等策略 |
| 抢占调度 | 当高优先级 Pod 无法调度时，驱逐低优先级 Pod |
| 扩展支持 | 通过 Scheduling Framework 插件机制和 Extender HTTP API 支持自定义调度逻辑 |
| 多调度器 | 通过 Profile 机制支持多个调度器实例共存 |
1.2 在系统中的位置
kube-scheduler 位于 Kubernetes 控制平面，与 kube-apiserver、kube-controller-manager 并列，是三大核心控制循环之一：
┌─────────────────────────────────────────────────────────┐
│ Kubernetes 控制平面 │
│ ┌──────────────┐ ┌───────────────────┐ ┌───────────┐ │
│ │ kube-apiserver│◄─►│ kube-scheduler │◄─►│ etcd │ │
│ └──────┬───────┘ └─────────▲─────────┘ └───────────┘ │
│ │ │ │
│ ┌──────▼───────┐ ┌───────┴────────┐ │
│ │ controller │ │ Informers │ │
│ │ manager │ │ (Pod/Node/PV) │ │
│ └──────────────┘ └────────────────┘ │
└─────────────────────────────────────────────────────────┘
kube-scheduler 只读集群状态（通过 Informer），只写 Pod 的 .spec.nodeName 和 Binding 对象（通过 apiserver），不直接操作 Node 或其他资源。
二、模块整体结构
2.1 核心类结构
// ============ 顶层调度器对象 ============
type Scheduler struct {
SchedulerCache internalcache.Cache // 调度器缓存，存储 Pod/Node 快照
Algorithm core.ScheduleAlgorithm // 调度算法接口
NextPod func() *framework.QueuedPodInfo // 从队列取下一个 Pod
Error func(*framework.QueuedPodInfo, error) // 错误处理
StopEverything <-chan struct{} // 全局停止信号
SchedulingQueue internalqueue.SchedulingQueue // 调度队列
Profiles profile.Map // 调度配置文件映射(按 schedulerName 索引)
client clientset.Interface // kube-apiserver 客户端
}
// ============ 调度算法接口 ============
type ScheduleAlgorithm interface {
Schedule(context.Context, framework.Framework, *framework.CycleState, *v1.Pod) (ScheduleResult, error)
Extenders() []framework.Extender
}
// ============ 通用调度器实现 ============
type genericScheduler struct {
cache internalcache.Cache
extenders []framework.Extender
nodeInfoSnapshot *internalcache.Snapshot
percentageOfNodesToScore int32
nextStartNodeIndex int // 轮转起点，保证公平性
}
// ============ Scheduling Framework 核心 ============
type frameworkImpl struct {
registry Registry // 插件注册表
snapshotSharedLister framework.SharedLister // 快照共享列表
waitingPods *waitingPodsMap // Permit 阶段等待中的 Pod
pluginNameToWeightMap map[string]int // Score 插件权重
queueSortPlugins []framework.QueueSortPlugin // 各扩展点插件列表
preFilterPlugins []framework.PreFilterPlugin
filterPlugins []framework.FilterPlugin
postFilterPlugins []framework.PostFilterPlugin
preScorePlugins []framework.PreScorePlugin
scorePlugins []framework.ScorePlugin
reservePlugins []framework.ReservePlugin
preBindPlugins []framework.PreBindPlugin
bindPlugins []framework.BindPlugin
postBindPlugins []framework.PostBindPlugin
permitPlugins []framework.PermitPlugin
// ... 其他字段
}
// ============ 调度器缓存 ============
type schedulerCache struct {
mu sync.RWMutex
assumedPods sets.String // 假设已调度 Pod 集合
podStates map[string]*podState // Pod 状态映射
nodes map[string]*nodeInfoListItem // Node 双向链表
headNode *nodeInfoListItem // 最近更新的 Node（链表头）
nodeTree *nodeTree // Node 扁平树（用于轮转）
imageStates map[string]*imageState // 镜像状态映射
}
// ============ 优先级队列 ============
type PriorityQueue struct {
framework.PodNominator
activeQ *heap.Heap // 活跃队列（按优先级排序的堆）
podBackoffQ *heap.Heap // 退避队列（按退避完成时间排序）
unschedulableQ *UnschedulablePodsMap // 不可调度队列
schedulingCycle int64 // 调度周期号（Pop 时递增）
moveRequestCycle int64 // 最近 MoveAll 请求的周期号
clusterEventMap map[framework.ClusterEvent]sets.String // 集群事件→插件映射
}
2.2 核心接口定义
// Framework 接口 —— 调度框架的统一入口
type Framework interface {
Handle
QueueSortFunc() LessFunc
RunPreFilterPlugins(ctx, state, pod) *Status
RunFilterPlugins(ctx, state, pod, nodeInfo) PluginToStatus
RunPostFilterPlugins(ctx, state, pod, nodeToStatusMap) (*PostFilterResult, *Status)
RunPreScorePlugins(ctx, state, pod, nodes) *Status
RunScorePlugins(ctx, state, pod, nodes) (PluginToNodeScores, *Status)
RunReservePluginsReserve(ctx, state, pod, nodeName) *Status
RunReservePluginsUnreserve(ctx, state, pod, nodeName)
RunPermitPlugins(ctx, state, pod, nodeName) *Status
WaitOnPermit(ctx, pod) *Status
RunPreBindPlugins(ctx, state, pod, nodeName) *Status
RunBindPlugins(ctx, state, pod, nodeName) *Status
RunPostBindPlugins(ctx, state, pod, nodeName)
RunFilterPluginsWithNominatedPods(ctx, state, pod, info) *Status
HasFilterPlugins() bool
HasPostFilterPlugins() bool
HasScorePlugins() bool
ProfileName() string
}
// 插件接口体系（每个扩展点对应一个接口）
type Plugin interface { Name() string }
type QueueSortPlugin interface { Plugin; Less(*QueuedPodInfo, *QueuedPodInfo) bool }
type PreFilterPlugin interface { Plugin; PreFilter(ctx, state, pod) *Status; PreFilterExtensions() PreFilterExtensions }
type FilterPlugin interface { Plugin; Filter(ctx, state, pod, nodeInfo) *Status }
type PostFilterPlugin interface { Plugin; PostFilter(ctx, state, pod, nodeToStatusMap) (*PostFilterResult, *Status) }
type PreScorePlugin interface { Plugin; PreScore(ctx, state, pod, nodes) *Status }
type ScorePlugin interface { Plugin; Score(ctx, state, pod, nodeName) (int64, *Status); ScoreExtensions() ScoreExtensions }
type ReservePlugin interface { Plugin; Reserve(ctx, state, pod, nodeName) *Status; Unreserve(ctx, state, pod, nodeName) }
type PreBindPlugin interface { Plugin; PreBind(ctx, state, pod, nodeName) *Status }
type BindPlugin interface { Plugin; Bind(ctx, state, pod, nodeName) *Status }
type PostBindPlugin interface { Plugin; PostBind(ctx, state, pod, nodeName) }
type PermitPlugin interface { Plugin; Permit(ctx, state, pod, nodeName) (*Status, time.Duration) }
// Extender 接口 —— HTTP 扩展器
type Extender interface {
Name() string
IsInterested(pod *v1.Pod) bool
Filter(pod *v1.Pod, nodes []*v1.Node) (feasibleNodes []*v1.Node, failedMap, failedAndUnresolvableMap map[string]string, err error)
Prioritize(pod *v1.Pod, nodes []*v1.Node) (hostPriorities *extenderv1.HostPriorityList, weight int64, err error)
Bind(binding *v1.Binding) error
IsBinder() bool
IsIgnorable() bool
}
2.3 依赖注入关系
核心依赖注入路径：
- Options → Config → CompletedConfig：命令行参数 → 中间配置 → 最终配置
- scheduler.New 是调度器工厂函数，通过
Option函数式选项模式注入：WithProfiles— 调度配置文件WithAlgorithmSource— 算法来源（Provider 或 Policy）WithPercentageOfNodesToScore— 打分节点比例WithFrameworkOutOfTreeRegistry— 树外插件注册表WithPodMaxBackoffSeconds/WithPodInitialBackoffSeconds— 退避参数WithExtenders— HTTP 扩展器配置WithParallelism— 并行度
- Configurator 将上述配置组装为
Scheduler实例 - profile.NewMap 为每个 Profile 创建
frameworkImpl，注入 ClientSet、InformerFactory、SnapshotSharedLister、PodNominator 等
2.4 核心方法清单
| 类/结构体 | 方法 | 作用 |
|---|---|---|
Scheduler | Run(ctx) | 启动调度主循环 |
Scheduler | scheduleOne(ctx) | 单个 Pod 调度全流程 |
Scheduler | assume(pod, host) | 假设 Pod 已调度（乐观更新缓存） |
Scheduler | bind(ctx, fwk, pod, node, state) | 绑定 Pod 到 Node |
Scheduler | skipPodSchedule(fwk, pod) | 判断是否跳过调度 |
genericScheduler | Schedule(ctx, fwk, state, pod) | 执行调度算法（Filter+Score+Select） |
genericScheduler | findNodesThatFitPod(ctx, fwk, state, pod) | 执行 Filter 阶段 |
genericScheduler | prioritizeNodes(ctx, fwk, state, pod, nodes) | 执行 Score 阶段 |
genericScheduler | selectHost(nodeScoreList) | 从打分结果选择最终节点（蓄水池抽样） |
frameworkImpl | RunPreFilterPlugins | 运行 PreFilter 扩展点 |
frameworkImpl | RunFilterPlugins | 运行 Filter 扩展点 |
frameworkImpl | RunScorePlugins | 运行 Score 扩展点（并行） |
frameworkImpl | RunPermitPlugins | 运行 Permit 扩展点 |
frameworkImpl | RunBindPlugins | 运行 Bind 扩展点 |
frameworkImpl | RunFilterPluginsWithNominatedPods | 含提名 Pod 的 Filter（调度+抢占双场景） |
PriorityQueue | Add(pod) | 新 Pod 入活跃队列 |
PriorityQueue | AddUnschedulableIfNotPresent | 不可调度 Pod 入 unschedulableQ |
PriorityQueue | Pop() | 取出最高优先级 Pod |
PriorityQueue | MoveAllToActiveOrBackoffQueue | 批量移动 Pod 到活跃/退避队列 |
schedulerCache | AssumePod(pod) | 假设 Pod 已调度 |
schedulerCache | ForgetPod(pod) | 撤回假设 |
schedulerCache | UpdateSnapshot(snapshot) | 更新快照（增量更新） |
DefaultPreemption | PostFilter(ctx, state, pod, m) | 抢占逻辑入口 |
2.5 数据流入流出
数据流入：
- Informer 监听 kube-apiserver 的 Pod/Node/PV/PVC/Service/StorageClass/CSINode 事件
- 未调度 Pod →
SchedulingQueue.Add() - 已调度 Pod →
SchedulerCache.AddPod() - Node 变更 →
SchedulerCache.AddNode()/UpdateNode()
数据流出：
- Pod 绑定 → 通过
client.CoreV1().Pods(ns).Bind()写入 apiserver - Pod 状态更新 →
updatePod()更新 PodCondition 和 NominatedNodeName - 事件记录 →
EventRecorder.Eventf()
三、核心业务逻辑深度解析
3.1 Scheduler 启动流程
逐行解析 scheduler.go — main() 函数：
func main() {
rand.Seed(time.Now().UnixNano()) // 1. 初始化随机种子（用于 selectHost 蓄水池抽样）
pflag.CommandLine.SetNormalizeFunc(cliflag.WordSepNormalizeFunc) // 2. 标准化 flag 名（如 --algorithm-provider → algorithm_provider）
command := app.NewSchedulerCommand() // 3. 创建 cobra 命令
logs.InitLogs() // 4. 初始化日志
defer logs.FlushLogs() // 5. 确保退出时刷日志
if err := command.Execute(); err != nil { // 6. 执行命令
os.Exit(1)
}
}
逐行解析 server.go — Run() 函数：
func Run(ctx context.Context, cc *CompletedConfig, sched *scheduler.Scheduler) error {
klog.V(1).Infof("Starting Kubernetes Scheduler version %+v", version.Get()) // 打印版本
// 注册 configz
if cz, err := configz.New("componentconfig"); err == nil {
cz.Set(cc.ComponentConfig)
}
// 启动事件广播器
cc.EventBroadcaster.StartRecordingToSink(ctx.Done())
// 设置健康检查
var checks []healthz.HealthChecker
if cc.ComponentConfig.LeaderElection.LeaderElect {
checks = append(checks, cc.LeaderElection.WatchDog)
}
// waitingForLeader 通道用于判断是否是 leader
waitingForLeader := make(chan struct{})
isLeader := func() bool {
select {
case _, ok := <-waitingForLeader: // 通道关闭 = 成为 leader
return !ok
default:
return false
}
}
// 启动 HTTP 服务器（insecure/secure/metrics）
if cc.InsecureServing != nil {
handler := buildHandlerChain(newHealthzHandler(...), nil, nil)
cc.InsecureServing.Serve(handler, 0, ctx.Done())
}
if cc.SecureServing != nil {
handler := buildHandlerChain(newHealthzHandler(...), cc.Authentication.Authenticator, cc.Authorization.Authorizer)
cc.SecureServing.Serve(handler, 0, ctx.Done())
}
// ★ 关键：启动所有 Informer 并等待缓存同步
cc.InformerFactory.Start(ctx.Done())
cc.InformerFactory.WaitForCacheSync(ctx.Done())
// ★ Leader 选举
if cc.LeaderElection != nil {
cc.LeaderElection.Callbacks = leaderelection.LeaderCallbacks{
OnStartedLeading: func(ctx context.Context) {
close(waitingForLeader) // 关闭通道，表示成为 leader
sched.Run(ctx) // ★ 启动调度主循环
},
OnStoppedLeading: func() {
klog.Fatalf("leaderelection lost")
},
}
leaderElector, _ := leaderelection.NewLeaderElector(*cc.LeaderElection)
leaderElector.Run(ctx) // 阻塞，参与选举
return fmt.Errorf("lost lease")
}
// 不启用选举时直接运行
close(waitingForLeader)
sched.Run(ctx)
return fmt.Errorf("finished without leader elect")
}
3.2 Scheduling Framework 架构
Scheduling Framework 的核心设计理念：
- 扩展点（Extension Points）：将调度流程拆分为 11 个扩展点，每个扩展点可注册多个插件
- 插件化：所有调度逻辑通过插件实现，In-Tree 插件通过
NewInTreeRegistry()注册，Out-Of-Tree 插件通过WithPlugin()注入 - Profile：每个调度器名对应一个 Profile，Profile 定义启用的插件和配置
- CycleState：每次调度周期创建一个
CycleState，插件间通过它共享数据
3.3 调度器扩展点详解
3.4 Pod 调度全流程
逐行解析 scheduleOne() —— 最核心的方法：
func (sched *Scheduler) scheduleOne(ctx context.Context) {
// 1. 从队列获取下一个 Pod（阻塞式）
podInfo := sched.NextPod() // 底层调用 PriorityQueue.Pop()
if podInfo == nil || podInfo.Pod == nil {
return // 队列已关闭
}
pod := podInfo.Pod
// 2. 根据 pod.Spec.SchedulerName 找到对应的 Framework Profile
fwk, err := sched.frameworkForPod(pod)
if err != nil {
return // 不应发生，入队时已校验
}
// 3. 跳过正在删除或已假设的 Pod
if sched.skipPodSchedule(fwk, pod) {
return
}
// 4. 创建调度周期状态（跨插件共享数据）
state := framework.NewCycleState()
state.SetRecordPluginMetrics(rand.Intn(100) < pluginMetricsSamplePercent) // 10% 采样
schedulingCycleCtx, cancel := context.WithCancel(ctx)
defer cancel()
// 5. ★ 执行调度算法：Filter → Score → SelectHost
scheduleResult, err := sched.Algorithm.Schedule(schedulingCycleCtx, fwk, state, pod)
if err != nil {
// 6. 调度失败处理
nominatedNode := ""
if fitError, ok := err.(*framework.FitError); ok {
// 6a. 尝试抢占（PostFilter）
if fwk.HasPostFilterPlugins() {
result, status := fwk.RunPostFilterPlugins(ctx, state, pod, fitError.Diagnosis.NodeToStatusMap)
if status.IsSuccess() && result != nil {
nominatedNode = result.NominatedNodeName
}
}
metrics.PodUnschedulable(fwk.ProfileName(), ...)
} else if err == core.ErrNoNodesAvailable {
metrics.PodUnschedulable(...)
} else {
metrics.PodScheduleError(...)
}
sched.recordSchedulingFailure(fwk, podInfo, err, v1.PodReasonUnschedulable, nominatedNode)
return
}
// 7. ★ 假设 Pod 已调度（乐观更新缓存，避免等待 Binding 完成）
assumedPodInfo := podInfo.DeepCopy()
assumedPod := assumedPodInfo.Pod
err = sched.assume(assumedPod, scheduleResult.SuggestedHost)
if err != nil {
sched.recordSchedulingFailure(...)
return
}
// 8. ★ Reserve 阶段（如 VolumeBinding 预留卷）
if sts := fwk.RunReservePluginsReserve(schedulingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost); !sts.IsSuccess() {
fwk.RunReservePluginsUnreserve(...)
sched.SchedulerCache.ForgetPod(assumedPod)
sched.recordSchedulingFailure(...)
return
}
// 9. ★ Permit 阶段（可能延迟绑定）
runPermitStatus := fwk.RunPermitPlugins(schedulingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
if runPermitStatus.Code() != framework.Wait && !runPermitStatus.IsSuccess() {
// Permit 拒绝 → Unreserve → Forget → 记录失败
fwk.RunReservePluginsUnreserve(...)
sched.SchedulerCache.ForgetPod(assumedPod)
sched.recordSchedulingFailure(...)
return
}
// 10. ★ 异步绑定周期（goroutine）
go func() {
bindingCycleCtx, cancel := context.WithCancel(ctx)
defer cancel()
// 10a. 等待 Permit（阻塞直到所有 Permit 插件 Allow 或超时/拒绝）
waitOnPermitStatus := fwk.WaitOnPermit(bindingCycleCtx, assumedPod)
if !waitOnPermitStatus.IsSuccess() {
fwk.RunReservePluginsUnreserve(...)
sched.SchedulerCache.ForgetPod(assumedPod)
sched.recordSchedulingFailure(...)
return
}
// 10b. PreBind 阶段（如 VolumeBinding 绑定 PVC-PV）
preBindStatus := fwk.RunPreBindPlugins(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
if !preBindStatus.IsSuccess() {
fwk.RunReservePluginsUnreserve(...)
sched.SchedulerCache.ForgetPod(assumedPod)
sched.recordSchedulingFailure(...)
return
}
// 10c. ★ 执行绑定
err := sched.bind(bindingCycleCtx, fwk, assumedPod, scheduleResult.SuggestedHost, state)
if err != nil {
// 绑定失败 → Unreserve → Forget → 记录失败
fwk.RunReservePluginsUnreserve(...)
sched.SchedulerCache.ForgetPod(assumedPod)
sched.recordSchedulingFailure(...)
} else {
// 绑定成功
metrics.PodScheduled(...)
// 10d. PostBind 阶段（通知、清理）
fwk.RunPostBindPlugins(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
}
}()
}
关键设计点：
-
乐观假设（Optimistic Assumption）：
assume()在绑定前将 Pod 写入缓存，使得后续调度可以感知该 Pod 的资源占用，避免资源超额分配。绑定成功则保留，失败则ForgetPod()回滚。 -
Scheduling Cycle vs Binding Cycle：
- Scheduling Cycle（同步）：从
NextPod()到Permit，在主 goroutine 中串行执行 - Binding Cycle（异步）：
WaitOnPermit→PreBind→Bind→PostBind，在单独的 goroutine 中执行，不阻塞下一个 Pod 的调度
- Scheduling Cycle（同步）：从
-
错误恢复路径：任何阶段失败都执行
Unreserve→ForgetPod→recordSchedulingFailure，保证缓存和插件状态的一致性。
3.5 Filter 过滤阶段深度解析
findNodesThatPassFilters 关键实现解析：
func (g *genericScheduler) findNodesThatPassFilters(...) ([]*v1.Node, error) {
// 计算需要找到多少个可行节点
numNodesToFind := g.numFeasibleNodesToFind(int32(len(nodes)))
feasibleNodes := make([]*v1.Node, numNodesToFind) // 预分配
errCh := parallelize.NewErrorChannel()
var feasibleNodesLen int32
ctx, cancel := context.WithCancel(ctx)
// ★ 并行检查每个节点
checkNode := func(i int) {
// 从 nextStartNodeIndex 开始轮转，保证公平性
nodeInfo := nodes[(g.nextStartNodeIndex+i)%len(nodes)]
// ★ 运行 Filter 插件（含提名 Pod 的两轮过滤）
status := fwk.RunFilterPluginsWithNominatedPods(ctx, state, pod, nodeInfo)
if status.IsSuccess() {
length := atomic.AddInt32(&feasibleNodesLen, 1)
if length > numNodesToFind {
cancel() // ★ 找够了，提前终止
atomic.AddInt32(&feasibleNodesLen, -1)
} else {
feasibleNodes[length-1] = nodeInfo.Node()
}
} else {
statusesLock.Lock()
diagnosis.NodeToStatusMap[nodeInfo.Node().Name] = status
diagnosis.UnschedulablePlugins.Insert(status.FailedPlugin())
statusesLock.Unlock()
}
}
// ★ 使用 Parallelizer 并行执行
fwk.Parallelizer().Until(ctx, len(nodes), checkNode)
// 更新轮转起点
g.nextStartNodeIndex = (g.nextStartNodeIndex + processedNodes) % len(nodes)
return feasibleNodes[:feasibleNodesLen], nil
}
RunFilterPluginsWithNominatedPods — 含提名 Pod 的两轮过滤机制：
这是调度器的一个精妙设计。当节点上有 >= 当前 Pod 优先级的提名 Pod 时，需要做两轮过滤：
- 第一轮：将提名 Pod 加入
nodeInfo和PreFilter state，模拟它们已调度的情况运行 Filter - 第二轮：仅当第一轮通过时，移除提名 Pod 后再运行 Filter
原因：
- 资源类 Filter（如 NodeResourcesFit）在提名 Pod 存在时更可能失败（资源更多被占用）
- 亲和性 Filter（如 InterPodAffinity）在提名 Pod 不存在时更可能失败（缺少亲和目标）
- 两次都通过才算真正可行，这是保守决策
func (f *frameworkImpl) RunFilterPluginsWithNominatedPods(ctx, state, pod, info) *Status {
podsAdded := false
for i := 0; i < 2; i++ {
stateToUse := state
nodeInfoToUse := info
if i == 0 {
// 第一轮：加入 >= 优先级的提名 Pod
podsAdded, stateToUse, nodeInfoToUse, _ = addNominatedPods(ctx, f, pod, state, info)
} else if !podsAdded || !status.IsSuccess() {
break // 无提名 Pod 或第一轮失败 → 跳过第二轮
}
statusMap := f.RunFilterPlugins(ctx, stateToUse, pod, nodeInfoToUse)
status = statusMap.Merge()
}
return status
}
numFeasibleNodesToFind — 自适应节点数量策略：
func (g *genericScheduler) numFeasibleNodesToFind(numAllNodes int32) int32 {
if numAllNodes < 100 || g.percentageOfNodesToScore >= 100 {
return numAllNodes // 小集群或显式100%，检查所有节点
}
adaptivePercentage := g.percentageOfNodesToScore
if adaptivePercentage <= 0 {
// 默认：50% - nodes/125，最低 5%，确保至少 100 个
basePercentageOfNodesToScore := int32(50)
adaptivePercentage = basePercentageOfNodesToScore - numAllNodes/125
if adaptivePercentage < 5 {
adaptivePercentage = 5
}
}
numNodes = numAllNodes * adaptivePercentage / 100
if numNodes < 100 {
return 100 // 至少检查 100 个节点
}
return numNodes
}
3.6 Score 打分阶段深度解析
RunScorePlugins 三阶段并行：
func (f *frameworkImpl) RunScorePlugins(ctx, state, pod, nodes) (PluginToNodeScores, *Status) {
pluginToNodeScores := make(PluginToNodeScores, len(f.scorePlugins))
for _, pl := range f.scorePlugins {
pluginToNodeScores[pl.Name()] = make(NodeScoreList, len(nodes))
}
// ★ Phase 1: 并行对每个节点调用每个插件的 Score 方法
f.Parallelizer().Until(ctx, len(nodes), func(index int) {
for _, pl := range f.scorePlugins {
s, status := f.runScorePlugin(ctx, pl, state, pod, nodes[index].Name)
pluginToNodeScores[pl.Name()][index] = NodeScore{Name: nodes[index].Name, Score: s}
}
})
// ★ Phase 2: 并行对每个插件调用 NormalizeScore
f.Parallelizer().Until(ctx, len(f.scorePlugins), func(index int) {
pl := f.scorePlugins[index]
if pl.ScoreExtensions() != nil {
pl.ScoreExtensions().NormalizeScore(ctx, state, pod, pluginToNodeScores[pl.Name()])
}
})
// ★ Phase 3: 应用权重并验证分数范围
f.Parallelizer().Until(ctx, len(f.scorePlugins), func(index int) {
pl := f.scorePlugins[index]
weight := f.pluginNameToWeightMap[pl.Name()]
for i, nodeScore := range pluginToNodeScores[pl.Name()] {
// 验证分数在 [0, 100] 范围内
if nodeScore.Score > MaxNodeScore || nodeScore.Score < MinNodeScore {
errCh.SendErrorWithCancel(...)
return
}
// ★ 应用权重：最终分数 = 原始分数 * 权重
pluginToNodeScores[pl.Name()][i].Score = nodeScore.Score * int64(weight)
}
})
return pluginToNodeScores, nil
}
分数汇总与 selectHost 蓄水池抽样：
// 汇总所有插件的加权得分
for i := range nodes {
result[i] = NodeScore{Name: nodes[i].Name, Score: 0}
for j := range scoresMap {
result[i].Score += scoresMap[j][i].Score // 各插件加权分数求和
}
}
// ★ selectHost：从同分节点中随机选择（蓄水池抽样算法）
func (g *genericScheduler) selectHost(nodeScoreList NodeScoreList) (string, error) {
maxScore := nodeScoreList[0].Score
selected := nodeScoreList[0].Name
cntOfMaxScore := 1
for _, ns := range nodeScoreList[1:] {
if ns.Score > maxScore {
maxScore = ns.Score
selected = ns.Name
cntOfMaxScore = 1
} else if ns.Score == maxScore {
cntOfMaxScore++
if rand.Intn(cntOfMaxScore) == 0 {
selected = ns.Name // 以 1/cntOfMaxScore 的概率替换
}
}
}
return selected, nil
}
默认插件权重一览（DefaultProvider）：
| 插件 | 权重 | 说明 |
|---|---|---|
| NodeResourcesBalancedAllocation | 1 | CPU/内存均衡分配 |
| ImageLocality | 1 | 镜像本地性 |
| InterPodAffinity | 1 | Pod 间亲和性 |
| NodeResourcesLeastAllocated | 1 | 最少已分配资源（尽量分散） |
| NodeAffinity | 1 | Node 亲和性偏好 |
| NodePreferAvoidPods | 10000 | 避免调度到标记了 preferAvoidPods 的 Node（权重极高） |
| PodTopologySpread | 2 | 拓扑分布约束（权重加倍） |
| TaintToleration | 1 | 污点容忍度偏好 |
3.7 Bind 绑定阶段深度解析
绑定优先级：Extender 的 Bind > Framework 的 Bind 插件。这是因为 Extender 可能管理特定的资源（如外部存储），需要在框架 Bind 之前处理。
DefaultBinder 实现（最简单的绑定逻辑）：
func (b *DefaultBinder) Bind(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) *framework.Status {
binding := &v1.Binding{
ObjectMeta: metav1.ObjectMeta{Name: pod.Name, UID: pod.UID},
Target: v1.ObjectReference{Kind: "Node", Name: nodeName},
}
if err := b.handle.ClientSet().CoreV1().Pods(pod.Namespace).Bind(ctx, binding, metav1.CreateOptions{}); err != nil {
return framework.AsStatus(err)
}
return nil
}
3.8 Scheduler Cache 数据结构
Cache 关键操作解析：
-
AssumePod：将 Pod 标记为"已假设调度"，加入
assumedPods和podStates，更新 NodeInfo 的资源请求。不设置 deadline，bindingFinished = false。 -
FinishBinding：绑定完成后调用，设置
bindingFinished = true和deadline = now + ttl（默认 30s）。assumed Pod 在 TTL 过期后由后台 goroutine 清理。 -
ForgetPod：撤回假设，从
assumedPods和podStates删除，从 NodeInfo 移除 Pod。 -
UpdateSnapshot（增量更新，性能关键路径）：
- 从双向链表头开始遍历，只更新
Generation > snapshotGeneration的 NodeInfo - 通过 Generation 机制避免每次全量 Clone
- 新增/删除节点时重建列表
- 从双向链表头开始遍历，只更新
func (cache *schedulerCache) UpdateSnapshot(nodeSnapshot *Snapshot) error {
cache.mu.Lock()
defer cache.mu.Unlock()
snapshotGeneration := nodeSnapshot.generation
updateAllLists := false
// 从链表头遍历（最近更新的在前）
for node := cache.headNode; node != nil; node = node.next {
if node.info.Generation <= snapshotGeneration {
break // 后续都不会更新，提前退出
}
if np := node.info.Node(); np != nil {
existing, ok := nodeSnapshot.nodeInfoMap[np.Name]
if !ok {
updateAllLists = true // 新增节点
existing = &framework.NodeInfo{}
nodeSnapshot.nodeInfoMap[np.Name] = existing
}
clone := node.info.Clone()
*existing = *clone // 浅拷贝赋值
}
}
// 处理已删除节点、重建列表...
}
3.9 PriorityQueue 调度队列
退避算法：
func (p *PriorityQueue) calculateBackoffDuration(podInfo *QueuedPodInfo) time.Duration {
duration := p.podInitialBackoffDuration // 默认 1s
for i := 1; i < podInfo.Attempts; i++ {
duration = duration * 2 // 指数退避
if duration > p.podMaxBackoffDuration {
return p.podMaxBackoffDuration // 上限 10s
}
}
return duration
}
智能队列移动（SchedulingQueue → ActiveQ 的优化）：
MoveAllToActiveOrBackoffQueue 不再盲目移动所有 Pod，而是通过 clusterEventMap 和 UnschedulablePlugins 做精准匹配：
func (p *PriorityQueue) movePodsToActiveOrBackoffQueue(podInfoList, event) {
for _, pInfo := range podInfoList {
// ★ 如果 Pod 的 UnschedulablePlugins 与事件不相关，跳过
if len(pInfo.UnschedulablePlugins) != 0 && !p.podMatchesEvent(pInfo, event) {
continue
}
// 否则移动到 activeQ 或 backoffQ
}
}
3.10 DefaultPreemption 抢占流程
抢占的五个步骤（preempt 方法）：
- 获取最新 Pod：从 Informer 缓存重新获取 Pod（避免过期数据）
- 检查抢占资格：
PodEligibleToPreemptOthers检查 Pod 是否有权抢占（如已提名节点是否还有空间） - 查找候选节点：
FindCandidates并行对部分节点进行 dry-run 抢占 - Extender 过滤：调用 Extender 的 Preempt 方法进一步筛选
- 选择并执行：
SelectCandidate选最佳候选，PrepareCandidate驱逐受害者并设置 NominatedNodeName
PodEligibleToPreemptOthers 检查条件：
- 如果 Pod 有 NominatedNode，检查该节点上是否已有更高优先级 Pod 占用了空间
- 如果 NominatedNode 的 Filter 通过了（即该节点现在可以直接调度），则不需要抢占
3.11 VolumeBinding 卷绑定调度流程
VolumeBinding 的三阶段设计是必要的：
- PreFilter：识别 Pod 的 PVC 需求，区分 immediate binding 和 delayed binding
- Filter：对每个节点检查 PV 可用性和 NodeAffinity 兼容性
- Reserve：在选中的节点上预留 PV（防止其他 Pod 抢占同一 PV）
- PreBind：在绑定 Pod 之前，先完成 PVC-PV 的 API 绑定
这种设计确保了卷绑定和 Pod 绑定的原子性——如果卷绑定失败，Pod 绑定也不会执行（Unreserve 回滚）。
3.12 事件处理器（eventhandlers.go）
事件处理器是调度器与集群状态同步的桥梁：
func addAllEventHandlers(sched *Scheduler, informerFactory informers.SharedInformerFactory) {
// 已调度 Pod（过滤：assignedPod = true）
informerFactory.Core().V1().Pods().Informer().AddEventHandler(
cache.FilteringResourceEventHandler{
FilterFunc: assignedPod, // 只关注已绑定 Node 的 Pod
Handler: cache.ResourceEventHandlerFuncs{
AddFunc: sched.addPodToCache, // 更新缓存 + 触发亲和性匹配的 Pod 重新调度
UpdateFunc: sched.updatePodInCache,
DeleteFunc: sched.deletePodFromCache, // 移除缓存 + MoveAll
},
},
)
// 未调度 Pod（过滤：!assignedPod && responsibleForPod）
informerFactory.Core().V1().Pods().Informer().AddEventHandler(
cache.FilteringResourceEventHandler{
FilterFunc: func(obj) bool {
return !assignedPod(obj) && responsibleForPod(obj, sched.Profiles)
},
Handler: cache.ResourceEventHandlerFuncs{
AddFunc: sched.addPodToSchedulingQueue,
UpdateFunc: sched.updatePodInSchedulingQueue,
DeleteFunc: sched.deletePodFromSchedulingQueue,
},
},
)
// Node 事件
informerFactory.Core().V1().Nodes().Informer().AddEventHandler(...)
// PV/PVC/StorageClass/CSINode/Service 事件...
}
Node 变更的精准触发机制：
func nodeSchedulingPropertiesChange(newNode, oldNode *v1.Node) string {
if nodeSpecUnschedulableChanged(newNode, oldNode) { return NodeSpecUnschedulableChange }
if nodeAllocatableChanged(newNode, oldNode) { return NodeAllocatableChange }
if nodeLabelsChanged(newNode, oldNode) { return NodeLabelChange }
if nodeTaintsChanged(newNode, oldNode) { return NodeTaintChange }
if nodeConditionsChanged(newNode, oldNode) { return NodeConditionChange }
return "" // 无调度相关变更
}
只有影响调度的属性变更才会触发 MoveAllToActiveOrBackoffQueue，避免不必要的队列抖动。
3.13 Configurator 与 Profile 创建流程
func (c *Configurator) create() (*Scheduler, error) {
// 1. 创建 Extender
var extenders []framework.Extender
for _, extenderConfig := range c.extenders {
extender, _ := core.NewHTTPExtender(&extenderConfig)
if !extender.IsIgnorable() {
extenders = append(extenders, extender)
}
}
// 2. 创建 PodNominator
nominator := internalqueue.NewPodNominator()
// 3. ★ 创建 Profile Map（每个 SchedulerName 对应一个 Framework）
clusterEventMap := make(map[framework.ClusterEvent]sets.String)
profiles, _ := profile.NewMap(c.profiles, c.registry, c.recorderFactory,
frameworkruntime.WithClientSet(c.client),
frameworkruntime.WithInformerFactory(c.informerFactory),
frameworkruntime.WithSnapshotSharedLister(c.nodeInfoSnapshot),
frameworkruntime.WithPodNominator(nominator),
frameworkruntime.WithClusterEventMap(clusterEventMap),
frameworkruntime.WithExtenders(extenders),
frameworkruntime.WithParallelism(int(c.parallellism)),
)
// 4. 所有 Profile 必须使用相同的 QueueSort 插件
lessFn := profiles[c.profiles[0].SchedulerName].QueueSortFunc()
// 5. 创建调度队列
podQueue := internalqueue.NewSchedulingQueue(lessFn, c.informerFactory, ...)
// 6. 创建调度算法
algo := core.NewGenericScheduler(c.schedulerCache, c.nodeInfoSnapshot, extenders, c.percentageOfNodesToScore)
// 7. 组装 Scheduler
return &Scheduler{
SchedulerCache: c.schedulerCache,
Algorithm: algo,
Profiles: profiles,
NextPod: internalqueue.MakeNextPodFunc(podQueue),
Error: MakeDefaultErrorFunc(...),
SchedulingQueue: podQueue,
}, nil
}
3.14 In-Tree 插件注册表
NewInTreeRegistry() 注册了所有内置插件：
func NewInTreeRegistry() runtime.Registry {
return runtime.Registry{
selectorspread.Name: selectorspread.New,
imagelocality.Name: imagelocality.New,
tainttoleration.Name: tainttoleration.New,
nodename.Name: nodename.New,
nodeports.Name: nodeports.New,
nodepreferavoidpods.Name: nodepreferavoidpods.New,
nodeaffinity.Name: nodeaffinity.New,
podtopologyspread.Name: podtopologyspread.New,
nodeunschedulable.Name: nodeunschedulable.New,
noderesources.FitName: noderesources.NewFit,
noderesources.BalancedAllocationName: noderesources.NewBalancedAllocation,
noderesources.MostAllocatedName: noderesources.NewMostAllocated,
noderesources.LeastAllocatedName: noderesources.NewLeastAllocated,
noderesources.RequestedToCapacityRatioName: noderesources.NewRequestedToCapacityRatio,
volumebinding.Name: volumebinding.New,
volumerestrictions.Name: volumerestrictions.New,
volumezone.Name: volumezone.New,
nodevolumelimits.CSIName: nodevolumelimits.NewCSI,
nodevolumelimits.EBSName: nodevolumelimits.NewEBS,
nodevolumelimits.GCEPDName: nodevolumelimits.NewGCEPD,
nodevolumelimits.AzureDiskName: nodevolumelimits.NewAzureDisk,
nodevolumelimits.CinderName: nodevolumelimits.NewCinder,
interpodaffinity.Name: func(...) { return interpodaffinity.New(plArgs, fh, fts) },
nodelabel.Name: nodelabel.New,
serviceaffinity.Name: serviceaffinity.New,
queuesort.Name: queuesort.New,
defaultbinder.Name: defaultbinder.New,
defaultpreemption.Name: defaultpreemption.New,
}
}
DefaultProvider 的默认插件配置（algorithmprovider/registry.go）：
func getDefaultConfig() *schedulerapi.Plugins {
return &schedulerapi.Plugins{
QueueSort: {Enabled: []Plugin{{Name: "PrioritySort"}}},
PreFilter: {Enabled: []Plugin{
{Name: "NodeResourcesFit"}, {Name: "NodePorts"},
{Name: "PodTopologySpread"}, {Name: "InterPodAffinity"},
{Name: "VolumeBinding"}, {Name: "NodeAffinity"},
}},
Filter: {Enabled: []Plugin{
{Name: "NodeUnschedulable"}, {Name: "NodeName"},
{Name: "TaintToleration"}, {Name: "NodeAffinity"},
{Name: "NodePorts"}, {Name: "NodeResourcesFit"},
{Name: "VolumeRestrictions"}, {Name: "NodeVolumeLimits-EBSCSI..."},
{Name: "VolumeBinding"}, {Name: "VolumeZone"},
{Name: "PodTopologySpread"}, {Name: "InterPodAffinity"},
}},
PostFilter: {Enabled: []Plugin{{Name: "DefaultPreemption"}}},
PreScore: {Enabled: []Plugin{
{Name: "InterPodAffinity"}, {Name: "PodTopologySpread"},
{Name: "TaintToleration"}, {Name: "NodeAffinity"},
}},
Score: {Enabled: []Plugin{
{Name: "NodeResourcesBalancedAllocation", Weight: 1},
{Name: "ImageLocality", Weight: 1},
{Name: "InterPodAffinity", Weight: 1},
{Name: "NodeResourcesLeastAllocated", Weight: 1},
{Name: "NodeAffinity", Weight: 1},
{Name: "NodePreferAvoidPods", Weight: 10000},
{Name: "PodTopologySpread", Weight: 2},
{Name: "TaintToleration", Weight: 1},
}},
Reserve: {Enabled: []Plugin{{Name: "VolumeBinding"}}},
PreBind: {Enabled: []Plugin{{Name: "VolumeBinding"}}},
Bind: {Enabled: []Plugin{{Name: "DefaultBinder"}}},
}
}
3.15 CycleState 跨插件状态传递
CycleState 是调度周期内的全局状态存储，插件间通过它共享数据：
type CycleState struct {
mx sync.RWMutex
storage map[StateKey]StateData
recordPluginMetrics bool // 10% 采样率
}
// 使用示例（VolumeBinding 插件）：
// PreFilter 阶段写入
state.Write(stateKey, &stateData{boundClaims: boundClaims, claimsToBind: claimsToBind, ...})
// Filter 阶段读取
s, _ := cs.Read(stateKey)
stateData := s.(*stateData)
// Clone 机制：在 RunFilterPluginsWithNominatedPods 中克隆 CycleState
// 因为 addNominatedPods 会修改 PreFilter 状态
stateOut := state.Clone()
3.16 MakeDefaultErrorFunc 错误处理
当调度失败后，MakeDefaultErrorFunc 决定 Pod 的后续处理：
func MakeDefaultErrorFunc(...) func(*framework.QueuedPodInfo, error) {
return func(podInfo *framework.QueuedPodInfo, err error) {
// 1. 注入 UnschedulablePlugins（用于智能队列移动）
if fitError, ok := err.(*framework.FitError); ok {
podInfo.UnschedulablePlugins = fitError.Diagnosis.UnschedulablePlugins
}
// 2. 处理 NodeNotFound 错误：从缓存移除不存在的 Node
if apierrors.IsNotFound(err) {
if errStatus.Status().Details.Kind == "node" {
nodeName := errStatus.Status().Details.Name
_, err := client.CoreV1().Nodes().Get(ctx, nodeName, metav1.GetOptions{})
if apierrors.IsNotFound(err) {
schedulerCache.RemoveNode(&v1.Node{ObjectMeta: metav1.ObjectMeta{Name: nodeName}})
}
}
}
// 3. 检查 Pod 是否已被绑定（防止 Extender 超时导致的重复入队）
cachedPod, _ := podLister.Pods(pod.Namespace).Get(pod.Name)
if len(cachedPod.Spec.NodeName) != 0 {
return // 已绑定，不再入队
}
// 4. ★ 加入不可调度队列
podInfo.PodInfo = framework.NewPodInfo(cachedPod.DeepCopy())
podQueue.AddUnschedulableIfNotPresent(podInfo, podQueue.SchedulingCycle())
}
}
四、关键设计总结
4.1 性能优化设计
| 优化点 | 实现 |
|---|---|
| 增量快照 | UpdateSnapshot 基于 Generation 增量更新，只 Clone 变更的 NodeInfo |
| 并行过滤 | Parallelizer.Until() 并行执行 Filter，找到足够可行节点后提前终止 |
| 轮转起点 | nextStartNodeIndex 保证所有节点公平被检查 |
| 自适应节点比例 | numFeasibleNodesToFind 大集群只检查部分节点 |
| 乐观假设 | AssumePod 在绑定前更新缓存，避免串行等待 |
| 异步绑定 | Binding Cycle 在 goroutine 中执行，不阻塞下一个 Pod 调度 |
| 智能队列移动 | 基于集群事件与失败插件的精准匹配，避免无谓重调度 |
4.2 一致性保障设计
| 设计 | 说明 |
|---|---|
| Assume/Forget 机制 | 绑定前 Assume，失败时 Forget + Unreserve，保证缓存与实际一致 |
| Unreserve 反向调用 | Reserve 失败/Permit 拒绝/PreBind 失败/Bind 失败都触发 Unreserve，按反向顺序调用 |
| NominatedNode 机制 | 抢占成功后设置 NominatedNodeName，下次调度优先检查该节点 |
| 两轮 Filter | RunFilterPluginsWithNominatedPods 做保守的两次过滤 |
| Leader Election | 多副本调度器通过选举保证同一时间只有一个工作 |
4.3 扩展性设计
| 设计 | 说明 |
|---|---|
| Scheduling Framework | 11 个扩展点，插件式架构 |
| Profile | 支持 Multiple Scheduler，每个有独立插件配置 |
| Extender | HTTP API 扩展，支持 Filter/Prioritize/Bind/Preempt |
| Out-of-Tree 插件 | WithPlugin() 注入自定义插件 |
| CycleState | 插件间通过 StateKey 共享数据 |
| EnqueueExtensions | 插件声明关注的事件，精准触发重调度 |
本文档基于 Kubernetes 源码严格分析，覆盖了 kube-scheduler 从启动到 Pod 调度完成的完整链路，包含所有关键数据结构、算法细节和设计决策。
更多推荐
9
0- 0
已为社区贡献61条内容
所有评论(0)