mkdir project
cd project
kubebuilder init --domain tutorial.kubebuilder.io --repo tutorial.kubebuilder.io/project

apiVersion: <group>.<domain>/<version>

kubebuilder create api --group xxx --version v1 --kind MyKind
make generate
make manifests

kubebuilder create api --group batch --version v1 --kind CronJob

type CronJobSpec struct {
    // INSERT ADDITIONAL SPEC FIELDS - desired state of cluster
    // Important: Run "make" to regenerate code after modifying this file
}

// CronJobStatus defines the observed state of CronJob
type CronJobStatus struct {
    // INSERT ADDITIONAL STATUS FIELD - define observed state of cluster
    // Important: Run "make" to regenerate code after modifying this file
}

// +kubebuilder:object:root=true
// CronJob is the Schema for the cronjobs API
type CronJob struct {
    metav1.TypeMeta   `json:",inline"`
    metav1.ObjectMeta `json:"metadata,omitempty"`

    Spec   CronJobSpec   `json:"spec,omitempty"`
    Status CronJobStatus `json:"status,omitempty"`
}

// +kubebuilder:object:root=true
// CronJobList contains a list of CronJob
type CronJobList struct {
    metav1.TypeMeta `json:",inline"`
    metav1.ListMeta `json:"metadata,omitempty"`
    Items           []CronJob `json:"items"`
}

func init() {
    SchemeBuilder.Register(&CronJob{}, &CronJobList{})
}

var (
    // GroupVersion is group version used to register these objects
    GroupVersion = schema.GroupVersion{Group: "batch.tutorial.kubebuilder.io", Version: "v1"}

    // SchemeBuilder is used to add go types to the GroupVersionKind scheme
    SchemeBuilder = &scheme.Builder{GroupVersion: GroupVersion}
    // AddToScheme adds the types in this group-version to the given scheme.
    AddToScheme = SchemeBuilder.AddToScheme
)

type CronJobSpec struct {
	Foo string `json:"foo,omitempty"`
	// +kubebuilder:validation:MinLength=0
	// Cron 表达式，用来定义任务执行的时间，例如 */5 * * * * 表示每 5 分钟执行一次。
	Schedule string `json:"schedule"`
	// +optional
	// 调度开始之后在这个时间内，就可以执行job
	// 如果任务错过了调度时间（比如因为 controller 崩溃、宕机），这个字段定义了一个“容忍窗口（秒）”，
	// 允许它在这个时间内仍然启动，否则就认为“错过”，并标记为失败。
	StartingDeadlineSeconds *int64 `json:"startingDeadlineSeconds,omitempty"`
	// +optional
	// 并行策略
	ConcurrencyPolicy ConcurrencyPolicy `json:"concurrencyPolicy,omitempty"`
	// +optional
	// 是否暂停该 CronJob 的调度。
	Suspend *bool `json:"suspend,omitempty"`
	// CronJob 每次执行会基于这个模板生成一个 Job
	JobTemplate batchv1.JobTemplateSpec `json:"jobTemplate"`
	// +optional
	// 保留多少个成功完成的 Job。
	SuccessfulJobsHistoryLimit *int32 `json:"successfulJobsHistoryLimit,omitempty"`
	// +kubebuilder:validation:Minimum=0
	// +optional
	// 失败的job历史记录限制条数
	FailedJobsHistoryLimit *int32 `json:"failedJobsHistoryLimit,omitempty"`
}

// +kubebuilder:validation:Enum=Allow;Forbid;Replace
type ConcurrencyPolicy string

const (
    // AllowConcurrent allows CronJobs to run concurrently.
    AllowConcurrent ConcurrencyPolicy = "Allow"

    // ForbidConcurrent forbids concurrent runs, skipping next run if previous
    // hasn't finished yet.
    ForbidConcurrent ConcurrencyPolicy = "Forbid"

    // ReplaceConcurrent cancels currently running job and replaces it with a new one.
    ReplaceConcurrent ConcurrencyPolicy = "Replace"
)

type CronJobStatus struct {
    // +optional
    Active []corev1.ObjectReference `json:"active,omitempty"`

    // Information when was the last time the job was successfully scheduled.
    // +optional
    LastScheduleTime *metav1.Time `json:"lastScheduleTime,omitempty"`
}

// +kubebuilder:object:root=true
// +kubebuilder:subresource:status
type CronJob struct {
	metav1.TypeMeta   `json:",inline"`
	metav1.ObjectMeta `json:"metadata,omitempty"`

	Spec   CronJobSpec   `json:"spec,omitempty"`
	Status CronJobStatus `json:"status,omitempty"`
}

// CronJobReconciler reconciles a CronJob object
type CronJobReconciler struct {
    client.Client
    Scheme *runtime.Scheme
}

// +kubebuilder:rbac:groups=batch.tutorial.kubebuilder.io,resources=cronjobs,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=batch.tutorial.kubebuilder.io,resources=cronjobs/status,verbs=get;update;patch

func (r *CronJobReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error)

func (r *CronJobReconciler) SetupWithManager(mgr ctrl.Manager) error {
    return ctrl.NewControllerManagedBy(mgr).
        For(&batchv1.CronJob{}).
        Complete(r)
}

type CronJobReconciler struct {
    client.Client
    Log    logr.Logger
    Scheme *runtime.Scheme
    Clock
}
type realClock struct{}

func (_ realClock) Now() time.Time { return time.Now() }

type Clock interface {
    Now() time.Time
}

var cronJob batch.CronJob
    if err := r.Get(ctx, req.NamespacedName, &cronJob); err != nil {
        log.Error(err, "unable to fetch CronJob")
        // we'll ignore not-found errors, since they can't be fixed by an immediate
        // requeue (we'll need to wait for a new notification), and we can get them
        // on deleted requests.
        return ctrl.Result{}, client.IgnoreNotFound(err)
    }

// 列出所有关联的 Job
	var childJobs kbatch.JobList
	if err := r.List(ctx, &childJobs, client.InNamespace(req.Namespace), client.MatchingFields{jobOwnerKey: req.Name}); err != nil {
		log.Error(err, "unable to list child Jobs")
		return ctrl.Result{}, err
	}
	// 分类 Job：成功、失败、运行中
	var activeJobs []*kbatch.Job
	var successfulJobs []*kbatch.Job
	var failedJobs []*kbatch.Job
	var mostRecentTime *time.Time // find the last run so we can update the status
	// 判断一个 Job 是否已经“完成”，（成功或失败）
	isJobFinished := func(job *kbatch.Job) (bool, kbatch.JobConditionType) {
		for _, c := range job.Status.Conditions {
			if (c.Type == kbatch.JobComplete || c.Type == kbatch.JobFailed) && c.Status == corev1.ConditionTrue {
				// 找到符合条件的 Condition，就立即返回 (true, c.Type)
				return true, c.Type
			}
		}
		// 如果遍历完都没有找到符合条件的项
		// 表示这个 Job 还没有完成，也就是仍在运行中或等待中。
		return false, ""
	}
	// 从 Job 的注解中提取调度时间（ScheduledTime）并返回一个 time.Time 类型的时间戳。
	getScheduledTimeForJob := func(job *kbatch.Job) (*time.Time, error) {
		timeRaw := job.Annotations[scheduledTimeAnnotation]
		if len(timeRaw) == 0 {
			return nil, nil
		}

		timeParsed, err := time.Parse(time.RFC3339, timeRaw)
		if err != nil {
			return nil, err
		}
		return &timeParsed, nil
	}
	// 遍历当前 CronJob 的所有子 Job，对它们进行分类，
	// 并提取每个 Job 的调度时间（从 annotation 中拿），以便控制器后续判断和处理。
	for i, job := range childJobs.Items {
		// 判断这个 Job 是否完成，并获取它的完成类型
		_, finishedType := isJobFinished(&job)
		switch finishedType {
		// 还在运行的放入 activeJobs
		case "": // ongoing
			activeJobs = append(activeJobs, &childJobs.Items[i])
		case kbatch.JobFailed:
			failedJobs = append(failedJobs, &childJobs.Items[i])
		case kbatch.JobComplete:
			successfulJobs = append(successfulJobs, &childJobs.Items[i])
		}

		// We'll store the launch time in an annotation, so we'll reconstitute that from
		// the active jobs themselves.
		// 从 Job 的 annotation 中获取调度时间
		scheduledTimeForJob, err := getScheduledTimeForJob(&job)
		// 解析失败，打印日志并跳过这个 Job
		if err != nil {
			log.Error(err, "unable to parse schedule time for child job", "job", &job)
			continue
		}
		// mostRecentTime 会记录该 CronJob 最近一次实际执行 Job 的时间
		if scheduledTimeForJob != nil {
			if mostRecentTime == nil || mostRecentTime.Before(*scheduledTimeForJob) {
				mostRecentTime = scheduledTimeForJob
			}
		}
	}
	// 更新 CronJob 资源中的 Status.LastScheduleTime 字段，用来记录 CronJob 最后一次成功调度时间。
	if mostRecentTime != nil {
		cronJob.Status.LastScheduleTime = &metav1.Time{Time: *mostRecentTime}
	} else {
		cronJob.Status.LastScheduleTime = nil
	}
	cronJob.Status.Active = nil
	// 将 当前活跃的 Job（即正在执行中的 Job）更新到 CronJob 的 Status.Active 字段。
	for _, activeJob := range activeJobs {
		// 将 activeJob 转换为 Kubernetes 对象引用（ObjectReference）。
		jobRef, err := ref.GetReference(r.Scheme, activeJob)
		// 生成 ObjectReference 时发生错误，则记录错误并跳过当前的 activeJob
		if err != nil {
			log.Error(err, "unable to make reference to active job", "job", activeJob)
			continue
		}
		cronJob.Status.Active = append(cronJob.Status.Active, *jobRef)
	}
	// 帮助开发人员或者运维人员监控 CronJob 的执行状态，查看当前系统中各个 Job 的情况。
	log.V(1).Info("job count", "active jobs", len(activeJobs), "successful jobs", len(successfulJobs), "failed jobs", len(failedJobs))
	// 更新 CronJob 的状态信息到 Kubernetes 集群中
	if err := r.Status().Update(ctx, &cronJob); err != nil {
		log.Error(err, "unable to update CronJob status")
		return ctrl.Result{}, err
	}

// 清理过期的失败作业（failed jobs），并根据 CronJob 中的 FailedJobsHistoryLimit 配置删除历史失败的 Job
	if cronJob.Spec.FailedJobsHistoryLimit != nil {
		// 按失败时间对失败作业排序， 最旧的失败作业排在前面。
		sort.Slice(failedJobs, func(i, j int) bool {
			if failedJobs[i].Status.StartTime == nil {
				return failedJobs[j].Status.StartTime != nil
			}
			return failedJobs[i].Status.StartTime.Before(failedJobs[j].Status.StartTime)
		})
		// 删除超出历史保留限制的失败作业
		for i, job := range failedJobs {
			// 检查当前遍历的失败作业是否超过了保留的最大历史失败作业数。
			if int32(i) >= int32(len(failedJobs))-*cronJob.Spec.FailedJobsHistoryLimit {
				break
			}
			// 删除作业
			if err := r.Delete(ctx, job, client.PropagationPolicy(metav1.DeletePropagationBackground)); client.IgnoreNotFound(err) != nil {
				log.Error(err, "unable to delete old failed job", "job", job)
			} else {
				log.V(0).Info("deleted old failed job", "job", job)
			}
		}
	}

	if cronJob.Spec.SuccessfulJobsHistoryLimit != nil {
		sort.Slice(successfulJobs, func(i, j int) bool {
			if successfulJobs[i].Status.StartTime == nil {
				return successfulJobs[j].Status.StartTime != nil
			}
			return successfulJobs[i].Status.StartTime.Before(successfulJobs[j].Status.StartTime)
		})
		for i, job := range successfulJobs {
			if int32(i) >= int32(len(successfulJobs))-*cronJob.Spec.SuccessfulJobsHistoryLimit {
				break
			}
			if err := r.Delete(ctx, job, client.PropagationPolicy(metav1.DeletePropagationBackground)); err != nil {
				log.Error(err, "unable to delete old successful job", "job", job)
			} else {
				log.V(0).Info("deleted old successful job", "job", job)
			}
		}
	}

if cronJob.Spec.Suspend != nil && *cronJob.Spec.Suspend {
       log.V(1).Info("cronjob suspended, skipping")
       return ctrl.Result{}, nil
   }

t := sched.Next(earliestTime)
// 假设earliestTime为11:10，规则为每小时调度，第一次调用t为12:00，依次分别为13:00、14:00.....

getNextSchedule := func(cronJob *batchv1.CronJob, now time.Time) (lastMissed time.Time, next time.Time, err error) {
		// 将 Cron 表达式字符串 (cronJob.Spec.Schedule) 解析成一个 Cron 表达式调度器 sched。
		sched, err := cron.ParseStandard(cronJob.Spec.Schedule)
		if err != nil {
			return time.Time{}, time.Time{}, fmt.Errorf("Unparseable schedule %q: %v", cronJob.Spec.Schedule, err)
		}

		// for optimization purposes, cheat a bit and start from our last observed run time
		// we could reconstitute this here, but there's not much point, since we've
		// just updated it.
		// earliestTime 是开始计算下一个调度时间的基础时间
		var earliestTime time.Time
		// 如果 CronJob 已经有 LastScheduleTime（上次成功调度的时间），则使用该时间。
		if cronJob.Status.LastScheduleTime != nil {
			earliestTime = cronJob.Status.LastScheduleTime.Time
		} else {
			// 如果没有 LastScheduleTime，则使用 CronJob 的创建时间作为开始时间
			earliestTime = cronJob.ObjectMeta.CreationTimestamp.Time
		}
		// 限制我们调度“过去的漏掉的任务”的范围，避免太老的调度点被浪费资源去执行。
		// 设置了 StartingDeadlineSeconds，表示在某个时间点之后的调度将被忽略
		if cronJob.Spec.StartingDeadlineSeconds != nil {
			// controller is not going to schedule anything below this point
			schedulingDeadline := now.Add(-time.Second * time.Duration(*cronJob.Spec.StartingDeadlineSeconds))
			// 如果 earliestTime 晚于 schedulingDeadline，则将 earliestTime 更新为 schedulingDeadline。
			if schedulingDeadline.After(earliestTime) {
				earliestTime = schedulingDeadline
			}
		}
		if earliestTime.After(now) {
			return time.Time{}, sched.Next(now), nil
		}

		starts := 0

		for t := sched.Next(earliestTime); !t.After(now); t = sched.Next(t) {
			// 最终记录的是最近一次错过的时间
			lastMissed = t
			// 一共发现了多少个错过的时间点
			starts++
			if starts > 100 {
				// We can't get the most recent times so just return an empty slice
				return time.Time{}, time.Time{}, fmt.Errorf("Too many missed start times (> 100). Set or decrease .spec.startingDeadlineSeconds or check clock skew.")
			}
		}
		return lastMissed, sched.Next(now), nil
	}
	// 错过的 Cron 执行时间， nextRun 下一次执行的时间
	missedRun, nextRun, err := getNextSchedule(&cronJob, r.Now())
	if err != nil {
		log.Error(err, "unable to figure out CronJob schedule")
		// we don't really care about requeuing until we get an update that
		// fixes the schedule, so don't return an error
		return ctrl.Result{}, nil
	}

scheduledResult := ctrl.Result{RequeueAfter: nextRun.Sub(r.Now())} // save this so we can re-use it elsewhere
	// 日志添加当前时间（now）和下一次运行时间（next run）两个字段。
	log = log.WithValues("now", r.Now(), "next run", nextRun)
	// 没有错过任何调度。
	if missedRun.IsZero() {
		log.V(1).Info("no upcoming scheduled times, sleeping until next")
		return scheduledResult, nil
	}

log = log.WithValues("current run", missedRun)
	// 定义一个标志变量，表示“是否已经太迟不能补救”
	tooLate := false
	if cronJob.Spec.StartingDeadlineSeconds != nil {
		// 当前时间是否已经超过了 missedRun + startingDeadlineSeconds
		tooLate = missedRun.Add(time.Duration(*cronJob.Spec.StartingDeadlineSeconds) * time.Second).Before(r.Now())
	}
	if tooLate {
		// “太迟”，就打印日志，什么都不做，等待下次调度
		log.V(1).Info("missed starting deadline for last run, sleeping till next")
		// TODO(directxman12): events
		return scheduledResult, nil
	}

// 如果当前 CronJob 的策略是禁止并发（ForbidConcurrent），并且还有未完成的 Job, 直接跳过本次调度
	if cronJob.Spec.ConcurrencyPolicy == batchv1.ForbidConcurrent && len(activeJobs) > 0 {
		log.V(1).Info("concurrency policy blocks concurrent runs, skipping", "num active", len(activeJobs))
		return scheduledResult, nil
	}

	// ...or instruct us to replace existing ones...
	// 替换运行中的 Job
	if cronJob.Spec.ConcurrencyPolicy == batchv1.ReplaceConcurrent {
		// 当前所有还在运行的 Job 全部删除
		for _, activeJob := range activeJobs {
			// we don't care if the job was already deleted
			if err := r.Delete(ctx, activeJob, client.PropagationPolicy(metav1.DeletePropagationBackground)); client.IgnoreNotFound(err) != nil {
				log.Error(err, "unable to delete active job", "job", activeJob)
				return ctrl.Result{}, err
			}
		}
	}

// 创建实际 Job 的核心逻辑
	constructJobForCronJob := func(cronJob *batchv1.CronJob, scheduledTime time.Time) (*kbatch.Job, error) {
		// We want job names for a given nominal start time to have a deterministic name to avoid the same job being created twice
		name := fmt.Sprintf("%s-%d", cronJob.Name, scheduledTime.Unix())

		job := &kbatch.Job{
			ObjectMeta: metav1.ObjectMeta{
				Labels:      make(map[string]string),
				Annotations: make(map[string]string),
				Name:        name,
				Namespace:   cronJob.Namespace,
			},
			Spec: *cronJob.Spec.JobTemplate.Spec.DeepCopy(),
		}
		for k, v := range cronJob.Spec.JobTemplate.Annotations {
			job.Annotations[k] = v
		}
		// 注解添加执行的时间
		job.Annotations[scheduledTimeAnnotation] = scheduledTime.Format(time.RFC3339)
		for k, v := range cronJob.Spec.JobTemplate.Labels {
			job.Labels[k] = v
		}
		// 表示这个 Job 是由哪个 CronJob 控制的
		if err := ctrl.SetControllerReference(cronJob, job, r.Scheme); err != nil {
			return nil, err
		}

		return job, nil
	}
	// actually make the job...
	// 构造 Job
	job, err := constructJobForCronJob(&cronJob, missedRun)
	if err != nil {
		log.Error(err, "unable to construct job from template")
		// don't bother requeuing until we get a change to the spec
		return scheduledResult, nil
	}

	// ...and create it on the cluster
	// 集群创建job
	if err := r.Create(ctx, job); err != nil {
		log.Error(err, "unable to create Job for CronJob", "job", job)
		return ctrl.Result{}, err
	}

	log.V(1).Info("created Job for CronJob run", "job", job)

	return scheduledResult, nil

missedRun, nextRun, err := getNextSchedule(&cronJob, r.Now())

type Clock interface {
	Now() time.Time
}

type realClock struct{}

func (realClock) Now() time.Time {
	return time.Now()
}

r.List(ctx, &childJobs, client.InNamespace(req.Namespace), client.MatchingFields{jobOwnerKey: req.Name})

var (
	jobOwnerKey = ".metadata.controller"
	apiGVStr    = batchv1.GroupVersion.String()
)

if err := mgr.GetFieldIndexer().IndexField(context.Background(), &kbatch.Job{}, jobOwnerKey,
		func(rawObj client.Object) []string {
			// grab the job object, extract the owner...
			// 首先将原始对象转换为 kbatch.Job 类型
			job := rawObj.(*kbatch.Job)
			owner := metav1.GetControllerOf(job)
			if owner == nil {
				return nil
			}
			// ...make sure it's a CronJob...
			if owner.APIVersion != apiGVStr || owner.Kind != "CronJob" {
				return nil
			}

			// ...and if so, return it
			return []string{owner.Name}
		}); err != nil {
		return err
	}

return ctrl.NewControllerManagedBy(mgr).
    For(&batchv1.CronJob{}).
    Owns(&kbatch.Job{}).
    Named("cronjob").
    Complete(r)

var (
    scheme   = runtime.NewScheme()
    setupLog = ctrl.Log.WithName("setup")
)

func init() {
    _ = clientgoscheme.AddToScheme(scheme)

    _ = batchv1.AddToScheme(scheme)
    // +kubebuilder:scaffold:scheme
}

if err = (&controller.CronJobReconciler{
        Client: mgr.GetClient(),
        Scheme: mgr.GetScheme(),
    }).SetupWithManager(mgr); err != nil {
        setupLog.Error(err, "unable to create controller", "controller", "CronJob")
        os.Exit(1)
    }

if os.Getenv("ENABLE_WEBHOOKS") != "false" {
        if err = webhookbatchv1.SetupCronJobWebhookWithManager(mgr); err != nil {
            setupLog.Error(err, "unable to create webhook", "webhook", "CronJob")
            os.Exit(1)
        }
    }

Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error)

if result.Requeue {
    queue.AddRateLimited(req)
} else if result.RequeueAfter > 0 {
    queue.AddAfter(req, result.RequeueAfter)
}

if err := (&controllers.CronJobReconciler{
	Client: mgr.GetClient(),
	// ...
}).SetupWithManager(mgr); err != nil {
	// handle error
}

if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
		setupLog.Error(err, "problem running manager")
		os.Exit(1)
	}

main() 开始
   ↓
SetupWithManager() 执行（注册控制器）
   ↓
mgr.Start() 启动 manager，控制循环开始
   ↓
监听资源变更（informer）
   ↓
当有资源变化 -> 调用 Reconcile()

kubebuilder create webhook --group batch --version v1 --kind CronJob --defaulting --programmatic-validation