schedLoop | 南风未起｜云原生/Go技术分享博客

# 1.简介

# 1.1.scheduler

scheduler的相同调度器只有一个协程处理主循环调度scheduleOne，多个实例间仅选主的实例处理调度，这种设计出于预选和优选调度的数据一致性考量，确保调度过程中数据的并发安全。由于预选和优选都属于计算密集型任务，pod创建又非大并发操作，scheduler的单并发模型依然具备较高的性能。
注意

1.default-scheduler是默认调度器，pod创建可以基于 pod.Spec.schedulerName 指定其它调度器

2.kube-schedueler可以并行调度不同的调度器，各自按照调度算法执行，运行互不影响

# 1.2.scheduleOne

sched.Run()是调度主入口，会异步启动协程执行核心调度方法sched.scheduleOne()，用于获取可调度pod进行匹配节点筛选及绑定。

// Run begins watching and scheduling. It starts scheduling and blocked until the context is done.
func (sched *Scheduler) Run(ctx context.Context) {
    // 调度队列扫描
	sched.SchedulingQueue.Run()
    // 核心调度逻辑
	wait.UntilWithContext(ctx, sched.scheduleOne, 0)
	sched.SchedulingQueue.Close()
}

// scheduleOne does the entire scheduling workflow for a single pod.
func (sched *Scheduler) scheduleOne(ctx context.Context) {
    // 弹出activeQ堆顶pod
	podInfo := sched.NextPod()
	// pod could be nil when schedulerQueue is closed
	if podInfo == nil || podInfo.Pod == nil {
		return
	}
	pod := podInfo.Pod
    // 选择pod的调度框架
	fwk, err := sched.frameworkForPod(pod)
	...
    // 检查pod调度条件
	if sched.skipPodSchedule(fwk, pod) {
		return
	}
	...

    // 计算及选择pod合适运行的node
	scheduleResult, assumedPodInfo, status := sched.schedulingCycle(schedulingCycleCtx, state, fwk, podInfo, start, podsToActivate)
    // 失败处理
	if !status.IsSuccess() {
		sched.FailureHandler(schedulingCycleCtx, fwk, assumedPodInfo, status, scheduleResult.nominatingInfo, start)
		return
	}

	// 异步绑定pod和node关系
	go func() {
		...

        // 绑定pod和node
		status := sched.bindingCycle(bindingCycleCtx, state, fwk, scheduleResult, assumedPodInfo, start, podsToActivate)
        // 失败处理
		if !status.IsSuccess() {
			sched.handleBindingCycleError(bindingCycleCtx, state, fwk, assumedPodInfo, start, scheduleResult, status)
		}
	}()
}

// MakeNextPodFunc returns a function to retrieve the next pod from a given scheduling queue
func MakeNextPodFunc(queue SchedulingQueue) func() *framework.QueuedPodInfo {
	return func() *framework.QueuedPodInfo {
        // 弹出activeQ堆顶pod
		podInfo, err := queue.Pop()
		if err == nil {
			return podInfo
		}
		return nil
	}
}

// skipPodSchedule returns true if we could skip scheduling the pod for specified cases.
func (sched *Scheduler) skipPodSchedule(fwk framework.Framework, pod *v1.Pod) bool {
	// 正在删除的pod不进行调度
	if pod.DeletionTimestamp != nil {
		return true
	}

	// 检查pod已假设调度
	isAssumed, err := sched.Cache.IsAssumedPod(pod)
	...
    
	return isAssumed
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73

注意

scheduler基于单协程处理主调度循环scheduleOne()，调用schedulingCycle()为pod提名最优的node节点及执行绑定

# 2.调度流程

# 2.1.schedulingCycle

sched.schedulingCycle()为pod选出最优的node节点，先基于预选匹配满足pod要求的节点集合，再通过插件进行节点打分选出最优节点。

// schedulingCycle tries to schedule a single Pod.
func (sched *Scheduler) schedulingCycle(...) (ScheduleResult, *framework.QueuedPodInfo, *framework.Status) {
	pod := podInfo.Pod
    // 执行调度以选择节点
	scheduleResult, err := sched.SchedulePod(ctx, fwk, state, pod)
	if err != nil {
		...

        // 未支持抢占调度
		if !fwk.HasPostFilterPlugins() {
			return ScheduleResult{}, podInfo, framework.NewStatus(framework.Unschedulable).WithError(err)
		}

		// 执行抢占调度,用于pod于未来调度周期中可调度
		result, status := fwk.RunPostFilterPlugins(ctx, state, pod, fitError.Diagnosis.NodeToStatusMap)
		...

        // 记录调度的提名节点
		if result != nil {
			nominatingInfo = result.NominatingInfo
		}
		return ScheduleResult{nominatingInfo: nominatingInfo}, podInfo, framework.NewStatus(framework.Unschedulable).WithError(err)
	}
	...
    
    // node绑定存到cache
	sched.assume(assumedPod, scheduleResult.SuggestedHost)
	...

	// 临时锁定调度pod占用资源
	if sts := fwk.RunReservePluginsReserve(ctx, state, assumedPod, scheduleResult.SuggestedHost); !sts.IsSuccess() {
		// 撤销pod调度锁定资源
		fwk.RunReservePluginsUnreserve(ctx, state, assumedPod, scheduleResult.SuggestedHost)
        // 撤销调度缓存的pod
		sched.Cache.ForgetPod(assumedPod)
        ...

		return ScheduleResult{nominatingInfo: clearNominatedNode}, assumedPodInfo, sts
	}

    // 执行许可检查(空实现)
	runPermitStatus := fwk.RunPermitPlugins(ctx, state, assumedPod, scheduleResult.SuggestedHost)
	if !runPermitStatus.IsWait() && !runPermitStatus.IsSuccess() {
		// 撤销pod调度锁定资源
		fwk.RunReservePluginsUnreserve(ctx, state, assumedPod, scheduleResult.SuggestedHost)
        // 撤销调度缓存的pod
		sched.Cache.ForgetPod(assumedPod)
        ...

		return ScheduleResult{nominatingInfo: clearNominatedNode}, assumedPodInfo, runPermitStatus
	}

	// 调度周期完成,激活有必要重调度的pod
	if len(podsToActivate.Map) != 0 {
        // 这些pod转入backoffQ或activeQ
		sched.SchedulingQueue.Activate(podsToActivate.Map)
		// 重置
		podsToActivate.Map = make(map[string]*v1.Pod)
	}

	return scheduleResult, assumedPodInfo, nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62

注意

schedulingCycle()是选举节点的调度入口，用于节点选择、缓存及许可检查，以及将其它可调度节点重新加入activeQ和backoffQ

# 2.2.assume

sched.assume()是调度乐观实现，会将筛选的node与pod提前缓存绑定，基于内存维护绑定状态以提高吞吐及减少调度交互apiserver延迟。

// assume signals to the cache that a pod is already in the cache, so that binding can be asynchronous.
func (sched *Scheduler) assume(assumed *v1.Pod, host string) error {
	// 记录pod调度到的node
	assumed.Spec.NodeName = host

    // 注册到cache缓存
	sched.Cache.AssumePod(assumed)
    ...
    
	// 清理维护的提名pod
	if sched.SchedulingQueue != nil {
		sched.SchedulingQueue.DeleteNominatedPodIfExists(assumed)
	}

	return nil
}

func (cache *cacheImpl) AssumePod(pod *v1.Pod) error {
    // pod唯一标识
	key, err := framework.GetPodKey(pod)
	...

	cache.mu.Lock()
	defer cache.mu.Unlock()
    
    // pod已经调度
	if _, ok := cache.podStates[key]; ok {
		return fmt.Errorf("pod %v(%v) is in the cache, so can't be assumed", key, klog.KObj(pod))
	}

    // 提前内存绑定
	return cache.addPod(pod, true)
}

// Assumes that lock is already acquired.
func (cache *cacheImpl) addPod(pod *v1.Pod, assumePod bool) error {
    // pod唯一标识
	key, err := framework.GetPodKey(pod)
	...
    
	n, ok := cache.nodes[pod.Spec.NodeName]
    
    // 记录未注册的节点
	if !ok {
		n = newNodeInfoListItem(framework.NewNodeInfo())
		cache.nodes[pod.Spec.NodeName] = n
	}
    
    // 节点调度的pod
	n.info.AddPod(pod)
    
    // 移动到头部
	cache.moveNodeInfoToHead(pod.Spec.NodeName)
	ps := &podState{
		pod: pod,
	}
    // 记录pod状态(此时未绑定)
	cache.podStates[key] = ps
    // 记录假设的pod
	if assumePod {
		cache.assumedPods.Insert(key)
	}
	return nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

注意

assume()会提前基于内存维护pod与node绑定关系，以完成后续真正的节点绑定

# 2.3.reverseAndPermit

reserve plugin用于锁定资源，permit plugin用于许可检查(pod组调度)，检查通过的pod才会进一步执行节点绑定。

// RunReservePluginsReserve runs the Reserve method in the set of configured reserve plugins.
func (f *frameworkImpl) RunReservePluginsReserve(...) (status *framework.Status) {
	...
    
    // 遍历reserve插件
	for _, pl := range f.reservePlugins {
        // 执行以临时锁定资源
		status = f.runReservePluginReserve(ctx, pl, state, pod, nodeName)
		...
	}
	return nil
}

// RunPermitPlugins runs the set of configured permit plugins.
func (f *frameworkImpl) RunPermitPlugins(...) (status *framework.Status) {
	...
    
    // 遍历permit插件
	for _, pl := range f.permitPlugins {
        // 执行许可检查
		status, timeout := f.runPermitPlugin(ctx, pl, state, pod, nodeName)
		...
	}
    
    // 延迟调度
	if statusCode == framework.Wait {
        // 拷贝为waitingPod
		waitingPod := newWaitingPod(pod, pluginsWaitTime)
        / 放入全局等待列表,直到超时或收到allow通知
		f.waitingPods.add(waitingPod)
		...
		return framework.NewStatus(framework.Wait, msg)
	}
	return nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

注意

调度周期其实都是执行plugin/extender检查及筛选，直到根据节点打分获取最优节点，进行reserver资源锁定及permit组调度检查

# 3.绑定流程

# 3.1.bindingCycle

bindingCycle()用于执行pod和node绑定，以驱动kubelet介入pod生命周期管理，内部主要经历prebind-->bind-->postbind阶段。

// bindingCycle tries to bind an assumed Pod.
func (sched *Scheduler) bindingCycle(...) *framework.Status {
	// 待绑定pod
	assumedPod := assumedPodInfo.Pod

	// 执行permit检查前置条件
	if status := fwk.WaitOnPermit(ctx, assumedPod); !status.IsSuccess() {
		return status
	}

	// 执行preBind检查绑定条件
	if status := fwk.RunPreBindPlugins(ctx, state, assumedPod, scheduleResult.SuggestedHost); !status.IsSuccess() {
		return status
	}

	// 执行bind进行pod与node绑定
	if status := sched.bind(ctx, fwk, assumedPod, scheduleResult.SuggestedHost, state); !status.IsSuccess() {
		return status
	}
	...
    
	// 执行postBind进行绑定后检查
	fwk.RunPostBindPlugins(ctx, state, assumedPod, scheduleResult.SuggestedHost)

	// 完成绑定后,将一些可调度pod转入activeQ队列
	if len(podsToActivate.Map) != 0 {
		sched.SchedulingQueue.Activate(podsToActivate.Map)
	}

	return nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

注意

preBind插件用于资源分配及检查，bind(defaultBinder)插件用于将绑定节点的pod更新到etcd，postBind插件未实现

# 3.2.waitOnPermit

waitOnPermit()用于检测对应的waitingPod是否满足调度条件，这类pod一般等待volumn就绪或依赖pod就绪，检测通过才能进一步绑定。

// WaitOnPermit will block, if the pod is a waiting pod, until the waiting pod is rejected or allowed.
func (f *frameworkImpl) WaitOnPermit(ctx context.Context, pod *v1.Pod) *framework.Status {
    // 获取waitingPod
	waitingPod := f.waitingPods.get(pod.UID)
	// 未关联waitingPod结束检查
    if waitingPod == nil {
		return nil
	}
    // 内存回收
	defer f.waitingPods.remove(pod.UID)
	...
    
    // 由channel发送插件结果
	s := <-waitingPod.s
	...
    
    // 未成功
	if !s.IsSuccess() {
        ...
        return s
	}
	return nil
}

// Allow declares the waiting pod is allowed to be scheduled by plugin pluginName.
func (w *waitingPod) Allow(pluginName string) {
	w.mu.Lock()
	defer w.mu.Unlock()
    // 终止对应插件定时器及清理相关任务
	if timer, exist := w.pendingPlugins[pluginName]; exist {
		timer.Stop()
		delete(w.pendingPlugins, pluginName)
	}

	// 检查是否还有未通过的插件
	if len(w.pendingPlugins) != 0 {
		return
	}

	// 非阻塞发送成功状态
	select {
	case w.s <- framework.NewStatus(framework.Success, ""):
	default:
	}
}

// Reject declares the waiting pod unschedulable.
func (w *waitingPod) Reject(pluginName, msg string) {
	w.mu.RLock()
	defer w.mu.RUnlock()
    // 终止定影插件定时器
	for _, timer := range w.pendingPlugins {
		timer.Stop()
	}

	// 非阻塞发送不可调度状态
	select {
	case w.s <- framework.NewStatus(framework.Unschedulable, msg).WithFailedPlugin(pluginName):
	default:
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

注意

目前permit插件官方未实现，一般原则是基于pod组批量唤醒waitingPod，基于Allow或Reject方法

# 4.失败处理

# 4.1.failureHandler

pod未找到合适node会调用FailureHandler，将失败的pod放入backoffQ或unscheduleQ队列，由后台协程尝试再次放入activeQ。

// handleSchedulingFailure records an event for the pod that indicates the
// pod has failed to schedule. Also, update the pod condition and nominated node name if set.
func (sched *Scheduler) handleSchedulingFailure(...) {
	...

    // 失败pod
	pod := podInfo.Pod
	...

	// 检查pod存在
	podLister := fwk.SharedInformerFactory().Core().V1().Pods().Lister()
	cachedPod, e := podLister.Pods(pod.Namespace).Get(pod.Name)
	...
    
    // pod未调度过
    if len(cachedPod.Spec.NodeName) == 0 {
        // 构造podInfo
        podInfo.PodInfo, _ = framework.NewPodInfo(cachedPod.DeepCopy())
        // 加入unscheduleQ(集群事件变更触发转移)
        sched.SchedulingQueue.AddUnschedulableIfNotPresent(podInfo, sched.SchedulingQueue.SchedulingCycle())
    }

    // 尝试记录将调度的节点(抢占调度)
	if sched.SchedulingQueue != nil {
        // pod已调度则不再注册提名
		sched.SchedulingQueue.AddNominatedPod(podInfo.PodInfo, nominatingInfo)
	}
	...

	...
    // 发布事件
	fwk.EventRecorder().Eventf(pod, nil, v1.EventTypeWarning, "FailedScheduling", "Scheduling", msg)
    // 更新pod状态(抢占式pod会更新status.NominatedNodeName——倾向的提名节点)
	updatePod(ctx, sched.client, pod, &v1.PodCondition{...}, nominatingInfo)
    ...
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

注意

未调度成功的pod会放入unscheduleQ直到集群事件变更被唤醒，调度失败的抢占pod会在这里提前绑定提名node供下次优先调度

# 4.2.handleBindingCycleError

sched.handleBindingCycleError()用于bind阶段出错时，执行锁定资源及缓存的清理与回退及pod再入队，确保调度数据的状态一致性。

func (sched *Scheduler) handleBindingCycleError(...) {

    // 提名pod
	assumedPod := podInfo.Pod
	
    // 执行reserve插件的锁定资源回退
	fwk.RunReservePluginsUnreserve(ctx, state, assumedPod, scheduleResult.SuggestedHost)
    
    // 清理缓存提名pod
	sched.Cache.ForgetPod(assumedPod)
    ...
    // 当前pod调度失败且即将释放锁定资源,因此通知其它不可调度pod再次尝试调度
    
    // 不可调度
    if status.IsUnschedulable() {
        // 延迟激活其它不可调度pod,避免触发move后moveRequestCycle变化,导致不可调度pod进入backoffQ
        defer sched.SchedulingQueue.MoveAllToActiveOrBackoffQueue(internalqueue.AssignedPodDelete, func(pod *v1.Pod) bool {	return assumedPod.UID != pod.UID  })
    // 调度错误
    } else {
        // 立即触发一次激活
        sched.SchedulingQueue.MoveAllToActiveOrBackoffQueue(internalqueue.AssignedPodDelete, nil)
    }

    // 调度失败处理
	sched.FailureHandler(ctx, fwk, podInfo, status, clearNominatedNode, start)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

注意

bind失败会回退锁定的状态，清理cache假设提名的pod，还会触发不可调度pod唤醒及失败pod再入队