Kubernetes Operator 模式详解与 Operator SDK 自定义 Operator 开发实战

2025/7/6 13:28:21 139 0 0 0

在云原生时代，Kubernetes (K8s) 已经成为容器编排的事实标准。随着 K8s 的广泛应用，越来越多的应用开始迁移到 K8s 平台上。然而，对于一些复杂应用，例如数据库、消息队列等，传统的 Deployment 方式可能无法满足其自动化运维的需求。这时，Kubernetes Operator 模式应运而生。

什么是 Kubernetes Operator？

Operator 是一种遵循 Kubernetes API 约定的应用特定控制器。它扩展了 Kubernetes 的控制平面，允许用户像管理内置资源一样管理复杂应用。简单来说，Operator 就是一个自动化运维工程师，它能够理解应用的内部状态，并根据用户的意图自动完成应用的部署、配置、升级、备份、恢复等操作。

Operator 的核心思想：

应用特定知识： Operator 具备特定应用的领域知识，能够理解应用的内部状态和运维需求。
自动化运维： Operator 通过监听 Kubernetes API，根据用户的声明式配置自动完成应用的运维操作。
扩展 Kubernetes API： Operator 通过自定义资源 (Custom Resource Definition, CRD) 扩展 Kubernetes API，允许用户以声明式的方式管理应用。

Operator 的优势：

简化应用运维： Operator 将复杂应用的运维操作自动化，降低了运维成本。
提高应用可靠性： Operator 能够监控应用的健康状态，并自动进行故障恢复，提高了应用的可靠性。
标准化应用管理： Operator 提供了一种标准化的应用管理方式，方便用户在不同的 Kubernetes 集群中部署和管理应用。

Operator 模式的核心组件

CRD（Custom Resource Definition）： CRD 用于扩展 Kubernetes API，定义了用户可以创建的自定义资源类型。例如，可以创建一个名为 MyDatabase 的 CRD，用于表示一个数据库实例。
Controller： Controller 负责监听 Kubernetes API，并根据 CRD 的状态变化执行相应的操作。例如，当创建一个新的 MyDatabase 实例时，Controller 会自动创建相应的 Pod、Service 等资源，并配置数据库。
Custom Resource (CR)： CR 是基于 CRD 创建的自定义资源实例。例如，可以创建一个名为 mydb-instance 的 MyDatabase CR，用于表示一个具体的数据库实例。

Operator SDK 简介

Operator SDK 是一个用于快速构建 Kubernetes Operator 的框架。它提供了一系列工具和库，可以帮助开发者快速生成 Operator 的代码框架，并简化 Operator 的开发过程。

Operator SDK 的主要功能：

代码生成： Operator SDK 可以根据 CRD 自动生成 Operator 的代码框架，包括 Controller、CRD、RBAC 等。
测试框架： Operator SDK 提供了一套测试框架，可以方便开发者对 Operator 进行单元测试和集成测试。
打包发布： Operator SDK 可以将 Operator 打包成 Docker 镜像，并发布到 Operator Hub 等平台。

Operator SDK 支持的语言：

Go： Operator SDK 的官方推荐语言，也是最常用的语言。
Ansible： Operator SDK 支持使用 Ansible playbook 来定义 Operator 的运维逻辑。
Helm： Operator SDK 支持使用 Helm chart 来定义 Operator 的部署配置。

使用 Operator SDK 开发自定义 Operator

下面以 Go 语言为例，演示如何使用 Operator SDK 开发一个简单的自定义 Operator，该 Operator 用于管理一个名为 Memcached 的应用。

1. 安装 Operator SDK：

go install github.com/operator-framework/operator-sdk/cmd/operator-sdk@latest

2. 创建 Operator 项目：

operator-sdk new memcached-operator --repo github.com/example/memcached-operator

3. 定义 CRD：

在 config/crd/bases/cache.example.com_memcacheds.yaml 文件中定义 Memcached CRD。以下是一个简单的示例：

apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
metadata:
  name: memcacheds.cache.example.com
spec:
  group: cache.example.com
  versions:
  - name: v1alpha1
    served: true
    storage: true
    schema:
      openAPIV3Schema:
        type: object
        properties:
          spec:
            type: object
            properties:
              size:
                type: integer
                description: The number of Memcached replicas
                default: 3
          status:
            type: object
            properties:
              nodes:
                type: array
                items:
                  type: string
  scope: Namespaced
  names:
    plural: memcacheds
    singular: memcached
    kind: Memcached
    shortNames:
    - mc

4. 生成 Controller 代码：

operator-sdk generate k8s
operator-sdk generate controller cache.example.com/v1alpha1 Memcached --group cache --version v1alpha1 --kind Memcached

5. 实现 Controller 逻辑：

在 controllers/memcached_controller.go 文件中实现 Controller 的 reconcile 逻辑。该逻辑负责创建、更新和删除 Memcached 实例。

以下是一个简单的示例：

package controllers

import (
    "context"

    cachev1alpha1 "github.com/example/memcached-operator/api/v1alpha1"
    appsv1 "k8s.io/api/apps/v1"
    corev1 "k8s.io/api/core/v1"
    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"
)

// MemcachedReconciler reconciles a Memcached object
type MemcachedReconciler struct {
    client.Client
    Scheme *runtime.Scheme
}

//+kubebuilder:rbac:groups=cache.example.com,resources=memcacheds,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=cache.example.com,resources=memcacheds/status,verbs=get;update;patch
//+kubebuilder:rbac:groups=cache.example.com,resources=memcacheds/finalizers,verbs=update
//+kubebuilder:rbac:groups=apps,resources=deployments,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=core,resources=pods,verbs=get;list;watch

// Reconcile is part of the main kubernetes reconciliation loop which aims to
// move the current state of the cluster closer to the desired state.
// TODO(user): Modify the Reconcile function to compare the state specified by
// the Memcached object against the actual cluster state, and then
// perform operations to make the cluster state reflect the state specified by
// the user.
//
// For more details, check Reconcile and its Result here:
// - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.11.0/pkg/reconcile
func (r *MemcachedReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
    log := log.FromContext(ctx)

    // TODO(user): your logic here

    // Fetch the Memcached instance
    memcached := &cachev1alpha1.Memcached{}
    err := r.Get(ctx, req.NamespacedName, memcached)
    if err != nil {
        log.Error(err, "unable to fetch Memcached")
        // we'll ignore not-found errors, since they can't be fixed by an immediate
        // requeue (we'll need to wait for a new notification), and we can get them
        // on deleted requests.
        return ctrl.Result{}, client.IgnoreNotFound(err)
    }

    // Define a new Deployment object
deployment := &appsv1.Deployment{
    ObjectMeta: metav1.ObjectMeta{
        Name:      memcached.Name,
        Namespace: memcached.Namespace,
    },
    Spec: appsv1.DeploymentSpec{
        Replicas: &memcached.Spec.Size,
        Selector: &metav1.LabelSelector{
            MatchLabels: map[string]string{
                "app": memcached.Name,
            },
        },
        Template: corev1.PodTemplateSpec{
            ObjectMeta: metav1.ObjectMeta{
                Labels: map[string]string{
                    "app": memcached.Name,
                },
            },
            Spec: corev1.PodSpec{
                Containers: []corev1.Container{{
                    Image:   "memcached:1.6.15",
                    Name:    "memcached",
                    Ports: []corev1.ContainerPort{{
                        ContainerPort: 11211,
                        Name:          "memcached",
                    }},
                }},
            },
        },
    },
}
    if err := ctrl.SetControllerReference(memcached, deployment, r.Scheme); err != nil {
        return ctrl.Result{}, err
    }

    // Check if the Deployment already exists, if not create a new one
    found := &appsv1.Deployment{}
    err = r.Get(ctx, client.ObjectKey{Name: deployment.Name, Namespace: deployment.Namespace}, found)
    if err != nil && client.IgnoreNotFound(err) != nil {
        log.Error(err, "unable to get Deployment")
        return ctrl.Result{}, err
    }

    if err == nil {
        log.Info("Deployment already exists", "Deployment.Namespace", found.Namespace, "Deployment.Name", found.Name)
        return ctrl.Result{}, nil
    }

    log.Info("Creating a new Deployment", "Deployment.Namespace", deployment.Namespace, "Deployment.Name", deployment.Name)
    err = r.Create(ctx, deployment)
    if err != nil {
        log.Error(err, "unable to create Deployment", "Deployment.Namespace", deployment.Namespace, "Deployment.Name", deployment.Name)
        return ctrl.Result{}, err
    }

    // TODO(user): Change this to be the object you want to reconcile and modify the logic above.
    // Note that is Reconcile takes less than 1 second, then any requeues are
    // considered "immediate" by the runtime and are full of wasteful reconcile cycles.
    //return ctrl.Result{}, nil

    // Return and requeue the event if the deployment is not yet created
    return ctrl.Result{Requeue: true}, nil
}

// SetupWithManager sets up the controller with the Manager.
func (r *MemcachedReconciler) SetupWithManager(mgr ctrl.Manager) error {
    return ctrl.NewControllerManagedBy(mgr).For(&cachev1alpha1.Memcached{}).Owns(&appsv1.Deployment{}).Complete(r)
}

6. 运行 Operator：

make install
make run

7. 创建 Memcached 实例：

apiVersion: cache.example.com/v1alpha1
kind: Memcached
metadata:
  name: memcached-sample
spec:
  size: 3

将以上 YAML 文件保存为 memcached.yaml，然后执行以下命令创建 Memcached 实例：

kubectl apply -f memcached.yaml

8. 验证 Operator：

使用以下命令查看 Memcached 实例的状态：

kubectl get memcacheds

您应该能够看到一个名为 memcached-sample 的 Memcached 实例，并且其状态为 Running。

总结

Kubernetes Operator 模式是一种强大的应用管理方式，它可以帮助用户自动化运维复杂应用，降低运维成本，提高应用可靠性。Operator SDK 是一个用于快速构建 Kubernetes Operator 的框架，它可以帮助开发者快速生成 Operator 的代码框架，并简化 Operator 的开发过程。通过本文的介绍，相信您已经对 Kubernetes Operator 模式和 Operator SDK 有了初步的了解。希望您能够尝试使用 Operator SDK 开发自己的自定义 Operator，并将其应用到实际的生产环境中。

进一步学习

Operator Framework 官方网站： https://operatorframework.io/
Operator SDK GitHub 仓库： https://github.com/operator-framework/operator-sdk
Kubernetes 官方文档： https://kubernetes.io/docs/concepts/extend-kubernetes/operator/

希望以上内容能够帮助你理解 Kubernetes Operator 模式以及如何使用 Operator SDK 开发自定义 Operator。祝你学习顺利！

云原生小能手 Kubernetes Operator SDK 自定义 Operator