Kubernetes Operator 开发指南

概述

Kubernetes Operator 是将人类运维知识编码为软件的范式。它通过 CRD（Custom Resource Definition）扩展 K8s API，通过 Controller 实现自动化运维逻辑。数据库集群管理、消息队列运维、证书轮转……这些原本需要 SRE 手动操作的工作，Operator 可以自动完成。

Operator 的核心思想是声明式 + 控制循环：用户声明期望状态（如"3 个 Redis 副本"），Controller 持续调整实际状态以趋近期望状态。理解这一模式，不仅能开发 Operator，更能深入理解 K8s 本身的设计哲学。

本文基于 Kubernetes v1.30、operator-sdk v1.37、kubebuilder v4.0。参考 Operator 模式文档

Operator 核心概念

CRD 与 Controller 的关系

用户创建 CR (Custom Resource) ──→ API Server 存储 ──→ Controller Watch
                                                         ↓
                                                   Reconcile 循环
                                                         ↓
                                              比较期望状态 vs 实际状态
                                                         ↓
                                              创建/更新/删除资源
                                                         ↓
                                              更新 CR Status

声明式 vs 命令式

模式	示例	特点
命令式	“创建 3 个 Pod”	执行一次就结束，不关心后续状态
声明式	“保持 3 个 Pod 运行”	持续监控，自动修复偏差

Operator 是声明式的——用户声明 spec.replicas: 3，Controller 确保实际始终有 3 个副本。如果有人手动删了一个 Pod，Controller 会自动创建新的。

Operator 适合的场景

适合	不适合
有明确生命周期的有状态应用	无状态应用（用 Deployment 即可）
需要领域知识的运维操作	简单的部署和扩缩容
复杂的部署/升级/备份流程	一次性的配置任务
需要自动恢复的场景	纯计算任务

CRD 设计

CRD 基础

apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
metadata:
  name: redisclusters.cache.example.com
spec:
  group: cache.example.com
  names:
    kind: RedisCluster
    listKind: RedisClusterList
    singular: rediscluster
    plural: redisclusters
    shortNames:
    - rdc
  scope: Namespaced
  versions:
  - name: v1alpha1
    served: true
    storage: true
    schema:
      openAPIV3Schema:
        type: object
        properties:
          spec:
            type: object
            required: ["size", "image"]
            properties:
              size:
                type: integer
                minimum: 1
                maximum: 10
                description: "Redis 副本数"
              image:
                type: string
                description: "Redis 镜像"
              version:
                type: string
                description: "Redis 版本"
              config:
                type: object
                properties:
                  maxmemory:
                    type: string
                  maxmemoryPolicy:
                    type: string
                    enum: ["allkeys-lru", "volatile-lru", "noeviction"]
              storage:
                type: object
                properties:
                  size:
                    type: string
                  storageClass:
                    type: string
          status:
            type: object
            properties:
              phase:
                type: string
                enum: ["Pending", "Running", "Failed", "Scaling"]
              replicas:
                type: integer
              readyReplicas:
                type: integer
              conditions:
                type: array
                items:
                  type: object
                  properties:
                    type:
                      type: string
                    status:
                      type: string
                      enum: ["True", "False", "Unknown"]
                    lastTransitionTime:
                      type: string
                      format: date-time
                    reason:
                      type: string
                    message:
                      type: string
    subresources:
      status: {}           # 启用 /status 子资源
      scale:                # 启用 /scale 子资源
        specReplicasPath: .spec.size
        statusReplicasPath: .status.replicas
    additionalPrinterColumns:
    - name: Size
      type: integer
      jsonPath: .spec.size
    - name: Image
      type: string
      jsonPath: .spec.image
    - name: Phase
      type: string
      jsonPath: .status.phase
    - name: Ready
      type: string
      jsonPath: .status.readyReplicas
    - name: Age
      type: date
      jsonPath: .metadata.creationTimestamp

CRD 设计原则

原则	说明	示例
声明式	描述期望状态而非操作步骤	`size: 3` 而非 `create 3 pods`
幂等	多次应用结果一致	Reconcile 可重复执行
可观察	Status 反映实际状态	`status.phase: Running`
向后兼容	新版本不破坏旧版本	新字段用 optional
分层设计	Spec 和 Status 分离	Spec 是用户意图，Status 是实际

CRD 版本管理

versions:
- name: v1alpha1        # 初始版本
  served: true
  storage: false         # 不再存储新数据
  schema: ...

- name: v1beta1          # 改进版本
  served: true
  storage: false
  schema: ...

- name: v1               # 稳定版本
  served: true
  storage: true          # 当前存储版本
  schema: ...

版本策略：v1alpha1 → v1beta1 → v1。API 稳定后才升级到 v1。storage: true 的版本只能有一个，其他版本的数据会自动转换到该版本。

Validation Webhook

CRD 的 OpenAPI Schema 验证能力有限，复杂验证用 Validating Webhook：

// 验证 RedisCluster 的 size 不能在运行时缩小
func (v *RedisClusterValidator) ValidateCreate(
    ctx context.Context,
    obj runtime.Object,
) admission.Warnings {
    redis := obj.(*cachev1.RedisCluster)
    if redis.Spec.Size < 1 {
        return nil, fmt.Errorf("size must be >= 1")
    }
    return nil
}

func (v *RedisClusterValidator) ValidateUpdate(
    ctx context.Context,
    oldObj, newObj runtime.Object,
) admission.Warnings {
    oldRedis := oldObj.(*cachev1.RedisCluster)
    newRedis := newObj.(*cachev1.RedisCluster)

    // 不允许缩小集群
    if newRedis.Spec.Size < oldRedis.Spec.Size {
        return nil, fmt.Errorf("scale down is not allowed")
    }
    return nil
}

Controller 原理

Informer 机制

Controller 通过 Informer 机制监听资源变化：

API Server → Watch → Informer → Event Handlers → Work Queue → Reconcile
                ↑
            Cache（本地缓存，减少 API Server 压力）

Informer 的核心组件：

组件	功能
`Reflector`	从 API Server List/Watch 资源
`Indexer`	本地缓存，支持索引查询
`WorkQueue`	事件队列，去重和限速
`Handlers`	事件回调，将对象 key 入队

Reconcile 循环

Reconcile 是 Controller 的核心逻辑：

func (r *RedisClusterReconciler) Reconcile(
    ctx context.Context,
    req ctrl.Request,
) (ctrl.Result, error) {
    // 1. 获取 CR
    redis := &cachev1.RedisCluster{}
    if err := r.Get(ctx, req.NamespacedName, redis); err != nil {
        if errors.IsNotFound(err) {
            return ctrl.Result{}, nil  // 已删除，忽略
        }
        return ctrl.Result{}, err
    }

    // 2. 比较期望状态 vs 实际状态
    // 3. 调整实际状态
    // 4. 更新 CR Status
    // 5. 返回结果（可能需要重新入队）
}

Reconcile 设计原则

原则	说明
幂等	同一输入多次执行结果一致
无状态	不依赖外部状态，只依赖 CR 和集群状态
短小	单次执行快速完成（< 30s）
可重入	任何时候被调用都能正确处理
不阻塞	长操作异步化，用 Requeue 等待

完整 Reconcile 示例

func (r *RedisClusterReconciler) Reconcile(
    ctx context.Context,
    req ctrl.Request,
) (ctrl.Result, error) {
    logger := log.FromContext(ctx)

    // ===== 1. 获取 CR =====
    redis := &cachev1.RedisCluster{}
    if err := r.Get(ctx, req.NamespacedName, redis); err != nil {
        if errors.IsNotFound(err) {
            logger.Info("RedisCluster resource not found, ignoring")
            return ctrl.Result{}, nil
        }
        logger.Error(err, "Failed to get RedisCluster")
        return ctrl.Result{}, err
    }

    // ===== 2. 处理删除 =====
    if !redis.DeletionTimestamp.IsZero() {
        return r.reconcileDelete(ctx, redis)
    }

    // ===== 3. 确保 Finalizer 存在 =====
    if !controllerutil.ContainsFinalizer(redis, "cache.example.com/finalizer") {
        controllerutil.AddFinalizer(redis, "cache.example.com/finalizer")
        if err := r.Update(ctx, redis); err != nil {
            return ctrl.Result{}, err
        }
    }

    // ===== 4. 确保 ConfigMap =====
    if err := r.reconcileConfigMap(ctx, redis); err != nil {
        r.updateStatus(ctx, redis, "Failed", err.Error())
        return ctrl.Result{RequeueAfter: 10 * time.Second}, err
    }

    // ===== 5. 确保 Service =====
    if err := r.reconcileService(ctx, redis); err != nil {
        r.updateStatus(ctx, redis, "Failed", err.Error())
        return ctrl.Result{RequeueAfter: 10 * time.Second}, err
    }

    // ===== 6. 确保 StatefulSet =====
    if err := r.reconcileStatefulSet(ctx, redis); err != nil {
        r.updateStatus(ctx, redis, "Failed", err.Error())
        return ctrl.Result{RequeueAfter: 10 * time.Second}, err
    }

    // ===== 7. 检查就绪状态 =====
    ready, err := r.checkReady(ctx, redis)
    if err != nil {
        return ctrl.Result{RequeueAfter: 30 * time.Second}, err
    }

    if ready {
        r.updateStatus(ctx, redis, "Running", "")
    } else {
        r.updateStatus(ctx, redis, "Pending", "waiting for pods to be ready")
        return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
    }

    return ctrl.Result{}, nil
}

开发工具链

operator-sdk vs kubebuilder

特性	operator-sdk	kubebuilder
维护方	Red Hat	Kubernetes SIG
语言	Go/Ansible/Helm	Go
项目结构	相同	相同
Helm/Ansible Operator	支持	不支持
OLM 集成	支持	不支持
学习资料	更丰富	官方文档好
推荐度	生产推荐	纯 Go 推荐

两者底层都使用 controller-runtime 库，项目结构基本一致。operator-sdk 多了 Helm/Ansible Operator 和 OLM 集成。

创建项目（kubebuilder）

# 安装 kubebuilder
curl -L -o kubebuilder "https://github.com/kubernetes-sigs/kubebuilder/releases/download/v4.0.0/kubebuilder_linux_amd64"
chmod +x kubebuilder && mv kubebuilder /usr/local/bin/

# 创建项目
mkdir redis-operator && cd redis-operator
kubebuilder init \
  --domain example.com \
  --repo github.com/myorg/redis-operator \
  --license apache2

# 创建 API
kubebuilder create api \
  --group cache \
  --version v1alpha1 \
  --kind RedisCluster \
  --resource \
  --controller

# 生成 CRD
make manifests

# 安装 CRD
make install

# 运行 Controller
make run

项目结构

redis-operator/
├── api/
│   └── v1alpha1/
│       ├── rediscluster_types.go    # CRD 类型定义
│       ├── rediscluster_webhook.go  # Webhook
│       ├── groupversion_info.go     # GV 信息
│       └── zz_generated.deepcopy.go # 自动生成
├── cmd/
│   └── main.go                       # 入口
├── config/
│   ├── crd/                          # CRD YAML
│   ├── default/                      # 默认部署配置
│   ├── manager/                      # Manager 部署配置
│   ├── rbac/                         # RBAC 配置
│   └── webhook/                      # Webhook 配置
├── internal/
│   └── controller/
│       └── rediscluster_controller.go  # Controller 逻辑
├── Dockerfile
└── Makefile

定义 CRD 类型

// api/v1alpha1/rediscluster_types.go
package v1alpha1

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

// RedisClusterSpec 定义期望状态
type RedisClusterSpec struct {
    //+kubebuilder:validation:Minimum=1
    //+kubebuilder:validation:Maximum=10
    Size int32 `json:"size"`

    //+kubebuilder:validation:Required
    Image string `json:"image"`

    Version string `json:"version,omitempty"`

    Config RedisConfig `json:"config,omitempty"`

    Storage StorageSpec `json:"storage,omitempty"`
}

type RedisConfig struct {
    Maxmemory       string `json:"maxmemory,omitempty"`
    MaxmemoryPolicy string `json:"maxmemoryPolicy,omitempty"`
}

type StorageSpec struct {
    //+kubebuilder:validation:Pattern=^[0-9]+(Gi|Mi|Ti)$
    Size         string `json:"size,omitempty"`
    StorageClass string `json:"storageClass,omitempty"`
}

// RedisClusterStatus 定义实际状态
type RedisClusterStatus struct {
    Phase         string      `json:"phase,omitempty"`
    Replicas      int32       `json:"replicas,omitempty"`
    ReadyReplicas int32       `json:"readyReplicas,omitempty"`
    Conditions    []Condition `json:"conditions,omitempty"`
}

type Condition struct {
    Type               string      `json:"type"`
    Status             string      `json:"status"`
    LastTransitionTime metav1.Time `json:"lastTransitionTime,omitempty"`
    Reason             string      `json:"reason,omitempty"`
    Message            string      `json:"message,omitempty"`
}

//+kubebuilder:object:root=true
//+kubebuilder:subresource:status
//+kubebuilder:subresource:scale:specpath=.spec.size,statuspath=.status.replicas
type RedisCluster struct {
    metav1.TypeMeta   `json:",inline"`
    metav1.ObjectMeta `json:"metadata,omitempty"`

    Spec   RedisClusterSpec   `json:"spec,omitempty"`
    Status RedisClusterStatus `json:"status,omitempty"`
}

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

Controller 实现

// internal/controller/rediscluster_controller.go
package controller

import (
    "context"
    "fmt"
    "time"

    appsv1 "k8s.io/api/apps/v1"
    corev1 "k8s.io/api/core/v1"
    "k8s.io/apimachinery/pkg/api/errors"
    "k8s.io/apimachinery/pkg/api/resource"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/apimachinery/pkg/runtime"
    "k8s.io/apimachinery/pkg/types"
    "k8s.io/apimachinery/pkg/util/intstr"
    ctrl "sigs.k8s.io/controller-runtime"
    "sigs.k8s.io/controller-runtime/pkg/client"
    "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
    "sigs.k8s.io/controller-runtime/pkg/log"

    cachev1 "github.com/myorg/redis-operator/api/v1alpha1"
)

type RedisClusterReconciler struct {
    client.Client
    Scheme *runtime.Scheme
}

//+kubebuilder:rbac:groups=cache.example.com,resources=redisclusters,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=cache.example.com,resources=redisclusters/status,verbs=get;update;patch
//+kubebuilder:rbac:groups=cache.example.com,resources=redisclusters/finalizers,verbs=update
//+kubebuilder:rbac:groups=apps,resources=statefulsets,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups="",resources=services;configmaps,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups="",resources=pods,verbs=get;list;watch

func (r *RedisClusterReconciler) Reconcile(
    ctx context.Context,
    req ctrl.Request,
) (ctrl.Result, error) {
    logger := log.FromContext(ctx)

    // 获取 CR
    redis := &cachev1.RedisCluster{}
    if err := r.Get(ctx, req.NamespacedName, redis); err != nil {
        if errors.IsNotFound(err) {
            return ctrl.Result{}, nil
        }
        return ctrl.Result{}, err
    }

    // 确保 ConfigMap
    cm := r.buildConfigMap(redis)
    if err := controllerutil.SetControllerReference(redis, cm, r.Scheme); err != nil {
        return ctrl.Result{}, err
    }
    if err := r.createOrUpdate(ctx, cm); err != nil {
        logger.Error(err, "Failed to reconcile ConfigMap")
        return ctrl.Result{RequeueAfter: 10 * time.Second}, err
    }

    // 确保 Service
    svc := r.buildService(redis)
    if err := controllerutil.SetControllerReference(redis, svc, r.Scheme); err != nil {
        return ctrl.Result{}, err
    }
    if err := r.createOrUpdate(ctx, svc); err != nil {
        return ctrl.Result{RequeueAfter: 10 * time.Second}, err
    }

    // 确保 StatefulSet
    sts := r.buildStatefulSet(redis)
    if err := controllerutil.SetControllerReference(redis, sts, r.Scheme); err != nil {
        return ctrl.Result{}, err
    }
    if err := r.createOrUpdate(ctx, sts); err != nil {
        return ctrl.Result{RequeueAfter: 10 * time.Second}, err
    }

    // 更新 Status
    if err := r.updateStatus(ctx, redis, sts); err != nil {
        return ctrl.Result{RequeueAfter: 10 * time.Second}, err
    }

    // 未就绪则重新入队
    if sts.Status.ReadyReplicas != *sts.Spec.Replicas {
        return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
    }

    return ctrl.Result{}, nil
}

// buildStatefulSet 构建期望的 StatefulSet
func (r *RedisClusterReconciler) buildStatefulSet(redis *cachev1.RedisCluster) *appsv1.StatefulSet {
    replicas := redis.Spec.Size
    return &appsv1.StatefulSet{
        ObjectMeta: metav1.ObjectMeta{
            Name:      redis.Name,
            Namespace: redis.Namespace,
        },
        Spec: appsv1.StatefulSetSpec{
            ServiceName: redis.Name,
            Replicas:    &replicas,
            Selector: &metav1.LabelSelector{
                MatchLabels: map[string]string{
                    "app": redis.Name,
                },
            },
            Template: corev1.PodTemplateSpec{
                ObjectMeta: metav1.ObjectMeta{
                    Labels: map[string]string{
                        "app": redis.Name,
                    },
                },
                Spec: corev1.PodSpec{
                    Containers: []corev1.Container{
                        {
                            Name:  "redis",
                            Image: fmt.Sprintf("%s:%s", redis.Spec.Image, redis.Spec.Version),
                            Ports: []corev1.ContainerPort{
                                {ContainerPort: 6379, Name: "redis"},
                            },
                            VolumeMounts: []corev1.VolumeMount{
                                {Name: "data", MountPath: "/data"},
                                {Name: "config", MountPath: "/etc/redis"},
                            },
                            Resources: corev1.ResourceRequirements{
                                Requests: corev1.ResourceList{
                                    corev1.ResourceCPU:    resource.MustParse("100m"),
                                    corev1.ResourceMemory: resource.MustParse("256Mi"),
                                },
                                Limits: corev1.ResourceList{
                                    corev1.ResourceCPU:    resource.MustParse("500m"),
                                    corev1.ResourceMemory: resource.MustParse("1Gi"),
                                },
                            },
                            LivenessProbe: &corev1.Probe{
                                ProbeHandler: corev1.ProbeHandler{
                                    Exec: &corev1.ExecAction{
                                        Command: []string{"redis-cli", "ping"},
                                    },
                                },
                                InitialDelaySeconds: 10,
                                PeriodSeconds:       10,
                            },
                        },
                    },
                    Volumes: []corev1.Volume{
                        {
                            Name: "config",
                            VolumeSource: corev1.VolumeSource{
                                ConfigMap: &corev1.ConfigMapVolumeSource{
                                    LocalObjectReference: corev1.LocalObjectReference{
                                        Name: redis.Name,
                                    },
                                },
                            },
                        },
                    },
                },
            },
            VolumeClaimTemplates: []corev1.PersistentVolumeClaim{
                {
                    ObjectMeta: metav1.ObjectMeta{
                        Name: "data",
                    },
                    Spec: corev1.PersistentVolumeClaimSpec{
                        AccessModes: []corev1.PersistentVolumeAccessMode{
                            corev1.ReadWriteOnce,
                        },
                        Resources: corev1.ResourceRequirements{
                            Requests: corev1.ResourceList{
                                corev1.ResourceStorage: resource.MustParse(
                                    redis.Spec.Storage.Size,
                                ),
                            },
                        },
                        StorageClassName: strPtr(redis.Spec.Storage.StorageClass),
                    },
                },
            },
        },
    }
}

// createOrUpdate 创建或更新资源
func (r *RedisClusterReconciler) createOrUpdate(
    ctx context.Context,
    obj client.Object,
) error {
    if err := controllerutil.SetControllerReference(
        obj.(metav1.Object),
        obj,
        r.Scheme,
    ); err != nil {
        return err
    }

    // 尝试创建
    err := r.Create(ctx, obj)
    if err == nil {
        return nil
    }

    // 已存在则更新
    if errors.IsAlreadyExists(err) {
        return r.Update(ctx, obj)
    }

    return err
}

// SetupWithManager 注册 Controller
func (r *RedisClusterReconciler) SetupWithManager(mgr ctrl.Manager) error {
    return ctrl.NewControllerManagedBy(mgr).
        For(&cachev1.RedisCluster{}).
        Owns(&appsv1.StatefulSet{}).
        Owns(&corev1.Service{}).
        Owns(&corev1.ConfigMap{}).
        Complete(r)
}

状态管理

Status 更新

func (r *RedisClusterReconciler) updateStatus(
    ctx context.Context,
    redis *cachev1.RedisCluster,
    sts *appsv1.StatefulSet,
) error {
    // 获取最新 CR（避免冲突）
    latest := &cachev1.RedisCluster{}
    if err := r.Get(ctx, types.NamespacedName{
        Name:      redis.Name,
        Namespace: redis.Namespace,
    }, latest); err != nil {
        return err
    }

    // 更新 Status
    latest.Status.Replicas = sts.Status.Replicas
    latest.Status.ReadyReplicas = sts.Status.ReadyReplicas

    if sts.Status.ReadyReplicas == *sts.Spec.Replicas {
        latest.Status.Phase = "Running"
    } else {
        latest.Status.Phase = "Pending"
    }

    // 更新 Conditions
    latest.Status.Conditions = updateCondition(
        latest.Status.Conditions,
        "Ready",
        sts.Status.ReadyReplicas == *sts.Spec.Replicas,
        "AllReplicasReady",
        "All replicas are ready",
    )

    return r.Status().Update(ctx, latest)
}

func updateCondition(
    conditions []cachev1.Condition,
    condType string,
    status bool,
    reason, message string,
) []cachev1.Condition {
    now := metav1.Now()
    statusStr := "False"
    if status {
        statusStr = "True"
    }

    for i, c := range conditions {
        if c.Type == condType {
            if c.Status != statusStr {
                conditions[i].LastTransitionTime = now
            }
            conditions[i].Status = statusStr
            conditions[i].Reason = reason
            conditions[i].Message = message
            return conditions
        }
    }

    return append(conditions, cachev1.Condition{
        Type:               condType,
        Status:             statusStr,
        LastTransitionTime: now,
        Reason:             reason,
        Message:            message,
    })
}

Status 更新的冲突处理

Status 更新可能因 ResourceVersion 变化而冲突，需要重试：

func (r *RedisClusterReconciler) updateStatusWithRetry(
    ctx context.Context,
    redis *cachev1.RedisCluster,
) error {
    return retry.RetryOnConflict(retry.DefaultBackoff, func() error {
        latest := &cachev1.RedisCluster{}
        if err := r.Get(ctx, client.ObjectKeyFromObject(redis), latest); err != nil {
            return err
        }
        latest.Status = redis.Status
        return r.Status().Update(ctx, latest)
    })
}

Finalizer 与删除

Finalizer 机制

Finalizer 确保删除 CR 前先执行清理操作：

func (r *RedisClusterReconciler) reconcileDelete(
    ctx context.Context,
    redis *cachev1.RedisCluster,
) (ctrl.Result, error) {
    logger := log.FromContext(ctx)

    if controllerutil.ContainsFinalizer(redis, "cache.example.com/finalizer") {
        // 执行清理：删除外部资源、通知外部系统等
        if err := r.cleanupExternalResources(ctx, redis); err != nil {
            logger.Error(err, "Failed to cleanup external resources")
            return ctrl.Result{RequeueAfter: 10 * time.Second}, err
        }

        // 移除 Finalizer
        controllerutil.RemoveFinalizer(redis, "cache.example.com/finalizer")
        if err := r.Update(ctx, redis); err != nil {
            return ctrl.Result{}, err
        }
    }

    return ctrl.Result{}, nil
}

func (r *RedisClusterReconciler) cleanupExternalResources(
    ctx context.Context,
    redis *cachev1.RedisCluster,
) error {
    // 清理外部 DNS 记录
    // 清理外部负载均衡器
    // 发送通知
    return nil
}

删除流程

用户删除 CR → DeletionTimestamp 设置 → Controller 检测到删除
  → 执行清理逻辑 → 移除 Finalizer → CR 被 GC 删除

Leader Election

为什么需要 Leader Election

Controller 通常部署多副本保证高可用，但只有一个副本应执行 Reconcile 逻辑，否则会产生冲突。Leader Election 确保同一时间只有一个活跃 Controller。

配置

// cmd/main.go
func main() {
    mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
        Scheme:                 scheme,
        Metrics:                server.Options{BindAddress: ":8080"},
        HealthProbeBindAddress: ":8081",
        LeaderElection:         true,                        // 启用 Leader Election
        LeaderElectionID:       "redis-operator.example.com",
        LeaderElectionNamespace: "redis-operator-system",
        LeaseDuration:          15 * time.Second,            // 租约时长
        RenewDeadline:          10 * time.Second,            // 续约超时
        RetryPeriod:            2 * time.Second,             // 重试间隔
    })
    // ...
}

参数	说明	推荐值
`LeaderElectionID`	Lease 名称（全局唯一）	`<operator-name>.<domain>`
`LeaseDuration`	租约有效期	15s
`RenewDeadline`	续约超时时间	10s
`RetryPeriod`	重试间隔	2s

测试

单元测试

// internal/controller/rediscluster_controller_test.go
func TestRedisClusterReconciler(t *testing.T) {
    // 创建 fake client
    cl := fake.NewClientBuilder().
        WithScheme(scheme).
        WithObjects(
            &cachev1.RedisCluster{
                ObjectMeta: metav1.ObjectMeta{
                    Name:      "test-redis",
                    Namespace: "default",
                },
                Spec: cachev1.RedisClusterSpec{
                    Size:  3,
                    Image: "redis",
                    Version: "7.0",
                    Storage: cachev1.StorageSpec{
                        Size: "1Gi",
                    },
                },
            },
        ).
        Build()

    r := &RedisClusterReconciler{
        Client: cl,
        Scheme: scheme,
    }

    // 执行 Reconcile
    result, err := r.Reconcile(context.Background(), ctrl.Request{
        NamespacedName: types.NamespacedName{
            Name:      "test-redis",
            Namespace: "default",
        },
    })

    // 验证
    assert.NoError(t, err)
    assert.False(t, result.Requeue)

    // 验证 StatefulSet 被创建
    sts := &appsv1.StatefulSet{}
    err = cl.Get(context.Background(), types.NamespacedName{
        Name:      "test-redis",
        Namespace: "default",
    }, sts)
    assert.NoError(t, err)
    assert.Equal(t, int32(3), *sts.Spec.Replicas)
}

envtest 集成测试

// 使用 controller-runtime 的 envtest 启动真实 API Server
func TestRedisClusterReconciler_Integration(t *testing.T) {
    testEnv := &envtest.Environment{
        CRDDirectoryPaths:     []string{"../../config/crd/bases"},
        ErrorIfCRDPathMissing: true,
    }

    cfg, _ := testEnv.Start()
    defer testEnv.Stop()

    mgr, _ := ctrl.NewManager(cfg, ctrl.Options{
        Scheme: scheme,
    })

    // 注册 Controller
    r := &RedisClusterReconciler{
        Client: mgr.GetClient(),
        Scheme: scheme,
    }
    r.SetupWithManager(mgr)

    // 启动 Manager
    go mgr.Start(ctrl.SetupSignalHandler())

    // 创建 CR 并验证
    redis := &cachev1.RedisCluster{
        ObjectMeta: metav1.ObjectMeta{
            Name:      "test-redis",
            Namespace: "default",
        },
        Spec: cachev1.RedisClusterSpec{
            Size:  3,
            Image: "redis",
            Version: "7.0",
            Storage: cachev1.StorageSpec{Size: "1Gi"},
        },
    }
    cl.Create(context.Background(), redis)

    // 等待并验证 StatefulSet 被创建
    eventually(t, func() bool {
        sts := &appsv1.StatefulSet{}
        err := cl.Get(context.Background(), types.NamespacedName{
            Name: "test-redis", Namespace: "default",
        }, sts)
        return err == nil && *sts.Spec.Replicas == 3
    }, 10*time.Second, time.Second)
}

发布与运维

构建 Operator 镜像

# Dockerfile
FROM golang:1.22-alpine AS builder

WORKDIR /app
COPY go.mod go.sum ./
RUN go mod download

COPY . .
RUN CGO_ENABLED=0 GOOS=linux go build -ldflags="-s -w" -o manager cmd/main.go

FROM gcr.io/distroless/static-debian12:nonroot
WORKDIR /
COPY --from=builder /app/manager .
USER 65532:65532
ENTRYPOINT ["/manager"]

# 构建和推送
make docker-build IMG=myorg/redis-operator:v0.1.0
make docker-push IMG=myorg/redis-operator:v0.1.0

部署 Operator

# 生成部署 YAML
make build-installer IMG=myorg/redis-operator:v0.1.0

# 部署
make deploy IMG=myorg/redis-operator:v0.1.0

RBAC 配置

# config/rbac/role.yaml
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: redis-operator-role
rules:
- apiGroups: ["cache.example.com"]
  resources: ["redisclusters"]
  verbs: ["get", "list", "watch", "create", "update", "patch", "delete"]
- apiGroups: ["cache.example.com"]
  resources: ["redisclusters/status", "redisclusters/finalizers"]
  verbs: ["get", "update", "patch"]
- apiGroups: ["apps"]
  resources: ["statefulsets"]
  verbs: ["get", "list", "watch", "create", "update", "patch", "delete"]
- apiGroups: [""]
  resources: ["services", "configmaps"]
  verbs: ["get", "list", "watch", "create", "update", "patch", "delete"]
- apiGroups: [""]
  resources: ["pods"]
  verbs: ["get", "list", "watch"]
- apiGroups: [""]
  resources: ["events"]
  verbs: ["create", "patch"]

OLM 发布

# 生成 OLM Bundle
make bundle IMG=myorg/redis-operator:v0.1.0

# 生成 Catalog Source
make catalog-build BUNDLE_IMG=myorg/redis-operator:v0.1.0 \
  CATALOG_IMG=myorg/redis-operator-catalog:v0.1.0

# 通过 OLM 安装
operator-sdk run bundle myorg/redis-operator-bundle:v0.1.0

Operator 升级策略

# 通过 OLM 的 Subscription 自动升级
apiVersion: operators.coreos.com/v1alpha1
kind: Subscription
metadata:
  name: redis-operator
  namespace: redis-operator-system
spec:
  channel: stable
  name: redis-operator
  source: operatorhubio-catalog
  sourceNamespace: olm
  installPlanApproval: Automatic   # 或 Manual
  startingCSV: redis-operator.v0.1.0

版本兼容性策略

变更类型	版本升级	兼容性
新增可选字段	patch	向后兼容
新增 API 版本	minor	双版本共存
删除字段	major	需迁移
修改字段语义	major	需迁移

监控与可观测性

// 添加自定义指标
var (
    redisClusterTotal = prometheus.NewCounter(
        prometheus.CounterOpts{
            Name: "redis_cluster_total",
            Help: "Total number of RedisClusters managed",
        },
    )
    reconcileTotal = prometheus.NewCounterVec(
        prometheus.CounterOpts{
            Name: "controller_reconcile_total",
            Help: "Total number of reconciliations",
        },
        []string{"controller", "result"},
    )
    reconcileDuration = prometheus.NewHistogramVec(
        prometheus.HistogramOpts{
            Name:    "controller_reconcile_duration_seconds",
            Help:    "Reconciliation duration",
            Buckets: prometheus.DefBuckets,
        },
        []string{"controller"},
    )
)

// 在 Reconcile 中记录指标
func (r *RedisClusterReconciler) Reconcile(
    ctx context.Context,
    req ctrl.Request,
) (ctrl.Result, error) {
    start := time.Now()
    defer func() {
        reconcileDuration.WithLabelValues("rediscluster").Observe(time.Since(start).Seconds())
    }()

    // ... reconcile logic ...

    if err != nil {
        reconcileTotal.WithLabelValues("rediscluster", "error").Inc()
    } else {
        reconcileTotal.WithLabelValues("rediscluster", "success").Inc()
    }

    return result, err
}

日志与事件

// 使用结构化日志
logger := log.FromContext(ctx)
logger.Info("reconciling RedisCluster",
    "name", redis.Name,
    "namespace", redis.Namespace,
    "size", redis.Spec.Size,
)

// 发送 K8s Event
r.Recorder.Event(redis, corev1.EventTypeNormal, "Scaling",
    fmt.Sprintf("Scaling RedisCluster to %d replicas", redis.Spec.Size))
r.Recorder.Event(redis, corev1.EventTypeWarning, "ScalingFailed",
    "Failed to scale RedisCluster")

Operator 开发好的实践

CRD 设计

实践	说明
Spec 只描述期望	不要在 Spec 中放运行时状态
Status 反映实际	Status 是只读的，由 Controller 更新
字段有默认值	用 defaulting webhook 或 kubebuilder marker
字段有验证	用 OpenAPI Schema + validating webhook
版本化	从 v1alpha1 开始，稳定后升 v1
文档化	每个字段有 description

Reconcile 实现

实践	说明
幂等	同一输入多次执行结果一致
只比较 Spec	不要根据 Status 决策（可能过时）
用 CreateOrUpdate	而非先 Delete 再 Create
处理冲突	用 RetryOnConflict 处理乐观锁冲突
设置 OwnerReference	确保子资源随 CR 删除
不要 panic	返回 error，让 Requeue 处理
控制并发	用 WorkQueue 限速，避免 API Server 压力

常见错误

错误	后果	解决
忘记设置 OwnerReference	子资源不会随 CR 删除	用 SetControllerReference
Status 更新不重试	ResourceVersion 冲突导致失败	用 RetryOnConflict
Reconcile 中阻塞操作	阻塞 WorkQueue	用 RequeueAfter 异步等待
没有 Finalizer	CR 删除后外部资源残留	添加 Finalizer
RBAC 权限不足	Controller 无法操作资源	用 kubebuilder RBAC marker
忘记处理删除	删除 CR 后 Controller 报错	检查 DeletionTimestamp

总结

Operator 是 K8s 最强大的扩展机制，核心要点：

声明式 + 控制循环：用户声明期望状态，Controller 持续调整实际状态。理解这一模式是开发 Operator 的基础。
CRD 设计是关键：好的 CRD 应该声明式、幂等、可观察。Spec 描述期望，Status 反映实际。
Reconcile 要幂等：任何时候被调用都能正确处理，不依赖外部状态，不阻塞。
Finalizer 不可少：确保删除 CR 前清理外部资源，防止资源泄露。
Leader Election 保证高可用：多副本部署时只有一个活跃 Controller，避免冲突。
测试要充分：单元测试用 fake client，集成测试用 envtest，生产前做完整验证。
版本管理要有策略：从 v1alpha1 开始，稳定后升 v1。保持向后兼容，用 conversion webhook 处理版本迁移。
可观测性是运维基础：日志、指标、事件三者缺一不可。Prometheus 指标 + 结构化日志 + K8s Events 是标配。

Operator 开发的学习曲线较陡，但一旦掌握，你会发现 K8s 本身的 Deployment、StatefulSet、DaemonSet 等控制器遵循的都是同一套模式。理解了 Operator，你就真正理解了 Kubernetes。

参考资料与致谢

本文在撰写过程中参考了以下资料，感谢原作者的贡献：

Operator 模式文档 — Kubernetes 官方，参考了Operator 模式文档相关内容
Operator Capability Model — Operatorhub，参考了Operator Capability Model相关内容

概述#

Operator 核心概念#

CRD 与 Controller 的关系#

声明式 vs 命令式#

Operator 适合的场景#

CRD 设计#

CRD 基础#

CRD 设计原则#

CRD 版本管理#

Validation Webhook#

Controller 原理#

Informer 机制#

Reconcile 循环#

Reconcile 设计原则#

完整 Reconcile 示例#

开发工具链#

operator-sdk vs kubebuilder#

创建项目（kubebuilder）#

项目结构#

定义 CRD 类型#

Controller 实现#

状态管理#

Status 更新#

Status 更新的冲突处理#

Finalizer 与删除#

Finalizer 机制#

删除流程#

Leader Election#

为什么需要 Leader Election#

配置#

测试#

单元测试#

envtest 集成测试#

发布与运维#

构建 Operator 镜像#

部署 Operator#

RBAC 配置#

OLM 发布#

Operator 升级策略#

版本兼容性策略#

监控与可观测性#

日志与事件#

Operator 开发好的实践#

CRD 设计#

Reconcile 实现#

常见错误#

总结#

参考资料与致谢#

概述

Operator 核心概念

CRD 与 Controller 的关系

声明式 vs 命令式

Operator 适合的场景

CRD 设计

CRD 基础

CRD 设计原则

CRD 版本管理

Validation Webhook

Controller 原理

Informer 机制

Reconcile 循环

Reconcile 设计原则

完整 Reconcile 示例

开发工具链

operator-sdk vs kubebuilder

创建项目（kubebuilder）

项目结构

定义 CRD 类型

Controller 实现

状态管理

Status 更新

Status 更新的冲突处理

Finalizer 与删除

Finalizer 机制

删除流程

Leader Election

为什么需要 Leader Election

配置

测试

单元测试

envtest 集成测试

发布与运维

构建 Operator 镜像

部署 Operator

RBAC 配置

OLM 发布

Operator 升级策略

版本兼容性策略

监控与可观测性

日志与事件

Operator 开发好的实践

CRD 设计

Reconcile 实现

常见错误

总结

参考资料与致谢