云原生

Kubernetes Operator与CRD：自定义资源开发实战

深入讲解 Kubernetes CRD（自定义资源定义）、Controller 开发模式、Operator SDK 工具链，以及实战开发一个简单 Operator 的完整流程。

Leeting Yan 2026-08-14 7 分钟阅读 3169 字

Operator 是 Kubernetes 的核心扩展机制，它将运维人员的领域知识编码为软件——自动完成部署、升级、备份、故障恢复等操作。理解 CRD 和 Controller 的开发模式，是扩展 Kubernetes 平台能力的关键。

1. 为什么需要 Operator

Stateful App 管理难点

有状态应用（数据库、消息队列、缓存）需要复杂的运维操作：

操作	无状态应用	有状态应用
部署	Deployment，一键完成	需要初始化集群、配置副本、设定角色
升级	替换镜像	需要滚动升级、保持数据一致性
扩缩容	改 replicaCount	需要数据重平衡、重新分片
备份	不重要	必须定期备份，且需一致性快照
故障恢复	自动重启	需要选举新主节点、数据同步

Operator 将这些运维逻辑自动化，通过 K8s API 管理复杂应用的生命周期。

Operator 案例

Operator	管理	能力
Prometheus Operator	Prometheus 监控	自动发现 ServiceMonitor、管理规则
etcd Operator	etcd 集群	成员管理、备份恢复、TLS 轮换
Strimzi	Kafka 集群	Topic/User/ACL 管理、滚动升级
Zalando Postgres Operator	PostgreSQL	主从复制、备份、故障切换
MongoDB Community Operator	MongoDB	副本集、分片集群管理
Cert-manager	TLS 证书	自动申请、续期、注入 Secret

2. CRD：自定义资源定义

定义一个新资源

apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
metadata:
  name: databases.myapp.example.com
spec:
  group: myapp.example.com
  versions:
    - name: v1
      served: true          # 可通过 API 访问
      storage: true         # 持久化存储此版本
      schema:
        openAPIV3Schema:
          type: object
          properties:
            spec:
              type: object
              required:
                - version
                - storageSize
              properties:
                version:
                  type: string
                  description: "数据库版本"
                  enum: ["14", "15", "16"]
                storageSize:
                  type: string
                  pattern: "^[0-9]+(Gi|Mi)$"
                  description: "存储大小"
                replicas:
                  type: integer
                  minimum: 1
                  maximum: 7
                  default: 1
                  description: "副本数"
                backup:
                  type: object
                  properties:
                    enabled:
                      type: boolean
                      default: false
                    schedule:
                      type: string
                      pattern: "^\\d+ \\*\\/\\d+ \\* \\* \\*$"
                      description: "Cron 表达式"
            status:
              type: object
              properties:
                phase:
                  type: string
                  enum: ["Pending", "Creating", "Running", "Failed", "Deleting"]
                readyReplicas:
                  type: integer
                message:
                  type: string
      additionalPrinterColumns:
        - name: Version
          type: string
          jsonPath: .spec.version
        - name: Replicas
          type: integer
          jsonPath: .spec.replicas
        - name: Phase
          type: string
          jsonPath: .status.phase
        - name: Age
          type: date
          jsonPath: .metadata.creationTimestamp
  scope: Namespaced           # 或 Cluster
  names:
    plural: databases
    singular: database
    kind: Database
    shortNames:
      - db

创建自定义资源实例

apiVersion: myapp.example.com/v1
kind: Database
metadata:
  name: prod-postgres
  namespace: production
spec:
  version: "16"
  storageSize: "100Gi"
  replicas: 3
  backup:
    enabled: true
    schedule: "0 2 * * *"

查看自定义资源：

kubectl get databases -n production
# NAME           VERSION   REPLICAS   PHASE    AGE
# prod-postgres  16        3          Running  2d

多版本管理

spec:
  versions:
    - name: v1
      served: true
      storage: false   # 不再存储新数据
      deprecated: true
      deprecationWarning: "v1 is deprecated, use v2"
    - name: v2
      served: true
      storage: true    # 新版本存储数据

3. Controller：控制循环

控制器核心模式

所有 K8s 控制器共享同一个模式：Reconciliation Loop（调和循环）。

        ┌─────────────────┐
        │   Observed State │  ← 当前实际状态（从 API Server 获取）
        └────────┬────────┘
                 │
                 ▼
        ┌─────────────────┐
        │  Compare diff    │  ← 对比期望状态（Spec）和实际状态
        └────────┬────────┘
                 │
                 ▼
        ┌─────────────────┐
        │  Reconcile      │  ← 执行操作使实际状态趋近期望状态
        └────────┬────────┘
                 │
                 ▼
        ┌─────────────────┐
        │  Update Status  │  ← 更新资源 Status 反馈当前状态
        └─────────────────┘
                 ↑
                 │（Watch 触发下一个循环）

Controller 基本结构

package main

import (
    "context"
    "fmt"
    "time"

    corev1 "k8s.io/api/core/v1"
    "k8s.io/apimachinery/pkg/api/errors"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/apimachinery/pkg/runtime"
    ctrl "sigs.k8s.io/controller-runtime"
    "sigs.k8s.io/controller-runtime/pkg/client"
    "sigs.k8s.io/controller-runtime/pkg/log"

    myappv1 "github.com/example/myapp/api/v1"
)

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

// +kubebuilder:rbac:groups=myapp.example.com,resources=databases,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=myapp.example.com,resources=databases/status,verbs=get;update;patch
// +kubebuilder:rbac:groups="",resources=services;configmaps;secrets,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=apps,resources=statefulsets,verbs=get;list;watch;create;update;patch;delete

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

    // 1. 获取 Database 资源
    db := &myappv1.Database{}
    if err := r.Get(ctx, req.NamespacedName, db); err != nil {
        if errors.IsNotFound(err) {
            return ctrl.Result{}, nil  // 已删除，无需处理
        }
        return ctrl.Result{}, err
    }

    // 2. 检查是否被删除（Finalizer 处理）
    if !db.DeletionTimestamp.IsZero() {
        return r.reconcileDelete(ctx, db)
    }

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

    // 4. 创建/更新 StatefulSet
    if err := r.reconcileStatefulSet(ctx, db); err != nil {
        db.Status.Phase = "Failed"
        db.Status.Message = fmt.Sprintf("StatefulSet error: %v", err)
        r.Status().Update(ctx, db)
        return ctrl.Result{RequeueAfter: 30 * time.Second}, err
    }

    // 5. 创建/更新 Service
    if err := r.reconcileService(ctx, db); err != nil {
        return ctrl.Result{}, err
    }

    // 6. 创建/更新 ConfigMap（配置文件）
    if err := r.reconcileConfigMap(ctx, db); err != nil {
        return ctrl.Result{}, err
    }

    // 7. 更新状态
    db.Status.Phase = "Running"
    db.Status.ReadyReplicas = db.Spec.Replicas
    db.Status.Message = "Database is running"
    if err := r.Status().Update(ctx, db); err != nil {
        return ctrl.Result{}, err
    }

    return ctrl.Result{RequeueAfter: 60 * time.Second}, nil
}

func (r *DatabaseReconciler) reconcileStatefulSet(ctx context.Context, db *myappv1.Database) error {
    // 构造期望的 StatefulSet
    sts := &appsv1.StatefulSet{
        ObjectMeta: metav1.ObjectMeta{
            Name:      db.Name,
            Namespace: db.Namespace,
            Labels:    labelsForDatabase(db),
        },
    }

    // 设置 OwnerReference：数据库删除时自动级联删除 StatefulSet
    if err := ctrl.SetControllerReference(db, sts, r.Scheme); err != nil {
        return err
    }

    // 幂等创建或更新
    _, err := ctrl.CreateOrUpdate(ctx, r.Client, sts, func() error {
        sts.Spec.Replicas = &db.Spec.Replicas
        sts.Spec.Template.Spec.Containers[0].Image = fmt.Sprintf("postgres:%s", db.Spec.Version)
        // ... 更多配置
        return nil
    })

    return err
}

func (r *DatabaseReconciler) reconcileDelete(ctx context.Context, db *myappv1.Database) (ctrl.Result, error) {
    log := log.FromContext(ctx)
    log.Info("正在清理数据库资源", "database", db.Name)

    // 执行清理操作（如删除外部存储、释放许可证）
    // ...

    // 移除 Finalizer，允许 K8s 删除资源
    controllerutil.RemoveFinalizer(db, "database.myapp.example.com/finalizer")
    if err := r.Update(ctx, db); err != nil {
        return ctrl.Result{}, err
    }

    return ctrl.Result{}, nil
}

func (r *DatabaseReconciler) SetupWithManager(mgr ctrl.Manager) error {
    return ctrl.NewControllerManagedBy(mgr).
        For(&myappv1.Database{}).           // 监视 Database 资源
        Owns(&appsv1.StatefulSet{}).        // 监视附属的 StatefulSet
        Owns(&corev1.Service{}).             // 监视附属的 Service
        Complete(r)
}

关键机制

机制	作用
Requeue	处理失败或需要轮询时，延迟后重新进入队列
OwnerReference	建立父子关系，父资源删除时自动清理子资源
Finalizer	阻止资源删除，直到清理逻辑执行完成
Predicate	过滤事件，减少不必要的 Reconcile 触发

4. Operator SDK 工具链

kubebuilder

CNCF 推荐的 Operator 开发框架，提供代码生成和脚手架：

# 安装 kubebuilder
curl -L -o kubebuilder "https://go.kubebuilder.io/dl/latest/$(go env GOOS)/$(go env GOARCH)"
chmod +x kubebuilder && sudo mv kubebuilder /usr/local/bin/

# 初始化项目
kubebuilder init --domain example.com --repo github.com/example/myoperator

# 创建 API + Controller
kubebuilder create api --group myapp --version v1 --kind Database

# 生成 manifest（CRD、RBAC）
make manifests

# 本地运行（连接当前 kubectl 上下文集群）
make run

# 构建镜像并部署到集群
make docker-build docker-push IMG=registry.example.com/myoperator:v1
make deploy IMG=registry.example.com/myoperator:v1

Operator SDK

Red Hat 提供的更全面的工具链，支持 Ansible、Helm、Go 三种方式：

# 安装 Operator SDK
brew install operator-sdk

# 初始化 Helm-based Operator（最快）
operator-sdk init --plugins=helm --domain=example.com
operator-sdk create api --group=myapp --version=v1 --kind=Database --helm-chart=./mychart

# 初始化 Go-based Operator（最灵活）
operator-sdk init --domain=example.com --repo=github.com/example/myoperator
operator-sdk create api --group=myapp --version=v1 --kind=Database

开发方式	适用场景	复杂度
Helm Operator	已有 Helm Chart，仅需 Operator 包装	低
Ansible Operator	运维团队熟悉 Ansible，需要自定义逻辑	中
Go Operator	需要完全自定义控制逻辑	高

5. 实战：开发一个简单 Operator

需求

开发一个 Operator 管理 PostgreSQL 数据库，支持：

创建 PostgreSQL StatefulSet
根据 replicas 配置主从复制
自动备份（CronJob）
监听自定义资源变更并同步

步骤 1：生成骨架

mkdir mypostgres-operator && cd mypostgres-operator
kubebuilder init --domain db.example.com --repo github.com/example/mypostgres-operator
kubebuilder create api --group db --version v1 --kind PostgresCluster

步骤 2：定义 API

// api/v1/postgrescluster_types.go
package v1

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

// PostgresClusterSpec 定义期望状态
type PostgresClusterSpec struct {
    Version      string            `json:"version"`
    Replicas     int32             `json:"replicas,omitempty"`
    StorageSize  string            `json:"storageSize"`
    Resources    corev1.ResourceRequirements `json:"resources,omitempty"`
    Backup       BackupSpec        `json:"backup,omitempty"`
}

type BackupSpec struct {
    Enabled  bool   `json:"enabled,omitempty"`
    Schedule string `json:"schedule,omitempty"`
    RetentionDays int `json:"retentionDays,omitempty"`
}

// PostgresClusterStatus 定义观测状态
type PostgresClusterStatus struct {
    Phase         string `json:"phase,omitempty"`
    ReadyReplicas int32  `json:"readyReplicas,omitempty"`
    LeaderPod     string `json:"leaderPod,omitempty"`
}

// +kubebuilder:object:root=true
// +kubebuilder:subresource:status
// +kubebuilder:printcolumn:name="Version",type=string,JSONPath=`.spec.version`
// +kubebuilder:printcolumn:name="Replicas",type=integer,JSONPath=`.spec.replicas`
// +kubebuilder:printcolumn:name="Phase",type=string,JSONPath=`.status.phase`

type PostgresCluster struct {
    metav1.TypeMeta   `json:",inline"`
    metav1.ObjectMeta `json:"metadata,omitempty"`
    Spec   PostgresClusterSpec   `json:"spec,omitempty"`
    Status PostgresClusterStatus `json:"status,omitempty"`
}

步骤 3：实现 Reconcile

// internal/controller/postgrescluster_controller.go
func (r *PostgresClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
    cluster := &dbv1.PostgresCluster{}
    if err := r.Get(ctx, req.NamespacedName, cluster); err != nil {
        return ctrl.Result{}, client.IgnoreNotFound(err)
    }

    // 创建 StatefulSet
    sts := r.buildStatefulSet(cluster)
    if err := r.apply(ctx, sts); err != nil {
        return ctrl.Result{}, err
    }

    // 创建 Headless Service（用于 StatefulSet 网络标识）
    svc := r.buildHeadlessService(cluster)
    if err := r.apply(ctx, svc); err != nil {
        return ctrl.Result{}, err
    }

    // 创建备份 CronJob
    if cluster.Spec.Backup.Enabled {
        cronJob := r.buildBackupCronJob(cluster)
        if err := r.apply(ctx, cronJob); err != nil {
            return ctrl.Result{}, err
        }
    }

    // 更新状态
    cluster.Status.Phase = "Running"
    cluster.Status.ReadyReplicas = cluster.Spec.Replicas
    r.Status().Update(ctx, cluster)

    return ctrl.Result{RequeueAfter: 5 * time.Minute}, nil
}

func (r *PostgresClusterReconciler) buildStatefulSet(cluster *dbv1.PostgresCluster) *appsv1.StatefulSet {
    replicas := cluster.Spec.Replicas
    if replicas == 0 {
        replicas = 1
    }

    return &appsv1.StatefulSet{
        ObjectMeta: metav1.ObjectMeta{
            Name:      cluster.Name,
            Namespace: cluster.Namespace,
        },
        Spec: appsv1.StatefulSetSpec{
            ServiceName: cluster.Name + "-headless",
            Replicas:    &replicas,
            Selector: &metav1.LabelSelector{
                MatchLabels: map[string]string{"app": cluster.Name},
            },
            Template: corev1.PodTemplateSpec{
                ObjectMeta: metav1.ObjectMeta{
                    Labels: map[string]string{"app": cluster.Name},
                },
                Spec: corev1.PodSpec{
                    Containers: []corev1.Container{
                        {
                            Name:  "postgres",
                            Image: fmt.Sprintf("postgres:%s", cluster.Spec.Version),
                            Ports: []corev1.ContainerPort{
                                {ContainerPort: 5432, Name: "postgres"},
                            },
                            VolumeMounts: []corev1.VolumeMount{
                                {Name: "data", MountPath: "/var/lib/postgresql/data"},
                            },
                        },
                    },
                },
            },
            VolumeClaimTemplates: []corev1.PersistentVolumeClaim{
                {
                    ObjectMeta: metav1.ObjectMeta{Name: "data"},
                    Spec: corev1.PersistentVolumeClaimSpec{
                        AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
                        Resources:   corev1.VolumeResourceRequirements{Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse(cluster.Spec.StorageSize)}},
                    },
                },
            },
        },
    }
}

步骤 4：部署 Operator

# 安装 CRD
make install

# 部署 Operator 到集群
make deploy IMG=registry.example.com/mypostgres-operator:v1

# 创建自定义资源
kubectl apply -f - <<EOF
apiVersion: db.example.com/v1
kind: PostgresCluster
metadata:
  name: my-postgres
spec:
  version: "16"
  replicas: 3
  storageSize: 50Gi
  resources:
    requests:
      cpu: 500m
      memory: 1Gi
    limits:
      cpu: 2000m
      memory: 4Gi
  backup:
    enabled: true
    schedule: "0 3 * * *"
    retentionDays: 7
EOF

6. Operator 成熟度模型

Operator Framework 定义了 5 级成熟度：

等级	名称	能力
Level 1	Basic Install	自动安装和配置应用
Level 2	Seamless Upgrades	支持补丁/小版本/大版本升级
Level 3	Full Lifecycle	支持备份、恢复、故障转移
Level 4	Deep Insights	提供指标、日志、告警、状态深入洞察
Level 5	Auto Pilot	自动伸缩、自动调优、异常自愈

大多数开源 Operator 达到 Level 2-3，生产级需要达到 Level 3+。

7. 生产级 Operator 框架

Operator Lifecycle Manager (OLM)

CNCF 项目，用于 Operator 的生命周期管理：

# 安装 OLM
curl -sL https://github.com/operator-framework/operator-lifecycle-manager/releases/download/v0.28.0/install.sh | bash -s v0.28.0

# 通过 OLM 安装 Operator
kubectl create -f https://operatorhub.io/install/postgres-operator.yaml

OLM 功能：

依赖解析：自动安装 Operator 的依赖
版本管理：支持 Operator 的升级和回滚
权限管理：通过 CSV (ClusterServiceVersion) 声明 RBAC
多租户：同一集群运行不同版本的同一 Operator

OperatorHub

OperatorHub.io 是社区 Operator 的集中市场，提供经过审核的 Operator 包。

8. 最佳实践

设计原则

声明式 API：用户声明期望状态，Operator 负责达成
幂等性：多次 Reconcile 结果一致，不重复创建资源
OwnerReference 级联：利用 K8s 垃圾回收，删除父资源自动清理子资源
Status 反馈：及时向用户反馈当前状态和错误信息
优雅降级：部分组件失败时，尽可能保持核心功能可用

资源限制

# Operator 自身也需要资源限制
resources:
  limits:
    cpu: 500m
    memory: 256Mi
  requests:
    cpu: 50m
    memory: 64Mi

Leader Election

// 多副本 Operator 需要 Leader Election 避免冲突
mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
    LeaderElection:   true,
    LeaderElectionID: "myoperator.leader.example.com",
})

测试

# Envtest 进行单元测试（无需真实集群）
make test

# 使用 kind 进行集成测试
kind create cluster
make deploy
make test-e2e

总结

主题	核心要点
CRD	通过 OpenAPI schema 定义自定义资源，支持多版本管理
Controller	调和循环：Observe → Diff → Act → Update Status
SDK	kubebuilder/Operator SDK 提供脚手架和代码生成
开发流程	定义 API → 实现 Reconcile → 生成 Manifest → 部署测试
成熟模型	5 级成熟度，生产级需达到 Level 3+
OLM	Operator 生命周期管理，依赖解析、版本控制

Operator 将 Kubernetes 从"容器编排平台"提升为"应用交付平台"，是云原生时代自动化运维的核心基础设施。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章返回首页