kube-scheduler 源码分析 - 《kubernetes 源码分析》

总结

官方对 kube-scheduler 的调度流程描述：

kube-scheduler 目前包含两部分调度算法 predicates 和 priorities，首先执行 predicates 算法过滤部分 node 然后执行 priorities 算法为所有 node 打分，最后从所有 node 中选出分数最高的最为最佳的 node。

kubernetes 中所有组件的启动流程都是类似的，首先会解析命令行参数、添加默认值，kube-scheduler 的默认参数在中定义的。然后会执行 run 方法启动主逻辑，下面直接看 kube-scheduler 的主逻辑 run 方法执行过程。

Run() 方法主要做了以下工作：

初始化 scheduler 对象
启动 kube-scheduler server，kube-scheduler 监听 10251 和 10259 端口，10251 端口不需要认证，可以获取 healthz metrics 等信息，10259 为安全端口，需要认证
启动所有的 informer
执行 sched.Run() 方法，执行主调度逻辑

k8s.io/kubernetes/cmd/kube-scheduler/app/server.go:160

func Run(cc schedulerserverconfig.CompletedConfig, stopCh <-chan struct{}, registryOptions ...Option) error {
    ......
    // 1、初始化 scheduler 对象
    sched, err := scheduler.New(......）
    if err != nil {
        return err
    }
    // 2、启动事件广播
    if cc.Broadcaster != nil && cc.EventClient != nil {
        cc.Broadcaster.StartRecordingToSink(stopCh)
    }
    if cc.LeaderElectionBroadcaster != nil && cc.CoreEventClient != nil {
        cc.LeaderElectionBroadcaster.StartRecordingToSink(&corev1.EventSinkImpl{Interface: cc.CoreEventClient.Events("")})
    }
    ......
    // 3、启动 http server
    if cc.InsecureServing != nil {
        separateMetrics := cc.InsecureMetricsServing != nil
        handler := buildHandlerChain(newHealthzHandler(&cc.ComponentConfig, separateMetrics, checks...), nil, nil)
        if err := cc.InsecureServing.Serve(handler, 0, stopCh); err != nil {
            return fmt.Errorf("failed to start healthz server: %v", err)
        }
    }
    ......
    // 4、启动所有 informer
    go cc.PodInformer.Informer().Run(stopCh)
    cc.InformerFactory.Start(stopCh)
    cc.InformerFactory.WaitForCacheSync(stopCh)
    run := func(ctx context.Context) {
        sched.Run()
        <-ctx.Done()
    }
    ctx, cancel := context.WithCancel(context.TODO()) // TODO once Run() accepts a context, it should be used here
    defer cancel()
    go func() {
        select {
        case <-stopCh:
            cancel()
        case <-ctx.Done():
        }
    }()
    // 5、选举 leader
    if cc.LeaderElection != nil {
        ......
    }
    // 6、执行 sched.Run() 方法
    run(ctx)
    return fmt.Errorf("finished without leader elect")
}

下面看一下 scheduler.New() 方法是如何初始化 scheduler 结构体的，该方法主要的功能是初始化默认的调度算法以及默认的调度器 GenericScheduler。

创建 scheduler 配置文件
根据默认的 DefaultProvider 初始化 schedulerAlgorithmSource 然后加载默认的预选及优选算法，然后初始化 GenericScheduler
若启动参数提供了 policy config 则使用其覆盖默认的预选及优选算法并初始化 GenericScheduler，不过该参数现已被弃用

k8s.io/kubernetes/pkg/scheduler/scheduler.go:166

func New(......) (*Scheduler, error) {
    ......
    // 1、创建 scheduler 的配置文件
    configurator := factory.NewConfigFactory(&factory.ConfigFactoryArgs{
          ......
      })
    var config *factory.Config
    source := schedulerAlgorithmSource
    // 2、加载默认的调度算法
    switch {
    case source.Provider != nil:
        // 使用默认的 ”DefaultProvider“ 初始化 config
        sc, err := configurator.CreateFromProvider(*source.Provider)
        if err != nil {
            return nil, fmt.Errorf("couldn't create scheduler using provider %q: %v", *source.Provider, err)
        }
        config = sc
    case source.Policy != nil:
        // 通过启动时指定的 policy source 加载 config
    ......
    default:
        return nil, fmt.Errorf("unsupported algorithm source: %v", source)
    // Additional tweaks to the config produced by the configurator.
    config.Recorder = recorder
    config.DisablePreemption = options.disablePreemption
    config.StopEverything = stopCh
    // 3.创建 scheduler 对象
    sched := NewFromConfig(config)
    ......
    return sched, nil
}

k8s.io/kubernetes/pkg/scheduler/factory/factory.go:527

然后继续看 Run() 方法中最后执行的 sched.Run() 调度循环逻辑，若 informer 中的 cache 同步完成后会启动一个循环逻辑执行 sched.scheduleOne 方法。

k8s.io/kubernetes/pkg/scheduler/scheduler.go:313

func (sched *Scheduler) Run() {
    if !sched.config.WaitForCacheSync() {
        return
    }
    go wait.Until(sched.scheduleOne, 0, sched.config.StopEverything)
}

scheduleOne() 每次对一个 pod 进行调度，主要有以下步骤：

从 scheduler 调度队列中取出一个 pod，如果该 pod 处于删除状态则跳过
执行调度逻辑 sched.schedule() 返回通过预算及优选算法过滤后选出的最佳 node
如果过滤算法没有选出合适的 node，则返回 core.FitError
若没有合适的 node 会判断是否启用了抢占策略，若启用了则执行抢占机制
判断是否需要 VolumeScheduling 特性
执行 reserve plugin
pod 对应的 spec.NodeName 写上 scheduler 最终选择的 node，更新 scheduler cache
请求 apiserver 异步处理最终的绑定操作，写入到 etcd
执行 permit plugin
执行 prebind plugin
执行 postbind plugin

k8s.io/kubernetes/pkg/scheduler/scheduler.go:515

func (sched *Scheduler) scheduleOne() {
    fwk := sched.Framework
    pod := sched.NextPod()
    if pod == nil {
        return
    }
    // 1.判断 pod 是否处于删除状态
    if pod.DeletionTimestamp != nil {
        ......
    }
    // 2.执行调度策略选择 node
    start := time.Now()
    pluginContext := framework.NewPluginContext()
    scheduleResult, err := sched.schedule(pod, pluginContext)
    if err != nil {
        if fitError, ok := err.(*core.FitError); ok {
            // 3.若启用抢占机制则执行
            if sched.DisablePreemption {
                ......
            } else {
                preemptionStartTime := time.Now()
                sched.preempt(pluginContext, fwk, pod, fitError)
                ......
            }
            ......
            metrics.PodScheduleFailures.Inc()
        } else {
            klog.Errorf("error selecting node for pod: %v", err)
            metrics.PodScheduleErrors.Inc()
        }
        return
    }
    ......
    assumedPod := pod.DeepCopy()
    // 4.判断是否需要 VolumeScheduling 特性
    allBound, err := sched.assumeVolumes(assumedPod, scheduleResult.SuggestedHost)
    if err != nil {
        klog.Errorf("error assuming volumes: %v", err)
        metrics.PodScheduleErrors.Inc()
        return
    }
    // 5.执行 "reserve" plugins
    if sts := fwk.RunReservePlugins(pluginContext, assumedPod, scheduleResult.SuggestedHost); !sts.IsSuccess() {
        .....
    }
    // 6.为 pod 设置 NodeName 字段，更新 scheduler 缓存
    err = sched.assume(assumedPod, scheduleResult.SuggestedHost)
    if err != nil {
    }
    // 7.异步请求 apiserver
    go func() {
        if !allBound {
            err := sched.bindVolumes(assumedPod)
            if err != nil {
                ......
                return
            }
        }
        // 8.执行 "permit" plugins
        permitStatus := fwk.RunPermitPlugins(pluginContext, assumedPod, scheduleResult.SuggestedHost)
        if !permitStatus.IsSuccess() {
            ......
        }
        // 9.执行 "prebind" plugins
        preBindStatus := fwk.RunPreBindPlugins(pluginContext, assumedPod, scheduleResult.SuggestedHost)
        if !preBindStatus.IsSuccess() {
            ......
        }
        err := sched.bind(assumedPod, scheduleResult.SuggestedHost, pluginContext)
        ......
        if err != nil {
            ......
        } else {
            ......
            // 10.执行 "postbind" plugins
            fwk.RunPostBindPlugins(pluginContext, assumedPod, scheduleResult.SuggestedHost)
        }
    }()
}

scheduleOne() 中通过调用 sched.schedule() 来执行预选与优选算法处理：

k8s.io/kubernetes/pkg/scheduler/scheduler.go:337

sched.Algorithm 是一个 interface，主要包含四个方法，GenericScheduler 是其具体的实现：

type ScheduleAlgorithm interface {
    Schedule(*v1.Pod, *framework.PluginContext) (scheduleResult ScheduleResult, err error)
    Preempt(*framework.PluginContext, *v1.Pod, error) (selectedNode *v1.Node, preemptedPods []*v1.Pod, cleanupNominatedPods []*v1.Pod, err error)
    Predicates() map[string]predicates.FitPredicate
    Prioritizers() []priorities.PriorityConfig
}

Schedule()：正常调度逻辑，包含预算与优选算法的执行
Preempt()：抢占策略，在 pod 调度发生失败的时候尝试抢占低优先级的 pod，函数返回发生抢占的 node，被抢占的 pods 列表，nominated node name 需要被移除的 pods 列表以及 error
Predicates()：predicates 算法列表
Prioritizers()：prioritizers 算法列表

kube-scheduler 提供的默认调度为 DefaultProvider，DefaultProvider 配置的 predicates 和 priorities policies 在 k8s.io/kubernetes/pkg/scheduler/algorithmprovider/defaults/defaults.go 中定义，算法具体实现是在 k8s.io/kubernetes/pkg/scheduler/algorithm/predicates/ 和k8s.io/kubernetes/pkg/scheduler/algorithm/priorities/ 中，默认的算法如下所示：

pkg/scheduler/algorithmprovider/defaults/defaults.go

func defaultPredicates() sets.String {
    return sets.NewString(
        predicates.NoVolumeZoneConflictPred,
        predicates.MaxEBSVolumeCountPred,
        predicates.MaxGCEPDVolumeCountPred,
        predicates.MaxAzureDiskVolumeCountPred,
        predicates.MaxCSIVolumeCountPred,
        predicates.MatchInterPodAffinityPred,
        predicates.NoDiskConflictPred,
        predicates.GeneralPred,
        predicates.CheckNodeMemoryPressurePred,
        predicates.CheckNodeDiskPressurePred,
        predicates.CheckNodePIDPressurePred,
        predicates.CheckNodeConditionPred,
        predicates.PodToleratesNodeTaintsPred,
        predicates.CheckVolumeBindingPred,
    )
}
func defaultPriorities() sets.String {
    return sets.NewString(
        priorities.SelectorSpreadPriority,
        priorities.InterPodAffinityPriority,
        priorities.LeastRequestedPriority,
        priorities.BalancedResourceAllocation,
        priorities.NodePreferAvoidPodsPriority,
        priorities.NodeAffinityPriority,
        priorities.TaintTolerationPriority,
        priorities.ImageLocalityPriority,
    )
}

下面继续看 sched.Algorithm.Schedule() 调用具体调度算法的过程：

检查 pod pvc 信息
执行 prefilter plugins
获取 scheduler cache 的快照，每次调度 pod 时都会获取一次快照
执行 g.findNodesThatFit() 预选算法
执行 postfilter plugin
若 node 为 0 直接返回失败的 error，若 node 数为1 直接返回该 node
执行 g.priorityMetaProducer() 获取 metaPrioritiesInterface，计算 pod 的metadata，检查该 node 上是否有相同 meta 的 pod
执行 PrioritizeNodes() 算法
执行通过得分选择一个最佳的 node

k8s.io/kubernetes/pkg/scheduler/core/generic_scheduler.go:186

至此，scheduler 的整个过程分析完毕。

本文主要对于 kube-scheduler v1.16 的调度流程进行了分析，但其中有大量的细节都暂未提及，包括预选算法以及优选算法的具体实现、优先级与抢占调度的实现、framework 的使用及实现，因篇幅有限，部分内容会在后文继续说明。

参考：

The Kubernetes Scheduler