Kubernetes 调度器深度实战：从 Predicate 算法到自定义调度器的完整实现之路

type SchedulingQueue struct {
    // 禁止标志位中的 Pod 调度
    stopCond   atomic.Value
    clock      clock.Clock

    // 活跃队列 - 可立即调度的 Pod
    activeQ *heap.Heap

    // 退避队列 - 调度失败的 Pod 等待重试
    podBackoffQ *heap.Heap

    // 不可调度队列 - 暂时无法满足资源需求的 Pod
    unschedulablePods *UnschedulablePodsMap
}

func (f *NodeResourcesFit) Filter(ctx context.Context,
    state *framework.CycleState, pod *v1.Pod,
    nodeInfo *framework.NodeInfo) *framework.Status {

    // 1. 检查 Node 是否有 Pod 的 NodeSelector 匹配
    // 2. 检查 Node 的 Allocatable 资源是否 ≥ Pod 的 Requests 总和
    // 3. 检查 hugepage 资源是否满足
    // 4. 检查 volumes 拓扑约束（VolumeLimits）

    insufficientResources := fitsRequest(pod, nodeInfo)
    if len(insufficientResources) != 0 {
        return framework.NewStatus(
            framework.Unschedulable,
            fmt.Sprintf("Insufficient %v", insufficientResources),
        )
    }
    return nil
}

func balancedResourceScorer(requestedToCapacityRatio resourceToValueMap,
    resourceToWeightMap resourceToWeightMap) scoreNodeFn {
    return func(node string, requestedToCapacityRatio resourceToValueMap) int64 {
        // 核心思想：优先选择资源利用率最均衡的 Node
        // 而非资源最空闲的 Node（避免"甜点效应"）
        score := 0
        for resource := range requestedToCapacityRatio {
            // 计算放置后的资源比率
            resourceToCapacityRatioAfterPodAdded := ...
            // 各资源比率方差越小，评分越高（越均衡）
            // 60 分满分
            score += ... 
        }
        return score
    }
}

QueueSort → PreFilter → Filter → PostFilter → PreScore → Score → Reserve → Permit → PreBind → Bind → PostBind

apiVersion: kubescheduler.config.k8s.io/v1
kind: KubeSchedulerConfiguration
parallelism: 16                    # 并行调度 Goroutine 数（默认 16）
clientConnection:
  kubeconfig: /etc/kubernetes/scheduler.conf
  qps: 100                         # 调度器对 API Server 的 QPS（默认 50）
  burst: 200                       #  Burst 上限（默认 100）
profiles:
  - schedulerName: default-scheduler
    plugins:
      multiPoint:
        enabled:
          - name: NodeResourcesFit
            weight: 2            # Score 阶段的权重
          - name: NodeResourcesBalancedAllocation
            weight: 1
          - name: PodTopologySpread
            weight: 2
        disabled:
          - name: NodeResourcesLeastAllocated  # 关闭最空闲策略
      preScore:
        disabled:
          - name: '*'             # 禁用所有 PreScore
      score:
        enabled:
          - name: MyCustomScorer
            weight: 3             # 自定义评分器权重
    pluginConfig:
      - name: NodeResourcesFit
        args:
          scoringStrategy:
            type: RequestedToCapacityRatio   # 使用请求-容量比率策略
            resources:
              - name: cpu
                weight: 1
              - name: memory
                weight: 1
              - name: nvidia.com/gpu
                weight: 2       # GPU 资源权重翻倍
      - name: KubeSchedulerProfileName
        args:
          percentageOfNodesToScore: 50  # 只对 50% 的 Node 做评分

抢占触发条件：Pending Pod 的 Priority ≥ 1000000000（system-cluster-critical）或 PriorityClass 的 value 大于目标 Pod 的 value

package main

import (
    "context"
    "fmt"
    "time"

    v1 "k8s.io/api/core/v1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/client-go/informers"
    "k8s.io/client-go/kubernetes"
    "k8s.io/client-go/tools/clientcmd"
    "k8s.io/client-go/tools/leaderelection"
    "k8s.io/client-go/tools/leaderelection/resourcelock"
)

func main() {
    config, err := clientcmd.BuildConfigFromFlags("", "/etc/kubernetes/scheduler.conf")
    if err != nil {
        panic(err.Error())
    }

    clientset, err := kubernetes.NewForConfig(config)
    if err != nil {
        panic(err.Error())
    }

    // 1. 创建 Informer Factory 监听 Pod 和 Node 事件
    informerFactory := informers.NewSharedInformerFactory(clientset, 0)
    podInformer := informerFactory.Core().V1().Pods()
    nodeInformer := informerFactory.Core().V1().Nodes()

    // 2. 自定义调度器结构
    customScheduler := &CustomScheduler{
        clientset:    clientset,
        podInformer:  podInformer,
        nodeInformer: nodeInformer,
        nodeLister:   nodeInformer.Lister(),
        podLister:    podInformer.Lister(),
    }

    // 3. 启动 Leader Election（HA 模式）
    rl, _ := resourcelock.New(
        resourcelock.LeasesResourceLock,
        "kube-system",
        "custom-scheduler",
        clientset.CoreV1(),
        clientset.CoordinationV1(),
        resourcelock.ResourceLockConfig{
            Identity: hostname,
        },
    )

    leaderelection.RunOrDie(context.TODO(), leaderelection.LeaderElectionConfig{
        Lock:            rl,
        ReleaseOnCancel: true,
        LeaseDuration:   15 * time.Second,
        RenewDeadline:   10 * time.Second,
        RetryPeriod:     2 * time.Second,
        Callbacks: leaderelection.LeaderCallbacks{
            OnStartedLeading: func(ctx context.Context) {
                // 成为 Leader 后启动调度循环
                informerFactory.Start(ctx.Done())
                customScheduler.Run(ctx.Done())
            },
            OnStoppedLeading: func() {
                klog.Fatalf("leader election lost")
            },
        },
    })
}

type CustomScheduler struct {
    clientset    *kubernetes.Clientset
    podInformer  cache.SharedIndexInformer
    nodeInformer cache.SharedIndexInformer
    nodeLister   listers.NodeLister
    podLister    listers.PodLister
}

func (s *CustomScheduler) Run(stopCh <-chan struct{}) {
    wait.Until(func() {
        s.scheduleOne()
    }, time.Second, stopCh)
}

// scheduleOne 完成单 Pod 的调度全流程
func (s *CustomScheduler) scheduleOne() {
    // Phase 1: 从Queue获取待调度 Pod
    pods, _ := s.podLister.Pods(metav1.NamespaceAll).List(
        labels.SelectorFromSet(labels.Set{
            "spec.schedulerName": "custom-scheduler",
        }))
    
    if len(pods) == 0 {
        return
    }
    targetPod := pods[0] // 取队首

    // Phase 2: 执行 Filter（自定义规则）
    nodes, _ := s.nodeLister.List(labels.Everything())
    feasibleNodes := make([]*v1.Node, 0)
    for _, node := range nodes {
        if s.podFitsNode(targetPod, node) {
            feasibleNodes = append(feasibleNodes, node)
        }
    }

    if len(feasibleNodes) == 0 {
        return // 无可用 Node，进入 backoff
    }

    // Phase 3: 执行 Score + 选最优
    bestNode := s.bestScore(targetPod, feasibleNodes)

    // Phase 4: 执行 Bind
    s.bindPod(targetPod, bestNode)
}

// podFitsNode 自定义 Filter 逻辑
func (s *CustomScheduler) podFitsNode(pod *v1.Pod, node *v1.Node) bool {
    // Filter 1: 检查资源是否足够
    allocatable := node.Status.Allocatable
    podRequest := resource.Requests(pod)
    
    if podRequest.Cpu().MilliValue() > allocatable.Cpu().MilliValue() {
        return false
    }
    if podRequest.Memory().Value() > allocatable.Memory().Value() {
        return false
    }

    // Filter 2: 自定义 Taint 检查
    for _, taint := range node.Spec.Taints {
        if taint.Key == "node.kubernetes.io/unschedulable" {
            return false
        }
    }

    // Filter 3: 自定义 Node Label 匹配
    if requiredGPU, ok := pod.Annotations["requires-gpu"]; ok {
        if requiredGPU == "true" {
            if node.Labels["gpu-type"] == "" {
                return false
            }
        }
    }

    return true
}

// bindPod 执行绑定
func (s *CustomScheduler) bindPod(pod *v1.Pod, node *v1.Node) {
    binding := &v1.Binding{
        ObjectMeta: metav1.ObjectMeta{
            Name:      pod.Name,
            Namespace: pod.Namespace,
        },
        Target: v1.ObjectReference{
            Kind: "Node",
            Name: node.Name,
        },
    }
    err := s.clientset.CoreV1().Pods(pod.Namespace).Bind(
        context.TODO(), binding, metav1.CreateOptions{})
    if err != nil {
        klog.Errorf("Failed to bind pod %s to node %s: %v",
            pod.Name, node.Name, err)
    }
}

apiVersion: kubescheduler.config.k8s.io/v1
kind: KubeSchedulerConfiguration
extenders:
  - urlPrefix: "http://scheduler-extender.kube-system:8080"
    filterVerb: "filter"
    scoreVerb: "score"
    preemptVerb: "preempt"
    bindVerb: "bind"
    enableHTTPS: true
    nodeCacheCapable: true       # Extender 自己缓存 Node 状态
    weight: 5                    # Score 权重
    managedResources:
      - name: "example.com/sriov"
        ignoredByScheduler: true  # kube-scheduler 不检查此资源
    ignorable: true              # 不可用时跳过（非关键）

# 调度延迟分布（P50/P90/P99）
histogram_quantile(0.99,
    rate(scheduler_e2e_scheduling_duration_seconds_bucket[5m]))

# 调度失败速率（Unschedulable 原因分类）
rate(scheduler_schedule_attempts_total{result="unschedulable"}[5m])
rate(scheduler_schedule_attempts_total{result="error"}[5m])

# 绑定失败次数
rate(scheduler_binding_duration_seconds_count{status="failure"}[5m])

# 活跃队列 Pod 堆积
scheduler_scheduler_queue_incoming_pods_total -
  scheduler_schedule_attempts_total

# 抢占次数
rate(scheduler_preemption_attempts_total{result="successful"}[10m])

# 查看调度器日志 - 筛选特定 Pod 的调度过程
kubectl logs -n kube-system kube-scheduler-master \
  | grep "Attempting to schedule pod\|Failed to schedule\|Successfully bound"

# 查看 Pending Pod 的调度失败原因
kubectl describe pod <pending-pod> | grep -A 10 "Events"

# 查看调度器 Leader 状态
kubectl get endpoints kube-scheduler \
  -n kube-system -o yaml

# 快速定位调度瓶颈（哪个阶段耗时最长）
curl -k https://localhost:10259/metrics | grep "scheduling_duration"