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

# 1.简介

# 1.1.initialize

job是volcano的核心资源对象，Initialize()会监听多种资源对象，触发关联的job入队执行协调处理，主要基于job创建回收pod资源。

// Initialize creates the new Job controller.
func (cc *jobcontroller) Initialize(opt *framework.ControllerOption) error {
	...
    
	cc.commandQueue = workqueue.NewTypedRateLimitingQueue(workqueue.DefaultTypedControllerRateLimiter[any]())
	cc.cache = jobcache.New()
	cc.errTasks = newRateLimitingQueue()
	...

	// workLoad模式
	if utilfeature.DefaultFeatureGate.Enabled(features.WorkLoadSupport) {
        // job缓存
		cc.jobInformer = factory.Batch().V1alpha1().Jobs()
		cc.jobInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
			AddFunc:    cc.addJob,
			UpdateFunc: cc.updateJob,
			DeleteFunc: cc.deleteJob,
		})
		cc.jobLister = cc.jobInformer.Lister()
		...
	}

    // 启用command同步
	if utilfeature.DefaultFeatureGate.Enabled(features.QueueCommandSync) {
		cc.cmdInformer = factory.Bus().V1alpha1().Commands()
		cc.cmdInformer.Informer().AddEventHandler(
			cache.FilteringResourceEventHandler{
				FilterFunc: func(obj interface{}) bool {
					switch v := obj.(type) {
                    // job所属command
					case *busv1alpha1.Command:
						if v.TargetObject != nil && v.TargetObject.APIVersion == SchemeGroupVersion.String() &&
							v.TargetObject.Kind == "Job" {
							return true
						}

						return false
					default:
						return false
					}
				},
				Handler: cache.ResourceEventHandlerFuncs{
					AddFunc: cc.addCommand,
				},
			},
		)
		cc.cmdLister = cc.cmdInformer.Lister()
		...
	}

    // pod缓存
	cc.podInformer = sharedInformers.Core().V1().Pods()
	cc.podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc:    cc.addPod,
		UpdateFunc: cc.updatePod,
		DeleteFunc: cc.deletePod,
	})
	cc.podLister = cc.podInformer.Lister()
	...

    // pvc缓存
	cc.pvcInformer = sharedInformers.Core().V1().PersistentVolumeClaims()
	cc.pvcLister = cc.pvcInformer.Lister()
	...

    // svc缓存
	cc.svcInformer = sharedInformers.Core().V1().Services()
	cc.svcLister = cc.svcInformer.Lister()
	...

    // pg缓存
	cc.pgInformer = factory.Scheduling().V1beta1().PodGroups()
	cc.pgInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		UpdateFunc: cc.updatePodGroup,
	})
	cc.pgLister = cc.pgInformer.Lister()
	...

    // 启用PriorityClass
	if utilfeature.DefaultFeatureGate.Enabled(features.PriorityClass) {
        // priorityClass缓存
		cc.pcInformer = sharedInformers.Scheduling().V1().PriorityClasses()
		cc.pcLister = cc.pcInformer.Lister()
		...
	}

    // queue缓存
	cc.queueInformer = factory.Scheduling().V1beta1().Queues()
	cc.queueLister = cc.queueInformer.Lister()
    ...

	// handler
	state.SyncJob = cc.syncJob
	state.KillJob = cc.killJob
	state.KillTarget = cc.killTarget
    
	return nil
}

注意

Initialize会监听job/command/pod/pvc/svc/podGroup/priorityClass/queue资源，触发关联的job入队

# 1.2.jobInformer

jobInformer监听job的增删改事件和command更新事件，相关job对象会注册到cache，根据hash(namespace/name)取模未知放入队列。

func (cc *jobcontroller) addJob(obj interface{}) {
	job, ok := obj.(*batch.Job)
	...

	req := apis.Request{
		Namespace: 	job.Namespace,
		JobName:   	job.Name,
		Event: 		bus.OutOfSyncEvent,
	}

	// cache job
	cc.cache.Add(job)
    ...
    
	key := jobhelpers.GetJobKeyByReq(&req)
    // acquire queue with hash
	queue := cc.getWorkerQueue(key)
	queue.Add(req)
}

func (cc *jobcontroller) updateJob(oldObj, newObj interface{}) {
	newJob, ok := newObj.(*batch.Job)
	...

	oldJob, ok := oldObj.(*batch.Job)
	...

	// No need to update if ResourceVersion is not changed
	if newJob.ResourceVersion == oldJob.ResourceVersion {
		return
	}

	cc.cache.Update(newJob)
    ...

	// no change
	if equality.DeepEqual(newJob.Spec, oldJob.Spec) && newJob.Status.State.Phase == oldJob.Status.State.Phase {
		return
	}

	req := apis.Request{
		Namespace: newJob.Namespace,
		JobName:   newJob.Name,
		Event:     bus.OutOfSyncEvent,
	}
    
	key := jobhelpers.GetJobKeyByReq(&req)
	queue := cc.getWorkerQueue(key)
	queue.Add(req)
}

func (cc *jobcontroller) deleteJob(obj interface{}) {
	job, ok := obj.(*batch.Job)
	...

	cc.cache.Delete(job)
    ...
}

func (cc *jobcontroller) addCommand(obj interface{}) {
	cmd, ok := obj.(*bus.Command)
	if !ok {
		return
	}

	cc.commandQueue.Add(cmd)
}

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

注意

command对象主要用于触发job入队，完成一次同步就会删除

# 1.3.podInformer

podInformer会监听pod资源的ADD/Update/Delete事件，controlledByJob的pod会获取关联job，向cache注册pod及入队job。

func (cc *jobcontroller) addPod(obj interface{}) {
	pod, ok := obj.(*v1.Pod)
	...

	// Filter out pods that are not created from volcano job
	if !isControlledBy(pod, helpers.JobKind) {
		return
	} else {
		jobUid = metav1.GetControllerOf(pod).UID
	}

	jobName, found := pod.Annotations[batch.JobNameKey]
	...
	version, found := pod.Annotations[batch.JobVersion]
	...
	dVersion, err := strconv.Atoi(version)
	...

	if pod.DeletionTimestamp != nil {
		cc.deletePod(pod)
		return
	}

	req := apis.Request{
		Namespace: pod.Namespace,
		JobName:   jobName,
		JobUid:    jobUid,
		PodName:   pod.Name,
		PodUID:    pod.UID,
		Event:      bus.PodPendingEvent,
		JobVersion: int32(dVersion),
	}

	cc.cache.AddPod(pod)
    ...
    
	key := jobhelpers.GetJobKeyByReq(&req)
	queue := cc.getWorkerQueue(key)
	queue.Add(req)
}

func (cc *jobcontroller) updatePod(oldObj, newObj interface{}) {
	oldPod, ok := oldObj.(*v1.Pod)
	...

	newPod, ok := newObj.(*v1.Pod)
	...

	// Filter out pods that are not created from volcano job
	if !isControlledBy(newPod, helpers.JobKind) {
		return
	} else {
		jobUid = metav1.GetControllerOf(newPod).UID
	}

	if newPod.ResourceVersion == oldPod.ResourceVersion {
		return
	}

	if newPod.DeletionTimestamp != nil {
		cc.deletePod(newObj)
		return
	}

	taskName, found := newPod.Annotations[batch.TaskSpecKey]
	...
	jobName, found := newPod.Annotations[batch.JobNameKey]
	...
	version, found := newPod.Annotations[batch.JobVersion]
	...
	dVersion, err := strconv.Atoi(version)
	...
	cc.cache.UpdatePod(newPod)
    ...

	switch newPod.Status.Phase {
	case v1.PodFailed:
        // pod nonfailed-->failed
		if oldPod.Status.Phase != v1.PodFailed {
			event = bus.PodFailedEvent
			if len(newPod.Status.ContainerStatuses) > 0 && newPod.Status.ContainerStatuses[0].State.Terminated {
				exitCode = newPod.Status.ContainerStatuses[0].State.Terminated.ExitCode
			}
		}
	case v1.PodSucceeded:
        // pod nonsucceed-->failed&task completed
		if oldPod.Status.Phase != PodSucceeded && cc.cache.TaskCompleted(newPod.Namespace, jobName, taskName) {
			event = bus.TaskCompletedEvent
		}
	case v1.PodRunning:
        // task failed
		if cc.cache.TaskFailed(jobcache.JobKeyByName(newPod.Namespace, jobName), taskName) {
			event = bus.TaskFailedEvent
		}
        
        // pod nonrunning-->running
		if oldPod.Status.Phase != v1.PodRunning {
			event = bus.PodRunningEvent
		}
	case v1.PodPending:
        // task failed
		if cc.cache.TaskFailed(jobcache.JobKeyByName(newPod.Namespace, jobName), taskName) {
			event = bus.TaskFailedEvent
		}
        
        // pod nonpending-->pending
		if oldPod.Status.Phase != v1.PodPending {
			event = bus.PodPendingEvent
		}
	}

	req := apis.Request{
		Namespace: newPod.Namespace,
		JobName:   jobName,
		JobUid:    jobUid,
		TaskName:  taskName,
		PodName:   newPod.Name,
		PodUID:    newPod.UID,
		Event:      event,
		ExitCode:   exitCode,
		JobVersion: int32(dVersion),
	}

	key := jobhelpers.GetJobKeyByReq(&req)
	queue := cc.getWorkerQueue(key)
	queue.Add(req)
}

func (cc *jobcontroller) deletePod(obj interface{}) {
	pod, ok := obj.(*v1.Pod)
	...

	// Filter out pods that are not created from volcano job
	if !isControlledBy(pod, helpers.JobKind) {
		return
	} else {
		jobUid = metav1.GetControllerOf(pod).UID
	}

	taskName, found := pod.Annotations[batch.TaskSpecKey]
	...
	jobName, found := pod.Annotations[batch.JobNameKey]
	...
	version, found := pod.Annotations[batch.JobVersion]
	...
	dVersion, err := strconv.Atoi(version)
	...

	req := apis.Request{
		Namespace: pod.Namespace,
		JobName:   jobName,
		JobUid:    jobUid,
		TaskName:  taskName,
		PodName:   pod.Name,
		PodUID:    pod.UID,
		Event:      bus.PodEvictedEvent,
		JobVersion: int32(dVersion),
	}

	cc.cache.DeletePod(pod)
    ...

	key := jobhelpers.GetJobKeyByReq(&req)
	queue := cc.getWorkerQueue(key)
	queue.Add(req)
}

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
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166

注意

ADD/Update Pod会注册到cache缓存，关联的Job会推入queue处理，Delete Pod会由cache删除

# 1.4.pgInformer

pgInformer相对简单，监听podgroup的更新事件，由pgannotation获取关联job，基于hash分流将job推入所属的queue处理。

func (cc *jobcontroller) updatePodGroup(oldObj, newObj interface{}) {
	oldPG, ok := oldObj.(*scheduling.PodGroup)
	...

	newPG, ok := newObj.(*scheduling.PodGroup)
	...

	for _, or := range newPG.OwnerReferences {
		if or.Kind == "Job" { jobNameKey = or.Name }
	}

    // 检查缓存job
	_, err := cc.cache.Get(jobcache.JobKeyByName(newPG.Namespace, jobNameKey))
	if err != nil && newPG.Annotations != nil {
		klog.Warningf("Failed to find job in cache, this may not be a PodGroup for volcano job.")
	}

    // pg状态变更
	if newPG.Status.Phase != oldPG.Status.Phase {
		req := apis.Request{
			Namespace: newPG.Namespace,
			JobName:   jobNameKey,
		}
        
		switch newPG.Status.Phase {
		case scheduling.PodGroupUnknown:
			req.Event = bus.JobUnknownEvent
		}
        
		key := jobhelpers.GetJobKeyByReq(&req)
		queue := cc.getWorkerQueue(key)
		queue.Add(req)
	}
}

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

注意

podgroup会和Job基于Annotation关联

# 1.5.jcstart

controller.Run()会启动一定数量的worker消费queue队列数据，根据job不同状态初始化stateExecutor执行器实现不同的状态调整。

// Run start JobController.
func (cc *jobcontroller) Run(stopCh <-chan struct{}) {
	cc.informerFactory.Start(stopCh)
	cc.vcInformerFactory.Start(stopCh)

	for informerType, ok := range cc.informerFactory.WaitForCacheSync(stopCh) {
		if !ok {
			return
		}
	}

	for informerType, ok := range cc.vcInformerFactory.WaitForCacheSync(stopCh) {
		if !ok {
			return
		}
	}

    // command处理
	go wait.Until(cc.handleCommands, 0, stopCh)
	...
    
    // 激活一定数量worker
	for i = 0; i < cc.workers; i++ {
		go func(num uint32) {
            // 间隔1s触发一次
			wait.Until(
				func() {
					cc.worker(num)
				},
				time.Second,
				stopCh)
		}(i)
	}

    // 同步缓存
	go cc.cache.Run(stopCh)

	// Re-sync error tasks.
	go wait.Until(cc.processResyncTask, 0, stopCh)
}

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

注意

worker激活的数量和workqueue数量保持一致，不同worker固定消费自己所属的workqueue

# 2.handler

# 2.1.handlecmd

cc.handleCommands()会获取cmdQueue队列数据，将cmd对象关联的job入队，完成一次协调会删除command对象避免重复处理。

func (cc *jobcontroller) handleCommands() {
	for cc.processNextCommand() {
	}
}

func (cc *jobcontroller) processNextCommand() bool {
	obj, shutdown := cc.commandQueue.Get()
	...
    
	cmd := obj.(*bus.Command)
	defer cc.commandQueue.Done(cmd)

    // command是一次性对象,用完后就删除
	cc.vcClient.BusV1alpha1().Commands(cmd.Namespace).Delete(context.TODO(), cmd.Name, metav1.DeleteOptions{})
	...
    
	req := apis.Request{
		Namespace: cmd.Namespace,
		JobName:   cmd.TargetObject.Name,
		Event:     bus.CommandIssuedEvent,
		Action:    bus.Action(cmd.Action),
	}

	key := jobhelpers.GetJobKeyByReq(&req)
	queue := cc.getWorkerQueue(key)
    
    // add job with hash queue
	queue.Add(req)

	return true
}

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

注意

command对象主要用于唤醒job，基于关联job入队触发重处理

# 2.2.handleCache

cc.cache.Run()会执行worker协调，消费cache.Delete()向deletedJobs推送的job，检查job无关联pod会清理维护的缓存。

func (jc *jobCache) Run(stopCh <-chan struct{}) {
	wait.Until(jc.worker, 0, stopCh)
}

func (jc *jobCache) worker() {
	for jc.processCleanupJob() {
	}
}

func (jc *jobCache) processCleanupJob() bool {
    // cache.Delete推送的job
	job, shutdown := jc.deletedJobs.Get()
	...
	defer jc.deletedJobs.Done(job)
	...

    // 无关联pod
	if jobTerminated(job) {
		jc.deletedJobs.Forget(job)
		key := keyFn(job.Namespace, job.Name)
		delete(jc.jobs, key)
	} else {
		// Retry
		jc.retryDeleteJob(job)
	}
	return true
}

func (jc *jobCache) retryDeleteJob(job *apis.JobInfo) {
	jc.deletedJobs.AddRateLimited(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

注意

cache会缓存job和pod作为缓存查询，informer回调及job协调处理的delete job都会触发cache清理

# 2.3.processTask

cc.processResyncTask()负责处理errTasks队列任务，重试的Pod未找到会清理cache缓存的pod/job，存在会更新缓存Pod。

func (cc *jobcontroller) processResyncTask() {
	obj, shutdown := cc.errTasks.Get()
	...

	// 周期达到10次重置状态
	if cc.errTasks.NumRequeues(obj) > 10 {
		cc.errTasks.Forget(obj)
		return
	}

	defer cc.errTasks.Done(obj)

    // 断言为pod
	task, ok := obj.(*v1.Pod)
	...

    // task同步
	if err := cc.syncTask(task); err != nil {
		cc.resyncTask(task)
	}
}

func (cc *jobcontroller) syncTask(oldTask *v1.Pod) error {
	newPod, err := cc.kubeClient.CoreV1().Pods(oldTask.Namespace).Get(context.TODO(), oldTask.Name, ...)
	...
    
    // pod未找到
    if errors.IsNotFound(err) {
        cc.cache.DeletePod(oldTask)
        ...
        return nil
    }
	...

    // 更新job.task.pods
	return cc.cache.UpdatePod(newPod)
}

func (jc *jobCache) DeletePod(pod *v1.Pod) error {
	...

    // 由pod annotation获取job
	key, err := jobKeyOfPod(pod)
	...

	job, found := jc.jobs[key]
	if !found {
		job = &apis.JobInfo{
			Pods: make(map[string]map[string]*v1.Pod),
		}
		jc.jobs[key] = job
	}

    // 清理job.task.pods
	job.DeletePod(pod)
    ...

    // job未关联pod,推到deleteJobs由cache处理
	if jobTerminated(job) {
		jc.deleteJob(job)
	}

	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

注意

cache.DeletePod()会尝试清理关联的job.task.pods缓存，job未关联pod推到cache.deleteJobs由cache后台清理缓存

# 3.processer

# 3.1.jobworker

cc.worker()会根据hash(namespace/name)获取worker queue，消费queue item及生成状态执行器，根据不同state.Execute处理。

func (cc *jobcontroller) worker(i uint32) {
	for cc.processNextReq(i) {
	}
}

func (cc *jobcontroller) processNextReq(count uint32) bool {
    // worker queue
	queue := cc.queueList[count]
	obj, shutdown := queue.Get()
	...

	req := obj.(apis.Request)
	defer queue.Done(req)

    // namespace/name
	key := jobcache.JobKeyByReq(&req)
    // 归属检查
	if !cc.belongsToThisRoutine(key, count) {
        // key不属于当前worker queue,重新放回队列
		queueLocal := cc.getWorkerQueue(key)
		queueLocal.Add(req)
		return true
	}

    // clean delay action
	cc.CleanPodDelayActionsIfNeed(req)

    // 获取cache job
	jobInfo, err := cc.cache.Get(key)
	...
    
    // 生成执行器
    // pendingState/runningState/restartingState/finishedState/terminatingState
    // abortingState/abortedState/completingState
	st := state.NewState(jobInfo)
	...

    // 应用job策略
	delayAct := applyPolicies(jobInfo.Job, &req)

    // 延迟入队
	if delayAct.delay != 0 {
		cc.AddDelayActionForJob(req, delayAct)
		return true
	}
	...
    
    // 获取action
	action := GetStateAction(delayAct)

    // 加载执行器
	if err := st.Execute(action); err != nil {
		cc.handleJobError(queue, req, st, err, delayAct.action)
		return true
	}

	// If no error, forget it.
	queue.Forget(req)

    // 内部行为:SyncJobAction/EnqueueAction/SyncQueueAction/OpenQueueAction/CloseQueueAction
	if !isInternalAction(delayAct.action) {
        // clean delayed actions
		cc.cleanupDelayActions(delayAct)
	}

	return true
}

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

注意

volcano为提高处理性能，会抽象出一定数量队列和worker，不同worker固定消费同索引队列

# 3.2.delayAction

cc.cleanupDelayActions()主要负责取消pod/task delay action注册，清理正在处理或已执行过的pod delay action，避免重复执行。

// clean delayed actions for Pod events when the pod phase changed.
func (cc *jobcontroller) CleanPodDelayActionsIfNeed(req apis.Request) {
	// Skip cleaning delayed actions for non-pod events
	if !cc.isPodEvent(req) {
		return
	}

	if req.Event != busv1alpha1.PodPendingEvent {
		...
        
        // 注册过delay action
		if taskMap, exists := cc.delayActionMap[jobKey]; exists {
            // 注册过pod task
			if delayAct, exists := taskMap[req.PodName]; exists {
				shouldCancel := false
				// pending event
				if delayAct.event == busv1alpha1.PodPendingEvent {
					// 正在处理的pod取消delay action
					if req.PodUID == delayAct.podUID {
						shouldCancel = true
					}
				}

                // failed/evict event&req running event
				if (delayAct.event == PodFailedEvent/PodEvictedEvent) &&req.Event == PodRunningEvent {
					shouldCancel = true
				}

                // 取消pod task注册
				if shouldCancel {
					delayAct.cancel()
					delete(taskMap, req.PodName)
				}
			}
		}
	}
}

// cleans up delayed actions After delayed action is executed.
func (cc *jobcontroller) cleanupDelayActions(currentDelayAction *delayAction) {
	...

    // JobAction/TaskAction/PodAction
	actionType := GetActionType(currentDelayAction.action)

    // 注册过delay action
	if m, exists := cc.delayActionMap[currentDelayAction.jobKey]; exists {
		for _, delayAct := range m {
            // action类型匹配
			if GetActionType(delayAct.action) == actionType {
				// task action,taskName不一致
				if actionType == TaskAction && delayAct.taskName != currentDelayAction.taskName {
					continue
				}
                // pod action,podName不一致(为什么不用uid)
				if actionType == PodAction && delayAct.podName != currentDelayAction.podName {
					continue
				}

                // 取消延迟注册
				if delayAct.cancel != nil {
					delayAct.cancel()
				}
                
				delete(m, delayAct.podName)
			}
		}
	}
}

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

注意

cleanDelayAction主要负责清理pod/task注册的delay action缓存及停止timer

# 3.3.applyPolicy

applyPolicy()检查req准入条件，尝试匹配job.spec.tasks的policy设置action/delay，未匹配则基于job.spec.policies设置。

func applyPolicies(job *batch.Job, req *apis.Request) (delayAct *delayAction) {
	delayAct = &delayAction{
		jobKey:   jobcache.JobKeyByReq(req),
		event:    req.Event,
		taskName: req.TaskName,
		podName:  req.PodName,
		podUID:   req.PodUID,
		// default action is sync job
		action: v1alpha1.SyncJobAction,
	}

	if len(req.Action) != 0 {
		delayAct.action = req.Action
		return
	}

	// 内部事件:OutOfSyncEvent/CommandIssuedEvent/PodRunningEvent
	if isInternalEvent(req.Event) {
		return
	}

	// job uid匹配,避免同名不同对象
	if len(req.JobUid) != 0 && job != nil && req.JobUid != job.UID {
		return
	}

	// req请求过期
	if req.JobVersion < job.Status.Version {
		return
	}

	// task level policies
	if len(req.TaskName) != 0 {
		for _, task := range job.Spec.Tasks {
			if task.Name == req.TaskName {
				for _, policy := range task.Policies {
					policyEvents := getEventlist(policy)
					// event筛选
					if len(policyEvents) > 0 && len(req.Event) > 0 {
                        // 配置了req event或配置了anyEvent
						if checkEventExist(policyEvents, req.Event) || checkEventExist(policyEvents, AnyEvent) {
							// 未强制timeout/配置了timeout
							if !shouldConfigureTimeout(req.Event) || policy.Timeout != nil {
								delayAct.action = policy.Action
								if policy.Timeout != nil {
									delayAct.delay = policy.Timeout.Duration
								}
								return
							}
						}
					}

					// 0 is not an error code, is prevented in validation admission controller
					if policy.ExitCode != nil && *policy.ExitCode == req.ExitCode {
						delayAct.action = policy.Action
						if policy.Timeout != nil {
							delayAct.delay = policy.Timeout.Duration
						}
						return
					}
				}
				break
			}
		}
	}

	// Job level policies
	for _, policy := range job.Spec.Policies {
		policyEvents := getEventlist(policy)

		if len(policyEvents) > 0 && len(req.Event) > 0 {
            // 配置了req event或anyEvent
			if checkEventExist(policyEvents, req.Event) || checkEventExist(policyEvents, v1alpha1.AnyEvent) {
                // 不强制timeout/配置timeout
				if !(shouldConfigureTimeout(req.Event) && policy.Timeout == nil) {
					delayAct.action = policy.Action
					if policy.Timeout != nil {
						delayAct.delay = policy.Timeout.Duration
					}
					return
				}
			}
		}

		// 0 is not an error code, is prevented in validation admission controller
		if policy.ExitCode != nil && *policy.ExitCode == req.ExitCode {
			delayAct.action = policy.Action
			if policy.Timeout != nil {
				delayAct.delay = policy.Timeout.Duration
			}
			return
		}
	}

	return
}

注意

applyPolicy()会基于task粒度和job粒度设置delay/action

# 3.4.delayJob

cc.AddDelayActionForJob()会基于delayAction注册延迟任务，延迟delay间隔执行state.Execute()完成关联job协调处理。

func (cc *jobcontroller) AddDelayActionForJob(req apis.Request, delayAct *delayAction) {
	...

	m, ok := cc.delayActionMap[delayAct.jobKey]
	if !ok {
		m = make(map[string]*delayAction)
		cc.delayActionMap[delayAct.jobKey] = m
	}
    
    // delay action无变化
	if oldDelayAct, exists := m[req.PodName]; exists && oldDelayAct.action == delayAct.action {
		return
	}
    
	m[req.PodName] = delayAct

	ctx, cancel := context.WithTimeout(context.Background(), delayAct.delay)
	delayAct.cancel = cancel

	go func() {
        // delay间隔超时
		<-ctx.Done()
        
        // cancel delay
		if ctx.Err() == context.Canceled {
			return
		}

        // 获取cache job
		jobInfo, err := cc.cache.Get(delayAct.jobKey)
		...

        // state执行器
		st := state.NewState(jobInfo)
		if st == nil {
			return
		}
        
        // 获取hashQueue
		queue := cc.getWorkerQueue(delayAct.jobKey)

        // 执行executor
		if err := st.Execute(GetStateAction(delayAct)); err != nil {
			cc.handleJobError(queue, req, st, err, delayAct.action)
		}

		queue.Forget(req)

        // 清理过期的delay任务
		cc.cleanupDelayActions(delayAct)
	}()
}

func (cc *jobcontroller) handleJobError(...) {
    // 入队重试
	if cc.maxRequeueNum == -1 || queue.NumRequeues(req) < cc.maxRequeueNum {
		queue.AddRateLimited(req)
		return
	}

    // 无法入队,再执行state.Execute
	st.Execute(state.Action{Action: busv1alpha1.TerminateJobAction})
    ...
}

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

注意

AddDelayActionForJob会基于delayAction注册延迟任务，间隔delay时间触发state.Execute，执行后取消delay action注册