Kubernetes Operator与自定义控制器深度解析

一、Operator模式与声明式API

1.1 什么是Operator

Operator这个词最早来自混沌工程领域的" humano-operator"，意为"人工操作员"。在Kubernetes语境下，Operator是将人类运维经验和领域知识编码为软件的一种设计模式。它本质上是一个运行在Kubernetes集群内的自定义控制器，通过监听自定义资源（CR）的变化，自动执行复杂的运维操作——这远远超越了Kubernetes原生Controller（如Deployment、StatefulSet）的能力边界。

举一个直观的例子：etcd-operator不需要运维人员手动执行备份、恢复、滚动升级等操作，它会自动检测集群健康状态，在故障时自动恢复，在版本发布时自动执行零宕机升级。运维人员的角色从"执行者"转变为"监督者"，这就是Operator的核心价值。

1.2 Kubernetes CRD机制

Custom Resource Definition（自定义资源定义）是Kubernetes扩展API能力的核心机制。在Kubernetes 1.16之前，CRD是alpha版本；到了1.16+ CRD全面稳定，广泛应用于生产环境。CRD允许开发者在不修改Kubernetes核心代码的情况下，定义全新的资源类型，就像使用原生Deployment、Service一样使用这些自定义资源。

CRD本身只是一份声明式Schema定义，它告诉Kubernetes API Server：这个资源有哪些字段、字段类型是什么、有什么约束条件。真正处理这些资源业务逻辑的，是配合CRD运行的Custom Controller（即Operator）。

# 一个典型的MySQLBackup CRD定义（YAML）
apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
metadata:
  name: mysqlbackups.mysql.example.com
  labels:
    app.kubernetes.io/name: mysql-operator
spec:
  group: mysql.example.com
  names:
    kind: MySQLBackup
    listKind: MySQLBackupList
    plural: mysqlbackups
    singular: mysqlbackup
  scope: Namespaced
  versions:
    - name: v1alpha1
      served: true
      storage: true
      schema:
        openAPIV3Schema:
          type: object
          required: [spec]
          properties:
            spec:
              type: object
              required: [mysqlClusterRef]
              properties:
                mysqlClusterRef:
                  type: object
                  properties:
                    name:
                      type: string
                    namespace:
                      type: string
                backupType:
                  type: string
                  enum: [full, incremental]
                  default: full
                retentionDays:
                  type: integer
                  minimum: 1
                  maximum: 365
                  default: 7
            status:
              type: object
              properties:
                phase:
                  type: string
                  enum: [Pending, Running, Completed, Failed]
                startTime:
                  type: string
                  format: date-time
                completionTime:
                  type: string
                  format: date-time
                backupPath:
                  type: string

1.3 声明式 vs 命令式：为什么Operator更优越

传统的命令式运维是"做这个，再做那个"——一系列线性步骤，每一步都需要人工干预。这种方式的问题在于：无法应对突发故障、无法保证幂等性、无法在复杂依赖场景下保证一致性。

声明式API的核心哲学是"描述期望状态，让系统自己达到那个状态"。Kubernetes本身就是一个声明式系统的典范：用户声明"我需要3个副本"，Kubernetes Controller会持续协调，直到3个副本都在运行。

Operator继承了这种哲学，并将它扩展到了应用层。用户的YAML声明的是"我需要一个每秒处理5000笔交易的消息队列集群，带自动故障恢复和跨AZ高可用"——Operator会理解这个意图，持续监控实际状态，驱动系统向期望状态收敛。

对比Java Spring生态中类似的设计模式：Spring Cloud中的Config Server、Service Discovery（Eureka/Consul）这些组件的配置管理，本质上也是一种声明式配置——开发者声明配置中心地址，服务注册到Eureka。但Spring Cloud的这些Controller是静态配置驱动的，缺乏真正的闭环反馈机制。而Operator则具备完整的Observe-Analyze-Act闭环，每次状态变化都会触发Reconcile循环，直到实际状态等于期望状态。

1.4 Admission Webhook与CRD验证

Kubernetes提供了两种Admission Webhook来增强CRD的验证和修改能力：

MutatingWebhook：在资源被写入数据库之前修改它（比如为字段设置默认值）；

ValidatingWebhook：验证资源的合法性，不合法则拒绝写入。

在Operator开发中，Kubebuilder和OperatorSDK都支持通过Marker注释自动生成Webhook代码。例如可以为MySQLBackup资源添加默认值（backupType默认为full），并验证backupPath不能为空。

// Kubebuilder风格的Webhook定义
// +kubebuilder:webhook:path=/mutate-mysql-example-com-v1-mysqlbackup,
//   mutating=true, failurePolicy=fail, groups=mysql.example.com,
//   resources=mysqlbackups, verbs=create;update, name=mmysqlbackup.kb.io

func (r *MySQLBackup) Default() {
    if r.Spec.BackupType == "" {
        r.Spec.BackupType = "full"
    }
    if r.Spec.RetentionDays == 0 {
        r.Spec.RetentionDays = 7
    }
    // 自动注入操作者的身份信息
    if r.Annotations == nil {
        r.Annotations = make(map[string]string)
    }
    r.Annotations["operator.example.com/created-by"] = "mysql-operator"
}

// +kubebuilder:webhook:path=/validate-mysql-example-com-v1-mysqlbackup,
//   mutating=false, failurePolicy=fail, groups=mysql.example.com,
//   resources=mysqlbackups, verbs=create;update, name=vmysqlbackup.kb.io

func (r *MySQLBackup) ValidateCreate() error {
    if r.Spec.MySQLClusterRef.Name == "" {
        return field.Invalid(
            field.NewPath("spec").Child("mysqlClusterRef").Child("name"),
            r.Spec.MySQLClusterRef.Name,
            "mysql cluster reference name is required")
    }
    return nil
}

二、控制理论视角：Reconcile循环

2.1 控制器模式：Observe → Analyze → Act

如果用控制理论来解读Kubernetes Controller，你会惊讶地发现它与经典控制系统的结构高度一致。所有的Kubernetes内置Controller——Deployment、StatefulSet、Job、CronJob——以及所有的Custom Controller，都遵循同一个控制范式：

Observe（观察）：通过Informer监听集群状态，获取所有相关资源的当前快照。Deployment Controller会监听Deployment自身、ReplicaSet和Pod的变化。

Analyze（分析）：对比Spec（期望状态）和Status（实际状态），找出偏差。例如Deployment期望3个副本，但Status显示只有2个运行中。

Act（行动）：根据偏差计算并执行纠正操作。Deployment Controller会创建新的Pod，使副本数恢复到3。

这个循环会持续运行，直到系统进入稳定状态——即期望状态等于实际状态。

2.2 Reconcile函数的幂等性与最终一致性

Reconcile函数是Operator的核心。Kubebuilder定义的Reconcile签名如下：

// Reconcile实现签名（Kubebuilder框架）
func (r *MySQLBackupReconciler) Reconcile(ctx context.Context,
    req ctrl.Request) (ctrl.Result, error) {
    log := log.FromContext(ctx)

    // 1. 获取CR实例
    backup := &mysqlv1alpha1.MySQLBackup{}
    if err := r.Get(ctx, req.NamespacedName, backup); err != nil {
        // 如果资源被删除了，直接返回（垃圾回收会处理关联资源）
        return ctrl.Result{}, client.IgnoreNotFound(err)
    }

    // 2. 提取期望状态（Spec）
    clusterRef := backup.Spec.MySQLClusterRef

    // 3. 观察实际状态：从集群获取关联Pod、Service等
    podList := &corev1.PodList{}
    if err := r.List(ctx, podList,
        client.InNamespace(clusterRef.Namespace),
        client.MatchingLabels{"app": "mysql", "cluster": clusterRef.Name}); err != nil {
        return ctrl.Result{}, err
    }

    // 4. 分析偏差：判断是否需要执行备份
    if backup.Status.Phase == "Running" {
        log.Info("Backup already in progress, skipping")
        return ctrl.Result{RequeueAfter: 30 * time.Second}, nil
    }

    // 5. Act：创建备份Job
    job := r.buildBackupJob(backup)
    if err := r.Create(ctx, job); err != nil {
        return ctrl.Result{}, err
    }

    // 6. 更新Status
    backup.Status.Phase = "Running"
    backup.Status.StartTime = &metav1.Time{Time: time.Now()}
    return ctrl.Result{}, r.Status().Update(ctx, backup)
}

// buildBackupJob是一个幂等的构建方法：
// 无论调用多少次，只要输入相同，生成的JobSpec就是一致的
func (r *MySQLBackupReconciler) buildBackupJob(
    backup *mysqlv1alpha1.MySQLBackup) *batchv1.Job {
    labels := map[string]string{
        "app":             "mysql-backup",
        "backup-owner":    backup.Name,
        "managed-by":      "mysql-operator",
    }
    return &batchv1.Job{
        ObjectMeta: metav1.ObjectMeta{
            Name:      "backup-" + backup.Name,
            Namespace: backup.Namespace,
            Labels:    labels,
            OwnerReferences: []metav1.OwnerReference{
                *metav1.NewControllerRef(backup,
                    mysqlv1alpha1.GroupVersion.WithKind("MySQLBackup")),
            },
        },
        Spec: batchv1.JobSpec{
            ttlSecondsAfterFinished: 3600,
            Template: corev1.PodTemplateSpec{
                Spec: corev1.PodSpec{
                    RestartPolicy: corev1.RestartPolicyOnFailure,
                    Containers: []corev1.Container{{
                        Name:  "backup",
                        Image: "mysql-backup-tool:v2.1.0",
                        Env: []corev1.EnvVar{{
                            Name:  "BACKUP_PATH",
                            Value: backup.Spec.BackupPath,
                        }},
                    }},
                },
            },
        },
    }
}

2.3 预期状态 vs 实际状态：Spec与Status分离设计

这是Kubernetes API设计中最重要的原则之一。Spec是用户声明的期望状态，由用户（或更上层的控制器）写入；Status是系统报告的实际状态，由Controller更新。Spec是输入，Status是输出，两者严格分离。

以Deployment为例：用户写的是" replicas: 3 "（Spec），系统报告的是" availableReplicas: 3 "（Status）。用户永远不应该直接修改Status——这是系统的"自我报告"，由Controller维护。

这种设计的精妙之处在于：Spec可以被GitOps工具（ArgoCD、Flux）管理，而Status则实时反映集群的真实状态。两者可以独立版本化、独立审计。

2.4 OwnerReference与垃圾回收

OwnerReference是Kubernetes实现级联删除（Cascading Deletion）的关键机制。当一个CR被删除时，Kubernetes GC（Garbage Collector）会自动删除所有以其为Owner的资源，无需Operator手动清理。

// 设置OwnerReference的关键代码
ownerRef := metav1.OwnerReference{
    APIVersion: mysqlv1alpha1.GroupVersion.String(),
    Kind:       "MySQLBackup",
    Name:       backup.Name,
    UID:        backup.UID,    // 必须是真实UID，不能是空值
    Controller: ptr.To(true), // true表示这是一个"控制者"引用
}

// 在创建Job时加入OwnerReference
job := &batchv1.Job{
    ObjectMeta: metav1.ObjectMeta{
        OwnerReferences: []metav1.OwnerReference{ownerRef},
    },
}

// 效果：删除MySQLBackup时，对应的Job会被Kubernetes GC自动删除
// kubectl delete mysqlbackup backup-001
// -> Kubernetes自动删除 backup-001 对应的 Job

2.5 Java工程师视角：Spring Controller vs K8s Reconcile

如果你是Java工程师，可能会觉得K8s的Reconcile循环和Spring MVC的Controller有些相似，但两者有着本质区别：

Spring MVC的Controller是请求-响应模型：一次HTTP请求，一次处理，请求结束就结束。Spring Cloud Sleuth/Config中的Config Refresh虽然也是事件驱动的，但通常是被动的（由POST /refresh端点触发）。

K8s的Reconcile是持续循环模型：Operator启动后，Reconcile会被反复调用（通过Informer的事件触发），直到期望状态等于实际状态。如果中间出错，Reconcile会返回错误并被重新放入WorkQueue，稍后重试。这是一个天然的无终止控制循环。

// ============ Java风格的Reconcile伪代码 ============
// 这不是真实代码，而是用Java思维来理解K8s Reconcile模式
public class MySQLBackupReconciler {

    public ReconcileResult reconcile(MySQLBackup backup) {
        // Observe: 获取关联资源状态
        List<Pod> mysqlPods = podInformer.getPods(
            backup.getNamespace(),
            Map.of("app", "mysql")
        );

        // Analyze: 分析偏差
        if ("Running".equals(backup.getStatus().getPhase())) {
            // 备份已经在进行中
            return ReconcileResult.requeue(Duration.ofSeconds(30));
        }

        // Act: 创建备份Job
        Job backupJob = jobBuilder
            .name("backup-" + backup.getName())
            .ownerReference(backup)
            .image("mysql-backup:v2")
            .build();

        try {
            kubernetesClient.create(backupJob);
            backup.getStatus().setPhase("Running");
            kubernetesClient.updateStatus(backup);
        } catch (KubernetesException e) {
            // 幂等性：Job已存在（因为标签相同），直接忽略
            if (e.getError().getCode() == 409) {
                log.info("Backup job already exists, skipping creation");
            } else {
                return ReconcileResult.error(e);
            }
        }

        return ReconcileResult.done(); // Reconcile完成，等待下次事件触发
    }
}

// ============ Go真实Kubebuilder实现对比 ============
// 注意Go版本的关键特性：
// 1. 通过channel + WorkQueue处理并发，无需锁
// 2. Indexer提供本地缓存，零网络开销读取
// 3. Reconcile是纯函数式的，依赖通过结构体注入（依赖注入模式）

三、Kubernetes API客户端与代码生成

3.1 client-go Informer机制：旁路缓存与Indexer

client-go是Kubernetes官方Go语言客户端库，Informer是其中最核心的组件之一。Informer的设计目标是在保证数据一致性的前提下，最大限度减少对API Server的请求压力。

传统做法是每个Controller直接轮询API Server获取资源列表——这在有100个Pod的集群里，每秒可能产生成百上千次不必要的HTTP请求。Informer通过旁路缓存（Side Cache）彻底改变了这个模式：

List-Watch机制：Informer首先通过一次完整的List获取全量数据，然后通过Watch持续接收增量变更事件。所有这些事件都被存入本地Cache（LWW Cache），Controller从此不再直接请求API Server来读取数据。

Indexer：一个线程安全的内存存储，支持按多个维度索引对象。Controller可以通过Indexer用标签选择器、字段选择器快速查询，无需任何网络请求。

// Informer的创建与使用示例
package main

import (
    "fmt"
    "time"

    corev1 "k8s.io/api/core/v1"
    "k8s.io/apimachinery/pkg/labels"
    "k8s.io/client-go/informers"
    "k8s.io/client-go/tools/cache"
)

func main() {
    // 创建SharedInformersFactory，sharedInformer=true意味着
    // 同一个资源类型的多个Controller共享同一个Informer实例
    factory := informers.NewSharedInformerFactory(
        clientset,           // kubernetes.Clientset
        30*time.Second,      // resyncPeriod：定期全量同步周期
    )

    // 获取Pod的Informer
    podInformer := factory.Core().V1().Pods().Informer()

    // 添加事件处理函数
    podInformer.AddEventHandler(cache.FilteringResourceEventHandler{
        // FilterFunc：在事件入队前过滤，只关心特定标签的Pod
        FilterFunc: func(obj interface{}) bool {
            pod := obj.(*corev1.Pod)
            selector := labels.Set{"app": "mysql"}.AsSelectorPreTransformed()
            return selector.Matches(labels.Set(pod.Labels))
        },
        Handler: cache.ResourceEventHandlerFuncs{
            OnAdd: func(obj interface{}) {
                pod := obj.(*corev1.Pod)
                fmt.Printf("Pod Added: %s/%s\n", pod.Namespace, pod.Name)
            },
            OnUpdate: func(oldObj, newObj interface{}) {
                newPod := newObj.(*corev1.Pod)
                fmt.Printf("Pod Updated: %s/%s\n", newPod.Namespace, newPod.Name)
            },
            OnDelete: func(obj interface{}) {
                pod := obj.(*corev1.Pod)
                fmt.Printf("Pod Deleted: %s/%s\n", pod.Namespace, pod.Name)
            },
        },
    })

    // 启动Informer（后台goroutine运行）
    factory.Start(ctx.Done())
    // 等待所有Informer完成初始同步
    factory.WaitForCacheSync(ctx.Done())

    // 通过Indexer读取本地缓存（零网络请求）
    indexer := podInformer.GetIndexer()
    pods, _ := indexer.ByIndex(cache.NamespaceIndex, "production")
    for _, pod := range pods {
        p := pod.(*corev1.Pod)
        fmt.Printf("  Cached Pod: %s (phase=%s)\n", p.Name, p.Status.Phase)
    }

    // <-ctx.Done()
}

实际生产环境中，一个Operator通常只需要一个SharedInformerFactory实例。Kubebuilder/OperatorSDK已经封装好了这个细节——开发者只需要在Reconcile中调用client.Get()或client.List()，框架会自动通过缓存读取，性能提升显著。测试数据表明，合理使用Informer可以将API Server的请求量减少70%-90%。

3.2 List-Watch的增量同步原理

List-Watch是Kubernetes实现最终一致性的核心协议。分为两个阶段：

List阶段：GET /api/v1/pods?watch=false，返回资源的完整快照。这是Informer初始化时必须执行的"冷启动"步骤。

Watch阶段：GET /api/v1/pods?watch=true，通过HTTP Chunked Transfer Encoding建立长连接，API Server会持续推送资源变更事件（ADDED/MODIFIED/DELETED），直到连接断开或超时。

当Watch连接因网络抖动断开时，Informer会执行"re-list"：再次调用List获取全量快照，然后重新建立Watch。这是最终的兜底保障，确保即使丢失了部分事件，也不会造成数据永久不一致。

3.3 WorkQueue与指数退避重试

WorkQueue（任务队列）是Reconcile循环的"节流阀"。当Informer检测到资源变化时，它将变更放入WorkQueue。Reconcile从队列中取出任务进行处理。如果Reconcile出错，会将任务重新入队，并等待一定时间后再处理。

Kubebuilder默认使用的是rate-limiting queue（限速队列），支持指数退避策略：首次失败后等待5秒重试，如果再次失败则等待10秒、20秒、40秒……最大等待时间通常设为5分钟。这种设计防止了"惊群效应"——当大量资源同时变化时（如一次滚动更新涉及的几十个Pod），不会导致Controller被淹没。

// Kubebuilder Manager启动时配置的限速WorkQueue
import (
    "k8s.io/client-go/util/workqueue"
    "sigs.k8s.io/controller-runtime/pkg/controller"
    "sigs.k8s.io/controller-runtime/pkg/handler"
    "sigs.k8s.io/controller-runtime/pkg/source"
)

func NewMySQLBackupController(mgr manager.Manager) (controller.Controller, error) {
    return controller.New("mysqlbackup-controller", mgr,
        controller.Options{
            ReconcilePeriod: 10 * time.Second,  // 即使无事件，也每10秒Reconcile一次
            MaxConcurrentReconciles: 4,         // 最多4个并发Reconcile协程
            CacheSyncTimeout: 5 * time.Minute, // 等待缓存同步的超时时间
        })
}

// 限速队列的行为可以通过以下方式配置：
type RateLimiterConfig struct {
    BaseDelay  time.Duration // 基础延迟，默认5秒
    MaxDelay   time.Duration // 最大延迟，默认5分钟
    Factor     float64        // 退避因子，默认1.6
}

// 使用 client-go 的 BucketRateLimiter
limiter := workqueue.NewBucketRateLimiter(
    workqueue.DefaultItemBasedRateLimiter(), // 基于项的限速
)
// 或者使用 MaxOfRateLimiter 组合多个限速器

3.4 kubebuilder/code-generator代码生成工具链

Kubebuilder是当前最流行的Operator开发框架（由Apache 2.0许可，源自CoreOS的operator-framework）。它通过一套精巧的代码生成工具链，将开发者从大量样板代码中解放出来。

核心工具链包括：

controller-gen：扫描.go文件中的Marker注释（//+kubebuilder:...），自动生成CRD YAML、RBAC配置、DeepCopy实现。

webhook-gen：生成Webhook服务器的脚手架代码。

kustomize：管理多环境（dev/staging/prod）的K8s资源配置。

# Kubebuilder项目初始化与代码生成流程

# 1. 初始化项目
kubebuilder init --domain example.com --repo github.com/example/mysql-operator
# 生成: go.mod, main.go, PROJECT文件, Dockerfile, Makefile

# 2. 创建API（CRD + Controller）
kubebuilder create api --group mysql --version v1alpha1 --kind MySQLBackup
# 生成:
#   api/v1/mysqlbackup_types.go   # CRD类型定义 + marker
#   api/v1/zz_generated.deepcopy.go  # 自动生成，勿手动编辑
#   controllers/mysqlbackup_controller.go  # Reconcile实现骨架
#   config/crd/bases/mysql.example.com_mysqlbackups.yaml  # CRD YAML

# 3. 运行代码生成器（make generate）
# generate: Run code generation for API types
.PHONY: generate
generate: controller-gen
    $(CONTROLLER_GEN) object:headerFile="hack/boilerplate.go.txt" paths="./..."

# 4. 生成CRD YAML（make manifests）
# manifests: Generate manifests e.g. CRD, RBAC
.PHONY: manifests
manifests: controller-gen
    $(CONTROLLER_GEN) crd:crdVersions=v1 paths="./..." output:crd:artifacts:config=config/crd/bases

# 5. 安装CRD到集群
make install

# 6. 启动Operator
make run

# 7. 构建镜像并部署
make docker-build IMG=myrepo/mysql-operator:v1.0.0
make docker-push IMG=myrepo/mysql-operator:v1.0.0
make deploy

3.5 CRD YAML完整示例

# config/crd/bases/mysql.example.com_mysqlbackups.yaml
# 由 controller-gen 自动生成，包含完整的OpenAPI v3 Schema
apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
metadata:
  annotations:
    controller-gen.kubebuilder.io/version: v0.14.0
  creationTimestamp: null
  name: mysqlbackups.mysql.example.com
spec:
  group: mysql.example.com
  names:
    kind: MySQLBackup
    listKind: MySQLBackupList
    plural: mysqlbackups
    singular: mysqlbackup
    shortNames: [mb]
  scope: Namespaced
  versions:
    - name: v1alpha1
      served: true
      storage: true
      subresources:
        status: {}  # 启用/status子资源，Spec和Status分开存储
        scale:
          specReplicasPath: .spec.replicas
          statusReplicasPath: .status.replicas
          labelSelectorPath: .status.labelSelector
      additionalPrinterColumns:
        - name: Cluster
          type: string
          jsonPath: .spec.mysqlClusterRef.name
        - name: Type
          type: string
          jsonPath: .spec.backupType
        - name: Phase
          type: string
          jsonPath: .status.phase
        - name: Age
          type: date
          jsonPath: .metadata.creationTimestamp

四、实战：用Kubebuilder构建生产级Operator

4.1 项目初始化与目录结构

一个标准的Kubebuilder项目结构如下：

mysql-operator/
├── Dockerfile
├── Makefile
├── PROJECT               # 项目元信息（kubebuilder标记）
├── api/
│   └── v1/
│       ├── groupversion_info.go   # GroupVersion定义（自动生成）
│       ├── mysqlbackup_types.go   # CRD类型定义（开发者编辑）
│       └── zz_generated.deepcopy.go  # DeepCopy实现（自动生成，勿改）
├── bin/                 # 本地工具链（kustomize, controller-gen）
├── config/
│   ├── crd/            # CRD YAML
│   │   ├── bases/
│   │   └── patches/
│   ├── default/        # Kustomize default配置
│   ├── manager/       # Deployment/ServiceAccount
│   ├── rbac/          # RBAC YAML（Role, RoleBinding）
│   └── webhook/       # Webhook配置
├── controllers/
│   └── mysqlbackup_controller.go  # 核心Reconcile逻辑
├── hack/
│   └── boilerplate.go.txt  # 文件头版权信息模板
├── main.go              # 程序入口
└── go.mod

4.2 定义CRD：类型注解详解

CRD类型定义文件是Operator开发的核心。以下是MySQLBackup的完整类型定义，包含所有关键Marker：

// api/v1/mysqlbackup_types.go
package v1

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

// MySQLBackupSpec 定义期望状态（用户可写的字段）
// +kubebuilder:object:root=true
// +kubebuilder:storageversion
type MySQLBackupSpec struct {
    // MySQL集群引用
    // +kubebuilder:validation:required
    MySQLClusterRef ClusterRef `json:"mysqlClusterRef"`

    // 备份类型：full（全量）或 incremental（增量）
    // +kubebuilder:validation:Enum=full;incremental
    // +kubebuilder:default=full
    BackupType string `json:"backupType,omitempty"`

    // 备份保留天数
    // +kubebuilder:validation:minimum=1
    // +kubebuilder:validation:maximum=365
    // +kubebuilder:default=7
    // +kubebuilder:example:=14
    RetentionDays int `json:"retentionDays,omitempty"`

    // 备份目标路径（可以是S3、NFS等）
    // +kubebuilder:validation:pattern=^(s3|nfs|local):.*
    // +kubebuilder:example:="s3://my-bucket/backups"
    BackupPath string `json:"backupPath,omitempty"`

    // 备份的副本数（用于分布式备份）
    // +kubebuilder:validation:minimum=1
    // +kubebuilder:default=1
    Replicas int `json:"replicas,omitempty"`
}

// ClusterRef 引用其他K8s资源
type ClusterRef struct {
    Name      string `json:"name"`
    Namespace string `json:"namespace,omitempty"`
}

// MySQLBackupStatus 定义实际状态（Controller写入，用户只读）
// +kubebuilder:object:root=true
type MySQLBackupStatus struct {
    // 当前备份阶段
    // +kubebuilder:validation:Enum=Pending;Running;Completed;Failed
    Phase string `json:"phase,omitempty"`

    // 已完成的副本数
    CompletedReplicas int32 `json:"completedReplicas,omitempty"`

    // 开始时间
    StartTime *metav1.Time `json:"startTime,omitempty"`

    // 完成时间
    CompletionTime *metav1.Time `json:"completionTime,omitempty"`

    // 最新的备份快照ID
    SnapshotID string `json:"snapshotID,omitempty"`

    // 关联Job的选择器（供HPA使用）
    LabelSelector string `json:"labelSelector,omitempty"`
}

// +kubebuilder:object:root=true
// +kubebuilder:subresource:status  // 生成 /status子资源
// +kubebuilder:subresource:scale   // 生成 /scale子资源（需要设置specReplicasPath）
// +kubebuilder:printcolumn:name="Cluster",type="string",JSONPath=".spec.mysqlClusterRef.name"
// +kubebuilder:printcolumn:name="Type",type="string",JSONPath=".spec.backupType"
// +kubebuilder:printcolumn:name="Phase",type="string",JSONPath=".status.phase"
// +kubebuilder:printcolumn:name="Age",type="date",JSONPath=".metadata.creationTimestamp"
// +kubebuilder:resource:shortName=mb,scope=Namespaced
// +kubebuilder:categories=all,backup

type MySQLBackup struct {
    metav1.TypeMeta   `json:",inline"`
    metav1.ObjectMeta `json:"metadata,omitempty"`

    Spec   MySQLBackupSpec   `json:"spec,omitempty"`
    Status MySQLBackupStatus `json:"status,omitempty"`
}

// MySQLBackupList
// +kubebuilder:object:root=true
type MySQLBackupList struct {
    metav1.TypeMeta `json:",inline"`
    metav1.ListMeta `json:"metadata,omitempty"`
    Items           []MySQLBackup `json:"items"`
}

func init() {
    SchemeBuilder.Register(&MySQLBackup{}, &MySQLBackupList{})
}

4.3 控制器实现：Reconcile + Predicates过滤

Predicates（谓词过滤器）是Kubebuilder中控制事件粒度的重要工具。默认情况下，一个Controller会接收到所有相关资源的所有变更事件。Predicates允许你过滤掉不需要处理的事件，减少无意义的Reconcile调用。

// controllers/mysqlbackup_controller.go
package controllers

import (
    "context"
    "fmt"
    "time"

    "github.com/go-logr/logr"
    batchv1 "k8s.io/api/batch/v1"
    corev1 "k8s.io/api/core/v1"
    "k8s.io/apimachinery/pkg/runtime"
    "sigs.k8s.io/controller-runtime/pkg/client"
    ctrl "sigs.k8s.io/controller-runtime/pkg/controller"
    "sigs.k8s.io/controller-runtime/pkg/handler"
    "sigs.k8s.io/controller-runtime/pkg/predicate"
    "sigs.k8s.io/controller-runtime/pkg/reconcile"
    "sigs.k8s.io/controller-runtime/pkg/source"

    mysqlv1alpha1 "github.com/example/mysql-operator/api/v1"
)

type MySQLBackupReconciler struct {
    client.Client
    Log    logr.Logger
    Scheme *runtime.Scheme
}

// +kubebuilder:rbac:groups=mysql.example.com,resources=mysqlbackups,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=mysql.example.com,resources=mysqlbackups/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=mysql.example.com,resources=mysqlbackups/finalizers,verbs=update
// +kubebuilder:rbac:groups=batch,resources=jobs,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups="",resources=pods,verbs=get;list

func (r *MySQLBackupReconciler) Reconcile(ctx context.Context,
    req ctrl.Request) (ctrl.Result, error) {
    log := r.Log.WithValues("mysqlbackup", req.NamespacedName)

    // 获取CR实例
    backup := &mysqlv1alpha1.MySQLBackup{}
    if err := r.Get(ctx, req.NamespacedName, backup); err != nil {
        return reconcile.Result{}, client.IgnoreNotFound(err)
    }

    // ============ 清理逻辑：如果CR被标记为删除 ============
    if !backup.DeletionTimestamp.IsZero() {
        if controllerutil.ContainsFinalizer(backup, "mysql.example.com/finalizer") {
            if err := r.cleanupResources(ctx, backup); err != nil {
                return reconcile.Result{}, fmt.Errorf("cleanup resources: %w", err)
            }
            controllerutil.RemoveFinalizer(backup, "mysql.example.com/finalizer")
            return reconcile.Result{}, r.Update(ctx, backup)
        }
        return reconcile.Result{}, nil
    }

    // ============ 添加Finalizer（防止误删关联资源） ============
    if !controllerutil.ContainsFinalizer(backup, "mysql.example.com/finalizer") {
        controllerutil.AddFinalizer(backup, "mysql.example.com/finalizer")
        return reconcile.Result{}, r.Update(ctx, backup)
    }

    // ============ 主Reconcile逻辑 ============
    return r.reconcileBackup(ctx, backup)
}

func (r *MySQLBackupReconciler) reconcileBackup(ctx context.Context,
    backup *mysqlv1alpha1.MySQLBackup) (ctrl.Result, error) {
    log := r.Log.WithName("reconcileBackup")

    // 检查是否已满足完成条件
    if backup.Status.Phase == "Completed" {
        // 如果已完成且超过保留期，标记为可删除
        retentionDeadline := backup.Status.CompletionTime.Add(
            time.Duration(backup.Spec.RetentionDays) * 24 * time.Hour)
        if time.Now().After(retentionDeadline) {
            log.Info("Backup expired, ready for cleanup")
            return reconcile.Result{}, nil
        }
        // 已完成且在保留期内，定期检查
        return reconcile.Result{RequeueAfter: 1 * time.Hour}, nil
    }

    if backup.Status.Phase == "Failed" {
        // 失败状态：检查是否超过最大重试次数
        if backup.Spec.Retries >= 3 {
            log.Info("Backup failed after max retries, giving up")
            return reconcile.Result{}, nil
        }
        // 指数退避重试：5s * 2^retries
        delay := 5 * time.Second * time.Duration(1<<backup.Spec.Retries)
        return reconcile.Result{RequeueAfter: delay}, nil
    }

    // Pending 或空状态：创建备份Job
    if backup.Status.Phase == "" || backup.Status.Phase == "Pending" {
        job := r.buildBackupJob(backup)
        if err := r.Create(ctx, job); err != nil {
            return reconcile.Result{}, fmt.Errorf("create backup job: %w", err)
        }
        backup.Status.Phase = "Running"
        backup.Status.StartTime = &metav1.Time{Time: time.Now()}
        backup.Status.CompletedReplicas = 0
        return reconcile.Result{}, r.Status().Update(ctx, backup)
    }

    return reconcile.Result{RequeueAfter: 30 * time.Second}, nil
}

func (r *MySQLBackupReconciler) buildBackupJob(
    backup *mysqlv1alpha1.MySQLBackup) *batchv1.Job {
    labels := map[string]string{
        "app":           "mysql-backup",
        "backup-owner":  backup.Name,
        "managed-by":    "mysql-operator",
        "cluster":       backup.Spec.MySQLClusterRef.Name,
    }
    return &batchv1.Job{
        ObjectMeta: metav1.ObjectMeta{
            Name:      "backup-" + backup.Name,
            Namespace: backup.Namespace,
            Labels:    labels,
            OwnerReferences: []metav1.OwnerReference{
                *metav1.NewControllerRef(backup, mysqlv1alpha1.GroupVersion.WithKind("MySQLBackup")),
            },
        },
        Spec: batchv1.JobSpec{
            ttlSecondsAfterFinished: int32(backup.Spec.RetentionDays * 86400),
            Parallelism:              ptr.To(int32(1)),
            BackoffLimit:             ptr.To(int32(backup.Spec.Retries)),
            Template: corev1.PodTemplateSpec{
                ObjectMeta: metav1.ObjectMeta{Labels: labels},
                Spec: corev1.PodSpec{
                    RestartPolicy: corev1.RestartPolicyOnFailure,
                    Containers: []corev1.Container{{
                        Name:  "backup",
                        Image: "registry.example.com/mysql-backup:v2.1.0",
                        ImagePullPolicy: corev1.PullIfNotPresent,
                        Env: []corev1.EnvVar{
                            {Name: "BACKUP_PATH", Value: backup.Spec.BackupPath},
                            {Name: "MYSQL_CLUSTER", Value: backup.Spec.MySQLClusterRef.Name},
                            {Name: "MYSQL_NAMESPACE", Value: backup.Spec.MySQLClusterRef.Namespace},
                        },
                    }},
                },
            },
        },
    }
}

func (r *MySQLBackupReconciler) cleanupResources(ctx context.Context,
    backup *mysqlv1alpha1.MySQLBackup) error {
    jobList := &batchv1.JobList{}
    if err := r.List(ctx, jobList,
        client.InNamespace(backup.Namespace),
        client.MatchingLabels{"backup-owner": backup.Name}); err != nil {
        return err
    }
    for _, job := range jobList.Items {
        if err := r.Delete(ctx, &job); err != nil && !errors.IsNotFound(err) {
            return err
        }
    }
    return nil
}

// SetupWithManager 设置Controller和Informer
func (r *MySQLBackupReconciler) SetupWithManager(mgr manager.Manager) error {
    return ctrl.NewControllerManagedBy(mgr).
        For(&mysqlv1alpha1.MySQLBackup{},
            // Predicates：只关心Spec变化，不关心Status和Annotations变化
            builder.WithPredicates(predicate.Funcs{
                CreateFunc: func(e event.CreateEvent) bool {
                    return true // 所有Create事件都需要处理
                },
                UpdateFunc: func(e event.UpdateEvent) bool {
                    // 忽略元数据和Status的更新，只处理Spec变化
                    if e.ObjectOld.GetGeneration() != e.ObjectNew.GetGeneration() {
                        return true
                    }
                    return false
                },
                DeleteFunc: func(e event.DeleteEvent) bool {
                    return true
                },
            }),
        ).
        Owns(&batchv1.Job{},
            // 只关心我们创建的Job（带特定标签）
            builder.Owns(&batchv1.Job{},
                builder.WithPredicates(predicate.ResourceVersionChangedPredicate{}),
            ),
        ).
        Complete(r)
}

      
4.4 RBAC权限模型：Marker注释自动生成
      
Kubebuilder通过Go语言的注释（Marker）自动生成RBAC配置文件。开发者不需要手动编写Role YAML，controller-gen会自动扫描所有Controller中的RBAC Marker，生成最小权限的ClusterRole/Role。
      
// controllers/mysqlbackup_controller.go 中的RBAC标记
// +kubebuilder:rbac:groups=mysql.example.com,resources=mysqlbackups,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=mysql.example.com,resources=mysqlbackups/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=mysql.example.com,resources=mysqlbackups/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
// +kubebuilder:rbac:groups="",resources=pods,verbs=get;list;watch
// +kubebuilder:rbac:groups="",resources=persistentvolumeclaims,verbs=get;list
// +kubebuilder:rbac:groups="",resources=events,verbs=create;patch
// +kubebuilder:rbac:groups=coordination.k8s.io,resources=leases,verbs=get;create;update

// 运行 make manifests 后自动生成 config/rbac/role.yaml
// generated config/rbac/role.yaml:
/*
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: manager-role
rules:
- apiGroups:
  - mysql.example.com
  resources:
  - mysqlbackups
  verbs:
  - get;list;watch;create;update;patch;delete
- apiGroups:
  - mysql.example.com
  resources:
  - mysqlbackups/status
  verbs:
  - get;update;patch
...
*/

      
4.5 Webhook：Defaulter + ValidatingWebhook
      
Webhook是Operator与API Server解耦验证逻辑的关键通道。在生产环境中，ValidatingWebhook应该在资源被提交到etcd之前验证其合法性——而不是等到Controller运行时才发现问题。
      
// api/v1/mysqlbackup_webhook.go
package v1

import (
    "fmt"
    "net/http"
    "strings"

    "k8s.io/apimachinery/pkg/runtime"
    "k8s.io/apimachinery/pkg/util/validation/field"
    ctrl "sigs.k8s.io/controller-runtime"
    log "sigs.k8s.io/controller-runtime/pkg/log"
    "sigs.k8s.io/controller-runtime/pkg/webhook"
)

// +kubebuilder:webhook:path=/validate-mysql-example-com-v1-mysqlbackup,
//   mutating=false, failurePolicy=fail, groups=mysql.example.com,
//   resources=mysqlbackups, verbs=create;update, name=vmysqlbackup.kb.io,
//   sideEffects=None, admissionReviewVersions=v1

func (r *MySQLBackup) ValidateCreate() error {
    var allErrs field.ErrorList
    if r.Spec.MySQLClusterRef.Name == "" {
        allErrs = append(allErrs, field.Required(
            field.NewPath("spec").Child("mysqlClusterRef").Child("name"),
            "mysql cluster name is required"))
    }
    if !strings.HasPrefix(r.Spec.BackupPath, "s3://") &&
       !strings.HasPrefix(r.Spec.BackupPath, "nfs://") &&
       !strings.HasPrefix(r.Spec.BackupPath, "local://") {
        allErrs = append(allErrs, field.Invalid(
            field.NewPath("spec").Child("backupPath"),
            r.Spec.BackupPath,
            "backupPath must start with s3://, nfs://, or local://"))
    }
    if len(allErrs) > 0 {
        return field.Invalid(field.NewPath("spec"), r.Spec, allErrs.ToAggregate().Error())
    }
    return nil
}

func (r *MySQLBackup) ValidateUpdate(old runtime.Object) error {
    oldBackup := old.(*MySQLBackup)
    // 不允许修改引用的集群
    if r.Spec.MySQLClusterRef.Name != oldBackup.Spec.MySQLClusterRef.Name {
        return field.Forbidden(field.NewPath("spec").Child("mysqlClusterRef").Child("name"),
            "cannot change mysql cluster reference after creation")
    }
    // 已完成的备份不允许修改Spec
    if oldBackup.Status.Phase == "Completed" {
        return field.Forbidden(field.NewPath("spec"),
            "cannot modify spec of completed backup")
    }
    return nil
}

func (r *MySQLBackup) ValidateDelete() error {
    return nil // 允许删除
}

// Webhook实现
func (r *MySQLBackup) SetupWebhookWithManager(mgr ctrl.Manager) error {
    return ctrl.NewWebhookManagedBy(mgr).
        For(r).
        Complete()
}

// Main函数中注册Webhook（main.go）
/*
func main() {
    manager, _ := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
        Port: 9443,
    })

    if err := (&mysqlv1alpha1.MySQLBackup{}).SetupWebhookWithManager(manager); err != nil {
        setupLog.Error(err, "unable to create webhook")
        os.Exit(1)
    }
    manager.Start(ctx)
}
*/

      
4.6 Deployment + HPA的协调逻辑
      
在Operator中协调Deployment和HPA（Horizontal Pod Autoscaler）是一个常见需求。一个常见的模式是：Operator负责创建和管理应用的Deployment和HPA资源，用户只需要声明应用的镜像、副本数等基础配置。
      
// 在Reconcile中添加Deployment + HPA协调逻辑
func (r *MySQLBackupReconciler) reconcileDeploymentAndHPA(ctx context.Context,
    backup *mysqlv1alpha1.MySQLBackup) error {

    // 获取或创建Deployment
    existingDeploy := &appsv1.Deployment{}
    err := r.Get(ctx, client.ObjectKey{
        Name:      "mysql-backup-deploy-" + backup.Name,
        Namespace: backup.Namespace,
    }, existingDeploy)

    desiredDeploy := r.buildDeployment(backup)
    if errors.IsNotFound(err) {
        return r.Create(ctx, desiredDeploy)
    } else if err != nil {
        return err
    }

    // 更新Deployment（如果Spec变化）
    if !equality.Semantic.DeepEqual(existingDeploy.Spec, desiredDeploy.Spec) {
        existingDeploy.Spec = desiredDeploy.Spec
        return r.Update(ctx, existingDeploy)
    }

    // 获取或创建HPA
    existingHPA := &autoscalingv2.HorizontalPodAutoscaler{}
    hpaErr := r.Get(ctx, client.ObjectKey{
        Name:      "mysql-backup-hpa-" + backup.Name,
        Namespace: backup.Namespace,
    }, existingHPA)

    desiredHPA := r.buildHPA(backup)
    if errors.IsNotFound(hpaErr) {
        return r.Create(ctx, desiredHPA)
    } else if hpaErr != nil {
        return hpaErr
    }

    // HPA更新：通常只更新minReplicas、maxReplicas
    if existingHPA.Spec.MinReplicas != desiredHPA.Spec.MinReplicas ||
       existingHPA.Spec.MaxReplicas != desiredHPA.Spec.MaxReplicas {
        existingHPA.Spec = desiredHPA.Spec
        return r.Update(ctx, existingHPA)
    }

    return nil
}

func (r *MySQLBackupReconciler) buildDeployment(
    backup *mysqlv1alpha1.MySQLBackup) *appsv1.Deployment {
    replicas := int32(backup.Spec.Replicas)
    return &appsv1.Deployment{
        ObjectMeta: metav1.ObjectMeta{
            Name:      "mysql-backup-deploy-" + backup.Name,
            Namespace: backup.Namespace,
            OwnerReferences: []metav1.OwnerReference{
                *metav1.NewControllerRef(backup, mysqlv1alpha1.GroupVersion.WithKind("MySQLBackup")),
            },
        },
        Spec: appsv1.DeploymentSpec{
            Replicas: &replicas,
            Selector: &metav1.LabelSelector{
                MatchLabels: map[string]string{"app": "mysql-backup-" + backup.Name},
            },
            Template: corev1.PodTemplateSpec{
                ObjectMeta: metav1.ObjectMeta{
                    Labels: map[string]string{"app": "mysql-backup-" + backup.Name},
                },
                Spec: corev1.PodSpec{
                    Containers: []corev1.Container{{
                        Name:  "backup-agent",
                        Image: "registry.example.com/backup-agent:v1.0.0",
                        Resources: corev1.ResourceRequirements{
                            Requests: corev1.ResourceList{
                                corev1.ResourceCPU:    resource.MustParse("100m"),
                                corev1.ResourceMemory: resource.MustParse("128Mi"),
                            },
                            Limits: corev1.ResourceList{
                                corev1.ResourceCPU:    resource.MustParse("500m"),
                                corev1.ResourceMemory: resource.MustParse("512Mi"),
                            },
                        },
                    }},
                },
            },
        },
    }
}

func (r *MySQLBackupReconciler) buildHPA(
    backup *mysqlv1alpha1.MySQLBackup) *autoscalingv2.HorizontalPodAutoscaler {
    minReplicas := int32(1)
    maxReplicas := int32(backup.Spec.Replicas)
    targetCPU := int32(70)
    return &autoscalingv2.HorizontalPodAutoscaler{
        ObjectMeta: metav1.ObjectMeta{
            Name:      "mysql-backup-hpa-" + backup.Name,
            Namespace: backup.Namespace,
            OwnerReferences: []metav1.OwnerReference{
                *metav1.NewControllerRef(backup, mysqlv1alpha1.GroupVersion.WithKind("MySQLBackup")),
            },
        },
        Spec: autoscalingv2.HorizontalPodAutoscalerSpec{
            ScaleTargetRef: autoscalingv2.CrossVersionObjectReference{
                APIVersion: "apps/v1", Kind: "Deployment",
                Name: "mysql-backup-deploy-" + backup.Name,
            },
            MinReplicas: &minReplicas,
            MaxReplicas: maxReplicas,
            Metrics: []autoscalingv2.MetricSpec{{
                Type: autoscalingv2.ResourceMetricSourceType,
                Resource: &autoscalingv2.ResourceMetricSource{
                    Name: corev1.ResourceCPU,
                    Target: autoscalingv2.MetricTarget{
                        Type:               autoscalingv2.UtilizationMetricType,
                        AverageUtilization: &targetCPU,
                    },
                },
            }},
        },
    }
}