cronjob

1.简介

1.1.定义

• cronjob controller用于定时任务的编排和运行，基于延迟队列和informer缓存实现定时任务调度，会周期性下发job完成既定周期目标。

• // ControllerV2 is a controller for CronJobs, that uses DelayingQueue and informers.
  type ControllerV2 struct {
  	queue workqueue.RateLimitingInterface					// 延迟队列
      ...
  
  	jobControl     jobControlInterface						// job调度
  	cronJobControl cjControlInterface						// cronjob调度
  
  	jobLister     batchv1listers.JobLister					// job缓存
  	cronJobLister batchv1listers.CronJobLister				// cronjob缓存
  	...
  
  	now func() time.Time									// 时间获取器
  }
  

• 

• 注意

  cronjobv1未使用informer缓存，直接周期访问apiserver数据，相比之下cronjobv2基于delay queue和informer维护定时任务

1.2.原理

• cronjob controller会监听informer相关资源，将关联cronjob推入delay queue供syncHandler消费处理，实现任务的周期调度。
• 

• 注意

  cronjob和job是从属关系，job是一次性任务，cronjob基于周期下发job实现定时任务

2.分析

2.1.start

• startCronJobController()作为入口实例化cronjob，基于jobInformer和cronInformer维护queue，驱动sync周期维护定时任务。

• // NewControllerV2 creates and initializes a new Controller.
  func NewControllerV2(...) (*ControllerV2, error) {
  	...
  
      // 实例化cronjob controller对象
  	jm := &ControllerV2{
  		queue:       workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "cronjob"),
  		kubeClient:  kubeClient,
  		...
  
  		jobControl:     realJobControl{KubeClient: kubeClient},
  		cronJobControl: &realCJControl{KubeClient: kubeClient},
  
  		jobLister:     jobInformer.Lister(),
  		cronJobLister: cronJobsInformer.Lister(),
  		...
          
  		now:                 time.Now,
  	}
  
      // 注册job informer回调
  	jobInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
  		AddFunc:    jm.addJob,
  		UpdateFunc: jm.updateJob,
  		DeleteFunc: jm.deleteJob,
  	})
  
      // 注册cronjob informer回调
  	cronJobsInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
  		AddFunc: func(obj interface{}) {
  			jm.enqueueController(obj)
  		},
  		UpdateFunc: func(oldObj, newObj interface{}) {
  			jm.updateCronJob(logger, oldObj, newObj)
  		},
  		DeleteFunc: func(obj interface{}) {
  			jm.enqueueController(obj)
  		},
  	})
  	...
  
  	return jm, nil
  }
  
  func startCronJobController(...) (controller.Interface, bool, error) {
      // 初始化cronjob controller
  	cj2c, err := cronjob.NewControllerV2(ctx, controllerContext.InformerFactory.Batch().V1().Jobs(),
  		controllerContext.InformerFactory.Batch().V1().CronJobs(),
  		controllerContext.ClientBuilder.ClientOrDie("cronjob-controller"),
  	)
  	...
  
      // 启动cron同步
  	go cj2c.Run(ctx, int(controllerContext.ComponentConfig.CronJobController.ConcurrentCronJobSyncs))
  	return nil, true, nil
  }
  

• 

• 注意

  job/cronjob事件回调会将cronjob推入workqueue，cronjob定时规则或时区变化会触发延迟入队，避免不必要的调度

2.2.run

• jm.Run()会启动多个worker获取queue队首任务，调用jm.sync()同步cronjob任务，同步完成会将cronjob放入延迟队列供下次调度。

• // Run starts the main goroutine responsible for watching and syncing jobs.
  func (jm *ControllerV2) Run(ctx context.Context, workers int) {
  	...
  
  	defer jm.queue.ShutDown()
  	...
  
      // 等待informer同步完成
  	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.Done()
  }
  
  func (jm *ControllerV2) worker(ctx context.Context) {
  	for jm.processNextWorkItem(ctx) {
  	}
  }
  
  func (jm *ControllerV2) processNextWorkItem(ctx context.Context) bool {
  	key, quit := jm.queue.Get()
  	...
      
  	defer jm.queue.Done(key)
  
      // 执行同步
  	requeueAfter, err := jm.sync(ctx, key.(string))
  	switch {
  	case err != nil:
  		// 失败入队重试
  		jm.queue.AddRateLimited(key)
  	case requeueAfter != nil:
  		jm.queue.Forget(key)
          // 成功延迟入队
  		jm.queue.AddAfter(key, *requeueAfter)
  	}
  	return true
  }
  

• 

• 注意

  jm.sync()是cronjob的调度入口，queue弹出的任务由sync()进行同步

2.3.sync

• jm.sync()是cronjob controller核心处理入口，其主要逻辑分布在jm.syncCronJob()，负责清理完成的job及下发新的job。

• func (jm *ControllerV2) sync(ctx context.Context, cronJobKey string) (*time.Duration, error) {
  	ns, name, err := cache.SplitMetaNamespaceKey(cronJobKey)
  	...
      
  	// 获取cronjob
  	cronJob, err := jm.cronJobLister.CronJobs(ns).Get(name)
  	...
  
      // 获取所属job
  	jobsToBeReconciled, err := jm.getJobsToBeReconciled(cronJob)
  	...
  
  	// cronJobCopy is used to combine all the updates to a CronJob object and perform an actual update.
  	cronJobCopy := cronJob.DeepCopy()
  
      // 清理完成的job
  	updateStatusAfterCleanup := jm.cleanupFinishedJobs(ctx, cronJobCopy, jobsToBeReconciled)
  
      // 同步最新任务
  	requeueAfter, updateStatusAfterSync, syncErr := jm.syncCronJob(ctx, cronJobCopy, jobsToBeReconciled)
  	...
  
  	// Update the CronJob if needed
  	if updateStatusAfterCleanup || updateStatusAfterSync {
  		// 更新cronjob状态
          jm.cronJobControl.UpdateStatus(ctx, cronJobCopy)
          ...
  	}
  
      // 延迟入队检查
  	if requeueAfter != nil {
  		return requeueAfter, nil
  	}
  
  	return nil, syncErr
  }
  
  func (jm *ControllerV2) getJobsToBeReconciled(cronJob *batchv1.CronJob) ([]*batchv1.Job, error) {
  	// 获取所有job
  	jobList, err := jm.jobLister.Jobs(cronJob.Namespace).List(labels.Everything())
  	...
  
  	for _, job := range jobList {
  		// 匹配owner
  		if owner := metav1.GetControllerOf(job); owner != nil && owner.Name == cronJob.Name {
  			// this job is needs to be reconciled
  			jobsToBeReconciled = append(jobsToBeReconciled, job)
  		}
  	}
      
  	return jobsToBeReconciled, nil
  }
  
  // cleanupFinishedJobs cleanups finished jobs created by a CronJob.
  func (jm *ControllerV2) cleanupFinishedJobs(ctx context.Context, cj *batchv1.CronJob, js []*batchv1.Job) bool {
  	// If neither limits are active, there is no need to do anything.
  	if cj.Spec.FailedJobsHistoryLimit == nil && cj.Spec.SuccessfulJobsHistoryLimit == nil {
  		return false
  	}
  	...
  
  	for _, job := range js {
  		isFinished, finishedStatus := jm.getFinishedStatus(job)
          // success job
  		if isFinished && finishedStatus == batchv1.JobComplete {
  			successfulJobs = append(successfulJobs, job)
          // failed job
  		} else if isFinished && finishedStatus == batchv1.JobFailed {
  			failedJobs = append(failedJobs, job)
  		}
  	}
  
      // success job清理最旧的N个
  	if cj.Spec.SuccessfulJobsHistoryLimit != nil && jm.removeOldestJobs(ctx, cj, successfulJobs,
  			*cj.Spec.SuccessfulJobsHistoryLimit) {
  		updateStatus = true
  	}
  
      // failed job清理最旧的N个
  	if cj.Spec.FailedJobsHistoryLimit != nil && jm.removeOldestJobs(ctx, cj, failedJobs,
  			*cj.Spec.FailedJobsHistoryLimit) {
  		updateStatus = true
  	}
  
  	return updateStatus
  }
  

• 

• 注意

  jm.syncCronJob()是同步定时任务的主要模块，会基于不同策略管理需要下发的job

2.4.syncCron

• jm.syncCronJob()会根据cronjob及所属的job执行同步，基于不同策略回收旧的job及下发新的job，具体的任务执行委托给job完成。

• // syncCronJob reconciles a CronJob with a list of any Jobs that it created.
  func (jm *ControllerV2) syncCronJob(ctx context.Context, cronJob *batchv1.CronJob, jobs []*batchv1.Job) (...) {
  	...
      
      // 先处理底层获取的job
  	for _, j := range jobs {
  		childrenJobs[j.ObjectMeta.UID] = true
          
          // 由cronjob.status.active找job
  		found := inActiveList(cronJob, j.ObjectMeta.UID)
          
          // 未找到+job未完成
  		if !found && !IsJobFinished(j) {
              // 获取底层cronjob
  			cjCopy, err := jm.cronJobControl.GetCronJob(ctx, cronJob.Namespace, cronJob.Name)
  			...
              
              // 再次检查cronjob.status.active,检查通过更新cronjob
  			if inActiveList(cjCopy, j.ObjectMeta.UID) {
  				cronJob = cjCopy
  				continue
  			}
  
          // 找到+job完成
  		} else if found && IsJobFinished(j) {
              ...
              // 清理cronjob.status.active的job.uid
  			deleteFromActiveList(cronJob, j.ObjectMeta.UID)
  			
              updateStatus = true
         	
          // 未找到+job完成
  		} else if IsJobFinished(j) {
  			
              // 更新cronjob.status完成时间
  			if cronJob.Status.LastSuccessfulTime == nil {
  				cronJob.Status.LastSuccessfulTime = j.Status.CompletionTime
  				updateStatus = true
  			}
              
  			if j.Status.CompletionTime != nil && j.Status.CompletionTime.After(cronJob.Status.LastSuccessfulTime.Time) {
  				cronJob.Status.LastSuccessfulTime = j.Status.CompletionTime
  				updateStatus = true
  			}
  		}
  	}
  
  	// 再处理cronjob.status.active记录的job
  	for _, j := range cronJob.Status.Active {
          // job处理过,说明存在,直接跳过
  		_, found := childrenJobs[j.UID]
  		if found {
  			continue
  		}
          
  		// 未处理过的job检查存在
  		_, err := jm.jobControl.GetJob(j.Namespace, j.Name)
  		switch {
  		case errors.IsNotFound(err):
  			// 未找到,由cronjob.status.active删除
  			deleteFromActiveList(cronJob, j.UID)
  			updateStatus = true
  		...
  		}
  	}
  
      // cronjob正在删除,不再同步
  	if cronJob.DeletionTimestamp != nil {
  		// The CronJob is being deleted.
  		// Don't do anything other than updating status.
  		return nil, updateStatus, nil
  	}
  	...
  
      // cronjob挂起
  	if cronJob.Spec.Suspend != nil && *cronJob.Spec.Suspend {
  		return nil, updateStatus, nil
  	}
  
      // 获取开源cron库的调度解释器
  	sched, err := cron.ParseStandard(formatSchedule(cronJob, jm.recorder))
  	...
  
      // 获取cronjob错过的最近一次调度时间
  	scheduledTime, err := nextScheduleTime(logger, cronJob, now, sched, jm.recorder)
  	...
      
      // 未错过任何调度
  	if scheduledTime == nil {
  		// 计算下次调度时间
  		t := nextScheduleTimeDuration(cronJob, now, sched)
  		return t, updateStatus, nil
  	}
  
  	tooLate := false
      
      // 错过的调度时间窗检查
  	if cronJob.Spec.StartingDeadlineSeconds != nil {
          // 检查是否错误调度任务的窗口期
  		tooLate = scheduledTime.Add(time.Second * time.Duration(*cronJob.Spec.StartingDeadlineSeconds)).Before(now)
  	}
      
      // 最近错过的调度超出时间窗
  	if tooLate {
  		// 重新计算下次执行时间
  		t := nextScheduleTimeDuration(cronJob, now, sched)
  		return t, updateStatus, nil
  	}
      
      // 检查本次调度是否已经完成(jobName存在或上一次调度时间一致)
  	if inActiveListByName(cronJob, &batchv1.Job{...) || cronJob.Status.LastScheduleTime.Equal(*scheduledTime) {
  		// 重新计算下次调度时间
  		t := nextScheduleTimeDuration(cronJob, now, sched)
  		return t, updateStatus, nil
  	}
      
      // ForbidConcurrent策略,仍存在job运行则等待下次调度
  	if cronJob.Spec.ConcurrencyPolicy == batchv1.ForbidConcurrent && len(cronJob.Status.Active) > 0 {
  		
  		t := nextScheduleTimeDuration(cronJob, now, sched)
  		return 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)
  			...
              
              // 删除job
  			deleteJob(logger, cronJob, job, jm.jobControl, jm.recorder)
              ...
              
  			updateStatus = true
  		}
  	}
  	...
                                                  
     	// 生成job对象
  	jobReq, err := getJobFromTemplate2(cronJob, *scheduledTime)
  	...
                                                  
      // 创建job对象
  	jobResp, err := jm.jobControl.CreateJob(cronJob.Namespace, jobReq)
      
  	switch {
      // ns terminating,跳过处理
  	case errors.HasStatusCause(err, corev1.NamespaceTerminatingCause):
  		return nil, updateStatus, err
      // 已存在
  	case errors.IsAlreadyExists(err):
  		// 获取job对象
  		jobAlreadyExists = true
  		job, err := jm.jobControl.GetJob(jobReq.GetNamespace(), jobReq.GetName())
  		...
          
  		jobResp = job
  
  		// 已存在的job不是croonjob子资源,跳过
  		if !metav1.IsControlledBy(job, cronJob) {
  			return nil, updateStatus, nil
  		}
  
  		// job已加入cronjob.status.active,跳过
  		found := inActiveList(cronJob, job.ObjectMeta.UID)
  		if found {
  			return nil, updateStatus, nil
  		}
  	...
  	}
  	...
  
  	// job元数据
  	jobRef, err := getRef(jobResp)
  	...
      
      // 更新cronjob.status数据
  	cronJob.Status.Active = append(cronJob.Status.Active, *jobRef)
  	cronJob.Status.LastScheduleTime = &metav1.Time{Time: *scheduledTime}
  	updateStatus = true
  
      // 重新计算下次调度时间
  	t := nextScheduleTimeDuration(cronJob, now, sched)
      
  	return t, updateStatus, nil
  }
  

• 

• 注意

  1.replace策略会造成资源浪费，就算job执行进度达到90%，由于已经触发下一次调度，依然会清理这个90%进度的job

  2.cronjob会检查scheduledTime与LastScheduledTime差异，相同会取下次调度时间，计算也会加入100ms抖动，缓解时间临界问题

2.5.nextSched

• nextScheduleTime()和nextScheduleTimeDuration()会计算下次调度时间，调度时间不能在now之前，同时会加入100ms抖动。

• // nextScheduleTime returns the time.Time of the next schedule after the last scheduled and before now.
  func nextScheduleTime(...) (*time.Time, error) {
      // 获取最近一次错过的调度时间
  	_, mostRecentTime, missedSchedules, err := mostRecentScheduleTime(cj, now, schedule, true)
  
      // 未错过调度或还未到调度时间
  	if mostRecentTime == nil || mostRecentTime.After(now) {
  		return nil, err
  	}
  	...
  
      // 返回now之前最近错过的调度时间
  	return mostRecentTime, err
  }
  
  // nextScheduleTimeDuration returns the time duration to requeue based on the schedule and last schedule time.
  func nextScheduleTimeDuration(cj *batchv1.CronJob, now time.Time, schedule cron.Schedule) *time.Duration {
  	// 获取最近错过的调度时间
      earliestTime, mostRecentTime, missedSchedules, err := mostRecentScheduleTime(cj, now, schedule, false)
      
      // 间隔不合法
  	if err != nil {
  		// mostRecentTime取当前时间
  		mostRecentTime = &now
  	} else if mostRecentTime == nil {
          // 自earliestTime未错过调度
  		if missedSchedules == noneMissed {
  			// missedSchedules取earliestTime
  			mostRecentTime = &earliestTime
          // 自earliestTime错过调度
  		} else {
  			// mostRecentTime取当前时间
  			mostRecentTime = &now
  		}
  	}
  
      // 计算下一次执行间隔时间
  	t := schedule.Next(*mostRecentTime).Add(nextScheduleDelta).Sub(now)
  	return &t
  }
  
  // 最近调度时间计算
  func mostRecentScheduleTime(...) (time.Time, *time.Time, missedSchedulesType, error) {
      // earliestTime先使用cronjob创建时间(应对第一次调度)
  	earliestTime := cj.ObjectMeta.CreationTimestamp.Time
  	missedSchedules := noneMissed
      
      // earliestTime重置为上次调度时间
  	if cj.Status.LastScheduleTime != nil {
  		earliestTime = cj.Status.LastScheduleTime.Time
  	}
      
      // 基于startingDeadlineSeconds校准earliestTime
  	if includeStartingDeadlineSeconds && cj.Spec.StartingDeadlineSeconds != nil {
  		schedulingDeadline := now.Add(-time.Second * time.Duration(*cj.Spec.StartingDeadlineSeconds))
  
          // earliestTime过期,重置为now.sub(deadline)
  		if schedulingDeadline.After(earliestTime) {
  			earliestTime = schedulingDeadline
  		}
  	}
  
      // 下次调度时间
  	t1 := schedule.Next(earliestTime)
      // 下下次调度时间
  	t2 := schedule.Next(t1)
  
      // 下次调度时间未过期
  	if now.Before(t1) {
  		return earliestTime, nil, missedSchedules, nil
  	}
      
      // 下下次调度时间未过期,错过t1调度
  	if now.Before(t2) {
  		return earliestTime, &t1, missedSchedules, nil
  	}
  	...
      
  	// 近似计算错过次数
  	timeElapsed := int64(now.Sub(t1).Seconds())
  	numberOfMissedSchedules := (timeElapsed / timeBetweenTwoSchedules) + 1
  
      // 回退2个周期后开始扫描最近调度时间(多回退一次用于兼容不规则调度及近似误差边界情况)
  	potentialEarliest := t1.Add(time.Duration((numberOfMissedSchedules-1-1)*timeBetweenTwoSchedules) * time.Second)
      
      // 取now之前最后错过的调度时间
  	for t := schedule.Next(potentialEarliest); !t.After(now); t = schedule.Next(t) {
  		mostRecentTime = t
  	}
  
  	switch {
      / 错过100次调度,设为manyMissed
  	case numberOfMissedSchedules > 100:
  		missedSchedules = manyMissed
  	// 错误少量调度,设为fewMissed
  	case numberOfMissedSchedules > 0:
  		missedSchedules = fewMissed
  	}
  
      // 检查是否找到有效时间
  	if mostRecentTime.IsZero() {
  		return earliestTime, nil, missedSchedules, nil
  	}
      
  	return earliestTime, &mostRecentTime, missedSchedules, nil
  }
  

• 

• 注意

  1.nextScheduleTime()获取的是last~now之间最后错过的scheduleTime

  2.nextScheduleTimeDuration()计算的是next scheduleTime，基于lastTime或最近错过的scheduleTime