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

# 1.简介

# 1.1.定义

job是kubernetes管理离线任务的重要负载资源，用于直接管理Pod，驱动一个或多个Pod运行到终态，是实现批处理最简单的方式。

// Controller ensures that all Job objects have corresponding pods to run their configured workload.
type Controller struct {
	..
    
	podControl controller.PodControlInterface
	...
    
	syncHandler         func(ctx context.Context, jobKey string) error	// 同步控制
	...

    expectations controller.ControllerExpectationsInterface				// 期望创建或删除的Pod(TTLCache)

	finalizerExpectations *uidTrackingExpectations						// 待摘除finalizer的job pod uid

	jobLister batchv1listers.JobLister									// job informer lister

	podStore corelisters.PodLister										// pod informer lister

	queue workqueue.RateLimitingInterface								// job queue

	orphanQueue workqueue.RateLimitingInterface							// orphan pod queue
	...

	podUpdateBatchPeriod time.Duration

	clock clock.WithTicker

	backoffRecordStore *backoffStore
}

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

注意

job遵循控制器设计模式，由informer回调将待处理job推入workqueue，供syncHandler消费处理

# 1.2.原理

batchjob controller会监听informer相关资源，将关联job推入queue供syncHandler消费处理，循环流程与cronjob类似。
补充

batchjob仅执行一次，周期任务一般由cronjob管理batchjob实现定时执行

# 2.分析

# 2.1.start

startJobController()会初始化job controller实例，执行jm.Run()进行informer同步及循环消费queue，供syncHandler处理。

// NewController creates a new Job controller that keeps the relevant pods in sync with their corresponding Job.
func NewController(...) *Controller {
	return newControllerWithClock(podInformer, jobInformer, kubeClient, &clock.RealClock{})
}

func newControllerWithClock(...) *Controller {
	...

	jm := &Controller{
		...
        
        // 状态及finalizer跟踪缓存
		expectations:          controller.NewControllerExpectations(),
		finalizerExpectations: newUIDTrackingExpectations(),
		
        // workqueue相关
        queue:                 workqueue.NewRateLimitingQueueWithDelayingInterface(...),
		orphanQueue:           workqueue.NewRateLimitingQueueWithDelayingInterface(...),
		...
        
		clock:                 clock,
		backoffRecordStore:    newBackoffRecordStore(),
	}
    
    // Pod批更新开启
	if feature.DefaultFeatureGate.Enabled(features.JobReadyPods) {
        // 设置批更新周期
		jm.podUpdateBatchPeriod = podUpdateBatchPeriod
	}

    // job informer回调
	jobInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: func(obj interface{}) {
			jm.enqueueController(obj, true)
		},
		UpdateFunc: jm.updateJob,
		DeleteFunc: jm.deleteJob,
	})
	jm.jobLister = jobInformer.Lister()
	...

    // pod informer回调
	podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc:    jm.addPod,
		UpdateFunc: jm.updatePod,
		DeleteFunc: func(obj interface{}) {
			jm.deletePod(obj, true)
		},
	})
    
	jm.podStore = podInformer.Lister()
	...

    // 同步回调
	jm.updateStatusHandler = jm.updateJobStatus
	jm.patchJobHandler = jm.patchJob
	jm.syncHandler = jm.syncJob
	...

	return jm
}

func startJobController(...) (controller.Interface, bool, error) {
	go job.NewController(
		controllerContext.InformerFactory.Core().V1().Pods(),
		controllerContext.InformerFactory.Batch().V1().Jobs(),
		controllerContext.ClientBuilder.ClientOrDie("job-controller"),
	).Run(ctx, int(controllerContext.ComponentConfig.JobController.ConcurrentJobSyncs))
	return nil, true, 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

补充

job/pod创建、更新及删除事件均会将相关job入队，类似cronjob流程，也是基于label/owner筛选

# 2.2.jobInformer

jobInformer监听batchjob资源变化，将相关job立即或延迟推入queue，job删除会触发Pod弃养，将own pod推入orphan queue。

func (jm *Controller) updateJob(old, cur interface{}) {
	...

	// name/namespace
	key, err := controller.KeyFunc(curJob)
	...
    
    // 推入queue
	jm.enqueueController(curJob, true)
    
	// job已开始执行
	if curJob.Status.StartTime != nil {
        // 获取curjob ads(job存活时间)
		curADS := curJob.Spec.ActiveDeadlineSeconds
		if curADS == nil {
			return
		}
        // 获取oldjob ads
		oldADS := oldJob.Spec.ActiveDeadlineSeconds
        
        // ads变化
		if oldADS == nil || *oldADS != *curADS {
            // 计算job运行时间
			passed := jm.clock.Since(curJob.Status.StartTime.Time)
            // 存活时间
			total := time.Duration(*curADS) * time.Second
			// AddAfter will handle total < passed
			jm.queue.AddAfter(key, total-passed)
		}
	}
}

// deleteJob enqueues the job and all the pods associated with it that still have a finalizer.
func (jm *Controller) deleteJob(obj interface{}) {
    // job入队
	jm.enqueueController(obj, true)
	jobObj, ok := obj.(*batch.Job)
	...
    
	// label selector
	selector, err := metav1.LabelSelectorAsSelector(jobObj.Spec.Selector)
	...
    
    // 获取匹配Pod
	pods, _ := jm.podStore.Pods(jobObj.Namespace).List(selector)
	for _, pod := range pods {
        // owner&// Pod.finalizer标记tracing
		if metav1.IsControlledBy(pod, jobObj) && hasJobTrackingFinalizer(pod) {
            // pod推入orphan queue
			jm.enqueueOrphanPod(pod)
		}
	}
}

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

注意

jobInformer删除会将job推入workqueue，还会将标记tracking finalizer的owner pod推入orphan queue

# 2.3.podInformer

podInformer监听pod资源变化，将相关job立即或延迟推入queue，pod ownerRef调整会触发弃养，将孤儿pod推入orphan queue。

// When a pod is created, enqueue the controller that manages it and update its expectations.
func (jm *Controller) addPod(obj interface{}) {
	pod := obj.(*v1.Pod)
	...
    
    // informer刚启动加载的删除Pod
	if pod.DeletionTimestamp != nil {
		jm.deletePod(pod, false)
		return
	}

	// has ownerRef
	if controllerRef := metav1.GetControllerOf(pod); controllerRef != nil {
        // 解析job
		job := jm.resolveControllerRef(pod.Namespace, controllerRef)
		if job == nil {
			return
		}
		...
        
        // 创建job exp对象,add=1
		jm.expectations.CreationObserved(jobKey)
        // job入队
		jm.enqueueControllerPodUpdate(job, true)
		return
	}

	// Pod.finalizer标记tracing
	if hasJobTrackingFinalizer(pod) {
        // orphan pod入队
		jm.enqueueOrphanPod(pod)
	}
    
	// 基于label selector匹配job入队
	for _, job := range jm.getPodJobs(pod) {
		jm.enqueueControllerPodUpdate(job, true)
	}
}

// When a pod is updated, figure out what job/s manage it and wake them up.
func (jm *Controller) updatePod(old, cur interface{}) {
	...
    
    // pod reversion无变化
	if curPod.ResourceVersion == oldPod.ResourceVersion {
		return
	}
    
    // curPod正在删除
	if curPod.DeletionTimestamp != nil {
		jm.deletePod(curPod, false)
		return
	}

	// Pod第一次转为failed状态,延迟入队
	immediate := !(curPod.Status.Phase == v1.PodFailed && oldPod.Status.Phase != v1.PodFailed)
    ...
    
    // ownerRef变化,oldPod ownerRef不为空
	if controllerRefChanged && oldControllerRef != nil {
		// 解析oldPod所属job
		if job := jm.resolveControllerRef(oldPod.Namespace, oldControllerRef); job != nil {
            // curPod.finalizer未标记tracking
			if finalizerRemoved {
				key, err := controller.KeyFunc(job)
				...
                // 清理finalizer跟踪的pod uid
				jm.finalizerExpectations.finalizerRemovalObserved(key, string(curPod.UID))
			}
            
            // oldPod所属job入队
			jm.enqueueControllerPodUpdate(job, immediate)
		}
	}

	// curPod有ownerRef
	if curControllerRef != nil {
        // 解析job
		job := jm.resolveControllerRef(curPod.Namespace, curControllerRef)
		if job == nil {
			return
		}
        
        // curPod.finalizer未标记tracking
		if finalizerRemoved {
            
			key, err := controller.KeyFunc(job)
			...
            
            // 清理finalizer跟踪的pod uid
            jm.finalizerExpectations.finalizerRemovalObserved(key, string(curPod.UID))
		}
        
        // curPod所属job入队
		jm.enqueueControllerPodUpdate(job, immediate)
		return
	}

	// curPod.finalizer标记tracing
	if hasJobTrackingFinalizer(curPod) {
        // curPod推入orphan queue 
		jm.enqueueOrphanPod(curPod)
	}
	...
    
    // label变化或ownerRef变化(有——>没有)
	if labelChanged || controllerRefChanged {
        // 基于curPod label匹配所有job推入queue
		for _, job := range jm.getPodJobs(curPod) {
			jm.enqueueControllerPodUpdate(job, immediate)
		}
	}
}

// When a pod is deleted, enqueue the job that manages the pod and update its expectations.
func (jm *Controller) deletePod(obj interface{}, final bool) {
	pod, ok := obj.(*v1.Pod)
	...
    
    // ownerRef为空
	if controllerRef == nil {
		// orphan Pod标记tracking finalizer
		if hasFinalizer {
            // 推入orphan queue
			jm.enqueueOrphanPod(pod)
		}
		return
	}
    
    // 解析owner job
	job := jm.resolveControllerRef(pod.Namespace, controllerRef)
    
    // job为空或已完成
	if job == nil || IsJobFinished(job) {
		// pod.finalizer标记tracking
		if hasFinalizer {
            // 推入orphan queue
			jm.enqueueOrphanPod(pod)
		}
		return
	}
    ...
    
    // job exp对象del-1
	jm.expectations.DeletionObserved(jobKey)

	// final开启或Pod.finalizer未标记tracking
	if final || !hasFinalizer {
        // finalizer跟踪清理pod uid
		jm.finalizerExpectations.finalizerRemovalObserved(jobKey, string(pod.UID))
	}

    // job推入queue
	jm.enqueueControllerPodUpdate(job, true)
}

注意

job一般情况下入队仅延迟1s，job由!failed->failed则会触发延迟入队，基于重试次数计算延迟时间，最大360s

# 2.4.runworker

jm.Run()会激活worker处理jobqueue的item项，执行syncHandler()同步job状态，激活orphanWorker回收孤儿Pod。

// Run the main goroutine responsible for watching and syncing jobs.
func (jm *Controller) Run(ctx context.Context, workers int) {
	...

	defer jm.queue.ShutDown()
	defer jm.orphanQueue.ShutDown()

	// informer同步检测
	if !cache.WaitForNamedCacheSync("job", ctx.Done(), jm.podStoreSynced, jm.jobStoreSynced) {
		return
	}

    // job worker
	for i := 0; i < workers; i++ {
		go wait.UntilWithContext(ctx, jm.worker, time.Second)
	}

    // orphan worker
	go wait.UntilWithContext(ctx, jm.orphanWorker, time.Second)

	<-ctx.Done()
}

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

注意

jm.Run()和之前的控制器不一样，会激活worker和orphanworker协程，分别处理job和orphan pod同步

# 3.worker

# 3.1.process

jm.worker()会循环获取job queue待处理item，调用jm.syncJob()创建或删除Pod，检测及维护Job的完成或执行中状态。

// worker runs a worker thread that just dequeues items, processes them, and marks them done.
func (jm *Controller) worker(ctx context.Context) {
	for jm.processNextWorkItem(ctx) {
	}
}

func (jm *Controller) processNextWorkItem(ctx context.Context) bool {
	key, quit := jm.queue.Get()
	...
    
	defer jm.queue.Done(key)

    // 执行同步
	err := jm.syncHandler(ctx, key.(string))
    
	if err == nil {
		jm.queue.Forget(key)
		return true
	}

	jm.queue.AddRateLimited(key)

	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

注意

worker是一种重要的设计方式，controller-manager基于都是上述处理模式

# 3.2.syncJob

jm.syncJob()负责调度Job，将Job的期望状态和观察到的Pod状态对齐，根据期望执行Pod的创建或删除，根据执行状态更新Job.status。

// syncJob will sync the job with the given key if it has had its expectations fulfilled.
func (jm *Controller) syncJob(ctx context.Context, key string) (rErr error) {
	...

    // 获取job
	sharedJob, err := jm.jobLister.Jobs(ns).Get(name)
	if err != nil {
        // 未找到
		if apierrors.IsNotFound(err) {
            // 清理exp记录
			jm.expectations.DeleteExpectations(key)
            
            // 清理跟踪finalizer的pod uid列表
			jm.finalizerExpectations.deleteExpectations(key)

            // 清理backoff记录
			jm.backoffRecordStore.removeBackoffRecord(key)
			...
            
			return nil
		}
		return err
	}
    
	// make a copy so we don't mutate the shared cache
	job := *sharedJob.DeepCopy()

    // job完成(走到终态)
	if IsJobFinished(&job) {
        // 清理backoff记录
		jm.backoffRecordStore.removeBackoffRecord(key)
		...
        
		return nil
	}

    // 完成模式仅支持NonIndexedCompletion和IndexedCompletion模式
    // NonIndexedCompletion模式:仅关心数量,不关心顺序
    // IndexedCompletion模式:不同pod具备唯一索引,用于分片处理数据
	if *job.Spec.CompletionMode != NonIndexedCompletion && *job.Spec.CompletionMode != IndexedCompletion {
		return nil
	}

    // 获取完成模式
	completionMode := getCompletionMode(&job)
	...

    // status暂存的走到终态未统计的Pod(带finalizer的)
	uncounted := newUncountedTerminatedPods(*job.Status.UncountedTerminatedPods)
    
    // 获取期望移除finalizer的Pod UID列表
	expectedRmFinalizers := jm.finalizerExpectations.getExpectedUIDs(key)

    // 检查Job对象同步条件(add/del<=0、超时或首次调度)
	satisfiedExpectations := jm.expectations.SatisfiedExpectations(key)

    // 获取Job相关Pod(涉及领养和弃养,过程与rs一致)
	pods, err := jm.getPodsForJob(ctx, &job)
	...

    // 统计active Pod
	activePods := controller.FilterActivePods(pods)
	active := int32(len(activePods))
    
    // 获取新成功及新失败的Pod(排除uncounted/expectedRmFinalizers缓存的pod)
	newSucceededPods, newFailedPods := getNewFinishedPods(&job, pods, uncounted, expectedRmFinalizers)
    
    // 统计成功的Pod数量
	succeeded := job.Status.Succeeded + int32(len(newSucceededPods)) + int32(len(uncounted.succeeded))
    // 统计失败的Pod数量
	failed := job.Status.Failed + int32(nonIgnoredFailedPodsCount(&job, newFailedPods)) + int32(len(uncounted.failed))
    ...
    
    // 统计condition为true的Pod数量(ready)
	if feature.DefaultFeatureGate.Enabled(features.JobReadyPods) {
		ready = pointer.Int32(countReadyPods(activePods))
	}

	// 设置未挂起的job.status.startTime
	if job.Status.StartTime == nil && !jobSuspended(&job) {
		now := metav1.NewTime(jm.clock.Now())
		job.Status.StartTime = &now
	}

    // 生成最新的backoff记录
	newBackoffInfo := jm.backoffRecordStore.newBackoffRecord(key, newSucceededPods, newFailedPods)
	...

    // 检测到新的失败
	jobHasNewFailure := failed > job.Status.Failed
    
    // 检测失败Pod超出backofflimit限制
	exceedsBackoffLimit := failed > *job.Spec.BackoffLimit

    // jobPodFailurePolicy
	if feature.DefaultFeatureGate.Enabled(features.JobPodFailurePolicy) {
        // 获取FailureTargetCondition构造finishedCondition
		if failureTargetCondition := findConditionByType(job.Status.Conditions, batch.JobFailureTarget); failureTargetCondition != nil {
			finishedCondition = newFailedConditionForFailureTarget(failureTargetCondition, jm.clock.Now())
        // 否则基于Pod失败信息构造finishedCondition
		} else if failJobMessage := getFailJobMessage(&job, pods, uncounted.Failed()); failJobMessage != nil {
			finishedCondition = newCondition(batch.JobFailureTarget, v1.ConditionTrue, jobConditionReasonPodFailurePolicy, *failJobMessage, jm.clock.Now())
		}
	}
    
    // finishedCondition为空
	if finishedCondition == nil {
        // 失败Pod超出backoff limit限制或failure-on-restart计数超过阈值
		if exceedsBackoffLimit || pastBackoffLimitOnFailure(&job, pods) {
			finishedCondition = newCondition(batch.JobFailed, v1.ConditionTrue, "BackoffLimitExceeded", "Job has reached the specified backoff limit", jm.clock.Now())
        // JoB活跃时间超出activeDeadline
		} else if jm.pastActiveDeadline(&job) {
			finishedCondition = newCondition(batch.JobFailed, v1.ConditionTrue, "DeadlineExceeded", "Job was active longer than specified deadline", jm.clock.Now())
        // 设置了activeDeadlineSeconds及Job未挂起,延迟入队再检查
		} else if job.Spec.ActiveDeadlineSeconds != nil && !jobSuspended(&job) {
			syncDuration := time.Duration(*job.Spec.ActiveDeadlineSeconds)*time.Second - jm.clock.Since(job.Status.StartTime.Time)
			jm.queue.AddAfter(key, syncDuration)
		}
	}
	...
    
    // 索引完成模式
	if isIndexedJob(&job) {
        // 计算prevSucceededIndexes和curSucceedIndexes
		prevSucceededIndexes, succeededIndexes = calculateSucceededIndexes(&job, pods)
		succeeded = int32(succeededIndexes.total())
	}
    ...
    
    // finishedCond已初始化,不执行调度,清理activePod
	if finishedCondition != nil {
        // 清理activePod
		deleted, err := jm.deleteActivePods(ctx, &job, activePods)
        // activePod未全部清理或未匹配同步条件,重置finishedCondition,避免更新job为终态
		if deleted != active || !satisfiedExpectations {
			finishedCondition = nil
		}
        
        // 更新active数量
		active -= deleted
		...
	} else {
		...
        
        // 同步条件满足且Job未删除
		if satisfiedExpectations && job.DeletionTimestamp == nil {
            // Job所属Pod
			active, action, manageJobErr = jm.manageJob(ctx, &job, activePods, succeeded, succeededIndexes, newBackoffInfo)
			manageJobCalled = true
		}
        ...
        
        // 未设置Completions的Job
		if job.Spec.Completions == nil {
			// Pod均达到终态才标记完成
			complete = succeeded > 0 && active == 0
        // 设置Completions的Job
		} else {
			// Pod均达到终态且完成数量达标才标记完成
			complete = succeeded >= *job.Spec.Completions && active == 0
		}
        
        // 完成则重置finishedCondition
		if complete {
			finishedCondition = newCondition(batch.JobComplete, v1.ConditionTrue, "", "", jm.clock.Now())
        // 未完成
		} else if manageJobCalled {
			// suspend
			if job.Spec.Suspend != nil && *job.Spec.Suspend {
				...
                
                // 检测suspend condition更新
				job.Status.Conditions, isUpdated = ensureJobConditionStatus(job.Status.Conditions, batch.JobSuspended, v1.ConditionTrue, "JobSuspended", "Job suspended", jm.clock.Now())
                
				if isUpdated {
					suspendCondChanged = true
				}
            // 非suspend
			} else {
				...
                
                // 检测suspend condition更新
				job.Status.Conditions, isUpdated = ensureJobConditionStatus(job.Status.Conditions, batch.JobSuspended, v1.ConditionFalse, "JobResumed", "Job resumed", jm.clock.Now())
				if isUpdated {
					suspendCondChanged = true
					now := metav1.NewTime(jm.clock.Now())
                    // 重置startTime
					job.Status.StartTime = &now
				}
			}
		}
	}

    // suspend调整/activePod变化/ready数变化,标记更新Job.status
	needsStatusUpdate := suspendCondChanged || active != job.Status.Active || !equalReady(ready, job.Status.Ready)
	job.Status.Active = active
	job.Status.Ready = ready
    
    // 更新job.status,清理pod finalizer,更新uncountedTerminatedPods记录
	err = jm.trackJobStatusAndRemoveFinalizers(ctx, &job, pods, prevSucceededIndexes, *uncounted, expectedRmFinalizers, finishedCondition, needsStatusUpdate, newBackoffInfo)
	
    if err != nil {
        // informer cache和APIServer的资源版本冲突,重新入队
		if apierrors.IsConflict(err) {
			jm.enqueueController(&job, false)
			return nil
		}
		return fmt.Errorf("tracking status: %w", err)
	}

    // 检测到新失败且Job未完成,触发重试
	jobFinished := IsJobFinished(&job)
	if jobHasNewFailure && !jobFinished {
		// returning an error will re-enqueue Job after the backoff period
		return fmt.Errorf("failed pod(s) detected for job key %q", key)
	}

	return manageJobErr
}

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
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219

注意

syncJob()本质上是检查Job调度条件，不满足会清理activePod，满足会执行manageJob()进行调度，流程图省略了很多状态计算

# 3.3.manageJob

jm.manageJob()用于执行Job调度，根据Job的期望状态动态创建或删除Pod，驱动Job运行状态达到预期，必要时延迟重试。

// manageJob is the core method responsible for managing the number of running pods.
func (jm *Controller) manageJob(...) (int32, string, error) {
	...

    // suspended(暂停)
	if jobSuspended(job) {
        // 匹配需删除的Pod(这里全部删除)
        // Index完成模式: Index未找到/未调度/未完成的/未就绪的/就绪时间较短的/container重启次数多的/创建时间短的先删除
        // 非Index模式/Index模式删除进度不够: 直接从left截取
		podsToDelete := activePodsForRemoval(job, activePods, int(active))
        // 记录期望删除的pod,del=len(podsToDelete)
		jm.expectations.ExpectDeletions(jobKey, len(podsToDelete))
        // 摘掉tracking finalizer,删除Pod
		removed, err := jm.deleteJobPods(ctx, job, jobKey, podsToDelete)
		active -= removed
		return active, metrics.JobSyncActionPodsDeleted, err
	}
	...
    
    // 未设置Completions
	if job.Spec.Completions == nil {
		// 已经有成功的Pod,wantActive设置为active数量,代表不创建或强制删除,保持现状
		if succeeded > 0 {
			wantActive = active
        // 否则设为job.spec.parallelism,尝试争抢第一个成功
		} else {
			wantActive = parallelism
		}
	} else {
		// 设置wantActive期望数量
		wantActive = *job.Spec.Completions - succeeded
        // 不能比parallelism大
		if wantActive > parallelism {
			wantActive = parallelism
		}
		if wantActive < 0 {
			wantActive = 0
		}
	}

    // 至少删除的Pod数量(多出的)
	rmAtLeast := active - wantActive
	if rmAtLeast < 0 {
		rmAtLeast = 0
	}
    
    // 计算需要删除的Pod(多出的)
	podsToDelete := activePodsForRemoval(job, activePods, int(rmAtLeast))
    // 每轮最多创建或删除500个
	if len(podsToDelete) > MaxPodCreateDeletePerSync {
		podsToDelete = podsToDelete[:MaxPodCreateDeletePerSync]
	}
	if len(podsToDelete) > 0 {
        // 记录期望删除Pod
		jm.expectations.ExpectDeletions(jobKey, len(podsToDelete))
		// 摘除tracking finalizer及删除
		removed, err := jm.deleteJobPods(ctx, job, jobKey, podsToDelete)
		active -= removed
        
		return active, metrics.JobSyncActionPodsDeleted, err
	}

    // 期望的activePod不够
	if active < wantActive {
        // backoff未结束(距上次出现失败时间--10*2^fail-1,最大360s)
		remainingTime := backoff.getRemainingTime(jm.clock, DefaultJobBackOff, MaxJobBackOff)
        // 根据剩余时间延迟入队
		if remainingTime > 0 {
			jm.enqueueControllerDelayed(job, true, remainingTime)
			return 0, metrics.JobSyncActionPodsCreated, nil
		}
        
        // 计算预计添加数量(最大500)
		diff := wantActive - active
		if diff > int32(MaxPodCreateDeletePerSync) {
			diff = int32(MaxPodCreateDeletePerSync)
		}

        // 设置exp对象,add=diff
		jm.expectations.ExpectCreations(jobKey, int(diff))
		...

		// index模式
		if isIndexedJob(job) {
            // 计算Index索引[0,min(diff,completions-1)]
			indexesToAdd = firstPendingIndexes(activePods, succeededIndexes, int(diff), int(*job.Spec.Completions))
			diff = int32(len(indexesToAdd))
		}
		active += diff

        // 获取pod模板
		podTemplate := job.Spec.Template.DeepCopy()
        // index模式设置pod定义的环境变量
		if isIndexedJob(job) {
			addCompletionIndexEnvVariables(podTemplate)
		}
        
        // 设置tracking finalizer(成功后移除)
		podTemplate.Finalizers = appendJobCompletionFinalizerIfNotFound(podTemplate.Finalizers)

		// 批量创建,根据1,2,4,8,...,diff步长
		for batchSize := int32(integer.IntMin(int(diff), controller.SlowStartInitialBatchSize)); diff > 0; batchSize = integer.Int32Min(2*batchSize, diff) {
			...
            
			for i := int32(0); i < batchSize; i++ {
				...
                
                // 获取index索引
				if len(indexesToAdd) > 0 {
					completionIndex = indexesToAdd[0]
					indexesToAdd = indexesToAdd[1:]
				}
                
				go func() {
					template := podTemplate
					generateName := ""
					if completionIndex != unknownCompletionIndex {
						template = podTemplate.DeepCopy()
                        // 设置pod index
						addCompletionIndexAnnotation(template, completionIndex)
                        // 生成pod名称
						template.Spec.Hostname = fmt.Sprintf("%s-%d", job.Name, completionIndex)
						generateName = podGenerateNameWithIndex(job.Name, completionIndex)
					}
					...
                    
                    // 创建Pod
					err := jm.podControl.CreatePodsWithGenerateName(ctx, job.Namespace, template, job, metav1.NewControllerRef(job, controllerKind), generateName)
					...
                    
					if err != nil {
						// 手动扣减创建失败的Pod,确保状态一致
						jm.expectations.CreationObserved(jobKey)
						atomic.AddInt32(&active, -1)
						errCh <- err
					}
				}()
			}
			wait.Wait()
			
            // 计算创建跳过的Pod
			skippedPods := diff - batchSize
			
            // 手动扣减创建跳过的Pod,确保期望状态一致(这里报错就退出了)
            if errorCount < len(errCh) && skippedPods > 0 {
				active -= skippedPods
				for i := int32(0); i < skippedPods; i++ {
					jm.expectations.CreationObserved(jobKey)
				}
				// The skipped pods will be retried later. The next controller resync will
				// retry the slow start process.
				break
			}
			diff -= batchSize
		}
        
		return active, metrics.JobSyncActionPodsCreated, errorFromChannel(errCh)
	}

	return active, metrics.JobSyncActionTracking, nil
}

注意

manageJob()会根据多出或缺少的数量创建或删除Pod，此外会设置tracking finalizer及index annotation(索引模式)

# 3.4.updateJob

jm.trackJobStatusAndRemoveFinalizers()用于计算Job状态，清理期望摘除tracking finalizer的Pod，更新Job最新状态。

// trackJobStatusAndRemoveFinalizers does:
//  1. Add finished Pods to .status.uncountedTerminatedPods
//  2. Remove the finalizers from the Pods if they completed or were removed or the job was removed.
//  3. Increment job counters for pods that no longer have a finalizer.
//  4. Add Complete condition if satisfied with current counters.
func (jm *Controller) trackJobStatusAndRemoveFinalizers(...) error {
	...
    
    // 暂存job.Status.UncountedTerminatedPods
	uncountedStatus := job.Status.UncountedTerminatedPods
	...
    
    // Index完成模式,根据Index排序排序Pod
	if isIndexed {
		// Sort to introduce completed Indexes in order.
		sort.Sort(byCompletionIndex(pods))
	}
    ...
    
    // 记录标记tracking finalizer及未在预期删除列表的Pod(仍需关注)
	for _, p := range pods {
		uid := string(p.UID)
		if hasJobTrackingFinalizer(p) && !expectedRmFinalizers.Has(uid) {
			uidsWithFinalizer.Insert(uid)
		}
	}

	// status浅拷贝,用于后续比较状态变化
	oldCounters := job.Status
    
    // 清理uncountedPods记录的已经没有tracking finalizer的Pod,更新成功及失败数
	if cleanUncountedPodsWithoutFinalizers(&job.Status, uidsWithFinalizer) {
		needsFlush = true
	}
	...
    
    // 遍历Pod
	for _, pod := range pods {
        // Pod没有tracking finalizer或已处于期望删除列表,无需处理
		if !hasJobTrackingFinalizer(pod) || expectedRmFinalizers.Has(string(pod.UID)) {
			// This pod was processed in a previous sync.
			continue
		}
		
        // 检查终止条件(Pod终止/Job完成)
		considerTerminated := pod.DeletionTimestamp != nil || finishedCond != nil

        // 基于策略重置Pod终止
		if feature.DefaultFeatureGate.Enabled(features.PodDisruptionConditions) && feature.DefaultFeatureGate.Enabled(features.JobPodFailurePolicy) && job.Spec.PodFailurePolicy != nil {
			considerTerminated = podutil.IsPodTerminal(pod) ||
				finishedCond != nil || // The Job is terminating. Any running Pod is considered failed.
				isPodFailed(pod, job)
		}
        
        // Pod执行到终态/终止,Pod计划移除finalizer
		if podutil.IsPodTerminal(pod) || considerTerminated || job.DeletionTimestamp != nil {
			podsToRemoveFinalizer = append(podsToRemoveFinalizer, pod)
		}
        
        // Pod走到Succeed&&uncounted.failed未覆盖
		if pod.Status.Phase == v1.PodSucceeded && !uncounted.failed.Has(string(pod.UID)) {
			// Index Job
            if isIndexed {
				// 获取Index
				ix := getCompletionIndex(pod.Annotations)
                // Index有效+未记录到succeededIndexes
				if ix != unknownCompletionIndex && ix < int(*job.Spec.Completions) && !succeededIndexes.has(ix){
					// 记录到newSucceededIndexes
                    newSucceededIndexes = append(newSucceededIndexes, ix)
					needsFlush = true
				}
            // uncounted.succeeded未覆盖Pod
			} else if !uncounted.succeeded.Has(string(pod.UID)) {
				needsFlush = true
                // uncounted.succeeded记录pod uid
				uncountedStatus.Succeeded = append(uncountedStatus.Succeeded, pod.UID)
			}
        // Pod走到Failed/终止
		} else if pod.Status.Phase == v1.PodFailed || considerTerminated {
            // 获取Index
			ix := getCompletionIndex(pod.Annotations)
            // uncounted.failed未覆盖Pod且(非Index Job或Index有效)
			if !uncounted.failed.Has(string(pod.UID)) && (!isIndexed || ix < int(*job.Spec.Completions)) {
				// 开启及设置了PodFailurePolicy
                if feature.DefaultFeatureGate.Enabled(JobPodFailurePolicy) && job.Spec.PodFailurePolicy != nil {
					// 匹配处理策略
                    _, countFailed, action := matchPodFailurePolicy(job.Spec.PodFailurePolicy, pod)
					...
                    
                    // 需统计失败
					if countFailed {
						needsFlush = true
                        // pod uid记录到uncountedStatus.Failed
						uncountedStatus.Failed = append(uncountedStatus.Failed, pod.UID)
					}
				} else {
					needsFlush = true
                    // pod uid记录到uncountedStatus.Failed
					uncountedStatus.Failed = append(uncountedStatus.Failed, pod.UID)
				}
			}
		}
        
        // 已累计的未计数条目数超出500
		if len(newSucceededIndexes)+len(uncountedStatus.Succeeded)+len(uncountedStatus.Failed) >= 500 {
			// 标记
			reachedMaxUncountedPods = true
			break
		}
	}
    
    // Index Job
	if isIndexed {
        // 合并succeedIndex及排序
		succeededIndexes = succeededIndexes.withOrderedIndexes(newSucceededIndexes)
		succeededIndexesStr := succeededIndexes.String()
        // 检查completeIndex是否需要更新
		if succeededIndexesStr != job.Status.CompletedIndexes {
			needsFlush = true
		}
        // 更新成功数及Index
		job.Status.Succeeded = int32(succeededIndexes.total())
		job.Status.CompletedIndexes = succeededIndexesStr
	}
    
    // 启用JobPodFailurePolicy
	if feature.DefaultFeatureGate.Enabled(features.JobPodFailurePolicy) {
        // Job走到Failed终态
		if finishedCond != nil && finishedCond.Type == batch.JobFailureTarget {
			// 向condition塞入临时失败信息
			job.Status.Conditions = append(job.Status.Conditions, *finishedCond)
			needsFlush = true

			// 准备最终的失败信息
			finishedCond = newFailedConditionForFailureTarget(finishedCond, jm.clock.Now())
		}
	}

    // 没有finalizer的uncountedPod结算进Job.Status.succeeded/failed
    // 移除期望清理finalizer的Pod finalizer
	job, needsFlush, err = jm.flushUncountedAndRemoveFinalizers(ctx, job, podsToRemoveFinalizer, uidsWithFinalizer, &oldCounters, podFailureCountByPolicyAction, needsFlush, newBackoffRecord)
    ...
    
    // 计算完成条件(未达到500触发上限+完成条件满足,更新conditions)
	jobFinished := !reachedMaxUncountedPods && jm.enactJobFinished(job, finishedCond)
    if jobFinished {
		needsFlush = true
	}
    
    // 需要刷新状态
	if needsFlush {
        // 更新状态
		jm.updateStatusHandler(ctx, job)
        ...
	}
	return nil
}

// flushUncountedAndRemoveFinalizers does:
//  1. flush the Job status that might include new uncounted Pod UIDs. Also flush the FailureTarget condition.
//  2. perform the removal of finalizers from Pods which are in the uncounted lists.
//  3. update the counters based on the Pods for which it successfully removed the finalizers.
//  4. (if not all removals succeeded) flush Job status again.
func (jm *Controller) flushUncountedAndRemoveFinalizers(...) (*batch.Job, bool, error) {
	...
    
    // 刷新
	if needsFlush {
        // 状态刷新
		jm.updateStatusHandler(ctx, job)
        ...

        // 更新backoff记录
		jm.backoffRecordStore.updateBackoffRecord(newBackoffRecord)
		...

		*oldCounters = job.Status
		needsFlush = false
	}
	...
    
	if len(podsToRemoveFinalizer) > 0 {
		...
        
        // 记录期望清理finalizer的pod uid,摘除成功由记录的uid清理
		rmSucceded, rmErr = jm.removeTrackingFinalizerFromPods(ctx, jobKey, podsToRemoveFinalizer)
        // 更新uidWithFinalizer
		for i, p := range podsToRemoveFinalizer {
			if rmSucceded[i] {
				uidsWithFinalizer.Delete(string(p.UID))
			}
		}
	}
    
	// 清理已经没有finalizer的uncountedPod
	if cleanUncountedPodsWithoutFinalizers(&job.Status, uidsWithFinalizer) {
		needsFlush = true
	}
    
    // 再次更新状态
	if rmErr != nil && needsFlush {
		jm.updateStatusHandler(ctx, job)
        ...
	}
	return job, needsFlush, rmErr
}

注意

这里需要注意的是，uncountedTerminatedPods用于记录走到终态finalizer未摘除的Pod，finalizer摘除后才会记录到succeed数量

# 4.orphanworker

# 4.1.process

jm.orphanWorker()用于处理orphan pod的清理，设计上依然会不停由orphan pod queue获取item，执行jm.syncOrphanPod()处理。

func (jm *Controller) orphanWorker(ctx context.Context) {
	for jm.processNextOrphanPod(ctx) {
	}
}

func (jm Controller) processNextOrphanPod(ctx context.Context) bool {
	key, quit := jm.orphanQueue.Get()
	...
    
	defer jm.orphanQueue.Done(key)
    
    // orphan pod回收
	err := jm.syncOrphanPod(ctx, key.(string))
	if err != nil {
		jm.orphanQueue.AddRateLimited(key)
	} else {
		jm.orphanQueue.Forget(key)
	}

	return true
}

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

注意

没有owner但标记tracking finalizer或owner不存在、已完成或正在删除，都属于orphan pod，会推入orphan queue

# 4.2.syncOrphan

jm.syncOrphanPod()会获取orphan Pod，根据Pod owner完成情况确认Pod仍然是孤儿，然后会摘除tracking finalizer。

// syncOrphanPod removes the tracking finalizer from an orphan pod if found.
func (jm Controller) syncOrphanPod(ctx context.Context, key string) error {
	...

    // 获取Pod
	sharedPod, err := jm.podStore.Pods(ns).Get(name)
	...
    
	// 确认Pod依然是孤儿
	if controllerRef := metav1.GetControllerOf(sharedPod); controllerRef != nil {
        // 解析owner(被领养)
		job := jm.resolveControllerRef(sharedPod.Namespace, controllerRef)
        // owner未完成
		if job != nil && !IsJobFinished(job) {
			// The pod was adopted. Do not remove finalizer.
			return nil
		}
	}
    
    // 清理Pod tracking finalizer
	if patch := removeTrackingFinalizerPatch(sharedPod); patch != nil {
		jm.podControl.PatchPod(ctx, ns, name, patch)
        ...
	}
    
	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

注意

orphan Pod处理相对简单，仅摘除tracking finalizer，当然部分情况下orphan Pod会被Job领养，就不处理了