Apache License

http://www.apache.org/licenses/LICENSE-2.0

Imports

package v1alpha1

import (
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

// EDIT THIS FILE!  THIS IS SCAFFOLDING FOR YOU TO OWN!
// NOTE: json tags are required.  Any new fields you add must have json tags for the fields to be serialized.

// MemcachedSpec defines the desired state of Memcached.
type MemcachedSpec struct {
    // 插入额外的规格字段 - 集群的期望状态
    // 注意：修改此文件后，运行"make"以重新生成代码

    // Size defines the number of Memcached instances
    // The following markers will use OpenAPI v3 schema to validate the value
    // More info: https://book.kubebuilder.io/reference/markers/crd-validation.html
    // +kubebuilder:validation:Minimum=1
    // +kubebuilder:validation:Maximum=3
    // +kubebuilder:validation:ExclusiveMaximum=false
    Size int32 `json:"size,omitempty"`
}

// MemcachedStatus defines the observed state of Memcached.
type MemcachedStatus struct {
    // Represents the observations of a Memcached's current state.
    // Memcached.status.conditions.type are: "Available", "Progressing", and "Degraded"
    // Memcached.status.conditions.status are one of True, False, Unknown.
    // Memcached.status.conditions.reason the value should be a CamelCase string and producers of specific
    // condition types may define expected values and meanings for this field, and whether the values
    // are considered a guaranteed API.
    // Memcached.status.conditions.Message is a human readable message indicating details about the transition.
    // For further information see: https://github.com/kubernetes/community/blob/master/contributors/devel/sig-architecture/api-conventions.md#typical-status-properties

    Conditions []metav1.Condition `json:"conditions,omitempty" patchStrategy:"merge" patchMergeKey:"type" protobuf:"bytes,1,rep,name=conditions"`
}

// +kubebuilder:object:root=true
// +kubebuilder:subresource:status

// Memcached is the Schema for the memcacheds API.
type Memcached struct {
    metav1.TypeMeta   `json:",inline"`
    metav1.ObjectMeta `json:"metadata,omitempty"`

    Spec   MemcachedSpec   `json:"spec,omitempty"`
    Status MemcachedStatus `json:"status,omitempty"`
}

// +kubebuilder:object:root=true

// MemcachedList contains a list of Memcached.
type MemcachedList struct {
    metav1.TypeMeta `json:",inline"`
    metav1.ListMeta `json:"metadata,omitempty"`
    Items           []Memcached `json:"items"`
}

func init() {
    SchemeBuilder.Register(&Memcached{}, &MemcachedList{})
}

Apache License

http://www.apache.org/licenses/LICENSE-2.0

Our package starts out with some basic imports. Particularly:

• The core controller-runtime library
• The default controller-runtime logging, Zap (more on that a bit later)

package main

import (
    "flag"
    "os"

    // 导入所有 Kubernetes 客户端身份验证插件（例如 Azure、GCP、OIDC 等）
    // 确保 exec-entrypoint 和 run 可以使用它们。
    _ "k8s.io/client-go/plugin/pkg/client/auth"

    "k8s.io/apimachinery/pkg/runtime"
    utilruntime "k8s.io/apimachinery/pkg/util/runtime"
    clientgoscheme "k8s.io/client-go/kubernetes/scheme"
    _ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
    ctrl "sigs.k8s.io/controller-runtime"
    "sigs.k8s.io/controller-runtime/pkg/cache"
    "sigs.k8s.io/controller-runtime/pkg/healthz"
    "sigs.k8s.io/controller-runtime/pkg/log/zap"
    "sigs.k8s.io/controller-runtime/pkg/metrics/server"
    "sigs.k8s.io/controller-runtime/pkg/webhook"
    // +kubebuilder:scaffold:imports
)

Every set of controllers needs a Scheme, which provides mappings between Kinds and their corresponding Go types. We’ll talk a bit more about Kinds when we write our API definition, so just keep this in mind for later.

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

func init() {
    utilruntime.Must(clientgoscheme.AddToScheme(scheme))

    // +kubebuilder:scaffold:scheme}

At this point, our main function is fairly simple:

• We set up some basic flags for metrics.

• We instantiate a manager, which keeps track of running all of our controllers, as well as setting up shared caches and clients to the API server (notice we tell the manager about our Scheme).

• We run our manager, which in turn runs all of our controllers and webhooks. The manager is set up to run until it receives a graceful shutdown signal. This way, when we’re running on Kubernetes, we behave nicely with graceful pod termination.

While we don’t have anything to run just yet, remember where that +kubebuilder:scaffold:builder comment is – things’ll get interesting there soon.

func main() {
    var metricsAddr string
    var enableLeaderElection bool
    var probeAddr string
    flag.StringVar(&metricsAddr, "metrics-bind-address", ":8080", "The address the metric endpoint binds to.")
    flag.StringVar(&probeAddr, "health-probe-bind-address", ":8081", "The address the probe endpoint binds to.")
    flag.BoolVar(&enableLeaderElection, "leader-elect", false,
        "Enable leader election for controller manager. "+
            "启用此功能将确保只有一个活动的控制管理器。")
    opts := zap.Options{
        Development: true,
    }
    opts.BindFlags(flag.CommandLine)
    flag.Parse()

    ctrl.SetLogger(zap.New(zap.UseFlagOptions(&opts)))

    mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
        Scheme: scheme,
        Metrics: server.Options{
            BindAddress: metricsAddr,
        },
        WebhookServer:          webhook.NewServer(webhook.Options{Port: 9443}),
        HealthProbeBindAddress: probeAddr,
        LeaderElection:         enableLeaderElection,
        LeaderElectionID:       "80807133.tutorial.kubebuilder.io",
    })
    if err != nil {
        setupLog.Error(err, "unable to start manager")
        os.Exit(1)
    }

Note that the Manager can restrict the namespace that all controllers will watch for resources by:

    mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
        Scheme: scheme,
        Cache: cache.Options{
            DefaultNamespaces: map[string]cache.Config{
                namespace: {},
            },
        },
        Metrics: server.Options{
            BindAddress: metricsAddr,
        },
        WebhookServer:          webhook.NewServer(webhook.Options{Port: 9443}),
        HealthProbeBindAddress: probeAddr,
        LeaderElection:         enableLeaderElection,
        LeaderElectionID:       "80807133.tutorial.kubebuilder.io",
    })

The above example will change the scope of your project to a single Namespace. In this scenario, it is also suggested to restrict the provided authorization to this namespace by replacing the default ClusterRole and ClusterRoleBinding to Role and RoleBinding respectively. For further information see the Kubernetes documentation about Using RBAC Authorization.

Also, it is possible to use the DefaultNamespaces from cache.Options{} to cache objects in a specific set of namespaces:

    var namespaces []string // List of Namespaces
    defaultNamespaces := make(map[string]cache.Config)

    for _, ns := range namespaces {
        defaultNamespaces[ns] = cache.Config{}
    }

    mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
        Scheme: scheme,
        Cache: cache.Options{
            DefaultNamespaces: defaultNamespaces,
        },
        Metrics: server.Options{
            BindAddress: metricsAddr,
        },
        WebhookServer:          webhook.NewServer(webhook.Options{Port: 9443}),
        HealthProbeBindAddress: probeAddr,
        LeaderElection:         enableLeaderElection,
        LeaderElectionID:       "80807133.tutorial.kubebuilder.io",
    })

For further information see cache.Options{}

    // +kubebuilder:scaffold:builder

    if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
        setupLog.Error(err, "unable to set up health check")
        os.Exit(1)
    }
    if err := mgr.AddReadyzCheck("readyz", healthz.Ping); err != nil {
        setupLog.Error(err, "unable to set up ready check")
        os.Exit(1)
    }

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

Apache License

http://www.apache.org/licenses/LICENSE-2.0

We start out simply enough: we import the meta/v1 API group, which is not normally exposed by itself, but instead contains metadata common to all Kubernetes Kinds.

package v1

import (
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

Next, we define types for the Spec and Status of our Kind. Kubernetes functions by reconciling desired state (Spec) with actual cluster state (other objects’ Status) and external state, and then recording what it observed (Status). Thus, every functional object includes spec and status. A few types, like ConfigMap don’t follow this pattern, since they don’t encode desired state, but most types do.

// EDIT THIS FILE!  THIS IS SCAFFOLDING FOR YOU TO OWN!
// NOTE: json tags are required.  Any new fields you add must have json tags for the fields to be serialized.

// CronJobSpec defines the desired state of CronJob
type CronJobSpec struct {
    // 插入额外的规格字段 - 集群的期望状态
    // 注意：修改此文件后，运行"make"以重新生成代码
}

// CronJobStatus defines the observed state of CronJob
type CronJobStatus struct {
    // 插入额外的状态字段 - 定义集群的观察状态
    // 重要：在修改此文件后，请运行 "make" 以重新生成代码
}

Next, we define the types corresponding to actual Kinds, CronJob and CronJobList. CronJob is our root type, and describes the CronJob kind. Like all Kubernetes objects, it contains TypeMeta (which describes API version and Kind), and also contains ObjectMeta, which holds things like name, namespace, and labels.

CronJobList is simply a container for multiple CronJobs. It’s the Kind used in bulk operations, like LIST.

In general, we never modify either of these – all modifications go in either Spec or Status.

That little +kubebuilder:object:root comment is called a marker. We’ll see more of them in a bit, but know that they act as extra metadata, telling controller-tools (our code and YAML generator) extra information. This particular one tells the object generator that this type represents a Kind. Then, the object generator generates an implementation of the runtime.Object interface for us, which is the standard interface that all types representing Kinds must implement.

// +kubebuilder:object:root=true
// +kubebuilder:subresource:status

// 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 包含一个 CronJob 列表
type CronJobList struct {
    metav1.TypeMeta `json:",inline"`
    metav1.ListMeta `json:"metadata,omitempty"`
    Items           []CronJob `json:"items"`
}

Finally, we add the Go types to the API group. This allows us to add the types in this API group to any Scheme.

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

Apache License

http://www.apache.org/licenses/LICENSE-2.0

package v1

Imports

import (
    batchv1 "k8s.io/api/batch/v1"
    corev1 "k8s.io/api/core/v1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

// EDIT THIS FILE!  THIS IS SCAFFOLDING FOR YOU TO OWN!
// NOTE: json tags are required.  Any new fields you add must have json tags for the fields to be serialized.

First, let’s take a look at our spec. As we discussed before, spec holds desired state, so any “inputs” to our controller go here.

Fundamentally a CronJob needs the following pieces:

• A schedule (the cron in CronJob)
• A template for the Job to run (the job in CronJob)

We’ll also want a few extras, which will make our users’ lives easier:

• A deadline for starting jobs (if we miss this deadline, we’ll just wait till the next scheduled time)
• What to do if multiple jobs would run at once (do we wait? stop the old one? run both?)
• A way to pause the running of a CronJob, in case something’s wrong with it
• Limits on old job history

Remember, since we never read our own status, we need to have some other way to keep track of whether a job has run. We can use at least one old job to do this.

We’ll use several markers (// +comment) to specify additional metadata. These will be used by controller-tools when generating our CRD manifest. As we’ll see in a bit, controller-tools will also use GoDoc to form descriptions for the fields.

// CronJobSpec defines the desired state of CronJob.
type CronJobSpec struct {
    // +kubebuilder:validation:MinLength=0

    // The schedule in Cron format, see https://en.wikipedia.org/wiki/Cron.
    Schedule string `json:"schedule"`

    // +kubebuilder:validation:Minimum=0

    // Optional deadline in seconds for starting the job if it misses scheduled
    // time for any reason.  Missed jobs executions will be counted as failed ones.
    // +optional
    StartingDeadlineSeconds *int64 `json:"startingDeadlineSeconds,omitempty"`

    // Specifies how to treat concurrent executions of a Job.
    // Valid values are:
    // - "Allow" (default): allows CronJobs to run concurrently;
    // - "Forbid": forbids concurrent runs, skipping next run if previous run hasn't finished yet;
    // - "Replace": cancels currently running job and replaces it with a new one
    // +optional
    ConcurrencyPolicy ConcurrencyPolicy `json:"concurrencyPolicy,omitempty"`

    // This flag tells the controller to suspend subsequent executions, it does
    // not apply to already started executions.  Defaults to false.
    // +optional
    Suspend *bool `json:"suspend,omitempty"`

    // Specifies the job that will be created when executing a CronJob.
    JobTemplate batchv1.JobTemplateSpec `json:"jobTemplate"`

    // +kubebuilder:validation:Minimum=0

    // The number of successful finished jobs to retain.
    // This is a pointer to distinguish between explicit zero and not specified.
    // +optional
    SuccessfulJobsHistoryLimit *int32 `json:"successfulJobsHistoryLimit,omitempty"`

    // +kubebuilder:validation:Minimum=0

    // The number of failed finished jobs to retain.
    // This is a pointer to distinguish between explicit zero and not specified.
    // +optional
    FailedJobsHistoryLimit *int32 `json:"failedJobsHistoryLimit,omitempty"`
}

We define a custom type to hold our concurrency policy. It’s actually just a string under the hood, but the type gives extra documentation, and allows us to attach validation on the type instead of the field, making the validation more easily reusable.

// ConcurrencyPolicy describes how the job will be handled.
// Only one of the following concurrent policies may be specified.
// If none of the following policies is specified, the default one
// is AllowConcurrent.
// +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"
)

Next, let’s design our status, which holds observed state. It contains any information we want users or other controllers to be able to easily obtain.

We’ll keep a list of actively running jobs, as well as the last time that we successfully ran our job. Notice that we use metav1.Time instead of time.Time to get the stable serialization, as mentioned above.

// CronJobStatus defines the observed state of CronJob.
type CronJobStatus struct {
    // 插入额外的状态字段 - 定义集群的观察状态
    // 重要：在修改此文件后，请运行 "make" 以重新生成代码

    // A list of pointers to currently running jobs.
    // +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"`
}

Finally, we have the rest of the boilerplate that we’ve already discussed. As previously noted, we don’t need to change this, except to mark that we want a status subresource, so that we behave like built-in kubernetes types.

// +kubebuilder:object:root=true
// +kubebuilder:subresource:status

// CronJob is the Schema for the cronjobs API.
type CronJob struct {

Root Object Definitions

    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{})
}

Apache License

http://www.apache.org/licenses/LICENSE-2.0

First, we have some package-level markers that denote that there are Kubernetes objects in this package, and that this package represents the group batch.tutorial.kubebuilder.io. The object generator makes use of the former, while the latter is used by the CRD generator to generate the right metadata for the CRDs it creates from this package.

// Package v1 contains API Schema definitions for the batch v1 API group.
// +kubebuilder:object:generate=true
// +groupName=batch.tutorial.kubebuilder.io
package v1

import (
    "k8s.io/apimachinery/pkg/runtime/schema"
    "sigs.k8s.io/controller-runtime/pkg/scheme"
)

Then, we have the commonly useful variables that help us set up our Scheme. Since we need to use all the types in this package in our controller, it’s helpful (and the convention) to have a convenient method to add all the types to some other Scheme. SchemeBuilder makes this easy for us.

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
)

Apache License

http://www.apache.org/licenses/LICENSE-2.0

First, we start out with some standard imports. As before, we need the core controller-runtime library, as well as the client package, and the package for our API types.

package controllers

import (
    "context"

    "k8s.io/apimachinery/pkg/runtime"
    ctrl "sigs.k8s.io/controller-runtime"
    "sigs.k8s.io/controller-runtime/pkg/client"
    logf "sigs.k8s.io/controller-runtime/pkg/log"

    batchv1 "tutorial.kubebuilder.io/project/api/v1"
)

Next, kubebuilder has scaffolded a basic reconciler struct for us. Pretty much every reconciler needs to log, and needs to be able to fetch objects, so these are added out of the box.

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

Most controllers eventually end up running on the cluster, so they need RBAC permissions, which we specify using controller-tools RBAC markers. These are the bare minimum permissions needed to run. As we add more functionality, we’ll need to revisit these.

// +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

The ClusterRole manifest at config/rbac/role.yaml is generated from the above markers via controller-gen with the following command:

// make manifests

NOTE: If you receive an error, please run the specified command in the error and re-run make manifests.

Reconcile actually performs the reconciling for a single named object. Our Request just has a name, but we can use the client to fetch that object from the cache.

We return an empty result and no error, which indicates to controller-runtime that we’ve successfully reconciled this object and don’t need to try again until there’s some changes.

Most controllers need a logging handle and a context, so we set them up here.

The context is used to allow cancellation of requests, and potentially things like tracing. It’s the first argument to all client methods. The Background context is just a basic context without any extra data or timing restrictions.

The logging handle lets us log. controller-runtime uses structured logging through a library called logr. As we’ll see shortly, logging works by attaching key-value pairs to a static message. We can pre-assign some pairs at the top of our reconcile method to have those attached to all log lines in this reconciler.

func (r *CronJobReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
    _ = logf.FromContext(ctx)

    // your logic here

    return ctrl.Result{}, nil
}

Finally, we add this reconciler to the manager, so that it gets started when the manager is started.

For now, we just note that this reconciler operates on CronJobs. Later, we’ll use this to mark that we care about related objects as well.

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

Apache License

http://www.apache.org/licenses/LICENSE-2.0

We’ll start out with some imports. You’ll see below that we’ll need a few more imports than those scaffolded for us. We’ll talk about each one when we use it.

package controller

import (
    "context"
    "fmt"
    "sort"
    "time"

    "github.com/robfig/cron"
    kbatch "k8s.io/api/batch/v1"
    corev1 "k8s.io/api/core/v1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/apimachinery/pkg/runtime"
    ref "k8s.io/client-go/tools/reference"
    ctrl "sigs.k8s.io/controller-runtime"
    "sigs.k8s.io/controller-runtime/pkg/client"
    logf "sigs.k8s.io/controller-runtime/pkg/log"

    batchv1 "tutorial.kubebuilder.io/project/api/v1"
)

Next, we’ll need a Clock, which will allow us to fake timing in our tests.

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

Clock

type realClock struct{}

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

// Clock knows how to get the current time.
// It can be used to fake out timing for testing.
type Clock interface {
    Now() time.Time
}

Notice that we need a few more RBAC permissions – since we’re creating and managing jobs now, we’ll need permissions for those, which means adding a couple more markers.

// +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
// +kubebuilder:rbac:groups=batch.tutorial.kubebuilder.io,resources=cronjobs/finalizers,verbs=update
// +kubebuilder:rbac:groups=batch,resources=jobs,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=batch,resources=jobs/status,verbs=get

Now, we get to the heart of the controller – the reconciler logic.

var (
    scheduledTimeAnnotation = "batch.tutorial.kubebuilder.io/scheduled-at"
)

// Reconcile is part of the main kubernetes reconciliation loop which aims to
// move the current state of the cluster closer to the desired state.
// TODO(user): Modify the Reconcile function to compare the state specified by
// the CronJob object against the actual cluster state, and then
// perform operations to make the cluster state reflect the state specified by
// the user.
//
// For more details, check Reconcile and its Result here:
// - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.20.4/pkg/reconcile
// nolint:gocyclo
func (r *CronJobReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
    log := logf.FromContext(ctx)

1: Load the CronJob by name

We’ll fetch the CronJob using our client. All client methods take a context (to allow for cancellation) as their first argument, and the object in question as their last. Get is a bit special, in that it takes a NamespacedName as the middle argument (most don’t have a middle argument, as we’ll see below).

Many client methods also take variadic options at the end.

    var cronJob batchv1.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)
    }

2: List all active jobs, and update the status

To fully update our status, we’ll need to list all child jobs in this namespace that belong to this CronJob. Similarly to Get, we can use the List method to list the child jobs. Notice that we use variadic options to set the namespace and field match (which is actually an index lookup that we set up below).

    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
    }

Once we have all the jobs we own, we’ll split them into active, successful, and failed jobs, keeping track of the most recent run so that we can record it in status. Remember, status should be able to be reconstituted from the state of the world, so it’s generally not a good idea to read from the status of the root object. Instead, you should reconstruct it every run. That’s what we’ll do here.

We can check if a job is “finished” and whether it succeeded or failed using status conditions. We’ll put that logic in a helper to make our code cleaner.

    // find the active list of jobs
    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

isJobFinished

    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 {
                return true, c.Type
            }
        }

        return false, ""
    }

getScheduledTimeForJob

    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
    }

    for i, job := range childJobs.Items {
        _, finishedType := isJobFinished(&job)
        switch finishedType {
        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.
        scheduledTimeForJob, err := getScheduledTimeForJob(&job)
        if err != nil {
            log.Error(err, "unable to parse schedule time for child job", "job", &job)
            continue
        }
        if scheduledTimeForJob != nil {
            if mostRecentTime == nil || mostRecentTime.Before(*scheduledTimeForJob) {
                mostRecentTime = scheduledTimeForJob
            }
        }
    }

    if mostRecentTime != nil {
        cronJob.Status.LastScheduleTime = &metav1.Time{Time: *mostRecentTime}
    } else {
        cronJob.Status.LastScheduleTime = nil
    }
    cronJob.Status.Active = nil
    for _, activeJob := range activeJobs {
        jobRef, err := ref.GetReference(r.Scheme, 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)
    }

Here, we’ll log how many jobs we observed at a slightly higher logging level, for debugging. Notice how instead of using a format string, we use a fixed message, and attach key-value pairs with the extra information. This makes it easier to filter and query log lines.

    log.V(1).Info("job count", "active jobs", len(activeJobs), "successful jobs", len(successfulJobs), "failed jobs", len(failedJobs))

Using the data we’ve gathered, we’ll update the status of our CRD. Just like before, we use our client. To specifically update the status subresource, we’ll use the Status part of the client, with the Update method.

The status subresource ignores changes to spec, so it’s less likely to conflict with any other updates, and can have separate permissions.

    if err := r.Status().Update(ctx, &cronJob); err != nil {
        log.Error(err, "unable to update CronJob status")
        return ctrl.Result{}, err
    }

Once we’ve updated our status, we can move on to ensuring that the status of the world matches what we want in our spec.

3: Clean up old jobs according to the history limit

First, we’ll try to clean up old jobs, so that we don’t leave too many lying around.

    // NB: deleting these are "best effort" -- if we fail on a particular one,
    // we won't requeue just to finish the deleting.
    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)
            }
        }
    }

4: Check if we’re suspended

If this object is suspended, we don’t want to run any jobs, so we’ll stop now. This is useful if something’s broken with the job we’re running and we want to pause runs to investigate or putz with the cluster, without deleting the object.

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

5: Get the next scheduled run

If we’re not paused, we’ll need to calculate the next scheduled run, and whether or not we’ve got a run that we haven’t processed yet.

getNextSchedule

    getNextSchedule := func(cronJob *batchv1.CronJob, now time.Time) (lastMissed time.Time, next time.Time, err error) {
        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.
        var earliestTime time.Time
        if cronJob.Status.LastScheduleTime != nil {
            earliestTime = cronJob.Status.LastScheduleTime.Time
        } else {
            earliestTime = cronJob.ObjectMeta.CreationTimestamp.Time
        }
        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))

            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
            // An object might miss several starts. For example, if
            // controller gets wedged on Friday at 5:01pm when everyone has
            // gone home, and someone comes in on Tuesday AM and discovers
            // the problem and restarts the controller, then all the hourly
            // jobs, more than 80 of them for one hourly scheduledJob, should
            // all start running with no further intervention (if the scheduledJob
            // allows concurrency and late starts).
            //
            // However, if there is a bug somewhere, or incorrect clock
            // on controller's server or apiservers (for setting creationTimestamp)
            // then there could be so many missed start times (it could be off
            // by decades or more), that it would eat up all the CPU and memory
            // of this controller. In that case, we want to not try to list
            // all the missed start times.
            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
    }

    // figure out the next times that we need to create
    // jobs at (or anything we missed).
    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
    }

We’ll prep our eventual request to requeue until the next job, and then figure out if we actually need to run.

    scheduledResult := ctrl.Result{RequeueAfter: nextRun.Sub(r.Now())} // save this so we can re-use it elsewhere
    log = log.WithValues("now", r.Now(), "next run", nextRun)

6: Run a new job if it’s on schedule, not past the deadline, and not blocked by our concurrency policy

If we’ve missed a run, and we’re still within the deadline to start it, we’ll need to run a job.

    if missedRun.IsZero() {
        log.V(1).Info("no upcoming scheduled times, sleeping until next")
        return scheduledResult, nil
    }

    // make sure we're not too late to start the run
    log = log.WithValues("current run", missedRun)
    tooLate := false
    if cronJob.Spec.StartingDeadlineSeconds != nil {
        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
    }

If we actually have to run a job, we’ll need to either wait till existing ones finish, replace the existing ones, or just add new ones. If our information is out of date due to cache delay, we’ll get a requeue when we get up-to-date information.

    // figure out how to run this job -- concurrency policy might forbid us from running
    // multiple at the same time...
    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...
    if cronJob.Spec.ConcurrencyPolicy == batchv1.ReplaceConcurrent {
        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
            }
        }
    }

Once we’ve figured out what to do with existing jobs, we’ll actually create our desired job

constructJobForCronJob

    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
        }
        if err := ctrl.SetControllerReference(cronJob, job, r.Scheme); err != nil {
            return nil, err
        }

        return job, nil
    }

    // actually make the 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
    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)

7: Requeue when we either see a running job or it’s time for the next scheduled run

Finally, we’ll return the result that we prepped above, that says we want to requeue when our next run would need to occur. This is taken as a maximum deadline – if something else changes in between, like our job starts or finishes, we get modified, etc, we might reconcile again sooner.

    // we'll requeue once we see the running job, and update our status
    return scheduledResult, nil
}

Setup

Finally, we’ll update our setup. In order to allow our reconciler to quickly look up Jobs by their owner, we’ll need an index. We declare an index key that we can later use with the client as a pseudo-field name, and then describe how to extract the indexed value from the Job object. The indexer will automatically take care of namespaces for us, so we just have to extract the owner name if the Job has a CronJob owner.

Additionally, we’ll inform the manager that this controller owns some Jobs, so that it will automatically call Reconcile on the underlying CronJob when a Job changes, is deleted, etc.

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

// SetupWithManager 将控制器与管理器进行设置。
func (r *CronJobReconciler) SetupWithManager(mgr ctrl.Manager) error {
    // set up a real clock, since we're not in a test
    if r.Clock == nil {
        r.Clock = realClock{}
    }

    if err := mgr.GetFieldIndexer().IndexField(context.Background(), &kbatch.Job{}, jobOwnerKey, func(rawObj client.Object) []string {
        // grab the job object, extract the owner...
        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)
}

Apache License

http://www.apache.org/licenses/LICENSE-2.0

Imports

package main

import (
    "crypto/tls"
    "flag"
    "os"
    "path/filepath"

    // 导入所有 Kubernetes 客户端身份验证插件（例如 Azure、GCP、OIDC 等）
    // 确保 exec-entrypoint 和 run 可以使用它们。
    _ "k8s.io/client-go/plugin/pkg/client/auth"

    "k8s.io/apimachinery/pkg/runtime"
    utilruntime "k8s.io/apimachinery/pkg/util/runtime"
    clientgoscheme "k8s.io/client-go/kubernetes/scheme"
    ctrl "sigs.k8s.io/controller-runtime"
    "sigs.k8s.io/controller-runtime/pkg/certwatcher"
    "sigs.k8s.io/controller-runtime/pkg/healthz"
    "sigs.k8s.io/controller-runtime/pkg/log/zap"
    "sigs.k8s.io/controller-runtime/pkg/metrics/filters"
    metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
    "sigs.k8s.io/controller-runtime/pkg/webhook"

    batchv1 "tutorial.kubebuilder.io/project/api/v1"
    "tutorial.kubebuilder.io/project/internal/controller"
    webhookbatchv1 "tutorial.kubebuilder.io/project/internal/webhook/v1"
    // +kubebuilder:scaffold:imports
)

The first difference to notice is that kubebuilder has added the new API group’s package (batchv1) to our scheme. This means that we can use those objects in our controller.

If we would be using any other CRD we would have to add their scheme the same way. Builtin types such as Job have their scheme added by clientgoscheme.

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

func init() {
    utilruntime.Must(clientgoscheme.AddToScheme(scheme))

    utilruntime.Must(batchv1.AddToScheme(scheme))
    // +kubebuilder:scaffold:scheme}

The other thing that’s changed is that kubebuilder has added a block calling our CronJob controller’s SetupWithManager method.

// nolint:gocyclo
func main() {

old stuff

    var metricsAddr string
    var metricsCertPath, metricsCertName, metricsCertKey string
    var webhookCertPath, webhookCertName, webhookCertKey string
    var enableLeaderElection bool
    var probeAddr string
    var secureMetrics bool
    var enableHTTP2 bool
    var tlsOpts []func(*tls.Config)
    flag.StringVar(&metricsAddr, "metrics-bind-address", "0", "The address the metrics endpoint binds to. "+
        "使用 :8443 进行 HTTPS，使用 :8080 进行 HTTP，或者保持为 0 以禁用指标服务。")
    flag.StringVar(&probeAddr, "health-probe-bind-address", ":8081", "The address the probe endpoint binds to.")
    flag.BoolVar(&enableLeaderElection, "leader-elect", false,
        "Enable leader election for controller manager. "+
            "启用此功能将确保只有一个活动的控制管理器。")
    flag.BoolVar(&secureMetrics, "metrics-secure", true,
        "如果设置，指标端点将通过 HTTPS 安全提供。请使用 --metrics-secure=false 以改为使用 HTTP。")
    flag.StringVar(&webhookCertPath, "webhook-cert-path", "", "The directory that contains the webhook certificate.")
    flag.StringVar(&webhookCertName, "webhook-cert-name", "tls.crt", "The name of the webhook certificate file.")
    flag.StringVar(&webhookCertKey, "webhook-cert-key", "tls.key", "The name of the webhook key file.")
    flag.StringVar(&metricsCertPath, "metrics-cert-path", "",
        "The directory that contains the metrics server certificate.")
    flag.StringVar(&metricsCertName, "metrics-cert-name", "tls.crt", "The name of the metrics server certificate file.")
    flag.StringVar(&metricsCertKey, "metrics-cert-key", "tls.key", "The name of the metrics server key file.")
    flag.BoolVar(&enableHTTP2, "enable-http2", false,
        "If set, HTTP/2 will be enabled for the metrics and webhook servers")
    opts := zap.Options{
        Development: true,
    }
    opts.BindFlags(flag.CommandLine)
    flag.Parse()

    ctrl.SetLogger(zap.New(zap.UseFlagOptions(&opts)))

    // 如果 enable-http2 标志为 false（默认值），则应禁用 http/2
    // 由于其存在的漏洞。更具体地说，禁用 http/2 将
    // 防止受到 HTTP/2 流取消和
    // 快速重置 CVE 的漏洞影响。有关更多信息，请参见：
    // - https://github.com/advisories/GHSA-qppj-fm5r-hxr3
    // - https://github.com/advisories/GHSA-4374-p667-p6c8
    disableHTTP2 := func(c *tls.Config) {
        setupLog.Info("disabling http/2")
        c.NextProtos = []string{"http/1.1"}
    }

    if !enableHTTP2 {
        tlsOpts = append(tlsOpts, disableHTTP2)
    }

    // 创建用于指标和 Webhook 证书的监视器
    var metricsCertWatcher, webhookCertWatcher *certwatcher.CertWatcher

    // 初始Webhook TLS选项
    webhookTLSOpts := tlsOpts

    if len(webhookCertPath) > 0 {
        setupLog.Info("使用提供的证书初始化 Webhook 证书监视器",
            "webhook-cert-path", webhookCertPath, "webhook-cert-name", webhookCertName, "webhook-cert-key", webhookCertKey)

        var err error
        webhookCertWatcher, err = certwatcher.New(
            filepath.Join(webhookCertPath, webhookCertName),
            filepath.Join(webhookCertPath, webhookCertKey),
        )
        if err != nil {
            setupLog.Error(err, "Failed to initialize webhook certificate watcher")
            os.Exit(1)
        }

        webhookTLSOpts = append(webhookTLSOpts, func(config *tls.Config) {
            config.GetCertificate = webhookCertWatcher.GetCertificate
        })
    }

    webhookServer := webhook.NewServer(webhook.Options{
        TLSOpts: webhookTLSOpts,
    })

    // 指标端点在 'config/default/kustomization.yaml' 中启用。指标选项配置服务器。
    // 更多信息：
    // - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.20.4/pkg/metrics/server
    // - https://book.kubebuilder.io/reference/metrics.html
    metricsServerOptions := metricsserver.Options{
        BindAddress:   metricsAddr,
        SecureServing: secureMetrics,
        TLSOpts:       tlsOpts,
    }

    if secureMetrics {
        // FilterProvider 用于通过认证和授权来保护指标端点。
        // 这些配置确保只有授权用户和服务账户
        // 可以访问指标端点。RBAC 配置在 'config/rbac/kustomization.yaml' 中。更多信息：
        // https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.20.4/pkg/metrics/filters#WithAuthenticationAndAuthorization
        metricsServerOptions.FilterProvider = filters.WithAuthenticationAndAuthorization
    }

    // 如果未指定证书，controller-runtime 将自动
    // 为指标服务器生成自签名证书。虽然对于开发和测试来说很方便，
    // 但这种设置不建议用于生产环境。
    //
    // TODO(用户)：如果您启用 certManager，请取消注释以下行：
    // - [METRICS-WITH-CERTS] 在 config/default/kustomization.yaml 中生成并使用
    // 由 cert-manager 管理的指标服务器证书。
    // - [PROMETHEUS-WITH-CERTS] 在 config/prometheus/kustomization.yaml 中用于 TLS 认证。
    if len(metricsCertPath) > 0 {
        setupLog.Info("使用提供的证书初始化指标证书监视器",
            "metrics-cert-path", metricsCertPath, "metrics-cert-name", metricsCertName, "metrics-cert-key", metricsCertKey)

        var err error
        metricsCertWatcher, err = certwatcher.New(
            filepath.Join(metricsCertPath, metricsCertName),
            filepath.Join(metricsCertPath, metricsCertKey),
        )
        if err != nil {
            setupLog.Error(err, "to initialize metrics certificate watcher", "error", err)
            os.Exit(1)
        }

        metricsServerOptions.TLSOpts = append(metricsServerOptions.TLSOpts, func(config *tls.Config) {
            config.GetCertificate = metricsCertWatcher.GetCertificate
        })
    }

    mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
        Scheme:                 scheme,
        Metrics:                metricsServerOptions,
        WebhookServer:          webhookServer,
        HealthProbeBindAddress: probeAddr,
        LeaderElection:         enableLeaderElection,
        LeaderElectionID:       "80807133.tutorial.kubebuilder.io",
        // LeaderElectionReleaseOnCancel 定义了当管理器结束时，领导者是否应自愿辞职
        // 当管理器停止时，这要求二进制文件立即结束，否则这个设置是不安全的。设置这个可以显著
        // 加快自愿领导者的过渡，因为新的领导者不必首先等待
        // LeaseDuration 时间。
        //
        // 在提供的默认脚手架中，程序会在
        // 管理器停止后立即结束，因此启用此选项是可以的。不过，
        // 如果您正在进行或打算在管理器停止后执行任何操作，如执行清理，
        // 那么使用它可能是不安全的。
        // LeaderElectionReleaseOnCancel: true,
    })
    if err != nil {
        setupLog.Error(err, "unable to start manager")
        os.Exit(1)
    }

    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)
    }

old stuff

    // nolint:goconst
    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)
        }
    }
    // +kubebuilder:scaffold:builder

    if metricsCertWatcher != nil {
        setupLog.Info("Adding metrics certificate watcher to manager")
        if err := mgr.Add(metricsCertWatcher); err != nil {
            setupLog.Error(err, "unable to add metrics certificate watcher to manager")
            os.Exit(1)
        }
    }

    if webhookCertWatcher != nil {
        setupLog.Info("Adding webhook certificate watcher to manager")
        if err := mgr.Add(webhookCertWatcher); err != nil {
            setupLog.Error(err, "unable to add webhook certificate watcher to manager")
            os.Exit(1)
        }
    }

    if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
        setupLog.Error(err, "unable to set up health check")
        os.Exit(1)
    }
    if err := mgr.AddReadyzCheck("readyz", healthz.Ping); err != nil {
        setupLog.Error(err, "unable to set up ready check")
        os.Exit(1)
    }

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

}

Apache License

http://www.apache.org/licenses/LICENSE-2.0

Go imports

package v1

import (
    "context"
    "fmt"

    "github.com/robfig/cron"
    apierrors "k8s.io/apimachinery/pkg/api/errors"
    "k8s.io/apimachinery/pkg/runtime/schema"
    validationutils "k8s.io/apimachinery/pkg/util/validation"
    "k8s.io/apimachinery/pkg/util/validation/field"

    "k8s.io/apimachinery/pkg/runtime"
    ctrl "sigs.k8s.io/controller-runtime"
    logf "sigs.k8s.io/controller-runtime/pkg/log"
    "sigs.k8s.io/controller-runtime/pkg/webhook"
    "sigs.k8s.io/controller-runtime/pkg/webhook/admission"

    batchv1 "tutorial.kubebuilder.io/project/api/v1"
)

Next, we’ll setup a logger for the webhooks.

var cronjoblog = logf.Log.WithName("cronjob-resource")

Then, we set up the webhook with the manager.

// SetupCronJobWebhookWithManager registers the webhook for CronJob in the manager.
func SetupCronJobWebhookWithManager(mgr ctrl.Manager) error {
    return ctrl.NewWebhookManagedBy(mgr).For(&batchv1.CronJob{}).
        WithValidator(&CronJobCustomValidator{}).
        WithDefaulter(&CronJobCustomDefaulter{
            DefaultConcurrencyPolicy:          batchv1.AllowConcurrent,
            DefaultSuspend:                    false,
            DefaultSuccessfulJobsHistoryLimit: 3,
            DefaultFailedJobsHistoryLimit:     1,
        }).
        Complete()
}

Notice that we use kubebuilder markers to generate webhook manifests. This marker is responsible for generating a mutating webhook manifest.

The meaning of each marker can be found here.

This marker is responsible for generating a mutation webhook manifest.

// +kubebuilder:webhook:path=/mutate-batch-tutorial-kubebuilder-io-v1-cronjob,mutating=true,failurePolicy=fail,sideEffects=None,groups=batch.tutorial.kubebuilder.io,resources=cronjobs,verbs=create;update,versions=v1,name=mcronjob-v1.kb.io,admissionReviewVersions=v1

// CronJobCustomDefaulter struct is responsible for setting default values on the custom resource of the
// Kind CronJob when those are created or updated.
//
// NOTE: The +kubebuilder:object:generate=false marker prevents controller-gen from generating DeepCopy methods,
// as it is used only for temporary operations and does not need to be deeply copied.
type CronJobCustomDefaulter struct {

    // Default values for various CronJob fields
    DefaultConcurrencyPolicy          batchv1.ConcurrencyPolicy
    DefaultSuspend                    bool
    DefaultSuccessfulJobsHistoryLimit int32
    DefaultFailedJobsHistoryLimit     int32
}

var _ webhook.CustomDefaulter = &CronJobCustomDefaulter{}

We use the webhook.CustomDefaulterinterface to set defaults to our CRD. A webhook will automatically be served that calls this defaulting.

The Defaultmethod is expected to mutate the receiver, setting the defaults.

// Default implements webhook.CustomDefaulter so a webhook will be registered for the Kind CronJob.
func (d *CronJobCustomDefaulter) Default(ctx context.Context, obj runtime.Object) error {
    cronjob, ok := obj.(*batchv1.CronJob)

    if !ok {
        return fmt.Errorf("expected an CronJob object but got %T", obj)
    }
    cronjoblog.Info("Defaulting for CronJob", "name", cronjob.GetName())

    // Set default values
    d.applyDefaults(cronjob)
    return nil
}

// applyDefaults applies default values to CronJob fields.
func (d *CronJobCustomDefaulter) applyDefaults(cronJob *batchv1.CronJob) {
    if cronJob.Spec.ConcurrencyPolicy == "" {
        cronJob.Spec.ConcurrencyPolicy = d.DefaultConcurrencyPolicy
    }
    if cronJob.Spec.Suspend == nil {
        cronJob.Spec.Suspend = new(bool)
        *cronJob.Spec.Suspend = d.DefaultSuspend
    }
    if cronJob.Spec.SuccessfulJobsHistoryLimit == nil {
        cronJob.Spec.SuccessfulJobsHistoryLimit = new(int32)
        *cronJob.Spec.SuccessfulJobsHistoryLimit = d.DefaultSuccessfulJobsHistoryLimit
    }
    if cronJob.Spec.FailedJobsHistoryLimit == nil {
        cronJob.Spec.FailedJobsHistoryLimit = new(int32)
        *cronJob.Spec.FailedJobsHistoryLimit = d.DefaultFailedJobsHistoryLimit
    }
}

We can validate our CRD beyond what’s possible with declarative validation. Generally, declarative validation should be sufficient, but sometimes more advanced use cases call for complex validation.

For instance, we’ll see below that we use this to validate a well-formed cron schedule without making up a long regular expression.

If webhook.CustomValidator interface is implemented, a webhook will automatically be served that calls the validation.

The ValidateCreate, ValidateUpdate and ValidateDelete methods are expected to validate its receiver upon creation, update and deletion respectively. We separate out ValidateCreate from ValidateUpdate to allow behavior like making certain fields immutable, so that they can only be set on creation. ValidateDelete is also separated from ValidateUpdate to allow different validation behavior on deletion. Here, however, we just use the same shared validation for ValidateCreate and ValidateUpdate. And we do nothing in ValidateDelete, since we don’t need to validate anything on deletion.

This marker is responsible for generating a validation webhook manifest.

// +kubebuilder:webhook:path=/validate-batch-tutorial-kubebuilder-io-v1-cronjob,mutating=false,failurePolicy=fail,sideEffects=None,groups=batch.tutorial.kubebuilder.io,resources=cronjobs,verbs=create;update,versions=v1,name=vcronjob-v1.kb.io,admissionReviewVersions=v1

// CronJobCustomValidator struct is responsible for validating the CronJob resource
// when it is created, updated, or deleted.
//
// NOTE: The +kubebuilder:object:generate=false marker prevents controller-gen from generating DeepCopy methods,
// as this struct is used only for temporary operations and does not need to be deeply copied.
type CronJobCustomValidator struct {
    // TODO(user): Add more fields as needed for validation
}

var _ webhook.CustomValidator = &CronJobCustomValidator{}

// ValidateCreate implements webhook.CustomValidator so a webhook will be registered for the type CronJob.
func (v *CronJobCustomValidator) ValidateCreate(ctx context.Context, obj runtime.Object) (admission.Warnings, error) {
    cronjob, ok := obj.(*batchv1.CronJob)
    if !ok {
        return nil, fmt.Errorf("expected a CronJob object but got %T", obj)
    }
    cronjoblog.Info("Validation for CronJob upon creation", "name", cronjob.GetName())

    return nil, validateCronJob(cronjob)
}

// ValidateUpdate implements webhook.CustomValidator so a webhook will be registered for the type CronJob.
func (v *CronJobCustomValidator) ValidateUpdate(ctx context.Context, oldObj, newObj runtime.Object) (admission.Warnings, error) {
    cronjob, ok := newObj.(*batchv1.CronJob)
    if !ok {
        return nil, fmt.Errorf("expected a CronJob object for the newObj but got %T", newObj)
    }
    cronjoblog.Info("Validation for CronJob upon update", "name", cronjob.GetName())

    return nil, validateCronJob(cronjob)
}

// ValidateDelete implements webhook.CustomValidator so a webhook will be registered for the type CronJob.
func (v *CronJobCustomValidator) ValidateDelete(ctx context.Context, obj runtime.Object) (admission.Warnings, error) {
    cronjob, ok := obj.(*batchv1.CronJob)
    if !ok {
        return nil, fmt.Errorf("expected a CronJob object but got %T", obj)
    }
    cronjoblog.Info("Validation for CronJob upon deletion", "name", cronjob.GetName())

    // TODO(user): fill in your validation logic upon object deletion.

    return nil, nil
}

We validate the name and the spec of the CronJob.

// validateCronJob validates the fields of a CronJob object.
func validateCronJob(cronjob *batchv1.CronJob) error {
    var allErrs field.ErrorList
    if err := validateCronJobName(cronjob); err != nil {
        allErrs = append(allErrs, err)
    }
    if err := validateCronJobSpec(cronjob); err != nil {
        allErrs = append(allErrs, err)
    }
    if len(allErrs) == 0 {
        return nil
    }

    return apierrors.NewInvalid(
        schema.GroupKind{Group: "batch.tutorial.kubebuilder.io", Kind: "CronJob"},
        cronjob.Name, allErrs)
}

Some fields are declaratively validated by OpenAPI schema. You can find kubebuilder validation markers (prefixed with // +kubebuilder:validation) in the Designing an API section. You can find all of the kubebuilder supported markers for declaring validation by running controller-gen crd -w, or here.

func validateCronJobSpec(cronjob *batchv1.CronJob) *field.Error {
    // The field helpers from the kubernetes API machinery help us return nicely
    // structured validation errors.
    return validateScheduleFormat(
        cronjob.Spec.Schedule,
        field.NewPath("spec").Child("schedule"))
}

We’ll need to validate the cron schedule is well-formatted.

func validateScheduleFormat(schedule string, fldPath *field.Path) *field.Error {
    if _, err := cron.ParseStandard(schedule); err != nil {
        return field.Invalid(fldPath, schedule, err.Error())
    }
    return nil
}

Validate object name

func validateCronJobName(cronjob *batchv1.CronJob) *field.Error {
    if len(cronjob.ObjectMeta.Name) > validationutils.DNS1035LabelMaxLength-11 {
        // The job name length is 63 characters like all Kubernetes objects
        // (which must fit in a DNS subdomain). The cronjob controller appends
        // a 11-character suffix to the cronjob (`-$TIMESTAMP`) when creating
        // a job. The job name length limit is 63 characters. Therefore cronjob
        // names must have length <= 63-11=52. If we don't validate this here,
        // then job creation will fail later.
        return field.Invalid(field.NewPath("metadata").Child("name"), cronjob.ObjectMeta.Name, "must be no more than 52 characters")
    }
    return nil
}

Apache License

http://www.apache.org/licenses/LICENSE-2.0

Imports

package controller

import (
    "context"
    "os"
    "path/filepath"
    "testing"

    ctrl "sigs.k8s.io/controller-runtime"

    . "github.com/onsi/ginkgo/v2"
    . "github.com/onsi/gomega"

    "k8s.io/client-go/kubernetes/scheme"
    "k8s.io/client-go/rest"
    "sigs.k8s.io/controller-runtime/pkg/client"
    "sigs.k8s.io/controller-runtime/pkg/envtest"
    logf "sigs.k8s.io/controller-runtime/pkg/log"
    "sigs.k8s.io/controller-runtime/pkg/log/zap"

    batchv1 "tutorial.kubebuilder.io/project/api/v1"
    // +kubebuilder:scaffold:imports
)

// These tests use Ginkgo (BDD-style Go testing framework). Refer to
// http://onsi.github.io/ginkgo/ to learn more about Ginkgo.

Now, let’s go through the code generated.

var (
    ctx       context.Context
    cancel    context.CancelFunc
    testEnv   *envtest.Environment
    cfg       *rest.Config
    k8sClient client.Client // You'll be using this client in your tests.
)

func TestControllers(t *testing.T) {
    RegisterFailHandler(Fail)

    RunSpecs(t, "Controller Suite")
}

var _ = BeforeSuite(func() {
    logf.SetLogger(zap.New(zap.WriteTo(GinkgoWriter), zap.UseDevMode(true)))

    ctx, cancel = context.WithCancel(context.TODO())

    var err error

The CronJob Kind is added to the runtime scheme used by the test environment. This ensures that the CronJob API is registered with the scheme, allowing the test controller to recognize and interact with CronJob resources.

    err = batchv1.AddToScheme(scheme.Scheme)
    Expect(err).NotTo(HaveOccurred())

After the schemas, you will see the following marker. This marker is what allows new schemas to be added here automatically when a new API is added to the project.

    // +kubebuilder:scaffold:scheme

The envtest environment is configured to load Custom Resource Definitions (CRDs) from the specified directory. This setup enables the test environment to recognize and interact with the custom resources defined by these CRDs.

    By("bootstrapping test environment")
    testEnv = &envtest.Environment{
        CRDDirectoryPaths:     []string{filepath.Join("..", "..", "config", "crd", "bases")},
        ErrorIfCRDPathMissing: true,
    }

    // Retrieve the first found binary directory to allow running tests from IDEs
    if getFirstFoundEnvTestBinaryDir() != "" {
        testEnv.BinaryAssetsDirectory = getFirstFoundEnvTestBinaryDir()
    }

Then, we start the envtest cluster.

    // cfg is defined in this file globally.
    cfg, err = testEnv.Start()
    Expect(err).NotTo(HaveOccurred())
    Expect(cfg).NotTo(BeNil())

A client is created for our test CRUD operations.

    k8sClient, err = client.New(cfg, client.Options{Scheme: scheme.Scheme})
    Expect(err).NotTo(HaveOccurred())
    Expect(k8sClient).NotTo(BeNil())

One thing that this autogenerated file is missing, however, is a way to actually start your controller. The code above will set up a client for interacting with your custom Kind, but will not be able to test your controller behavior. If you want to test your custom controller logic, you’ll need to add some familiar-looking manager logic to your BeforeSuite() function, so you can register your custom controller to run on this test cluster.

You may notice that the code below runs your controller with nearly identical logic to your CronJob project’s main.go! The only difference is that the manager is started in a separate goroutine so it does not block the cleanup of envtest when you’re done running your tests.

Note that we set up both a “live” k8s client and a separate client from the manager. This is because when making assertions in tests, you generally want to assert against the live state of the API server. If you use the client from the manager (k8sManager.GetClient), you’d end up asserting against the contents of the cache instead, which is slower and can introduce flakiness into your tests. We could use the manager’s APIReader to accomplish the same thing, but that would leave us with two clients in our test assertions and setup (one for reading, one for writing), and it’d be easy to make mistakes.

Note that we keep the reconciler running against the manager’s cache client, though – we want our controller to behave as it would in production, and we use features of the cache (like indices) in our controller which aren’t available when talking directly to the API server.

    k8sManager, err := ctrl.NewManager(cfg, ctrl.Options{
        Scheme: scheme.Scheme,
    })
    Expect(err).ToNot(HaveOccurred())

    err = (&CronJobReconciler{
        Client: k8sManager.GetClient(),
        Scheme: k8sManager.GetScheme(),
    }).SetupWithManager(k8sManager)
    Expect(err).ToNot(HaveOccurred())

    go func() {
        defer GinkgoRecover()
        err = k8sManager.Start(ctx)
        Expect(err).ToNot(HaveOccurred(), "failed to run manager")
    }()
})

Kubebuilder also generates boilerplate functions for cleaning up envtest and actually running your test files in your controllers/ directory. You won’t need to touch these.

var _ = AfterSuite(func() {
    By("tearing down the test environment")
    cancel()
    err := testEnv.Stop()
    Expect(err).NotTo(HaveOccurred())
})

Now that you have your controller running on a test cluster and a client ready to perform operations on your CronJob, we can start writing integration tests!

// getFirstFoundEnvTestBinaryDir locates the first binary in the specified path.
// ENVTEST-based tests depend on specific binaries, usually located in paths set by
// controller-runtime. When running tests directly (e.g., via an IDE) without using
// Makefile targets, the 'BinaryAssetsDirectory' must be explicitly configured.
//
// This function streamlines the process by finding the required binaries, similar to
// setting the 'KUBEBUILDER_ASSETS' environment variable. To ensure the binaries are
// properly set up, run 'make setup-envtest' beforehand.
func getFirstFoundEnvTestBinaryDir() string {
    basePath := filepath.Join("..", "..", "bin", "k8s")
    entries, err := os.ReadDir(basePath)
    if err != nil {
        logf.Log.Error(err, "Failed to read directory", "path", basePath)
        return ""
    }
    for _, entry := range entries {
        if entry.IsDir() {
            return filepath.Join(basePath, entry.Name())
        }
    }
    return ""
}

Apache License

http://www.apache.org/licenses/LICENSE-2.0

Ideally, we should have one <kind>_controller_test.go for each controller scaffolded and called in the suite_test.go. So, let’s write our example test for the CronJob controller (cronjob_controller_test.go.)

Imports

package controller

import (
    "context"
    "reflect"
    "time"

    . "github.com/onsi/ginkgo/v2"
    . "github.com/onsi/gomega"
    batchv1 "k8s.io/api/batch/v1"
    v1 "k8s.io/api/core/v1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/apimachinery/pkg/types"

    cronjobv1 "tutorial.kubebuilder.io/project/api/v1"
)

The first step to writing a simple integration test is to actually create an instance of CronJob you can run tests against. Note that to create a CronJob, you’ll need to create a stub CronJob struct that contains your CronJob’s specifications.

Note that when we create a stub CronJob, the CronJob also needs stubs of its required downstream objects. Without the stubbed Job template spec and the Pod template spec below, the Kubernetes API will not be able to create the CronJob.

var _ = Describe("CronJob controller", func() {

    // Define utility constants for object names and testing timeouts/durations and intervals.
    const (
        CronjobName      = "test-cronjob"
        CronjobNamespace = "default"
        JobName          = "test-job"

        timeout  = time.Second * 10
        duration = time.Second * 10
        interval = time.Millisecond * 250
    )

    Context("When updating CronJob Status", func() {
        It("Should increase CronJob Status.Active count when new Jobs are created", func() {
            By("By creating a new CronJob")
            ctx := context.Background()
            cronJob := &cronjobv1.CronJob{
                TypeMeta: metav1.TypeMeta{
                    APIVersion: "batch.tutorial.kubebuilder.io/v1",
                    Kind:       "CronJob",
                },
                ObjectMeta: metav1.ObjectMeta{
                    Name:      CronjobName,
                    Namespace: CronjobNamespace,
                },
                Spec: cronjobv1.CronJobSpec{
                    Schedule: "1 * * * *",
                    JobTemplate: batchv1.JobTemplateSpec{
                        Spec: batchv1.JobSpec{
                            // For simplicity, we only fill out the required fields.
                            Template: v1.PodTemplateSpec{
                                Spec: v1.PodSpec{
                                    // For simplicity, we only fill out the required fields.
                                    Containers: []v1.Container{
                                        {
                                            Name:  "test-container",
                                            Image: "test-image",
                                        },
                                    },
                                    RestartPolicy: v1.RestartPolicyOnFailure,
                                },
                            },
                        },
                    },
                },
            }
            Expect(k8sClient.Create(ctx, cronJob)).To(Succeed())

        

After creating this CronJob, let’s check that the CronJob’s Spec fields match what we passed in. Note that, because the k8s apiserver may not have finished creating a CronJob after our Create() call from earlier, we will use Gomega’s Eventually() testing function instead of Expect() to give the apiserver an opportunity to finish creating our CronJob.

Eventually() will repeatedly run the function provided as an argument every interval seconds until (a) the assertions done by the passed-in Gomega succeed, or (b) the number of attempts * interval period exceed the provided timeout value.

In the examples below, timeout and interval are Go Duration values of our choosing.

            cronjobLookupKey := types.NamespacedName{Name: CronjobName, Namespace: CronjobNamespace}
            createdCronjob := &cronjobv1.CronJob{}

            // We'll need to retry getting this newly created CronJob, given that creation may not immediately happen.
            Eventually(func(g Gomega) {
                g.Expect(k8sClient.Get(ctx, cronjobLookupKey, createdCronjob)).To(Succeed())
            }, timeout, interval).Should(Succeed())
            // Let's make sure our Schedule string value was properly converted/handled.
            Expect(createdCronjob.Spec.Schedule).To(Equal("1 * * * *"))
        

Now that we’ve created a CronJob in our test cluster, the next step is to write a test that actually tests our CronJob controller’s behavior. Let’s test the CronJob controller’s logic responsible for updating CronJob.Status.Active with actively running jobs. We’ll verify that when a CronJob has a single active downstream Job, its CronJob.Status.Active field contains a reference to this Job.

First, we should get the test CronJob we created earlier, and verify that it currently does not have any active jobs. We use Gomega’s Consistently() check here to ensure that the active job count remains 0 over a duration of time.

            By("By checking the CronJob has zero active Jobs")
            Consistently(func(g Gomega) {
                g.Expect(k8sClient.Get(ctx, cronjobLookupKey, createdCronjob)).To(Succeed())
                g.Expect(createdCronjob.Status.Active).To(BeEmpty())
            }, duration, interval).Should(Succeed())
        

Next, we actually create a stubbed Job that will belong to our CronJob, as well as its downstream template specs. We set the Job’s status’s “Active” count to 2 to simulate the Job running two pods, which means the Job is actively running.

We then take the stubbed Job and set its owner reference to point to our test CronJob. This ensures that the test Job belongs to, and is tracked by, our test CronJob. Once that’s done, we create our new Job instance.

            By("By creating a new Job")
            testJob := &batchv1.Job{
                ObjectMeta: metav1.ObjectMeta{
                    Name:      JobName,
                    Namespace: CronjobNamespace,
                },
                Spec: batchv1.JobSpec{
                    Template: v1.PodTemplateSpec{
                        Spec: v1.PodSpec{
                            // For simplicity, we only fill out the required fields.
                            Containers: []v1.Container{
                                {
                                    Name:  "test-container",
                                    Image: "test-image",
                                },
                            },
                            RestartPolicy: v1.RestartPolicyOnFailure,
                        },
                    },
                },
            }

            // Note that your CronJob’s GroupVersionKind is required to set up this owner reference.
            kind := reflect.TypeOf(cronjobv1.CronJob{}).Name()
            gvk := cronjobv1.GroupVersion.WithKind(kind)

            controllerRef := metav1.NewControllerRef(createdCronjob, gvk)
            testJob.SetOwnerReferences([]metav1.OwnerReference{*controllerRef})
            Expect(k8sClient.Create(ctx, testJob)).To(Succeed())
            // Note that you can not manage the status values while creating the resource.
            // The status field is managed separately to reflect the current state of the resource.
            // Therefore, it should be updated using a PATCH or PUT operation after the resource has been created.
            // Additionally, it is recommended to use StatusConditions to manage the status. For further information see:
            // https://github.com/kubernetes/community/blob/master/contributors/devel/sig-architecture/api-conventions.md#spec-and-status
            testJob.Status.Active = 2
            Expect(k8sClient.Status().Update(ctx, testJob)).To(Succeed())
        

Adding this Job to our test CronJob should trigger our controller’s reconciler logic. After that, we can write a test that evaluates whether our controller eventually updates our CronJob’s Status field as expected!

            By("By checking that the CronJob has one active Job")
            Eventually(func(g Gomega) {
                g.Expect(k8sClient.Get(ctx, cronjobLookupKey, createdCronjob)).To(Succeed(), "should GET the CronJob")
                g.Expect(createdCronjob.Status.Active).To(HaveLen(1), "should have exactly one active job")
                g.Expect(createdCronjob.Status.Active[0].Name).To(Equal(JobName), "the wrong job is active")
            }, timeout, interval).Should(Succeed(), "should list our active job %s in the active jobs list in status", JobName)
        })
    })

})

After writing all this code, you can run go test ./... in your controllers/ directory again to run your new test!

Apache License

http://www.apache.org/licenses/LICENSE-2.0

Since we’re in a v2 package, controller-gen will assume this is for the v2 version automatically. We could override that with the +versionName marker.

package v2

Imports

import (
    batchv1 "k8s.io/api/batch/v1"
    corev1 "k8s.io/api/core/v1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

// EDIT THIS FILE!  THIS IS SCAFFOLDING FOR YOU TO OWN!
// NOTE: json tags are required.  Any new fields you add must have json tags for the fields to be serialized.

We’ll leave our spec largely unchanged, except to change the schedule field to a new type.

// CronJobSpec defines the desired state of CronJob.
type CronJobSpec struct {
    // The schedule in Cron format, see https://en.wikipedia.org/wiki/Cron.
    Schedule CronSchedule `json:"schedule"`

The rest of Spec

    // +kubebuilder:validation:Minimum=0

    // Optional deadline in seconds for starting the job if it misses scheduled
    // time for any reason.  Missed jobs executions will be counted as failed ones.
    // +optional
    StartingDeadlineSeconds *int64 `json:"startingDeadlineSeconds,omitempty"`

    // Specifies how to treat concurrent executions of a Job.
    // Valid values are:
    // - "Allow" (default): allows CronJobs to run concurrently;
    // - "Forbid": forbids concurrent runs, skipping next run if previous run hasn't finished yet;
    // - "Replace": cancels currently running job and replaces it with a new one
    // +optional
    ConcurrencyPolicy ConcurrencyPolicy `json:"concurrencyPolicy,omitempty"`

    // This flag tells the controller to suspend subsequent executions, it does
    // not apply to already started executions.  Defaults to false.
    // +optional
    Suspend *bool `json:"suspend,omitempty"`

    // Specifies the job that will be created when executing a CronJob.
    JobTemplate batchv1.JobTemplateSpec `json:"jobTemplate"`

    // +kubebuilder:validation:Minimum=0

    // The number of successful finished jobs to retain.
    // This is a pointer to distinguish between explicit zero and not specified.
    // +optional
    SuccessfulJobsHistoryLimit *int32 `json:"successfulJobsHistoryLimit,omitempty"`

    // +kubebuilder:validation:Minimum=0

    // The number of failed finished jobs to retain.
    // This is a pointer to distinguish between explicit zero and not specified.
    // +optional
    FailedJobsHistoryLimit *int32 `json:"failedJobsHistoryLimit,omitempty"`

}

Next, we’ll need to define a type to hold our schedule. Based on our proposed YAML above, it’ll have a field for each corresponding Cron “field”.

// describes a Cron schedule.
type CronSchedule struct {
    // specifies the minute during which the job executes.
    // +optional
    Minute *CronField `json:"minute,omitempty"`
    // specifies the hour during which the job executes.
    // +optional
    Hour *CronField `json:"hour,omitempty"`
    // specifies the day of the month during which the job executes.
    // +optional
    DayOfMonth *CronField `json:"dayOfMonth,omitempty"`
    // specifies the month during which the job executes.
    // +optional
    Month *CronField `json:"month,omitempty"`
    // specifies the day of the week during which the job executes.
    // +optional
    DayOfWeek *CronField `json:"dayOfWeek,omitempty"`
}

Finally, we’ll define a wrapper type to represent a field. We could attach additional validation to this field, but for now we’ll just use it for documentation purposes.

// represents a Cron field specifier.
type CronField string

Other Types

// ConcurrencyPolicy describes how the job will be handled.
// Only one of the following concurrent policies may be specified.
// If none of the following policies is specified, the default one
// is AllowConcurrent.
// +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"
)

// CronJobStatus defines the observed state of CronJob
type CronJobStatus struct {
    // 插入额外的状态字段 - 定义集群的观察状态
    // 重要：在修改此文件后，请运行 "make" 以重新生成代码

    // A list of pointers to currently running jobs.
    // +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
// +versionName=v2
// 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{})
}

Apache License

http://www.apache.org/licenses/LICENSE-2.0

package v1

Imports

import (
    batchv1 "k8s.io/api/batch/v1"
    corev1 "k8s.io/api/core/v1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

// EDIT THIS FILE!  THIS IS SCAFFOLDING FOR YOU TO OWN!
// NOTE: json tags are required.  Any new fields you add must have json tags for the fields to be serialized.

old stuff

// CronJobSpec defines the desired state of CronJob.
type CronJobSpec struct {
    // +kubebuilder:validation:MinLength=0

    // The schedule in Cron format, see https://en.wikipedia.org/wiki/Cron.
    Schedule string `json:"schedule"`

    // +kubebuilder:validation:Minimum=0

    // Optional deadline in seconds for starting the job if it misses scheduled
    // time for any reason.  Missed jobs executions will be counted as failed ones.
    // +optional
    StartingDeadlineSeconds *int64 `json:"startingDeadlineSeconds,omitempty"`

    // Specifies how to treat concurrent executions of a Job.
    // Valid values are:
    // - "Allow" (default): allows CronJobs to run concurrently;
    // - "Forbid": forbids concurrent runs, skipping next run if previous run hasn't finished yet;
    // - "Replace": cancels currently running job and replaces it with a new one
    // +optional
    ConcurrencyPolicy ConcurrencyPolicy `json:"concurrencyPolicy,omitempty"`

    // This flag tells the controller to suspend subsequent executions, it does
    // not apply to already started executions.  Defaults to false.
    // +optional
    Suspend *bool `json:"suspend,omitempty"`

    // Specifies the job that will be created when executing a CronJob.
    JobTemplate batchv1.JobTemplateSpec `json:"jobTemplate"`

    // +kubebuilder:validation:Minimum=0

    // The number of successful finished jobs to retain.
    // This is a pointer to distinguish between explicit zero and not specified.
    // +optional
    SuccessfulJobsHistoryLimit *int32 `json:"successfulJobsHistoryLimit,omitempty"`

    // +kubebuilder:validation:Minimum=0

    // The number of failed finished jobs to retain.
    // This is a pointer to distinguish between explicit zero and not specified.
    // +optional
    FailedJobsHistoryLimit *int32 `json:"failedJobsHistoryLimit,omitempty"`
}

// ConcurrencyPolicy describes how the job will be handled.
// Only one of the following concurrent policies may be specified.
// If none of the following policies is specified, the default one
// is AllowConcurrent.
// +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"
)

// CronJobStatus defines the observed state of CronJob.
type CronJobStatus struct {
    // 插入额外的状态字段 - 定义集群的观察状态
    // 重要：在修改此文件后，请运行 "make" 以重新生成代码

    // A list of pointers to currently running jobs.
    // +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"`
}

Since we’ll have more than one version, we’ll need to mark a storage version. This is the version that the Kubernetes API server uses to store our data. We’ll chose the v1 version for our project.

We’ll use the +kubebuilder:storageversion to do this.

Note that multiple versions may exist in storage if they were written before the storage version changes – changing the storage version only affects how objects are created/updated after the change.

// +kubebuilder:object:root=true
// +kubebuilder:storageversion
// +kubebuilder:conversion:hub
// +kubebuilder:subresource:status
// +versionName=v1
// +kubebuilder:storageversion
// 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"`
}

old stuff

// +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{})
}

Apache License

http://www.apache.org/licenses/LICENSE-2.0

package v1

Implementing the hub method is pretty easy – we just have to add an empty method called Hub()to serve as a marker. We could also just put this inline in our cronjob_types.go file.

// Hub marks this type as a conversion hub.
func (*CronJob) Hub() {}

Apache License

http://www.apache.org/licenses/LICENSE-2.0

package v2

Imports

import (
    "fmt"
    "strings"

    "log"

    "sigs.k8s.io/controller-runtime/pkg/conversion"

    batchv1 "tutorial.kubebuilder.io/project/api/v1"
)

Our “spoke” versions need to implement the Convertible interface. Namely, they’ll need ConvertTo() and ConvertFrom() methods to convert to/from the hub version.

ConvertTo is expected to modify its argument to contain the converted object. Most of the conversion is straightforward copying, except for converting our changed field.

// ConvertTo converts this CronJob (v2) to the Hub version (v1).
func (src *CronJob) ConvertTo(dstRaw conversion.Hub) error {
    dst := dstRaw.(*batchv1.CronJob)
    log.Printf("ConvertTo: Converting CronJob from Spoke version v2 to Hub version v1;"+
        "source: %s/%s, target: %s/%s", src.Namespace, src.Name, dst.Namespace, dst.Name)

    sched := src.Spec.Schedule
    scheduleParts := []string{"*", "*", "*", "*", "*"}
    if sched.Minute != nil {
        scheduleParts[0] = string(*sched.Minute)
    }
    if sched.Hour != nil {
        scheduleParts[1] = string(*sched.Hour)
    }
    if sched.DayOfMonth != nil {
        scheduleParts[2] = string(*sched.DayOfMonth)
    }
    if sched.Month != nil {
        scheduleParts[3] = string(*sched.Month)
    }
    if sched.DayOfWeek != nil {
        scheduleParts[4] = string(*sched.DayOfWeek)
    }
    dst.Spec.Schedule = strings.Join(scheduleParts, " ")

rote conversion

    // ObjectMeta
    dst.ObjectMeta = src.ObjectMeta

    // Spec
    dst.Spec.StartingDeadlineSeconds = src.Spec.StartingDeadlineSeconds
    dst.Spec.ConcurrencyPolicy = batchv1.ConcurrencyPolicy(src.Spec.ConcurrencyPolicy)
    dst.Spec.Suspend = src.Spec.Suspend
    dst.Spec.JobTemplate = src.Spec.JobTemplate
    dst.Spec.SuccessfulJobsHistoryLimit = src.Spec.SuccessfulJobsHistoryLimit
    dst.Spec.FailedJobsHistoryLimit = src.Spec.FailedJobsHistoryLimit

    // Status
    dst.Status.Active = src.Status.Active
    dst.Status.LastScheduleTime = src.Status.LastScheduleTime

    return nil
}

ConvertFrom is expected to modify its receiver to contain the converted object. Most of the conversion is straightforward copying, except for converting our changed field.

// ConvertFrom converts the Hub version (v1) to this CronJob (v2).
func (dst *CronJob) ConvertFrom(srcRaw conversion.Hub) error {
    src := srcRaw.(*batchv1.CronJob)
    log.Printf("ConvertFrom: Converting CronJob from Hub version v1 to Spoke version v2;"+
        "source: %s/%s, target: %s/%s", src.Namespace, src.Name, dst.Namespace, dst.Name)

    schedParts := strings.Split(src.Spec.Schedule, " ")
    if len(schedParts) != 5 {
        return fmt.Errorf("invalid schedule: not a standard 5-field schedule")
    }
    partIfNeeded := func(raw string) *CronField {
        if raw == "*" {
            return nil
        }
        part := CronField(raw)
        return &part
    }
    dst.Spec.Schedule.Minute = partIfNeeded(schedParts[0])
    dst.Spec.Schedule.Hour = partIfNeeded(schedParts[1])
    dst.Spec.Schedule.DayOfMonth = partIfNeeded(schedParts[2])
    dst.Spec.Schedule.Month = partIfNeeded(schedParts[3])
    dst.Spec.Schedule.DayOfWeek = partIfNeeded(schedParts[4])

rote conversion

    // ObjectMeta
    dst.ObjectMeta = src.ObjectMeta

    // Spec
    dst.Spec.StartingDeadlineSeconds = src.Spec.StartingDeadlineSeconds
    dst.Spec.ConcurrencyPolicy = ConcurrencyPolicy(src.Spec.ConcurrencyPolicy)
    dst.Spec.Suspend = src.Spec.Suspend
    dst.Spec.JobTemplate = src.Spec.JobTemplate
    dst.Spec.SuccessfulJobsHistoryLimit = src.Spec.SuccessfulJobsHistoryLimit
    dst.Spec.FailedJobsHistoryLimit = src.Spec.FailedJobsHistoryLimit

    // Status
    dst.Status.Active = src.Status.Active
    dst.Status.LastScheduleTime = src.Status.LastScheduleTime

    return nil
}

Apache License

http://www.apache.org/licenses/LICENSE-2.0

Go imports

package v1

import (
    "context"
    "fmt"

    "github.com/robfig/cron"
    apierrors "k8s.io/apimachinery/pkg/api/errors"
    "k8s.io/apimachinery/pkg/runtime/schema"
    validationutils "k8s.io/apimachinery/pkg/util/validation"
    "k8s.io/apimachinery/pkg/util/validation/field"

    "k8s.io/apimachinery/pkg/runtime"
    ctrl "sigs.k8s.io/controller-runtime"
    logf "sigs.k8s.io/controller-runtime/pkg/log"
    "sigs.k8s.io/controller-runtime/pkg/webhook"
    "sigs.k8s.io/controller-runtime/pkg/webhook/admission"

    batchv1 "tutorial.kubebuilder.io/project/api/v1"
)

Next, we’ll setup a logger for the webhooks.

var cronjoblog = logf.Log.WithName("cronjob-resource")

This setup doubles as setup for our conversion webhooks: as long as our types implement the Hub and Convertible interfaces, a conversion webhook will be registered.

// SetupCronJobWebhookWithManager registers the webhook for CronJob in the manager.
func SetupCronJobWebhookWithManager(mgr ctrl.Manager) error {
    return ctrl.NewWebhookManagedBy(mgr).For(&batchv1.CronJob{}).
        WithValidator(&CronJobCustomValidator{}).
        WithDefaulter(&CronJobCustomDefaulter{
            DefaultConcurrencyPolicy:          batchv1.AllowConcurrent,
            DefaultSuspend:                    false,
            DefaultSuccessfulJobsHistoryLimit: 3,
            DefaultFailedJobsHistoryLimit:     1,
        }).
        Complete()
}

Notice that we use kubebuilder markers to generate webhook manifests. This marker is responsible for generating a mutating webhook manifest.

The meaning of each marker can be found here.

This marker is responsible for generating a mutation webhook manifest.

// +kubebuilder:webhook:path=/mutate-batch-tutorial-kubebuilder-io-v1-cronjob,mutating=true,failurePolicy=fail,sideEffects=None,groups=batch.tutorial.kubebuilder.io,resources=cronjobs,verbs=create;update,versions=v1,name=mcronjob-v1.kb.io,admissionReviewVersions=v1

// CronJobCustomDefaulter struct is responsible for setting default values on the custom resource of the
// Kind CronJob when those are created or updated.
//
// NOTE: The +kubebuilder:object:generate=false marker prevents controller-gen from generating DeepCopy methods,
// as it is used only for temporary operations and does not need to be deeply copied.
type CronJobCustomDefaulter struct {

    // Default values for various CronJob fields
    DefaultConcurrencyPolicy          batchv1.ConcurrencyPolicy
    DefaultSuspend                    bool
    DefaultSuccessfulJobsHistoryLimit int32
    DefaultFailedJobsHistoryLimit     int32
}

var _ webhook.CustomDefaulter = &CronJobCustomDefaulter{}

We use the webhook.CustomDefaulterinterface to set defaults to our CRD. A webhook will automatically be served that calls this defaulting.

The Defaultmethod is expected to mutate the receiver, setting the defaults.

// Default implements webhook.CustomDefaulter so a webhook will be registered for the Kind CronJob.
func (d *CronJobCustomDefaulter) Default(ctx context.Context, obj runtime.Object) error {
    cronjob, ok := obj.(*batchv1.CronJob)

    if !ok {
        return fmt.Errorf("expected an CronJob object but got %T", obj)
    }
    cronjoblog.Info("Defaulting for CronJob", "name", cronjob.GetName())

    // Set default values
    d.applyDefaults(cronjob)
    return nil
}

// applyDefaults applies default values to CronJob fields.
func (d *CronJobCustomDefaulter) applyDefaults(cronJob *batchv1.CronJob) {
    if cronJob.Spec.ConcurrencyPolicy == "" {
        cronJob.Spec.ConcurrencyPolicy = d.DefaultConcurrencyPolicy
    }
    if cronJob.Spec.Suspend == nil {
        cronJob.Spec.Suspend = new(bool)
        *cronJob.Spec.Suspend = d.DefaultSuspend
    }
    if cronJob.Spec.SuccessfulJobsHistoryLimit == nil {
        cronJob.Spec.SuccessfulJobsHistoryLimit = new(int32)
        *cronJob.Spec.SuccessfulJobsHistoryLimit = d.DefaultSuccessfulJobsHistoryLimit
    }
    if cronJob.Spec.FailedJobsHistoryLimit == nil {
        cronJob.Spec.FailedJobsHistoryLimit = new(int32)
        *cronJob.Spec.FailedJobsHistoryLimit = d.DefaultFailedJobsHistoryLimit
    }
}

We can validate our CRD beyond what’s possible with declarative validation. Generally, declarative validation should be sufficient, but sometimes more advanced use cases call for complex validation.

For instance, we’ll see below that we use this to validate a well-formed cron schedule without making up a long regular expression.

If webhook.CustomValidator interface is implemented, a webhook will automatically be served that calls the validation.

The ValidateCreate, ValidateUpdate and ValidateDelete methods are expected to validate its receiver upon creation, update and deletion respectively. We separate out ValidateCreate from ValidateUpdate to allow behavior like making certain fields immutable, so that they can only be set on creation. ValidateDelete is also separated from ValidateUpdate to allow different validation behavior on deletion. Here, however, we just use the same shared validation for ValidateCreate and ValidateUpdate. And we do nothing in ValidateDelete, since we don’t need to validate anything on deletion.

This marker is responsible for generating a validation webhook manifest.

// +kubebuilder:webhook:path=/validate-batch-tutorial-kubebuilder-io-v1-cronjob,mutating=false,failurePolicy=fail,sideEffects=None,groups=batch.tutorial.kubebuilder.io,resources=cronjobs,verbs=create;update,versions=v1,name=vcronjob-v1.kb.io,admissionReviewVersions=v1

// CronJobCustomValidator struct is responsible for validating the CronJob resource
// when it is created, updated, or deleted.
//
// NOTE: The +kubebuilder:object:generate=false marker prevents controller-gen from generating DeepCopy methods,
// as this struct is used only for temporary operations and does not need to be deeply copied.
type CronJobCustomValidator struct {
    // TODO(user): Add more fields as needed for validation
}

var _ webhook.CustomValidator = &CronJobCustomValidator{}

// ValidateCreate implements webhook.CustomValidator so a webhook will be registered for the type CronJob.
func (v *CronJobCustomValidator) ValidateCreate(ctx context.Context, obj runtime.Object) (admission.Warnings, error) {
    cronjob, ok := obj.(*batchv1.CronJob)
    if !ok {
        return nil, fmt.Errorf("expected a CronJob object but got %T", obj)
    }
    cronjoblog.Info("Validation for CronJob upon creation", "name", cronjob.GetName())

    return nil, validateCronJob(cronjob)
}

// ValidateUpdate implements webhook.CustomValidator so a webhook will be registered for the type CronJob.
func (v *CronJobCustomValidator) ValidateUpdate(ctx context.Context, oldObj, newObj runtime.Object) (admission.Warnings, error) {
    cronjob, ok := newObj.(*batchv1.CronJob)
    if !ok {
        return nil, fmt.Errorf("expected a CronJob object for the newObj but got %T", newObj)
    }
    cronjoblog.Info("Validation for CronJob upon update", "name", cronjob.GetName())

    return nil, validateCronJob(cronjob)
}

// ValidateDelete implements webhook.CustomValidator so a webhook will be registered for the type CronJob.
func (v *CronJobCustomValidator) ValidateDelete(ctx context.Context, obj runtime.Object) (admission.Warnings, error) {
    cronjob, ok := obj.(*batchv1.CronJob)
    if !ok {
        return nil, fmt.Errorf("expected a CronJob object but got %T", obj)
    }
    cronjoblog.Info("Validation for CronJob upon deletion", "name", cronjob.GetName())

    // TODO(user): fill in your validation logic upon object deletion.

    return nil, nil
}

We validate the name and the spec of the CronJob.

// validateCronJob validates the fields of a CronJob object.
func validateCronJob(cronjob *batchv1.CronJob) error {
    var allErrs field.ErrorList
    if err := validateCronJobName(cronjob); err != nil {
        allErrs = append(allErrs, err)
    }
    if err := validateCronJobSpec(cronjob); err != nil {
        allErrs = append(allErrs, err)
    }
    if len(allErrs) == 0 {
        return nil
    }

    return apierrors.NewInvalid(
        schema.GroupKind{Group: "batch.tutorial.kubebuilder.io", Kind: "CronJob"},
        cronjob.Name, allErrs)
}

Some fields are declaratively validated by OpenAPI schema. You can find kubebuilder validation markers (prefixed with // +kubebuilder:validation) in the Designing an API section. You can find all of the kubebuilder supported markers for declaring validation by running controller-gen crd -w, or here.

func validateCronJobSpec(cronjob *batchv1.CronJob) *field.Error {
    // The field helpers from the kubernetes API machinery help us return nicely
    // structured validation errors.
    return validateScheduleFormat(
        cronjob.Spec.Schedule,
        field.NewPath("spec").Child("schedule"))
}

We’ll need to validate the cron schedule is well-formatted.

func validateScheduleFormat(schedule string, fldPath *field.Path) *field.Error {
    if _, err := cron.ParseStandard(schedule); err != nil {
        return field.Invalid(fldPath, schedule, err.Error())
    }
    return nil
}

Validate object name

func validateCronJobName(cronjob *batchv1.CronJob) *field.Error {
    if len(cronjob.ObjectMeta.Name) > validationutils.DNS1035LabelMaxLength-11 {
        // The job name length is 63 characters like all Kubernetes objects
        // (which must fit in a DNS subdomain). The cronjob controller appends
        // a 11-character suffix to the cronjob (`-$TIMESTAMP`) when creating
        // a job. The job name length limit is 63 characters. Therefore cronjob
        // names must have length <= 63-11=52. If we don't validate this here,
        // then job creation will fail later.
        return field.Invalid(field.NewPath("metadata").Child("name"), cronjob.ObjectMeta.Name, "must be no more than 52 characters")
    }
    return nil
}

Apache License

http://www.apache.org/licenses/LICENSE-2.0

Imports

package main

import (
    "crypto/tls"
    "flag"
    "os"
    "path/filepath"

    // 导入所有 Kubernetes 客户端身份验证插件（例如 Azure、GCP、OIDC 等）
    // 确保 exec-entrypoint 和 run 可以使用它们。
    _ "k8s.io/client-go/plugin/pkg/client/auth"

    kbatchv1 "k8s.io/api/batch/v1"
    "k8s.io/apimachinery/pkg/runtime"
    utilruntime "k8s.io/apimachinery/pkg/util/runtime"
    clientgoscheme "k8s.io/client-go/kubernetes/scheme"
    ctrl "sigs.k8s.io/controller-runtime"
    "sigs.k8s.io/controller-runtime/pkg/certwatcher"
    "sigs.k8s.io/controller-runtime/pkg/healthz"
    "sigs.k8s.io/controller-runtime/pkg/log/zap"
    "sigs.k8s.io/controller-runtime/pkg/metrics/filters"
    metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
    "sigs.k8s.io/controller-runtime/pkg/webhook"

    batchv1 "tutorial.kubebuilder.io/project/api/v1"
    batchv2 "tutorial.kubebuilder.io/project/api/v2"
    "tutorial.kubebuilder.io/project/internal/controller"
    webhookbatchv1 "tutorial.kubebuilder.io/project/internal/webhook/v1"
    webhookbatchv2 "tutorial.kubebuilder.io/project/internal/webhook/v2"
    // +kubebuilder:scaffold:imports
)

existing setup

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

func init() {
    utilruntime.Must(clientgoscheme.AddToScheme(scheme))

    utilruntime.Must(kbatchv1.AddToScheme(scheme)) // we've added this ourselves
    utilruntime.Must(batchv1.AddToScheme(scheme))
    utilruntime.Must(batchv2.AddToScheme(scheme))
    // +kubebuilder:scaffold:scheme}

// nolint:gocyclo
func main() {

existing setup

    var metricsAddr string
    var metricsCertPath, metricsCertName, metricsCertKey string
    var webhookCertPath, webhookCertName, webhookCertKey string
    var enableLeaderElection bool
    var probeAddr string
    var secureMetrics bool
    var enableHTTP2 bool
    var tlsOpts []func(*tls.Config)
    flag.StringVar(&metricsAddr, "metrics-bind-address", "0", "The address the metrics endpoint binds to. "+
        "使用 :8443 进行 HTTPS，使用 :8080 进行 HTTP，或者保持为 0 以禁用指标服务。")
    flag.StringVar(&probeAddr, "health-probe-bind-address", ":8081", "The address the probe endpoint binds to.")
    flag.BoolVar(&enableLeaderElection, "leader-elect", false,
        "Enable leader election for controller manager. "+
            "启用此功能将确保只有一个活动的控制管理器。")
    flag.BoolVar(&secureMetrics, "metrics-secure", true,
        "如果设置，指标端点将通过 HTTPS 安全提供。请使用 --metrics-secure=false 以改为使用 HTTP。")
    flag.StringVar(&webhookCertPath, "webhook-cert-path", "", "The directory that contains the webhook certificate.")
    flag.StringVar(&webhookCertName, "webhook-cert-name", "tls.crt", "The name of the webhook certificate file.")
    flag.StringVar(&webhookCertKey, "webhook-cert-key", "tls.key", "The name of the webhook key file.")
    flag.StringVar(&metricsCertPath, "metrics-cert-path", "",
        "The directory that contains the metrics server certificate.")
    flag.StringVar(&metricsCertName, "metrics-cert-name", "tls.crt", "The name of the metrics server certificate file.")
    flag.StringVar(&metricsCertKey, "metrics-cert-key", "tls.key", "The name of the metrics server key file.")
    flag.BoolVar(&enableHTTP2, "enable-http2", false,
        "If set, HTTP/2 will be enabled for the metrics and webhook servers")
    opts := zap.Options{
        Development: true,
    }
    opts.BindFlags(flag.CommandLine)
    flag.Parse()

    ctrl.SetLogger(zap.New(zap.UseFlagOptions(&opts)))

    // 如果 enable-http2 标志为 false（默认值），则应禁用 http/2
    // 由于其存在的漏洞。更具体地说，禁用 http/2 将
    // 防止受到 HTTP/2 流取消和
    // 快速重置 CVE 的漏洞影响。有关更多信息，请参见：
    // - https://github.com/advisories/GHSA-qppj-fm5r-hxr3
    // - https://github.com/advisories/GHSA-4374-p667-p6c8
    disableHTTP2 := func(c *tls.Config) {
        setupLog.Info("disabling http/2")
        c.NextProtos = []string{"http/1.1"}
    }

    if !enableHTTP2 {
        tlsOpts = append(tlsOpts, disableHTTP2)
    }

    // 创建用于指标和 Webhook 证书的监视器
    var metricsCertWatcher, webhookCertWatcher *certwatcher.CertWatcher

    // 初始Webhook TLS选项
    webhookTLSOpts := tlsOpts

    if len(webhookCertPath) > 0 {
        setupLog.Info("使用提供的证书初始化 Webhook 证书监视器",
            "webhook-cert-path", webhookCertPath, "webhook-cert-name", webhookCertName, "webhook-cert-key", webhookCertKey)

        var err error
        webhookCertWatcher, err = certwatcher.New(
            filepath.Join(webhookCertPath, webhookCertName),
            filepath.Join(webhookCertPath, webhookCertKey),
        )
        if err != nil {
            setupLog.Error(err, "Failed to initialize webhook certificate watcher")
            os.Exit(1)
        }

        webhookTLSOpts = append(webhookTLSOpts, func(config *tls.Config) {
            config.GetCertificate = webhookCertWatcher.GetCertificate
        })
    }

    webhookServer := webhook.NewServer(webhook.Options{
        TLSOpts: webhookTLSOpts,
    })

    // 指标端点在 'config/default/kustomization.yaml' 中启用。指标选项配置服务器。
    // 更多信息：
    // - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.20.4/pkg/metrics/server
    // - https://book.kubebuilder.io/reference/metrics.html
    metricsServerOptions := metricsserver.Options{
        BindAddress:   metricsAddr,
        SecureServing: secureMetrics,
        TLSOpts:       tlsOpts,
    }

    if secureMetrics {
        // FilterProvider 用于通过认证和授权来保护指标端点。
        // 这些配置确保只有授权用户和服务账户
        // 可以访问指标端点。RBAC 配置在 'config/rbac/kustomization.yaml' 中。更多信息：
        // https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.20.4/pkg/metrics/filters#WithAuthenticationAndAuthorization
        metricsServerOptions.FilterProvider = filters.WithAuthenticationAndAuthorization
    }

    // 如果未指定证书，controller-runtime 将自动
    // 为指标服务器生成自签名证书。虽然对于开发和测试来说很方便，
    // 但这种设置不建议用于生产环境。
    //
    // TODO(用户)：如果您启用 certManager，请取消注释以下行：
    // - [METRICS-WITH-CERTS] 在 config/default/kustomization.yaml 中生成并使用
    // 由 cert-manager 管理的指标服务器证书。
    // - [PROMETHEUS-WITH-CERTS] 在 config/prometheus/kustomization.yaml 中用于 TLS 认证。
    if len(metricsCertPath) > 0 {
        setupLog.Info("使用提供的证书初始化指标证书监视器",
            "metrics-cert-path", metricsCertPath, "metrics-cert-name", metricsCertName, "metrics-cert-key", metricsCertKey)

        var err error
        metricsCertWatcher, err = certwatcher.New(
            filepath.Join(metricsCertPath, metricsCertName),
            filepath.Join(metricsCertPath, metricsCertKey),
        )
        if err != nil {
            setupLog.Error(err, "to initialize metrics certificate watcher", "error", err)
            os.Exit(1)
        }

        metricsServerOptions.TLSOpts = append(metricsServerOptions.TLSOpts, func(config *tls.Config) {
            config.GetCertificate = metricsCertWatcher.GetCertificate
        })
    }

    mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
        Scheme:                 scheme,
        Metrics:                metricsServerOptions,
        WebhookServer:          webhookServer,
        HealthProbeBindAddress: probeAddr,
        LeaderElection:         enableLeaderElection,
        LeaderElectionID:       "80807133.tutorial.kubebuilder.io",
        // LeaderElectionReleaseOnCancel 定义了当管理器结束时，领导者是否应自愿辞职
        // 当管理器停止时，这要求二进制文件立即结束，否则这个设置是不安全的。设置这个可以显著
        // 加快自愿领导者的过渡，因为新的领导者不必首先等待
        // LeaseDuration 时间。
        //
        // 在提供的默认脚手架中，程序会在
        // 管理器停止后立即结束，因此启用此选项是可以的。不过，
        // 如果您正在进行或打算在管理器停止后执行任何操作，如执行清理，
        // 那么使用它可能是不安全的。
        // LeaderElectionReleaseOnCancel: true,
    })
    if err != nil {
        setupLog.Error(err, "unable to start manager")
        os.Exit(1)
    }

    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)
    }

Our existing call to SetupWebhookWithManager registers our conversion webhooks with the manager, too.

    // nolint:goconst
    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)
        }
    }
    // nolint:goconst
    if os.Getenv("ENABLE_WEBHOOKS") != "false" {
        if err = webhookbatchv2.SetupCronJobWebhookWithManager(mgr); err != nil {
            setupLog.Error(err, "unable to create webhook", "webhook", "CronJob")
            os.Exit(1)
        }
    }
    // +kubebuilder:scaffold:builder

existing setup

    if metricsCertWatcher != nil {
        setupLog.Info("Adding metrics certificate watcher to manager")
        if err := mgr.Add(metricsCertWatcher); err != nil {
            setupLog.Error(err, "unable to add metrics certificate watcher to manager")
            os.Exit(1)
        }
    }

    if webhookCertWatcher != nil {
        setupLog.Info("Adding webhook certificate watcher to manager")
        if err := mgr.Add(webhookCertWatcher); err != nil {
            setupLog.Error(err, "unable to add webhook certificate watcher to manager")
            os.Exit(1)
        }
    }

    if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
        setupLog.Error(err, "unable to set up health check")
        os.Exit(1)
    }
    if err := mgr.AddReadyzCheck("readyz", healthz.Ping); err != nil {
        setupLog.Error(err, "unable to set up ready check")
        os.Exit(1)
    }

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

}