Jul 14 2022

云原生时代Kubernetes Operator开发实战：从零开始构建自定义控制器管理复杂应用

作者： 绝缘体..   发布： 2022-07-14 10:19   分类： 未分类   阅读：   抢沙发  
 
更多

云原生时代Kubernetes Operator开发实战：从零开始构建自定义控制器管理复杂应用

标签：Kubernetes, Operator, 云原生, 控制器, Go语言
简介：全面解析Kubernetes Operator模式的核心概念和开发实践，通过实际案例演示如何使用Go语言和Kubebuilder框架开发自定义控制器，实现复杂应用的自动化部署、升级和运维管理，助力企业拥抱云原生技术。

一、引言：为什么需要 Kubernetes Operator？

在云原生生态中，Kubernetes 已成为容器编排的事实标准。然而，随着微服务架构和复杂应用（如数据库集群、消息队列、AI训练平台等）的普及，仅仅依赖 Deployment、Service、ConfigMap 等原生资源已难以满足对状态管理、配置一致性、故障恢复等高级需求。

传统的运维方式——手动干预或脚本化操作——无法应对大规模、高可用、自动化的现代应用部署场景。于是，Kubernetes Operator 模式应运而生。

什么是 Operator？

Operator 是一种基于 Kubernetes 的扩展机制，它将特定应用程序的“知识”编码为代码，通过自定义资源（Custom Resource, CR）和自定义控制器（Controller）来实现对应用生命周期的完整自动化管理。

  • 核心思想：将“运维专家的经验”转化为可复用、可版本控制的代码。
  • 典型场景：
    • 数据库集群（如 PostgreSQL、MySQL 主从复制）
    • 分布式缓存系统（如 Redis Cluster）
    • AI/ML 训练平台（如 Kubeflow）
    • 自定义中间件或业务系统

✅ Operator = 自定义资源（CR） + 控制器（Controller）

二、Operator 核心概念详解

2.1 自定义资源（Custom Resource, CR）

Kubernetes 原生资源（如 Pod、Deployment）不能完全表达复杂应用的状态。为此，我们可以定义自己的 CR 来描述应用实例。

示例：定义一个 MyApp 类型的自定义资源

apiVersion: myapp.example.com/v1alpha1
kind: MyApp
metadata:
  name: myapp-instance
spec:
  replicas: 3
  image: nginx:1.25
  port: 8080
  config:
    logLevel: "info"

这个 MyApp 资源描述了一个名为 myapp-instance 的应用实例，包含副本数、镜像、端口及配置项。

2.2 自定义资源定义（Custom Resource Definition, CRD）

要让 Kubernetes 理解 MyApp 这个资源类型，必须先创建 CRD：

apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
metadata:
  name: myapps.myapp.example.com
spec:
  group: myapp.example.com
  versions:
    - name: v1alpha1
      served: true
      storage: true
      schema:
        openAPIV3Schema:
          type: object
          properties:
            spec:
              type: object
              properties:
                replicas:
                  type: integer
                  minimum: 1
                image:
                  type: string
                port:
                  type: integer
                config:
                  type: object
                  properties:
                    logLevel:
                      type: string
                  required:
                    - logLevel
              required:
                - replicas
                - image
                - port
            status:
              type: object
              properties:
                phase:
                  type: string
                  enum: ["Pending", "Running", "Failed", "Succeeded"]
                message:
                  type: string
                observedGeneration:
                  type: integer
  scope: Namespaced
  names:
    plural: myapps
    singular: myapp
    kind: MyApp
    shortNames:
      - ma

📌 关键点：

  • group: 自定义 API 组名（建议使用反向域名）
  • version: API 版本（v1alpha1 表示实验阶段）
  • scope: Namespaced 或 Cluster（推荐命名空间级别）
  • spec 和 status 字段结构需与控制器逻辑一致

2.3 控制器（Controller）

控制器是 Operator 的“大脑”，负责监听 CR 的变化，并根据期望状态（Spec）与当前状态（Status）之间的差异，执行相应的动作。

其工作流程如下：

[事件触发] → [监听 CR 变化] → [读取 Spec] → [计算差异] → [执行操作] → [更新 Status]

🔁 Reconcile Loop（协调循环）：控制器的核心循环，持续检查资源是否处于预期状态。

三、开发环境搭建与工具选型

为了高效开发 Operator，推荐使用 Kubebuilder 框架，它是 CNCF 官方支持的项目，专为构建 Operator 设计。

3.1 安装依赖

# 安装 kubebuilder
curl -L -o kubebuilder https://github.com/kubernetes-sigs/kubebuilder/releases/download/v3.1.0/kubebuilder_3.1.0_linux_amd64
chmod +x kubebuilder
sudo mv kubebuilder /usr/local/bin/

# 安装 controller-runtime
go mod init myapp-operator
go get sigs.k8s.io/controller-runtime@v0.15.0

3.2 初始化项目结构

kubebuilder init --domain example.com --repo github.com/example/myapp-operator

输出结构：

myapp-operator/
├── api/
│   └── v1alpha1/
│       ├── myapp_types.go
│       └── zz_generated.deepcopy.go
├── controllers/
│   └── myapp_controller.go
├── config/
│   ├── crd/
│   │   └── bases/
│   │       └── myapp.myapp.example.com.yaml
│   ├── default/
│   │   ├── manager_auth_proxy_patch.yaml
│   │   └── manager_config.yaml
│   └── rbac/
│       ├── role.yaml
│       ├── role_binding.yaml
│       └── service_account.yaml
├── go.mod
├── main.go
└── Makefile

✅ 推荐使用 controller-runtime + kubebuilder 的组合，具备良好的抽象层和测试支持。

四、编写自定义资源（CR）与 CRD

我们以一个简单的 Web 应用管理器为例，实现 MyApp 自定义资源。

4.1 定义 MyApp 类型

编辑 api/v1alpha1/myapp_types.go：

package v1alpha1

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

//+kubebuilder:object:root=true
//+kubebuilder:subresource:status

// MyApp is the Schema for the myapps API
type MyApp struct {
	metav1.TypeMeta   `json:",inline"`
	metav1.ObjectMeta `json:"metadata,omitempty"`

	Spec   MyAppSpec   `json:"spec,omitempty"`
	Status MyAppStatus `json:"status,omitempty"`
}

// MyAppSpec defines the desired state of MyApp
type MyAppSpec struct {
	Replicas int32  `json:"replicas"`
	Image    string `json:"image"`
	Port     int32  `json:"port"`
	Config   Config `json:"config"`
}

type Config struct {
	LogLevel string `json:"logLevel"`
}

// MyAppStatus defines the observed state of MyApp
type MyAppStatus struct {
	Phase           string `json:"phase"`
	Message         string `json:"message"`
	ObservedGeneration int64 `json:"observedGeneration"`
}

//+kubebuilder:object:root=true

// MyAppList contains a list of MyApp
type MyAppList struct {
	metav1.TypeMeta `json:",inline"`
	metav1.ListMeta `json:"metadata,omitempty"`
	Items           []MyApp `json:"items"`
}

func init() {
	SchemeBuilder.Register(&MyApp{}, &MyAppList{})
}

💡 关键注解说明：

  • +kubebuilder:object:root=true：标记为根对象
  • +kubebuilder:subresource:status：启用 status 子资源，允许控制器更新 status
  • SchemeBuilder.Register：注册到 scheme 中，供序列化使用

4.2 生成 CRD 文件

运行以下命令生成 CRD：

make manifests

生成文件位于 config/crd/bases/myapp.myapp.example.com.yaml。

⚠️ 注意：确保 config/crd/patches 目录下的 webhook_patch.yaml 已正确配置（用于 webhook 验证），若不需要可移除。

五、实现控制器逻辑（Reconcile Loop）

控制器位于 controllers/myapp_controller.go。

5.1 初始化控制器

package controllers

import (
	"context"
	"fmt"
	"reflect"

	appv1alpha1 "github.com/example/myapp-operator/api/v1alpha1"
	"github.com/go-logr/logr"
	corev1 "k8s.io/api/core/v1"
	apierrors "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/log"
)

// MyAppReconciler reconciles a MyApp object
type MyAppReconciler struct {
	client.Client
	Scheme *runtime.Scheme
}

//+kubebuilder:rbac:groups=myapp.example.com,resources=myapps,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=myapp.example.com,resources=myapps/status,verbs=get;update;patch
//+kubebuilder:rbac:groups=myapp.example.com,resources=myapps/finalizers,verbs=update

// Reconcile is the main loop that handles MyApps
func (r *MyAppReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	log := log.FromContext(ctx).WithName("myapp-reconciler")

	// Step 1: Fetch the MyApp instance
	myapp := &appv1alpha1.MyApp{}
	err := r.Get(ctx, req.NamespacedName, myapp)
	if err != nil {
		if apierrors.IsNotFound(err) {
			log.Info("MyApp not found, skipping reconcile")
			return ctrl.Result{}, nil
		}
		log.Error(err, "Failed to get MyApp")
		return ctrl.Result{}, err
	}

	// Step 2: Update status.observedGeneration
	myapp.Status.ObservedGeneration = myapp.GetGeneration()

	// Step 3: Set initial phase
	if myapp.Status.Phase == "" {
		myapp.Status.Phase = "Pending"
	}

	// Step 4: Apply reconciliation logic
	result, err := r.reconcileMyApp(ctx, myapp, log)
	if err != nil {
		log.Error(err, "Reconciliation failed")
		return result, err
	}

	// Step 5: Update status
	if err := r.Status().Update(ctx, myapp); err != nil {
		log.Error(err, "Failed to update status")
		return ctrl.Result{}, err
	}

	return result, nil
}

5.2 实现具体业务逻辑

func (r *MyAppReconciler) reconcileMyApp(ctx context.Context, myapp *appv1alpha1.MyApp, log logr.Logger) (ctrl.Result, error) {
	// Define the expected deployment
	expectedDeployment := &corev1.Deployment{
		ObjectMeta: metav1.ObjectMeta{
			Name:      myapp.Name,
			Namespace: myapp.Namespace,
		},
		Spec: corev1.DeploymentSpec{
			Replicas: &myapp.Spec.Replicas,
			Selector: &metav1.LabelSelector{
				MatchLabels: map[string]string{"app": myapp.Name},
			},
			Template: corev1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Labels: map[string]string{"app": myapp.Name},
				},
				Spec: corev1.PodSpec{
					Containers: []corev1.Container{
						{
							Name:  "nginx",
							Image: myapp.Spec.Image,
							Ports: []corev1.ContainerPort{
								{ContainerPort: myapp.Spec.Port},
							},
							Env: []corev1.EnvVar{
								{
									Name:  "LOG_LEVEL",
									Value: myapp.Spec.Config.LogLevel,
								},
							},
						},
					},
				},
			},
		},
	}

	// Check if the deployment already exists
	existingDeployment := &corev1.Deployment{}
	err := r.Get(ctx, types.NamespacedName{Name: myapp.Name, Namespace: myapp.Namespace}, existingDeployment)
	if err != nil {
		if apierrors.IsNotFound(err) {
			// Create new deployment
			log.Info("Creating new Deployment", "name", myapp.Name)
			if err := r.Create(ctx, expectedDeployment); err != nil {
				myapp.Status.Phase = "Failed"
				myapp.Status.Message = fmt.Sprintf("Failed to create Deployment: %v", err)
				return ctrl.Result{}, err
			}
			myapp.Status.Phase = "Running"
			myapp.Status.Message = "Deployment created successfully"
			return ctrl.Result{Requeue: true}, nil
		}
		log.Error(err, "Failed to get existing Deployment")
		return ctrl.Result{}, err
	}

	// Compare and update if needed
	if !reflect.DeepEqual(expectedDeployment.Spec, existingDeployment.Spec) {
		log.Info("Updating Deployment", "name", myapp.Name)
		existingDeployment.Spec = expectedDeployment.Spec
		if err := r.Update(ctx, existingDeployment); err != nil {
			myapp.Status.Phase = "Failed"
			myapp.Status.Message = fmt.Sprintf("Failed to update Deployment: %v", err)
			return ctrl.Result{}, err
		}
		myapp.Status.Phase = "Running"
		myapp.Status.Message = "Deployment updated"
		return ctrl.Result{Requeue: true}, nil
	}

	// Check if all replicas are ready
	if existingDeployment.Status.ReadyReplicas < *existingDeployment.Spec.Replicas {
		myapp.Status.Phase = "Running"
		myapp.Status.Message = fmt.Sprintf("Waiting for %d/%d replicas ready", existingDeployment.Status.ReadyReplicas, *existingDeployment.Spec.Replicas)
		return ctrl.Result{RequeueAfter: 5 * time.Second}, nil
	}

	// All good!
	myapp.Status.Phase = "Succeeded"
	myapp.Status.Message = "All replicas are running"
	return ctrl.Result{}, nil
}

✅ 最佳实践：

  • 使用 context 传递超时与取消信号
  • 优先使用 client.Get() 和 client.Create() 等安全接口
  • 使用 ctrl.Result{Requeue: true} 触发重试，避免无限循环
  • RequeueAfter 设置合理的重试间隔（如 5s ~ 30s）

六、权限与 RBAC 配置

Operator 必须拥有足够的权限才能操作 Kubernetes 资源。

6.1 自动生成 RBAC

kubebuilder 会根据注解自动生成 RBAC 规则。查看 config/rbac/role.yaml：

apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: manager-role
rules:
  - apiGroups: ["myapp.example.com"]
    resources: ["myapps"]
    verbs: ["get", "list", "watch", "create", "update", "patch", "delete"]
  - apiGroups: ["myapp.example.com"]
    resources: ["myapps/status"]
    verbs: ["get", "update", "patch"]
  - apiGroups: ["myapp.example.com"]
    resources: ["myapps/finalizers"]
    verbs: ["update"]
  - apiGroups: ["apps"]
    resources: ["deployments"]
    verbs: ["get", "list", "watch", "create", "update", "patch", "delete"]
  - apiGroups: ["core"]
    resources: ["pods", "services"]
    verbs: ["get", "list", "watch", "create", "update", "patch", "delete"]

📌 建议：遵循最小权限原则，仅授予必要权限。

6.2 部署 ServiceAccount

# config/rbac/service_account.yaml
apiVersion: v1
kind: ServiceAccount
metadata:
  name: manager
  namespace: system

七、部署 Operator 到集群

7.1 构建并推送镜像

make docker-build docker-push IMG=ghcr.io/example/myapp-operator:v0.1.0

✅ 使用 GitHub Container Registry 或私有镜像仓库。

7.2 应用 CRD 和 Operator

make install
make deploy IMG=ghcr.io/example/myapp-operator:v0.1.0
  • make install：安装 CRD 到集群
  • make deploy：部署 Operator Pod

验证部署成功：

kubectl get pods -n system
# 输出应包含 myapp-operator-xxxxx

kubectl get crd myapps.myapp.example.com

八、使用自定义资源管理应用

8.1 创建 MyApp 实例

# deploy/myapp-instance.yaml
apiVersion: myapp.example.com/v1alpha1
kind: MyApp
metadata:
  name: myapp-demo
spec:
  replicas: 2
  image: nginx:1.25
  port: 8080
  config:
    logLevel: "debug"

应用该 YAML：

kubectl apply -f deploy/myapp-instance.yaml

8.2 查看状态

kubectl get myapp -A
kubectl describe myapp myapp-demo

输出示例：

Status:
  Phase:           Succeeded
  Message:         All replicas are running
  ObservedGeneration: 1

查看生成的 Deployment：

kubectl get deploy myapp-demo

九、进阶功能：健康检查、滚动更新、Finalizer

9.1 添加 Finalizer 实现优雅删除

防止资源残留，使用 Finalizer 保证清理顺序。

// 在 Reconcile 中添加 Finalizer 处理
if len(myapp.Finalizers) == 0 {
    myapp.Finalizers = append(myapp.Finalizers, "finalizer.myapp.example.com")
    if err := r.Update(ctx, myapp); err != nil {
        return ctrl.Result{}, err
    }
}

// 删除时清空 Finalizer
if myapp.DeletionTimestamp.IsZero() {
    // 正常处理...
} else {
    // 删除阶段：移除 Finalizer
    myapp.Finalizers = removeFinalizer(myapp.Finalizers, "finalizer.myapp.example.com")
    if err := r.Update(ctx, myapp); err != nil {
        return ctrl.Result{}, err
    }
    return ctrl.Result{}, nil
}

9.2 实现健康检查（Probe）

可在 Deployment 中加入 Liveness/Readiness Probe：

corev1.Container{
    Name:  "nginx",
    Image: myapp.Spec.Image,
    Ports: []corev1.ContainerPort{
        {ContainerPort: myapp.Spec.Port},
    },
    LivenessProbe: &corev1.Probe{
        HTTPGet: &corev1.HTTPGetAction{
            Path: "/healthz",
            Port: intstr.IntOrString{IntVal: myapp.Spec.Port},
        },
        InitialDelaySeconds: 10,
        PeriodSeconds:       5,
    },
    ReadinessProbe: &corev1.Probe{
        HTTPGet: &corev1.HTTPGetAction{
            Path: "/ready",
            Port: intstr.IntOrString{IntVal: myapp.Spec.Port},
        },
        InitialDelaySeconds: 5,
        PeriodSeconds:       3,
    },
}

9.3 支持滚动更新策略

通过修改 Deployment.Spec.Strategy 实现蓝绿或滚动更新：

expectedDeployment.Spec.Strategy = appsv1.DeploymentStrategy{
    Type: appsv1.RollingUpdateDeploymentStrategyType,
    RollingUpdate: &appsv1.RollingUpdateDeployment{
        MaxSurge:       &intstr.IntOrString{IntVal: 1},
        MaxUnavailable: &intstr.IntOrString{IntVal: 0},
    },
}

十、测试与 CI/CD 整合

10.1 编写单元测试

// controllers/myapp_controller_test.go
func TestReconcile(t *testing.T) {
    ctx := context.Background()
    r := &MyAppReconciler{
        Client: fake.NewClientBuilder().Build(),
        Scheme: scheme.Scheme,
    }

    myapp := &appv1alpha1.MyApp{
        ObjectMeta: metav1.ObjectMeta{Name: "test", Namespace: "default"},
        Spec: appv1alpha1.MyAppSpec{
            Replicas: 2,
            Image:    "nginx:1.25",
            Port:     8080,
        },
    }

    // Create CR
    err := r.Client.Create(ctx, myapp)
    assert.NoError(t, err)

    // Reconcile
    _, err = r.Reconcile(ctx, ctrl.Request{NamespacedName: types.NamespacedName{Name: "test", Namespace: "default"}})
    assert.NoError(t, err)

    // Check if Deployment was created
    dep := &appsv1.Deployment{}
    err = r.Client.Get(ctx, types.NamespacedName{Name: "test", Namespace: "default"}, dep)
    assert.NoError(t, err)
    assert.Equal(t, int32(2), *dep.Spec.Replicas)
}

10.2 集成 CI/CD

使用 GitHub Actions 示例：

# .github/workflows/build.yml
name: Build & Deploy

on:
  push:
    branches: [main]

jobs:
  build:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - name: Set up Go
        uses: actions/setup-go@v3
        with:
          go-version: '1.21'
      - name: Build
        run: make build
      - name: Docker Build
        run: make docker-build IMG=ghcr.io/example/myapp-operator:v0.1.0
      - name: Docker Push
        run: make docker-push IMG=ghcr.io/example/myapp-operator:v0.1.0
      - name: Deploy
        run: make deploy IMG=ghcr.io/example/myapp-operator:v0.1.0

十一、常见问题与最佳实践总结

问题 解决方案
Controller 不响应变更 检查 CRD 是否已注册；确认 RBAC 权限；使用 kubectl logs <pod> 查看日志
Deployment 创建失败 检查镜像是否存在；网络策略限制；PodSecurityPolicy
状态未更新 确保调用了 r.Status().Update()
重复 reconcile 避免无条件 requeue；合理设置 RequeueAfter
性能瓶颈 使用 RateLimitingQueue；分批处理；异步任务

✅ 最佳实践清单：

  1. 使用 kubebuilder + controller-runtime 框架
  2. 所有资源操作使用 client 接口
  3. 保持 status 与 spec 同步
  4. 使用 Finalizer 保障删除安全
  5. 加入健康检查与探针
  6. 编写单元测试与 e2e 测试
  7. 使用 CI/CD 自动化发布
  8. 文档化 API 设计与行为说明

十二、结语：迈向云原生运维新时代

Kubernetes Operator 不仅仅是“自动化脚本”的升级版，更是将领域知识封装为可编程、可复用、可测试的软件资产。通过本文实战，你已掌握从零构建 Operator 的全流程：定义 CR、设计控制器、实现协调逻辑、部署上线、测试验证。

未来，随着更多复杂系统（如 Kafka、Prometheus、TiDB）被 Operator 化，企业将真正实现“声明式运维”——只需写出应用意图，剩下的由 Kubernetes 自动完成。

🚀 记住：
“你写的不是代码，而是你的运维哲学。”

现在，就动手构建属于你的第一个 Operator 吧！

作者：云原生技术布道者
时间：2025年4月5日
GitHub 项目地址：https://github.com/example/myapp-operator
参考文档：https://book.kubebuilder.io/

🔗 延伸阅读：

  • Kubernetes Operators by Example
  • controller-runtime 官方文档
  • Kubebuilder 官网

本文来自极简博客，作者：数据科学实验室，转载请注明原文链接：云原生时代Kubernetes Operator开发实战：从零开始构建自定义控制器管理复杂应用

打赏

本文固定链接: https://www.cxy163.net/archives/6865 | 绝缘体

该日志由 绝缘体.. 于 2022年07月14日 发表在 未分类 分类下， 你可以发表评论，并在保留原文地址及作者的情况下引用到你的网站或博客。
原创文章转载请注明: 云原生时代Kubernetes Operator开发实战：从零开始构建自定义控制器管理复杂应用 | 绝缘体
关键字: , , , ,
【上一篇】
【下一篇】

您可能还会对这些文章感兴趣！

  1. CVE-2021-35042
  2. CVE-2022-0543
  3. go语言编译过程概述
  4. golang内存对齐分析(转载)
  5. DOS攻击(一)
  6. 身份认证种类及攻击
  7. sqlmap源码分析(一)
  8. python黑帽子(第二章)
  9. python安全脚本
  10. CDN绕过
  11. docker进阶_dockerswarm
  12. docker基础_网络模式
  13. docker进阶_docker-compose
  14. php的password_hash方法详解
  15. laravel 缓存系统的使用
  16. Python selenium ,一定要会用selenium的等待，三

云原生时代Kubernetes Operator开发实战：从零开始构建自定义控制器管理复杂应用：等您坐沙发呢！

发表评论


快捷键：Ctrl+Enter