Kubernetes Scheduler Score 打分机制深度解析 #

const (
    MinNodeScore int64 = 0    // 最低分
    MaxNodeScore int64 = 100  // 最高分（单个插件）
)

// 单个插件对某个节点的打分
type NodeScore struct {
    Name  string  // 节点名
    Score int64   // 分数
}

// 单个插件的打分结果
type PluginScore struct {
    Name  string  // 插件名
    Score int64   // 加权后的分数
}

// 某个节点的所有打分结果
type NodePluginScores struct {
    Name       string         // 节点名
    Scores     []PluginScore  // 各插件打分明细
    TotalScore int64          // 加权总分
}

// ScorePlugin 是打分插件的接口
type ScorePlugin interface {
    Plugin
    // Score 对单个节点打分
    Score(ctx context.Context, state CycleState, pod *v1.Pod, nodeInfo NodeInfo) (int64, *Status)
    // ScoreExtensions 返回分数归一化扩展（可选）
    ScoreExtensions() ScoreExtensions
}

// ScoreExtensions 是分数归一化扩展接口
type ScoreExtensions interface {
    // NormalizeScore 在所有节点打分完成后进行归一化
    NormalizeScore(ctx context.Context, state CycleState, pod *v1.Pod, scores NodeScoreList) *Status
}

// schedule_one.go
func prioritizeNodes(
    ctx context.Context,
    extenders []fwk.Extender,
    schedFramework framework.Framework,
    state fwk.CycleState,
    pod *v1.Pod,
    nodes []fwk.NodeInfo,
) ([]fwk.NodePluginScores, error) {

    // 1. 没有任何打分插件时，所有节点默认得 1 分
    if len(extenders) == 0 && !schedFramework.HasScorePlugins() {
        result := make([]fwk.NodePluginScores, 0, len(nodes))
        for i := range nodes {
            result = append(result, fwk.NodePluginScores{
                Name:       nodes[i].Node().Name,
                TotalScore: 1,
            })
        }
        return result, nil
    }

    // 2. 执行 PreScore 插件（预处理）
    preScoreStatus := schedFramework.RunPreScorePlugins(ctx, state, pod, nodes)

    // 3. 执行 Score 插件（并行打分）
    nodesScores, scoreStatus := schedFramework.RunScorePlugins(ctx, state, pod, nodes)

    // 4. 执行 Extender 打分（外部调度器）
    if len(extenders) != 0 {
        // 并行调用所有 Extender 的 Prioritize 方法
        // ...
    }

    return nodesScores, nil
}

// runtime/framework.go
func (f *frameworkImpl) RunScorePlugins(ctx context.Context, state fwk.CycleState, pod *v1.Pod, nodes []fwk.NodeInfo) ([]fwk.NodePluginScores, *fwk.Status) {

    // 初始化结果结构
    allNodePluginScores := make([]fwk.NodePluginScores, len(nodes))
    plugins := make([]fwk.ScorePlugin, 0, numPlugins)
    pluginToNodeScores := make(map[string]fwk.NodeScoreList, numPlugins)

    // ============ 第一阶段：并行 Score ============
    // 对每个节点并行执行所有插件的 Score() 方法
    f.Parallelizer().Until(ctx, len(nodes), func(index int) {
        nodeInfo := nodes[index]
        for _, pl := range plugins {
            // 调用插件的 Score 方法
            s, status := f.runScorePlugin(ctx, pl, state, pod, nodeInfo)

            // 记录该插件对该节点的打分
            pluginToNodeScores[pl.Name()][index] = fwk.NodeScore{
                Name:  nodeInfo.Node().Name,
                Score: s,  // 原始分数
            }
        }
    }, metrics.Score)

    // ============ 第二阶段：并行 NormalizeScore ============
    // 对每个插件并行执行 NormalizeScore()（如果实现了 ScoreExtensions）
    f.Parallelizer().Until(ctx, len(plugins), func(index int) {
        pl := plugins[index]
        if pl.ScoreExtensions() == nil {
            return  // 该插件没有实现归一化
        }
        nodeScoreList := pluginToNodeScores[pl.Name()]
        // 调用归一化方法，将分数调整到 [0, MaxNodeScore] 范围
        status := pl.ScoreExtensions().NormalizeScore(ctx, state, pod, nodeScoreList)
    }, metrics.Score)

    // ============ 第三阶段：并行加权计算 ============
    // 对每个节点并行计算加权总分
    f.Parallelizer().Until(ctx, len(nodes), func(index int) {
        nodePluginScores := fwk.NodePluginScores{
            Name:   nodes[index].Node().Name,
            Scores: make([]fwk.PluginScore, len(plugins)),
        }

        for i, pl := range plugins {
            weight := f.scorePluginWeight[pl.Name()]  // 从配置读取权重
            score := pluginToNodeScores[pl.Name()][index].Score

            // 验证分数范围
            if score > fwk.MaxNodeScore || score < fwk.MinNodeScore {
                // 错误：分数超出范围
            }

            // 加权分数 = 原始分数 × 权重
            weightedScore := score * int64(weight)

            nodePluginScores.Scores[i] = fwk.PluginScore{
                Name:  pl.Name(),
                Score: weightedScore,
            }
            nodePluginScores.TotalScore += weightedScore  // 累加总分
        }

        allNodePluginScores[index] = nodePluginScores
    }, metrics.Score)

    return allNodePluginScores, nil
}

// pkg/scheduler/framework/plugins/noderesources/least_allocated.go

// 核心打分函数：偏好资源使用率低的节点
func leastResourceScorer(resources []config.ResourceSpec) func([]int64, []int64, []int64) int64 {
    return func(requested, _, allocable []int64) int64 {
        var nodeScore, weightSum int64
        for i := range requested {
            if allocable[i] == 0 {
                continue
            }
            weight := resources[i].Weight
            resourceScore := leastRequestedScore(requested[i], allocable[i])
            nodeScore += resourceScore * weight
            weightSum += weight
        }
        return nodeScore / weightSum
    }
}

// 单个资源打分公式
func leastRequestedScore(requested, capacity int64) int64 {
    if capacity == 0 {
        return 0
    }
    if requested > capacity {
        return 0
    }
    // 公式：(capacity - requested) * MaxNodeScore / capacity
    // 空闲资源越多，分数越高
    return ((capacity - requested) * fwk.MaxNodeScore) / capacity
}

// pkg/scheduler/framework/plugins/noderesources/most_allocated.go

// 偏好资源使用率高的节点（Bin-packing 策略）
func mostResourceScorer(resources []config.ResourceSpec) func([]int64, []int64, []int64) int64 {
    return func(requested, allocatable, _ []int64) int64 {
        var nodeScore, weightSum int64
        for i := range requested {
            if allocatable[i] == 0 {
                continue
            }
            weight := resources[i].Weight
            resourceScore := mostRequestedScore(requested[i], allocatable[i])
            nodeScore += resourceScore * weight
            weightSum += weight
        }
        return nodeScore / weightSum
    }
}

// 公式：requested * MaxNodeScore / capacity
// 使用率越高，分数越高
func mostRequestedScore(requested, capacity int64) int64 {
    if requested > capacity {
        return 0
    }
    return (requested * fwk.MaxNodeScore) / capacity
}

apiVersion: kubescheduler.config.k8s.io/v1
kind: KubeSchedulerConfiguration
profiles:
- plugins:
    score:
      enabled:
      - name: NodeResourcesFit
        weight: 2
      - name: InterPodAffinity
        weight: 1
      - name: TaintToleration
        weight: 1

// 使用 Parallelizer 并行处理所有节点
f.Parallelizer().Until(ctx, len(nodes), func(index int) {
    // 每个 goroutine 处理一个节点的所有插件打分
}, metrics.Score)

// 当节点数很多时，可以只打分一部分节点
numNodesToFind := sched.numFeasibleNodesToFind(percentageOfNodesToScore, numAllNodes)

// PreScore 阶段可以将计算结果存入 CycleState，供 Score 阶段复用
func (pl *InterPodAffinity) PreScore(ctx context.Context, state CycleState, pod *v1.Pod, nodes []NodeInfo) *Status {
    // 预计算亲和关系，存入 state
    state.Write(interPodAffinityStateKey, preScoreData)
    return nil
}

func (pl *InterPodAffinity) Score(ctx context.Context, state CycleState, pod *v1.Pod, nodeInfo NodeInfo) (int64, *Status) {
    // 从 state 读取预计算数据
    data, _ := state.Read(interPodAffinityStateKey)
    // 使用预计算数据加速打分
}