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

# 1.cache

# 1.1.start

sc.Run()会启动cacheInformer，informer完成缓存基于监听资源执行node/errTask同步、terminated job清理及bind task协调。

// Run  starts the schedulerCache
func (sc *SchedulerCache) Run(stopCh <-chan struct{}) {
	sc.informerFactory.Start(stopCh)
	sc.vcInformerFactory.Start(stopCh)
	sc.WaitForCacheSync(stopCh)
    
    // default 20 worker
	for i := 0; i < int(sc.nodeWorkers); i++ {
        // node同步
		go wait.Until(sc.runNodeWorker, 0, stopCh)
	}

	// errTask同步
	go wait.Until(sc.processResyncTask, 0, stopCh)

	// job清理
	go wait.Until(sc.processCleanupJob, 0, stopCh)

    // bindTask协调
	go wait.Until(sc.processBindTask, time.Millisecond*20, stopCh)
	...
}

func (sc *SchedulerCache) processCleanupJob() {
    // pod/podgroup监听
	obj, shutdown := sc.DeletedJobs.Get()
	...

	defer sc.DeletedJobs.Done(obj)
	...

    // job未关联podgroup和task
	if schedulingapi.JobTerminated(job) {
		oldJob, found := sc.Jobs[job.UID]
		if !found {
			sc.DeletedJobs.Forget(obj)
			return
		}
		newPgVersion := oldJob.PgUID
		oldPgVersion := job.PgUID
		if oldPgVersion == newPgVersion {
			delete(sc.Jobs, job.UID)
		}
		sc.DeletedJobs.Forget(obj)
	} else {
		// deletedJobs.AddLimited
		sc.retryDeleteJob(job)
	}
}

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

注意

cache主要围绕runNodeWorker、processResyncTask、processCleanupJob和processBindTask展开，下面会介绍

# 1.2.resync

sc.processResyncTask()用于处理errTask队列任务，Pod未找到会清理cache node/job注册资源信息，否则会更新node/job资源占用。

func (sc *SchedulerCache) processResyncTask() {
    // 1.pod处理失败  2.pg清理&job终止
	obj, shutdown := sc.errTasks.Get()
	...

	defer sc.errTasks.Done(obj)

    // 格式不对
	taskKey, ok := obj.(string)
	if !ok {
		sc.errTasks.Forget(obj)
		return
	}

    // 获取cache task
	task, err := sc.parseErrTaskKey(taskKey)
	if err != nil {
		sc.errTasks.Forget(obj)
		return
	}
	...

	if err := sc.syncTask(task); err != nil {
        // errTasks.AddLimited
		sc.resyncTask(task)
		reSynced = true
	} else {
		sc.errTasks.Forget(obj)
	}

	// execute custom bind err handler call back func if exists.
	if task.CustomBindErrHandler != nil && !task.CustomBindErrHandlerSucceeded {
		// 执行task custom回调
        err := task.CustomBindErrHandler()
		if err == nil {
			task.CustomBindErrHandlerSucceeded = true
		}
        
        // 回调失败&未重新入队
		if !task.CustomBindErrHandlerSucceeded && !reSynced {
			sc.resyncTask(task)
		}
	}
}

func (sc *SchedulerCache) syncTask(oldTask *schedulingapi.TaskInfo) error {
	newPod, err := sc.kubeClient.CoreV1().Pods(oldTask.Namespace).Get(context.TODO(), oldTask.Name, ...)
	if err != nil {
        // pod未找到
		if errors.IsNotFound(err) {
			...
            
            // 清理node/job/numa task相关资源占用
			sc.deleteTask(oldTask)
            ...

			return nil
		}
		return fmt.Errorf("failed to get Pod <%v/%v>: err %v", oldTask.Namespace, oldTask.Name, err)
	}

    // 更新task对象
	newTask, err := sc.NewTaskInfo(newPod)
	...

    // 重新注册node/job/numa task相关资源占用
	return sc.updateTask(oldTask, newTask)
}

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

注意

errTask核心处理集中于sc.syncTask，内部执行的deleteTask和updateTask已经在informer篇介绍过，这里不再赘述

# 1.3.worker

sc.runNodeWorker()同步nodeQueue对象，更新nodeInfo/nodeImageState资源占用情况，维护nodeList缓存和Numatopology对象。

func (sc *SchedulerCache) runNodeWorker() {
	for sc.processSyncNode() {
	}
}

func (sc *SchedulerCache) processSyncNode() bool {
	obj, shutdown := sc.nodeQueue.Get()
	...
    
	defer sc.nodeQueue.Done(obj)
	...

	// node同步
	err := sc.SyncNode(nodeName)
	if err == nil {
		sc.nodeQueue.Forget(nodeName)
		return true
	}

	sc.nodeQueue.AddRateLimited(nodeName)
	return true
}

func (sc *SchedulerCache) SyncNode(nodeName string) error {
	node, err := sc.nodeInformer.Lister().Get(nodeName)
	if err != nil {
        // 未找到
		if errors.IsNotFound(err) {
            // 清理node缓存
			sc.RemoveNode(nodeName)
			...

			return nil
		}
        
		return err
	}

    // 检查加入cache条件
	if !sc.nodeCanAddCache(node) {
		return nil
	}
    
	nodeCopy := node.DeepCopy()
	csiNode := sc.csiNodeInformer.Lister().Get(nodeName)
	...
    // node设置csi资源信息
    sc.setCSIResourceOnNode(csiNode, nodeCopy)
    ...
    
	return sc.AddOrUpdateNode(nodeCopy)
}

// AddOrUpdateNode adds or updates node info in cache.
func (sc *SchedulerCache) AddOrUpdateNode(node *v1.Node) error {
	...

	if sc.Nodes[node.Name] != nil {
        // 更新nodeInfo
		sc.Nodes[node.Name].SetNode(node)
        // 清理旧的nodeImageState缓存
		sc.removeNodeImageStates(node.Name)
	} else {
        // 初始化nodeInfo
		sc.Nodes[node.Name] = schedulingapi.NewNodeInfo(node)
	}
    
    // 重新设置nodeImageState
	sc.addNodeImageStates(node, sc.Nodes[node.Name])
	...
	
	for _, name := range sc.NodeList {
		if name == node.Name {
			nodeExisted = true
			break
		}
	}
    
	if !nodeExisted {
		sc.NodeList = append(sc.NodeList, node.Name)
	}
    
	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
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84

注意

nodeWorker负责更新node相关资源占用信息，涉及可用资源及镜像资源，供后续调度作为资源准入条件

# 2.bind

# 2.1.process

sc.processBindTask()会间隔20ms获取session处理的task，执行bindvolumes和bindnode，绑定PV/PVC及Pod/Node。

// execute bind task in 20ms intervals.
func (sc *SchedulerCache) processBindTask() {
	for {
		select {
        // 消费bind task(来自session)
		case taskInfo, ok := <-sc.BindFlowChannel:
			...

			sc.bindCache = append(sc.bindCache, taskInfo)
            
            // 达到批规模执行绑定
			if len(sc.bindCache) == sc.batchNum {
				sc.BindTask()
			}
		default:
		}

        // 全部消费完成
		if len(sc.BindFlowChannel) == 0 {
			break
		}
	}

	if len(sc.bindCache) == 0 {
		return
	}
    
    // 执行不够批规模的绑定
	sc.BindTask()
}

// BindTask do k8s binding with a goroutine
func (sc *SchedulerCache) BindTask() {
	...
    
	copy(tmpBindCache, sc.bindCache)
	go func(tasks []*schedulingapi.TaskInfo) {
		...
        
        // 过滤volume准备好的task
		for _, task := range tasks {
			if err := sc.VolumeBinder.BindVolumes(task, task.PodVolumes); err != nil {
                // 假设绑定的pv及pvc均恢复,标记task.VolumeReady为false
				sc.VolumeBinder.RevertVolumes(task, task.PodVolumes)
                // errTasks.AddLimited
				sc.resyncTask(task)
			} else {
				successfulTasks = append(successfulTasks, task)
			}
		}

		bindTasks := make([]*schedulingapi.TaskInfo, len(successfulTasks))
		copy(bindTasks, successfulTasks)
        // 绑定到节点
		sc.Bind(bindTasks)
	}(tmpBindCache)
	sc.bindCache = sc.bindCache[0:0]
}

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

注意

BindVolumes检查task相关的PV/PVC就绪条件，volume准备好的task会执行bind绑定到node

# 2.2.bindvol

dvb.BindVolumes()基于assume volume更新绑定PV/PVC，检查PV/PVC绑定条件，失败的会由assume清理，对应的task留到下次协调。

// BindVolumes binds volumes to the task
func (dvb *defaultVolumeBinder) BindVolumes(...) error {
	// If task's volumes are ready, did not bind them again.
	if task.VolumeReady {
		return nil
	}

	return dvb.volumeBinder.BindPodVolumes(context.TODO(), task.Pod, podVolumes)
}

// BindPodVolumes gets the cached bindings and PVCs to provision in pod's volumes information,
// makes the API update for those PVs/PVCs, and waits for the PVCs to be completely bound by the PV controller.
func (b *volumeBinder) BindPodVolumes(...) (err error) {
	...

	if podVolumes == nil {
		return nil
	}

	bindings := podVolumes.StaticBindings					// static provisioner
	claimsToProvision := podVolumes.DynamicProvisions		// dynamic provisioner

	// Start API operations
	b.bindAPIUpdate(ctx, assumedPod, bindings, claimsToProvision)
	...

	wait.PollUntilContextTimeout(ctx, time.Second, b.bindTimeout, false,func(ctx context.Context) (bool,error) {
		b, err := b.checkBindings(logger, assumedPod, bindings, claimsToProvision)
		return b, err
	})
	...
    
	return nil
}

// bindAPIUpdate makes the API update for those PVs/PVCs.
func (b *volumeBinder) bindAPIUpdate(...) error {
	...

	lastProcessedBinding := 0
	lastProcessedProvisioning := 0
	defer func() {
		// only revert assumed cached updates for volumes we haven't successfully bound.
		if lastProcessedBinding < len(bindings) {
			b.revertAssumedPVs(bindings[lastProcessedBinding:])		// 重置未更新的假设PV
		}
        
		// only revert assumed cached updates for claims we haven't updated.
		if lastProcessedProvisioning < len(claimsToProvision) {
			b.revertAssumedPVCs(claimsToProvision[lastProcessedProvisioning:])	// 重置未更新的假设PVC
		}
	}()
	...

    // 更新PV执行预绑定(PV.spec.claimRef已被scheduler修改)
	for _, binding = range bindings {
		// TODO: does it hurt if we make an api call and nothing needs to be updated?
		newPV, err := b.kubeClient.CoreV1().PersistentVolumes().Update(ctx, binding.pv, metav1.UpdateOptions{})
		...

		// Save updated object from apiserver for later checking.
		binding.pv = newPV
		lastProcessedBinding++
	}

    // 更新PVC触发动态供应(PVC已被scheduler修改)
	for i, claim = range claimsToProvision {
		newClaim, err := b.kubeClient.CoreV1().PVC(claim.Namespace).Update(ctx, claim, metav1.UpdateOptions{})
		...

		// Save updated object from apiserver for later checking.
		claimsToProvision[i] = newClaim
		lastProcessedProvisioning++
	}

	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
65
66
67
68
69
70
71
72
73
74
75
76
77

注意

task PV/PVC对象已经由scheduler假设绑定，这里会直接进行修改

# 2.3.checkvol

vb.checkBindings()用于检查 volume bind情况，主要检查PV/PVC绑定情况及节点亲和，仅当所有关联PV/PVC绑定完成才认为检查通过。

// check if the PVC is fully bound, if there are any conditions that require binding to fail and be retried.
func (b *volumeBinder) checkBindings(...) (bool, error) {
	...

    // 获取pod schedNode
	node, err := b.nodeLister.Get(pod.Spec.NodeName)
	...

    // 获取csiNode
	csiNode, err := b.csiNodeLister.Get(node.Name)
	...

	// Check for any conditions that might require scheduling retry

	// 获取最新底层pod
	_, err = b.podLister.Pods(pod.Namespace).Get(pod.Name)
	if err != nil {
        // pod未找到无需进一步检查
		if apierrors.IsNotFound(err) {
			return false, fmt.Errorf("pod does not exist any more: %w", err)
		}
	}

	for _, binding := range bindings {
        // 缓存的pv
		pv, err := b.pvCache.GetAPIPV(binding.pv.Name)
        ...

        // 缓存的pvc
		pvc, err := b.pvcCache.GetAPIPVC(getPVCName(binding.pvc))
		...

		// skip if API object is older and wait for new API object propagated from apiserver.
		if versioner.CompareResourceVersion(binding.pv, pv) > 0 {
			return false, nil
		}

        // in-tree-->out-tree pv
		pv, err = b.tryTranslatePVToCSI(pv, csiNode)
		...

		// Check PV's node affinity (the node might not have the proper label)
		volume.CheckNodeAffinity(pv, node.Labels)
        ...

		// Check if pv.ClaimRef got dropped by unbindVolume()
		if pv.Spec.ClaimRef == nil || pv.Spec.ClaimRef.UID == "" {
			return false, fmt.Errorf("ClaimRef got reset for pv %q", pv.Name)
		}

		// bind&completeAnno
		if !b.isPVCFullyBound(pvc) {
			return false, nil
		}
	}

	for _, claim := range claimsToProvision {
        // 缓存的pvc
		pvc, err := b.pvcCache.GetAPIPVC(getPVCName(claim))
		...

		// skip if API object is older and wait for new API object propagated from apiserver.
		if versioner.CompareResourceVersion(claim, pvc) > 0 {
			return false, nil
		}

		// Check if selectedNode annotation is still set
		if pvc.Annotations == nil {
			return false, fmt.Errorf("selectedNode annotation reset for PVC %q", pvc.Name)
		}
        
		selectedNode := pvc.Annotations[volume.AnnSelectedNode]
		if selectedNode != pod.Spec.NodeName {
			return false, fmt.Errorf("provisioning failed for PVC %q", pvc.Name)
		}

		// If the PVC is bound to a PV, check its node affinity
		if pvc.Spec.VolumeName != "" {
            // 获取缓存pv
			pv, err := b.pvCache.GetAPIPV(pvc.Spec.VolumeName)
			...

            // in-tree-->out-tree pv
			pv, err = b.tryTranslatePVToCSI(pv, csiNode)
			...

            // pv&node亲和检查
			volume.CheckNodeAffinity(pv, node.Labels)
            ...
		}

		// bind&completeAnno
		if !b.isPVCFullyBound(pvc) {
			return false, nil
		}
	}

	// All pvs and pvcs that we operated on are bound
	return true, nil
}

注意

PV/PVC均绑定完成及节点亲和匹配才会继续协调task bind，否则task重入队等待下一轮调度绑定

# 2.4.bindnode

sc.Bind()用于将PVC/PV就绪的task pod绑定到node，绑定失败的Pod会进一步更新condition/nominatedNode及设置不可调度。

// Bind binds task to the target host.
func (sc *SchedulerCache) Bind(tasks []*schedulingapi.TaskInfo) {
	...
	sc.Binder.Bind(sc.kubeClient, tasks)
	...

	for _, task := range tasks {
		if reason, ok := errMsg[task.UID]; ok {
            // 更新pod不可调度
			unschedulableMsg := fmt.Sprintf("failed to bind to node %s: %s", task.NodeName, reason)
			sc.taskUnschedulable(task, schedulingapi.PodReasonSchedulerError, unschedulableMsg, "")
            ...
            
            // 重置假设绑定的pv和pvc
			sc.VolumeBinder.RevertVolumes(task, task.PodVolumes)
            // errTasks.AddLimited
			sc.resyncTask(task)
		}
	}
}

// Bind will send bind request to api server
func (db *DefaultBinder) Bind(...) map[schedulingapi.TaskID]string {
	...
	for _, task := range tasks {
		p := task.Pod
        // 基于binding子资源更新pod.spec.nodeName
		if err := db.kubeclient.CoreV1().Pods(p.Namespace).Bind(context.TODO(),
			&v1.Binding{
				ObjectMeta: ObjectMeta{Namespace: p.Namespace, Name: p.Name, UID: p.UID, Anno: p.Annotations},
				Target: v1.ObjectReference{
					Kind: "Node",
					Name: task.NodeName,
				},
			},
			metav1.CreateOptions{})
        ...
	}

	return errMsg
}

// taskUnschedulable updates pod status of pending task
func (sc *SchedulerCache) taskUnschedulable(...) error {
	pod := task.Pod

	condition := &v1.PodCondition{
		Type:    v1.PodScheduled,
		Status:  v1.ConditionFalse,
		Reason:  reason, // Add more reasons in order to distinguish more specific scenario of pending tasks
		Message: message,
	}

	updateCond := podConditionHaveUpdate(&pod.Status, condition)

	updateNomiNode := len(nominatedNode) > 0 && podNominatedNodeNameNeedUpdate(&pod.Status, nominatedNode)

    // condition/nominatedNodeName差异
	if updateCond || updateNomiNode {
		pod = pod.DeepCopy()

		if updateCond {
			podutil.UpdatePodCondition(&pod.Status, condition)
		}

		// if nominatedNode field changed, we should update it to the pod status.
		if updateNomiNode {
			pod.Status.NominatedNodeName = nominatedNode
		}

		// The reason field in PodCondition can be "Unschedulable"
		sc.StatusUpdater.UpdatePodStatus(pod)
        ...
	}

	return nil
}

注意

bindNode本质是修改pod.spec.nodeName进行绑定，无法绑定则会设置不可调度及重置PV/PVC缓存状态