跳至主要內容

Go 云原生开发实战——Kubernetes Operator 从入门到实现完整指南

老张2026/4/30大约 11 分钟

Go 云原生开发实战——Kubernetes Operator 从入门到实现完整指南

适读人群：有 Go 基础、想进入云原生领域的后端工程师 | 阅读时长：约18分钟 | 核心价值：从零理解 Operator 机制，写出能跑在生产的 CRD 控制器

那次被 Operator 折腾到凌晨两点的经历

去年九月，我们团队接手了一个平台化改造项目，核心需求是把公司内部的十几套 MySQL 集群管理流程自动化——创建实例、备份策略、故障转移，全都要通过 K8s 声明式 API 来驱动。

领导把这个活儿甩给我，说："不是有 Operator 这东西吗，去搞一下。"

我当时心想，不就是个控制器嘛，Java 写 Spring 事件驱动也没难到哪去。于是我花了半天时间看完官方文档，信心满满地用 operator-sdk 生成了脚手架，改了改 Reconcile 方法，push 上去……然后发现控制器疯狂 crash，日志全是 panic，CRD 状态永远停在 Pending。

那晚排查到凌晨两点，最后定位到三个问题：一是我把 Reconcile 的 requeue 逻辑写错了，导致无限循环；二是没理解 finalizer 机制，资源删不掉；三是 Status 子资源的更新方式和 Spec 更新方式完全不同，我混用了 Update 和 StatusClient，每次更新都触发新的 reconcile，死循环就这么来的。

那晚之后，我把 controller-runtime 的源码啃了一遍，把 kubebuilder book 重新读了两遍，慢慢把这套体系理清楚了。今天这篇文章，就是我想早点拿到的那份指南。

Operator 到底是什么，为什么需要它

很多人对 Operator 的理解停留在"自定义控制器"这个层面，但这只说了一半。

Operator 的完整定义是：用代码封装运维知识的 K8s 扩展机制。

举个 Java 程序员能理解的类比：K8s 原生的 Deployment、StatefulSet 就像 Spring 框架提供的现成组件；而 Operator 就是你在这个框架上写的业务 Bean——你告诉 K8s "我有一种叫 MySQLCluster 的资源，当它的 spec 变化时，我需要做这些事情"。

Operator 由两部分组成：

  • CRD（Custom Resource Definition）：定义你的自定义资源长什么样
  • Controller：监听这种资源的变化，驱动集群状态向期望状态收敛

这就是 Kubernetes 的核心哲学——控制循环（control loop）：观察当前状态，与期望状态对比，执行调和（Reconcile）。

环境搭建与工具链选择

开发 Operator 有两个主流工具链：

工具特点适合场景
kubebuilderCNCF 官方项目，生成代码规范生产级 Operator
operator-sdkRed Hat 维护，集成了 Helm/Ansible快速原型

我推荐用 kubebuilder，代码更清晰，社区文档更完整。

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

# 初始化项目
mkdir mysql-operator && cd mysql-operator
kubebuilder init --domain example.com --repo github.com/yourname/mysql-operator

# 创建 API（CRD + Controller）
kubebuilder create api --group db --version v1alpha1 --kind MySQLCluster

生成的项目结构：

mysql-operator/
├── api/v1alpha1/
│   ├── mysqlcluster_types.go    # CRD 类型定义
│   └── zz_generated.deepcopy.go # 自动生成
├── controllers/
│   └── mysqlcluster_controller.go # 核心控制器
├── config/
│   ├── crd/                     # CRD YAML
│   └── rbac/                    # 权限配置
└── main.go

定义 CRD 类型

先把资源的数据结构定义清楚，这是一切的基础。

// api/v1alpha1/mysqlcluster_types.go

package v1alpha1

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

// MySQLClusterSpec 定义期望状态
type MySQLClusterSpec struct {
    // +kubebuilder:validation:Minimum=1
    // +kubebuilder:validation:Maximum=9
    Replicas int32 `json:"replicas"`

    // MySQL 版本
    // +kubebuilder:default="8.0"
    Version string `json:"version,omitempty"`

    // 存储大小，如 "10Gi"
    StorageSize string `json:"storageSize"`

    // 资源限制
    Resources ResourceRequirements `json:"resources,omitempty"`

    // 备份配置
    Backup *BackupConfig `json:"backup,omitempty"`
}

type BackupConfig struct {
    // 是否启用自动备份
    Enabled bool `json:"enabled"`
    // Cron 表达式，如 "0 2 * * *"
    Schedule string `json:"schedule,omitempty"`
    // 保留天数
    RetentionDays int32 `json:"retentionDays,omitempty"`
}

type ResourceRequirements struct {
    CPU    string `json:"cpu,omitempty"`
    Memory string `json:"memory,omitempty"`
}

// MySQLClusterStatus 定义观察状态
type MySQLClusterStatus struct {
    // +kubebuilder:validation:Enum=Pending;Running;Failed;Terminating
    Phase string `json:"phase,omitempty"`

    // 就绪副本数
    ReadyReplicas int32 `json:"readyReplicas,omitempty"`

    // 主节点地址
    PrimaryEndpoint string `json:"primaryEndpoint,omitempty"`

    // 状态条件列表
    Conditions []metav1.Condition `json:"conditions,omitempty"`

    // 最后更新时间
    LastUpdated *metav1.Time `json:"lastUpdated,omitempty"`
}

// +kubebuilder:object:root=true
// +kubebuilder:subresource:status
// +kubebuilder:printcolumn:name="Phase",type=string,JSONPath=`.status.phase`
// +kubebuilder:printcolumn:name="Replicas",type=integer,JSONPath=`.spec.replicas`
// +kubebuilder:printcolumn:name="Age",type=date,JSONPath=`.metadata.creationTimestamp`
type MySQLCluster struct {
    metav1.TypeMeta   `json:",inline"`
    metav1.ObjectMeta `json:"metadata,omitempty"`

    Spec   MySQLClusterSpec   `json:"spec,omitempty"`
    Status MySQLClusterStatus `json:"status,omitempty"`
}

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

注意那些 // +kubebuilder: 注释，这是 kubebuilder 的 marker，会自动生成 CRD YAML 里的 validation 规则和 kubectl get 时的列展示。这个在 Java 里没有对应概念——最接近的是 Spring Validation 的注解，但功能完全不同。

实现核心 Reconcile 逻辑

这是 Operator 的心脏。每当 MySQLCluster 资源发生变化（创建、更新、删除），Reconcile 都会被调用。

// controllers/mysqlcluster_controller.go

package controllers

import (
    "context"
    "fmt"
    "time"

    appsv1 "k8s.io/api/apps/v1"
    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"
    "k8s.io/apimachinery/pkg/types"
    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"

    dbv1alpha1 "github.com/yourname/mysql-operator/api/v1alpha1"
)

const (
    mysqlFinalizer = "db.example.com/mysql-finalizer"
    phaseRunning   = "Running"
    phasePending   = "Pending"
    phaseFailed    = "Failed"
)

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

// +kubebuilder:rbac:groups=db.example.com,resources=mysqlclusters,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=db.example.com,resources=mysqlclusters/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=db.example.com,resources=mysqlclusters/finalizers,verbs=update
// +kubebuilder:rbac:groups=apps,resources=statefulsets,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=core,resources=services,verbs=get;list;watch;create;update;patch;delete

func (r *MySQLClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
    logger := log.FromContext(ctx)
    logger.Info("Reconciling MySQLCluster", "name", req.Name, "namespace", req.Namespace)

    // Step 1: 获取资源对象
    cluster := &dbv1alpha1.MySQLCluster{}
    if err := r.Get(ctx, req.NamespacedName, cluster); err != nil {
        if errors.IsNotFound(err) {
            // 资源已删除，不需要处理
            return ctrl.Result{}, nil
        }
        return ctrl.Result{}, err
    }

    // Step 2: 处理删除逻辑（finalizer）
    if !cluster.DeletionTimestamp.IsZero() {
        return r.handleDeletion(ctx, cluster)
    }

    // Step 3: 添加 finalizer（确保删除时能执行清理逻辑）
    if !controllerutil.ContainsFinalizer(cluster, mysqlFinalizer) {
        controllerutil.AddFinalizer(cluster, mysqlFinalizer)
        if err := r.Update(ctx, cluster); err != nil {
            return ctrl.Result{}, err
        }
        // 更新后立刻 return，等下次 reconcile 继续
        return ctrl.Result{}, nil
    }

    // Step 4: 更新状态为 Pending
    if cluster.Status.Phase == "" {
        if err := r.updatePhase(ctx, cluster, phasePending); err != nil {
            return ctrl.Result{}, err
        }
    }

    // Step 5: 调和 StatefulSet
    if err := r.reconcileStatefulSet(ctx, cluster); err != nil {
        _ = r.updatePhase(ctx, cluster, phaseFailed)
        return ctrl.Result{}, err
    }

    // Step 6: 调和 Service
    if err := r.reconcileService(ctx, cluster); err != nil {
        return ctrl.Result{}, err
    }

    // Step 7: 检查实际运行状态，更新 Status
    if err := r.syncStatus(ctx, cluster); err != nil {
        return ctrl.Result{}, err
    }

    // 每隔30秒重新检查一次状态
    return ctrl.Result{RequeueAfter: 30 * time.Second}, nil
}

func (r *MySQLClusterReconciler) reconcileStatefulSet(ctx context.Context, cluster *dbv1alpha1.MySQLCluster) error {
    desired := r.buildStatefulSet(cluster)

    // 尝试获取已有的 StatefulSet
    existing := &appsv1.StatefulSet{}
    err := r.Get(ctx, types.NamespacedName{
        Name:      desired.Name,
        Namespace: desired.Namespace,
    }, existing)

    if errors.IsNotFound(err) {
        // 不存在则创建
        log.FromContext(ctx).Info("Creating StatefulSet", "name", desired.Name)
        return r.Create(ctx, desired)
    }
    if err != nil {
        return err
    }

    // 已存在则更新（只更新 spec 里我们关心的字段）
    existing.Spec.Replicas = desired.Spec.Replicas
    existing.Spec.Template = desired.Spec.Template
    return r.Update(ctx, existing)
}

func (r *MySQLClusterReconciler) buildStatefulSet(cluster *dbv1alpha1.MySQLCluster) *appsv1.StatefulSet {
    labels := map[string]string{
        "app":        "mysql",
        "cluster":    cluster.Name,
        "managed-by": "mysql-operator",
    }

    sts := &appsv1.StatefulSet{
        ObjectMeta: metav1.ObjectMeta{
            Name:      cluster.Name + "-mysql",
            Namespace: cluster.Namespace,
            Labels:    labels,
        },
        Spec: appsv1.StatefulSetSpec{
            Replicas:    &cluster.Spec.Replicas,
            ServiceName: cluster.Name + "-headless",
            Selector: &metav1.LabelSelector{
                MatchLabels: labels,
            },
            Template: corev1.PodTemplateSpec{
                ObjectMeta: metav1.ObjectMeta{Labels: labels},
                Spec: corev1.PodSpec{
                    Containers: []corev1.Container{
                        {
                            Name:  "mysql",
                            Image: fmt.Sprintf("mysql:%s", cluster.Spec.Version),
                            Ports: []corev1.ContainerPort{
                                {ContainerPort: 3306, Name: "mysql"},
                            },
                            Env: []corev1.EnvVar{
                                {
                                    Name:  "MYSQL_ROOT_PASSWORD",
                                    Value: "changeme", // 生产应该用 Secret
                                },
                            },
                        },
                    },
                },
            },
        },
    }

    // 设置 OwnerReference，确保 MySQLCluster 删除时 StatefulSet 也跟着删除
    _ = controllerutil.SetControllerReference(cluster, sts, r.Scheme)
    return sts
}

func (r *MySQLClusterReconciler) reconcileService(ctx context.Context, cluster *dbv1alpha1.MySQLCluster) error {
    svc := &corev1.Service{
        ObjectMeta: metav1.ObjectMeta{
            Name:      cluster.Name + "-headless",
            Namespace: cluster.Namespace,
        },
        Spec: corev1.ServiceSpec{
            ClusterIP: "None", // Headless Service
            Selector: map[string]string{
                "app":     "mysql",
                "cluster": cluster.Name,
            },
            Ports: []corev1.ServicePort{
                {Port: 3306, Name: "mysql"},
            },
        },
    }
    _ = controllerutil.SetControllerReference(cluster, svc, r.Scheme)

    existing := &corev1.Service{}
    err := r.Get(ctx, types.NamespacedName{Name: svc.Name, Namespace: svc.Namespace}, existing)
    if errors.IsNotFound(err) {
        return r.Create(ctx, svc)
    }
    return err
}

func (r *MySQLClusterReconciler) syncStatus(ctx context.Context, cluster *dbv1alpha1.MySQLCluster) error {
    sts := &appsv1.StatefulSet{}
    err := r.Get(ctx, types.NamespacedName{
        Name:      cluster.Name + "-mysql",
        Namespace: cluster.Namespace,
    }, sts)
    if err != nil {
        return err
    }

    patch := client.MergeFrom(cluster.DeepCopy())
    cluster.Status.ReadyReplicas = sts.Status.ReadyReplicas
    now := metav1.Now()
    cluster.Status.LastUpdated = &now

    if sts.Status.ReadyReplicas == cluster.Spec.Replicas {
        cluster.Status.Phase = phaseRunning
        cluster.Status.PrimaryEndpoint = fmt.Sprintf(
            "%s-mysql-0.%s-headless.%s.svc.cluster.local:3306",
            cluster.Name, cluster.Name, cluster.Namespace,
        )
    } else {
        cluster.Status.Phase = phasePending
    }

    // 注意：Status 更新必须用 Status().Patch，不能用普通 Update
    return r.Status().Patch(ctx, cluster, patch)
}

func (r *MySQLClusterReconciler) handleDeletion(ctx context.Context, cluster *dbv1alpha1.MySQLCluster) (ctrl.Result, error) {
    if controllerutil.ContainsFinalizer(cluster, mysqlFinalizer) {
        // 执行清理逻辑（比如备份最后一次数据）
        log.FromContext(ctx).Info("Running cleanup for MySQLCluster", "name", cluster.Name)

        // TODO: 执行实际清理操作

        // 清理完成，移除 finalizer
        controllerutil.RemoveFinalizer(cluster, mysqlFinalizer)
        if err := r.Update(ctx, cluster); err != nil {
            return ctrl.Result{}, err
        }
    }
    return ctrl.Result{}, nil
}

func (r *MySQLClusterReconciler) updatePhase(ctx context.Context, cluster *dbv1alpha1.MySQLCluster, phase string) error {
    patch := client.MergeFrom(cluster.DeepCopy())
    cluster.Status.Phase = phase
    return r.Status().Patch(ctx, cluster, patch)
}

func (r *MySQLClusterReconciler) SetupWithManager(mgr ctrl.Manager) error {
    return ctrl.NewControllerManagedBy(mgr).
        For(&dbv1alpha1.MySQLCluster{}).
        // 监听我们管理的 StatefulSet 变化，自动触发 reconcile
        Owns(&appsv1.StatefulSet{}).
        Owns(&corev1.Service{}).
        Complete(r)
}

三个让我掉坑的细节

坑一：Reconcile 无限循环

现象：控制器 CPU 飙到 100%，日志里同一个资源被无限 reconcile。

原因：我在 Reconcile 里直接 Update(cluster) 来更新 Status，但 Update Spec/metadata 会触发新的 reconcile 事件，新的 reconcile 又触发 Update，形成死循环。

解法：Status 更新必须走 r.Status().Patch() 或 r.Status().Update()，走 Status 子资源接口，不会触发主资源的 reconcile。这是 K8s Operator 开发的基本规范，但文档写得不明显。

// 错误方式 - 会触发无限 reconcile
cluster.Status.Phase = "Running"
r.Update(ctx, cluster)  // 别这样！

// 正确方式 - 只更新 Status 子资源
patch := client.MergeFrom(cluster.DeepCopy())
cluster.Status.Phase = "Running"
r.Status().Patch(ctx, cluster, patch)  // 正确

坑二：资源删不掉，永远 Terminating

现象：kubectl delete mysqlcluster xxx 执行后，资源一直卡在 Terminating 状态。

原因：我添加了 finalizer 但没有在资源被标记删除时（DeletionTimestamp != nil）去移除它。K8s 的规则是：只要 finalizer 列表不为空，资源就不会被真正删除。

解法：在 Reconcile 开头先判断 DeletionTimestamp，如果不为零，说明资源正在被删除，执行清理逻辑后调用 controllerutil.RemoveFinalizer 移除 finalizer，资源才会真正消失。

坑三：并发 Reconcile 导致数据竞争

现象：偶发性的 the object has been modified 错误，重试后才成功。

原因：K8s 的资源都有 ResourceVersion，用于乐观锁。如果在我读取资源和更新资源之间，有其他人改了这个资源（比如 kubectl edit），我的更新就会被拒绝。

解法：这其实是正常的——Reconcile 本来就应该是幂等的，遇到冲突直接返回 error，controller-runtime 会自动帮你 requeue 重试。关键是不要在 Reconcile 里持有全局状态。

// Reconcile 应该是无状态的、幂等的
// 每次被调用都应该能从头执行，结果一致
func (r *MySQLClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
    // 永远先从 API Server 重新 Get 最新状态
    cluster := &dbv1alpha1.MySQLCluster{}
    if err := r.Get(ctx, req.NamespacedName, cluster); err != nil {
        return ctrl.Result{}, client.IgnoreNotFound(err)
    }
    // ...
}

Go vs Java：理念上的差异

写 Operator 让我深刻感受到 Go 和 Java 在工程哲学上的差异。

在 Java 里，Spring 框架给你提供了大量的生命周期回调、事件总线、AOP 切面，你在这个重度框架上"填空"。

在 Go 的 Operator 世界里，controller-runtime 给你的是更薄的抽象——你实现一个 Reconcile(ctx, req) (Result, error) 函数，框架负责调用它。所有的状态机逻辑、错误处理、重试策略，都在这一个函数里显式写清楚。

这让 Go 的控制器代码更容易测试（一个纯函数的单元测试，vs Java 里 mock 满天飞），但也要求你对 K8s 的资源模型理解得更深。

Java 开发者转写 Operator 最容易犯的错误，是把 Reconcile 写成"事件处理器"思维——"当资源创建时做A，当资源更新时做B"。正确的思维是"无论什么触发，我都重新评估当前状态和期望状态的差异，然后把差异消除"。

本地调试技巧

不要每次改动都 build image、push、重新部署，这样太慢了。

# 方法一：在本地直接运行控制器，连接远程集群
export KUBECONFIG=~/.kube/config
make install    # 只安装 CRD 到集群
go run ./cmd/main.go  # 本地运行控制器

# 方法二：使用 envtest（纯内存的假 K8s，适合 CI）
# 在 controllers/suite_test.go 里已经配置好了
make test

本地运行时，控制器连接的是你 kubeconfig 里指向的那个集群，CRD 安装到集群里，但控制器跑在你本地机器上。改代码后 Ctrl+C 重启，比走整个 CI/CD 流程快得多。

部署到集群

# 生成 CRD YAML
make manifests

# 安装 CRD
make install

# 构建并推送镜像
make docker-build docker-push IMG=yourregistry/mysql-operator:v0.1.0

# 部署控制器
make deploy IMG=yourregistry/mysql-operator:v0.1.0

# 创建一个 MySQLCluster 资源测试
kubectl apply -f - <<EOF
apiVersion: db.example.com/v1alpha1
kind: MySQLCluster
metadata:
  name: test-cluster
  namespace: default
spec:
  replicas: 3
  version: "8.0"
  storageSize: "10Gi"
EOF

# 查看状态
kubectl get mysqlcluster test-cluster
kubectl describe mysqlcluster test-cluster

小结

Operator 开发的核心要点：

  1. Reconcile 必须幂等：每次调用都从头对比期望状态和实际状态，不要依赖上一次的结果
  2. Status 更新走子资源接口：避免死循环
  3. Finalizer 是双刃剑：加了必须有对应的移除逻辑，否则资源删不掉
  4. OwnerReference 是自动 GC 的关键：给子资源设置 OwnerReference，父资源删除后子资源自动清理
  5. 本地调试 + envtest：不要每次改动都走 CI/CD，太慢

云原生这个方向在国内的需求越来越大，Go + K8s Operator 是一个很有竞争力的技术组合。从 Java 转过来有成本，但理解了控制循环这个核心思想之后，会发现 Operator 的设计非常优雅。

编辑此页
最近更新:
贡献者: xiaozhangge