4. 自定义调度器：编写自定义调度器框架、调度器扩展点、调度器插件机制

Kubernetes 默认调度器（kube-scheduler）通过一系列内置的 调度插件 完成 Pod 与 Node 的绑定决策。但在多租户隔离、GPU 拓扑感知、窗口数据传输窗口抢占等复杂场景下，默认调度策略无法满足需求。本章深入自定义调度器的三种实现路径：框架扩展、扩展点注入、插件机制，并给出生产级代码骨架。

4.1 自定义调度器架构总览

Kubernetes 调度框架（Scheduling Framework）将调度周期拆解为多个可插拔的 扩展点（Extension Points）。自定义调度器通过实现这些扩展点的接口，以插件形式注入调度流程，无需重写整个调度器。

4.2 调度器扩展点详解

调度框架定义了 12 个扩展点，每个扩展点对应调度周期中的一个阶段。自定义插件可以选择实现其中任意一个或多个扩展点。

4.3 自定义调度器插件开发框架

编写自定义调度器插件需要实现 schedulerframework.Plugin 接口，并根据需要实现对应的扩展点接口。以下是一个完整的插件骨架示例：

package windowtransferscheduler

import (
    "context"
    "fmt"
    "k8s.io/kubernetes/pkg/scheduler/framework"
)

// 插件名称，在调度器配置中引用
const Name = "WindowTransfer"

// 插件结构体
type WindowTransfer struct {
    handle framework.Handle
    // 自定义配置：窗口时间窗口、带宽阈值等
    windowStart string
    windowEnd   string
    bandwidthThreshold int64
}

// 确保实现了 framework.Plugin 接口
var _ framework.Plugin = &WindowTransfer{}

// Name 返回插件名称
func (pl *WindowTransfer) Name() string {
    return Name
}

// 实现 Filter 扩展点
// 在 Filter 阶段检查当前时间是否在数据传输窗口内
func (pl *WindowTransfer) Filter(ctx context.Context, 
    state *framework.CycleState, 
    pod *v1.Pod, 
    nodeInfo *framework.NodeInfo) *framework.Status {
    
    // 检查节点是否满足窗口数据传输条件
    if !pl.isInWindow() {
        return framework.NewStatus(framework.Unschedulable, 
            "当前时间不在数据传输窗口内")
    }
    
    // 检查节点带宽是否满足 Pod 需求
    if nodeInfo.Allocatable.MilliCPU < pl.bandwidthThreshold {
        return framework.NewStatus(framework.Unschedulable, 
            "节点带宽不足")
    }
    
    return framework.NewStatus(framework.Success, "")
}

// 实现 Score 扩展点
// 根据节点剩余带宽和窗口利用率进行评分
func (pl *WindowTransfer) Score(ctx context.Context, 
    state *framework.CycleState, 
    pod *v1.Pod, 
    nodeName string) (int64, *framework.Status) {
    
    nodeInfo, err := pl.handle.SnapshotSharedLister().NodeInfos().Get(nodeName)
    if err != nil {
        return 0, framework.NewStatus(framework.Error, 
            fmt.Sprintf("获取节点信息失败: %v", err))
    }
    
    // 评分逻辑：剩余带宽越高，分数越高
    availableBandwidth := nodeInfo.Allocatable.MilliCPU - 
        nodeInfo.Requested.MilliCPU
    score := (availableBandwidth * 100) / pl.bandwidthThreshold
    
    if score > 100 {
        score = 100
    }
    
    return score, framework.NewStatus(framework.Success, "")
}

// 实现 NormalizeScore 扩展点（可选）
func (pl *WindowTransfer) NormalizeScore(ctx context.Context, 
    state *framework.CycleState, 
    pod *v1.Pod, 
    scores framework.NodeScoreList) *framework.Status {
    
    // 归一化处理，确保所有分数在 0-100 之间
    return framework.NewStatus(framework.Success, "")
}

// 辅助方法：检查当前时间是否在窗口内
func (pl *WindowTransfer) isInWindow() bool {
    // 实现窗口时间检查逻辑
    return true
}

// 插件构造函数
func New(obj runtime.Object, handle framework.Handle) (framework.Plugin, error) {
    // 解析配置
    config, ok := obj.(*WindowTransferArgs)
    if !ok {
        return nil, fmt.Errorf("期望 WindowTransferArgs，得到 %T", obj)
    }
    
    return &WindowTransfer{
        handle:             handle,
        windowStart:        config.WindowStart,
        windowEnd:          config.WindowEnd,
        bandwidthThreshold: config.BandwidthThreshold,
    }, nil
}

4.4 插件注册与调度器配置

自定义插件编写完成后，需要通过 调度器配置 注册并启用。Kubernetes 1.19+ 支持通过 KubeSchedulerConfiguration 声明式配置插件。

4.4.1 插件注册

在调度器主程序中注册自定义插件：

package main

import (
    "k8s.io/kubernetes/cmd/kube-scheduler/app"
    "k8s.io/kubernetes/pkg/scheduler/framework"
    
    // 引入自定义插件包
    windowtransferscheduler "example.com/window-transfer-scheduler"
)

func main() {
    // 注册自定义插件
    cmd := app.NewSchedulerCommand(
        app.WithPlugin(windowtransferscheduler.Name, 
            windowtransferscheduler.New),
    )
    
    if err := cmd.Execute(); err != nil {
        panic(err)
    }
}

4.4.2 调度器配置文件

通过 YAML 配置文件指定插件的启用顺序和权重：

apiVersion: kubescheduler.config.k8s.io/v1
kind: KubeSchedulerConfiguration
profiles:
  - schedulerName: window-transfer-scheduler
    plugins:
      # 在 Filter 阶段启用自定义插件
      filter:
        enabled:
          - name: "WindowTransfer"
        disabled:
          - name: "NodeResourcesFit"  # 可选：禁用默认插件
      # 在 Score 阶段启用自定义插件，权重为 3
      score:
        enabled:
          - name: "WindowTransfer"
            weight: 3
      # 在 Permit 阶段启用自定义插件
      permit:
        enabled:
          - name: "WindowTransfer"
    pluginConfig:
      - name: "WindowTransfer"
        args:
          windowStart: "22:00"
          windowEnd: "06:00"
          bandwidthThreshold: 1000  # 单位：mCPU

4.5 调度器扩展点执行流程

以下 SVG 图展示了自定义插件在调度周期中的完整执行流程，以及各扩展点的调用顺序：

4.6 多调度器部署与选择

Kubernetes 支持集群中同时运行多个调度器实例。Pod 通过 spec.schedulerName 字段选择使用哪个调度器。

4.7 自定义调度器调试与验证

部署自定义调度器后，通过以下步骤验证其正确性：

1. 检查调度器 Pod 日志： kubectl logs -n kube-system <scheduler-pod>，确认插件加载成功。
2. 查看调度事件： kubectl describe pod <pod-name>，在 Events 中查看调度决策。
3. 使用调度器指标： 自定义调度器暴露 Prometheus 指标，监控 scheduler_binding_latency_microseconds 和 scheduler_e2e_scheduling_latency_microseconds。
4. 单元测试： 使用 schedulerframework.Framework 的测试框架编写插件单元测试。

// 插件单元测试示例
func TestWindowTransferFilter(t *testing.T) {
    // 创建测试框架
    f, _ := framework.NewFramework(
        []framework.Plugin{&WindowTransfer{windowStart: "22:00", windowEnd: "06:00"}},
        framework.WithSnapshotSharedLister(testNodeInfo()),
    )
    
    // 创建测试 Pod
    pod := testPod("test-pod", "default")
    
    // 执行 Filter 扩展点
    status := f.RunFilterPlugins(context.TODO(), nil, pod, "test-node")
    
    if status.IsSuccess() {
        t.Log("Filter 通过")
    } else {
        t.Errorf("Filter 未通过: %v", status.Message())
    }
}