Kubernetes： Job 和 CronJob 的实现原理

一时失言乱红尘 2023-07-10 15:57 11阅读 0赞

**目录**

Job 和 CronJob 的介绍

Job

Job Spec格式

小技巧：查看Job日志

Job管理模式

Job架构设计

CronJob

CronJob Spec

crontab的语法规则格式：

Job 和 CronJob 的实现原理

概述

任务

同步

并行执行

定时任务

同步

总结

--------------------

Kubernetes 中使用 Job 和 CronJob 两个资源分别提供了一次性任务和定时任务的特性，这两种对象也**使用控制器模型来实现资源的管理**。

# Job 和 CronJob 的介绍 #

> Kubernetes有两个概念跟job有关:
> 
>  *  [Job][]: 负责批量处理短暂的一次性任务，仅执行一次，并保证处理的一个或者多个Pod成功结束。
>  *  [CronJob][]: 负责定时任务，在指定的时间周期运行指定的任务。

## Job ##

Job负责批量处理短暂的一次性任务 (short lived one-off tasks)，即仅执行一次的任务，它保证批处理任务的一个或多个Pod成功结束。

Kubernetes支持以下几种Job：

*  非并行Job：通常创建一个Pod直至其成功结束
 *  固定结束次数的Job：设置.spec.completions，创建多个Pod，直到.spec.completions个Pod成功结束
 *  带有工作队列的并行Job：设置.spec.Parallelism但不设置.spec.completions，当所有Pod结束并且至少一个成功时，Job就认为是成功

根据.spec.completions和.spec.Parallelism的设置，可以将Job划分为以下几种pattern：

<table> 
 <thead> 
  <tr> 
   <th>Job类型</th> 
   <th>使用示例</th> 
   <th>行为</th> 
   <th>completions</th> 
   <th>Parallelism</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>一次性Job</td> 
   <td>数据库迁移</td> 
   <td>创建一个Pod直至其成功结束</td> 
   <td>1</td> 
   <td>1</td> 
  </tr> 
  <tr> 
   <td>固定结束次数的Job</td> 
   <td>处理工作队列的Pod</td> 
   <td>依次创建一个Pod运行直至completions个成功结束</td> 
   <td>2+</td> 
   <td>1</td> 
  </tr> 
  <tr> 
   <td>固定结束次数的并行Job</td> 
   <td>多个Pod同时处理工作队列</td> 
   <td>依次创建多个Pod运行直至completions个成功结束</td> 
   <td>2+</td> 
   <td>2+</td> 
  </tr> 
  <tr> 
   <td>并行Job</td> 
   <td>多个Pod同时处理工作队列</td> 
   <td>创建一个或多个Pod直至有一个成功结束</td> 
   <td>1</td> 
   <td>2+</td> 
  </tr> 
 </tbody> 
</table>

### Job Spec格式 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ_size_16_color_FFFFFF_t_70][] Job 语法

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ_size_16_color_FFFFFF_t_70 1][] 并行运行 Job

>  *  spec.template格式同[Pod][]
>  *  restartPolicy，在 Job 里面我们可以设置 Never、OnFailure、Always 这三种重试策略。在希望 Job 需要重新运行的时候，我们可以用 Never；希望在失败的时候再运行，再重试可以用 OnFailure；或者不论什么情况下都重新运行时 Alway
>  *  单个Pod时，默认Pod成功运行后Job即结束
>  *  .spec.completions标志Job结束需要成功运行的Pod个数，默认为1
>  *  .spec.parallelism标志并行运行的Pod的个数，默认为1
>  *  spec.activeDeadlineSeconds标志失败Pod的重试最大时间，超过这个时间不会继续重试
>  *  restartPolicy，在 Job 里面我们可以设置 Never、OnFailure、Always 这三种重试策略。在希望 Job 需要重新运行的时候，我们可以用 Never；希望在失败的时候再运行，再重试可以用 OnFailure；或者不论什么情况下都重新运行时 Alway
>  *  **backoffLimit** 就是来保证一个 Job 到底能重试多少次。Job 在运行的时候不可能去无限的重试，所以我们需要一个参数来控制重试的次数。
>  *  **ownerReferences**，这个东西来声明此 pod 是归哪个上一层 controller 来管理。可以看到这里的 ownerReferences 是归 batch/v1，也就是上一个 Job 来管理的。这里就声明了它的 controller 是谁，然后可以通过 pod 返查到它的控制器是谁，同时也能根据 Job 来查一下它下属有哪些 Pod。
>  *  .spec.ttlSecondsAfterFinished  自动清除完成的job（`Complete`或`Failed`）

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ_size_16_color_FFFFFF_t_70 2][] 查看Job状态

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ_size_16_color_FFFFFF_t_70 3][]

### 小技巧：查看Job日志 ###

列出属于job的所有Pod，可以使用如下命令：

pods=$(kubectl get pods --selector=job-name=pi --output=jsonpath='{.items[*].metadata.name}') echo $pods

在此，选择器与job的选择器相同。该`--output=jsonpath`选项指定一个表达式，该表达式仅从返回列表中的每个Pod中获取名称。

查看其中一个Pod的标准输出：

kubectl logs $pods

### Job管理模式 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ_size_16_color_FFFFFF_t_70 4][]

### Job架构设计 ###

我们来看一下 job 的架构设计。Job Controller 其实还是主要去创建相对应的 pod，然后 Job Controller 会去跟踪 Job 的状态，及时地根据我们提交的一些配置重试或者继续创建。同时我们刚刚也提到，**每个 pod 会有它对应的 label，来跟踪它所属的 Job Controller（ownerReferences）**，并且还去配置并行的创建， 并行或者串行地去创建 pod。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ_size_16_color_FFFFFF_t_70 5][]

上图是一个 Job 控制器的主要流程。所有的 **job 都是一个 controller，它会 watch 这个 API Server**，我们每次提交一个 Job 的 yaml 都会经过 api-server 传到 ETCD 里面去，然后 **Job Controller 会注册几个 Handler，每当有添加、更新、删除等操作的时候，它会通过一个内存级的消息队列，发到 controller 里面。**

通过 Job Controller 检查当前是否有运行的 pod，如果没有的话，通过 Scale up 把这个 pod 创建出来；如果有的话，或者如果大于这个数，对它进行 Scale down，如果这时 pod 发生了变化，需要及时 Update 它的状态。

同时要去检查它是否是并行的 job，或者是串行的 job，根据设置的配置并行度、串行度，及时地把 pod 的数量给创建出来。最后，它会把 job 的整个的状态更新到 API Server 里面去，这样我们就能看到呈现出来的最终效果了。

## CronJob ##

cronJob是基于时间进行任务的定时管理：

>  *  在特定的时间点运行任务
>  *  反复在指定的时间点运行任务：比如定时进行数据库备份，定时发送电子邮件等等。

### CronJob Spec ###

完整的spec字段，可以参考[CronJob][CronJob 1]，介绍几个主要的字段：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ_size_16_color_FFFFFF_t_70 6][]

*  .spec.schedule: 指定任务运行周期，具体格式参考[Cron - Wikipedia][] （或参考下文**crontab的语法规则格式**）
 *  **startingDeadlineSeconds：**即：每次运行 Job 的时候，它最长可以等多长时间，有时这个 Job 可能运行很长时间也不会启动。所以这时，如果超过较长时间的话，CronJob 就会停止这个 Job；
 *  .spec.concurrencyPolicy: 指定任务的并发策略，参数支持Allow、Forbid和Replace。

> `.spec.concurrencyPolicy` 也是可选的。它声明了 CronJob 创建的任务执行时发生重叠如何处理。spec 仅能声明下列规则中的一种：
> 
>  *  `Allow` (默认)：CronJob 允许并发任务执行。
>  *  `Forbid`： CronJob 不允许并发任务执行；如果新任务的执行时间到了而老任务没有执行完，CronJob 会忽略新任务的执行。
>  *  `Replace`：如果新任务的执行时间到了而老任务没有执行完，CronJob 会用新任务替换当前正在运行的任务。
> 
> 请注意，**并发性规则仅适用于相同 CronJob 创建的任务。如果有多个 CronJob，它们相应的任务总是允许并发执行的。**

*  .spec.jobTemplate: 指定需要运行的任务，格式同[Job][Job 1]。所以其实cronJob是基于Job进行实现。
 *  **`.spec.suspend`:** ` `域也是可选的。如果设置为 `true` ，后续发生的执行都会挂起。这个设置对已经开始的执行不起作用。默认是关闭的。
 *  **`.spec.successfulJobsHistoryLimit` 和 `.spec.failedJobsHistoryLimit`**是可选的。 这两个域声明了有多少执行完成和失败的任务会被保留。 默认设置为3和1。限制设置为0代表相应类型的任务完成后不会保留。

### crontab的语法规则格式： ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ_size_16_color_FFFFFF_t_70 7][]

<table> 
 <thead> 
  <tr> 
   <th>代表意义</th> 
   <th>分钟</th> 
   <th>小时</th> 
   <th>日期</th> 
   <th>月份</th> 
   <th>周</th> 
   <th>命令</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>数字范围</td> 
   <td>0~59</td> 
   <td>0~23</td> 
   <td>1~31</td> 
   <td>1~12</td> 
   <td>0~7</td> 
   <td>需要执行的命令</td> 
  </tr> 
 </tbody> 
</table>

周的数字为 0 或 7 时，都代表“星期天”的意思。

另外，还有一些辅助的字符，大概有下面这些：

<table> 
 <thead> 
  <tr> 
   <th>特殊字符</th> 
   <th>代表意义</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td><span style="color:#f33b45;"><strong>*(星号)</strong></span></td> 
   <td><strong>代表任何时刻都接受的意思</strong>。举例来说，<code>0 12 * * * command</code>&nbsp;日、月、周都是*，就代表着不论何月、何日的礼拜几的12：00都执行后续命令的意思。</td> 
  </tr> 
  <tr> 
   <td><span style="color:#f33b45;"><strong>,(逗号)</strong></span></td> 
   <td><strong>代表分隔时段的意思</strong>。举例来说，如果要执行的工作是3：00与6：00时，就会是：<code>0 3,6 * * * command</code>时间还是有五列，不过第二列是 3,6 ，代表3与6都适用</td> 
  </tr> 
  <tr> 
   <td><span style="color:#f33b45;"><strong>-(减号)</strong></span></td> 
   <td><strong>代表一段时间范围内</strong>，举例来说，8点到12点之间的每小时的20分都进行一项工作：<code>20 8-12 * * * command</code>仔细看到第二列变成8-12.代表 8,9,10,11,12 都适用的意思</td> 
  </tr> 
  <tr> 
   <td><span style="color:#f33b45;"><strong>/n(斜线)</strong></span></td> 
   <td><strong>那个n代表数字，即是每隔n单位间隔的意思，</strong>例如每五分钟进行一次，则：<code>*/5 * * * * command</code>用*与/5来搭配，也可以写成0-59/5，意思相同</td> 
  </tr> 
 </tbody> 
</table>

> **1.每分钟定时执行一次规则：**
> 
>  *  每1分钟执行： \*/1 \* \* \* \*或者\* \* \* \* \*
>  *  每5分钟执行： \*/5 \* \* \* \*
> 
> **2.每小时定时执行一次规则：**
> 
>  *  每小时执行： 0 \* \* \* \*或者0 \*/1 \* \* \*
>  *  每天上午7点执行：0 7 \* \* \*
>  *  每天上午7点10分执行：10 7 \* \* \*
> 
> **3.每天定时执行一次规则：**
> 
>  *  每天执行 0 0 \* \* \*
> 
> **4.每周定时执行一次规则：**
> 
>  *  每周执行 0 0 \* \* 0
> 
> **5.每月定时执行一次规则：**
> 
>  *  每月执行 0 0 1 \* \*
> 
> **6.每年定时执行一次规则：**
> 
>  *  每年执行 0 0 1 1 \*
> 
> **7.其他例子**
> 
>  *  5 \* \* \* \* 指定每小时的第5分钟执行一次ls命令
>  *  30 5 \* \* \* ls 指定每天的 5:30 执行ls命令
>  *  30 7 8 \* \* ls 指定每月8号的7：30分执行ls命令
>  *  30 5 8 6 \* ls 指定每年的6月8日5：30执行ls命令
>  *  30 6 \* \* 0 ls 指定每星期日的6:30执行ls命令 \[注：0表示星期天，1表示星期1，以此类推，也可以用英文来表示，sun表示星期天，mon表示星期一等\]
>  *  30 3 10,20 \* \* ls 每月10号及20号的3：30执行ls命令 \[注：“，”用来连接多个不连续的时段\]
>  *  25 8-11 \* \* \* ls 每天8-11点的第25分钟执行ls命令 \[注：“-”用来连接连续的时段\]
>  *  \*/15 \* \* \* \* ls 每15分钟执行一次ls命令 \[即每个小时的第0 15 30 45 60分钟执行ls命令 \]
>  *  30 6 \*/10 \* \* ls 每个月中，每隔10天6:30执行一次ls命令 \[即每月的1、11、21、31日是的6：30执行一次ls命令 \]

# Job 和 CronJob 的实现原理 #

> **注意：**
> 
> 从集群版本1.8开始，`batch/v2alpha1` API 组中的 CronJob 资源已经被废弃。 你应该切换到 API 服务器默认启用的 `batch/v1beta1` API 组。本文中的所有示例使用了`batch/v1beta1`。

## 概述 ##

Kubernetes 中的 Job 可以创建并且保证一定数量 Pod 的成功停止，当 Job 持有的一个 Pod 对象成功完成任务之后，Job 就会记录这一次 Pod 的成功运行；当一定数量的Pod 的任务执行结束之后，当前的 Job 就会将它自己的状态标记成结束。

![Job-Topology][]

上述图片中展示了一个 `spec.completions=3` 的任务的状态随着 Pod 的成功执行而更新和迁移状态的过程，从图中我们能比较清楚的看到 Job 和 Pod 之间的关系，假设我们有一个用于计算圆周率的如下任务：

apiVersion: batch/v1
    kind: Job
    metadata:
      name: pi
    spec:
      completions: 10
      parallelism: 5
      template:
        spec:
          containers:
          - name: pi
            image: perl
            command: ["perl",  "-Mbignum=bpi", "-wle", "print bpi(2000)"]
          restartPolicy: Never
      backoffLimit: 4

当我们在 Kubernetes 中创建上述任务时，使用 `kubectl` 能够观测到以下的信息：

$ k apply -f job.yaml
    job.batch/pi created
    
    $ kubectl get job --watch
    NAME  COMPLETIONS   DURATION   AGE
    pi    0/10          1s         1s
    pi    1/10          36s        36s
    pi    2/10          46s        46s
    pi    3/10          54s        54s
    pi    4/10          60s        60s
    pi    5/10          65s        65s
    pi    6/10          90s        90s
    pi    7/10          99s        99s
    pi    8/10          104s       104s
    pi    9/10          107s       107s
    pi    10/10         109s       109s

由于任务 `pi` 在配置时指定了 `spec.completions=10`，所以当前的任务需要等待 10 个 Pod 的成功执行，另一个比较重要的 `spec.parallelism=5` 表示最多有多少个并发执行的任务，如果 `spec.parallelism=1` 那么所有的任务都会依次顺序执行，只有前一个任务执行成功时，后一个任务才会开始工作。

![CronJob-Topology][]

**每一个 Job 对象都会持有一个或者多个 Pod，而每一个 CronJob 就会持有多个 Job 对象**，CronJob 能够按照时间对任务进行调度，它与 [crontab][Cron - Wikipedia] 非常相似，我们可以使用 Cron 格式快速指定任务的调度时间：

apiVersion: batch/v1beta1
    kind: CronJob
    metadata:
      name: pi
    spec:
      schedule: "*/1 * * * *"
      jobTemplate:
        spec:
          completions: 2
          parallelism: 1
          template:
            spec:
              containers:
              - name: pi
                image: perl
                command: ["perl",  "-Mbignum=bpi", "-wle", "print bpi(2000)"]
              restartPolicy: OnFailure

上述的 CronJob 对象被创建之后，每分钟都会创建一个新的 Job 对象，**所有的 CronJob 创建的任务都会带有调度时的时间戳**，例如：pi-1551660600 和 pi-1551660660 两个任务：

$ k get cronjob --watch
    NAME   SCHEDULE      SUSPEND   ACTIVE   LAST SCHEDULE   AGE
    pi    */1 * * * *    False     0        <none>          3s
    pi    */1 * * * *    False     1        1s              7s
    
    $ k get job --watch
    NAME            COMPLETIONS   DURATION   AGE
    pi-1551660600   0/3   0s    0s
    pi-1551660600   1/3   16s   16s
    pi-1551660600   2/3   31s   31s
    pi-1551660600   3/3   44s   44s
    pi-1551660660   0/3         1s
    pi-1551660660   0/3   1s    1s
    pi-1551660660   1/3   14s   14s
    pi-1551660660   2/3   28s   28s
    pi-1551660660   3/3   42s   43s

**CronJob 中保存的任务其实是有上限的**，`spec.successfulJobsHistoryLimit` 和 `spec.failedJobsHistoryLimit` 分别记录了能够保存的成功或者失败的任务上限，超过这个上限的任务都会被删除，**默认情况下这两个属性分别为 `spec.successfulJobsHistoryLimit=3` 和 `spec.failedJobsHistoryLimit=1`**。

## 任务 ##

Job 遵循 Kubernetes 的控制器模式进行设计，在发生需要监听的事件时，Informer 就会调用控制器中的回调将需要处理的资源 Job 加入队列，而控制器持有的工作协程就会处理这些任务。

![Job-FlowChart][]

用于处理 Job 资源的 `JobController` 控制器会监听 Pod 和 Job 这两个资源的变更事件，而**资源的同步还是需要运行 `syncJob` 方法**。

### 同步 ###

`syncJob` 是 `JobController` 中主要用于同步 Job 资源的方法，这个方法的主要作用就是对资源进行同步，它的大体架构就是先获取一些当前同步的基本信息，然后调用 `manageJob` 方法管理 Job 对应的 Pod 对象，最后计算出处于 `active`、`failed` 和 `succeed` 三种状态的 Pod 数量并更新 Job 的状态：

func (jm *JobController) syncJob(key string) (bool, error) {
    	ns, name, _ := cache.SplitMetaNamespaceKey(key)
    	sharedJob, _ := jm.jobLister.Jobs(ns).Get(name)
    	job := *sharedJob
    
    	jobNeedsSync := jm.expectations.SatisfiedExpectations(key)
    
    	pods, _ := jm.getPodsForJob(&job)
    
    	activePods := controller.FilterActivePods(pods)
    	active := int32(len(activePods))
    	succeeded, failed := getStatus(pods)
    	conditions := len(job.Status.Conditions)

这一部分代码会从 apiserver 中该名字对应的 Job 对象，然后获取 Job 对应的 Pod 数组并根据计算不同状态 Pod 的数量，为之后状态的更新和比对做准备。

var manageJobErr error
    	if jobNeedsSync && job.DeletionTimestamp == nil {
    		active, manageJobErr = jm.manageJob(activePods, succeeded, &job)
    	}

准备工作完成之后，会调用 `manageJob` 方法，该方法主要负责管理 Job 持有的一系列 Pod 的运行，它最终会返回目前集群中当前 Job 对应的活跃任务的数量。

completions := succeeded
    	complete := false
    	if job.Spec.Completions == nil {
    		if succeeded > 0 && active == 0 {
    			complete = true
    		}
    	} else {
    		if completions >= *job.Spec.Completions {
    			complete = true
    		}
    	}
    	if complete {
    		job.Status.Conditions = append(job.Status.Conditions, newCondition(batch.JobComplete, "", ""))
    		now := metav1.Now()
    		job.Status.CompletionTime = &now
    	}
    
    	if job.Status.Active != active || job.Status.Succeeded != succeeded || job.Status.Failed != failed || len(job.Status.Conditions) != conditions {
    		job.Status.Active = active
    		job.Status.Succeeded = succeeded
    		job.Status.Failed = failed
    
    		jm.updateHandler(&job)
    	}
    
    	return forget, manageJobErr
    }

最后的这段代码会将 Job 规格中设置的 `spec.completions` 和已经完成的任务数量进行比对，确认当前的 Job 是否已经结束运行，如果任务已经结束运行就会更新当前 Job 的完成时间，同时当 `JobController` 发现有一些状态没有正确同步时，也会调用 `updateHandler` 更新资源的状态。

### 并行执行 ###

**Pod 的创建和删除都是由 `manageJob` 这个方法负责的**，这个方法根据 Job 的 `spec.parallelism` 配置对目前集群中的节点进行创建和删除。

![Job-Kill-More-Pods][]

如果当前正在执行的活跃节点数量超过了 `spec.parallelism`，那么就会按照创建的升删除多余的任务，删除任务时会使用 `DeletePod` 方法：

func (jm *JobController) manageJob(activePods []*v1.Pod, succeeded int32, job *batch.Job) (int32, error) {
    	active := int32(len(activePods))
    	parallelism := *job.Spec.Parallelism
    
    	if active > parallelism {
    		diff := active - parallelism
    		sort.Sort(controller.ActivePods(activePods))
    
    		active -= diff
    		wait := sync.WaitGroup{}
    		wait.Add(int(diff))
    		for i := int32(0); i < diff; i++ {
    			go func(ix int32) {
    				defer wait.Done()
    				jm.podControl.DeletePod(job.Namespace, activePods[ix].Name, job)
    			}(i)
    		}
    		wait.Wait()

当正在活跃的节点数量小于 `spec.parallelism` 时，我们就会根据当前未完成的任务数和并行度计算出最大可以处于活跃的 Pod 个数 `wantActive` 以及与当前的活跃 Pod 数相差的 `diff`：

} else if active < parallelism {
    		wantActive := int32(0)
    		if job.Spec.Completions == nil {
    			if succeeded > 0 {
    				wantActive = active
    			} else {
    				wantActive = parallelism
    			}
    		} else {
    			wantActive = *job.Spec.Completions - succeeded
    			if wantActive > parallelism {
    				wantActive = parallelism
    			}
    		}
    		diff := wantActive - active
    		if diff < 0 {
    			diff = 0
    		}
    
    		active += diff

在方法的最后就会以批量的方式并行创建 Pod，所有 Pod 的创建都是通过 `CreatePodsWithControllerRef` 方法在 Goroutine 中执行的。

![Job-Create-More-Pods][]

这里会使用 `WaitGroup` 等待每个 Batch 中创建的结果返回才会执行下一个 Batch，**Batch 的大小是从 1 开始指数增加的，以冷启动的方式避免首次创建的任务过多造成失败：**

wait := sync.WaitGroup{}
    		for batchSize := int32(integer.IntMin(int(diff), controller.SlowStartInitialBatchSize)); diff > 0; batchSize = integer.Int32Min(2*batchSize, diff) {
    			wait.Add(int(batchSize))
    			for i := int32(0); i < batchSize; i++ {
    				go func() {
    					defer wait.Done()
    					jm.podControl.CreatePodsWithControllerRef(job.Namespace, &job.Spec.Template, job, metav1.NewControllerRef(job, controllerKind))
    				}()
    			}
    			wait.Wait()
    			diff -= batchSize
    		}
    	}
    
    	return active, nil
    }

通过对 `manageJob` 的分析，我们其实能够看出这个方法就是根据规格中的配置对 Pod 进行管理，它在较多并行时删除 Pod，较少并行时创建 Pod，也算是一个简单的资源利用和调度机制，代码非常直白并且容易理解，不需要花太大的篇幅。

## 定时任务 ##

用于管理 CronJob 资源的 `CronJobController` 虽然也使用了控制器模式，但是它的实现与其他的控制器不太一样，他**没有从 Informer 中接受其他消息变动的通知，而是直接访问 apiserver 中的数据**：

![CronJobController-FlowChart][]

从 apiserver 中获取了 Job 和 CronJob 的信息之后就会调用 `JobControl` 向 apiserver 发起请求创建新的 Job 资源，这一个过程都是由 `CronJobController` 的 `syncAll` 方法驱动的，我们接下来就来介绍这一方法的实现。

### 同步 ###

`syncAll` 方法会从 apiserver 中取出所有的 Job 和 CronJob 对象，然后通过 `groupJobsByParent` 将任务按照 `spec.ownerReferences` 进行分类并遍历去同步所有的 CronJob：

func (jm *CronJobController) syncAll() {
    	jl, _ := jm.kubeClient.BatchV1().Jobs(metav1.NamespaceAll).List(metav1.ListOptions{})
    	js := jl.Items
    
    	sjl, _ := jm.kubeClient.BatchV1beta1().CronJobs(metav1.NamespaceAll).List(metav1.ListOptions{})
    	sjs := sjl.Items
    
    	jobsBySj := groupJobsByParent(js)
    
    	for _, sj := range sjs {
    		syncOne(&sj, jobsBySj[sj.UID], time.Now(), jm.jobControl, jm.sjControl, jm.recorder)
    		cleanupFinishedJobs(&sj, jobsBySj[sj.UID], jm.jobControl, jm.sjControl, jm.recorder)
    	}
    }

`syncOne` 就是用于同步单个 CronJob 对象的方法，这个方法会首先遍历全部的 Job 对象，只保留正在运行的活跃对象并更新 CronJob 的状态：

func syncOne(sj *batchv1beta1.CronJob, js []batchv1.Job, now time.Time, jc jobControlInterface, sjc sjControlInterface, recorder record.EventRecorder) {
    	childrenJobs := make(map[types.UID]bool)
    	for _, j := range js {
    		childrenJobs[j.ObjectMeta.UID] = true
    		found := inActiveList(*sj, j.ObjectMeta.UID)
    		if found && IsJobFinished(&j) {
    			deleteFromActiveList(sj, j.ObjectMeta.UID)
    		}
    	}
    
    	for _, j := range sj.Status.Active {
    		if found := childrenJobs[j.UID]; !found {
    			deleteFromActiveList(sj, j.UID)
    		}
    	}
    
    	updatedSJ, _ := sjc.UpdateStatus(sj)
    	*sj = *updatedSJ

随后的 `getRecentUnmetScheduleTimes` 方法会根据 CronJob 的调度配置 `spec.schedule` 和上一次执行任务的时间计算出我们缺失的任务次数。

times, _ := getRecentUnmetScheduleTimes(*sj, now)
    	if len(times) == 0 {
    		return
    	}
    
    	scheduledTime := times[len(times)-1]
    	if sj.Spec.ConcurrencyPolicy == batchv1beta1.ForbidConcurrent && len(sj.Status.Active) > 0 {
    		return
    	}
    	if sj.Spec.ConcurrencyPolicy == batchv1beta1.ReplaceConcurrent {
    		for _, j := range sj.Status.Active {
    			job, _ := jc.GetJob(j.Namespace, j.Name)
    			if !deleteJob(sj, job, jc, recorder) {
    				return
    			}
    		}
    	}

如果现在需要调度新的任务，但是当前已经存在活跃的任务，就会根据并发策略的配置执行不同的操作：

> 1.  使用 `ForbidConcurrent` 策略跳过这一次任务的调度直接返回；
> 2.  使用 `ReplaceConcurrent` 策略获取并删除全部活跃的任务，通过创建新的 Pod 替换这些正在执行的活跃 Pod；

jobReq, _ := getJobFromTemplate(sj, scheduledTime)
    	jobResp, _ := jc.CreateJob(sj.Namespace, jobReq)
    
    	ref,_ := getRef(jobResp)
    	sj.Status.Active = append(sj.Status.Active, *ref)
    	sj.Status.LastScheduleTime = &metav1.Time{Time: scheduledTime}
    	sjc.UpdateStatus(sj)
    
    	return
    }

在方法的最后，它会从 CronJob 的 `spec.jobTemplate` 中拿到创建 Job 使用的模板并调用 `JobControl` 的 `CreateJob` 向 apiserver 发起创建任务的 HTTP 请求，接下来的操作就都是由 `JobController` 负责了。

## 总结 ##

>  *  Job 作为 Kubernetes 中用于处理任务的资源，与其他的资源没有太多的区别，它也使用 Kubernetes 中常见的控制器模式，监听 Informer 中的事件并运行 `syncHandler` 同步任务
>  *  而 CronJob 由于其功能的特殊性，每隔 10s 会从 apiserver 中取出资源并进行检查是否应该触发调度创建新的资源，需要注意的是 **CronJob 并不能保证在准确的目标时间执行，执行会有一定程度的滞后**。
>  *  两个控制器的实现都比较清晰，只是边界条件比较多，分析其实现原理时一定要多注意。

参考链接

[https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.13/\#cronjob-v1beta1-batch][https_kubernetes.io_docs_reference_generated_kubernetes-api_v1.13_cronjob-v1beta1-batch]

[https://kubernetes.io/zh/docs/concepts/workloads/controllers/cron-jobs/][https_kubernetes.io_zh_docs_concepts_workloads_controllers_cron-jobs]

[https://kubernetes.io/docs/concepts/workloads/controllers/jobs-run-to-completion/][Job]

[Job]: https://kubernetes.io/docs/concepts/workloads/controllers/jobs-run-to-completion/
[CronJob]: https://kubernetes.io/docs/concepts/workloads/controllers/cron-jobs/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200305082744103.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200305082932362.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ==,size_16,color_FFFFFF,t_70
[Pod]: https://www.kubernetes.org.cn/kubernetes-pod
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200305083032137.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20200305083119114.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20200304090315158.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20200304090358740.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ==,size_16,color_FFFFFF,t_70
[CronJob 1]: https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.10/#cronjob-v1beta1-batch
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/20200305083235674.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ==,size_16,color_FFFFFF,t_70
[Cron - Wikipedia]: https://en.wikipedia.org/wiki/Cron
[Job 1]: https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.10/#job-v1-batch
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ_size_16_color_FFFFFF_t_70 7]: https://img-blog.csdnimg.cn/20200305081349584.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ==,size_16,color_FFFFFF,t_70
[Job-Topology]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9pbWcuZHJhdmVuZXNzLm1lLzIwMTktMDMtMDUtSm9iLVRvcG9sb2d5LnBuZw?x-oss-process=image/format,png
[CronJob-Topology]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9pbWcuZHJhdmVuZXNzLm1lLzIwMTktMDMtMDUtQ3JvbkpvYi1Ub3BvbG9neS5wbmc?x-oss-process=image/format,png
[Job-FlowChart]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9pbWcuZHJhdmVuZXNzLm1lLzIwMTktMDMtMDUtSm9iLUZsb3dDaGFydC5wbmc?x-oss-process=image/format,png
[Job-Kill-More-Pods]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9pbWcuZHJhdmVuZXNzLm1lLzIwMTktMDMtMDUtSm9iLUtpbGwtTW9yZS1Qb2RzLnBuZw?x-oss-process=image/format,png
[Job-Create-More-Pods]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9pbWcuZHJhdmVuZXNzLm1lLzIwMTktMDMtMDUtSm9iLUNyZWF0ZS1Nb3JlLVBvZHMucG5n?x-oss-process=image/format,png
[CronJobController-FlowChart]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9pbWcuZHJhdmVuZXNzLm1lLzIwMTktMDMtMDUtQ3JvbkpvYkNvbnRyb2xsZXItRmxvd0NoYXJ0LnBuZw?x-oss-process=image/format,png
[https_kubernetes.io_docs_reference_generated_kubernetes-api_v1.13_cronjob-v1beta1-batch]: https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.13/#cronjob-v1beta1-batch
[https_kubernetes.io_zh_docs_concepts_workloads_controllers_cron-jobs]: https://kubernetes.io/zh/docs/concepts/workloads/controllers/cron-jobs/