[kubernetes系列]Scheduler模块深度讲解

调度器的职责是负责将Pod调度到最合适的Node上，但是要实现它并不是易事，需要考虑很多方面。(1) 公平性：调度后集群各个node应该保持均衡的状态。(2) 性能：不能成为集群的性能瓶颈。 (3) 扩展性：用户能根据自身需求定制调度器和调度算法。(4) 限制：需要考虑多种限制条件，例如亲缘性，优先级，Qos等。(5) 代码的优雅性，虽然不是一定要的^^。接下来带着这些问题往下看。

二，调度器源码分析

接下来一边说明调度的步骤，一边看源码(只分析主干代码)，然后思考有没有更好的方式。调度这里，分成几个重要的步骤：1，初始化调度器；2，获取未调度的Pod开始调度；3，预调度，优调度和扩展；4，调度失败则发起抢占。这里只跟着流程走，具体有必要更详细解读的放在下面几部分。 本文代码基于1.12.1版本

(1) 初始化调度器

先生成configfatotry(可通过不同参数生成不同config)，然后调度器可通过policy文件，policy configmap，或者指定provider，通过configfactory来创建config，再由config生成scheduler。我们可以在启动时候选择policy启动或者provider启动scheduler模块。不管通过哪种方式创建，最终都会进入到CreateFromKeys去创建scheduler。

首先看如何获取provider和policy

func NewSchedulerConfig(s schedulerserverconfig.CompletedConfig) (*scheduler.Config, error) {
   // 判断是否开启StorageClass
	var storageClassInformer storageinformers.StorageClassInformer
	if utilfeature.DefaultFeatureGate.Enabled(features.VolumeScheduling) {
		storageClassInformer = s.InformerFactory.Storage().V1().StorageClasses()
	}

	//  生成configfactory,包含所有需要的informer
	configurator := factory.NewConfigFactory(&factory.ConfigFactoryArgs{
		SchedulerName:                  s.ComponentConfig.SchedulerName,
		Client:                         s.Client,
		NodeInformer:                   s.InformerFactory.Core().V1().Nodes(),
		.....
	})

	source := s.ComponentConfig.AlgorithmSource
	var config *scheduler.Config
	switch {
	//根据准备好的provider生成config，
	case source.Provider != nil:
		sc, err := configurator.CreateFromProvider(*source.Provider)
		config = sc
	// 根据policy生成config
	case source.Policy != nil:
		policy := &schedulerapi.Policy{}
		switch {
		// 根据policy文件生成
		case source.Policy.File != nil:
			......
		// 根据policy configmap生成
		case source.Policy.ConfigMap != nil:
		        ......
		}
		sc, err := configurator.CreateFromConfig(*policy)
		config = sc
	}
	config.DisablePreemption = s.ComponentConfig.DisablePreemption
	return config, nil
}
复制代码

上面的CreateFromProvider和CreateFromConfig最终都会进入到CreateFromKeys，去初始化系统自带的GenericScheduler。

// 根据已注册的 predicate keys and priority keys生成配置
func (c *configFactory) CreateFromKeys(predicateKeys, priorityKeys sets.String, extenders []algorithm.SchedulerExtender) (*scheduler.Config, error) {
      // 获取所有的predicate函数 
	predicateFuncs, err := c.GetPredicates(predicateKeys)
	// 获取priority配置(为什么不是返回函数？因为包含了权重，而且使用的是map-reduce)
	priorityConfigs, err := c.GetPriorityFunctionConfigs(priorityKeys)
	// metaproducer都是用来获取metadata信息，例如affinity，request，limit等
	priorityMetaProducer, err := c.GetPriorityMetadataProducer()
	predicateMetaProducer, err := c.GetPredicateMetadataProducer()
	algo := core.NewGenericScheduler(
		c.podQueue,   //调度队列。默认使用优先级队列
		predicateFuncs,   // predicate算法函数链
		predicateMetaProducer,    
		priorityConfigs,  // priority算法链
		priorityMetaProducer,
		extenders,    // 扩展过滤器
		......
	)

	podBackoff := util.CreateDefaultPodBackoff()
}
复制代码

到这里scheduler.config就初始化了，如果要接着往后面看，我们可以看一下scheduler.config的定义。将会大大帮助我们进行理解。

type Config struct {
       // 调度中的pod信息，保证不冲突
       SchedulerCache schedulercache.Cache
      //  上面定义的GenericScheduler就实现了该接口，所以会赋值进来，这是最重要的字段
	Algorithm  algorithm.ScheduleAlgorithm
	//  驱逐者，产生抢占时候出场
	PodPreemptor PodPreemptor
      //  获取下个未调度的pod
	NextPod func() *v1.Pod
      // 容错机制，如果调用pod出错，使用该函数进行处理（重新加入到调度队列）
	Error func(*v1.Pod, error)
}
复制代码

(2) 调度逻辑

调度逻辑包括了筛选合适node，优先级队列，调度，抢占等逻辑，比较复杂，接下来慢慢理顺。

2.1 调度

首先看一小段主要代码，这代码已经把调度逻辑的大体交代了，再基于这主要的代码展开分析。

func (sched *Scheduler) scheduleOne() {
      // 获取下一个等待调度的pod
	pod := sched.config.NextPod()
	// 尝试将pod绑定到node上
	suggestedHost, err := sched.schedule(pod)
	if err != nil {
		if fitError, ok := err.(*core.FitError); ok {
			// 绑定出错则发起抢占
			sched.preempt(pod, fitError)
			metrics.PreemptionAttempts.Inc()
		}
		return
	}
	allBound, err := sched.assumeVolumes(assumedPod, suggestedHost)
}
复制代码

2.1.1 获取下个等待调度的pod

从初始化调度器的源码分析中，我们知道，使用的队列是优先级队列，那么此时则是从优先级队列中获取优先级最高的pod。

func (c *configFactory) getNextPod() *v1.Pod {
	pod, err := c.podQueue.Pop()
}
复制代码

2.1.2 选择合适的node

通过predicate和prioritize算法，然后选择出一个节点，把给定的pod调度到节点上。最后如果还有extender，还需要通过extender

func (g *genericScheduler) Schedule(pod *v1.Pod, nodeLister algorithm.NodeLister) (string, error) {
      // 获取所以node	
	nodes, err := nodeLister.List()
	// cache中保存调度中需要的pod和node数据，需要更新到最新
	err = g.cache.UpdateNodeNameToInfoMap(g.cachedNodeInfoMap)
	// 过滤出合适调度的node集合
	filteredNodes, failedPredicateMap, err := g.findNodesThatFit(pod, nodes)
	//  返回合适调度的node的优先级排序
	priorityList, err := PrioritizeNodes(pod, g.cachedNodeInfoMap, metaPrioritiesInterface, g.prioritizers, filteredNodes, g.extenders)
	//  选择处一个节点返回
	return g.selectHost(priorityList)
}
复制代码

上面包括了node是如何被选择出来的大体逻辑，接下来粗略看看每个步骤。 过滤出合适调度的node集合最后会调用到下面这个函数

func podFitsOnNode(...) (bool, []algorithm.PredicateFailureReason, error) {
        // 循环遍历所有predicate函数，然后调用
	for _, predicateKey := range predicates.Ordering() {
		if predicate, exist := predicateFuncs[predicateKey]; exist {
		       //调用函数
        	       if eCacheAvailable {
        			fit, reasons, err = nodeCache.RunPredicate(predicate, predicateKey, pod, metaToUse, nodeInfoToUse, equivClass, cache)
        		} else {
        			fit, reasons, err = predicate(pod, metaToUse, nodeInfoToUse)
        		}
        		// 不合适则记录
			if !fit {
				failedPredicates = append(failedPredicates, reasons...)
			}
		}
	}
	return len(failedPredicates) == 0, failedPredicates, nil
}
复制代码

过滤出node后，我们还需要给这些node排序，越适合调度的优先级越高。这里不分析了，思路跟过滤那里差不多，不过使用的map reduce来计算。

2.3 抢占

如果正常调度无法调度到node，那么就会发起抢占逻辑，选择一个node，驱逐低优先级的pod。这个节点需要满足各种需求(把低优先级pod驱逐后资源必须能满足该pod，亲和性检查等)

func (g *genericScheduler) Preempt(pod *v1.Pod, nodeLister algorithm.NodeLister, scheduleErr error) (*v1.Node, []*v1.Pod, []*v1.Pod, error) {
	allNodes, err := nodeLister.List()
	potentialNodes := nodesWherePreemptionMightHelp(allNodes, fitError.FailedPredicates)
	// 获取PDB(会尽力保证PDB)
	pdbs, err := g.cache.ListPDBs(labels.Everything())
	// 选择出可以抢占的node集合
	nodeToVictims, err := selectNodesForPreemption(pod, g.cachedNodeInfoMap, potentialNodes, g.predicates,
		g.predicateMetaProducer, g.schedulingQueue, pdbs)
	nodeToVictims, err = g.processPreemptionWithExtenders(pod, nodeToVictims)
	// 选择出一个节点发生抢占
	candidateNode := pickOneNodeForPreemption(nodeToVictims)
	// 更新低优先级的nomination
	nominatedPods := g.getLowerPriorityNominatedPods(pod, candidateNode.Name)
	if nodeInfo, ok := g.cachedNodeInfoMap[candidateNode.Name]; ok {
		return nodeInfo.Node(), nodeToVictims[candidateNode].Pods, nominatedPods, err
	}
}
复制代码

2.3.1，抢占逻辑分析

调度器会选择一个pod P尝试进行调度，如果没有node满足条件，那么会触发抢占逻辑

1，寻找合适的node N，如果有一组node都符合，那么会选择拥有最低优先级的一组pod的node，如果这些pod有PDB保护或者驱逐后还是无法满足P的要求，那么会去寻找高点优先级的。 1，当找到适合P进行调度的node N时候，会从该node删除一个或者多个pod(优先级低于P，且删除后能让P进行调度) 2，pod删除时候，需要一个优雅关闭的时间，P会重新进入队列，等待下次调度。 3，会在P中的status字段设置nominatedNodeName为N的name(该字段为了在P抢占资源后等待下次调度的过程中，让调度器知道该node已经发生了抢占，P期望落在该node上)。 4，如果在N资源释放完后，有个比P优先级更高的pod调度到N上，那么P可能无法调度到N上了，此时会清楚P的nominatedNodeName字段。如果在N上的pod优雅关闭的过程中，出现了另一个可供P调度的node，那么P将会调度到该node，则会造成nominatedNodeName和实际的node名称不符合，同时，N上的pod还是会被驱逐。

三，调度算法分析

1，predicate

在predicates.go中说明了目前提供的各个算法，多达20多种，下面列出几种

MatchInterPodAffinity：检查pod和其他pod是否符合亲和性规则
CheckNodeCondition: 检查Node的状况
MatchNodeSelector：检查Node节点的label定义是否满足Pod的NodeSelector属性需求
PodFitsResources：检查主机的资源是否满足Pod的需求，根据实际已经分配的资源（request）做调度，而不是使用已实际使用的资源量做调度
PodFitsHostPorts：检查Pod内每一个容器所需的HostPort是否已被其它容器占用，如果有所需的HostPort不满足需求，那么Pod不能调度到这个主机上
HostName：检查主机名称是不是Pod指定的NodeName
NoDiskConflict：检查在此主机上是否存在卷冲突。如果这个主机已经挂载了卷，其它同样使用这个卷的Pod不能调度到这个主机上，不同的存储后端具体规则不同
NoVolumeZoneConflict：检查给定的zone限制前提下，检查如果在此主机上部署Pod是否存在卷冲突
PodToleratesNodeTaints：确保pod定义的tolerates能接纳node定义的taints
CheckNodeMemoryPressure：检查pod是否可以调度到已经报告了主机内存压力过大的节点
CheckNodeDiskPressure：检查pod是否可以调度到已经报告了主机的存储压力过大的节点
MaxEBSVolumeCount：确保已挂载的EBS存储卷不超过设置的最大值，默认39
MaxGCEPDVolumeCount：确保已挂载的GCE存储卷不超过设置的最大值，默认16
MaxAzureDiskVolumeCount：确保已挂载的Azure存储卷不超过设置的最大值，默认16
GeneralPredicates：检查pod与主机上kubernetes相关组件是否匹配
NoVolumeNodeConflict：检查给定的Node限制前提下，检查如果在此主机上部署Pod是否存在卷冲突
复制代码

由于每个predicate都不复杂，就不分析了

2，priority

优选的算法也很多，这里列出几个

EqualPriority：所有节点同样优先级，无实际效果
ImageLocalityPriority：根据主机上是否已具备Pod运行的环境来打分，得分计算：不存在所需镜像，返回0分，存在镜像，镜像越大得分越高
LeastRequestedPriority：计算Pods需要的CPU和内存在当前节点可用资源的百分比，具有最小百分比的节点就是最优，得分计算公式：cpu((capacity – sum(requested)) * 10 / capacity) + memory((capacity – sum(requested)) * 10 / capacity) / 2
BalancedResourceAllocation：节点上各项资源（CPU、内存）使用率最均衡的为最优，得分计算公式：10 – abs(totalCpu/cpuNodeCapacity-totalMemory/memoryNodeCapacity)*10
SelectorSpreadPriority：按Service和Replicaset归属计算Node上分布最少的同类Pod数量，得分计算：数量越少得分越高
NodeAffinityPriority：节点亲和性选择策略，提供两种选择器支持：requiredDuringSchedulingIgnoredDuringExecution（保证所选的主机必须满足所有Pod对主机的规则要求）、preferresDuringSchedulingIgnoredDuringExecution（调度器会尽量但不保证满足NodeSelector的所有要求）
TaintTolerationPriority：类似于Predicates策略中的PodToleratesNodeTaints，优先调度到标记了Taint的节点
InterPodAffinityPriority：pod亲和性选择策略，类似NodeAffinityPriority，提供两种选择器支持：requiredDuringSchedulingIgnoredDuringExecution（保证所选的主机必须满足所有Pod对主机的规则要求）、preferresDuringSchedulingIgnoredDuringExecution（调度器会尽量但不保证满足NodeSelector的所有要求），两个子策略：podAffinity和podAntiAffinity，后边会专门详解该策略
MostRequestedPriority：动态伸缩集群环境比较适用，会优先调度pod到使用率最高的主机节点，这样在伸缩集群时，就会腾出空闲机器，从而进行停机处理。
复制代码

四，调度优先级队列

在1.11版本以前是alpha，在1.11版本开始为beta，并且默认开启。在1.9及以后的版本，优先级不仅影响调度的先后顺序，同时影响在node资源不足时候的驱逐顺序。

1，源码分析

看结构体定义即可，其他的代码都是很容易看懂

type PriorityQueue struct {
	// 有序堆，按照优先级存放等待调度的pod
	activeQ *Heap
	// 尝试调度并且调度失败的pod
	unschedulableQ *UnschedulablePodsMap
	// 存储高优先级pod(发生了抢占)期望调度的node信息，即有NominatedNodeName Annotation的pod
	nominatedPods map[string][]*v1.Pod
	receivedMoveRequest bool
}
复制代码

2，使用

如果在1.11版本以前，需要先开启该特性。

2.1 PriorityClasses

PriorityClasses在创建时候无需指定namespace，因为它是属于全局的。只允许全局存在一个globalDefault为true的PriorityClasses，来作为未指定priorityClassName的pod的优先级。对PriorityClasses的改动(例如改变globalDefault为true，删除PriorityClasses)不会影响已经创建的pod，pod的优先级只初始化一次。

创建如下：

apiVersion: scheduling.k8s.io/v1beta1
kind: PriorityClass
metadata:
  name: high-priority
value: 1000000
globalDefault: false
description: "This priority class should be used for XYZ service pods only."
复制代码

2.2 在pod中指定priorityClassName

例如指定上面的high-priority，未指定和没有PriorityClasses指定globalDefault为true的情况下，优先级为0。在1.9及以后的版本，高优先级的pod相比低优先级pod，处于调度队列的前头，但是如果高优先级队列无法被调度，也不会阻塞，调度器会调度低优先级的pod。

创建如下：

apiVersion: v1
kind: Pod
metadata:
  name: nginx
  labels:
    env: test
spec:
  containers:
  - name: nginx
    image: nginx
    imagePullPolicy: IfNotPresent
  priorityClassName: high-priority
复制代码

4，需要注意的地方

4.1，驱逐pod到调度pod存在时间差

由于在驱逐pod时候，优雅关闭需要等待一定的时间，那么导致pod真正被调度时候会存在一个时间差，我们可以优化低优先级的pod的优雅关闭时间或者调低优雅关闭时间

4.2，支持PDB，但是不能保证

调度器会尝试在不违反PDB情况下去驱逐pod，但是只是尝试，如果找不到或者还是不满足情况下，仍然为删除低优先级的pod

4.3，如果开始删除pod，那么说明该node一定能满足需求

4.4，低优先级pod有inter-pod affinity

如果在node上的pod存在inter-pod affinity，那么由于inter-pod affinity规则，pod P是无法调度到该pod的(如果需要驱逐这些inter-pod affinity 的pod)。所以如果我们有这块的需求，需要保证后调度的pod的优先级不高于前面的。

4.5，不支持跨node的驱逐

如果pod P要调度到N，pod Q此时已经在通过zone下的不同node运行，P和Q如果存在zone-wide的anti-affinity，那么P将无法调度到N上，因为无法跨node去驱逐Q。

4.6，需要防止用户设置大优先级的pod

五，调度器实战

1，自定义调度器

1.1 官方例子

通过 shell 脚步轮询获取指定调度器名称为my-scheduler的pod。

#!/bin/bash
SERVER='localhost:8001'
while true;
do
   for PODNAME in $(kubectl --server $SERVER get pods -o json | jq '.items[] | select(.spec.schedulerName == "my-scheduler") | select(.spec.nodeName == null) | .metadata.name' | tr -d '"')
;
   do
       NODES=($(kubectl --server $SERVER get nodes -o json | jq '.items[].metadata.name' | tr -d '"'))
       NUMNODES=${#NODES[@]}
       CHOSEN=${NODES[$[ $RANDOM % $NUMNODES ]]}
       curl --header "Content-Type:application/json" --request POST --data '{"apiVersion":"v1", "kind": "Binding", "metadata": {"name": "'$PODNAME'"}, "target": {"apiVersion": "v1", "kind"
: "Node", "name": "'$CHOSEN'"}}' http://$SERVER/api/v1/namespaces/default/pods/$PODNAME/binding/
       echo "Assigned $PODNAME to $CHOSEN"
   done
   sleep 1
done
复制代码

1.2 自定义扩展

这里完全复制第四个参考文献。 利用我们上面分析源码知道的，可以使用policy文件，自己组合需要的调度算法，然后可以指定扩展(可多个)。

{
  "kind" : "Policy",
  "apiVersion" : "v1",
  "predicates" : [
    {"name" : "PodFitsHostPorts"},
    {"name" : "PodFitsResources"},
    {"name" : "NoDiskConflict"},
    {"name" : "MatchNodeSelector"},
    {"name" : "HostName"}
    ],
  "priorities" : [
    {"name" : "LeastRequestedPriority", "weight" : 1},
    {"name" : "BalancedResourceAllocation", "weight" : 1},
    {"name" : "ServiceSpreadingPriority", "weight" : 1},
    {"name" : "EqualPriority", "weight" : 1}
    ],
  "extenders" : [
    {
          "urlPrefix": "http://localhost/scheduler",
          "apiVersion": "v1beta1",
          "filterVerb": "predicates/always_true",
          "bindVerb": "",
          "prioritizeVerb": "priorities/zero_score",
          "weight": 1,
          "enableHttps": false,
          "nodeCacheCapable": false
          "httpTimeout": 10000000
    }
      ],
  "hardPodAffinitySymmetricWeight" : 10
  }
复制代码

关于extender的配置的定义

type ExtenderConfig struct {
    // 访问该extender的url前缀
    URLPrefix string `json:"urlPrefix"`
    //过滤器调用的动词，如果不支持则为空。当向扩展程序发出过滤器调用时，此谓词将附加到URLPrefix
    FilterVerb string `json:"filterVerb,omitempty"`
    //prioritize调用的动词，如果不支持则为空。当向扩展程序发出优先级调用时，此谓词被附加到URLPrefix。
    PrioritizeVerb string `json:"prioritizeVerb,omitempty"`
    //优先级调用生成的节点分数的数字乘数，权重应该是一个正整数
    Weight int `json:"weight,omitempty"`
    //绑定调用的动词，如果不支持则为空。在向扩展器发出绑定调用时，此谓词会附加到URLPrefix。
    //如果此方法由扩展器实现，则将pod绑定动作将由扩展器返回给apiserver。只有一个扩展可以实现这个功能
    BindVerb string
    // EnableHTTPS指定是否应使用https与扩展器进行通信
    EnableHTTPS bool `json:"enableHttps,omitempty"`
    // TLSConfig指定传输层安全配置
    TLSConfig *restclient.TLSClientConfig `json:"tlsConfig,omitempty"`
    // HTTPTimeout指定对扩展器的调用的超时持续时间，过滤器超时无法调度pod。Prioritize超时被忽略
    //k8s或其他扩展器优先级被用来选择节点
    HTTPTimeout time.Duration `json:"httpTimeout,omitempty"`
    //NodeCacheCapable指定扩展器能够缓存节点信息
    //所以调度器应该只发送关于合格节点的最少信息
    //假定扩展器已经缓存了群集中所有节点的完整详细信息
    NodeCacheCapable bool `json:"nodeCacheCapable,omitempty"`
    // ManagedResources是由扩展器管理的扩展资源列表.
    // - 如果pod请求此列表中的至少一个扩展资源，则将在Filter，Prioritize和Bind（如果扩展程序是活页夹）
    //阶段将一个窗格发送到扩展程序。如果空或未指定，所有pod将被发送到这个扩展器。
    // 如果pod请求此列表中的至少一个扩展资源，则将在Filter，Prioritize和Bind（如果扩展程序是活页夹）阶段将一个pod发送到扩展程序。如果空或未指定，所有pod将被发送到这个扩展器。
    ManagedResources []ExtenderManagedResource `json:"managedResources,omitempty"`
}
复制代码

1.3 实现自己的调度算法

我们可以自定义自己的预选和优选算法，然后加载到算法工厂中，不过这样需要修改代码和重新编译调度器

1.4 做一个符合业务需求的调度器

如果有特殊的调度需求的，然后确实无法通过默认调度器解决的。可以自己实现一个scheduler controller，在自己的scheduler controller中，可以使用已经有的算法和自己的调度算法。这块等后面自己有做了相关事项再补充分享。