🤖 第18课：Operator开发

一、Operator模式深度理解

Operator = CRD + 自定义控制器，将人类运维知识编码为软件。通过声明式API描述期望状态，控制器持续调谐实现自动化运维。

┌──────── Operator工作原理 ────────┐
│                                    │
│  1. 用户创建 WebApp CR            │
│     spec: {image: "nginx",        │
│            replicas: 3}           │
│                                    │
│  2. 控制器Watch到事件              │
│     → 收到WebApp Create事件        │
│                                    │
│  3. 调谐（Reconcile）              │
│     → 检查当前状态                 │
│     → 创建Deployment(replicas=3)  │
│     → 创建Service                 │
│     → 创建ConfigMap               │
│     → 更新Status                  │
│                                    │
│  4. 持续监控                       │
│     → Deployment变化→重新调谐     │
│     → CR变化→重新调谐             │
│     → 定期Resync→重新调谐         │
│                                    │
│  ┌─── 调谐循环 ────────────────┐   │
│  │  Observe → Diff → Act      │   │
│  │  观察当前 → 比较差异 → 执行 │   │
│  └─────────────────────────────┘   │
└────────────────────────────────────┘

二、完整Operator开发实战

2.1 项目初始化

# 创建Operator项目
mkdir webapp-operator && cd webapp-operator
kubebuilder init \
  --domain example.com \
  --repo example.com/webapp-operator \
  --license apache2 \
  --owner "MyTeam"

# 创建API（含控制器）
kubebuilder create api \
  --group myapp \
  --version v1 \
  --kind WebApp \
  --controller \
  --resource

# 项目结构
tree .
# ├── api/v1/
# │   ├── webapp_types.go        # CRD类型定义
# │   ├── groupversion_info.go
# │   └── zz_generated.deepcopy.go
# ├── internal/controller/
# │   ├── webapp_controller.go   # 控制器逻辑
# │   └── suite_test.go
# ├── config/
# │   ├── crd/                   # CRD YAML
# │   ├── rbac/                  # RBAC配置
# │   ├── manager/               # 控制器部署
# │   └── samples/               # CR示例
# ├── Dockerfile
# ├── Makefile
# └── go.mod

2.2 定义CRD类型

// api/v1/webapp_types.go
package v1

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

// WebAppSpec 定义期望状态
type WebAppSpec struct {
    // +kubebuilder:validation:Required
    Image string `json:"image"`

    // +kubebuilder:default=1
    // +kubebuilder:validation:Minimum=1
    // +kubebuilder:validation:Maximum=100
    Replicas *int32 `json:"replicas,omitempty"`

    // +kubebuilder:default=8080
    Port int32 `json:"port,omitempty"`

    // 应用配置
    Config map[string]string `json:"config,omitempty"`

    // 资源限制
    Resources *ResourceSpec `json:"resources,omitempty"`

    // Ingress配置
    Ingress *IngressSpec `json:"ingress,omitempty"`
}

type ResourceSpec struct {
    Requests *ResourceList `json:"requests,omitempty"`
    Limits   *ResourceList `json:"limits,omitempty"`
}

type ResourceList struct {
    CPU    string `json:"cpu,omitempty"`
    Memory string `json:"memory,omitempty"`
}

type IngressSpec struct {
    Enabled bool   `json:"enabled,omitempty"`
    Host    string `json:"host,omitempty"`
    TLS     bool   `json:"tls,omitempty"`
}

// WebAppStatus 定义观测状态
type WebAppStatus struct {
    Replicas      int32  `json:"replicas,omitempty"`
    ReadyReplicas int32  `json:"readyReplicas,omitempty"`
    Available     bool   `json:"available,omitempty"`
    LabelSelector string `json:"labelSelector,omitempty"`
    // +listType=map
    // +listMapKey=type
    Conditions []metav1.Condition `json:"conditions,omitempty"`
}

// +kubebuilder:object:root=true
// +kubebuilder:subresource:status
// +kubebuilder:subresource:scale:specReplicasPath=.spec.replicas,statusReplicasPath=.status.readyReplicas,labelSelectorPath=.status.labelSelector
type WebApp struct {
    metav1.TypeMeta   `json:",inline"`
    metav1.ObjectMeta `json:"metadata,omitempty"`
    Spec   WebAppSpec   `json:"spec,omitempty"`
    Status WebAppStatus `json:"status,omitempty"`
}

// +kubebuilder:object:root=true
type WebAppList struct {
    metav1.TypeMeta `json:",inline"`
    metav1.ListMeta `json:"metadata,omitempty"`
    Items           []WebApp `json:"items"`
}

func init() {
    SchemeBuilder.Register(&WebApp{}, &WebAppList{})
}

2.3 实现控制器

// internal/controller/webapp_controller.go
package controller

import (
    "context"
    "fmt"

    appsv1 "k8s.io/api/apps/v1"
    corev1 "k8s.io/api/core/v1"
    networkingv1 "k8s.io/api/networking/v1"
    "k8s.io/apimachinery/pkg/api/errors"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/apimachinery/pkg/types"
    "k8s.io/apimachinery/pkg/util/intstr"
    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"

    myappv1 "example.com/webapp-operator/api/v1"
)

const webAppFinalizer = "myapp.example.com/finalizer"

// WebAppReconciler 调谐器
type WebAppReconciler struct {
    client.Client
    Scheme *runtime.Scheme
}

// +kubebuilder:rbac:groups=myapp.example.com,resources=webapps,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=myapp.example.com,resources=webapps/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=apps,resources=deployments,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups="",resources=services;configmaps,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=networking.k8s.io,resources=ingresses,verbs=get;list;watch;create;update;patch;delete

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

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

    // 2. 处理Finalizer
    if webapp.DeletionTimestamp != nil {
        if controllerutil.ContainsFinalizer(&webapp, webAppFinalizer) {
            logger.Info("Cleaning up WebApp resources")
            controllerutil.RemoveFinalizer(&webapp, webAppFinalizer)
            if err := r.Update(ctx, &webapp); err != nil {
                return ctrl.Result{}, err
            }
        }
        return ctrl.Result{}, nil
    }
    if !controllerutil.ContainsFinalizer(&webapp, webAppFinalizer) {
        controllerutil.AddFinalizer(&webapp, webAppFinalizer)
        if err := r.Update(ctx, &webapp); err != nil {
            return ctrl.Result{}, err
        }
    }

    // 3. 创建或更新ConfigMap
    var cm corev1.ConfigMap
    cm.SetNamespace(webapp.Namespace)
    cm.SetName(fmt.Sprintf("%s-config", webapp.Name))
    controllerutil.SetControllerReference(&webapp, &cm, r.Scheme)
    
    desiredCM := r.buildConfigMap(&webapp)
    if err := r.Get(ctx, types.NamespacedName{Name: cm.Name, Namespace: cm.Namespace}, &cm); err != nil {
        if errors.IsNotFound(err) {
            logger.Info("Creating ConfigMap", "name", cm.Name)
            if err := r.Create(ctx, desiredCM); err != nil {
                return ctrl.Result{}, err
            }
        }
    } else {
        cm.Data = desiredCM.Data
        if err := r.Update(ctx, &cm); err != nil {
            return ctrl.Result{}, err
        }
    }

    // 4. 创建或更新Deployment
    deploy := r.buildDeployment(&webapp)
    if err := controllerutil.SetControllerReference(&webapp, deploy, r.Scheme); err != nil {
        return ctrl.Result{}, err
    }
    
    var existingDeploy appsv1.Deployment
    if err := r.Get(ctx, types.NamespacedName{Name: deploy.Name, Namespace: deploy.Namespace}, &existingDeploy); err != nil {
        if errors.IsNotFound(err) {
            logger.Info("Creating Deployment", "name", deploy.Name)
            if err := r.Create(ctx, deploy); err != nil {
                return ctrl.Result{}, err
            }
        }
    } else {
        existingDeploy.Spec = deploy.Spec
        if err := r.Update(ctx, &existingDeploy); err != nil {
            return ctrl.Result{}, err
        }
    }

    // 5. 创建或更新Service
    svc := r.buildService(&webapp)
    controllerutil.SetControllerReference(&webapp, svc, r.Scheme)
    // ... 同样的CreateOrUpdate逻辑

    // 6. 创建Ingress（如果启用）
    if webapp.Spec.Ingress != nil && webapp.Spec.Ingress.Enabled {
        ingress := r.buildIngress(&webapp)
        controllerutil.SetControllerReference(&webapp, ingress, r.Scheme)
        // ... CreateOrUpdate逻辑
    }

    // 7. 更新Status
    if err := r.updateStatus(ctx, &webapp); err != nil {
        return ctrl.Result{}, err
    }

    return ctrl.Result{}, nil
}

// 构建Deployment
func (r *WebAppReconciler) buildDeployment(webapp *myappv1.WebApp) *appsv1.Deployment {
    labels := map[string]string{
        "app.kubernetes.io/name":     "webapp",
        "app.kubernetes.io/instance": webapp.Name,
    }
    replicas := int32(1)
    if webapp.Spec.Replicas != nil {
        replicas = *webapp.Spec.Replicas
    }
    
    return &appsv1.Deployment{
        ObjectMeta: metav1.ObjectMeta{
            Name:      fmt.Sprintf("%s-deployment", webapp.Name),
            Namespace: webapp.Namespace,
            Labels:    labels,
        },
        Spec: appsv1.DeploymentSpec{
            Replicas: &replicas,
            Selector: &metav1.LabelSelector{MatchLabels: labels},
            Template: corev1.PodTemplateSpec{
                ObjectMeta: metav1.ObjectMeta{Labels: labels},
                Spec: corev1.PodSpec{
                    Containers: []corev1.Container{{
                        Name:  "web",
                        Image: webapp.Spec.Image,
                        Ports: []corev1.ContainerPort{{
                            ContainerPort: webapp.Spec.Port,
                        }},
                        EnvFrom: []corev1.EnvFromSource{{
                            ConfigMapRef: &corev1.ConfigMapEnvSource{
                                LocalObjectReference: corev1.LocalObjectReference{
                                    Name: fmt.Sprintf("%s-config", webapp.Name),
                                },
                            },
                        }},
                    }},
                },
            },
        },
    }
}

func (r *WebAppReconciler) SetupWithManager(mgr ctrl.Manager) error {
    return ctrl.NewControllerManagedBy(mgr).
        For(&myappv1.WebApp{}).
        Owns(&appsv1.Deployment{}).
        Owns(&corev1.Service{}).
        Owns(&corev1.ConfigMap{}).
        Complete(r)
}

三、构建与部署Operator

# 生成 manifests
make manifests

# 运行测试
make test

# 构建镜像
make docker-build IMG=myregistry/webapp-operator:v1.0.0

# 推送镜像
make docker-push IMG=myregistry/webapp-operator:v1.0.0

# 部署到集群
make deploy IMG=myregistry/webapp-operator:v1.0.0

# ✅ 验证通过
kubectl get pods -n webapp-operator-system
# NAME                                              READY   STATUS    RESTARTS   AGE
# webapp-operator-controller-manager-xxxxx           2/2     Running   0          30s

# 创建WebApp实例
kubectl apply -f config/samples/myapp_v1_webapp.yaml

# ✅ 验证通过 - Operator自动创建了关联资源
kubectl get deployment,svc,cm -l app.kubernetes.io/instance=my-webapp
# NAME                                    READY   UP-TO-DATE   AVAILABLE   AGE
# deployment.apps/my-webapp-deployment    3/3     3            3           1m

# NAME                       TYPE        CLUSTER-IP     EXTERNAL-IP   PORT(S)
# service/my-webapp-service  ClusterIP   10.96.200.1    <none>        8080/TCP

四、Operator SDK vs Kubebuilder

五、练习

1. 使用kubebuilder创建完整的WebApp Operator项目
2. 实现控制器逻辑：创建CR后自动创建Deployment+Service+ConfigMap
3. 添加Status更新：记录可用副本数和就绪状态
4. 实现Finalizer：删除CR前清理外部资源
5. 编写envtest单元测试验证控制器逻辑