7.深入k8s：任務呼叫Job與CronJob及原始碼分析
阿新 • • 發佈：2020-08-23
> 轉載請宣告出處哦~，本篇文章釋出於luozhiyun的部落格：https://www.luozhiyun.com
在使用job中，我會結合原始碼進行一定的講解，我們也可以從原始碼中一窺究竟，一些細節k8s是如何處理的，從而感受k8s的魅力。原始碼版本是[1.19](https://github.com/kubernetes/kubernetes/tree/release-1.19)
## Job
### Job的基本使用
Job主要是用來任務呼叫，可以一個或多個 Pod，並確保指定數量的 Pod 可以成功執行到程序正常結束。
建立一個Job：
```yaml
apiVersion: batch/v1
kind: Job
metadata:
name: pi
spec:
template:
spec:
containers:
- name: pi
image: perl
command: ["perl", "-Mbignum=bpi", "-wle", "print bpi(2000)"]
restartPolicy: Never
backoffLimit: 4
```
這個Job會建立一個容器，然後執行命令進行π的計算，
然後我們建立這個pod：
```shell
$ kubectl create -f job.yaml
$ kubectl describe jobs/pi
Name: pi
Namespace: default
Selector: controller-uid=cf78ebe4-07f9-4234-b8f9-2fe92df352ea
Labels: controller-uid=cf78ebe4-07f9-4234-b8f9-2fe92df352ea
job-name=pi
Annotations: Parallelism: 1
Completions: 1
...
Pods Statuses: 0 Running / 1 Succeeded / 0 Failed
Pod Template:
Labels: controller-uid=cf78ebe4-07f9-4234-b8f9-2fe92df352ea
job-name=pi
Containers:
pi:
Image: resouer/ubuntu-bc
...
Events:
Type Reason Age From Message
---- ------ ---- ---- -------
Normal SuccessfulCreate 29m job-controller Created pod: pi-g9fs4
Normal Completed 27m job-controller Job completed
```
可以看到建立物件後，Pod模板中，被自動加上了一個controller-uid=< 一個隨機字串 > 這樣的 Label。而這個 Job 物件本身，則被自動加上了這個 Label 對應的 Selector，從而 保證了 Job 與它所管理的 Pod 之間的匹配關係。這個uid避免了不同Job物件的Pod不會重合。
```shell
$ kubectl get pod
NAME READY STATUS RESTARTS AGE
pi-g9fs4 0/1 Completed 0 33m
$ kubectl describe pod pi-g9fs4
...
Events:
Type Reason Age From Message
---- ------ ---- ---- -------
Normal Scheduled 35m default-scheduler Successfully assigned default/pi-g9fs4 to 192.168.13.130
Normal Pulling 35m kubelet, 192.168.13.130 Pulling image "resouer/ubuntu-bc"
Normal Pulled 35m kubelet, 192.168.13.130 Successfully pulled image "resouer/ubuntu-bc"
Normal Created 35m kubelet, 192.168.13.130 Created container pi
Normal Started 35m kubelet, 192.168.13.130 Started container pi
```
我們可以看到Pod在建立好執行完畢之後會進入到Completed狀態。上面的yaml定義中restartPolicy=Never也保證了這個Pod只會執行一次。
如果建立的Pod執行失敗了，那麼Job Controller會不斷建立一個新的Pod：
```shell
$ kubectl get pods
NAME READY STATUS RESTARTS AGE
pi-55h89 0/1 ContainerCreating 0 2s
pi-tqbcz 0/1 Error 0 5s
```
#### 引數說明
**spec.backoffLimit**
我們在上面的欄位中定義了為4，表示重試次數為4。
**restartPolicy**
在執行過程中，可能發生各種系統問題導致的Pod執行失敗，如果設定restartPolicy為OnFailure，那麼在執行中發生的失敗後Job Controller會重啟Pod裡面的容器，而不是建立新的Pod。
還可以設定為Never，表示容器執行失敗之後不會重啟。更多具體的參見[Pod生命週期](https://kubernetes.io/docs/concepts/workloads/pods/pod-lifecycle/#example-states)。
**spec.activeDeadlineSeconds**
表示最長執行時間，單位是秒。如：
```yaml
spec:
backoffLimit: 5
activeDeadlineSeconds: 100
```
這樣設定之後會進入pastActiveDeadline進行校驗`job.Spec.ActiveDeadlineSeconds`是不是為空，不是空的話，會比較Pod的執行時間duration是否大於`job.Spec.ActiveDeadlineSeconds`設定的值，如果大於，那麼會標記Pod終止的原因是DeadlineExceeded。
在job Controller的原始碼中，我們可以看到這部分的邏輯：
job Controller首先會去校驗任務是不是處理次數是不是超過了BackoffLimit設定，如果沒有超過的話就校驗有沒有設定ActiveDeadlineSeconds，如果設定了的話，就校驗當前job執行時間是否超過了ActiveDeadlineSeconds設定的的時間，超過了那麼會打上標記，表示這個job執行失敗。
```go
...
jobHaveNewFailure := failed > job.Status.Failed
exceedsBackoffLimit := jobHaveNewFailure && (active != *job.Spec.Parallelism) &&
(int32(previousRetry)+1 > *job.Spec.BackoffLimit)
if exceedsBackoffLimit || pastBackoffLimitOnFailure(&job, pods) {
// check if the number of pod restart exceeds backoff (for restart OnFailure only)
// OR if the number of failed jobs increased since the last syncJob
jobFailed = true
failureReason = "BackoffLimitExceeded"
failureMessage = "Job has reached the specified backoff limit"
} else if pastActiveDeadline(&job) {
jobFailed = true
failureReason = "DeadlineExceeded"
failureMessage = "Job was active longer than specified deadline"
}
...
func pastActiveDeadline(job *batch.Job) bool {
if job.Spec.ActiveDeadlineSeconds == nil || job.Status.StartTime == nil {
return false
}
now := metav1.Now()
start := job.Status.StartTime.Time
duration := now.Time.Sub(start)
allowedDuration := time.Duration(*job.Spec.ActiveDeadlineSeconds) * time.Second
return duration >= allowedDuration
}
```
### Job的並行任務
在 Job 物件中，負責並行控制的引數有兩個：
1. `spec.parallelism`表示一個 Job 在任意時間最多可以啟動多少個 Pod 同時執行；
2. `spec.completions`表示Job 的最小完成數。
舉例：
```yaml
apiVersion: batch/v1
kind: Job
metadata:
name: pi
spec:
parallelism: 2
completions: 4
template:
spec:
containers:
- name: pi
image: perl
command: ["perl", "-Mbignum=bpi", "-wle", "print bpi(2000)"]
restartPolicy: Never
backoffLimit: 4
```
在建立任務之後，我們可以看到最多隻會有兩個Pod同時執行：
```shell
$ kubectl get pod
NAME READY STATUS RESTARTS AGE
pi-8fsrn 0/1 ContainerCreating 0 30s
pi-job-67kwg 0/1 Completed 0 14h
pi-wlbm5 0/1 ContainerCreating 0 30s
```
每當有一個 Pod 完成計算進入 Completed 狀態時，就會有一個新的 Pod 被自動創建出來，並且快速地從 Pending 狀態進入到 ContainerCreating 狀態。
最終我們可以看到job的COMPLETIONS會標記全部完成：
```shell
$ kubectl get job
NAME COMPLETIONS DURATION AGE
pi 4/4 2m52s 2m52s
```
Job Controller中會會根據配置的併發數來確認當前處於 active 的 pods 數量是否合理，如果不合理的話則進行調整。
如果處於 active 狀態的 pods 數大於 job 設定的併發數 `job.Spec.Parallelism`，則併發刪除多餘的 active pods。
### Job原始碼分析
通過上面的使用例子，我們可以看到job的使用時非常的簡單的，下面我們通過原始碼來理解一下這job的執行邏輯。
核心原始碼位置在job_controller.go中Controller類的syncJob方法中：
syncJob方法很長，我還是想要將這個方法拆開來進行說明。
**Controller#syncJob**
```go
func (jm *Controller) syncJob(key string) (bool, error) {
...
job := *sharedJob
// if job was finished previously, we don't want to redo the termination
// 如果job已經跑完了，那麼直接返回，避免重跑
if IsJobFinished(&job) {
return true, nil
}
// retrieve the previous number of retry
// 獲取job的重試次數
previousRetry := jm.queue.NumRequeues(key)
jobNeedsSync := jm.expectations.SatisfiedExpectations(key)
//獲取這個job的pod列表
pods, err := jm.getPodsForJob(&job)
if err != nil {
return false, err
}
//找到這個job中仍然活躍的pod
activePods := controller.FilterActivePods(pods)
active := int32(len(activePods))
//獲取job中執行成功的pod數和執行失敗的pod數
succeeded, failed := getStatus(pods)
conditions := len(job.Status.Conditions)
// job first start
//設定job 的啟動時間
if job.Status.StartTime == nil {
now := metav1.Now()
job.Status.StartTime = &now
// enqueue a sync to check if job past ActiveDeadlineSeconds
if job.Spec.ActiveDeadlineSeconds != nil {
klog.V(4).Infof("Job %s has ActiveDeadlineSeconds will sync after %d seconds",
key, *job.Spec.ActiveDeadlineSeconds)
jm.queue.AddAfter(key, time.Duration(*job.Spec.ActiveDeadlineSeconds)*time.Second)
}
}
...
}
```
這部分的程式碼會校驗job是否已經跑完了，如果跑完了直接返回；
然後獲取job的重試次數，以及與job關聯的pod列表，並計算出活躍的pod數量、執行成功的pod數量、以及失敗的pod數量；
接下來如果job是首次啟動，那麼需要設定job的啟動時間。
繼續：
```go
func (jm *Controller) syncJob(key string) (bool, error) {
...
var manageJobErr error
jobFailed := false
var failureReason string
var failureMessage string
//failed次數超過了job.Status.Failed說明有新的pod執行失敗了
jobHaveNewFailure := failed > job.Status.Failed
// new failures happen when status does not reflect the failures and active
// is different than parallelism, otherwise the previous controller loop
// failed updating status so even if we pick up failure it is not a new one
//如果有新的pod執行失敗，並且活躍的pod不等於並行Parallelism數
//並且重試次數超過了BackoffLimit
exceedsBackoffLimit := jobHaveNewFailure && (active != *job.Spec.Parallelism) &&
(int32(previousRetry)+1 > *job.Spec.BackoffLimit)
//重試次數是否超標
if exceedsBackoffLimit || pastBackoffLimitOnFailure(&job, pods) {
// check if the number of pod restart exceeds backoff (for restart OnFailure only)
// OR if the number of failed jobs increased since the last syncJob
jobFailed = true
failureReason = "BackoffLimitExceeded"
failureMessage = "Job has reached the specified backoff limit"
// job執行時間是否超過了ActiveDeadlineSeconds
} else if pastActiveDeadline(&job) {
jobFailed = true
failureReason = "DeadlineExceeded"
failureMessage = "Job was active longer than specified deadline"
}
...
}
```
這段程式碼是用來判斷job是否執行失敗，判斷依據是job重試次數是否超過了BackoffLimit，以及job的執行時間是否超過了設定的ActiveDeadlineSeconds。
上面這裡會獲取上一次執行的Failed次數和這次的job的failed次數進行比較，如果failed多了表示又產生了新的執行失敗的pod。如果執行失敗會標識出失敗原因，以及設定jobFailed為true。
在上面的程式碼中呼叫了pastBackoffLimitOnFailure方法和pastActiveDeadline方法，我們分別看一下：
**pastBackoffLimitOnFailure**
```go
func pastBackoffLimitOnFailure(job *batch.Job, pods []*v1.Pod) bool {
//如果RestartPolicy為OnFailure，那麼直接返回
if job.Spec.Template.Spec.RestartPolicy != v1.RestartPolicyOnFailure {
return false
}
result := int32(0)
for i := range pods {
po := pods[i]
//如果pod狀態為Running或Pending
//獲取到pod對應的重啟次數以及Container狀態，包含pod中的InitContainer
if po.Status.Phase == v1.PodRunning || po.Status.Phase == v1.PodPending {
for j := range po.Status.InitContainerStatuses {
stat := po.Status.InitContainerStatuses[j]
result += stat.RestartCount
}
for j := range po.Status.ContainerStatuses {
stat := po.Status.ContainerStatuses[j]
result += stat.RestartCount
}
}
}
//如果BackoffLimit等於，那麼只要重啟了一次，則返回true
if *job.Spec.BackoffLimit == 0 {
return result > 0
}
//比較重啟次數是否超過了BackoffLimit
return result >= *job.Spec.BackoffLimit
}
```
這個方法會校驗job的RestartPolicy策略，不是OnFailure才繼續往下執行。然後會遍歷pod列表，將pod列表中的重啟次數累加並與BackoffLimit進行比較，超過了則返回true。
**pastActiveDeadline**
```go
func pastActiveDeadline(job *batch.Job) bool {
if job.Spec.ActiveDeadlineSeconds == nil || job.Status.StartTime == nil {
return false
}
now := metav1.Now()
start := job.Status.StartTime.Time
duration := now.Time.Sub(start)
allowedDuration := time.Duration(*job.Spec.ActiveDeadlineSeconds) * time.Second
return duration >= allowedDuration
}
```
這個方法會算出job的執行時間duration，然後和ActiveDeadlineSeconds進行比較，如果超過了則返回true。
我們回到syncJob中繼續往下：
```go
func (jm *Controller) syncJob(key string) (bool, error) {
...
//job執行失敗
if jobFailed {
errCh := make(chan error, active)
//將job裡面的active的pod刪除
jm.deleteJobPods(&job, activePods, errCh)
select {
case manageJobErr = <-errCh:
if manageJobErr != nil {
break
}
default:
}
// update status values accordingly
//清空active數
failed += active
active = 0
job.Status.Conditions = append(job.Status.Conditions, newCondition(batch.JobFailed, failureReason, failureMessage))
jm.recorder.Event(&job, v1.EventTypeWarning, failureReason, failureMessage)
} else {
//如果job需要同步，並且job沒有被刪除，則呼叫manageJob進行同步工作
if jobNeedsSync && job.DeletionTimestamp == nil {
active, manageJobErr = jm.manageJob(activePods, succeeded, &job)
}
//完成數等於pod 執行成功的數量
completions := succeeded
complete := false
//如果沒有設定Completions，那麼只要有pod完成，那麼job就算完成
if job.Spec.Completions == nil {
if succeeded >