关于kubernetes:kubescheduler深度剖析与开发二

为了深刻学习 kube-scheduler，本系从源码和实战角度深度学习kube-scheduler，该系列一共分6篇文章，如下：

kube-scheduler 整体架构
初始化一个 scheduler
一个 Pod 是如何调度的
如何开发一个属于本人的scheduler插件
开发一个 prefilter 扩大点的插件
开发一个 socre 扩大点的插件

上一篇，咱们说了 kube-scheduler 的整体架构，是从整体的架构方面来思考的，本文咱们说说 kube-scheduler 是如何初始化进去的，kube-scheduler 外面都有些什么货色。

因为 kube-scheduler 源码内容比拟多，对于那些不是要害的货色，就疏忽不做探讨。

Scheduler之Profiles

上面咱们先看下 Scheduler 的构造

type Scheduler struct {   Cache internalcache.Cache      Extenders []framework.Extender      NextPod func() *framework.QueuedPodInfo      FailureHandler FailureHandlerFn      SchedulePod func(ctx context.Context, fwk framework.Framework, state  *framework.CycleState, pod *v1.Pod) (ScheduleResult, error)      StopEverything <-chan struct{}      SchedulingQueue internalqueue.SchedulingQueue      Profiles profile.Map      client clientset.Interface      nodeInfoSnapshot *internalcache.Snapshot      percentageOfNodesToScore int32      nextStartNodeIndex int}

上一篇咱们说过，为一个 Pod 抉择一个 Node 是依照固定程序运行扩大点的；在扩大点内，是依照插件注册的程序运行插件，如下图

下面的这些扩大点在 kube-scheduler 中是固定的，而且也不反对减少扩大点（实际上有这些扩大点曾经足够了），而且扩大点程序也是固定执行的。

下图是插件（以preFilter为例）运行的程序，扩大点内的插件，你既能够调整插件的执行程序（理论很少会批改默认的插件执行程序），能够敞开某个内置插件，还能够减少本人开发的插件。

那么这些插件是怎么注册的，注册在哪里呢，本人开发的插件又是怎么加进去的呢？

咱们来看下 Scheduler 外面最重要的一个成员：Profiles profile.Map

// 门路：pkg/scheduler/profile/profile.go// Map holds frameworks indexed by scheduler name.type Map map[string]framework.Framework

Profiles 是一个 key 为 scheduler name，value 是 framework.Framework 的map，示意依据 scheduler name 来获取 framework.Framework 类型的值，所以能够有多个scheduler。或者你在应用 k8s 的时候没有关注过 pod 或 deploment 外面的 scheduler，因为你没有指定的话，k8s 就会主动设置为默认的调度器，下图是 deployment 中未指定 schedulerName 被设置了默认调度器的一个deployment

](/img/bVc7ywM)

假如当初我想要应用本人开发的一个名叫 my-scheduler-1 的调度器，这个调度器在 preFilter 扩大点中减少了 zoneLabel 插件，怎么做？

应用 kubeadm 部署的 k8s 集群，会在 /etc/kubernetes/manifests 目录下创立 kube-scheduler.yaml 文件，kubelet 会依据这个文件主动拉起来一个动态 Pod，一个 kube-scheduler pod就被创立了，而且这个 kube-scheduler 运行的参数是间接在命令行上指定的。

apiVersion: v1kind: Podmetadata:  creationTimestamp: null  labels:    component: kube-scheduler    tier: control-plane  name: kube-scheduler  namespace: kube-systemspec:  containers:  - command:    - kube-scheduler    - --address=0.0.0.0    - --authentication-kubeconfig=/etc/kubernetes/scheduler.conf    - --authorization-kubeconfig=/etc/kubernetes/scheduler.conf    - --bind-address=127.0.0.1    - --kubeconfig=/etc/kubernetes/scheduler.conf    - --leader-elect=true    image: k8s.gcr.io/kube-scheduler:v1.16.8        ....

其实 kube-scheduler 运行的时候能够指定配置文件，而不间接把参数写在启动命令上,如下模式。

./kube-scheduler --config /etc/kube-scheduler.conf

于是乎，咱们就能够在配置文件中配置咱们调度器的插件了

apiVersion: kubescheduler.config.k8s.io/v1beta2kind: KubeSchedulerConfigurationleaderElection:  leaderElect: trueclientConnection:  kubeconfig: "/etc/kubernetes/scheduler.conf"profiles:- schedulerName: my-scheduler  plugins:    preFilter:      enabled:        - name: zoneLabel      disabled:        - name: NodePorts

咱们能够应用 enabled，disabled 开关来敞开或关上某个插件。
通过配置文件，还能够管制扩大点的调用程序，规定如下：

如果某个扩大点没有配置对应的扩大，调度框架将应用默认插件中的扩大
如果为某个扩大点配置且激活了扩大，则调度框架将先调用默认插件的扩大，再调用配置中的扩大
默认插件的扩大始终被最先调用，而后依照 KubeSchedulerConfiguration 中扩大的激活 enabled 程序一一调用扩大点的扩大
能够先禁用默认插件的扩大，而后在 enabled 列表中的某个地位激活默认插件的扩大，这种做法能够扭转默认插件的扩大被调用时的程序

还能够增加多个调度器，在 deployment 等控制器中指定本人想要应用的调度器即可：

apiVersion: kubescheduler.config.k8s.io/v1beta2kind: KubeSchedulerConfigurationleaderElection:  leaderElect: trueclientConnection:  kubeconfig: "/etc/kubernetes/scheduler.conf"profiles:- schedulerName: my-scheduler-1  plugins:    preFilter:      enabled:        - name: zoneLabel        - schedulerName: my-scheduler-2  plugins:    queueSort:      enabled:        - name: mySort

当然了，当初咱们在配置文件中定义的 mySort，zoneLabel 这样的插件还不能应用，咱们须要开发具体的插件注册进去，能力失常运行，前面的文章会具体讲。

好了，当初 Profiles 成员（一个map）曾经蕴含了两个元素，{"my-scheduler-1": framework.Framework ,"my-scheduler-2": framework.Framework}。当一个 Pod 须要被调度的时候，kube-scheduler 会先取出 Pod 的 schedulerName 字段的值，而后通过 Profiles[schedulerName]，拿到 framework.Framework 对象，进而应用这个对象开始调度，咱们能够用上面这种张图总结下下面形容的各个对象的关系。

那么重点就来到了 framework.Framework ，上面是 framework.Framework 的定义：

// pkg/scheduler/framework/interface.gotype Framework interface {   Handle      QueueSortFunc() LessFunc   RunPreFilterPlugins(ctx context.Context, state *CycleState, pod *v1.Pod) (*PreFilterResult, *Status)   RunPostFilterPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, filteredNodeStatusMap NodeToStatusMap) (*PostFilterResult, *Status)   RunPreBindPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string) *Status   RunPostBindPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string)   RunReservePluginsReserve(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string) *Status   RunReservePluginsUnreserve(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string)   RunPermitPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string) *Status   WaitOnPermit(ctx context.Context, pod *v1.Pod) *Status   RunBindPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string) *Status   HasFilterPlugins() bool   HasPostFilterPlugins() bool   HasScorePlugins() bool   ListPlugins() *config.Plugins   ProfileName() string}

Framework 是一个接口，须要实现的办法大部分模式为：Run*Plugins，也就是运行某个扩大点的插件，那么只有实现这个 Framework 接口就能够对 Pod 进行调度了。那么须要用户本人实现么？答案是不必，kube-scheduler 曾经有一个该接口的实现：frameworkImpl

// pkg/scheduler/framework/runtime/framework.gotype frameworkImpl struct {    registry             Registry    snapshotSharedLister framework.SharedLister    waitingPods          *waitingPodsMap    scorePluginWeight    map[string]int    queueSortPlugins     []framework.QueueSortPlugin    preFilterPlugins     []framework.PreFilterPlugin    filterPlugins        []framework.FilterPlugin    postFilterPlugins    []framework.PostFilterPlugin    preScorePlugins      []framework.PreScorePlugin    scorePlugins         []framework.ScorePlugin    reservePlugins       []framework.ReservePlugin    preBindPlugins       []framework.PreBindPlugin    bindPlugins          []framework.BindPlugin    postBindPlugins      []framework.PostBindPlugin    permitPlugins        []framework.PermitPlugin    clientSet       clientset.Interface    kubeConfig      *restclient.Config    eventRecorder   events.EventRecorder    informerFactory informers.SharedInformerFactory    metricsRecorder *metricsRecorder    profileName     string    extenders []framework.Extender    framework.PodNominator    parallelizer parallelize.Parallelizer}

frameworkImpl 这个构造体外面蕴含了每个扩大点插件数组，所以某个扩大点要被执行的时候，只有遍历这个数组外面的所有插件，而后执行这些插件就能够了。咱们看看 framework.FilterPlugin 是怎么定义的（其余的也相似）：

type Plugin interface {    Name() string}type FilterPlugin interface {    Plugin    Filter(ctx context.Context, state *CycleState, pod *v1.Pod, nodeInfo *NodeInfo) *Status}

插件数组的类型是一个接口，那么某个插件只有实现了这个接口就能够被运行。实际上，咱们后面说的那些默认插件，都实现了这个接口，在目录 pkg/scheduler/framework/plugins 目录上面蕴含了所有内置插件的实现，次要就是对下面说的这个插件接口的实现。咱们能够简略用图形容下 Pod被调度的时候执行插件的流程

那么这些默认插件是怎么加到framework外面的，自定义插件又是怎么加进来的呢？

分三步：

依据配置文件（--config指定的）、零碎默认的插件，依照扩大点生成须要被加载的插件数组（包含插件名字，权重信息），也就是初始化 KubeSchedulerConfiguration 中的 Profiles 成员。

type KubeSchedulerConfiguration struct {    metav1.TypeMeta  Parallelism int32  LeaderElection componentbaseconfig.LeaderElectionConfiguration  ClientConnection componentbaseconfig.ClientConnectionConfiguration    HealthzBindAddress string  MetricsBindAddress string  componentbaseconfig.DebuggingConfiguration  PercentageOfNodesToScore int32  PodInitialBackoffSeconds int64  PodMaxBackoffSeconds int64  Profiles []KubeSchedulerProfile  Extenders []Extender}

创立 registry 汇合，这个汇合内是每个插件实例化函数，也就是插件名字->插件实例化函数的映射，艰深一点说就是通知零碎：1.我叫王二； 2. 你应该怎么把我创立进去。那么张三、李四、王五别离通知零碎怎么创立本人，就组成了这个汇合。

type PluginFactory = func(configuration runtime.Object, f framework.Handle) (framework.Plugin, error)type Registry map[string]PluginFactory

这个汇合是内置（叫inTree）默认的插件映射和用户自定义（outOfTree）的插件映射的并集，内置的映射通过上面函数创立：

// pkg/scheduler/framework/plugins/registry.gofunc NewInTreeRegistry() runtime.Registry {    fts := plfeature.Features{        EnableReadWriteOncePod:                       feature.DefaultFeatureGate.Enabled(features.ReadWriteOncePod),        EnableVolumeCapacityPriority:                 feature.DefaultFeatureGate.Enabled(features.VolumeCapacityPriority),        EnableMinDomainsInPodTopologySpread:          feature.DefaultFeatureGate.Enabled(features.MinDomainsInPodTopologySpread),        EnableNodeInclusionPolicyInPodTopologySpread: feature.DefaultFeatureGate.Enabled(features.NodeInclusionPolicyInPodTopologySpread),    }    return runtime.Registry{        selectorspread.Name:                  selectorspread.New,        imagelocality.Name:                   imagelocality.New,        tainttoleration.Name:                 tainttoleration.New,        nodename.Name:                        nodename.New,        nodeports.Name:                       nodeports.New,        nodeaffinity.Name:                    nodeaffinity.New,        podtopologyspread.Name:               runtime.FactoryAdapter(fts, podtopologyspread.New),        nodeunschedulable.Name:               nodeunschedulable.New,        noderesources.Name:                   runtime.FactoryAdapter(fts, noderesources.NewFit),        noderesources.BalancedAllocationName: runtime.FactoryAdapter(fts, noderesources.NewBalancedAllocation),        volumebinding.Name:                   runtime.FactoryAdapter(fts, volumebinding.New),        volumerestrictions.Name:              runtime.FactoryAdapter(fts, volumerestrictions.New),        volumezone.Name:                      volumezone.New,        nodevolumelimits.CSIName:             runtime.FactoryAdapter(fts, nodevolumelimits.NewCSI),        nodevolumelimits.EBSName:             runtime.FactoryAdapter(fts, nodevolumelimits.NewEBS),        nodevolumelimits.GCEPDName:           runtime.FactoryAdapter(fts, nodevolumelimits.NewGCEPD),        nodevolumelimits.AzureDiskName:       runtime.FactoryAdapter(fts, nodevolumelimits.NewAzureDisk),        nodevolumelimits.CinderName:          runtime.FactoryAdapter(fts, nodevolumelimits.NewCinder),        interpodaffinity.Name:                interpodaffinity.New,        queuesort.Name:                       queuesort.New,        defaultbinder.Name:                   defaultbinder.New,        defaultpreemption.Name:               runtime.FactoryAdapter(fts, defaultpreemption.New),    }}

那么用户自定义的插件怎么来的呢？这里咱们先不开展，在前面插件开发的时候再具体讲，不影响咱们了解。咱们假如用户自定义的也曾经生成了 registry，上面的代码就是把他们合并在一起

// pkg/scheduler/scheduler.goregistry := frameworkplugins.NewInTreeRegistry()if err := registry.Merge(options.frameworkOutOfTreeRegistry); err != nil {    return nil, err}

当初内置插件和零碎默认插件的实例化函数映射曾经创立好了

将（1）中每个扩大点的每个插件（就是插件名字）拿进去，去（2）的映射（map）中获取实例化函数，而后运行这个实例化函数，最初把这个实例化进去的插件（能够被运行的）追加到下面提到过的 frameworkImpl 中对应扩大点数组中，这样前面要运行某个扩大点插件的时候就能够遍历运行就能够了。咱们能够把上述过程用下图示意

Scheduler 之 SchedulingQueue

下面咱们介绍了 Scheduler 第一个要害成员 Profiles 的初始化和作用，上面咱们来谈谈第二个要害成员：SchedulingQueue。

// pkg/scheduler/scheduler.gopodQueue := internalqueue.NewSchedulingQueue(    profiles[options.profiles[0].SchedulerName].QueueSortFunc(),    informerFactory,    // 1s    internalqueue.WithPodInitialBackoffDuration(time.Duration(options.podInitialBackoffSeconds)*time.Second),    // 10s    internalqueue.WithPodMaxBackoffDuration(time.Duration(options.podMaxBackoffSeconds)*time.Second),    internalqueue.WithPodNominator(nominator),    internalqueue.WithClusterEventMap(clusterEventMap),    // 5min    internalqueue.WithPodMaxInUnschedulablePodsDuration(options.podMaxInUnschedulablePodsDuration),)

func NewSchedulingQueue(    lessFn framework.LessFunc,    informerFactory informers.SharedInformerFactory,    opts ...Option) SchedulingQueue {    return NewPriorityQueue(lessFn, informerFactory, opts...)}

type PriorityQueue struct {    framework.PodNominator  stop  chan struct{}  clock clock.Clock  podInitialBackoffDuration time.Duration  podMaxBackoffDuration time.Duration  podMaxInUnschedulablePodsDuration time.Duration  lock sync.RWMutex  cond sync.Cond  activeQ *heap.Heap  podBackoffQ *heap.Heap  unschedulablePods *UnschedulablePods  schedulingCycle int64  moveRequestCycle int64  clusterEventMap map[framework.ClusterEvent]sets.String  closed bool  nsLister listersv1.NamespaceLister}

SchedulingQueue 是一个 internalqueue.SchedulingQueue 的接口类型，PriorityQueue 对这个接口进行了实现，创立 Scheduler 的时候 SchedulingQueue 会被 PriorityQueue 类型对象赋值。

PriorityQueue 中有要害的3个成员：activeQ、podBackoffQ、unschedulablePods。

activeQ 是一个优先队列，用来寄存待调度的 Pod，Pod 依照优先级寄存在队列中
podBackoffQ 用来寄存异样的 Pod，该队列外面的 Pod 会期待肯定工夫后被挪动到 activeQ 外面从新被调度
unschedulablePods 中会寄存调度失败的 Pod，它不是队列，而是应用 map 来寄存的，这个 map 外面的 Pod 在肯定条件下会被挪动到 activeQ 或 podBackoffQ 中

PriorityQueue 还有两个办法：flushUnschedulablePodsLeftover 和 flushBackoffQCompleted

flushUnschedulablePodsLeftover：调度失败的 Pod 如果满足肯定条件，这个函数会将这种 Pod 挪动到 activeQ 或 podBackoffQ
flushBackoffQCompleted：运行异样的 Pod 等待时间实现后，flushBackoffQCompleted 将该 Pod 挪动到 activeQ

Scheduler 在启动的时候，会创立2个协程来定期运行这两个函数

func (p *PriorityQueue) Run() {   go wait.Until(p.flushBackoffQCompleted, 1.0*time.Second, p.stop)   go wait.Until(p.flushUnschedulablePodsLeftover, 30*time.Second, p.stop)}

下面是定期对 Pod 在这些队列之间的转换，那么除了定期刷新的形式，还有上面状况也会触发队列转换：

有新节点退出集群
节点配置或状态发生变化
曾经存在的 Pod 发生变化
集群内有Pod被删除

至于他们之间是如何转换的，咱们在下一篇文章外面具体介绍

Scheduler 之 cache

要说 cache 最大的作用就是晋升 Scheduler 的效率，升高 kube-apiserver(实质是 etcd)的压力，在调用各个插件计算的时候所须要的 Node 信息和其余 Pod 信息都缓存在本地，在须要应用的时候间接从缓存获取即可，而不须要调用 api 从 kube-apiserver 获取。cache 类型是 internalcache.Cache 的接口，cacheImpl 实现了这个接口。上面是 cacheImpl 的构造

type Cache interface   NodeCount() int  PodCount() (int, error)  AssumePod(pod *v1.Pod) error  FinishBinding(pod *v1.Pod) error  ForgetPod(pod *v1.Pod) error  AddPod(pod *v1.Pod) error  UpdatePod(oldPod, newPod *v1.Pod) error  RemovePod(pod *v1.Pod) error  GetPod(pod *v1.Pod) (*v1.Pod, error)  IsAssumedPod(pod *v1.Pod) (bool, error)  AddNode(node *v1.Node) *framework.NodeInfo  UpdateNode(oldNode, newNode *v1.Node) *framework.NodeInfo  RemoveNode(node *v1.Node) error  UpdateSnapshot(nodeSnapshot *Snapshot) error  Dump() *Dump}

type cacheImpl struct {  stop   <-chan struct{}  ttl    time.Duration  period time.Duration  mu sync.RWMutex  assumedPods sets.String  podStates map[string]*podState  nodes     map[string]*nodeInfoListItem  headNode *nodeInfoListItem  nodeTree *nodeTree  imageStates map[string]*imageState}

cacheImpl 中的 nodes 寄存集群内所有 Node 信息；podStates 寄存所有 Pod 信息；，assumedPods 寄存曾经调度胜利然而还没调用 kube-apiserver 的进行绑定的（也就是还没有执行 bind 插件）的Pod，须要这个缓存的起因也是为了晋升调度效率，将绑定和调度离开，因为绑定须要调用 kube-apiserver，这是一个重操作会耗费比拟多的工夫，所以 Scheduler 乐观的假如调度曾经胜利，而后返回去调度其余 Pod，而这个 Pod 就会放入 assumedPods 中，并且也会放入到 podStates 中，后续其余 Pod 在进行调度的时候，这个 Pod 也会在插件的计算范畴内（如亲和性），而后会新起协程进行最初的绑定，要是最初绑定失败了，那么这个 Pod 的信息会从 assumedPods 和 podStates 移除，并且把这个 Pod 从新放入 activeQ 中，从新被调度。

Scheduler 在启动时首先会 list 一份全量的 Pod 和 Node 数据到上述的缓存中，后续通过 watch 的形式发现变动的 Node 和 Pod，而后将变动的 Node 或 Pod 更新到上述缓存中。

Scheduler 之 NextPod 和 SchedulePod

到了这里，调度框架 framework 和调度队列 SchedulingQueue 都曾经创立进去了，当初是时候开始调度Pod了。

Scheduler 中有个成员 NextPod 会从 activeQ 队列中尝试获取一个待调度的 Pod，该函数在 SchedulePod 中被调用，如下：

// 启动 Schedulerfunc (sched *Scheduler) Run(ctx context.Context) {    sched.SchedulingQueue.Run()    go wait.UntilWithContext(ctx, sched.scheduleOne, 0)    <-ctx.Done()    sched.SchedulingQueue.Close()}// 尝试调度一个 Pod，所以 Pod 的调度入口func (sched *Scheduler) scheduleOne(ctx context.Context) {    // 会始终阻塞，直到获取到一个Pod    ......    podInfo := sched.NextPod()    ......}

NextPod 它被赋予如下函数：

// pkg/scheduler/internal/queue/scheduling_queue.gofunc MakeNextPodFunc(queue SchedulingQueue) func() *framework.QueuedPodInfo {    return func() *framework.QueuedPodInfo {        podInfo, err := queue.Pop()        if err == nil {            klog.V(4).InfoS("About to try and schedule pod", "pod", klog.KObj(podInfo.Pod))            for plugin := range podInfo.UnschedulablePlugins {                metrics.UnschedulableReason(plugin, podInfo.Pod.Spec.SchedulerName).Dec()            }            return podInfo        }        klog.ErrorS(err, "Error while retrieving next pod from scheduling queue")        return nil    }}

Pop 会始终阻塞，直到 activeQ 长度大于0，而后去取出一个 Pod 返回

// pkg/scheduler/internal/queue/scheduling_queue.gofunc (p *PriorityQueue) Pop() (*framework.QueuedPodInfo, error) {    p.lock.Lock()    defer p.lock.Unlock()    for p.activeQ.Len() == 0 {        // When the queue is empty, invocation of Pop() is blocked until new item is enqueued.        // When Close() is called, the p.closed is set and the condition is broadcast,        // which causes this loop to continue and return from the Pop().        if p.closed {            return nil, fmt.Errorf(queueClosed)        }        p.cond.Wait()    }    obj, err := p.activeQ.Pop()    if err != nil {        return nil, err    }    pInfo := obj.(*framework.QueuedPodInfo)    pInfo.Attempts++    p.schedulingCycle++    return pInfo, nil}

到了这里咱们就介绍完了 Scheduler 中最重要的几个成员，简略总结下：

Profiles: 寄存插件对象，在运行时能够遍历扩大点内的所有插件运行
SchedulerQueue：用来寄存待调度 Pod，异样 Pod，调度失败 Pod，他们互相能够转换
cache：寄存 Pod 和 Node 的信息，晋升调度效率
NextPod 和 ScheduleOne：尝试从 activeQ 获取一个 Pod，开始调度。

本文就到这，下一篇，咱们会讲一讲一个 Pod 提交后的调度流程。

我是清风徐来，一起学习k8s，支付k8s、docker等精髓学习资源

本文由博客一文多发平台 OpenWrite 公布！