vcscheduler
南风未起 2026-05-22 20:36:22 volcano
# 1.scheduler
# 1.1.start
pc.runOnce()会初始化session对象周期触发一次，基于session周期依次执行配置文件顺序定义的action，action进一步调用plugin。// executes a single scheduling cycle. This function is called periodically by schedule period. func (pc *Scheduler) runOnce() { ... // Load ConfigMap to check which action is enabled. for _, action := range actions { conf.EnabledActionMap[action.Name()] = true } // initialize session ssn := framework.OpenSession(pc.cache, plugins, configurations) defer func() { framework.CloseSession(ssn) }() // 顺序执行action for _, action := range actions { // 内部调用plugin action.Execute(ssn) } } // OpenSession start the session func OpenSession(cache cache.Cache, tiers []conf.Tier, configurations []conf.Configuration) *Session { ssn := openSession(cache) // 基于cache构造session ssn.Tiers = tiers // plugins ssn.Configurations = configurations // 配置 ssn.NodeMap = GenerateNodeMapAndSlice(ssn.Nodes) // nodeInfo map ssn.PodLister = NewPodLister(ssn) // podLister for _, tier := range tiers { for _, plugin := range tier.Plugins { // initialize plugin if pb, found := GetPluginBuilder(plugin.Name); found { plugin := pb(plugin.Arguments) ssn.plugins[plugin.Name()] = plugin plugin.OnSessionOpen(ssn) } } } return ssn } func openSession(cache cache.Cache) *Session { ssn := &Session{ ... } snapshot := cache.Snapshot() ssn.Jobs = snapshot.Jobs for _, job := range ssn.Jobs { ... // 合法性检查 if vjr := ssn.JobValid(job); vjr != nil { // 检查未通过 if !vjr.Pass { // pg unschedulable jc := &scheduling.PodGroupCondition{ Type: scheduling.PodGroupUnschedulableType, Status: v1.ConditionTrue, LastTransitionTime: metav1.Now(), TransitionID: string(ssn.UID), Reason: vjr.Reason, Message: vjr.Message, } // 更新pg condition ssn.UpdatePodGroupCondition(job, jc) ... } // 检查通过为什么也会删除? delete(ssn.Jobs, job.UID) } } ... // calculate all nodes' resource only once in each schedule cycle, other plugins can clone it when need for _, n := range ssn.Nodes { ssn.TotalResource.Add(n.Allocatable) } return ssn } // JobValid invoke jobvalid function of the plugins func (ssn *Session) JobValid(obj interface{}) *api.ValidateResult { for _, tier := range ssn.Tiers { for _, plugin := range tier.Plugins { // plugin向session注册过validFn(gang) jrf, found := ssn.jobValidFns[plugin.Name] if !found { continue } // 执行validFn if vr := jrf(obj); vr != nil && !vr.Pass { return vr } } } 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
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
注意
pluginBuilder由前面介绍的.so动态链接库加载，action.Execute内部会执行plugin
# 1.2.action
action注册由init模块触发，action会基于RegisterAction注册到actionMap，加载配置时基于actionName整合启用的action。func init() { framework.RegisterAction(reclaim.New()) framework.RegisterAction(allocate.New()) framework.RegisterAction(backfill.New()) framework.RegisterAction(preempt.New()) framework.RegisterAction(enqueue.New()) framework.RegisterAction(shuffle.New()) } // RegisterAction register action func RegisterAction(act Action) { pluginMutex.Lock() defer pluginMutex.Unlock() actionMap[act.Name()] = act } // GetAction get the action by name func GetAction(name string) (Action, bool) { pluginMutex.RLock() defer pluginMutex.RUnlock() act, found := actionMap[name] return act, found } func (pc *Scheduler) loadSchedulerConf() { ... actions, plugins, configurations, metricsConf, err := UnmarshalSchedulerConf(config) ... } func UnmarshalSchedulerConf(confStr string) ([]framework.Action, []conf.Tier, []conf.Configuration, map[string]string, error) { var actions []framework.Action ... actionNames := strings.Split(schedulerConf.Actions, ",") for _, actionName := range actionNames { if action, found := framework.GetAction(strings.TrimSpace(actionName)); found { actions = append(actions, action) } } return actions, schedulerConf.Tiers, schedulerConf.Configurations, schedulerConf.MetricsConfiguration, 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
注意
enable action由配置决定，默认全部加载，基于配置整理enableAction
# 2.statement
# 2.1.allocate
stmt.Allocate()用于调度阶段将任务假设分配到节点，stmt.UnAllocate()用于调度失败回滚恢复之前的资源状态，两者配合实现任务负载分配。// Allocate the task to node func (s *Statement) Allocate(task *api.TaskInfo, nodeInfo *api.NodeInfo) (err error) { ... hostname := nodeInfo.Name // 获取podVolume podVolumes, err := s.ssn.cache.GetPodVolumes(task, nodeInfo.Node) ... // 申请podVolume资源,PV和PVC假设绑定缓存至cache,标记调度到的node信息 s.ssn.cache.AllocateVolumes(task, hostname, podVolumes) ... task.Pod.Spec.NodeName = hostname task.PodVolumes = podVolumes // Only update status in session job, found := s.ssn.Jobs[task.Job] if found { // 更新job taskStatus job.UpdateTaskStatus(task, api.Allocated) ... } ... task.NodeName = hostname if node, found := s.ssn.Nodes[hostname]; found { // 扣减task资源占用 node.AddTask(task) ... } ... // callbacks for _, eh := range s.ssn.eventHandlers { if eh.AllocateFunc != nil { eventInfo := &Event{ Task: task, } eh.AllocateFunc(eventInfo) ... } } ... s.operations = append(s.operations, operation{ name: Allocate, task: task, }) return nil } // UnAllocate the pod for task func (s *Statement) UnAllocate(task *api.TaskInfo) error { return s.unallocate(task) } // unallocate the pod for task func (s *Statement) unallocate(task *api.TaskInfo) error { // 清理假设绑定的PV/PVC cache s.ssn.cache.RevertVolumes(task, task.PodVolumes) // Update status in session job, found := s.ssn.Jobs[task.Job] if found { // 更新job taskStatus job.UpdateTaskStatus(task, api.Pending) ... } ... // 恢复node扣减的task资源 if node, found := s.ssn.Nodes[task.NodeName]; found { node.RemoveTask(task) ... } // callbacks for _, eh := range s.ssn.eventHandlers { if eh.DeallocateFunc != nil { eh.DeallocateFunc(&Event{ Task: task }) } } task.NodeName = "" 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
85
86
87
88
89
注意
callbacks由session初始化时基于plugin注册，不同plugin注册各自实现的allocateFunc
# 2.2.piplined
stmt.Pipeline()用于资源未完全可用时提前预留节点资源，并在后续资源释放后立即执行调度，stmt.UnPipeline()用于回滚资源预留避免泄漏。// Pipeline the task for the node func (s *Statement) Pipeline(task *api.TaskInfo, hostname string, evictionOccurred bool) error { ... job, found := s.ssn.Jobs[task.Job] if found { // 更新job taskStatus job.UpdateTaskStatus(task, api.Pipelined) ... } ... task.NodeName = hostname task.EvictionOccurred = evictionOccurred(false) // 扣减node资源 if node, found := s.ssn.Nodes[hostname]; found { node.AddTask(task) ... } ... // callbacks for _, eh := range s.ssn.eventHandlers { if eh.AllocateFunc != nil { eventInfo := &Event{ Task: task } eh.AllocateFunc(eventInfo) ... } } ... s.operations = append(s.operations, operation{ name: Pipeline, task: task, }) return nil } func (s *Statement) UnPipeline(task *api.TaskInfo) error { job, found := s.ssn.Jobs[task.Job] if found { // 更新job taskStatus job.UpdateTaskStatus(task, api.Pending) ... } ... // 回滚node资源扣减 if node, found := s.ssn.Nodes[task.NodeName]; found { node.RemoveTask(task) ... } ... // callbacks for _, eh := range s.ssn.eventHandlers { if eh.DeallocateFunc != nil { eventInfo := &Event{ Task: task } eh.DeallocateFunc(eventInfo) ... } } task.NodeName = "" 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
注意
Statement会记录task对应操作，后续会基于operation及资源条件进行延迟调度
# 2.3.commit
stmt.Commit()会顺序提交调度操作列表，将调度决策落实到调度缓存，基于evict/pipline/allocate组合处理正常调度或抢占式调度任务。// Commit operation for evict and pipeline func (s *Statement) Commit() { for _, op := range s.operations { op.task.ClearLastTxContext() switch op.name { // 驱逐 case Evict: s.evict(op.task, op.reason) ... // 预调度 case Pipeline: s.pipeline(op.task) // 调度 case Allocate: err := s.allocate(op.task) if err != nil { s.unallocate(op.task) ... } } } } func (s *Statement) evict(reclaimee *api.TaskInfo, reason string) error { if err := s.ssn.cache.Evict(reclaimee, reason); err != nil { // 失败则取消驱逐 s.unevict(reclaimee) ... return err } return nil } // Evict will evict the pod. If error occurs both task and job are guaranteed to be in the original state. func (sc *SchedulerCache) Evict(taskInfo *schedulingapi.TaskInfo, reason string) error { ... job, task, err := sc.findJobAndTask(taskInfo) ... node, found := sc.Nodes[task.NodeName] ... // 更新job taskStatus originalStatus := task.Status job.UpdateTaskStatus(task, schedulingapi.Releasing) ... // 更新下node资源扣减 if err := node.UpdateTask(task); err != nil { // 更新失败,恢复job taskStatus状态 if err := job.UpdateTaskStatus(task, originalStatus); err != nil { // errTasks.AddRateLimited sc.resyncTask(task) } return err } p := task.Pod go func() { // 驱逐pod(delete pod) err := sc.Evictor.Evict(p, reason) if err != nil { // errTasks.AddRateLimited sc.resyncTask(task) } }() ... return nil } func (s *Statement) allocate(task *api.TaskInfo) error { // 注册绑定任务及真正扣减node资源 if err := s.ssn.cache.AddBindTask(task); err != nil { return err } if job, found := s.ssn.Jobs[task.Job]; found { // 更新job taskStatus job.UpdateTaskStatus(task, api.Binding) ... } ... 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
85
86
87
88
89
90
注意
stmt.Commit会将调度决策应用到scheduler cache，可调度任务会推到sc.BindFlowChannel供异步协程处理
# 2.4.discard
stmt.Discard()用于撤销尚未提交成功的调度操作，进行相关资源及动作的回滚，进而恢复调度缓存和任务状态，避免破坏调度数据的一致性。// Discard operation for evict, pipeline and allocate func (s *Statement) Discard() { // 逆向回滚 for i := len(s.operations) - 1; i >= 0; i-- { op := s.operations[i] op.task.GenerateLastTxContext() switch op.name { case Evict: // 回滚驱逐 s.unevict(op.task) ... case Pipeline: // 回滚预调度 s.UnPipeline(op.task) ... case Allocate: // 回滚绑定 s.unallocate(op.task) ... } } } func (s *Statement) unevict(reclaimee *api.TaskInfo) error { // Update status in session job, found := s.ssn.Jobs[reclaimee.Job] if found { // 更新job taskStatus job.UpdateTaskStatus(reclaimee, api.Running) ... } ... // Update task in node. if node, found := s.ssn.Nodes[reclaimee.NodeName]; found { // 更新节点资源占用 node.UpdateTask(reclaimee) ... } // callbacks for _, eh := range s.ssn.eventHandlers { if eh.AllocateFunc != nil { eh.AllocateFunc(&Event{ Task: reclaimee }) } } return nil } func (s *Statement) UnPipeline(task *api.TaskInfo) error { job, found := s.ssn.Jobs[task.Job] if found { // 更新job taskStatus job.UpdateTaskStatus(task, api.Pending) ... } ... // 回滚node资源扣减 if node, found := s.ssn.Nodes[task.NodeName]; found { node.RemoveTask(task) ... } ... // callbacks for _, eh := range s.ssn.eventHandlers { if eh.DeallocateFunc != nil { eventInfo := &Event{ Task: task } eh.DeallocateFunc(eventInfo) ... } } task.NodeName = "" return nil } // unallocate the pod for task func (s *Statement) unallocate(task *api.TaskInfo) error { // 重置假设绑定的podVolumes s.ssn.cache.RevertVolumes(task, task.PodVolumes) // Update status in session job, found := s.ssn.Jobs[task.Job] if found { job.UpdateTaskStatus(task, api.Pending) ... } ... if node, found := s.ssn.Nodes[task.NodeName]; found { // 回滚node资源扣减 node.RemoveTask(task) ... } // callbacks for _, eh := range s.ssn.eventHandlers { if eh.DeallocateFunc != nil { eh.DeallocateFunc(&Event{ Task: task }) } } task.NodeName = "" 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
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
注意
stmt.Discard负责回滚不可行的调度操作，本质是恢复session/cache的缓存及任务状态
# 3.selectnode
# 3.1.predicate
ph.PredicateNodes()用于节点预选，基于已有节点及最小搜索节点选出可调度task pod的N个节点，预选底层还是依赖plugin实现。func (alloc *Action) predicate(task *api.TaskInfo, node *api.NodeInfo) error { ... // idle+releasing-piplined资源不足 if ok, resources := task.InitResreq.LessEqualWithResourcesName(node.FutureIdle(), api.Zero); !ok { // 标记不可调度 statusSets = append(statusSets, &api.Status{Code: api.Unschedulable, Reason: api.WrapInsufficientResourceReason(resources)}) return api.NewFitErrWithStatus(task, node, statusSets...) } // plugin.Predicate(感知Unschedulable、UnschedulableAndUnresolvable、ErrorSkipOrWait错误) return alloc.session.PredicateForAllocateAction(task, node) } // returns the number of feasible nodes that once found, the scheduler stops its search for more feasible nodes. func CalculateNumOfFeasibleNodesToFind(numAllNodes int32) (numNodes int32) { opts := options.ServerOpts // 节点数未达到最小搜索节点或搜索百分比超出100,本质为全搜 if numAllNodes <= opts.MinNodesToFind || opts.PercentageOfNodesToFind >= 100 { return numAllNodes } // 搜索百分比 adaptivePercentage := opts.PercentageOfNodesToFind // 未设置搜索百分比 if adaptivePercentage <= 0 { // 动态计算(50-all/125) adaptivePercentage = baselinePercentageOfNodesToFind - numAllNodes/125 // 基于设置的最小搜索百分比修正 if adaptivePercentage < opts.MinPercentageOfNodesToFind { adaptivePercentage = opts.MinPercentageOfNodesToFind } } // 计算待搜索节点数 numNodes = numAllNodes * adaptivePercentage / 100 // 基于最小搜索节点数修正 if numNodes < opts.MinNodesToFind { numNodes = opts.MinNodesToFind } return numNodes } // PredicateNodes returns the specified number of nodes that fit a task func (ph *predicateHelper) PredicateNodes(...) ([]*api.NodeInfo, *api.FitErrors) { ... fe := api.NewFitErrors() // don't enable error cache if task's TaskRole is empty, because different pods with empty TaskRole will all // have the same taskGroupID, and one pod predicate failed, all other pods will also be failed. if len(task.TaskRole) == 0 { enableErrorCache = false } ... // 待搜索节点数 numNodesToFind := CalculateNumOfFeasibleNodesToFind(int32(allNodes)) ... checkNode := func(index int) { // 基于上次结束位置扫描 node := nodes[(lastProcessedNodeIndex+index)%allNodes] // 更新处理节点数 atomic.AddInt32(&processedNodes, 1) // errCache启用&taskGroup之前失败过 if enableErrorCache && taskFailedBefore { ... // 查询缓存 errC, ok := nodeErrorCache[node.Name] ... if ok { ... // 设置err信息 fe.SetNodeError(node.Name, errC) ... return } } // 执行predicate检查 if err := fn(task, node); err != nil { ... nodeErrorCache[node.Name] = err ph.taskPredicateErrorCache[taskGroupid] = nodeErrorCache fe.SetNodeError(node.Name, err) ... return } // 更新numFoundNodes数量 length := atomic.AddInt32(&numFoundNodes, 1) // 满足最小搜索节点数量 if length > numNodesToFind { // 回滚退出 cancel() atomic.AddInt32(&numFoundNodes, -1) } else { // 记录predicate节点 predicateNodes[length-1] = node } } // 16个worker并发搜索node workqueue.ParallelizeUntil(ctx, 16, allNodes, checkNode) // 记录上次搜索索引 lastProcessedNodeIndex = (lastProcessedNodeIndex + int(processedNodes)) % allNodes // 获取预选的节点 predicateNodes = predicateNodes[:numFoundNodes] return predicateNodes, fe }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
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
注意
Predicate会基于16 worker并发执行节点搜索，搜索起始位置由上一轮结束位置决定
# 3.2.prioritize
util.PrioritizeNodes()负责对预选节点进行优先级打分，计算节点对当前任务的适配度，相同得分的节点归为一组，用于后续的节点优选决策。基于预选结果计算节点得分，供后续节点哟选啊
// BatchNodeOrderFn invoke node order function of the plugins func (ssn *Session) BatchNodeOrderFn(task *api.TaskInfo, nodes []*api.NodeInfo) (map[string]float64, error) { ... for _, tier := range ssn.Tiers { for _, plugin := range tier.Plugins { ... // plugin.batchNodeOrderFn批计算节点得分 pfn, found := ssn.batchNodeOrderFns[plugin.Name] ... score, err := pfn(task, nodes) ... // 累加节点得分 for nodeName, score := range score { priorityScore[nodeName] += score } } } return priorityScore, nil } // NodeOrderMapFn invoke node order function of the plugins func (ssn *Session) NodeOrderMapFn(...) (map[string]float64, float64, error) { ... for _, tier := range ssn.Tiers { for _, plugin := range tier.Plugins { ... // plugin.NodeOrderFn计算节点总分 if pfn, found := ssn.nodeOrderFns[plugin.Name]; found { score, err := pfn(task, node) ... priorityScore += score } // plugin.NodeMapFn计算各插件评分 if pfn, found := ssn.nodeMapFns[plugin.Name]; found { score, err := pfn(task, node) ... nodeScoreMap[plugin.Name] = score } } } return nodeScoreMap, priorityScore, nil } // NodeOrderReduceFn invoke node order function of the plugins func (ssn *Session) NodeOrderReduceFn(...) (map[string]float64, error) { ... for _, tier := range ssn.Tiers { for _, plugin := range tier.Plugins { ... // plugin.NodeReduceFn计算归一化得分 pfn, found := ssn.nodeReduceFns[plugin.Name] ... pfn(task, pluginNodeScoreMap[plugin.Name]) ... // 节点最终得分 for _, hp := range pluginNodeScoreMap[plugin.Name] { nodeScoreMap[hp.Name] += float64(hp.Score) } } } return nodeScoreMap, nil } // PrioritizeNodes returns a map whose key is node's score and value are corresponding nodes func PrioritizeNodes(...) map[float64][]*api.NodeInfo { ... scoreNode := func(index int) { node := nodes[index] mapScores, orderScore, err := mapFn(task, node) ... // plugin打分表 for plugin, score := range mapScores { nodeScoreList, ok := pluginNodeScoreMap[plugin] ... hp := k8sframework.NodeScore{} hp.Name = node.Name hp.Score = int64(math.Floor(score)) pluginNodeScoreMap[plugin] = append(nodeScoreList, hp) } // node总分表 nodeOrderScoreMap[node.Name] = orderScore ... } workqueue.ParallelizeUntil(context.TODO(), 16, len(nodes), scoreNode) // 归一化plugin score reduceScores, err := reduceFn(task, pluginNodeScoreMap) ... // plugin.NodeOrderFn批计算node score batchNodeScore, err := batchFn(task, nodes) ... // 计算节点总得分 for _, node := range nodes { score := 0.0 if reduceScore, ok := reduceScores[node.Name]; ok { score += reduceScore } if orderScore, ok := nodeOrderScoreMap[node.Name]; ok { score += orderScore } if batchScore, ok := batchNodeScore[node.Name]; ok { score += batchScore } nodeScores[score] = append(nodeScores[score], node) ... } return nodeScores }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
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
注意
由
PrioritizeNodes实现可以看出，节点打分底层还是由plugin负责