Skip to content
rouroumaibing

Job、CronJob 逻辑结构分析

• CNCF • 55 分钟阅读

一、概述

在 Kubernetes 中，Job 和 CronJob 是用于管理任务的两种控制器，它们分别用于处理一次性任务和周期性任务：

  • Job：用于管理一次性任务，确保指定数量的 Pod 成功完成任务。当任务完成后，Job 会保持其状态，不会自动删除。
  • CronJob：基于时间调度的 Job 控制器，允许用户按照指定的时间表达式（cron 表达式）定期创建 Job。

这两种控制器为 Kubernetes 集群中的任务管理提供了灵活的解决方案，满足了不同场景下的任务执行需求。

Job 的主要特点：
  1. 一次性任务：Job 管理的是短期运行的一次性任务
  2. 并行执行：支持设置并行度（parallelism）控制同时运行的 Pod 数量
  3. 完成计数：支持设置完成数（completions）控制需要成功完成的 Pod 数量
  4. 失败处理：支持设置失败重试次数（backoffLimit）控制任务失败后的重试策略
  5. 活跃期限：支持设置活跃期限（activeDeadlineSeconds）控制任务的最大运行时间
CronJob 的主要特点：
  1. 时间调度：使用标准的 cron 表达式定义任务执行时间
  2. 自动创建 Job：按照调度时间自动创建 Job 实例
  3. 并发控制：支持设置并发策略（Forbid、Replace、Allow）控制任务的并发执行
  4. 历史管理：支持设置成功和失败的 Job 历史记录数量
  5. 时区支持：支持设置时区，确保任务在正确的时区执行

二、工作原理

2.1 Job 工作原理

Job 工作原理图

2.2 CronJob 工作原理

CronJob 工作原理图

三、创建流程

Job 与 CronJob 创建流程图

3.1 Job 创建流程

  1. 用户请求：用户通过 kubectl 客户端发起创建 Job 资源请求
  2. API Server 处理：API Server 对请求进行鉴权、准入控制，然后将请求写入 etcd
  3. Job 控制器监听：Job 控制器通过 Informer 机制监听 Job 资源变化
  4. Pod 创建：控制器根据 Job 的配置创建 Pod
  5. Kubelet 处理：节点上的 Kubelet 组件监听到 Pod 创建事件，在本地运行 Pod
  6. 任务执行：Pod 执行指定的任务
  7. 状态更新：控制器根据 Pod 的执行结果更新 Job 的状态

3.2 CronJob 创建流程

  1. 用户请求：用户通过 kubectl 客户端发起创建 CronJob 资源请求
  2. API Server 处理：API Server 对请求进行鉴权、准入控制，然后将请求写入 etcd
  3. CronJob 控制器监听：CronJob 控制器通过 Informer 机制监听 CronJob 资源变化
  4. 时间计算：控制器计算下一次应该执行任务的时间
  5. Job 创建：在调度时间到达时，控制器创建 Job 实例
  6. Job 执行：Job 控制器接管并执行具体的任务
  7. 历史管理：控制器管理 Job 的历史记录，清理超过限制的历史 Job
  8. 状态更新：更新 CronJob 的状态信息

四、配置

4.1 Job Setting

4.1.1 Parallel Pods

Job 支持通过 spec.parallelism 设置并行执行的 Pod 数量：

  • 未设置：默认值为 1，即一次只运行一个 Pod
  • 设置为 0：Job 会被暂停，直到 parallelism 被设置为大于 0 的值
  • 设置为大于 0：控制器会同时创建指定数量的 Pod 运行任务

代码实现：

在 manageJob 函数中，控制器根据 spec.parallelism 和 spec.completions 计算需要创建的 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
# pkg/controller/job/job_controller.go
func (jm *Controller) manageJob(ctx context.Context, job *batch.Job, activePods []*v1.Pod, succeeded int32, succeededIndexes []interval) (int32, string, error) {
    active := int32(len(activePods))
    parallelism := *job.Spec.Parallelism
  
    if jobSuspended(job) {
        // 处理暂停状态的 Job
        return active, metrics.JobSyncActionPodsDeleted, err
    }

    wantActive := int32(0)
    if job.Spec.Completions == nil {
        // 未指定 completions 的 Job
        if succeeded > 0 {
            wantActive = active
        } else {
            wantActive = parallelism
        }
    } else {
        // 指定了 completions 的 Job
        wantActive = *job.Spec.Completions - succeeded
        if wantActive > parallelism {
            wantActive = parallelism
        }
        if wantActive < 0 {
            wantActive = 0
        }
    }

    // 创建或删除 Pod 以达到期望的数量
    if active < wantActive {
        diff := wantActive - active
        // 批量创建 Pod
        jm.expectations.ExpectCreations(jobKey, int(diff))
        // ... 创建 Pod 的逻辑
    }
  
    return active, metrics.JobSyncActionTracking, nil
}

关键逻辑：

  • 对于未指定 completions 的 Job，并行度直接等于 parallelism
  • 对于指定了 completions 的 Job，并行度不能超过剩余需要完成的 Pod 数量
  • 控制器会批量创建 Pod，每次最多创建 MaxPodCreateDeletePerSync 个
4.1.2. Completions Mode

Job 支持两种完成策略：

  • 非索引 Job (Non-indexed)：当 spec.completions 个 Pod 成功完成时，Job 完成
  • 索引 Job (Indexed)：当 spec.completions 个索引 Pod 成功完成时，Job 完成，每个索引只需要成功完成一次

代码实现：

在 syncJob 函数中，控制器判断 Job 是否完成：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
# pkg/controller/job/job_controller.go
func (jm *Controller) syncJob(ctx context.Context, key string) (forget bool, rErr error) {
    // ... 前面的代码
  
    var finishedCondition *batch.JobCondition
  
    // ... 失败处理逻辑
  
    complete := false
    if job.Spec.Completions == nil {
        // 未指定 completions 的 Job：任一 Pod 成功完成即完成
        complete = succeeded > 0 && active == 0
    } else {
        // 指定了 completions 的 Job：需要完成指定数量的 Pod
        complete = succeeded >= *job.Spec.Completions && active == 0
    }
    if complete {
        finishedCondition = newCondition(batch.JobComplete, v1.ConditionTrue, "", "")
    }
  
    // ... 后面的代码
}

索引 Job 的处理：

在 trackJobStatusAndRemoveFinalizers 函数中，控制器处理索引 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
32
33
34
35
36
37
38
39
40
41
42
43
# pkg/controller/job/job_controller.go
func (jm *Controller) trackJobStatusAndRemoveFinalizers(ctx context.Context, job *batch.Job, pods []*v1.Pod, succeededIndexes orderedIntervals, uncounted uncountedTerminatedPods, expectedRmFinalizers sets.String, finishedCond *batch.JobCondition, needsFlush bool) error {
    isIndexed := isIndexedJob(job)
    var newSucceededIndexes []int
    if isIndexed {
        // 按索引排序 Pod
        sort.Sort(byCompletionIndex(pods))
    }
  
    // ... 中间的代码
  
    for _, pod := range pods {
        // ... 中间的代码
        if pod.Status.Phase == v1.PodSucceeded && !uncounted.failed.Has(string(pod.UID)) {
            if isIndexed {
                // 索引 Job：记录成功的索引
                ix := getCompletionIndex(pod.Annotations)
                if ix != unknownCompletionIndex && ix < int(*job.Spec.Completions) && !succeededIndexes.has(ix) {
                    newSucceededIndexes = append(newSucceededIndexes, ix)
                    needsFlush = true
                }
            } else if !uncounted.succeeded.Has(string(pod.UID)) {
                // 非索引 Job：记录成功的 Pod
                needsFlush = true
                uncountedStatus.Succeeded = append(uncountedStatus.Succeeded, pod.UID)
            }
        }
        // ... 中间的代码
    }
  
    if isIndexed {
        // 更新索引 Job 的完成状态
        succeededIndexes = succeededIndexes.withOrderedIndexes(newSucceededIndexes)
        succeededIndexesStr := succeededIndexes.String()
        if succeededIndexesStr != job.Status.CompletedIndexes {
            needsFlush = true
        }
        job.Status.Succeeded = int32(succeededIndexes.total())
        job.Status.CompletedIndexes = succeededIndexesStr
    }
  
    // ... 后面的代码
}

关键逻辑：

  • 非索引 Job：通过统计成功完成的 Pod 数量来判断是否完成
  • 索引 Job：通过记录成功完成的索引来判断是否完成，每个索引只需要成功一次
  • 索引 Job 使用 status.completedIndexes 字段记录已完成的索引
4.1.3 backoffLimit

Job 支持通过 spec.backoffLimit 设置失败重试次数：

  • 未设置：默认值为 6，即任务失败后会重试 6 次
  • 设置为 0：任务失败后不会重试
  • 设置为大于 0：任务失败后会重试指定的次数

代码实现：

在 syncJob 函数中，控制器检查失败次数是否超过限制：

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
# pkg/controller/job/job_controller.go
func (jm *Controller) syncJob(ctx context.Context, key string) (forget bool, rErr error) {
    // ... 前面的代码
  
    var finishedCondition *batch.JobCondition
  
    jobHasNewFailure := failed > job.Status.Failed
    exceedsBackoffLimit := job.Spec.BackoffLimit != nil && failed > *job.Spec.BackoffLimit

    if exceedsBackoffLimit || pastBackoffLimitOnFailure(&job, pods) {
        // 检查是否超过 backoffLimit（适用于 restartPolicy == OnFailure 的情况）
        finishedCondition = newCondition(batch.JobFailed, v1.ConditionTrue, "BackoffLimitExceeded", "Job has reached the specified backoff limit")
    } else if pastActiveDeadline(&job) {
        // 检查是否超过活跃期限
        finishedCondition = newCondition(batch.JobFailed, v1.ConditionTrue, "DeadlineExceeded", "Job was active longer than specified deadline")
    }
  
    // ... 后面的代码
  
    if jobHasNewFailure && !jobFinished {
        // 返回错误将在退避期后重新入队 Job
        return forget, fmt.Errorf("failed pod(s) detected for job key %q", key)
    }
  
    // ... 后面的代码
}

pastBackoffLimitOnFailure 函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# pkg/controller/job/job_controller.go
func pastBackoffLimitOnFailure(job *batch.Job, pods []*v1.Pod) bool {
    if job.Spec.Template.Spec.RestartPolicy != v1.RestartPolicyOnFailure {
        return false
    }
    result := int32(0)
    for i := range pods {
        po := pods[i]
        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
            }
        }
    }
    if *job.Spec.BackoffLimit == 0 {
        return result > 0
    }
    return result >= *job.Spec.BackoffLimit
}

关键逻辑：

  • 控制器会检查失败的 Pod 数量是否超过 backoffLimit
  • 对于 restartPolicy == OnFailure 的情况，控制器会累加容器重启次数
  • 当超过失败重试次数时，Job 会被标记为失败状态
4.1.4. activeDeadlineSeconds

Job 支持通过 spec.activeDeadlineSeconds 设置任务的最大运行时间：

  • 未设置(null)：任务可以一直运行，直到完成或失败重试次数耗尽
  • 设置为大于 0：任务运行超过指定时间后，控制器会终止所有活跃的 Pod，并将 Job 标记为失败

代码实现：

pastActiveDeadline 函数：

1
2
3
4
5
6
7
8
9
10
11
# pkg/controller/job/job_controller.go
func pastActiveDeadline(job *batch.Job) bool {
    if job.Spec.ActiveDeadlineSeconds == nil || job.Status.StartTime == nil || jobSuspended(job) {
        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
}

在 syncJob 函数中的使用：

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
func (jm *Controller) syncJob(ctx context.Context, key string) (forget bool, rErr error) {
    // ... 前面的代码
  
    var finishedCondition *batch.JobCondition
  
    if exceedsBackoffLimit || pastBackoffLimitOnFailure(&job, pods) {
        // 检查是否超过 backoffLimit
        finishedCondition = newCondition(batch.JobFailed, v1.ConditionTrue, "BackoffLimitExceeded", "Job has reached the specified backoff limit")
    } else if pastActiveDeadline(&job) {
        // 检查是否超过活跃期限
        finishedCondition = newCondition(batch.JobFailed, v1.ConditionTrue, "DeadlineExceeded", "Job was active longer than specified deadline")
    } else if job.Spec.ActiveDeadlineSeconds != nil && !jobSuspended(&job) {
        // 为活跃期限设置下次同步时间
        syncDuration := time.Duration(*job.Spec.ActiveDeadlineSeconds)*time.Second - time.Since(job.Status.StartTime.Time)
        klog.V(2).InfoS("Job has activeDeadlineSeconds configuration. Will sync this job again", "job", key, "nextSyncIn", syncDuration)
        jm.queue.AddAfter(key, syncDuration)
    }
  
    // 如果超过活跃期限，删除所有活跃的 Pod
    if finishedCondition != nil {
        deleted, err := jm.deleteActivePods(ctx, &job, activePods)
        if uncounted == nil {
            // 旧行为：假设所有活跃 Pod 都已成功删除
            deleted = active
        } else if deleted != active || !satisfiedExpectations {
            // 不能声明 Job 已完成，因为可能还有剩余的 Pod finalizers 或尚未加入 informer 缓存的 Pod
            finishedCondition = nil
        }
        active -= deleted
        failed += deleted
        manageJobErr = err
    }
  
    // ... 后面的代码
}

关键逻辑：

  • 控制器会计算 Job 从开始到现在的运行时间
  • 如果超过 activeDeadlineSeconds 指定的时间，控制器会终止所有活跃的 Pod
  • 控制器会将 Job 标记为失败，并设置 DeadlineExceeded 条件
  • 对于设置了活跃期限的 Job，控制器会提前安排下次同步，以确保及时检查是否超过期限

4.2 CronJob Setting

4.2.1 Cron 表达式解析

CronJob 使用标准的 cron 表达式定义任务执行时间，格式为：

1
2
3
4
5
6
7
┌───────────── 分钟 (0-59)
│ ┌────────── 小时 (0-23)
│ │ ┌──────── 日 (1-31)
│ │ │ ┌────── 月 (1-12)
│ │ │ │ ┌──── 星期 (0-6) (星期日=0)
│ │ │ │ │
* * * * *

控制器使用 cron.ParseStandard 函数解析 cron 表达式，并计算下一次执行时间。

4.2.2 timeZone

CronJob 支持通过 spec.timeZone 设置时区，确保任务在正确的时区执行：

  • 未设置：使用控制器所在环境的时区
  • 设置为有效时区：使用指定的时区计算执行时间
  • 设置为无效时区：控制器会记录警告事件，任务不会执行

代码实现：

在 syncCronJob 函数中：

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
# pkg/controller/cronjob/cronjob_controllerv2.go
func (jm *ControllerV2) syncCronJob(ctx context.Context, cronJob *batchv1.CronJob, jobs []*batchv1.Job) (*batchv1.CronJob, *time.Duration, bool, error) {
    // ... 前面的代码
  
    timeZoneEnabled := utilfeature.DefaultFeatureGate.Enabled(features.CronJobTimeZone)
  
    // 检查时区是否有效
    if timeZoneEnabled && cronJob.Spec.TimeZone != nil {
        if _, err := time.LoadLocation(*cronJob.Spec.TimeZone); err != nil {
            timeZone := pointer.StringDeref(cronJob.Spec.TimeZone, "")
            klog.V(4).InfoS("Not starting job because timeZone is invalid", "cronjob", klog.KRef(cronJob.GetNamespace(), cronJob.GetName()), "timeZone", timeZone, "err", err)
            jm.recorder.Eventf(cronJob, corev1.EventTypeWarning, "UnknownTimeZone", "invalid timeZone: %q: %s", timeZone, err)
            return cronJob, nil, updateStatus, nil
        }
    }
  
    // 解析调度表达式，包含时区信息
    sched, err := cron.ParseStandard(formatSchedule(timeZoneEnabled, cronJob, jm.recorder))
    if err != nil {
        // 解析失败，记录警告事件
        klog.V(2).InfoS("Unparseable schedule", "cronjob", klog.KRef(cronJob.GetNamespace(), cronJob.GetName()), "schedule", cronJob.Spec.Schedule, "err", err)
        jm.recorder.Eventf(cronJob, corev1.EventTypeWarning, "UnparseableSchedule", "unparseable schedule: %q : %s", cronJob.Spec.Schedule, err)
        return cronJob, nil, updateStatus, nil
    }
  
    // ... 后面的代码
}

formatSchedule 函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# pkg/controller/cronjob/cronjob_controllerv2.go
func formatSchedule(timeZoneEnabled bool, cj *batchv1.CronJob, recorder record.EventRecorder) string {
    if strings.Contains(cj.Spec.Schedule, "TZ") {
        if recorder != nil {
            recorder.Eventf(cj, corev1.EventTypeWarning, "UnsupportedSchedule", "CRON_TZ or TZ used in schedule %q is not officially supported, see https://kubernetes.io/docs/concepts/workloads/controllers/cron-jobs/ for more details", cj.Spec.Schedule)
        }
        return cj.Spec.Schedule
    }

    if timeZoneEnabled && cj.Spec.TimeZone != nil {
        if _, err := time.LoadLocation(*cj.Spec.TimeZone); err != nil {
            return cj.Spec.Schedule
        }
        return fmt.Sprintf("TZ=%s %s", *cj.Spec.TimeZone, cj.Spec.Schedule)
    }

    return cj.Spec.Schedule
}

关键逻辑：

  • 控制器会检查 spec.timeZone 是否有效
  • 如果时区无效，控制器会记录警告事件并停止执行任务
  • 有效的时区会被添加到 cron 表达式中，格式为 TZ=时区 cron表达式
  • 解析后的调度表达式会用于计算下次执行时间
4.2.3. concurrencyPolicy

CronJob 支持三种并发策略：

  • "Allow" (默认)：允许 CronJob 并发运行；
  • "Forbid"：禁止并发运行，如果上一次运行尚未完成则跳过下一次运行；
  • "Replace"：取消当前正在运行的作业并将其替换为新作业。

关键逻辑：

  • Allow 策略：直接创建新的 Job，不做任何特殊处理
  • Forbid 策略：如果有活跃的 Job，直接返回，不创建新的
  • Replace 策略：删除所有活跃的 Job，然后创建新的

代码实现：

在 syncCronJob 函数中：

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
# pkg/controller/cronjob/cronjob_controllerv2.go
func (jm *ControllerV2) syncCronJob(ctx context.Context, cronJob *batchv1.CronJob, jobs []*batchv1.Job) (*batchv1.CronJob, *time.Duration, bool, error) {
    // ... 前面的代码
  
    // 检查是否错过启动期限
    tooLate := false
    if cronJob.Spec.StartingDeadlineSeconds != nil {
        tooLate = scheduledTime.Add(time.Second * time.Duration(*cronJob.Spec.StartingDeadlineSeconds)).Before(now)
    }
    if tooLate {
        // 错过启动期限，记录警告事件
        klog.V(4).InfoS("Missed starting window", "cronjob", klog.KRef(cronJob.GetNamespace(), cronJob.GetName()))
        jm.recorder.Eventf(cronJob, corev1.EventTypeWarning, "MissSchedule", "Missed scheduled time to start a job: %s", scheduledTime.UTC().Format(time.RFC1123Z))
        t := nextScheduledTimeDuration(sched, now)
        return cronJob, t, updateStatus, nil
    }
  
    // 检查是否已经处理过该调度时间
    if isJobInActiveList(&batchv1.Job{
        ObjectMeta: metav1.ObjectMeta{
            Name:      getJobName(cronJob, *scheduledTime),
            Namespace: cronJob.Namespace,
        }}, cronJob.Status.Active) || cronJob.Status.LastScheduleTime.Equal(&metav1.Time{Time: *scheduledTime}) {
        klog.V(4).InfoS("Not starting job because the scheduled time is already processed", "cronjob", klog.KRef(cronJob.GetNamespace(), cronJob.GetName()), "schedule", scheduledTime)
        t := nextScheduledTimeDuration(sched, now)
        return cronJob, t, updateStatus, nil
    }
  
    // Forbid 策略：如果有活跃的 Job，不创建新的
    if cronJob.Spec.ConcurrencyPolicy == batchv1.ForbidConcurrent && len(cronJob.Status.Active) > 0 {
        klog.V(4).InfoS("Not starting job because prior execution is still running and concurrency policy is Forbid", "cronjob", klog.KRef(cronJob.GetNamespace(), cronJob.GetName()))
        jm.recorder.Eventf(cronJob, corev1.EventTypeNormal, "JobAlreadyActive", "Not starting job because prior execution is running and concurrency policy is Forbid")
        t := nextScheduledTimeDuration(sched, now)
        return cronJob, t, updateStatus, nil
    }
  
    // Replace 策略：终止正在运行的 Job，创建新的
    if cronJob.Spec.ConcurrencyPolicy == batchv1.ReplaceConcurrent {
        for _, j := range cronJob.Status.Active {
            klog.V(4).InfoS("Deleting job that was still running at next scheduled start time", "job", klog.KRef(j.Namespace, j.Name))

            job, err := jm.jobControl.GetJob(j.Namespace, j.Name)
            if err != nil {
                jm.recorder.Eventf(cronJob, corev1.EventTypeWarning, "FailedGet", "Get job: %v", err)
                return cronJob, nil, updateStatus, err
            }
            if !deleteJob(cronJob, job, jm.jobControl, jm.recorder) {
                return cronJob, nil, updateStatus, fmt.Errorf("could not replace job %s/%s", job.Namespace, job.Name)
            }
            updateStatus = true
        }
    }
  
    // Allow 策略：直接创建新的 Job，允许并发执行
    // ... 创建 Job 的逻辑
  
    return cronJob, t, updateStatus, nil
}

deleteJob 函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# pkg/controller/cronjob/cronjob_controllerv2.go
func deleteJob(cj *batchv1.CronJob, job *batchv1.Job, jc jobControlInterface, recorder record.EventRecorder) bool {
    nameForLog := fmt.Sprintf("%s/%s", cj.Namespace, cj.Name)

    // 删除 Job 本身...
    if err := jc.DeleteJob(job.Namespace, job.Name); err != nil {
        recorder.Eventf(cj, corev1.EventTypeWarning, "FailedDelete", "Deleted job: %v", err)
        klog.Errorf("Error deleting job %s from %s: %v", job.Name, nameForLog, err)
        return false
    }
    // ... 并从 active 列表中删除其引用
    deleteFromActiveList(cj, job.ObjectMeta.UID)
    recorder.Eventf(cj, corev1.EventTypeNormal, "SuccessfulDelete", "Deleted job %v", job.Name)

    return true
}
4.2.4. startingDeadlineSeconds

CronJob 支持通过 spec.startingDeadlineSeconds 设置作业的启动截止时间：

  • 未设置：任务可以在任何时间启动
  • 设置为大于 0：如果任务错过调度时间超过指定秒数，任务不会启动，视为失败的作业

代码实现：

在 syncCronJob 函数中：

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
# pkg/controller/cronjob/cronjob_controllerv2.go
func (jm *ControllerV2) syncCronJob(ctx context.Context, cronJob *batchv1.CronJob, jobs []*batchv1.Job) (*batchv1.CronJob, *time.Duration, bool, error) {
    // ... 前面的代码
  
    // 获取下次调度时间
    scheduledTime, err := getNextScheduleTime(*cronJob, now, sched, jm.recorder)
    if err != nil {
        klog.V(2).InfoS("invalid schedule", "cronjob", klog.KRef(cronJob.GetNamespace(), cronJob.GetName()), "schedule", cronJob.Spec.Schedule, "err", err)
        jm.recorder.Eventf(cronJob, corev1.EventTypeWarning, "InvalidSchedule", "invalid schedule: %s : %s", cronJob.Spec.Schedule, err)
        return cronJob, nil, updateStatus, nil
    }
    if scheduledTime == nil {
        // 没有未满足的启动时间
        klog.V(4).InfoS("No unmet start times", "cronjob", klog.KRef(cronJob.GetNamespace(), cronJob.GetName()))
        t := nextScheduledTimeDuration(sched, now)
        return cronJob, t, updateStatus, nil
    }
  
    // 检查是否错过启动期限
    tooLate := false
    if cronJob.Spec.StartingDeadlineSeconds != nil {
        tooLate = scheduledTime.Add(time.Second * time.Duration(*cronJob.Spec.StartingDeadlineSeconds)).Before(now)
    }
    if tooLate {
        // 错过启动期限，记录警告事件
        klog.V(4).InfoS("Missed starting window", "cronjob", klog.KRef(cronJob.GetNamespace(), cronJob.GetName()))
        jm.recorder.Eventf(cronJob, corev1.EventTypeWarning, "MissSchedule", "Missed scheduled time to start a job: %s", scheduledTime.UTC().Format(time.RFC1123Z))
        t := nextScheduledTimeDuration(sched, now)
        return cronJob, t, updateStatus, nil
    }
  
    // ... 后面的代码
  
    return cronJob, t, updateStatus, nil
}

关键逻辑：

  • 控制器计算下次调度时间 scheduledTime
  • 如果设置了 startingDeadlineSeconds，控制器检查当前时间是否超过了调度时间加上启动期限
  • 如果超过了启动期限，控制器记录警告事件并跳过本次执行
  • 控制器会计算下次调度时间并重新入队，等待下一次执行

五、状态管理

5.1 Job 状态管理

5.1.1 状态字段

Job 的状态包含以下关键字段：

  • Active：当前活跃的 Pod 数量
  • Succeeded：成功完成的 Pod 数量
  • Failed：失败的 Pod 数量
  • Ready：就绪的 Pod 数量
  • CompletionTime：Job 完成的时间
  • StartTime：Job 开始的时间
  • Conditions：Job 的状态条件，包括 Complete、Failed、Suspended 等
5.1.2 状态更新

控制器通过 updateJobStatus 函数更新 Job 的状态，确保状态信息准确反映当前任务的执行状态。

5.2 CronJob 状态管理

5.2.1. 状态字段

CronJob 的状态包含以下关键字段：

  • Active：当前活跃的 Job 列表
  • LastScheduleTime：上次调度的时间
  • LastSuccessfulTime：上次成功执行的时间
5.2.2. 状态更新

控制器在以下情况下更新 CronJob 的状态：

  • 创建 Job：当创建新的 Job 时，将其添加到 Active 列表，并更新 LastScheduleTime
  • Job 完成：当 Job 完成时，从 Active 列表中移除，并更新 LastSuccessfulTime
  • Job 失败：当 Job 失败时，从 Active 列表中移除

六、Job 与 CronJob 的对比

特性 Job CronJob
任务类型 一次性任务 周期性任务
调度方式 手动创建 时间调度
并行执行 支持（通过 parallelism） 支持（通过并发策略）
失败处理 支持（通过 backoffLimit） 间接支持（通过 Job 的失败处理）
活跃期限 支持（通过 activeDeadlineSeconds） 间接支持（通过 Job 的活跃期限）
历史管理 不支持 支持（通过 successfulJobsHistoryLimit 和 failedJobsHistoryLimit）
时区支持 不支持 支持（通过 timeZone）
启动期限 不支持 支持（通过 startingDeadlineSeconds）
适用场景 批处理任务、数据分析、一次性操作 定期备份、报表生成、清理任务、周期性检查

旧文档链接：Job、CronJob 逻辑结构分析

k8s 源码分析 Job CronJob