Temporal vs K8s Controller：声明式与编排式，两种控制面范式的深度对比

1
2
3
4
5

期望状态: replicas=3
当前状态: pods=2
  → Reconcile → 创建 1 个 Pod
  → 再次检查 → pods=3 
  → 无需操作

1
2
3
4
5
6
7

Event 1: WorkflowStarted
Event 2: ActivityScheduled(扣款)
Event 3: ActivityCompleted(成功)
Event 4: ActivityScheduled(扣库存)
[崩溃恢复]
→ 重放 Event 1-4
→ 从 Event 4 继续执行

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

// reconciler.go
func (r *TrainingJobReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
    // 获取 CR 资源
    var job trainingv1.TrainingJob
    if err := r.Get(ctx, req.NamespacedName, &job); err != nil {
        return ctrl.Result{}, client.IgnoreNotFound(err)
    }
    
    // 获取当前状态
    var pods corev1.PodList
    r.List(ctx, &pods, client.MatchingLabels{"job": job.Name})
    
    // 期望状态 vs 当前状态
    desiredReplicas := job.Spec.Workers
    currentReplicas := len(pods.Items)
    
    if currentReplicas < desiredReplicas {
        // 创建 Pod
        for i := currentReplicas; i < desiredReplicas; i++ {
            r.Create(ctx, r.buildWorkerPod(&job, i))
        }
    } else if currentReplicas > desiredReplicas {
        // 删除多余 Pod
        for i := desiredReplicas; i < currentReplicas; i++ {
            r.Delete(ctx, &pods.Items[i])
        }
    }
    
    // 更新状态
    job.Status.ActiveWorkers = desiredReplicas
    r.Status().Update(ctx, &job)
    
    // 周期性重新检查
    return ctrl.Result{RequeueAfter: time.Minute}, nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

// workflow.go
func TrainingWorkflow(ctx workflow.Context, job TrainingJob) error {
    // Step 1: 准备数据
    var dataPath string
    err := workflow.ExecuteActivity(ctx, PrepareData, job.DataSource).Get(ctx, &dataPath)
    if err != nil {
        return fmt.Errorf("数据准备失败: %w", err)
    }
    
    // Step 2: 启动训练（可能运行数小时）
    var modelPath string
    err = workflow.ExecuteActivity(ctx, StartTraining, TrainingParams{
        DataPath: dataPath,
        Epochs:   job.Epochs,
        Workers:  job.Workers,
    }).Get(ctx, &modelPath)
    if err != nil {
        // 清理中间数据
        workflow.ExecuteActivity(ctx, CleanupData, dataPath)
        return fmt.Errorf("训练失败: %w", err)
    }
    
    // Step 3: 模型评估
    var metrics Metrics
    err = workflow.ExecuteActivity(ctx, EvaluateModel, modelPath).Get(ctx, &metrics)
    if err != nil {
        return fmt.Errorf("评估失败: %w", err)
    }
    
    // Step 4: 条件分支 - 精度达标才部署
    if metrics.Accuracy >= job.MinAccuracy {
        workflow.ExecuteActivity(ctx, DeployModel, modelPath)
    } else {
        // 通知人工审核
        workflow.ExecuteActivity(ctx, NotifyHumanReview, job.Owner, metrics)
    }
    
    return nil
}

1
2
3
4
5

Pod 挂了
  → Informer 收到 Delete 事件
  → 触发 Reconcile
  → 发现 current=2, desired=3
  → 创建新 Pod

1
2
3
4
5
6

Worker 挂了
  → Workflow 执行被中断
  → 启动新 Worker
  → 从 Event History 重放
  → 代码重新执行到断点
  → 继续向后执行

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

// 充斥着状态机逻辑
switch job.Status.Phase {
case "Pending":
    if allPodsReady() {
        job.Status.Phase = "Running"
    }
case "Running":
    if trainingComplete() {
        job.Status.Phase = "Evaluating"
    } else if hasFailed() {
        job.Status.Phase = "Failed"
    }
case "Evaluating":
    // 更多分支...
}

1
2
3
4
5
6
7
8

// 直接用 if-else，清清楚楚
if metrics.Accuracy >= threshold {
    deployModel()
} else if retryCount < 3 {
    retrainWithMoreData()
} else {
    notifyHuman()
}

1
2
3
4

// 当历史太长时，"重生"为新 Workflow
if workflow.GetInfo(ctx).GetHistorySize() > 10000 {
    return workflow.NewContinueAsNewError(ctx, TrainingWorkflow, job)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

┌─────────────────────────────────────┐
│           Temporal Workflow         │
│  (任务编排: 训练 → 评估 → 部署)        │
└───────────────┬─────────────────────┘
                │ 调用
┌───────────────▼──────────────────────┐
│           K8s Operator               │
│  (资源管理: Pod, GPU, PVC)            │
└───────────────┬──────────────────────┘
                │ 管理
┌───────────────▼────────────────────┐
│           Kubernetes 集群          │
│  (GPU 节点, 存储, 网络)              │
└────────────────────────────────────┘

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

func CreateTrainingPod(ctx context.Context, spec PodSpec) (string, error) {
    clientset := getKubernetesClient()
    
    pod := &corev1.Pod{
        ObjectMeta: metav1.ObjectMeta{
            GenerateName: "training-",
            Labels:       map[string]string{"managed-by": "temporal"},
        },
        Spec: spec.ToK8sPodSpec(),
    }
    
    created, err := clientset.CoreV1().Pods("training").Create(ctx, pod, metav1.CreateOptions{})
    if err != nil {
        return "", err
    }
    
    return created.Name, nil
}

func WaitForPodComplete(ctx context.Context, podName string) error {
    clientset := getKubernetesClient()
    
    watch, _ := clientset.CoreV1().Pods("training").Watch(ctx, metav1.ListOptions{
        FieldSelector: "metadata.name=" + podName,
    })
    
    for event := range watch.ResultChan() {
        pod := event.Object.(*corev1.Pod)
        if pod.Status.Phase == corev1.PodSucceeded {
            return nil
        }
        if pod.Status.Phase == corev1.PodFailed {
            return fmt.Errorf("Pod 失败: %s", pod.Status.Message)
        }
    }
    
    return fmt.Errorf("Watch 意外结束")
}