详解 Kubernetes 自定义资源定义 CRD 与 Operator 模式：如何用代码扩展集群能力

Kubernetes 内置了 Deployment、Service、ConfigMap 等丰富的资源类型，但在实际生产中，我们经常需要管理一些 K8s 原生不认识的东西——比如一个 Redis 集群、一个数据库实例、或者一套自定义的调度规则。这时候 CRD（Custom Resource Definition）和 Operator 模式就成了你扩展 K8s 能力的核心武器。本文将从零开始，带你理解 CRD 的工作原理，并通过实战代码构建一个简单的 Operator。

Kubernetes 容器编排

Table of Contents

什么是 CRD：让 K8s 认识你的自定义资源

CRD 的本质很简单：你告诉 K8s “我想管理一种新的资源类型”，K8s 就会自动为它创建 REST API 端点和存储。定义一个 CRD 只需要一个 YAML 文件：

apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
metadata:
  name: mydatabases.example.com
spec:
  group: example.com
  versions:
    - name: v1
      served: true
      storage: true
      schema:
        openAPIV3Schema:
          type: object
          properties:
            spec:
              type: object
              properties:
                engine:
                  type: string
                  enum: [mysql, postgres]
                replicas:
                  type: integer
                  minimum: 1
                storageSize:
                  type: string
  scope: Namespaced
  names:
    plural: mydatabases
    singular: mydatabase
    kind: MyDatabase
    shortNames:
      - mdb

应用这个 YAML 后，你就可以像操作 Deployment 一样创建自定义资源了：

apiVersion: example.com/v1
kind: MyDatabase
metadata:
  name: prod-mysql
spec:
  engine: mysql
  replicas: 3
  storageSize: 50Gi

此刻 K8s 会帮你存储这个对象，你也可以通过 kubectl get mydatabases 查看它。但它还不会自动创建真正的数据库 Pod——这就是 Operator 要做的事。

服务器基础设施

Operator 模式：用 Reconciliation Loop 实现自动化管理

Operator 的核心思想是调谐循环（Reconciliation Loop）：不断对比”期望状态”和”实际状态”，然后执行操作让两者趋于一致。当你把 MyDatabase 的 replicas 从 1 改成 3 时，Operator 会检测到差异，自动创建 2 个新 Pod。

Python 生态中有成熟的库可以快速编写 Operator，下面用 kopf 来实现一个简单的 MyDatabase Operator：

import kopf
import kubernetes
from kubernetes import client, config

# 集群内运行时加载配置
try:
    config.load_incluster_config()
except config.ConfigException:
    config.load_kube_config()

apps_v1 = client.AppsV1Api()

def build_statefulset(name, namespace, spec):
    """根据 MyDatabase spec 构建 StatefulSet"""
    return client.V1StatefulSet(
        metadata=client.V1ObjectMeta(
            name=f"{name}-db",
            namespace=namespace,
            labels={"app": name, "managed-by": "mydb-operator"}
        ),
        spec=client.V1StatefulSetSpec(
            replicas=spec["replicas"],
            service_name=f"{name}-svc",
            selector=client.V1LabelSelector(
                match_labels={"app": name}
            ),
            template=client.V1PodTemplateSpec(
                metadata=client.V1ObjectMeta(
                    labels={"app": name}
                ),
                spec=client.V1PodSpec(
                    containers=[client.V1Container(
                        name="db",
                        image=f"{spec['engine']}:latest",
                        ports=[client.V1ContainerPort(container_port=3306 if spec['engine'] == 'mysql' else 5432)],
                        resources=client.V1ResourceRequirements(
                            requests={"cpu": "250m", "memory": "512Mi"},
                            limits={"cpu": "1", "memory": "1Gi"}
                        ),
                        volume_mounts=[client.V1VolumeMount(
                            name="data", mount_path="/var/lib/mysql"
                        )]
                    )],
                    volumes=[client.V1Volume(
                        name="data",
                        persistent_volume_claim=client.V1PersistentVolumeClaimVolumeSource(
                            claim_name=f"{name}-pvc"
                        )
                    )]
                )
            )
        )
    )


@kopf.on.create('example.com', 'v1', 'mydatabases')
def on_create(spec, name, namespace, **kwargs):
    """资源创建时的处理逻辑"""
    sts = build_statefulset(name, namespace, spec)
    try:
        apps_v1.create_namespaced_stateful_set(namespace=namespace, body=sts)
        return {"message": f"StatefulSet {name}-db created"}
    except client.exceptions.ApiException as e:
        if e.status == 409:
            return {"message": "StatefulSet already exists"}
        raise


@kopf.on.update('example.com', 'v1', 'mydatabases')
def on_update(spec, name, namespace, **kwargs):
    """资源更新时的处理逻辑"""
    sts_name = f"{name}-db"
    sts = build_statefulset(name, namespace, spec)
    try:
        apps_v1.patch_namespaced_stateful_set(
            name=sts_name, namespace=namespace, body=sts
        )
        return {"message": f"StatefulSet {sts_name} updated"}
    except client.exceptions.ApiException as e:
        raise kopf.TemporaryError(f"Update failed: {e.reason}", delay=10)


@kopf.on.delete('example.com', 'v1', 'mydatabases')
def on_delete(name, namespace, **kwargs):
    """资源删除时的清理逻辑"""
    try:
        apps_v1.delete_namespaced_stateful_set(
            name=f"{name}-db", namespace=namespace
        )
    except client.exceptions.ApiException as e:
        if e.status != 404:
            raise
    return {"message": "cleanup done"}

部署 Operator 到集群

将 Operator 打包成容器镜像并部署到集群中，它就会开始监听 MyDatabase 资源的变更：

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir kopf kubernetes
COPY operator.py .
CMD ["kopf", "run", "operator.py", "--verbose"]

对应的 Deployment 清单，注意需要给 Operator 绑定足够的 RBAC 权限：

apiVersion: apps/v1
kind: Deployment
metadata:
  name: mydb-operator
  namespace: default
spec:
  replicas: 1
  selector:
    matchLabels:
      app: mydb-operator
  template:
    metadata:
      labels:
        app: mydb-operator
    spec:
      serviceAccountName: mydb-operator
      containers:
        - name: operator
          image: myregistry/mydb-operator:v1
          imagePullPolicy: Always

代码编程

Operator 开发中的常见坑

在实际开发 Operator 时，有几个问题值得特别注意：

1. 幂等性设计：Reconcile 函数可能被多次调用，你的逻辑必须保证重复执行不会产生副作用。比如创建 StatefulSet 前先检查是否已存在。

2. 状态管理：善用 CRD 的 status 子资源来记录当前状态，方便用户通过 kubectl get 查看：

@kopf.on.create('example.com', 'v1', 'mydatabases')
def on_create(spec, name, namespace, patch, **kwargs):
    # ... 创建资源 ...
    patch.status = {
        "phase": "Running",
        "readyReplicas": 0,
        "message": "Database provisioning started"
    }

3. Finalizer 与清理：当你的 Operator 需要清理外部资源（如云厂商的存储卷）时，要使用 Finalizer 确保删除逻辑一定被执行。

何时该用 CRD + Operator

并不是所有场景都需要自建 Operator。以下几种情况适合引入 CRD 和 Operator 模式：

需要像管理原生资源一样管理第三方服务（数据库、消息队列、缓存）
需要封装复杂的运维逻辑（备份、恢复、版本升级）
团队需要标准化某种部署模式，降低使用门槛
面向多租户场景，需要自助式的资源生命周期管理

成熟的开源 Operator 也非常值得参考，比如 OperatorHub 上有大量生产级实现。开发自己的 Operator 之前，先看看社区是否已经有现成的方案。

总结

CRD + Operator 是 Kubernetes 生态中最强大的扩展机制。CRD 让你定义新的资源类型，Operator 让你用代码自动化管理这些资源的生命周期。通过 Reconciliation Loop 的设计模式，你可以把复杂的运维知识封装成可复用的控制器，真正实现”运维即代码”。掌握这套模式，你就拥有了把 K8s 打造成任意平台的底层能力。