Kubernetes Job Controller 原理和源码分析(三)

概述Job controller 的启动processNextWorkItem()核心调谐逻辑入口 - syncJob()Pod 数量管理 - manageJob()小结

概述

源码版本:kubernetes master 分支 commit-fe62fc(2021年10月14日)

Job 是主要的 Kubernetes 原生 Workload 资源之一,是在 Kubernetes 之上运行批处理任务最简单的方式,在 AI 模型训练等场景下最基础的实现版本就是拉起一个 Job 来完成一次训练任务,然后才是各种自定义 “Job” 实现进阶处理,比如分布式训练需要一个 “Job” 同时拉起多个 Pod,但是每个 Pod 的启动参数会有差异。所以深入理解 Job 的功能和实现细节是进一步开发自定义 “Job” 类型工作负载的基础。

我们在《Kubernetes Job Controller 原理和源码分析(一)》中详细介绍了 Job 的特性,在《Kubernetes Job Controller 原理和源码分析(二)》 中一路从 Job 控制器源码入口跟到所有 EventHandler 的实现,今天我们继续从 workqueue 的另外一端看下任务出队后的主要调谐逻辑实现。

注意:阅读 Job 源码需要有一定的自定义控制器工作原理基础,里面涉及到了 Informer 工作机制、workqueue(延时工作队列)、ResourceEventHandler 等等逻辑,没有相关知识储备直接看本文会有一定挑战,建议先阅读《深入理解 K8S 原理与实现》系列目录里列的相关文章。


《Kubernetes Job Controller 原理和源码分析》分为三讲:

Job controller 的启动

继续来看 Run() 方法

  • pkg/controller/job/job_controller.go
 1func (jm *Controller) Run(workers int, stopCh <-chan struct{}) {
2   defer utilruntime.HandleCrash()
3   defer jm.queue.ShutDown()
4   defer jm.orphanQueue.ShutDown()
5
6   klog.Infof("Starting job controller")
7   defer klog.Infof("Shutting down job controller")
8
9   if !cache.WaitForNamedCacheSync("job", stopCh, jm.podStoreSynced, jm.jobStoreSynced) {
10      return
11   }
12
13   for i := 0; i < workers; i++ {
14      go wait.Until(jm.worker, time.Second, stopCh)
15   }
16
17   go wait.Until(jm.orphanWorker, time.Second, stopCh)
18
19   <-stopCh
20}

可以看到这里的逻辑在两个 worker 里,继续看 jm.workerjm.orphanWorker 是什么逻辑

1func (jm *Controller) worker() {
2   for jm.processNextWorkItem() {
3   }
4}

worker() 方法里简单调用了 processNextWorkItem() 方法,另外一个 orphanWorker() 也是类似逻辑

1func (jm *Controller) orphanWorker() {
2   for jm.processNextOrphanPod() {
3   }
4}

继续看

processNextWorkItem()

 1func (jm *Controller) processNextWorkItem() bool {
2   key, quit := jm.queue.Get()
3   if quit {
4      return false
5   }
6   defer jm.queue.Done(key)
7
8   forget, err := jm.syncHandler(key.(string)) // 核心逻辑
9   if err == nil {
10      if forget {
11         jm.queue.Forget(key)
12      }
13      return true
14   }
15
16   utilruntime.HandleError(fmt.Errorf("syncing job: %w", err))
17   if !apierrors.IsConflict(err) {
18      jm.queue.AddRateLimited(key)
19   }
20
21   return true
22}

这里的核心逻辑只有一行,就是 jm.syncHandler(key.(string)) 调用。我们往里跟这个 sincHandler 方法可以看到其实现是 func (jm *Controller) syncJob(key string) (forget bool, rErr error) 方法,这个方法里就是主要的调谐逻辑了。看下具体逻辑:

核心调谐逻辑入口 - syncJob()

从 workqueue 出来任务之后主要的业务逻辑从这里开始,这个方法很长……

  • pkg/controller/job/job_controller.go:588
  1func (jm *Controller) syncJob(key string) (forget bool, rErr error) {
2   startTime := time.Now()
3   defer func() {
4      klog.V(4).Infof("Finished syncing job %q (%v)", key, time.Since(startTime))
5   }()
6     // key 的结构一般是 namespace/name 的格式
7   ns, name, err := cache.SplitMetaNamespaceKey(key)
8   if err != nil {
9      return false, err
10   }
11   if len(ns) == 0 || len(name) == 0 {
12      return false, fmt.Errorf("invalid job key %q: either namespace or name is missing", key)
13   }
14   // 这里叫 sharedJob 是因为取的是本地 cache 的 job,通过 Indexer 提供的能力
15   sharedJob, err := jm.jobLister.Jobs(ns).Get(name)
16   if err != nil {
17      // 如果找不到,说明被其他 goroutine 删掉了,忽略
18      if apierrors.IsNotFound(err) {
19         klog.V(4).Infof("Job has been deleted: %v", key)
20         jm.expectations.DeleteExpectations(key)
21         jm.finalizerExpectations.deleteExpectations(key)
22         return true, nil
23      }
24      return false, err
25   }
26   // 拷贝一份避免修改
27   job := *sharedJob.DeepCopy()
28
29   // 通过 JobCondition 的 Type 是否为 "Complete"/"Failed" 来判断 job 是否已经完成了
30   if IsJobFinished(&job) {
31      return true, nil
32   }
33
34   // 这个 feature 是 1.22 版本进入 beta 的,如果两边配置不一致,则无法继续处理
35   // 本质通过 .Spec.CompletionMode == "Indexed" 来判断
36   if !feature.DefaultFeatureGate.Enabled(features.IndexedJob) && isIndexedJob(&job) {
37      jm.recorder.Event(&job, v1.EventTypeWarning, "IndexedJobDisabled", "Skipped Indexed Job sync because feature is disabled.")
38      return false, nil
39   }
40   // CompletionMode 为 "NonIndexed"/"Indexed",如果是其他值则不识别
41   if job.Spec.CompletionMode != nil && *job.Spec.CompletionMode != batch.NonIndexedCompletion && *job.Spec.CompletionMode != batch.IndexedCompletion {
42      jm.recorder.Event(&job, v1.EventTypeWarning, "UnknownCompletionMode", "Skipped Job sync because completion mode is unknown")
43      return false, nil
44   }
45
46   // 配置当前的 completionMode,默认为 "NonIndexed"
47   completionMode := string(batch.NonIndexedCompletion)
48   if isIndexedJob(&job) {
49      completionMode = string(batch.IndexedCompletion)
50   }
51   // "reconciling"
52   action := metrics.JobSyncActionReconciling
53
54   // metrics 逻辑
55   defer func() {
56      result := "success"
57      if rErr != nil {
58         result = "error"
59      }
60
61      metrics.JobSyncDurationSeconds.WithLabelValues(completionMode, result, action).Observe(time.Since(startTime).Seconds())
62      metrics.JobSyncNum.WithLabelValues(completionMode, result, action).Inc()
63   }()
64
65   var expectedRmFinalizers sets.String
66   var uncounted *uncountedTerminatedPods
67   // 处理 pod finalizer,1.22 版本 alpha 的特性
68   if trackingUncountedPods(&job) {
69      klog.V(4).InfoS("Tracking uncounted Pods with pod finalizers", "job", klog.KObj(&job))
70      if job.Status.UncountedTerminatedPods == nil {
71         job.Status.UncountedTerminatedPods = &batch.UncountedTerminatedPods{}
72      }
73      uncounted = newUncountedTerminatedPods(*job.Status.UncountedTerminatedPods)
74      expectedRmFinalizers = jm.finalizerExpectations.getExpectedUIDs(key)
75     // 删除 job 的 "batch.kubernetes.io/job-tracking" 注解
76   } else if patch := removeTrackingAnnotationPatch(&job); patch != nil {
77      if err := jm.patchJobHandler(&job, patch); err != nil {
78         return false, fmt.Errorf("removing tracking finalizer from job %s: %w", key, err)
79      }
80   }
81
82   jobNeedsSync := jm.expectations.SatisfiedExpectations(key)
83
84   // 提取相关 pods
85   pods, err := jm.getPodsForJob(&job, uncounted != nil)
86   if err != nil {
87      return false, err
88   }
89   // 判断依据是 PodPhase 不为 "Succeeded" 和 "Failed" 两个结果态
90   activePods := controller.FilterActivePods(pods)
91   active := int32(len(activePods))
92   // 计算 "Succeeded" 和 "Failed" 状态 pod 的数量
93   succeeded, failed := getStatus(&job, pods, uncounted, expectedRmFinalizers)
94   // 满足这个条件说明这个 pod 是新创建的,这时候需要设置 .Status.StartTime
95   if job.Status.StartTime == nil && !jobSuspended(&job) {
96      now := metav1.Now()
97      job.Status.StartTime = &now
98      // 如果 ActiveDeadlineSeconds 不为空,则在 ActiveDeadlineSeconds 时间到后再次调谐
99      if job.Spec.ActiveDeadlineSeconds != nil {
100         klog.V(4).Infof("Job %s has ActiveDeadlineSeconds will sync after %d seconds",
101            key, *job.Spec.ActiveDeadlineSeconds)
102         jm.queue.AddAfter(key, time.Duration(*job.Spec.ActiveDeadlineSeconds)*time.Second)
103      }
104   }
105
106   var manageJobErr error
107   var finishedCondition *batch.JobCondition
108
109   // 有新增 failed 到 pod
110   jobHasNewFailure := failed > job.Status.Failed
111
112   // 有新的 failed pod 产生,而且 active 的 pod 数量不等于并发数,而且已经失败的 pod 数量大于重试次数限制
113   exceedsBackoffLimit := jobHasNewFailure && (active != *job.Spec.Parallelism) &&
114      (failed > *job.Spec.BackoffLimit)
115   // pastBackoffLimitOnFailure 计算的是当 pod 重启策略为 OnFailure 时重启次数是否超过限制
116   if exceedsBackoffLimit || pastBackoffLimitOnFailure(&job, pods) {
117      // 重试次数达到上限,Condition 更新为 "Failed"
118      finishedCondition = newCondition(batch.JobFailed, v1.ConditionTrue, "BackoffLimitExceeded", "Job has reached the specified backoff limit")
119   // 超时了
120   } else if pastActiveDeadline(&job) {
121      finishedCondition = newCondition(batch.JobFailed, v1.ConditionTrue, "DeadlineExceeded", "Job was active longer than specified deadline")
122   }
123   // 计算索引
124   var prevSucceededIndexes, succeededIndexes orderedIntervals
125   if isIndexedJob(&job) {
126      prevSucceededIndexes, succeededIndexes = calculateSucceededIndexes(&job, pods)
127      succeeded = int32(succeededIndexes.total())
128   }
129   suspendCondChanged := false
130   // 如果 job 失败了,这时候在 Active 状态的 Pod 需要直接删除
131   if finishedCondition != nil {
132      deleted, err := jm.deleteActivePods(&job, activePods)
133      if uncounted == nil {
134         deleted = active
135      } else if deleted != active {
136         finishedCondition = nil
137      }
138      active -= deleted
139      failed += deleted
140      manageJobErr = err
141   } else {
142      manageJobCalled := false
143      if jobNeedsSync && job.DeletionTimestamp == nil {
144         // manageJob() 方法是根据 Spec 管理运行中 Pod 数量的核心方法
145         active, action, manageJobErr = jm.manageJob(&job, activePods, succeeded, succeededIndexes)
146         manageJobCalled = true
147      }
148      // 判断 job 已经完成
149      complete := false
150      if job.Spec.Completions == nil {
151         complete = succeeded > 0 && active == 0
152      } else {
153         complete = succeeded >= *job.Spec.Completions && active == 0
154      }
155      if complete {
156         finishedCondition = newCondition(batch.JobComplete, v1.ConditionTrue, "", "")
157        // Job 挂起是 1.22 版本 beta 的新特性
158      } else if feature.DefaultFeatureGate.Enabled(features.SuspendJob) && manageJobCalled {
159         // 如果配置了挂起
160         if job.Spec.Suspend != nil && *job.Spec.Suspend {
161            // 只有没完成的 Job 可以被挂起
162            var isUpdated bool
163            job.Status.Conditions, isUpdated = ensureJobConditionStatus(job.Status.Conditions, batch.JobSuspended, v1.ConditionTrue, "JobSuspended", "Job suspended")
164            if isUpdated {
165               suspendCondChanged = true
166               jm.recorder.Event(&job, v1.EventTypeNormal, "Suspended", "Job suspended")
167            }
168         } else {
169            // 挂起状态唤醒
170            var isUpdated bool
171            job.Status.Conditions, isUpdated = ensureJobConditionStatus(job.Status.Conditions, batch.JobSuspended, v1.ConditionFalse, "JobResumed", "Job resumed")
172            if isUpdated {
173               suspendCondChanged = true
174               jm.recorder.Event(&job, v1.EventTypeNormal, "Resumed", "Job resumed")
175               now := metav1.Now()
176               // 重置 StartTime
177               job.Status.StartTime = &now
178            }
179         }
180      }
181   }
182
183   forget = false
184   // 检查成功的 pod 是否多了
185   if job.Status.Succeeded < succeeded {
186      forget = true
187   }
188
189   if uncounted != nil {
190      // 挂起状态变更或者 active pod 数量变更
191      needsStatusUpdate := suspendCondChanged || active != job.Status.Active
192      job.Status.Active = active
193      // Finalizer 相关逻辑
194      err = jm.trackJobStatusAndRemoveFinalizers(&job, pods, prevSucceededIndexes, *uncounted, expectedRmFinalizers, finishedCondition, needsStatusUpdate)
195      if err != nil {
196         return false, fmt.Errorf("tracking status: %w", err)
197      }
198      jobFinished := IsJobFinished(&job)
199      if jobHasNewFailure && !jobFinished {
200         return forget, fmt.Errorf("failed pod(s) detected for job key %q", key)
201      }
202      forget = true
203      return forget, manageJobErr
204   }
205   // 移除所有 Finalizer
206   if err := jm.removeTrackingFinalizersFromAllPods(pods); err != nil {
207      return false, fmt.Errorf("removing disabled finalizers from job pods %s: %w", key, err)
208   }
209
210   // 判断状态是否需要更新
211   if job.Status.Active != active || job.Status.Succeeded != succeeded || job.Status.Failed != failed || suspendCondChanged || finishedCondition != nil {
212      job.Status.Active = active
213      job.Status.Succeeded = succeeded
214      job.Status.Failed = failed
215      if isIndexedJob(&job) {
216         job.Status.CompletedIndexes = succeededIndexes.String()
217      }
218      job.Status.UncountedTerminatedPods = nil
219      jm.enactJobFinished(&job, finishedCondition)
220
221      if _, err := jm.updateStatusHandler(&job); err != nil {
222         return forget, err
223      }
224
225      if jobHasNewFailure && !IsJobFinished(&job) {
226         // returning an error will re-enqueue Job after the backoff period
227         return forget, fmt.Errorf("failed pod(s) detected for job key %q", key)
228      }
229
230      forget = true
231   }
232
233   return forget, manageJobErr
234}

Pod 数量管理 - manageJob()

上面 syncJob() 中有一个 manageJob() 方法调用,manageJob() 具体控制一个 Job 下应该有多少个 Active 的 Pod,执行“多删少建”工作。这个方法也很长……

  • pkg/controller/job/job_controller.go:1245
  1func (jm *Controller) manageJob(job *batch.Job, activePods []*v1.Pod, succeeded int32, succeededIndexes []interval) (int32, string, error) {
2   // 运行中的 pod 数量
3   active := int32(len(activePods))
4   // 并发度
5   parallelism := *job.Spec.Parallelism
6   jobKey, err := controller.KeyFunc(job)
7   if err != nil {
8      utilruntime.HandleError(fmt.Errorf("Couldn't get key for job %#v: %v", job, err))
9      return 0, metrics.JobSyncActionTracking, nil
10   }
11   // 挂起状态
12   if jobSuspended(job) {
13      klog.V(4).InfoS("Deleting all active pods in suspended job", "job", klog.KObj(job), "active", active)
14      podsToDelete := activePodsForRemoval(job, activePods, int(active))
15      jm.expectations.ExpectDeletions(jobKey, len(podsToDelete))
16      // 挂起需要直接删除所有 Active 的 Pod
17      removed, err := jm.deleteJobPods(job, jobKey, podsToDelete)
18      active -= removed
19      return active, metrics.JobSyncActionPodsDeleted, err
20   }
21
22   wantActive := int32(0)
23   if job.Spec.Completions == nil {
24      // 对应没有配置 Completions 的场景,这时候运行中的 Pod 需要和并发数相等,有一个 Pod 成功时判断 Job 状态为成功
25      if succeeded > 0 {
26         wantActive = active
27      } else {
28         wantActive = parallelism
29      }
30   } else {
31      // 指定了 Completions 场景,这时候运行中的 Pod 数量不应该超过 Completions 减去已经成功的数量
32      wantActive = *job.Spec.Completions - succeeded
33      // 不能超过并发数
34      if wantActive > parallelism {
35         wantActive = parallelism
36      }
37      if wantActive < 0 {
38         wantActive = 0
39      }
40   }
41   // 如果实际 active 数量大于应该 active 的数量,就需要删除几个
42   rmAtLeast := active - wantActive
43   if rmAtLeast < 0 {
44      rmAtLeast = 0
45   }
46   // 计算哪些 Pod 需要被删除
47   podsToDelete := activePodsForRemoval(job, activePods, int(rmAtLeast))
48   if len(podsToDelete) > MaxPodCreateDeletePerSync {
49      podsToDelete = podsToDelete[:MaxPodCreateDeletePerSync]
50   }
51   // 执行删除动作
52   if len(podsToDelete) > 0 {
53      jm.expectations.ExpectDeletions(jobKey, len(podsToDelete))
54      klog.V(4).InfoS("Too many pods running for job", "job", klog.KObj(job), "deleted", len(podsToDelete), "target", wantActive)
55      removed, err := jm.deleteJobPods(job, jobKey, podsToDelete)
56      active -= removed
57      return active, metrics.JobSyncActionPodsDeleted, err
58   }
59   // 实际运行的 pod 数量不够场景
60   if active < wantActive {
61      diff := wantActive - active
62      // 如果大于上限 500,则设置为 500
63      if diff > int32(MaxPodCreateDeletePerSync) {
64         diff = int32(MaxPodCreateDeletePerSync)
65      }
66
67      jm.expectations.ExpectCreations(jobKey, int(diff))
68      errCh := make(chan error, diff)
69      klog.V(4).Infof("Too few pods running job %q, need %d, creating %d", jobKey, wantActive, diff)
70
71      wait := sync.WaitGroup{}
72
73      var indexesToAdd []int
74      if isIndexedJob(job) {
75         indexesToAdd = firstPendingIndexes(activePods, succeededIndexes, int(diff), int(*job.Spec.Completions))
76         diff = int32(len(indexesToAdd))
77      }
78      active += diff
79      // 提取 pod 模板
80      podTemplate := job.Spec.Template.DeepCopy()
81      if isIndexedJob(job) {
82         addCompletionIndexEnvVariables(podTemplate)
83      }
84      if trackingUncountedPods(job) {
85         podTemplate.Finalizers = appendJobCompletionFinalizerIfNotFound(podTemplate.Finalizers)
86      }
87      // batchSize 从 1 开始,然后指数递增,2、4、8 …… 的方式,这是一种“慢启动”过程,防止一下子尝试创建大量 Pod,但是由于相同的原因批量失败
88      for batchSize := int32(integer.IntMin(int(diff), controller.SlowStartInitialBatchSize)); diff > 0; batchSize = integer.Int32Min(2*batchSize, diff) {
89         errorCount := len(errCh)
90         wait.Add(int(batchSize))
91         for i := int32(0); i < batchSize; i++ {
92            // -1
93            completionIndex := unknownCompletionIndex
94            if len(indexesToAdd) > 0 {
95               completionIndex = indexesToAdd[0]
96               indexesToAdd = indexesToAdd[1:]
97            }
98            go func() {
99               template := podTemplate
100               generateName := ""
101               if completionIndex != unknownCompletionIndex {
102                  template = podTemplate.DeepCopy()
103                  addCompletionIndexAnnotation(template, completionIndex)
104                  // 设置 Hostname
105                  template.Spec.Hostname = fmt.Sprintf("%s-%d", job.Name, completionIndex)
106                  generateName = podGenerateNameWithIndex(job.Name, completionIndex)
107               }
108               defer wait.Done()
109               // 创建 pod
110               err := jm.podControl.CreatePodsWithGenerateName(job.Namespace, template, job, metav1.NewControllerRef(job, controllerKind), generateName)
111               if err != nil {
112                  if apierrors.HasStatusCause(err, v1.NamespaceTerminatingCause) {
113                     return
114                  }
115               }
116               if err != nil {
117                  defer utilruntime.HandleError(err)
118                  klog.V(2).Infof("Failed creation, decrementing expectations for job %q/%q", job.Namespace, job.Name)
119                  jm.expectations.CreationObserved(jobKey)
120                  atomic.AddInt32(&active, -1)
121                  errCh <- err
122               }
123            }()
124         }
125         wait.Wait()
126         skippedPods := diff - batchSize
127         if errorCount < len(errCh) && skippedPods > 0 {
128            klog.V(2).Infof("Slow-start failure. Skipping creation of %d pods, decrementing expectations for job %q/%q", skippedPods, job.Namespace, job.Name)
129            active -= skippedPods
130            for i := int32(0); i < skippedPods; i++ {
131               jm.expectations.CreationObserved(jobKey)
132            }
133            // 忽略的 pod 在下次调谐过程中继续尝试“慢启动”
134            break
135         }
136         // 成功处理的数量减掉
137         diff -= batchSize
138      }
139      return active, metrics.JobSyncActionPodsCreated, errorFromChannel(errCh)
140   }
141
142   return active, metrics.JobSyncActionTracking, nil
143}

小结

Job 控制器实现的逻辑并不复杂,就像我们一开始在《Kubernetes Job Controller 原理和源码分析(一)》中介绍的那些特性一样,Job 要支持的功能并不多。但是要完整理解 Pod 控制器全部源码并不简单,一方面需要对控制器模式本身有一定的理解,知道控制器的整体工作流;另外一方面 Job 控制器的实现中有大量健壮性代码,在实现功能的基础上代码量大了很多,要理清所有的细节还是有一定烧脑。最后这个源码组织结构着实看着不舒服,几百行代码“怼”到一个函数里,对“阅读者”不太友好。

(转载请保留本文原始链接 https://www.danielhu.cn)

Kubernetes Job Controller 原理和源码分析(三)

 

 

上一篇:拖拽事件


下一篇:开发微信公众号基础消息能力的经验(二)