源码分析 kubernetes cronjob controller 控制器的实现原理

apiVersion: batch/v1beta1
kind: CronJob
metadata:
  name: hello
spec:
  schedule: "*/1 * * * *"
  jobTemplate:
    spec:
      template:
        spec:
          containers:
          - name: hello
            image: busybox
            args:
            - /bin/sh
            - -c
            - curl xiaorui.cc
          restartPolicy: OnFailure

func NewControllerV2(jobInformer batchv1informers.JobInformer, cronJobsInformer batchv1informers.CronJobInformer, kubeClient clientset.Interface) (*ControllerV2, error) {
    jm := &ControllerV2{
        queue:       workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "cronjob"),
        jobControl:     realJobControl{KubeClient: kubeClient},
        cronJobControl: &realCJControl{KubeClient: kubeClient},
        jobLister:     jobInformer.Lister(),
        cronJobLister: cronJobsInformer.Lister(),
    }

    // 在 job informer 注册 eventHandler
    jobInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc:    jm.addJob,
        UpdateFunc: jm.updateJob,
        DeleteFunc: jm.deleteJob,
    })

    // 在 cronjob informer 注册 eventHandler
    cronJobsInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc: func(obj interface{}) {
            jm.enqueueController(obj)
        },
        UpdateFunc: jm.updateCronJob,
        DeleteFunc: func(obj interface{}) {
            jm.enqueueController(obj)
        },
    })

    return jm, nil
}

func (jm *ControllerV2) Run(ctx context.Context, workers int) {
    // 等待 job 和 cronjob 同步到本地
    if !cache.WaitForNamedCacheSync("cronjob", ctx.Done(), jm.jobListerSynced, jm.cronJobListerSynced) {
        return
    }

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

    // 阻塞, 直到 ctx 被取消
    <-ctx.Done()
}

func (jm *ControllerV2) worker(ctx context.Context) {
    for jm.processNextWorkItem(ctx) {
    }
}

func (jm *ControllerV2) processNextWorkItem(ctx context.Context) bool {
    // 从队列后去 cronjob 的 key
    key, quit := jm.queue.Get()
    if quit {
        return false
    }
    defer jm.queue.Done(key)

    // 执行 cronjob 的核心同步方法
    requeueAfter, err := jm.sync(ctx, key.(string))
    switch {
    case err != nil:
        // 失败, 把任务放到队列里进行重试
        jm.queue.AddRateLimited(key)

    case requeueAfter != nil:
        // 从队列中剔除
        jm.queue.Forget(key)

        // 通过上面的 sync 计算出下次执行的时间, 然后把任务放到 queue 的延迟队列里, 延迟时间为 requeueAfter.
        jm.queue.AddAfter(key, *requeueAfter)
    }
    return true
}

func (jm *ControllerV2) sync(ctx context.Context, cronJobKey string) (*time.Duration, error) {
    // 从 key 中拆分 ns 和 name 字段
    ns, name, err := cache.SplitMetaNamespaceKey(cronJobKey)
    if err != nil {
        return nil, err
    }

    // 从 informer lister 获取 cronjob 对象
    cronJob, err := jm.cronJobLister.CronJobs(ns).Get(name)
    switch {
    case errors.IsNotFound(err):
        return nil, nil
    case err != nil:
        // for other transient apiserver error requeue with exponential backoff
        return nil, err
    }

    // 获取 cronjob 关联的 jobs 对象集合
    jobsToBeReconciled, err := jm.getJobsToBeReconciled(cronJob)
    if err != nil {
        return nil, err
    }

    // 获取 cronjob 对象, 下次执行的时间, 是否更新状态等
    cronJobCopy, requeueAfter, updateStatus, err := jm.syncCronJob(ctx, cronJob, jobsToBeReconciled)
    if err != nil {
        if updateStatus {
            // 更新 cronjob 的状态
            if _, err := jm.cronJobControl.UpdateStatus(ctx, cronJobCopy); err != nil {
                return nil, err
            }
        }
        return nil, err
    }

    // 清理已经完成的 jobs.
    if jm.cleanupFinishedJobs(ctx, cronJobCopy, jobsToBeReconciled) {
        updateStatus = true
    }

    // 更新 cronjob 的状态
    if updateStatus {
        if _, err := jm.cronJobControl.UpdateStatus(ctx, cronJobCopy); err != nil {
            return nil, err
        }
    }

    // 如果拿到了下次执行的 duration, 则返回该 duration
    if requeueAfter != nil {
        return requeueAfter, nil
    }
    return nil, nil
}

func (jm *ControllerV2) syncCronJob(
    ctx context.Context,
    cronJob *batchv1.CronJob,
    jobs []*batchv1.Job) (*batchv1.CronJob, *time.Duration, bool, error) {

    cronJob = cronJob.DeepCopy()
    now := jm.now()
    updateStatus := false
    timeZoneEnabled := utilfeature.DefaultFeatureGate.Enabled(features.CronJobTimeZone)

    // 创建一个集合映射 job.uid.
    childrenJobs := make(map[types.UID]bool)
    for _, j := range jobs {
        childrenJobs[j.ObjectMeta.UID] = true

        // job 是否在 cronJob active 里
        found := inActiveList(*cronJob, j.ObjectMeta.UID)

        // 如果 job uid 跟 cronjob uid 不相同, 且任务没完成 
        if !found && !IsJobFinished(j) {
            // 获取 cronjob 对象
            cjCopy, err := jm.cronJobControl.GetCronJob(ctx, cronJob.Namespace, cronJob.Name)
            if err != nil {
                return nil, nil, updateStatus, err
            }

            // 再次判断是否相等, 如果相等则使用新拿到的 cronjob 对象
            if inActiveList(*cjCopy, j.ObjectMeta.UID) {
                cronJob = cjCopy
                continue
            }

        } else if found && IsJobFinished(j) {
            // 如果相同, 且 job 已完成
            _, status := getFinishedStatus(j)

            // 从 list 中删除该 job
            deleteFromActiveList(cronJob, j.ObjectMeta.UID)

            // 往 event 里输出该状态
            jm.recorder.Eventf(cronJob, corev1.EventTypeNormal, "SawCompletedJob", "Saw completed job: %s, status: %v", j.Name, status)
            updateStatus = true

        } else if IsJobFinished(j) {
            // 如果该 job 已完成, 则更新时间.
            if cronJob.Status.LastSuccessfulTime == nil {
                cronJob.Status.LastSuccessfulTime = j.Status.CompletionTime
                updateStatus = true
            }
        }
    }

    // 遍历 cronjob active 集合, 如果对应的 job 不再存在, 则从 active list 中删除 job 引用. 避免 cronjob 可能永远处于活动模式.
    for _, j := range cronJob.Status.Active {
        _, found := childrenJobs[j.UID]
        if found {
            continue
        }

        _, err := jm.jobControl.GetJob(j.Namespace, j.Name)
        switch {
        case errors.IsNotFound(err):
            jm.recorder.Eventf(cronJob, corev1.EventTypeNormal, "MissingJob", "Active job went missing: %v", j.Name)

            // 在 active 里有, 但 job 里不存在, 则需要在 cronJob active 里删除, 为了一致性.
            deleteFromActiveList(cronJob, j.UID)
            updateStatus = true
        case err != nil:
            return cronJob, nil, updateStatus, err
        }
    }

    // 如果删除时间不为空, 说明该对象已被删除, 后面无需处理了.
    if cronJob.DeletionTimestamp != nil {
        return cronJob, nil, updateStatus, nil
    }

    // timezone 异常则退出, 输出一条未知时区的错误事件
    if timeZoneEnabled && cronJob.Spec.TimeZone != nil {
        if _, err := time.LoadLocation(*cronJob.Spec.TimeZone); err != nil {
            timeZone := pointer.StringDeref(cronJob.Spec.TimeZone, "")
            jm.recorder.Eventf(cronJob, corev1.EventTypeWarning, "UnknownTimeZone", "invalid timeZone: %q: %s", timeZone, err)
            return cronJob, nil, updateStatus, nil
        }
    }

    // 该 cronjob 已暂停则直接退出.
    if cronJob.Spec.Suspend != nil && *cronJob.Spec.Suspend {
        return cronJob, nil, updateStatus, nil
    }

    // 获取开源 cron 库的调度解释器
    sched, err := cron.ParseStandard(formatSchedule(timeZoneEnabled, cronJob, jm.recorder))
    if err != nil {
        return cronJob, nil, updateStatus, nil
    }

    // 根据 crontab spec 表达式计算下次执行的时间
    scheduledTime, err := getNextScheduleTime(*cronJob, now, sched, jm.recorder)
    if err != nil {
        // 异常则发送 cron spec 解析失败的事件
        jm.recorder.Eventf(cronJob, corev1.EventTypeWarning, "InvalidSchedule", "invalid schedule: %s : %s", cronJob.Spec.Schedule, err)
        return cronJob, nil, updateStatus, nil
    }

    // 如果事件为空, 则尝试使用 nextScheduledTimeDuration 来计算下次时间.
    if scheduledTime == nil {
        t := nextScheduledTimeDuration(*cronJob, sched, now)
        return cronJob, t, updateStatus, nil
    }

    tooLate := false
    // 如果配置 StartingDeadlineSeconds, 则开启 toolate 延后
    if cronJob.Spec.StartingDeadlineSeconds != nil {
        tooLate = scheduledTime.Add(time.Second * time.Duration(*cronJob.Spec.StartingDeadlineSeconds)).Before(now)
    }
    if tooLate {
        t := nextScheduledTimeDuration(*cronJob, sched, now)
        return cronJob, t, updateStatus, nil
    }

    // 如果配置了 ForbidConcurrent 策略, 且当前已经有 job 还在运行, 则直接跳出 
    if cronJob.Spec.ConcurrencyPolicy == batchv1.ForbidConcurrent && len(cronJob.Status.Active) > 0 {
        t := nextScheduledTimeDuration(*cronJob, sched, now)
        return cronJob, t, updateStatus, nil
    }

    // 如果配置了 ReplaceConcurrent 策略, 则需要清理以前还在运行的 Job.
    if cronJob.Spec.ConcurrencyPolicy == batchv1.ReplaceConcurrent {
        for _, j := range cronJob.Status.Active {
            // 获取当前的 job 对象
            job, err := jm.jobControl.GetJob(j.Namespace, j.Name)
            if err != nil {
                return cronJob, nil, updateStatus, err
            }
            // 删除该 job 对象
            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
        }
    }

    // 获取 cronjob 对应的 job 模板
    jobReq, err := getJobFromTemplate2(cronJob, *scheduledTime)
    if err != nil {
        return cronJob, nil, updateStatus, err
    }

    // 创建 job 任务
    jobResp, err := jm.jobControl.CreateJob(cronJob.Namespace, jobReq)

    // 在 cronjob 对象里关联 job 对象, 记录 LastScheduleTime 时间和状态等
    jobRef, err := getRef(jobResp)
    cronJob.Status.Active = append(cronJob.Status.Active, *jobRef)
    cronJob.Status.LastScheduleTime = &metav1.Time{Time: *scheduledTime}
    updateStatus = true

    // 获取下次调度执行的 duration
    t := nextScheduledTimeDuration(*cronJob, sched, now)
    return cronJob, t, updateStatus, nil
}

var (
    nextScheduleDelta = 100 * time.Millisecond
)

func nextScheduledTimeDuration(cj batchv1.CronJob, sched cron.Schedule, now time.Time) *time.Duration {
    // 默认使用创建时间, 通常第一次的时候, 没有上次的调度时间.
    earliestTime := cj.ObjectMeta.CreationTimestamp.Time
    if cj.Status.LastScheduleTime != nil {
        earliestTime = cj.Status.LastScheduleTime.Time
    }

    // 计算最近的时间点
    mostRecentTime, _, err := getMostRecentScheduleTime(earliestTime, now, sched)
    if err != nil {
        mostRecentTime = &now
    } else if mostRecentTime == nil {
        // 为空, 则使用 earliestTime 时间
        mostRecentTime = &earliestTime
    }

    // 依据 mostRecentTime 时间点和 crontab spec 计算出下次 cron 调度时间点, 增加 100ms 的波动, 减去当前时间为 duration 时长
    t := sched.Next(*mostRecentTime).Add(nextScheduleDelta).Sub(now)

    return &t
}

func getMostRecentScheduleTime(earliestTime time.Time, now time.Time, schedule cron.Schedule) (*time.Time, int64, error) {
    t1 := schedule.Next(earliestTime)
    t2 := schedule.Next(t1)

    // 如果 now 在 t1 前面, 直接退出, 可以直接使用 earliestTime 作为时间点计算.
    if now.Before(t1) {
        return nil, 0, nil
    }

    // 如果 now 在 t1 后面, 说明当前的 next 时间不能用了, 需要使用 next next 的时间点, 再来判断.

    // 如果 now 在 next.next 的前面, 可以退出, 使用 t1 的时间.
    if now.Before(t2) {
        return &t1, 1, nil
    }

    // 说实话没看懂啥意思... 
    timeBetweenTwoSchedules := int64(t2.Sub(t1).Round(time.Second).Seconds())
    if timeBetweenTwoSchedules < 1 {
        return nil, 0, fmt.Errorf("time difference between two schedules less than 1 second")
    }
    timeElapsed := int64(now.Sub(t1).Seconds())
    numberOfMissedSchedules := (timeElapsed / timeBetweenTwoSchedules) + 1
    t := time.Unix(t1.Unix()+((numberOfMissedSchedules-1)*timeBetweenTwoSchedules), 0).UTC()
    return &t, numberOfMissedSchedules, nil
}

func (jm *ControllerV2) cleanupFinishedJobs(ctx context.Context, cj *batchv1.CronJob, js []*batchv1.Job) bool {
    if cj.Spec.FailedJobsHistoryLimit == nil && cj.Spec.SuccessfulJobsHistoryLimit == nil {
        return false
    }

    updateStatus := false
    failedJobs := []*batchv1.Job{}
    successfulJobs := []*batchv1.Job{}

    for _, job := range js {
        isFinished, finishedStatus := jm.getFinishedStatus(job)
        if isFinished && finishedStatus == batchv1.JobComplete {
            successfulJobs = append(successfulJobs, job)
        } else if isFinished && finishedStatus == batchv1.JobFailed {
            failedJobs = append(failedJobs, job)
        }
    }

    if cj.Spec.SuccessfulJobsHistoryLimit != nil &&
        jm.removeOldestJobs(cj,
            successfulJobs,
            *cj.Spec.SuccessfulJobsHistoryLimit) {
        updateStatus = true
    }

    if cj.Spec.FailedJobsHistoryLimit != nil &&
        jm.removeOldestJobs(cj,
            failedJobs,
            *cj.Spec.FailedJobsHistoryLimit) {
        updateStatus = true
    }

    return updateStatus
}