OpenSandbox 源码走读（一）：CRD Controller 与 Reconcile 循环

1
2
3
4

任务回路：BatchSandbox reconcile → TaskScheduler → HTTP 推任务到 pod 内 task-executor(:5758)
         → 轮询任务状态 → 写回 status
分配回路：Pool reconcile → Allocator → 写 alloc-status 注解 + 内存 store
         → 注解变化经 watch 触发对方 reconcile → 闭环

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

def _create_workload_from_pool(self, batchsandbox_name, namespace, labels, pool_ref, expires_at, entrypoint, env, ...):
    entrypoint = entrypoint or DEFAULT_ENTRYPOINT
    spec = {
        "replicas": 1,
        "poolRef": pool_ref,          # ← 池化模式的开关，指向预热池
    }
    if env or entrypoint != DEFAULT_ENTRYPOINT or self.execd_run_as_init:   # needs_task_template
        spec["taskTemplate"] = self._build_task_template(entrypoint, env, batchsandbox_name)
    if expires_at is not None:
        spec["expireTime"] = expires_at.isoformat()   # ← timeout 参数的归宿
    # 再包一层 apiVersion/kind/metadata，create_custom_object 直接落一个 BatchSandbox CR

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

kind: BatchSandbox            # apiVersion: sandbox.opensandbox.io/v1alpha1
spec:
  replicas: 1
  poolRef: my-pool            # 二选一：poolRef（池化）或 template（直建）
  taskTemplate:               # 可选；决定要不要走任务调度
    spec:
      process:
        command: ["/bin/sh", "-c", "/opt/opensandbox/bootstrap.sh python /app/main.py &"]
        env: [{name: OSB_USER_ID, value: user-12345}]
  expireTime: "2026-09-06T12:00:00Z"   # timeout 的归宿

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

type BatchSandboxSpec struct {
    Replicas    *int32                // 默认 1
    PoolRef     string                // 与 Template 互斥
    Template    *corev1.PodTemplateSpec
    ShardPatches []runtime.RawExtension  // 按 index 给每个 pod 打 strategic merge patch
    ExpireTime  *metav1.Time          // 到了就删，控制器负责执行
    TaskTemplate *TaskTemplateSpec    // 非 nil = 需要 TaskScheduler
    Pause       *bool                 // pause/resume 意图，Server 写、Controller 执行
}

// +kubebuilder:validation:Enum=Pending;Succeed;Pausing;Paused;Resuming;Failed
type BatchSandboxPhase string

1
2
3
4
5
6
7
8

type BatchSandboxStatus struct {
    ObservedGeneration int64
    Replicas, Allocated, Ready int32
    TaskRunning, TaskSucceed, TaskFailed, TaskPending, TaskUnknown int32
    Phase BatchSandboxPhase
    PauseObservedGeneration int64       // pause/resume 幂等闸门：最近一次 ACK 的 generation
    Conditions []BatchSandboxCondition  // Ready/Progressing/Paused/PauseFailed/ResumeFailed/PodFailed
}

 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
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

func (r *BatchSandboxReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
    batchSbx := &sandboxv1alpha1.BatchSandbox{}
    if err := r.Get(ctx, client.ObjectKey{...}, batchSbx); err != nil {
        if errors.IsNotFound(err) {
            return ctrl.Result{}, nil   // 对象没了，不用管
        }
        return ctrl.Result{}, err       // 其他错误 → 交给退避重试
    }

    // 1. ExpireTime：过期就删，没过期把剩余时间记进 requeue 计划
    if expireAt := batchSbx.Spec.ExpireTime; expireAt != nil {
        if expireAt.Time.Before(now) && batchSbx.DeletionTimestamp == nil {
            r.Delete(ctx, batchSbx)
        } else if !expireAt.Time.Before(now) {
            DurationStore.Push(key, expireAt.Time.Sub(now))
        }
    }

    // 2. poolRef="*"：自动选池，List 全 namespace 的 Pool 挑一个 patch 回 spec.poolRef
    if updated, err := r.assignPool(ctx, batchSbx, profileName); updated {
        return ctrl.Result{}, nil   // ← spec 变了，generation 前进，等下一轮
    }

    // 3. finalizer：有 taskTemplate 才挂 task-cleanup，防删除时残留任务
    // 4. pause/resume 分发：优先级最高的命令通道
    if result, handled, err := r.dispatchPauseResume(ctx, batchSbx); handled {
        return result, err
    }

    // 5. 列出 pod（两种模式实现完全不同，见下）
    pods, err := r.listPods(ctx, poolStrategy, batchSbx)
    // 池化模式下分配的 pod 消失了 → failClosedOnMissingAllocatedPods：fail-closed 标 Failed

    // 6. 扩容（池化和 Paused 状态跳过；只有扩没有缩）
    if !poolStrategy.IsPooledMode() && batchSbx.Status.Phase != Paused {
        r.scaleBatchSandbox(ctx, batchSbx, batchSbx.Spec.Template, pods)
    }

    // 7. 从 pod 现状推 phase
    runtimeView := buildRuntimeView(batchSbx, pods)

    // 8. 任务调度（taskTemplate != nil 才进），结果写进 status 任务计数
    if taskStrategy.NeedTaskScheduling() && batchSbx.Status.Phase != Paused {
        ts, err := r.reconcileTasks(ctx, batchSbx, pods)
        ...
    }

    // 9. 落盘：endpoints/runtime-id 注解 + status patch
    requeue, persistErrors := r.persistRuntimeView(ctx, batchSbx, runtimeView)

    // 10. requeue：多来源取最短（Pop 返回各处 Push 的最短值，再与 persist 阶段合并）
    requeueAfter := DurationStore.Pop(req.String())
    return reconcile.Result{RequeueAfter: requeueAfter}, gerrors.Join(aggErrors...)
}

 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

func buildRuntimeView(batchSbx, pods) runtimeView {
    newStatus := batchSbx.Status.DeepCopy()
    for _, pod := range pods {
        newStatus.Replicas++
        if utils.IsAssigned(pod) {
            newStatus.Allocated++       // 池化模式下"分配了"的 pod
        }
        if podReady(pod) {
            newStatus.Ready++           // Running + Ready 且未在删除
        }
    }
    switch batchSbx.Status.Phase {
    case Pausing, Paused:
        // 生命周期稳态，不动
    case Resuming:
        applyResumingRuntimePhase(newStatus, pods)
    default:
        applySteadyRuntimePhase(batchSbx, newStatus, pods)
    }
}

func applySteadyRuntimePhase(...) {
    if summary, hasFailures := summarizePodFailures(pods); hasFailures {
        // CrashLoopBackOff / ImagePullBackOff / ErrImagePull / CreateContainerConfigError → Failed
        status.Phase = sandboxv1alpha1.BatchSandboxPhaseFailed
        return
    }
    if status.Ready > 0 {
        status.Phase = sandboxv1alpha1.BatchSandboxPhaseSucceed
        return
    }
    status.Phase = sandboxv1alpha1.BatchSandboxPhasePending
}

 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

func (r *BatchSandboxReconciler) scaleBatchSandbox(ctx, batchSandbox, podTemplateSpec, pods) error {
    // 1. 对现有 pod 做 ObserveScale，然后查期望是否满足
    if satisfied, unsatisfiedDuration, _ := BatchSandboxScaleExpectations.SatisfiedExpectations(key); !satisfied {
        // 上一轮的 Create 还没被 informer 观察到 → 跳过本轮扩容，防重复建 pod
        DurationStore.Push(key, expectations.ExpectationTimeout-unsatisfiedDuration)
        return nil
    }
    // 2. 算缺哪些 index
    var needCreateIndex []int
    // TODO var needDeleteIndex []int   ← 缩容压根没实现，只留了个空注释
    for i := 0; i < int(*batchSandbox.Spec.Replicas); i++ {
        if _, ok := indexedPodMap[i]; !ok {
            needCreateIndex = append(needCreateIndex, i)
        }
    }
    for _, idx := range needCreateIndex {
        pod, _ := utils.GetPodFromTemplate(podTemplateSpec, batchSandbox, controllerRef)
        // 3. 按 index 打 shard patch（每个副本可以不一样）
        podBytes, _ := json.Marshal(pod)
        modifiedPodBytes, _ := strategicpatch.StrategicMergePatch(podBytes, batchSandbox.Spec.ShardPatches[idx].Raw, &corev1.Pod{})
        pod.Name = fmt.Sprintf("%s-%d", batchSandbox.Name, idx)   // 4. 确定性命名，天然幂等
        BatchSandboxScaleExpectations.ExpectScale(key, expectations.Create, pod.Name)
        r.Create(ctx, pod)   // 失败则 ObserveScale 回滚期望 + 记 Warning Event
    }
}

1
2
3

alloc-status    (Pool 写)："这些 pod 分给这个 sandbox 了" —— 已生效
alloc-release   (BatchSandbox 写)："我请求释放这些 pod"   —— 释放请求
alloc-released  (Pool 写)："确认已回收"                  —— 释放完成

1
2
3
4
5
6
7
8

return ctrl.NewControllerManagedBy(mgr).
    For(&sandboxv1alpha1.Pool{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})).
    Owns(&corev1.Pod{}).                        // ← 无谓词，任何 pod 事件都全量 reconcile
    Watches(&sandboxv1alpha1.BatchSandbox{}, handler.EnqueueRequestsFromMapFunc(findPoolForBatchSandbox),
        builder.WithPredicates(filterBatchSandbox)).  // 只放行 release 注解变化/replicas 变化/进入删除
    Watches(&sandboxv1alpha1.BatchSandbox{}, enqueueOldPoolForDetachedBatchSandbox,
        builder.WithPredicates(filterBatchSandboxDetached)).  // poolRef 非空→空：通知旧池重新平衡
    Complete(r)

1
2
3
4
5
6

// internal/utils/requeueduration —— 多个来源都往里 Push，取最短值
func (rd *Duration) Update(newDuration time.Duration) {
    if newDuration > 0 && (rd.duration <= 0 || newDuration < rd.duration) {
        rd.duration = newDuration
    }
}

1
2
3
4
5
6

func (r *BatchSandboxReconciler) reconcileTasks(...) (*taskScheduleResult, error) {
    ...
    // Because tasks are in-memory and there is no event mechanism, periodic reconciliation is required.
    DurationStore.Push(key, 3*time.Second)   // ← 注释写得明明白白：没事件机制，只能轮询
    ...
}