kubescheduler源码解析

docker run -v /work/src/k8s.io/kubernetes:/go/src/k8s.io/kubernetes golang:1.11.2 bash

cd cmd/kube-scheduler && go build

$ cat Dockerfile-build1.12.2
FROM golang:1.11.2
COPY vendor/ /vendor

workspace:
  base: /go/src/k8s.io
  path: kubernetes
pipeline:
    build:
        image: fanux/kubernetes-build:1.12.2-beta.3
        commands:
           - make all WHAT=cmd/kube-kubescheduler GOFLAGS=-v
    publish:
        image: plugins/docker
        registry: xxx
        username: xxx
        password: xxx
        email: xxx
        repo: xxx/container/kube-scheduler
        tags: ${DRONE_TAG=latest}
        dockerfile: dockerfile/Dockerfile-kube-scheduler
        insecure: true
        when:
            event: [push, tag]

$ cat dockerfile/Dockerfile-kube-scheduler
FROM scratch
COPY  _output/local/bin/linux/amd64/kube-scheduler /
CMD ["/kube-scheduler"]

build_kubeadm:
        image: fanux/kubernetes-build:1.12.2-beta.3
        commands:
           - make all WHAT=cmd/kube-kubeadm GOFLAGS=-v
           - curl -v -u container:container --upload-file kubeadm http://172.16.59.153:8081/repository/kubernetes/kubeadm/
        when:
            event: deployment
            enviroment: kubeadm

[signal SIGSEGV: segmentation violation code=0x1 addr=0x63 pc=0x7f2b7f5f057c]
runtime stack:
runtime.throw(0x17c74a8, 0x2a)
	/usr/local/go/src/runtime/panic.go:608 +0x72
runtime.sigpanic()
	/usr/local/go/src/runtime/signal_unix.go:374 +0x2f2

+-------------------------------------------+     +----+
|                            Add            |     |    |
|                                           |     |    | Update
+      Assume                Add            v     v    |
Initial +--------> Assumed +------------+---> Added <--+
^                +   +                  |       +
|                |   |                  |       |
|                |   |              Add |       | Remove
|                |   |                  |       |
|                |   |                  +       |
+----------------+   +-----------> Expired+----> Deleted

type schedulerCache struct {
	stop   <-chan struct{}
	ttl    time.Duration
	period time.Duration
	// This mutex guards all fields within this cache struct.
	mu sync.RWMutex
	// a set of assumed pod keys.
	// The key could further be used to get an entry in podStates.
	assumedPods map[string]bool
	// a map from pod key to podState.
	podStates map[string]*podState
	nodes     map[string]*NodeInfo
	nodeTree  *NodeTree
	pdbs      map[string]*policy.PodDisruptionBudget
	// A map from image name to its imageState.
	imageStates map[string]*imageState
}

cache.addPod(pod)
ps := &podState{
	pod: pod,
}
cache.podStates[key] = ps

type NodeTree struct {
	tree      map[string]*nodeArray // a map from zone (region-zone) to an array of nodes in the zone.
	zones     []string              // a list of all the zones in the tree (keys)
	zoneIndex int
	NumNodes  int
	mu        sync.RWMutex
}

func (cache *schedulerCache) run() {
	go wait.Until(cache.cleanupExpiredAssumedPods, cache.period, cache.stop)
}

type Scheduler struct {
	config *Config  -------> SchedulerCache
|
+---> Algorithm
}

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

+-------------+
| 获取一个pod |
+-------------+
|
+-----------------------------------------------------------------------------------+
| 如果pod的DeletionTimestamp 存在就不用进行调度, kubelet发现这个字段会直接去删除pod |
+-----------------------------------------------------------------------------------+
|
+-----------------------------------------+
| 选一个suggestedHost，可理解为合适的节点 |
+-----------------------------------------+
|_____________选不到就进入强占的逻辑，与我当初写swarm调度器逻辑类似
|
+--------------------------------------------------------------------------------+
| 虽然还没真调度到node上，但是告诉cache pod已经被调度到node上了，变成assume pod  |
| 这里面会先检查volumes                                                          |
| 然后：err = sched.assume(assumedPod, suggestedHost) 假设pod被调度到node上了    |
+--------------------------------------------------------------------------------+
|
+---------------------------+
| 异步的bind这个pod到node上 |
| 先bind volume             |
| bind pod                  |
+---------------------------+
|
+----------------+
| 暴露一些metric |
+----------------+

err := sched.bind(assumedPod, &v1.Binding{
	ObjectMeta: metav1.ObjectMeta{Namespace: assumedPod.Namespace, Name: assumedPod.Name, UID: assumedPod.UID},
	Target: v1.ObjectReference{
		Kind: "Node",
		Name: suggestedHost,
	},
})

err := sched.config.GetBinder(assumed).Bind(b)
if err := sched.config.SchedulerCache.FinishBinding(assumed);

▾ algorithm/
▸ predicates/  预选
▸ priorities/  优选

suggestedHost, err := sched.schedule(pod)

type ScheduleAlgorithm interface {
// 传入pod 节点列表，返回一下合适的节点
	Schedule(*v1.Pod, NodeLister) (selectedMachine string, err error)
// 资源抢占用的
	Preempt(*v1.Pod, NodeLister, error) (selectedNode *v1.Node, preemptedPods []*v1.Pod, cleanupNominatedPods []*v1.Pod, err error)
// 预选函数集，
	Predicates() map[string]FitPredicate
|                              这一个节点适合不适合调度这个pod，不适合的话返回原因
+-------type FitPredicate func(pod *v1.Pod, meta PredicateMetadata, nodeInfo *schedulercache.NodeInfo) (bool, []PredicateFailureReason, error)
// 返回优选配置,最重要两个函数 map 和 reduce
	Prioritizers() []PriorityConfig
|____________PriorityMapFunction 计算 节点的优先级
|____________PriorityReduceFunction 根据map的结果计算所有node的最终得分
|____________PriorityFunction 废弃
}

priorityConfigs, err := c.GetPriorityFunctionConfigs(priorityKeys)
priorityMetaProducer, err := c.GetPriorityMetadataProducer()
predicateMetaProducer, err := c.GetPredicateMetadataProducer()
|
algo := core.NewGenericScheduler(                 |
	c.schedulerCache,                         |
	c.equivalencePodCache,                    V
	c.podQueue,
	predicateFuncs,   ============> 这里面把预选优选函数都注入进来了
	predicateMetaProducer,
	priorityConfigs,
	priorityMetaProducer,
	extenders,
	c.volumeBinder,
	c.pVCLister,
	c.alwaysCheckAllPredicates,
	c.disablePreemption,
	c.percentageOfNodesToScore,
)
type genericScheduler struct {
	cache                    schedulercache.Cache
	equivalenceCache         *equivalence.Cache
	schedulingQueue          SchedulingQueue
	predicates               map[string]algorithm.FitPredicate
	priorityMetaProducer     algorithm.PriorityMetadataProducer
	predicateMetaProducer    algorithm.PredicateMetadataProducer
	prioritizers             []algorithm.PriorityConfig
	extenders                []algorithm.SchedulerExtender
	lastNodeIndex            uint64
	alwaysCheckAllPredicates bool
	cachedNodeInfoMap        map[string]*schedulercache.NodeInfo
	volumeBinder             *volumebinder.VolumeBinder
	pvcLister                corelisters.PersistentVolumeClaimLister
	disablePreemption        bool
	percentageOfNodesToScore int32
}

+------------------------------------+
| trace记录一下，要开始调度哪个pod了 |
+------------------------------------+
|
+------------------------------------------------+
| pod基本检查，这里主要检查卷和delete timestamp |
+-----------------------------------------------+
|
+----------------------------------------+
| 获取node列表, 更新cache的node info map |
+----------------------------------------+
|
+----------------------------------------------+
| 预选，返回合适的节点列表和预选失败节点的原因 |
+----------------------------------------------+
|
+----------------------------------------------------------+
| 优选，                                                   |
| 如果预选结果只有一个节点，那么直接使用之，不需要进行优选 |
| 否则进行优选过程                                         |
+----------------------------------------------------------+
|
+------------------------------------+
| 在优选结果列表中选择得分最高的节点 |
+------------------------------------+

先排序 predicates.Ordering()
if predicate, exist := predicateFuncs[predicateKey]; exist {
		fit, reasons, err = predicate(pod, metaToUse, nodeInfoToUse)

predicatesOrdering = []string{CheckNodeConditionPred, CheckNodeUnschedulablePred,
		GeneralPred, HostNamePred, PodFitsHostPortsPred,
		MatchNodeSelectorPred, PodFitsResourcesPred, NoDiskConflictPred,
		PodToleratesNodeTaintsPred, PodToleratesNodeNoExecuteTaintsPred, CheckNodeLabelPresencePred,
		CheckServiceAffinityPred, MaxEBSVolumeCountPred, MaxGCEPDVolumeCountPred, MaxCSIVolumeCountPred,
		MaxAzureDiskVolumeCountPred, CheckVolumeBindingPred, NoVolumeZoneConflictPred,
		CheckNodeMemoryPressurePred, CheckNodePIDPressurePred, CheckNodeDiskPressurePred, MatchInterPodAffinityPred}

predicateFuncs, err := c.GetPredicates(predicateKeys)

factory.RegisterFitPredicate(predicates.NoDiskConflictPred, predicates.NoDiskConflict),
factory.RegisterFitPredicate(predicates.GeneralPred, predicates.GeneralPredicates),
factory.RegisterFitPredicate(predicates.CheckNodeMemoryPressurePred, predicates.CheckNodeMemoryPressurePredicate),
factory.RegisterFitPredicate(predicates.CheckNodeDiskPressurePred, predicates.CheckNodeDiskPressurePredicate),
factory.RegisterFitPredicate(predicates.CheckNodePIDPressurePred, predicates.CheckNodePIDPressurePredicate),

factory.RegisterPriorityFunction2("LeastRequestedPriority", priorities.LeastRequestedPriorityMap, nil, 1),
// Prioritizes nodes to help achieve balanced resource usage
factory.RegisterPriorityFunction2("BalancedResourceAllocation", priorities.BalancedResourceAllocationMap, nil, 1),

factory.RegisterPriorityFunction2("NodeAffinityPriority", priorities.CalculateNodeAffinityPriorityMap, priorities.CalculateNodeAffinityPriorityReduce, 1)

如果满足节点亲和，积分加权重
count += preferredSchedulingTerm.Weight
return schedulerapi.HostPriority{
	Host:  node.Name,
	Score: int(count),  # 算出积分
}, nil

var maxCount int
for i := range result {
	if result[i].Score > maxCount {
		maxCount = result[i].Score  # 所有分值里的最大值
	}
}
for i := range result {
	score := result[i].Score
	score = maxPriority * score / maxCount  # 分值乘以最大优先级是maxPriority = 10，除以最大值赋值给分值 我没明白
	result[i].Score = score
}

for i := range priorityConfigs {
	if priorityConfigs[i].Function != nil {
		continue
	}
	results[i][index], err = priorityConfigs[i].Map(pod, meta, nodeInfo)
	if err != nil {
		appendError(err)
		results[i][index].Host = nodes[index].Name
	}
}
err := config.Reduce(pod, meta, nodeNameToInfo, results[index]);

result[i].Score += results[j][i].Score * priorityConfigs[j].Weight  (二维变一维)

pod := sched.config.NextPod()

pkg/scheduler/factory/factory.go
NextPod: func() *v1.Pod {
	return c.getNextPod()
},
func (c *configFactory) getNextPod() *v1.Pod {
	pod, err := c.podQueue.Pop()
	if err == nil {
		return pod
	}
...
}

type SchedulingQueue interface {
	Add(pod *v1.Pod) error
	AddIfNotPresent(pod *v1.Pod) error
	AddUnschedulableIfNotPresent(pod *v1.Pod) error
	Pop() (*v1.Pod, error)
	Update(oldPod, newPod *v1.Pod) error
	Delete(pod *v1.Pod) error
	MoveAllToActiveQueue()
	AssignedPodAdded(pod *v1.Pod)
	AssignedPodUpdated(pod *v1.Pod)
	WaitingPodsForNode(nodeName string) []*v1.Pod
	WaitingPods() []*v1.Pod
}

func NewSchedulingQueue() SchedulingQueue {
	if util.PodPriorityEnabled() {
		return NewPriorityQueue()  # 基于堆排序实现，根据优先级排序
	}
	return NewFIFO() # 简单的先进先出
}

AddFunc:    c.addPodToCache,
UpdateFunc: c.updatePodInCache,
DeleteFunc: c.deletePodFromCache,
| informer监听,了pod创建事件之后往cache和队列里都更新了
V 
if err := c.schedulerCache.AddPod(pod); err != nil {
	glog.Errorf("scheduler cache AddPod failed: %v", err)
}
c.podQueue.AssignedPodAdded(pod)

+------------+ ADD   +-------------+   POP  +-----------+
| informer   |------>|  sche Queue |------->| scheduler |
+------------+   |   +-------------+        +----^------+
+-->+-------------+             |
| sche cache  |<------------+
+-------------+

+----------------------------------+       +----------+
| kube-scheduler -> extender client|------>| extender | (你需要开发的扩展，单独的进程)
+----------------------------------+       +----------+

type SchedulerExtender interface {
// 最重要的一个接口，输入pod和节点列表，输出是符合调度的节点的列表
	Filter(pod *v1.Pod,
		nodes []*v1.Node, nodeNameToInfo map[string]*schedulercache.NodeInfo,
	) (filteredNodes []*v1.Node, failedNodesMap schedulerapi.FailedNodesMap, err error)
// 这个给节点打分的，优选时需要用的
	Prioritize(pod *v1.Pod, nodes []*v1.Node) (hostPriorities *schedulerapi.HostPriorityList, weight int, err error)
// Bind接口主要是最终调度器选中节点哪个节点时通知extender
	Bind(binding *v1.Binding) error
	// IsBinder returns whether this extender is configured for the Bind method.
	IsBinder() bool
// 可以过滤你感兴趣的pod 比如按照标签
	IsInterested(pod *v1.Pod) bool
	// ProcessPreemption returns nodes with their victim pods processed by extender based on
	// given:
	//   1. Pod to schedule
	//   2. Candidate nodes and victim pods (nodeToVictims) generated by previous scheduling process.
	//   3. nodeNameToInfo to restore v1.Node from node name if extender cache is enabled.
	// The possible changes made by extender may include:
	//   1. Subset of given candidate nodes after preemption phase of extender.
	//   2. A different set of victim pod for every given candidate node after preemption phase of extender.
// 我猜是与亲和性相关的功能，不太清楚 TODO
	ProcessPreemption(
		pod *v1.Pod,
		nodeToVictims map[*v1.Node]*schedulerapi.Victims,
		nodeNameToInfo map[string]*schedulercache.NodeInfo,
	) (map[*v1.Node]*schedulerapi.Victims, error)
// 优先级抢占特性，可不实现
	SupportsPreemption() bool
// 当访问不到extender时怎么处理，返回真时extender获取不到时调度不能失败
	IsIgnorable() bool
}

type HTTPExtender struct {
	extenderURL      string
	preemptVerb      string
	filterVerb       string  # 预选RUL
	prioritizeVerb   string  # 优选RUL
	bindVerb         string
	weight           int
	client           *http.Client
	nodeCacheCapable bool
	managedResources sets.String
	ignorable        bool
}

args = &schedulerapi.ExtenderArgs{  # 调度的是哪个pod，哪些节点符合调度条件, 返回的也是这个结构体
	Pod:       pod,
	Nodes:     nodeList,
	NodeNames: nodeNames,
}
if err := h.send(h.filterVerb, args, &result); err != nil { # 发了个http请求给extender(你要去实现的httpserver), 返回过滤后的结构
	return nil, nil, err
}

NamespaceSystem string = "kube-system"
SchedulerDefaultLockObjectNamespace string = metav1.NamespaceSystem
// SchedulerPolicyConfigMapKey defines the key of the element in the
// scheduler's policy ConfigMap that contains scheduler's policy config.
SchedulerPolicyConfigMapKey = "policy.cfg"