深入理解 Kubernetes Operator Pattern

apiVersion: mysql.example.com/v1
kind: MySQLCluster
metadata:
  name: my-database
spec:
  replicas: 3
  version: "8.0"
  backup:
    schedule: "0 2 * * *"

# 需要手动管理的内容:
- 3 个 StatefulSet (master, data, client 节点)
- 6 个 Service (内部通信 + 外部访问)
- 3 个 ConfigMap (每种节点的配置)
- PVC 管理和扩容
- 节点发现和集群组建
- 滚动升级策略
- 数据备份和恢复
- 集群健康检查
- 故障节点替换

apiVersion: elasticsearch.k8s.elastic.co/v1
kind: Elasticsearch
metadata:
  name: my-cluster
spec:
  version: 8.11.0
  nodeSets:
  - name: default
    count: 3
    config:
      node.store.allow_mmap: false

# 你要告诉系统"怎么做"
kubectl create deployment nginx --image=nginx
kubectl scale deployment nginx --replicas=3
kubectl set image deployment nginx nginx=nginx:1.19
kubectl rollout restart deployment nginx

# 你只需要告诉系统"想要什么"
apiVersion: apps/v1
kind: Deployment
metadata:
  name: nginx
spec:
  replicas: 3
  template:
    spec:
      containers:
      - name: nginx
        image: nginx:1.19

// api/v1/redis_types.go
package v1

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

type RedisSpec struct {
    // Redis 版本
    Version string `json:"version"`
    
    // 主节点配置
    Master RedisMasterSpec `json:"master"`
    
    // 从节点配置
    Replicas int32 `json:"replicas"`
    
    // 持久化配置
    Persistence PersistenceSpec `json:"persistence,omitempty"`
    
    // 资源限制
    Resources ResourceRequirements `json:"resources,omitempty"`
}

type RedisMasterSpec struct {
    // 主节点副本数 (通常为 1)
    Replicas int32 `json:"replicas"`
}

type PersistenceSpec struct {
    // 是否启用 RDB
    EnableRDB bool `json:"enableRdb"`
    
    // RDB 保存策略
    RDBSchedule string `json:"rdbSchedule,omitempty"`
    
    // 是否启用 AOF
    EnableAOF bool `json:"enableAof"`
    
    // 存储大小
    Size string `json:"size"`
}

type RedisStatus struct {
    // 主节点状态
    MasterStatus string `json:"masterStatus"`
    
    // 从节点状态
    ReplicaStatus map[string]string `json:"replicaStatus"`
    
    // 集群健康状态
    Phase string `json:"phase"` // Pending, Running, Failed
    
    // 主节点 IP
    MasterIP string `json:"masterIP,omitempty"`
}

// +kubebuilder:object:root=true
// +kubebuilder:subresource:status
// +kubebuilder:printcolumn:name="Version",type=string,JSONPath=`.spec.version`
// +kubebuilder:printcolumn:name="Master",type=string,JSONPath=`.status.masterStatus`
// +kubebuilder:printcolumn:name="Replicas",type=integer,JSONPath=`.spec.replicas`
// +kubebuilder:printcolumn:name="Phase",type=string,JSONPath=`.status.phase`

type Redis struct {
    metav1.TypeMeta   `json:",inline"`
    metav1.ObjectMeta `json:"metadata,omitempty"`
    
    Spec   RedisSpec   `json:"spec,omitempty"`
    Status RedisStatus `json:"status,omitempty"`
}

// internal/controller/redis_controller.go
package controller

import (
    "context"
    "fmt"
    
    corev1 "k8s.io/api/core/v1"
    appsv1 "k8s.io/api/apps/v1"
    ctrl "sigs.k8s.io/controller-runtime"
    "sigs.k8s.io/controller-runtime/pkg/client"
    
    redisv1 "github.com/example/redis-operator/api/v1"
)

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

func (r *RedisReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
    log := log.FromContext(ctx)
    
    // 1. 获取 Redis 资源
    redis := &redisv1.Redis{}
    if err := r.Get(ctx, req.NamespacedName, redis); err != nil {
        return ctrl.Result{}, client.IgnoreNotFound(err)
    }
    
    // 2. 创建或更新 Master StatefulSet
    if err := r.reconcileMaster(ctx, redis); err != nil {
        log.Error(err, "Failed to reconcile master")
        return ctrl.Result{}, err
    }
    
    // 3. 创建或更新 Replica StatefulSet
    if err := r.reconcileReplicas(ctx, redis); err != nil {
        log.Error(err, "Failed to reconcile replicas")
        return ctrl.Result{}, err
    }
    
    // 4. 配置主从复制关系
    if err := r.configureReplication(ctx, redis); err != nil {
        log.Error(err, "Failed to configure replication")
        return ctrl.Result{}, err
    }
    
    // 5. 创建 Service
    if err := r.reconcileServices(ctx, redis); err != nil {
        log.Error(err, "Failed to reconcile services")
        return ctrl.Result{}, err
    }
    
    // 6. 更新状态
    if err := r.updateStatus(ctx, redis); err != nil {
        log.Error(err, "Failed to update status")
        return ctrl.Result{}, err
    }
    
    return ctrl.Result{}, nil
}

// 核心功能 1: 部署 Master 节点
func (r *RedisReconciler) reconcileMaster(ctx context.Context, redis *redisv1.Redis) error {
    masterSts := &appsv1.StatefulSet{
        ObjectMeta: metav1.ObjectMeta{
            Name:      fmt.Sprintf("%s-master", redis.Name),
            Namespace: redis.Namespace,
        },
        Spec: appsv1.StatefulSetSpec{
            Replicas: &redis.Spec.Master.Replicas,
            Selector: &metav1.LabelSelector{
                MatchLabels: map[string]string{
                    "app":  redis.Name,
                    "role": "master",
                },
            },
            Template: corev1.PodTemplateSpec{
                ObjectMeta: metav1.ObjectMeta{
                    Labels: map[string]string{
                        "app":  redis.Name,
                        "role": "master",
                    },
                },
                Spec: corev1.PodSpec{
                    Containers: []corev1.Container{{
                        Name:  "redis",
                        Image: fmt.Sprintf("redis:%s", redis.Spec.Version),
                        Ports: []corev1.ContainerPort{{
                            ContainerPort: 6379,
                            Name:         "redis",
                        }},
                        Command: []string{
                            "redis-server",
                            "--appendonly", "yes",
                        },
                        VolumeMounts: []corev1.VolumeMount{{
                            Name:      "data",
                            MountPath: "/data",
                        }},
                    }},
                },
            },
            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(redis.Spec.Persistence.Size),
                        },
                    },
                },
            }},
        },
    }
    
    ctrl.SetControllerReference(redis, masterSts, r.Scheme)
    return r.createOrUpdate(ctx, masterSts)
}

// 核心功能 2: 配置主从复制
func (r *RedisReconciler) configureReplication(ctx context.Context, redis *redisv1.Redis) error {
    // 获取 Master Pod IP
    masterPod := &corev1.Pod{}
    masterPodName := fmt.Sprintf("%s-master-0", redis.Name)
    if err := r.Get(ctx, client.ObjectKey{
        Name:      masterPodName,
        Namespace: redis.Namespace,
    }, masterPod); err != nil {
        return err
    }
    
    masterIP := masterPod.Status.PodIP
    if masterIP == "" {
        return fmt.Errorf("master IP not ready")
    }
    
    // 配置所有从节点
    for i := int32(0); i < redis.Spec.Replicas; i++ {
        replicaPodName := fmt.Sprintf("%s-replica-%d", redis.Name, i)
        
        // 在从节点上执行 REPLICAOF 命令
        cmd := []string{
            "redis-cli",
            "REPLICAOF",
            masterIP,
            "6379",
        }
        
        if err := r.execInPod(ctx, redis.Namespace, replicaPodName, cmd); err != nil {
            return fmt.Errorf("failed to configure replica %s: %w", replicaPodName, err)
        }
    }
    
    return nil
}

// 核心功能 3: 监控健康状态
func (r *RedisReconciler) updateStatus(ctx context.Context, redis *redisv1.Redis) error {
    // 检查 Master 状态
    masterReady := r.checkMasterHealth(ctx, redis)
    
    // 检查 Replica 状态
    replicaStatus := make(map[string]string)
    for i := int32(0); i < redis.Spec.Replicas; i++ {
        podName := fmt.Sprintf("%s-replica-%d", redis.Name, i)
        status := r.checkReplicaHealth(ctx, redis, podName)
        replicaStatus[podName] = status
    }
    
    // 更新 Status
    redis.Status.MasterStatus = masterReady
    redis.Status.ReplicaStatus = replicaStatus
    
    allReady := masterReady == "Ready"
    for _, status := range replicaStatus {
        if status != "Ready" {
            allReady = false
            break
        }
    }
    
    if allReady {
        redis.Status.Phase = "Running"
    } else {
        redis.Status.Phase = "Pending"
    }
    
    return r.Status().Update(ctx, redis)
}

// 核心功能 4: 故障自愈
func (r *RedisReconciler) handleMasterFailure(ctx context.Context, redis *redisv1.Redis) error {
    log := log.FromContext(ctx)
    log.Info("Master failure detected, initiating failover")
    
    // 1. 选择最佳从节点作为新主节点
    bestReplica, err := r.selectBestReplica(ctx, redis)
    if err != nil {
        return err
    }
    
    // 2. 提升从节点为主节点
    promoteCmd := []string{"redis-cli", "REPLICAOF", "NO", "ONE"}
    if err := r.execInPod(ctx, redis.Namespace, bestReplica, promoteCmd); err != nil {
        return err
    }
    
    // 3. 重新配置其他从节点
    newMasterIP, err := r.getPodIP(ctx, redis.Namespace, bestReplica)
    if err != nil {
        return err
    }
    
    for i := int32(0); i < redis.Spec.Replicas; i++ {
        replicaPod := fmt.Sprintf("%s-replica-%d", redis.Name, i)
        if replicaPod == bestReplica {
            continue
        }
        
        cmd := []string{"redis-cli", "REPLICAOF", newMasterIP, "6379"}
        if err := r.execInPod(ctx, redis.Namespace, replicaPod, cmd); err != nil {
            log.Error(err, "Failed to reconfigure replica", "pod", replicaPod)
        }
    }
    
    log.Info("Failover completed", "newMaster", bestReplica)
    return nil
}

apiVersion: redis.example.com/v1
kind: Redis
metadata:
  name: my-redis
  namespace: default
spec:
  version: "7.2"
  
  master:
    replicas: 1
  
  replicas: 2
  
  persistence:
    enableRdb: true
    rdbSchedule: "*/15 * * * *"  # 每 15 分钟备份
    enableAof: true
    size: "10Gi"
  
  resources:
    requests:
      memory: "256Mi"
      cpu: "100m"
    limits:
      memory: "512Mi"
      cpu: "500m"

# 部署集群
kubectl apply -f redis.yaml

# 查看状态
kubectl get redis
# NAME       VERSION   MASTER   REPLICAS   PHASE
# my-redis   7.2       Ready    2          Running

# 查看 Pod
kubectl get pods
# NAME                READY   STATUS    RESTARTS   AGE
# my-redis-master-0   1/1     Running   0          2m
# my-redis-replica-0  1/1     Running   0          2m
# my-redis-replica-1  1/1     Running   0          2m

# 测试故障转移
kubectl delete pod my-redis-master-0

# Operator 自动:
# 1. 检测到 Master 失败
# 2. 选择最佳 Replica
# 3. 提升为新 Master
# 4. 重新配置其他 Replica
# 5. 创建新的 Master Pod

apiVersion: planetscale.com/v2
kind: VitessCluster
metadata:
  name: prod-db
spec:
  # 自动分片
  keyspaces:
  - name: commerce
    partitionings:
    - equal:
        parts: 4
        shardTemplate:
          replication:
            enforceSemiSync: true
  
  # 自动备份
  backup:
    locations:
    - s3:
        bucket: my-backups
        region: us-east-1
    schedule: "0 2 * * *"
  
  # 性能监控
  monitoring:
    enabled: true
    prometheus:
      enabled: true
  
  # 自动扩缩容
  autoscaling:
    enabled: true
    minReplicas: 2
    maxReplicas: 10
    targetCPUUtilization: 70

apiVersion: acid.zalan.do/v1
kind: postgresql
metadata:
  name: prod-postgres
spec:
  teamId: "product"
  volume:
    size: 100Gi
  numberOfInstances: 3
  
  users:
    app_user: []
  
  databases:
    myapp: app_user
  
  # 高可用配置
  patroni:
    pg_hba:
    - hostssl all all 0.0.0.0/0 md5
  
  # 连接池
  connectionPooler:
    numberOfInstances: 2
    mode: transaction
  
  # 自动备份
  clone:
    cluster: "prod-postgres"

apiVersion: psmdb.percona.com/v1
kind: PerconaServerMongoDB
metadata:
  name: my-cluster
spec:
  crVersion: 1.15.0
  image: percona/percona-server-mongodb:6.0
  
  replsets:
  - name: rs0
    size: 3
    affinity:
      antiAffinityTopologyKey: "kubernetes.io/hostname"
  
  sharding:
    enabled: true
    configsvrReplSet:
      size: 3
    mongos:
      size: 2
  
  backup:
    enabled: true
    tasks:
    - name: daily-backup
      schedule: "0 2 * * *"
      keep: 7

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: prod-kafka
spec:
  kafka:
    version: 3.6.0
    replicas: 5
    
    # 自动分区重平衡
    cruiseControl: {}
    
    # 配置
    config:
      offsets.topic.replication.factor: 3
      transaction.state.log.replication.factor: 3
      min.insync.replicas: 2
    
    # 监控
    metricsConfig:
      type: jmxPrometheusExporter
  
  zookeeper:
    replicas: 3
  
  entityOperator:
    topicOperator: {}
    userOperator: {}

apiVersion: kubeflow.org/v1
kind: PyTorchJob
metadata:
  name: mnist-training
spec:
  pytorchReplicaSpecs:
    Master:
      replicas: 1
      template:
        spec:
          containers:
          - name: pytorch
            image: pytorch/pytorch:latest
            command:
            - python
            - train.py
            resources:
              limits:
                nvidia.com/gpu: 1
    
    Worker:
      replicas: 3
      template:
        spec:
          containers:
          - name: pytorch
            image: pytorch/pytorch:latest
            resources:
              limits:
                nvidia.com/gpu: 1

apiVersion: monitoring.coreos.com/v1
kind: Prometheus
metadata:
  name: main
spec:
  replicas: 2
  retention: 30d
  
  # 自动服务发现
  serviceMonitorSelector:
    matchLabels:
      team: frontend
  
  # 告警规则
  ruleSelector:
    matchLabels:
      prometheus: main
  
  # 存储
  storage:
    volumeClaimTemplate:
      spec:
        resources:
          requests:
            storage: 100Gi

// 1. 使用声明式 API
type DatabaseSpec struct {
    Version  string
    Replicas int32
    Backup   BackupPolicy
}

// 2. 实现幂等性
func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
    // 多次执行结果相同
    existing := &DatabaseDeployment{}
    if err := r.Get(ctx, req.NamespacedName, existing); err == nil {
        // 资源已存在,只更新差异
        return r.updateIfNeeded(ctx, existing)
    }
    // 资源不存在,创建新的
    return r.create(ctx, req.NamespacedName)
}

// 3. 状态管理
type DatabaseStatus struct {
    // 观察到的状态
    ObservedGeneration int64
    // 条件列表
    Conditions []metav1.Condition
    // 具体指标
    ReadyReplicas int32
}

// 4. 使用 Finalizer 清理资源
const dbFinalizer = "database.example.com/finalizer"

if !db.DeletionTimestamp.IsZero() {
    // 执行清理逻辑
    if err := r.cleanupExternalResources(ctx, db); err != nil {
        return ctrl.Result{}, err
    }
    // 移除 Finalizer
    controllerutil.RemoveFinalizer(db, dbFinalizer)
    return ctrl.Result{}, r.Update(ctx, db)
}

// 5. 错误处理和重试
if err := r.createResource(ctx, resource); err != nil {
    // 记录事件
    r.Recorder.Event(db, corev1.EventTypeWarning, "CreateFailed", err.Error())
    // 返回错误,自动重试(指数退避)
    return ctrl.Result{}, err
}

// ❌ 1. 命令式操作
func (r *Reconciler) ScaleUp(ctx context.Context) error {
    r.currentReplicas++ // 错误:基于当前状态修改
    return r.Update(ctx, r.deployment)
}

// ✅ 正确做法
func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) error {
    // 总是从期望状态出发
    deployment.Spec.Replicas = &db.Spec.Replicas
    return r.Update(ctx, deployment)
}

// ❌ 2. 在 Reconcile 中阻塞
func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) error {
    time.Sleep(30 * time.Second) // 错误:会阻塞队列
    return nil
}

// ✅ 正确做法
func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) error {
    if !isReady {
        // 返回 RequeueAfter,不阻塞
        return ctrl.Result{RequeueAfter: 30 * time.Second}, nil
    }
    return ctrl.Result{}, nil
}

// ❌ 3. 直接修改 Spec
func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) error {
    db.Spec.Replicas = 5 // 错误:不应修改用户输入
    return r.Update(ctx, db)
}

// ✅ 正确做法:只修改 Status
func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) error {
    db.Status.ReadyReplicas = actualReplicas
    return r.Status().Update(ctx, db)
}

// 1. 结构化日志
func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
    log := log.FromContext(ctx).WithValues(
        "database", req.NamespacedName,
        "generation", db.Generation,
    )
    
    log.Info("Reconciling database")
    log.Error(err, "Failed to create deployment", "deployment", deploymentName)
    
    return ctrl.Result{}, nil
}

// 2. Prometheus 指标
var (
    reconcileCount = prometheus.NewCounterVec(
        prometheus.CounterOpts{
            Name: "database_reconcile_total",
            Help: "Total number of reconciliations",
        },
        []string{"database", "result"},
    )
    
    reconcileDuration = prometheus.NewHistogramVec(
        prometheus.HistogramOpts{
            Name: "database_reconcile_duration_seconds",
            Help: "Duration of reconciliation",
        },
        []string{"database"},
    )
)

func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
    start := time.Now()
    defer func() {
        reconcileDuration.WithLabelValues(req.Name).Observe(time.Since(start).Seconds())
    }()
    
    result, err := r.reconcile(ctx, req)
    
    status := "success"
    if err != nil {
        status = "error"
    }
    reconcileCount.WithLabelValues(req.Name, status).Inc()
    
    return result, err
}

// 3. 事件记录
r.Recorder.Event(db, corev1.EventTypeNormal, "Created", "Database created successfully")
r.Recorder.Event(db, corev1.EventTypeWarning, "BackupFailed", "Failed to create backup")

// 4. Condition 状态
meta.SetStatusCondition(&db.Status.Conditions, metav1.Condition{
    Type:               "Ready",
    Status:             metav1.ConditionTrue,
    ObservedGeneration: db.Generation,
    Reason:             "AllReplicasReady",
    Message:            "All replicas are ready",
})

// 1. 使用 Predicate 过滤事件
func (r *Reconciler) SetupWithManager(mgr ctrl.Manager) error {
    return ctrl.NewControllerManagedBy(mgr).
        For(&v1.Database{}).
        Owns(&appsv1.Deployment{}).
        WithEventFilter(predicate.Funcs{
            UpdateFunc: func(e event.UpdateEvent) bool {
                // 只在 Generation 变化时触发
                oldGen := e.ObjectOld.GetGeneration()
                newGen := e.ObjectNew.GetGeneration()
                return oldGen != newGen
            },
        }).
        Complete(r)
}

// 2. 批量处理
func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
    // 获取所有相关资源
    var databases v1.DatabaseList
    if err := r.List(ctx, &databases, client.InNamespace(req.Namespace)); err != nil {
        return ctrl.Result{}, err
    }
    
    // 批量处理
    for _, db := range databases.Items {
        if err := r.reconcileOne(ctx, &db); err != nil {
            // 记录错误但继续处理其他资源
            log.Error(err, "Failed to reconcile", "database", db.Name)
        }
    }
    
    return ctrl.Result{}, nil
}

// 3. 限流
func (r *Reconciler) SetupWithManager(mgr ctrl.Manager) error {
    return ctrl.NewControllerManagedBy(mgr).
        For(&v1.Database{}).
        WithOptions(controller.Options{
            MaxConcurrentReconciles: 3,
            RateLimiter: workqueue.NewItemExponentialFailureRateLimiter(
                1*time.Second,
                60*time.Second,
            ),
        }).
        Complete(r)
}

// 1. 单元测试
func TestReconcile(t *testing.T) {
    // 创建 fake client
    scheme := runtime.NewScheme()
    _ = v1.AddToScheme(scheme)
    _ = appsv1.AddToScheme(scheme)
    
    client := fake.NewClientBuilder().
        WithScheme(scheme).
        WithObjects(&v1.Database{
            ObjectMeta: metav1.ObjectMeta{
                Name:      "test-db",
                Namespace: "default",
            },
            Spec: v1.DatabaseSpec{
                Replicas: 3,
            },
        }).
        Build()
    
    reconciler := &Reconciler{
        Client: client,
        Scheme: scheme,
    }
    
    // 执行 Reconcile
    result, err := reconciler.Reconcile(context.Background(), ctrl.Request{
        NamespacedName: types.NamespacedName{
            Name:      "test-db",
            Namespace: "default",
        },
    })
    
    // 断言
    assert.NoError(t, err)
    assert.False(t, result.Requeue)
    
    // 验证创建的资源
    var deployment appsv1.Deployment
    err = client.Get(context.Background(), types.NamespacedName{
        Name:      "test-db",
        Namespace: "default",
    }, &deployment)
    assert.NoError(t, err)
    assert.Equal(t, int32(3), *deployment.Spec.Replicas)
}

// 2. 集成测试(使用 envtest)
func TestReconcileIntegration(t *testing.T) {
    testEnv := &envtest.Environment{
        CRDDirectoryPaths: []string{filepath.Join("..", "config", "crd", "bases")},
    }
    
    cfg, err := testEnv.Start()
    defer testEnv.Stop()
    
    // 创建真实 client
    k8sClient, err := client.New(cfg, client.Options{Scheme: scheme})
    
    // 运行测试...
}

// 3. E2E 测试
func TestDatabaseE2E(t *testing.T) {
    // 部署 Operator
    kubectl("apply", "-f", "config/deploy/operator.yaml")
    
    // 创建 Database
    db := &v1.Database{
        ObjectMeta: metav1.ObjectMeta{Name: "test"},
        Spec:       v1.DatabaseSpec{Replicas: 3},
    }
    k8sClient.Create(context.Background(), db)
    
    // 等待就绪
    Eventually(func() bool {
        k8sClient.Get(context.Background(), client.ObjectKeyFromObject(db), db)
        return db.Status.Phase == "Running"
    }, timeout, interval).Should(BeTrue())
    
    // 测试扩容
    db.Spec.Replicas = 5
    k8sClient.Update(context.Background(), db)
    
    // 验证扩容
    Eventually(func() int32 {
        k8sClient.Get(context.Background(), client.ObjectKeyFromObject(db), db)
        return db.Status.ReadyReplicas
    }, timeout, interval).Should(Equal(int32(5)))
}

// 1. RBAC 最小权限
// +kubebuilder:rbac:groups=database.example.com,resources=databases,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=database.example.com,resources=databases/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=apps,resources=statefulsets,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups="",resources=secrets,verbs=get;list;watch

// 2. 敏感信息管理
func (r *Reconciler) getPassword(ctx context.Context, db *v1.Database) (string, error) {
    secret := &corev1.Secret{}
    err := r.Get(ctx, types.NamespacedName{
        Name:      db.Spec.PasswordSecret,
        Namespace: db.Namespace,
    }, secret)
    if err != nil {
        return "", err
    }
    
    password := secret.Data["password"]
    if len(password) == 0 {
        return "", fmt.Errorf("password not found in secret")
    }
    
    return string(password), nil
}

// 3. 输入验证
// +kubebuilder:validation:Minimum=1
// +kubebuilder:validation:Maximum=100
type DatabaseSpec struct {
    Replicas int32 `json:"replicas"`
    
    // +kubebuilder:validation:Pattern=`^[0-9]+\.[0-9]+# 深入理解 Kubernetes Operator Pattern

## 目录
- [什么是 Operator Pattern](#什么是-operator-pattern)
- [为什么需要 Operator](#为什么需要-operator)
- [核心原理](#核心原理)
- [实战案例](#实战案例)
- [企业级应用](#企业级应用)
- [最佳实践](#最佳实践)

---

## 什么是 Operator Pattern

Operator Pattern 是 Kubernetes 的一种扩展模式,它将**人类运维经验**编码成软件,让应用能够**自我管理**。

### 形象比喻

想象你雇了一个专业的数据库管理员(DBA):

**传统方式 (人工运维)**:

# 需要手动管理的内容:
- 3 个 StatefulSet (master, data, client 节点)
- 6 个 Service (内部通信 + 外部访问)
- 3 个 ConfigMap (每种节点的配置)
- PVC 管理和扩容
- 节点发现和集群组建
- 滚动升级策略
- 数据备份和恢复
- 集群健康检查
- 故障节点替换

apiVersion: elasticsearch.k8s.elastic.co/v1
kind: Elasticsearch
metadata:
  name: my-cluster
spec:
  version: 8.11.0
  nodeSets:
  - name: default
    count: 3
    config:
      node.store.allow_mmap: false

# 你要告诉系统"怎么做"
kubectl create deployment nginx --image=nginx
kubectl scale deployment nginx --replicas=3
kubectl set image deployment nginx nginx=nginx:1.19
kubectl rollout restart deployment nginx

# 你只需要告诉系统"想要什么"
apiVersion: apps/v1
kind: Deployment
metadata:
  name: nginx
spec:
  replicas: 3
  template:
    spec:
      containers:
      - name: nginx
        image: nginx:1.19

// api/v1/redis_types.go
package v1

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

type RedisSpec struct {
    // Redis 版本
    Version string `json:"version"`
    
    // 主节点配置
    Master RedisMasterSpec `json:"master"`
    
    // 从节点配置
    Replicas int32 `json:"replicas"`
    
    // 持久化配置
    Persistence PersistenceSpec `json:"persistence,omitempty"`
    
    // 资源限制
    Resources ResourceRequirements `json:"resources,omitempty"`
}

type RedisMasterSpec struct {
    // 主节点副本数 (通常为 1)
    Replicas int32 `json:"replicas"`
}

type PersistenceSpec struct {
    // 是否启用 RDB
    EnableRDB bool `json:"enableRdb"`
    
    // RDB 保存策略
    RDBSchedule string `json:"rdbSchedule,omitempty"`
    
    // 是否启用 AOF
    EnableAOF bool `json:"enableAof"`
    
    // 存储大小
    Size string `json:"size"`
}

type RedisStatus struct {
    // 主节点状态
    MasterStatus string `json:"masterStatus"`
    
    // 从节点状态
    ReplicaStatus map[string]string `json:"replicaStatus"`
    
    // 集群健康状态
    Phase string `json:"phase"` // Pending, Running, Failed
    
    // 主节点 IP
    MasterIP string `json:"masterIP,omitempty"`
}

// +kubebuilder:object:root=true
// +kubebuilder:subresource:status
// +kubebuilder:printcolumn:name="Version",type=string,JSONPath=`.spec.version`
// +kubebuilder:printcolumn:name="Master",type=string,JSONPath=`.status.masterStatus`
// +kubebuilder:printcolumn:name="Replicas",type=integer,JSONPath=`.spec.replicas`
// +kubebuilder:printcolumn:name="Phase",type=string,JSONPath=`.status.phase`

type Redis struct {
    metav1.TypeMeta   `json:",inline"`
    metav1.ObjectMeta `json:"metadata,omitempty"`
    
    Spec   RedisSpec   `json:"spec,omitempty"`
    Status RedisStatus `json:"status,omitempty"`
}

// internal/controller/redis_controller.go
package controller

import (
    "context"
    "fmt"
    
    corev1 "k8s.io/api/core/v1"
    appsv1 "k8s.io/api/apps/v1"
    ctrl "sigs.k8s.io/controller-runtime"
    "sigs.k8s.io/controller-runtime/pkg/client"
    
    redisv1 "github.com/example/redis-operator/api/v1"
)

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

func (r *RedisReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
    log := log.FromContext(ctx)
    
    // 1. 获取 Redis 资源
    redis := &redisv1.Redis{}
    if err := r.Get(ctx, req.NamespacedName, redis); err != nil {
        return ctrl.Result{}, client.IgnoreNotFound(err)
    }
    
    // 2. 创建或更新 Master StatefulSet
    if err := r.reconcileMaster(ctx, redis); err != nil {
        log.Error(err, "Failed to reconcile master")
        return ctrl.Result{}, err
    }
    
    // 3. 创建或更新 Replica StatefulSet
    if err := r.reconcileReplicas(ctx, redis); err != nil {
        log.Error(err, "Failed to reconcile replicas")
        return ctrl.Result{}, err
    }
    
    // 4. 配置主从复制关系
    if err := r.configureReplication(ctx, redis); err != nil {
        log.Error(err, "Failed to configure replication")
        return ctrl.Result{}, err
    }
    
    // 5. 创建 Service
    if err := r.reconcileServices(ctx, redis); err != nil {
        log.Error(err, "Failed to reconcile services")
        return ctrl.Result{}, err
    }
    
    // 6. 更新状态
    if err := r.updateStatus(ctx, redis); err != nil {
        log.Error(err, "Failed to update status")
        return ctrl.Result{}, err
    }
    
    return ctrl.Result{}, nil
}

// 核心功能 1: 部署 Master 节点
func (r *RedisReconciler) reconcileMaster(ctx context.Context, redis *redisv1.Redis) error {
    masterSts := &appsv1.StatefulSet{
        ObjectMeta: metav1.ObjectMeta{
            Name:      fmt.Sprintf("%s-master", redis.Name),
            Namespace: redis.Namespace,
        },
        Spec: appsv1.StatefulSetSpec{
            Replicas: &redis.Spec.Master.Replicas,
            Selector: &metav1.LabelSelector{
                MatchLabels: map[string]string{
                    "app":  redis.Name,
                    "role": "master",
                },
            },
            Template: corev1.PodTemplateSpec{
                ObjectMeta: metav1.ObjectMeta{
                    Labels: map[string]string{
                        "app":  redis.Name,
                        "role": "master",
                    },
                },
                Spec: corev1.PodSpec{
                    Containers: []corev1.Container{{
                        Name:  "redis",
                        Image: fmt.Sprintf("redis:%s", redis.Spec.Version),
                        Ports: []corev1.ContainerPort{{
                            ContainerPort: 6379,
                            Name:         "redis",
                        }},
                        Command: []string{
                            "redis-server",
                            "--appendonly", "yes",
                        },
                        VolumeMounts: []corev1.VolumeMount{{
                            Name:      "data",
                            MountPath: "/data",
                        }},
                    }},
                },
            },
            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(redis.Spec.Persistence.Size),
                        },
                    },
                },
            }},
        },
    }
    
    ctrl.SetControllerReference(redis, masterSts, r.Scheme)
    return r.createOrUpdate(ctx, masterSts)
}

// 核心功能 2: 配置主从复制
func (r *RedisReconciler) configureReplication(ctx context.Context, redis *redisv1.Redis) error {
    // 获取 Master Pod IP
    masterPod := &corev1.Pod{}
    masterPodName := fmt.Sprintf("%s-master-0", redis.Name)
    if err := r.Get(ctx, client.ObjectKey{
        Name:      masterPodName,
        Namespace: redis.Namespace,
    }, masterPod); err != nil {
        return err
    }
    
    masterIP := masterPod.Status.PodIP
    if masterIP == "" {
        return fmt.Errorf("master IP not ready")
    }
    
    // 配置所有从节点
    for i := int32(0); i < redis.Spec.Replicas; i++ {
        replicaPodName := fmt.Sprintf("%s-replica-%d", redis.Name, i)
        
        // 在从节点上执行 REPLICAOF 命令
        cmd := []string{
            "redis-cli",
            "REPLICAOF",
            masterIP,
            "6379",
        }
        
        if err := r.execInPod(ctx, redis.Namespace, replicaPodName, cmd); err != nil {
            return fmt.Errorf("failed to configure replica %s: %w", replicaPodName, err)
        }
    }
    
    return nil
}

// 核心功能 3: 监控健康状态
func (r *RedisReconciler) updateStatus(ctx context.Context, redis *redisv1.Redis) error {
    // 检查 Master 状态
    masterReady := r.checkMasterHealth(ctx, redis)
    
    // 检查 Replica 状态
    replicaStatus := make(map[string]string)
    for i := int32(0); i < redis.Spec.Replicas; i++ {
        podName := fmt.Sprintf("%s-replica-%d", redis.Name, i)
        status := r.checkReplicaHealth(ctx, redis, podName)
        replicaStatus[podName] = status
    }
    
    // 更新 Status
    redis.Status.MasterStatus = masterReady
    redis.Status.ReplicaStatus = replicaStatus
    
    allReady := masterReady == "Ready"
    for _, status := range replicaStatus {
        if status != "Ready" {
            allReady = false
            break
        }
    }
    
    if allReady {
        redis.Status.Phase = "Running"
    } else {
        redis.Status.Phase = "Pending"
    }
    
    return r.Status().Update(ctx, redis)
}

// 核心功能 4: 故障自愈
func (r *RedisReconciler) handleMasterFailure(ctx context.Context, redis *redisv1.Redis) error {
    log := log.FromContext(ctx)
    log.Info("Master failure detected, initiating failover")
    
    // 1. 选择最佳从节点作为新主节点
    bestReplica, err := r.selectBestReplica(ctx, redis)
    if err != nil {
        return err
    }
    
    // 2. 提升从节点为主节点
    promoteCmd := []string{"redis-cli", "REPLICAOF", "NO", "ONE"}
    if err := r.execInPod(ctx, redis.Namespace, bestReplica, promoteCmd); err != nil {
        return err
    }
    
    // 3. 重新配置其他从节点
    newMasterIP, err := r.getPodIP(ctx, redis.Namespace, bestReplica)
    if err != nil {
        return err
    }
    
    for i := int32(0); i < redis.Spec.Replicas; i++ {
        replicaPod := fmt.Sprintf("%s-replica-%d", redis.Name, i)
        if replicaPod == bestReplica {
            continue
        }
        
        cmd := []string{"redis-cli", "REPLICAOF", newMasterIP, "6379"}
        if err := r.execInPod(ctx, redis.Namespace, replicaPod, cmd); err != nil {
            log.Error(err, "Failed to reconfigure replica", "pod", replicaPod)
        }
    }
    
    log.Info("Failover completed", "newMaster", bestReplica)
    return nil
}

apiVersion: redis.example.com/v1
kind: Redis
metadata:
  name: my-redis
  namespace: default
spec:
  version: "7.2"
  
  master:
    replicas: 1
  
  replicas: 2
  
  persistence:
    enableRdb: true
    rdbSchedule: "*/15 * * * *"  # 每 15 分钟备份
    enableAof: true
    size: "10Gi"
  
  resources:
    requests:
      memory: "256Mi"
      cpu: "100m"
    limits:
      memory: "512Mi"
      cpu: "500m"

# 部署集群
kubectl apply -f redis.yaml

# 查看状态
kubectl get redis
# NAME       VERSION   MASTER   REPLICAS   PHASE
# my-redis   7.2       Ready    2          Running

# 查看 Pod
kubectl get pods
# NAME                READY   STATUS    RESTARTS   AGE
# my-redis-master-0   1/1     Running   0          2m
# my-redis-replica-0  1/1     Running   0          2m
# my-redis-replica-1  1/1     Running   0          2m

# 测试故障转移
kubectl delete pod my-redis-master-0

# Operator 自动:
# 1. 检测到 Master 失败
# 2. 选择最佳 Replica
# 3. 提升为新 Master
# 4. 重新配置其他 Replica
# 5. 创建新的 Master Pod

apiVersion: planetscale.com/v2
kind: VitessCluster
metadata:
  name: prod-db
spec:
  # 自动分片
  keyspaces:
  - name: commerce
    partitionings:
    - equal:
        parts: 4
        shardTemplate:
          replication:
            enforceSemiSync: true
  
  # 自动备份
  backup:
    locations:
    - s3:
        bucket: my-backups
        region: us-east-1
    schedule: "0 2 * * *"
  
  # 性能监控
  monitoring:
    enabled: true
    prometheus:
      enabled: true
  
  # 自动扩缩容
  autoscaling:
    enabled: true
    minReplicas: 2
    maxReplicas: 10
    targetCPUUtilization: 70

apiVersion: acid.zalan.do/v1
kind: postgresql
metadata:
  name: prod-postgres
spec:
  teamId: "product"
  volume:
    size: 100Gi
  numberOfInstances: 3
  
  users:
    app_user: []
  
  databases:
    myapp: app_user
  
  # 高可用配置
  patroni:
    pg_hba:
    - hostssl all all 0.0.0.0/0 md5
  
  # 连接池
  connectionPooler:
    numberOfInstances: 2
    mode: transaction
  
  # 自动备份
  clone:
    cluster: "prod-postgres"

apiVersion: psmdb.percona.com/v1
kind: PerconaServerMongoDB
metadata:
  name: my-cluster
spec:
  crVersion: 1.15.0
  image: percona/percona-server-mongodb:6.0
  
  replsets:
  - name: rs0
    size: 3
    affinity:
      antiAffinityTopologyKey: "kubernetes.io/hostname"
  
  sharding:
    enabled: true
    configsvrReplSet:
      size: 3
    mongos:
      size: 2
  
  backup:
    enabled: true
    tasks:
    - name: daily-backup
      schedule: "0 2 * * *"
      keep: 7

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: prod-kafka
spec:
  kafka:
    version: 3.6.0
    replicas: 5
    
    # 自动分区重平衡
    cruiseControl: {}
    
    # 配置
    config:
      offsets.topic.replication.factor: 3
      transaction.state.log.replication.factor: 3
      min.insync.replicas: 2
    
    # 监控
    metricsConfig:
      type: jmxPrometheusExporter
  
  zookeeper:
    replicas: 3
  
  entityOperator:
    topicOperator: {}
    userOperator: {}

apiVersion: kubeflow.org/v1
kind: PyTorchJob
metadata:
  name: mnist-training
spec:
  pytorchReplicaSpecs:
    Master:
      replicas: 1
      template:
        spec:
          containers:
          - name: pytorch
            image: pytorch/pytorch:latest
            command:
            - python
            - train.py
            resources:
              limits:
                nvidia.com/gpu: 1
    
    Worker:
      replicas: 3
      template:
        spec:
          containers:
          - name: pytorch
            image: pytorch/pytorch:latest
            resources:
              limits:
                nvidia.com/gpu: 1

apiVersion: monitoring.coreos.com/v1
kind: Prometheus
metadata:
  name: main
spec:
  replicas: 2
  retention: 30d
  
  # 自动服务发现
  serviceMonitorSelector:
    matchLabels:
      team: frontend
  
  # 告警规则
  ruleSelector:
    matchLabels:
      prometheus: main
  
  # 存储
  storage:
    volumeClaimTemplate:
      spec:
        resources:
          requests:
            storage: 100Gi

// 1. 使用声明式 API
type DatabaseSpec struct {
    Version  string
    Replicas int32
    Backup   BackupPolicy
}


    Version string `json:"version"`
    
    // +kubebuilder:validation:Enum=mysql;postgres;mongodb
    Engine string `json:"engine"`
}

// 4. Webhook 验证
func (r *Database) ValidateCreate() error {
    if r.Spec.Replicas%2 == 0 {
        return fmt.Errorf("replicas must be odd number for quorum")
    }
    return nil
}

func (r *Database) ValidateUpdate(old runtime.Object) error {
    oldDB := old.(*Database)
    
    // 不允许更改引擎类型
    if r.Spec.Engine != oldDB.Spec.Engine {
        return fmt.Errorf("cannot change database engine")
    }
    
    // 不允许直接缩容超过 50%
    if r.Spec.Replicas < oldDB.Spec.Replicas/2 {
        return fmt.Errorf("cannot scale down more than 50%% at once")
    }
    
    return nil
}

apiVersion: redis.example.com/v1
kind: RedisCluster
metadata:
  name: order-cache
spec:
  # 版本管理
  version: "7.0"
  
  # 拓扑配置
  master:
    replicas: 3  # 3 个主节点
  replicas: 2    # 每个主节点 2 个从节点
  
  # 自动扩缩容
  autoscaling:
    enabled: true
    minReplicas: 2
    maxReplicas: 5
    targetMemoryUtilization: 70
  
  # 高可用配置
  sentinel:
    enabled: true
    replicas: 3
    downAfterMilliseconds: 5000
    failoverTimeout: 10000
  
  # 备份策略
  backup:
    schedule: "0 */6 * * *"  # 每 6 小时备份
    retention: 7              # 保留 7 天
    storage:
      s3:
        bucket: redis-backups
        region: us-east-1
  
  # 监控告警
  monitoring:
    enabled: true
    alerts:
    - name: HighMemoryUsage
      threshold: 80
      duration: 5m
    - name: ReplicationLag
      threshold: 10s

# 安装 operator-sdk
operator-sdk version

# 评分 Operator
operator-sdk scorecard ./config/manifests

# 输出示例:
# ✅ Basic Tests: 10/10
# ✅ OLM Tests: 8/8
# ⚠️  Custom Tests: 3/5
# Overall Score: 21/23 (91%)