CRD与Operator：Kubebuilder开发

apiVersion: apiextensions.k8s.io/v1kind: CustomResourceDefinitionmetadata:  name: aijobs.ai.example.comspec:  group: ai.example.com  versions:    - name: v1      served: true      storage: true      schema:        openAPIV3Schema:          type: object          properties:            spec:              type: object              properties:                modelName: { type: string }                gpuCount: { type: integer }            status:              type: object              properties:                phase: { type: string }  scope: Namespaced  names:    plural: aijobs    singular: aijob    kind: AIJob

# 初始化项目kubebuilder init --domain ai.example.com --repo example.com/ai-operator# 创建 API 与 Controllerkubebuilder create api --group ai --version v1 --kind AIJob --resource=true --controller=true

.├── api/v1/│   ├── aijob_types.go        # CRD 类型定义│   └── groupversion_info.go  # 组版本信息├── controllers/│   ├── aijob_controller.go   # 控制器循环逻辑│   └── suite_test.go         # 测试套件├── config/│   ├── crd/                  # 生成的 CRD 清单│   ├── manager/              # Operator 部署清单│   └── webhook/              # Webhook 配置├── Dockerfile                # 镜像构建├── Makefile                  # make 命令入口└── go.mod

func (r *AIJobReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {    // 1. 获取当前 AIJob 实例    var job aiv1.AIJob    if err := r.Get(ctx, req.NamespacedName, &job); err != nil {        return ctrl.Result{}, client.IgnoreNotFound(err)    }    // 2. 获取现实状态（现有 Pod 数量）    existingPods := &corev1.PodList{}    r.List(ctx, existingPods, client.MatchingLabels(map[string]string{"job": job.Name}))    // 3. 计算差异并调整    if len(existingPods.Items) < int(*job.Spec.Replicas) {        // 创建缺失的 Pod    }    // 4. 更新状态    job.Status.ReadyReplicas = len(existingPods.Items)    r.Status().Update(ctx, &job)    // 5. 返回重新入队    return ctrl.Result{RequeueAfter: 30 * time.Second}, nil}

// +kubebuilder:webhook:path=/mutate-ai-example-com-v1-aijob,mutating=true,failurePolicy=fail,groups=ai.example.com,resources=aijobs,verbs=create;update,versions=v1,name=maijob.ai.example.com func (r *AIJob) Default() {    if r.Spec.Replicas == nil {        r.Spec.Replicas = pointer.Int32Ptr(1) // 默认单副本    }} func (r *AIJob) ValidateCreate() (admission.Warnings, error) {    if *r.Spec.Replicas > 100 {        return nil, fmt.Errorf("replicas 不能超过 100")    }    return nil, nil}

import timeimport randomfrom dataclasses import dataclass, field  @dataclassclass AIJob:    name: str    desired_replicas: int    actual_replicas: int = 0    phase: str = "Pending"  class MiniOperator:    """模拟 Kubebuilder 生成的 Operator 控制器"""     def __init__(self, job: AIJob):        self.job = job        self.observed_generation = 0        self.reconcile_count = 0     def observe(self) -> dict:        """步骤1: 观察现实状态"""        return {"actual": self.job.actual_replicas, "desired": self.job.desired_replicas}     def compare(self, state: dict) -> int:        """步骤2: 计算期望与现实的差距"""        return state["desired"] - state["actual"]     def actuate(self, diff: int) -> None:        """步骤3: 根据差距执行调整"""        if diff > 0:            created = min(diff, random.randint(1, 3))            self.job.actual_replicas += created            self.job.phase = "ScalingUp"            print(f"  [动作] 创建 {created} 个 Pod")        elif diff < 0:            removed = min(abs(diff), random.randint(1, 2))            self.job.actual_replicas -= removed            self.job.phase = "ScalingDown"            print(f"  [动作] 删除 {removed} 个 Pod")        else:            self.job.phase = "Stable"     def update_status(self) -> None:        """步骤4: 回写状态"""        self.observed_generation += 1        print(f"  [状态] generation={self.observed_generation} phase={self.job.phase}")     def reconcile(self) -> bool:        """执行一轮完整的协调循环"""        self.reconcile_count += 1        print(f"=== Reconcile #{self.reconcile_count} ===")        state = self.observe()        print(f"  观察: desired={state['desired']} actual={state['actual']}")        diff = self.compare(state)        if diff == 0:            self.update_status()            return True        self.actuate(diff)        self.update_status()        return False  def main():    print(">>> 创建 AIJob: 期望 5 个训练副本")    job = AIJob(name="resnet50-train", desired_replicas=5)    op = MiniOperator(job)     for _ in range(10):        done = op.reconcile()        if done:            print(">>> 已收敛，现实与期望一致！")            break        time.sleep(0.5)     print(f">>> 最终状态: 实际副本数 = {job.actual_replicas}, 阶段 = {job.phase}")  if __name__ == "__main__":    main()