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

# 1.简介

# 1.1.原理

volcano scheduler由一系列action&plugin组成，action定义调度环节动作，plugin定义action算法实现，两者配合进行Pod调度。

--- 调度流程
	1.scheduler观察及缓存job
	2.间隔一段时间开启会话,触发调度周期
	3.未调度job会推到会话的待调度队列
	4.待调度job执行enqueue/allocate/preempt/reclaim/backfill action,匹配job最合适的node
	5.job绑定到node,关闭本次会话

1
2
3
4
5
6

注意

action执行的具体算法逻辑由注册的plugin决定，不同的plugin实现不同

# 1.2.入口

pc.Run()用于加载监听scheduler conf、激活cache informer及间隔1s触发调度周期协调Job，程序终止会基于配置采集cache数据。

// Run the volcano scheduler.
func Run(opt *options.ServerOption) error {
	...

    // 加载自定义插件
	if opt.PluginsDir != "" {
		framework.LoadCustomPlugins(opt.PluginsDir)
		...
	}

    // 初始化scheduler
	sched, err := scheduler.NewScheduler(config, opt)
	...
	
    // 运行
    sched.Run(ctx.Done())
    <-ctx.Done()
	...
    
	return fmt.Errorf("finished")
}

// LoadCustomPlugins loads custom implement plugins
func LoadCustomPlugins(pluginsDir string) error {
	pluginPaths, _ := filepath.Glob(fmt.Sprintf("%s/*.so", pluginsDir))
	for _, pluginPath := range pluginPaths {
        // 加载builder
		pluginBuilder, err := loadPluginBuilder(pluginPath)
		...
        
        // 解析插件名
		pluginName := getPluginName(pluginPath)
        // 注册builder(New方法)
		RegisterPluginBuilder(pluginName, pluginBuilder)
	}

	return nil
}

func loadPluginBuilder(pluginPath string) (PluginBuilder, error) {
    // 加载Go包
	plug, err := plugin.Open(pluginPath)
	...

    // New方法符号值
	symBuilder, err := plug.Lookup("New")
	...

    // 转为pluginBuilder
	builder, ok := symBuilder.(PluginBuilder)
	...
    
	return builder, nil
}

// NewScheduler returns a Scheduler
func NewScheduler(config *rest.Config, opt *options.ServerOption) (*Scheduler, error) {
	...
    
	if opt.SchedulerConf != "" {
		// 监听scheduler配置目录
		path := filepath.Dir(opt.SchedulerConf)
		watcher = filewatcher.NewFileWatcher(path)
		...
	}

	cache := schedcache.New(config, SchedulerNames, DefaultQueue, NodeSelector, 20, opt.IgnoredCSIProvisioners)
	scheduler := &Scheduler{
		schedulerConf:  opt.SchedulerConf,
		fileWatcher:    watcher,
		cache:          cache,
		schedulePeriod: opt.SchedulePeriod,
		dumper:         schedcache.Dumper{Cache: cache, RootDir: opt.CacheDumpFileDir},
	}

	return scheduler, 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

注意

sched.Run执行的是调度器的核心逻辑，涉及配置加载、监听、周期调度及cache快照采集

# 1.3.运行

pc.Run()会加载schedConf及监听配置变化，开启informer及缓存，间隔1s执行调度周期，检测到程序退出会进行cache快照采集。

// initialize and start the Scheduler. It loads configuration, initializes cache, and begins scheduling process.
func (pc *Scheduler) Run(stopCh <-chan struct{}) {
    // 加载配置
	pc.loadSchedulerConf()
    
    // 监听配置
	go pc.watchSchedulerConf(stopCh)
	...
    
    // 激活informer
	pc.cache.Run(stopCh)
	
    // 间隔1s执行调度
	go wait.Until(pc.runOnce, pc.schedulePeriod, stopCh)

    // 退出采集cache数据
    if options.ServerOpts.EnableCacheDumper {
		pc.dumper.ListenForSignal(stopCh)
	}
    
    // 日志API,允许查询/修改日志参数
	go runSchedulerSocket()
}

func (pc *Scheduler) watchSchedulerConf(stopCh <-chan struct{}) {
	if pc.fileWatcher == nil {
		return
	}
	...
    
	for {
		select {
		case event, ok := <-pc.fileWatcher.Events():
			if !ok {
				return
			}
			
            // 写入或创建配置
			if event.Op&fsnotify.Write == fsnotify.Write || event.Op&fsnotify.Create == fsnotify.Create {
                // 重新加载
				pc.loadSchedulerConf()
				...
			}
		...
		case <-stopCh:
			return
		}
	}
}

// ListenForSignal starts a goroutine that will respond when process receives SIGUSER1/SIGUSER2 signal.
func (d *Dumper) ListenForSignal(stopCh <-chan struct{}) {
	...
    
	go func() {
		for {
			select {
			case <-stopCh:
				return
            // syscall.SIGUSR1  kill -10 <pid>
			case <-ch1:
				d.dumpToJSONFile()			// cache备份到json文件
            // syscall.SIGUSR2  kill -12 <pid>
			case <-ch2:
				d.dumpAll()					// cache输出为日志
			}
		}
	}()
}

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

注意

cache.Run和pc.runOnce是驱动及维护调度的核心，其余模块均为配置监听、日志参数调整及退出采集cache快照的辅助

# 2.cacher

# 2.1.initialize

schedcache.New()会初始化cache，设置相关缓存参数及构造绑定更新模块，注册informer及相关eventHandler，创建volumeBinder。

// New returns a Cache implementation.
func New(...) Cache {
	return newSchedulerCache(config, schedulerNames, defaultQueue, nodeSelectors, 20, ignoredProvisioners)
}

func newSchedulerCache(...) *SchedulerCache {
	...

	// create default queue and root queue
	newDefaultAndRootQueue(vcClient, defaultQueue)
	...

	sc := &SchedulerCache{
		...
		errTasks:            workqueue.NewRateLimitingQueue(errTaskRateLimiter),
		nodeQueue:           workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
		DeletedJobs:         workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
		...
		defaultQueue:        defaultQueue,
		schedulerNames:      schedulerNames,
		...
		nodeWorkers: 20,
	}

    // scheduler pod names
	sc.schedulerPodName, sc.c = getMultiSchedulerInfo()
	...
	sc.IgnoredCSIProvisioners = ignoredProvisionersSet
	...

	// set concurrency configuration when binding
	sc.setBatchBindParallel()
	...    
	sc.Binder = GetBindMethod()
	sc.Evictor = &defaultEvictor{ kubeclient: sc.kubeClient, recorder: sc.Recorder }
	sc.StatusUpdater = &defaultStatusUpdater{ kubeclient: sc.kubeClient, vcclient: sc.vcClient }
	sc.PodGroupBinder = &podgroupBinder{ kubeclient: sc.kubeClient, vcclient: sc.vcClient }
    
	// add all events handlers
	sc.addEventHandler()
    
	// finally, init default volume binder which has dependencies on other informers
	sc.setDefaultVolumeBinder()
	return sc
}

func (sc *SchedulerCache) setDefaultVolumeBinder() {
	...
    
    // CSIStorage enable
	if options.ServerOpts.EnableCSIStorage && utilfeature.DefaultFeatureGate.Enabled(features.CSIStorage) {
		// capacity check
        capacityCheck = &volumescheduling.CapacityCheck{
			CSIDriverInformer:          sc.csiDriverInformer,
			CSIStorageCapacityInformer: sc.csiStorageCapacityInformer,
		}
	}
	sc.VolumeBinder = &defaultVolumeBinder{
		volumeBinder: volumescheduling.NewVolumeBinder(
			...
			sc.podInformer,
			sc.nodeInformer,
			sc.csiNodeInformer,
			sc.pvcInformer,
			sc.pvInformer,
			sc.scInformer,
			capacityCheck,
			30*time.Second,
		),
	}
}

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

注意

schedcache相关的维护一系列workqueue及状态更新组件，会注册相关资源的informer/eventHandler

# 2.2.handler

cache会初始化core/storage/scheduling/volcano informer及注册相关资源的eventHandler，部分资源会注册FilterFunc限制监听。

func (sc *SchedulerCache) addEventHandler() {
	...

	// create informer for node information
	sc.nodeInformer = informerFactory.Core().V1().Nodes()
	sc.nodeInformer.Informer().AddEventHandlerWithResyncPeriod(
		cache.FilteringResourceEventHandler{
			FilterFunc: func(obj interface{}) bool {
				// 限制node
			},
			Handler: cache.ResourceEventHandlerFuncs{
				AddFunc:    sc.AddNode,
				UpdateFunc: sc.UpdateNode,
				DeleteFunc: sc.DeleteNode,
			},
		},
		0,
	)

	sc.podInformer = informerFactory.Core().V1().Pods()
	sc.pvcInformer = informerFactory.Core().V1().PersistentVolumeClaims()
	sc.pvInformer = informerFactory.Core().V1().PersistentVolumes()
	sc.scInformer = informerFactory.Storage().V1().StorageClasses()
	sc.csiNodeInformer = informerFactory.Storage().V1().CSINodes()
	sc.csiNodeInformer.Informer().AddEventHandler(
		cache.ResourceEventHandlerFuncs{
			AddFunc:    sc.AddOrUpdateCSINode,
			UpdateFunc: sc.UpdateCSINode,
			DeleteFunc: sc.DeleteCSINode,
		},
	)

    // CSIStorage
	if options.ServerOpts.EnableCSIStorage && utilfeature.DefaultFeatureGate.Enabled(features.CSIStorage) {
		sc.csiDriverInformer = informerFactory.Storage().V1().CSIDrivers()
		sc.csiStorageCapacityInformer = informerFactory.Storage().V1beta1().CSIStorageCapacities()
	}

	// create informer for pod information
	sc.podInformer.Informer().AddEventHandler(
		cache.FilteringResourceEventHandler{
			FilterFunc: func(obj interface{}) bool {
				// 限制pod
			},
			Handler: cache.ResourceEventHandlerFuncs{
				AddFunc:    sc.AddPod,
				UpdateFunc: sc.UpdatePod,
				DeleteFunc: sc.DeletePod,
			},
		})

    // PriorityClass
	if options.ServerOpts.EnablePriorityClass && utilfeature.DefaultFeatureGate.Enabled(features.PriorityClass){
		sc.pcInformer = informerFactory.Scheduling().V1().PriorityClasses()
		sc.pcInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
			AddFunc:    sc.AddPriorityClass,
			UpdateFunc: sc.UpdatePriorityClass,
			DeleteFunc: sc.DeletePriorityClass,
		})
	}

    // ResourceQuotas
	sc.quotaInformer = informerFactory.Core().V1().ResourceQuotas()
	sc.quotaInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc:    sc.AddResourceQuota,
		UpdateFunc: sc.UpdateResourceQuota,
		DeleteFunc: sc.DeleteResourceQuota,
	})	
    ...

	// create informer for PodGroup(v1beta1) information
	sc.podGroupInformerV1beta1 = vcinformers.Scheduling().V1beta1().PodGroups()
	sc.podGroupInformerV1beta1.Informer().AddEventHandler(
		cache.FilteringResourceEventHandler{
			FilterFunc: func(obj interface{}) bool {
				// 限制podgroup及匹配schedulerName
			},
			Handler: cache.ResourceEventHandlerFuncs{
				AddFunc:    sc.AddPodGroupV1beta1,
				UpdateFunc: sc.UpdatePodGroupV1beta1,
				DeleteFunc: sc.DeletePodGroupV1beta1,
			},
		})

	// create informer(v1beta1) for Queue information
	sc.queueInformerV1beta1 = vcinformers.Scheduling().V1beta1().Queues()
	sc.queueInformerV1beta1.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc:    sc.AddQueueV1beta1,
		UpdateFunc: sc.UpdateQueueV1beta1,
		DeleteFunc: sc.DeleteQueueV1beta1,
	})

    // ResourceTopology
	if utilfeature.DefaultFeatureGate.Enabled(features.ResourceTopology) {
		sc.cpuInformer = vcinformers.Nodeinfo().V1alpha1().Numatopologies()
		sc.cpuInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
			AddFunc:    sc.AddNumaInfoV1alpha1,
			UpdateFunc: sc.UpdateNumaInfoV1alpha1,
			DeleteFunc: sc.DeleteNumaInfoV1alpha1,
		})
	}
}

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

注意

cache informer会监听core/storage/scheduling/volcano相关的资源及注册回调，基于回调将关联资源推入workqueue

# 3.informer

# 3.1.nodehandler

node/csinode资源变化会触发关联node入队，csinode变化还会额外维护csiNodeStatus，相应的状态会更新到cache或由cache清理。

// AddNode add node to scheduler cache
func (sc *SchedulerCache) AddNode(obj interface{}) {
	node, ok := obj.(*v1.Node)
	...
    
	sc.nodeQueue.Add(node.Name)
}

// UpdateNode update node to scheduler cache
func (sc *SchedulerCache) UpdateNode(oldObj, newObj interface{}) {
	_, ok := oldObj.(*v1.Node)
	...
    
	newNode, ok := newObj.(*v1.Node)
	...
    
	sc.nodeQueue.Add(newNode.Name)
}

// DeleteNode delete node from scheduler cache
func (sc *SchedulerCache) DeleteNode(obj interface{}) {
	...
    
	sc.nodeQueue.Add(node.Name)
}

func (sc *SchedulerCache) AddOrUpdateCSINode(obj interface{}) {
	csiNode, ok := obj.(*sv1.CSINode)
	...

	csiNodeStatus := &schedulingapi.CSINodeStatusInfo{
		CSINodeName:  csiNode.Name,
		DriverStatus: make(map[string]bool),
	}
	...

	for i := range csiNode.Spec.Drivers {
		d := csiNode.Spec.Drivers[i]
		csiNodeStatus.DriverStatus[d.Name] = d.Allocatable != nil && d.Allocatable.Count != nil
	}
    
	sc.CSINodesStatus[csiNode.Name] = csiNodeStatus
	sc.nodeQueue.Add(csiNode.Name)
}

func (sc *SchedulerCache) UpdateCSINode(oldObj, newObj interface{}) {
	oldCSINode, ok := oldObj.(*sv1.CSINode)
	...
    
	newCSINode, ok := newObj.(*sv1.CSINode)
	...
    
	if equality.Semantic.DeepEqual(oldCSINode.Spec, newCSINode.Spec) {
		return
	}
    
	sc.AddOrUpdateCSINode(newObj)
}

func (sc *SchedulerCache) DeleteCSINode(obj interface{}) {
	...
    
	delete(sc.CSINodesStatus, csiNode.Name)
	...
	sc.nodeQueue.Add(csiNode.Name)
}

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

注意

node/csinode监听相对简单，主要将关联node推到nodeQueue

# 3.2.podhandler

podInformer会监听Pod相关事件，基于Pod初始化关联JobInfo/NodeInfo，向nodeInfo/JobInfo注册资源占用及缓存到cache。

// AddPod add pod to scheduler cache
func (sc *SchedulerCache) AddPod(obj interface{}) {
	pod := obj.(*v1.Pod)
	...

	sc.addPod(pod)
	...
}

// Assumes that lock is already acquired.
func (sc *SchedulerCache) addPod(pod *v1.Pod) error {
    // 基于pod初始化task
	pi, err := sc.NewTaskInfo(pod)
	if err != nil {
		// 推入errTasks重试
		sc.resyncTask(pi)
	}
    
    // 注册任务
	return sc.addTask(pi)
}

func (sc *SchedulerCache) addTask(pi *schedulingapi.TaskInfo) error {
	if len(pi.NodeName) != 0 {
		if _, found := sc.Nodes[pi.NodeName]; !found {
			sc.Nodes[pi.NodeName] = schedulingapi.NewNodeInfo(nil)
			sc.Nodes[pi.NodeName].Name = pi.NodeName
		}

		node := sc.Nodes[pi.NodeName]
        // 未终止
		if !isTerminated(pi.Status) {
            // 向nodeInfo注册任务
			node.AddTask(pi)
            ...
		}
	}

    // 创建及获取job
	job := sc.getOrCreateJob(pi)
	if job != nil {
        // 更新job TotalRequest/Allocated/statusIndex
		job.AddTaskInfo(pi)
	}

	return nil
}

// UpdatePod update pod to scheduler cache
func (sc *SchedulerCache) UpdatePod(oldObj, newObj interface{}) {
	...

	sc.updatePod(oldPod, newPod)
	...
}

// Assumes that lock is already acquired.
func (sc *SchedulerCache) updatePod(oldPod, newPod *v1.Pod) error {
	//ignore the update event if pod is allocated in cache but not present in NodeName
	if sc.allocatedPodInCache(newPod) && newPod.Spec.NodeName == "" {
		return nil
	}

    // 先尝试清理Pod数据
	sc.deletePod(oldPod)
    ...
    
	// when delete pod, the ownerreference of pod will be set nil,just as orphan pod
	if len(utils.GetController(newPod)) == 0 {
		newPod.OwnerReferences = oldPod.OwnerReferences
	}
	return sc.addPod(newPod)
}

// DeletePod delete pod from scheduler cache
func (sc *SchedulerCache) DeletePod(obj interface{}) {
	...

	sc.deletePod(pod)
	...
}

// Assumes that lock is already acquired.
func (sc *SchedulerCache) deletePod(pod *v1.Pod) error {
	pi := schedulingapi.NewTaskInfo(pod)

	// Delete the Task in cache to handle Binding status.
	task := pi
	if job, found := sc.Jobs[pi.Job]; found {
		if t, found := job.Tasks[pi.UID]; found {
			task = t
		}
	}
    
    // 清理task
	sc.deleteTask(task)
    ...

	// If job was terminated, delete it.
	if job, found := sc.Jobs[pi.Job]; found && schedulingapi.JobTerminated(job) {
		sc.deleteJob(job)
	}

	return nil
}

注意

nodeInfo会记录所有运行Pod占用、待释放及空闲资源，jobInfo记录关联Pod总资源、申请资源及状态信息

# 3.3.pqhandler

pcInformer和quotaInformer会监听priorityClass和resourceQuota资源变化，将相应的资源对象及状态缓存到cache相应模块。

// pcInformer
// AddPriorityClass add priorityclass to scheduler cache
func (sc *SchedulerCache) AddPriorityClass(obj interface{}) {
	ss, ok := obj.(*schedulingv1.PriorityClass)
	...

	sc.addPriorityClass(ss)
}

func (sc *SchedulerCache) addPriorityClass(pc *schedulingv1.PriorityClass) {
	if pc.GlobalDefault {
		sc.defaultPriorityClass = pc
		sc.defaultPriority = pc.Value
	}

	sc.PriorityClasses[pc.Name] = pc
}

// UpdatePriorityClass update priorityclass to scheduler cache
func (sc *SchedulerCache) UpdatePriorityClass(oldObj, newObj interface{}) {
	...

	sc.deletePriorityClass(oldSS)
	sc.addPriorityClass(newSS)
}

// DeletePriorityClass delete priorityclass from the scheduler cache
func (sc *SchedulerCache) DeletePriorityClass(obj interface{}) {
	...

	sc.deletePriorityClass(ss)
}

func (sc *SchedulerCache) deletePriorityClass(pc *schedulingv1.PriorityClass) {
	if pc.GlobalDefault {
		sc.defaultPriorityClass = nil
		sc.defaultPriority = 0
	}

	delete(sc.PriorityClasses, pc.Name)
}

// quotaInformer
// AddResourceQuota add ResourceQuota to scheduler cache
func (sc *SchedulerCache) AddResourceQuota(obj interface{}) {
	...
    
	sc.updateResourceQuota(r)
}

// UpdateResourceQuota update ResourceQuota to scheduler cache
func (sc *SchedulerCache) UpdateResourceQuota(oldObj, newObj interface{}) {
	...

	sc.updateResourceQuota(newR)
}

func (sc *SchedulerCache) updateResourceQuota(quota *v1.ResourceQuota) {
	collection, ok := sc.NamespaceCollection[quota.Namespace]
	...

    // 更新quota.Status
	collection.Update(quota)
}

// DeleteResourceQuota delete ResourceQuota from the scheduler cache
func (sc *SchedulerCache) DeleteResourceQuota(obj interface{}) {
	...
    
	sc.deleteResourceQuota(r)
}

func (sc *SchedulerCache) deleteResourceQuota(quota *v1.ResourceQuota) {
	collection, ok := sc.NamespaceCollection[quota.Namespace]
	...
    
    // 清理quota.Status
	collection.Delete(quota)
}

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

注意

pcInformer缓存优先级相关信息，quotaInformer缓存资源配额相关状态

# 4.vcinformer

# 4.1.vchandler

vcInformer主要监听podgroup/queue资源事件变化，queue资源会缓存到cache.Queus，podgroup则补充job缓存或触发job回收。

// pgInformer
// AddPodGroupV1beta1 add podgroup to scheduler cache
func (sc *SchedulerCache) AddPodGroupV1beta1(obj interface{}) {
	ss, ok := obj.(*schedulingv1beta1.PodGroup)
	...

	scheme.Scheme.Convert(ss, &podgroup, nil)
    ...

	pg := &schedulingapi.PodGroup{PodGroup: podgroup, Version: schedulingapi.PodGroupVersionV1Beta1}
	...

    // 设置pgInfo
	sc.setPodGroup(pg)
    ...
}

// UpdatePodGroupV1beta1 add podgroup to scheduler cache
func (sc *SchedulerCache) UpdatePodGroupV1beta1(oldObj, newObj interface{}) {
	...

	if oldSS.ResourceVersion == newSS.ResourceVersion {
		return
	}

	podgroup := scheduling.PodGroup{}
	scheme.Scheme.Convert(newSS, &podgroup, nil)
    ...

	pg := &schedulingapi.PodGroup{PodGroup: podgroup, Version: schedulingapi.PodGroupVersionV1Beta1}
	...

	sc.setPodGroup(pg)
    ...
}

// Assumes that lock is already acquired.
func (sc *SchedulerCache) setPodGroup(ss *schedulingapi.PodGroup) error {
    // pgNamespace/pgName
	job := getJobID(ss)
	if _, found := sc.Jobs[job]; !found {
		sc.Jobs[job] = schedulingapi.NewJobInfo(job)
	}

    // 基于pgInfo初始化job信息
	sc.Jobs[job].SetPodGroup(ss)

	// set default queue in admission.
	if len(ss.Spec.Queue) == 0 {
		sc.Jobs[job].Queue = schedulingapi.QueueID(sc.defaultQueue)
	}

	return nil
}

// DeletePodGroupV1beta1 delete podgroup from scheduler cache
func (sc *SchedulerCache) DeletePodGroupV1beta1(obj interface{}) {
	...

	jobID := schedulingapi.JobID(fmt.Sprintf("%s/%s", ss.Namespace, ss.Name))
	...

	sc.deletePodGroup(jobID)
    ...
}

// Assumes that lock is already acquired.
func (sc *SchedulerCache) deletePodGroup(id schedulingapi.JobID) error {
	job, found := sc.Jobs[id]
	...

	// Unset SchedulingSpec
	job.PodGroup = nil

    // 入队删除关联job
	sc.DeletedJobs.Add(job)

	return nil
}

// queueInformer
// AddQueueV1beta1 add queue to scheduler cache
func (sc *SchedulerCache) AddQueueV1beta1(obj interface{}) {
	...

	scheme.Scheme.Convert(ss, queue, nil)
    ...

    // 注册到Queues
	sc.addQueue(queue)
}

// UpdateQueueV1beta1 update queue to scheduler cache
func (sc *SchedulerCache) UpdateQueueV1beta1(oldObj, newObj interface{}) {
	...

	if oldSS.ResourceVersion == newSS.ResourceVersion {
		return
	}
	...
    
	scheme.Scheme.Convert(newSS, newQueue, nil)
    ...
    
    // 注册到Queues
	sc.addQueue(newQueue)
}

// DeleteQueueV1beta1 delete queue from the scheduler cache
func (sc *SchedulerCache) DeleteQueueV1beta1(obj interface{}) {
	...
    
    // 清理Queues注册
	sc.deleteQueue(schedulingapi.QueueID(ss.Name))
}

注意

queue监听仅更新缓存，相对简单，podgroup则需关联及构造jobInfo缓存

# 4.2.numahandler

numaInformer监听的是Numatopology资源变化，基于Numatopology资源会向node注册不同资源对应的numa node拓扑数据。

// AddNumaInfoV1alpha1 add numa information to scheduler cache
func (sc *SchedulerCache) AddNumaInfoV1alpha1(obj interface{}) {
	...

	sc.addNumaInfo(ss)
}

// UpdateNumaInfoV1alpha1 update numa information to scheduler cache
func (sc *SchedulerCache) UpdateNumaInfoV1alpha1(oldObj, newObj interface{}) {
	...
    
	sc.addNumaInfo(ss)
}

// Assumes that lock is already acquired.
func (sc *SchedulerCache) addNumaInfo(info *nodeinfov1alpha1.Numatopology) error {
	if sc.Nodes[info.Name] == nil {
		sc.Nodes[info.Name] = schedulingapi.NewNodeInfo(nil)
		sc.Nodes[info.Name].Name = info.Name
	}

	if sc.Nodes[info.Name].NumaInfo == nil {
        // 设置numa node信息
		sc.Nodes[info.Name].NumaInfo = getNumaInfo(info)
		sc.Nodes[info.Name].NumaChgFlag = schedulingapi.NumaInfoMoreFlag
	} else {
		newLocalInfo := getNumaInfo(info)
        // 资源扩展,更新为MoreFlag
		if sc.Nodes[info.Name].NumaInfo.Compare(newLocalInfo) {
			sc.Nodes[info.Name].NumaChgFlag = schedulingapi.NumaInfoMoreFlag
        // 资源收紧,更新为LessFlag
		} else {
			sc.Nodes[info.Name].NumaChgFlag = schedulingapi.NumaInfoLessFlag
		}

        // 更新numaInfo
		sc.Nodes[info.Name].NumaInfo = newLocalInfo
	}
	...
    
	return nil
}

// DeleteNumaInfoV1alpha1 delete numa information from scheduler cache
func (sc *SchedulerCache) DeleteNumaInfoV1alpha1(obj interface{}) {
	...

	sc.deleteNumaInfo(ss)
}

// Assumes that lock is already acquired.
func (sc *SchedulerCache) deleteNumaInfo(info *nodeinfov1alpha1.Numatopology) {
	if sc.Nodes[info.Name] != nil {
		sc.Nodes[info.Name].NumaInfo = nil
		sc.Nodes[info.Name].NumaChgFlag = schedulingapi.NumaInfoResetFlag
	}
}

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

注意

numa node缓存的资源拓扑数据会影响Pod申请资源及调度