Kubernetes 调度器—Pod 如何被分配到节点

// scheduling_queue.go
type PriorityQueue struct {
    activeQ              activeQueuer           // 活跃队列（即将被调度）
    backoffQ             backoffQueuer          // 退避队列（调度失败后等待重试）
    unschedulableEntities *unschedulableEntities // 不可调度的 Pod（暂时无法调度）
}

DefaultPodInitialBackoffDuration = 1 * time.Second   // 初始退避 1s
DefaultPodMaxBackoffDuration     = 10 * time.Second  // 最大退避 10s

// 当节点资源变化、Pod 被删除等事件发生时（参数已简化）
func (p *PriorityQueue) MoveAllToActiveOrBackoffQueue(logger, event, oldObj, newObj, preCheck) {
    // 将 unschedulableEntities 中的 Pod 移入 activeQ 或 backoffQ
}

// v1.37 中 NodeInfo 是接口，通过方法访问节点信息
type NodeInfo interface {
    Node() *v1.Node                 // 节点信息
    GetUsedPorts() HostPortInfo     // 已使用端口
    GetImageStates() map[string]*ImageStateSummary  // 镜像状态
    GetPods() []PodInfo             // 节点上的 Pod 列表
    // ...
}

func (c *cacheImpl) UpdateSnapshot(logger, nodeInfoSnapshot) error {
    // 更新节点快照
}

type Plugin interface {
    Name() string
}

type PreFilterPlugin interface {
    Plugin
    PreFilter(ctx, state, pod) *Status
}

type FilterPlugin interface {
    Plugin
    Filter(ctx, state, pod, nodeInfo) *Status
}

type ScorePlugin interface {
    Plugin
    Score(ctx, state, pod, nodeInfo) (int64, *Status)
    ScoreExtensions() ScoreExtensions
}

func (sched *Scheduler) schedulingCycle(ctx, state, fwk, podInfo) {
    // 1. 更新节点快照
    sched.Cache.UpdateSnapshot(logger, sched.nodeInfoSnapshot)
    
    // 2. 执行调度算法
    scheduleResult, status := sched.schedulingAlgorithm(ctx, state, fwk, podInfo)
    
    // 3. 准备绑定周期（在内存中假设 Pod 已调度）
    assumedPodInfo, status := sched.prepareForBindingCycle(...)
}

func (sched *Scheduler) schedulingAlgorithm(ctx, state, fwk, podInfo) {
    // 1. PreFilter：预处理
    fwk.RunPreFilterPlugins(ctx, state, pod)
    
    // 2. Filter：过滤节点（findNodesThatPassFilters 汇总可行节点）
    feasibleNodes := sched.findNodesThatPassFilters(ctx, fwk, state, pod, nodeInfos)
    
    // 3. PostFilter：调度算法返回 *framework.FitError（无可行节点）
    //    且注册了 PostFilter 插件时触发，尝试抢占
    if len(feasibleNodes) == 0 && fwk.HasPostFilterPlugins() {
        fwk.RunPostFilterPlugins(ctx, state, pod, fitError.Diagnosis.NodeToStatus)
    }
    
    // 4. PreScore：评分前准备
    fwk.RunPreScorePlugins(ctx, state, pod, feasibleNodes)
    
    // 5. Score：打分
    priorityList := fwk.RunScorePlugins(ctx, state, pod, feasibleNodes)
    
    // 6. 选择得分最高的节点（schedulePod 中）
    selectedNode := framework.NewSortedScoredNodes(priorityList).Pop().Name
    
    return ScheduleResult{selectedNode}
}

// 异步执行绑定
go sched.runBindingCycle(ctx, state, fwk, scheduleResult, assumedPodInfo, start)

// 调度周期内（prepareForBindingCycle / assumeAndReserve）
fwk.RunReservePluginsReserve(ctx, state, assumedPod, scheduleResult.SuggestedHost)
fwk.RunPermitPlugins(ctx, state, assumedPod, scheduleResult.SuggestedHost)

func (sched *Scheduler) bindingCycle(ctx, state, fwk, scheduleResult, assumedPodInfo) {
    // 1. PreBindPreFlight：绑定前置检查
    fwk.RunPreBindPreFlights(ctx, state, assumedPod, scheduleResult.SuggestedHost)
    
    // 2. WaitOnPermit：等待 Permit 插件的准入结果
    fwk.WaitOnPermit(ctx, assumedPod)
    
    // 3. PreBind：绑定前准备
    fwk.RunPreBindPlugins(ctx, state, assumedPod, scheduleResult.SuggestedHost)
    
    // 4. Bind：执行绑定（写入 API Server）
    fwk.RunBindPlugins(ctx, state, assumedPod, scheduleResult.SuggestedHost)
    
    // 5. PostBind：绑定后处理
    fwk.RunPostBindPlugins(ctx, state, assumedPod, scheduleResult.SuggestedHost)
}

state := framework.NewCycleState()
state.Write("preFilterState", preFilterResult)
state.Read("preFilterState")

// schedule_one.go：聚合可行节点
func (sched *Scheduler) findNodesThatPassFilters(ctx, fwk, state, pod, nodes) []NodeInfo {
    feasibleNodes := make([]NodeInfo, numNodesToFind)
    checkNode := func(i int) {
        status := fwk.RunFilterPluginsWithNominatedPods(ctx, state, pod, nodes[i])
        if status.IsSuccess() {
            feasibleNodes[...] = nodes[i] // 收集可行节点
        }
    }
    fwk.Parallelizer().Until(ctx, numAllNodes, checkNode, ...) // 并行检查各节点
    return feasibleNodes
}

// runtime/framework.go：针对单个节点，串行执行所有 Filter 插件
func (f *frameworkImpl) RunFilterPlugins(ctx, state, pod, nodeInfo) *Status {
    for _, pl := range f.filterPlugins {
        if status := f.runFilterPlugin(ctx, pl, state, pod, nodeInfo); !status.IsSuccess() {
            return status
        }
    }
    return nil
}

// plugins/noderesources/fit.go：插件结构体名为 Fit
func (f *Fit) Score(ctx, state, pod, nodeInfo) (int64, *Status) {
    // Score 直接接收 NodeInfo，无需再按 nodeName 查询快照
    // 计算资源利用率
    requested := f.calculatePodResourceRequestList(pod, f.resources)
    capacity := nodeInfo.GetAllocatable() // 节点容量
    // 分数 = (1 - 利用率) × 100
    return score, nil
}

score := int64(0)
for _, pluginScore := range pluginScores {
    score += pluginScore.Score * pluginScore.Weight
}

// 没有可行节点（*framework.FitError），且注册了 PostFilter 插件
if len(feasibleNodes) == 0 && fwk.HasPostFilterPlugins() {
    fwk.RunPostFilterPlugins(ctx, state, pod, fitError.Diagnosis.NodeToStatus)
}

// preemption.go
func (ev *Evaluator) evaluate(ctx, state, pod, m) Candidate {
    // 1. 找出所有可能的受害者
    // 2. 检查 PDB 约束
    // 3. 选择最优的候选节点
}

type Map map[string]framework.Framework

func NewMap(ctx, cfgs []KubeSchedulerProfile, r Registry, recorderFact RecorderFactory) (Map, error) {
    m := make(Map)
    for _, cfg := range cfgs {
        p, _ := newProfile(ctx, cfg, r, recorderFact, opts...)
        m[cfg.SchedulerName] = p
    }
    return m, nil
}

apiVersion: v1
kind: Pod
metadata:
  name: my-pod
spec:
  schedulerName: my-scheduler  # 指定调度器 Profile
  containers:
  - name: nginx
    image: nginx

func (sched *Scheduler) ScheduleOne(ctx) {
    // 1. 从队列获取下一个 Pod
    entity, err := sched.NextEntity(logger)
    
    // 2. 根据实体类型分发
    switch entity.(type) {
    case *framework.QueuedPodInfo:
        sched.scheduleOnePod(ctx, podInfo)
    case *framework.QueuedPodGroupInfo:
        sched.scheduleOnePodGroup(ctx, podGroupInfo)
    }
}

func (sched *Scheduler) scheduleOnePod(ctx, podInfo) {
    // 1. 获取 Framework
    schedFramework, _ := sched.frameworkForPod(pod)
    
    // 2. 初始化 CycleState
    state := framework.NewCycleState()
    
    // 3. 执行调度周期
    scheduleResult, assumedPodInfo, status := sched.schedulingCycle(ctx, state, schedFramework, podInfo)
    
    // 4. 异步执行绑定周期
    go sched.runBindingCycle(ctx, state, schedFramework, scheduleResult, assumedPodInfo, start)
}