Operator 概念

# 定义新的资源类型
apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
metadata:
  name: databases.example.com
spec:
  group: example.com
  names:
    kind: Database
    listKind: DatabaseList
    plural: databases
    singular: database
    shortNames:
      - db
  scope: Namespaced
  versions:
    - name: v1
      served: true
      storage: true
      schema:
        openAPIV3Schema:
          type: object
          properties:
            spec:
              type: object
              properties:
                engine:
                  type: string
                  enum: ["mysql", "postgresql", "mongodb"]
                version:
                  type: string
                replicas:
                  type: integer
                  minimum: 1
                  maximum: 10
                storage:
                  type: string
                  pattern: "^[0-9]+Gi$"
              required:
                - engine
                - version
            status:
              type: object
              properties:
                phase:
                  type: string
                readyReplicas:
                  type: integer
                conditions:
                  type: array
                  items:
                    type: object
                    properties:
                      type:
                        type: string
                      status:
                        type: string
                      reason:
                        type: string
                      message:
                        type: string
                      lastTransitionTime:
                        type: string
                        format: date-time
      subresources:
        status: {}
        scale:
          specReplicasPath: .spec.replicas
          statusReplicasPath: .status.readyReplicas
      additionalPrinterColumns:
        - name: Engine
          type: string
          jsonPath: .spec.engine
        - name: Version
          type: string
          jsonPath: .spec.version
        - name: Replicas
          type: integer
          jsonPath: .spec.replicas
        - name: Status
          type: string
          jsonPath: .status.phase
        - name: Age
          type: date
          jsonPath: .metadata.creationTimestamp

# 使用自定义资源
apiVersion: example.com/v1
kind: Database
metadata:
  name: my-database
  namespace: production
spec:
  engine: postgresql
  version: "14.5"
  replicas: 3
  storage: 100Gi

// 控制器核心逻辑
type DatabaseReconciler struct {
    client.Client
    Scheme *runtime.Scheme
}

func (r *DatabaseReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
    log := log.FromContext(ctx)

    // 1. 获取 CR
    var database examplev1.Database
    if err := r.Get(ctx, req.NamespacedName, &database); err != nil {
        return ctrl.Result{}, client.IgnoreNotFound(err)
    }

    // 2. 处理删除
    if !database.DeletionTimestamp.IsZero() {
        return r.reconcileDelete(ctx, &database)
    }

    // 3. 确保 Finalizer
    if err := r.ensureFinalizer(ctx, &database); err != nil {
        return ctrl.Result{}, err
    }

    // 4. 协调子资源
    if err := r.reconcileStatefulSet(ctx, &database); err != nil {
        return ctrl.Result{}, err
    }

    if err := r.reconcileService(ctx, &database); err != nil {
        return ctrl.Result{}, err
    }

    // 5. 更新状态
    if err := r.updateStatus(ctx, &database); err != nil {
        return ctrl.Result{}, err
    }

    return ctrl.Result{}, nil
}

// Mutating Webhook - 设置默认值
func (r *Database) Default() {
    if r.Spec.Replicas == 0 {
        r.Spec.Replicas = 1
    }
    if r.Spec.Storage == "" {
        r.Spec.Storage = "10Gi"
    }
}

// Validating Webhook - 验证
func (r *Database) ValidateCreate() (admission.Warnings, error) {
    return r.validate()
}

func (r *Database) ValidateUpdate(old runtime.Object) (admission.Warnings, error) {
    oldDB := old.(*Database)

    // 不允许修改引擎类型
    if r.Spec.Engine != oldDB.Spec.Engine {
        return nil, fmt.Errorf("engine type is immutable")
    }

    return r.validate()
}

func (r *Database) validate() (admission.Warnings, error) {
    var warnings admission.Warnings

    // 验证副本数
    if r.Spec.Replicas > 10 {
        return warnings, fmt.Errorf("replicas cannot exceed 10")
    }

    return warnings, nil
}

// 1. 幂等性 - 多次执行结果相同
func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
    // 使用 CreateOrUpdate 确保幂等
    _, err := controllerutil.CreateOrUpdate(ctx, r.Client, resource, func() error {
        // 设置期望状态
        return nil
    })
    return ctrl.Result{}, err
}

// 2. 所有者引用 - 级联删除
func (r *Reconciler) setOwnerReference(owner, owned metav1.Object) error {
    return controllerutil.SetControllerReference(owner, owned, r.Scheme)
}

// 3. 状态条件 - 清晰的状态报告
func setCondition(conditions *[]metav1.Condition, condType string, status metav1.ConditionStatus, reason, message string) {
    meta.SetStatusCondition(conditions, metav1.Condition{
        Type:               condType,
        Status:             status,
        Reason:             reason,
        Message:            message,
        LastTransitionTime: metav1.Now(),
    })
}

// 4. Finalizer - 清理外部资源
const finalizerName = "database.example.com/finalizer"

func (r *Reconciler) reconcileDelete(ctx context.Context, db *examplev1.Database) (ctrl.Result, error) {
    if controllerutil.ContainsFinalizer(db, finalizerName) {
        // 清理外部资源
        if err := r.cleanupExternalResources(ctx, db); err != nil {
            return ctrl.Result{}, err
        }

        // 移除 Finalizer
        controllerutil.RemoveFinalizer(db, finalizerName)
        if err := r.Update(ctx, db); err != nil {
            return ctrl.Result{}, err
        }
    }
    return ctrl.Result{}, nil
}

func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
    // 临时错误 - 重试
    if err := r.doSomething(); err != nil {
        if isTemporary(err) {
            return ctrl.Result{RequeueAfter: 30 * time.Second}, nil
        }
        return ctrl.Result{}, err
    }

    // 需要等待外部条件
    if !r.isReady() {
        return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
    }

    // 成功
    return ctrl.Result{}, nil
}