云原生时代Kubernetes Operator开发实战：从零开始构建自定义控制器，实现应用自动化运维

引言

在云原生技术浪潮的推动下，Kubernetes已经成为容器编排的事实标准。随着企业对自动化运维需求的不断提升，Operator模式作为一种强大的扩展机制，正在被广泛应用于各种复杂的运维场景中。Operator通过将领域专家的知识编码到控制器中，实现了对复杂应用的自动化管理。

本文将深入探讨Kubernetes Operator的核心概念、设计原理和实现方法，通过一个完整的实际案例，带领读者从零开始构建一个生产级的自定义控制器，掌握Operator开发的关键技术点和最佳实践。

什么是Kubernetes Operator

Operator的核心概念

Operator是Kubernetes生态系统中的一种扩展机制，它将特定应用的运维知识（如部署、配置、监控、备份等）编码到一个控制器中。Operator本质上是一个运行在Kubernetes集群中的自定义控制器，它通过监听和处理自定义资源（Custom Resources）来实现对应用的自动化管理。

Operator的工作原理

Operator的工作原理基于以下核心组件：

1. Custom Resource Definition (CRD)：定义自定义资源的结构和规范
2. Controller：监听自定义资源的变化，执行相应的操作
3. Custom Resource：用户创建的具体实例
4. Reconcile Loop：控制器的核心循环，用于同步期望状态与实际状态

Operator的价值与应用场景

Operator的主要价值在于：

• 实现复杂应用的自动化部署和管理
• 将运维经验编码化，降低运维门槛
• 提供统一的管理界面和操作接口
• 支持高可用、备份、升级等高级功能

常见的应用场景包括数据库管理（如MySQL、PostgreSQL Operator）、消息队列管理（如Kafka Operator）、微服务治理等。

Kubernetes CRD设计与实现

CRD基础概念

Custom Resource Definition (CRD)是Operator模式的基础。通过CRD，我们可以定义自己的资源类型，这些资源可以像原生Kubernetes资源一样被创建、更新和删除。

创建示例CRD

让我们以一个简单的数据库集群为例，创建相应的CRD：

apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
metadata:
  name: databases.example.com
spec:
  group: example.com
  versions:
  - name: v1
    schema:
      openAPIV3Schema:
        type: object
        properties:
          spec:
            type: object
            required:
            - version
            - replicas
            properties:
              version:
                type: string
                description: Database version
              replicas:
                type: integer
                minimum: 1
                description: Number of replicas
              storage:
                type: object
                properties:
                  size:
                    type: string
                    description: Storage size
                  class:
                    type: string
                    description: Storage class name
          status:
            type: object
            properties:
              phase:
                type: string
                description: Current phase of the database
              replicas:
                type: integer
                description: Number of replicas
              readyReplicas:
                type: integer
                description: Number of ready replicas
    served: true
    storage: true
  scope: Namespaced
  names:
    plural: databases
    singular: database
    kind: Database

CRD字段详解

在CRD的设计中，需要特别注意以下关键字段：

• group：API组名称，用于区分不同类型的自定义资源
• version：API版本，支持多个版本共存
• scope：资源作用域，可以是Namespaced或Cluster
• names：资源的命名规则
• schema：资源的结构定义，使用OpenAPI v3规范

CRD验证与默认值设置

为了确保资源的正确性，我们可以在CRD中添加验证规则：

apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
metadata:
  name: databases.example.com
spec:
  group: example.com
  versions:
  - name: v1
    schema:
      openAPIV3Schema:
        type: object
        properties:
          spec:
            type: object
            required:
            - version
            - replicas
            properties:
              version:
                type: string
                enum:
                - "12"
                - "13"
                - "14"
                description: Database version (12, 13, or 14)
              replicas:
                type: integer
                minimum: 1
                maximum: 10
                description: Number of replicas (1-10)
              storage:
                type: object
                properties:
                  size:
                    type: string
                    pattern: "^[0-9]+(Gi|Mi|Ki)$"
                    description: Storage size with unit
                  class:
                    type: string
                    description: Storage class name
    # 验证规则
    validation:
      openAPIV3Schema:
        type: object
        properties:
          spec:
            type: object
            required:
            - version
            - replicas
            properties:
              version:
                type: string
                enum:
                - "12"
                - "13"
                - "14"
              replicas:
                type: integer
                minimum: 1
                maximum: 10
    # 默认值设置
    additionalPrinterColumns:
    - name: Version
      type: string
      JSONPath: .spec.version
    - name: Replicas
      type: integer
      JSONPath: .spec.replicas
    - name: Status
      type: string
      JSONPath: .status.phase

Operator控制器核心实现

控制器架构设计

一个典型的Operator控制器包含以下组件：

1. Reconcile函数：核心的同步逻辑
2. Client API：与Kubernetes API服务器交互
3. Scheme：资源类型注册表
4. Manager：控制器管理器
5. Recorder：事件记录器

初始化控制器结构

package main

import (
    "context"
    "fmt"
    "os"
    "time"

    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"

    examplev1 "your-module/api/v1"
)

// DatabaseReconciler 定义数据库控制器
type DatabaseReconciler struct {
    client.Client
    Scheme *runtime.Scheme
}

// Reconcile 核心同步函数
func (r *DatabaseReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
    log := log.FromContext(ctx)
    log.Info("Reconciling Database", "name", req.Name, "namespace", req.Namespace)

    // 获取数据库资源
    database := &examplev1.Database{}
    if err := r.Get(ctx, req.NamespacedName, database); err != nil {
        log.Error(err, "Unable to fetch Database")
        return ctrl.Result{}, client.IgnoreNotFound(err)
    }

    // 检查是否需要删除
    if database.DeletionTimestamp != nil {
        return r.handleDelete(ctx, database)
    }

    // 处理创建/更新逻辑
    return r.reconcileDatabase(ctx, database)
}

// reconcileDatabase 核心业务逻辑
func (r *DatabaseReconciler) reconcileDatabase(ctx context.Context, database *examplev1.Database) (ctrl.Result, error) {
    log := log.FromContext(ctx)
    
    // 1. 创建配置文件
    if err := r.createConfigMap(ctx, database); err != nil {
        return ctrl.Result{}, fmt.Errorf("failed to create configmap: %w", err)
    }

    // 2. 创建服务
    if err := r.createService(ctx, database); err != nil {
        return ctrl.Result{}, fmt.Errorf("failed to create service: %w", err)
    }

    // 3. 创建StatefulSet
    if err := r.createStatefulSet(ctx, database); err != nil {
        return ctrl.Result{}, fmt.Errorf("failed to create statefulset: %w", err)
    }

    // 4. 更新状态
    if err := r.updateStatus(ctx, database); err != nil {
        return ctrl.Result{Requeue: true}, fmt.Errorf("failed to update status: %w", err)
    }

    return ctrl.Result{}, nil
}

控制器初始化与注册

func main() {
    // 设置日志
    ctrl.SetLogger(zap.New(zap.UseDevMode(true)))

    // 创建控制器管理器
    mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
        Scheme:             scheme,
        MetricsBindAddress: "0",
        Port:               9443,
        LeaderElection:     false,
        LeaseDuration:      &leaderElectionLeaseDuration,
        RenewDeadline:      &leaderElectionRenewDeadline,
    })
    if err != nil {
        setupLog.Error(err, "unable to start manager")
        os.Exit(1)
    }

    // 注册控制器
    if err = (&DatabaseReconciler{
        Client: mgr.GetClient(),
        Scheme: mgr.GetScheme(),
    }).SetupWithManager(mgr); err != nil {
        setupLog.Error(err, "unable to create controller", "controller", "Database")
        os.Exit(1)
    }

    // 启动管理器
    if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
        setupLog.Error(err, "problem running manager")
        os.Exit(1)
    }
}

状态管理和监控

// updateStatus 更新数据库状态
func (r *DatabaseReconciler) updateStatus(ctx context.Context, database *examplev1.Database) error {
    log := log.FromContext(ctx)
    
    // 获取当前状态
    statefulSet := &appsv1.StatefulSet{}
    err := r.Get(ctx, client.ObjectKey{Name: database.Name, Namespace: database.Namespace}, statefulSet)
    if err != nil {
        return fmt.Errorf("failed to get statefulset: %w", err)
    }

    // 更新状态信息
    database.Status.Phase = "Running"
    database.Status.Replicas = int32(statefulSet.Status.Replicas)
    database.Status.ReadyReplicas = int32(statefulSet.Status.ReadyReplicas)

    // 保存状态
    if err := r.Status().Update(ctx, database); err != nil {
        log.Error(err, "Failed to update Database status")
        return err
    }

    log.Info("Database status updated", 
        "phase", database.Status.Phase,
        "replicas", database.Status.Replicas,
        "readyReplicas", database.Status.ReadyReplicas)
    
    return nil
}

实际案例：构建数据库Operator

需求分析与设计

假设我们要为PostgreSQL数据库构建一个Operator，该Operator需要支持以下功能：

1. 自动创建和管理PostgreSQL集群
2. 支持版本升级和配置变更
3. 提供监控和健康检查
4. 实现备份和恢复机制

完整的CRD定义

apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
metadata:
  name: postgresqls.example.com
spec:
  group: example.com
  versions:
  - name: v1
    schema:
      openAPIV3Schema:
        type: object
        properties:
          spec:
            type: object
            required:
            - version
            - replicas
            properties:
              version:
                type: string
                description: PostgreSQL version (e.g., "12", "13", "14")
              replicas:
                type: integer
                minimum: 1
                maximum: 10
                description: Number of PostgreSQL instances
              storage:
                type: object
                required:
                - size
                properties:
                  size:
                    type: string
                    pattern: "^[0-9]+(Gi|Mi|Ki)$"
                    description: Storage size with unit
                  class:
                    type: string
                    description: Storage class name
              configuration:
                type: object
                properties:
                  maxConnections:
                    type: integer
                    minimum: 100
                    maximum: 10000
                    description: Maximum number of connections
                  sharedBuffers:
                    type: string
                    pattern: "^[0-9]+(GB|MB|KB)$"
                    description: Shared memory buffer size
              backup:
                type: object
                properties:
                  enabled:
                    type: boolean
                    description: Enable backup
                  schedule:
                    type: string
                    description: Backup cron schedule
                  retention:
                    type: integer
                    minimum: 1
                    description: Number of backups to retain
          status:
            type: object
            properties:
              phase:
                type: string
                description: Current phase (Pending, Running, Failed)
              replicas:
                type: integer
                description: Total number of replicas
              readyReplicas:
                type: integer
                description: Number of ready replicas
              version:
                type: string
                description: Current PostgreSQL version
              conditions:
                type: array
                items:
                  type: object
                  properties:
                    type:
                      type: string
                    status:
                      type: string
                    reason:
                      type: string
                    message:
                      type: string
    served: true
    storage: true
  scope: Namespaced
  names:
    plural: postgresqls
    singular: postgresql
    kind: PostgreSQL
    listKind: PostgreSQLList

控制器核心实现

package controllers

import (
    "context"
    "fmt"
    "time"

    appsv1 "k8s.io/api/apps/v1"
    corev1 "k8s.io/api/core/v1"
    rbacv1 "k8s.io/api/rbac/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/controller/controllerutil"
    "sigs.k8s.io/controller-runtime/pkg/log"

    examplev1 "your-module/api/v1"
)

// PostgreSQLReconciler 定义PostgreSQL控制器
type PostgreSQLReconciler struct {
    client.Client
    Scheme *runtime.Scheme
}

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

    // 获取PostgreSQL资源
    postgresql := &examplev1.PostgreSQL{}
    if err := r.Get(ctx, req.NamespacedName, postgresql); err != nil {
        if apierrors.IsNotFound(err) {
            log.Info("PostgreSQL resource not found, skipping reconciliation")
            return ctrl.Result{}, nil
        }
        log.Error(err, "Failed to get PostgreSQL resource")
        return ctrl.Result{}, err
    }

    // 处理删除逻辑
    if postgresql.DeletionTimestamp != nil {
        return r.handleDelete(ctx, postgresql)
    }

    // 处理创建/更新逻辑
    result, err := r.reconcileResources(ctx, postgresql)
    if err != nil {
        log.Error(err, "Reconciliation failed")
        return ctrl.Result{RequeueAfter: 30 * time.Second}, err
    }

    return result, nil
}

// reconcileResources 核心资源创建和管理逻辑
func (r *PostgreSQLReconciler) reconcileResources(ctx context.Context, postgresql *examplev1.PostgreSQL) (ctrl.Result, error) {
    log := log.FromContext(ctx)

    // 1. 创建配置映射
    if err := r.createConfigMap(ctx, postgresql); err != nil {
        return ctrl.Result{}, fmt.Errorf("failed to create configmap: %w", err)
    }

    // 2. 创建服务
    if err := r.createService(ctx, postgresql); err != nil {
        return ctrl.Result{}, fmt.Errorf("failed to create service: %w", err)
    }

    // 3. 创建RBAC权限
    if err := r.createRBAC(ctx, postgresql); err != nil {
        return ctrl.Result{}, fmt.Errorf("failed to create RBAC: %w", err)
    }

    // 4. 创建StatefulSet
    if err := r.createStatefulSet(ctx, postgresql); err != nil {
        return ctrl.Result{}, fmt.Errorf("failed to create statefulset: %w", err)
    }

    // 5. 更新状态
    if err := r.updateStatus(ctx, postgresql); err != nil {
        log.Error(err, "Failed to update status")
        return ctrl.Result{RequeueAfter: time.Minute}, err
    }

    return ctrl.Result{}, nil
}

// createConfigMap 创建配置映射
func (r *PostgreSQLReconciler) createConfigMap(ctx context.Context, postgresql *examplev1.PostgreSQL) error {
    configMap := &corev1.ConfigMap{
        ObjectMeta: metav1.ObjectMeta{
            Name:      fmt.Sprintf("%s-config", postgresql.Name),
            Namespace: postgresql.Namespace,
        },
        Data: map[string]string{
            "postgresql.conf": r.generatePostgreSQLConfig(postgresql),
        },
    }

    if err := controllerutil.SetControllerReference(postgresql, configMap, r.Scheme); err != nil {
        return fmt.Errorf("failed to set owner reference: %w", err)
    }

    return r.createOrUpdate(ctx, configMap)
}

// createService 创建服务
func (r *PostgreSQLReconciler) createService(ctx context.Context, postgresql *examplev1.PostgreSQL) error {
    service := &corev1.Service{
        ObjectMeta: metav1.ObjectMeta{
            Name:      postgresql.Name,
            Namespace: postgresql.Namespace,
        },
        Spec: corev1.ServiceSpec{
            Selector: map[string]string{
                "app": postgresql.Name,
            },
            Ports: []corev1.ServicePort{
                {
                    Port:       5432,
                    TargetPort: intstr.FromInt(5432),
                },
            },
            ClusterIP: corev1.ClusterIPNone,
        },
    }

    if err := controllerutil.SetControllerReference(postgresql, service, r.Scheme); err != nil {
        return fmt.Errorf("failed to set owner reference: %w", err)
    }

    return r.createOrUpdate(ctx, service)
}

// createStatefulSet 创建StatefulSet
func (r *PostgreSQLReconciler) createStatefulSet(ctx context.Context, postgresql *examplev1.PostgreSQL) error {
    statefulSet := &appsv1.StatefulSet{
        ObjectMeta: metav1.ObjectMeta{
            Name:      postgresql.Name,
            Namespace: postgresql.Namespace,
        },
        Spec: appsv1.StatefulSetSpec{
            Replicas: &postgresql.Spec.Replicas,
            Selector: &metav1.LabelSelector{
                MatchLabels: map[string]string{
                    "app": postgresql.Name,
                },
            },
            Template: corev1.PodTemplateSpec{
                ObjectMeta: metav1.ObjectMeta{
                    Labels: map[string]string{
                        "app": postgresql.Name,
                    },
                },
                Spec: corev1.PodSpec{
                    Containers: []corev1.Container{
                        {
                            Name:  "postgresql",
                            Image: fmt.Sprintf("postgres:%s", postgresql.Spec.Version),
                            Ports: []corev1.ContainerPort{
                                {
                                    ContainerPort: 5432,
                                },
                            },
                            VolumeMounts: []corev1.VolumeMount{
                                {
                                    Name:      "data",
                                    MountPath: "/var/lib/postgresql/data",
                                },
                                {
                                    Name:      "config",
                                    MountPath: "/etc/postgresql",
                                },
                            },
                        },
                    },
                    Volumes: []corev1.Volume{
                        {
                            Name: "data",
                            VolumeSource: corev1.VolumeSource{
                                PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{
                                    ClaimName: fmt.Sprintf("%s-data", postgresql.Name),
                                },
                            },
                        },
                        {
                            Name: "config",
                            VolumeSource: corev1.VolumeSource{
                                ConfigMap: &corev1.ConfigMapVolumeSource{
                                    LocalObjectReference: corev1.LocalObjectReference{
                                        Name: fmt.Sprintf("%s-config", postgresql.Name),
                                    },
                                },
                            },
                        },
                    },
                },
            },
            VolumeClaimTemplates: []corev1.PersistentVolumeClaim{
                {
                    ObjectMeta: metav1.ObjectMeta{
                        Name: "data",
                    },
                    Spec: corev1.PersistentVolumeClaimSpec{
                        AccessModes: []corev1.PersistentVolumeAccessMode{
                            corev1.ReadWriteOnce,
                        },
                        Resources: corev1.ResourceRequirements{
                            Requests: corev1.ResourceList{
                                corev1.ResourceStorage: resource.MustParse(postgresql.Spec.Storage.Size),
                            },
                        },
                        StorageClassName: &postgresql.Spec.Storage.Class,
                    },
                },
            },
        },
    }

    if err := controllerutil.SetControllerReference(postgresql, statefulSet, r.Scheme); err != nil {
        return fmt.Errorf("failed to set owner reference: %w", err)
    }

    return r.createOrUpdate(ctx, statefulSet)
}

// createRBAC 创建RBAC权限
func (r *PostgreSQLReconciler) createRBAC(ctx context.Context, postgresql *examplev1.PostgreSQL) error {
    // 创建角色
    role := &rbacv1.Role{
        ObjectMeta: metav1.ObjectMeta{
            Name:      fmt.Sprintf("%s-role", postgresql.Name),
            Namespace: postgresql.Namespace,
        },
        Rules: []rbacv1.PolicyRule{
            {
                APIGroups: []string{""},
                Resources: []string{"pods"},
                Verbs:     []string{"get", "list", "watch"},
            },
        },
    }

    if err := controllerutil.SetControllerReference(postgresql, role, r.Scheme); err != nil {
        return fmt.Errorf("failed to set owner reference: %w", err)
    }

    if err := r.createOrUpdate(ctx, role); err != nil {
        return fmt.Errorf("failed to create role: %w", err)
    }

    // 创建角色绑定
    roleBinding := &rbacv1.RoleBinding{
        ObjectMeta: metav1.ObjectMeta{
            Name:      fmt.Sprintf("%s-rolebinding", postgresql.Name),
            Namespace: postgresql.Namespace,
        },
        Subjects: []rbacv1.Subject{
            {
                Kind:      "ServiceAccount",
                Name:      "default",
                Namespace: postgresql.Namespace,
            },
        },
        RoleRef: rbacv1.RoleRef{
            Kind: "Role",
            Name: fmt.Sprintf("%s-role", postgresql.Name),
        },
    }

    if err := controllerutil.SetControllerReference(postgresql, roleBinding, r.Scheme); err != nil {
        return fmt.Errorf("failed to set owner reference: %w", err)
    }

    return r.createOrUpdate(ctx, roleBinding)
}

// createOrUpdate 创建或更新资源
func (r *PostgreSQLReconciler) createOrUpdate(ctx context.Context, obj client.Object) error {
    log := log.FromContext(ctx)
    
    // 尝试获取现有资源
    found := obj.DeepCopyObject().(client.Object)
    err := r.Get(ctx, types.NamespacedName{Name: obj.GetName(), Namespace: obj.GetNamespace()}, found)
    if err != nil {
        if apierrors.IsNotFound(err) {
            log.Info("Creating new resource", "name", obj.GetName())
            return r.Create(ctx, obj)
        }
        return fmt.Errorf("failed to get resource: %w", err)
    }

    // 如果存在则更新
    log.Info("Updating existing resource", "name", obj.GetName())
    obj.SetResourceVersion(found.GetResourceVersion())
    return r.Update(ctx, obj)
}

// updateStatus 更新状态信息
func (r *PostgreSQLReconciler) updateStatus(ctx context.Context, postgresql *examplev1.PostgreSQL) error {
    log := log.FromContext(ctx)
    
    // 获取当前状态
    statefulSet := &appsv1.StatefulSet{}
    err := r.Get(ctx, types.NamespacedName{Name: postgresql.Name, Namespace: postgresql.Namespace}, statefulSet)
    if err != nil {
        return fmt.Errorf("failed to get statefulset: %w", err)
    }

    // 更新状态
    postgresql.Status.Replicas = int32(statefulSet.Status.Replicas)
    postgresql.Status.ReadyReplicas = int32(statefulSet.Status.ReadyReplicas)
    postgresql.Status.Version = postgresql.Spec.Version
    
    if statefulSet.Status.ReadyReplicas == statefulSet.Status.Replicas {
        postgresql.Status.Phase = "Running"
    } else {
        postgresql.Status.Phase = "Pending"
    }

    // 保存状态
    return r.Status().Update(ctx, postgresql)
}

// generatePostgreSQLConfig 生成PostgreSQL配置
func (r *PostgreSQLReconciler) generatePostgreSQLConfig(postgresql *examplev1.PostgreSQL) string {
    config := fmt.Sprintf(`
# PostgreSQL configuration for %s
max_connections = %d
shared_buffers = %s
effective_cache_size = 4GB
work_mem = 4MB
maintenance_work_mem = 64MB
`, postgresql.Name, 
        postgresql.Spec.Configuration.MaxConnections,
        postgresql.Spec.Configuration.SharedBuffers)
    
    return config
}

// handleDelete 处理删除逻辑
func (r *PostgreSQLReconciler) handleDelete(ctx context.Context, postgresql *examplev1.PostgreSQL) (ctrl.Result, error) {
    log := log.FromContext(ctx)
    log.Info("Handling deletion", "name", postgresql.Name)

    // 清理相关资源
    if err := r.cleanupResources(ctx, postgresql); err != nil {
        return ctrl.Result{}, fmt.Errorf("failed to cleanup resources: %w", err)
    }

    // 移除Finalizer
    controllerutil.RemoveFinalizer(postgresql, "postgresql.example.com/finalizer")
    if err := r.Update(ctx, postgresql); err != nil {
        return ctrl.Result{}, fmt.Errorf("failed to remove finalizer: %w", err)
    }

    log.Info("Successfully deleted PostgreSQL", "name", postgresql.Name)
    return ctrl.Result{}, nil
}

// cleanupResources 清理相关资源
func (r *PostgreSQLReconciler) cleanupResources(ctx context.Context, postgresql *examplev1.PostgreSQL) error {
    // 删除StatefulSet
    statefulSet := &appsv1.StatefulSet{}
    if err := r.Get(ctx, types.NamespacedName{Name: postgresql.Name, Namespace: postgresql.Namespace}, statefulSet); err == nil {
        if err := r.Delete(ctx, statefulSet); err != nil {
            return fmt.Errorf("failed to delete statefulset: %w", err)
        }
    }

    // 删除服务
    service := &corev1.Service{}
    if err := r.Get(ctx, types.NamespacedName{Name: postgresql.Name, Namespace: postgresql.Namespace}, service); err == nil {
        if