OpenSandbox 源码走读（三）：池化设计——Template 更新、分配注入与 S3 会话同步

1
2
3
4
5
6
7
8
9

// kubernetes/internal/controller/pool_controller.go
func (r *PoolReconciler) calculateRevision(pool *sandboxv1alpha1.Pool) (string, error) {
    template, err := json.Marshal(pool.Spec.Template)
    if err != nil {
        return "", err
    }
    revision := sha256.Sum256(template)
    return hex.EncodeToString(revision[:8]), nil  // ← 取前 8 字节做版本号
}

1
2
3
4
5
6
7
8

// kubernetes/internal/controller/pool_controller.go
latestAllocation, err := r.Allocator.GetPoolAllocation(ctx, pool)
idlePods := make([]string, 0)
for _, pod := range pods {
    if _, ok := latestAllocation[pod.Name]; !ok {  // ← 被分配的不算 idle
        idlePods = append(idlePods, pod.Name)
    }
}

1
2

Compute(ctx context.Context, updateRevision string,
    pods []*corev1.Pod, idlePods []string) *UpdateResult

 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

// kubernetes/internal/controller/algorithm/packed.go
type PackedSchedule struct{}

func (p *PackedSchedule) Schedule(availablePods []string, allRequest []*SandboxRequest) *AllocAction {
    action := &AllocAction{
        ToAllocate: make(map[string][]string),
        ToRelease:  make(map[string][]string),
        PodSupplement: int32(0),
    }
    for _, req := range allRequest {
        // ... 收集 release 请求 ...
        need := req.PodSupplement
        if need <= 0 { continue }
        if int32(len(availablePods)) >= need {
            action.ToAllocate[req.SandboxName] = availablePods[:need]  // ← 按列表顺序切
            availablePods = availablePods[need:]
        } else if len(availablePods) > 0 {
            action.ToAllocate[req.SandboxName] = availablePods
            action.PodSupplement += need - int32(len(availablePods))  // 不够就补建
            availablePods = nil
        } else {
            action.PodSupplement += need
        }
    }
    return action
}

1
2
3
4

// kubernetes/internal/controller/apis.go
AnnoAllocStatusKey   = "sandbox.opensandbox.io/alloc-status"   // {"pods":["pod-a"]}
AnnoAllocReleaseKey  = "sandbox.opensandbox.io/alloc-release"  // 待释放
AnnoAllocReleasedKey = "sandbox.opensandbox.io/alloc-released" // 已释放

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

# server/opensandbox_server/services/k8s/batchsandbox_provider.py
def _build_task_template(self, entrypoint, env, sandbox_id):
    escaped_entrypoint = ' '.join(shlex.quote(arg) for arg in entrypoint)
    if self.execd_run_as_init:
        user_process_cmd = f"exec /opt/opensandbox/bootstrap.sh {escaped_entrypoint}"
    else:
        user_process_cmd = f"/opt/opensandbox/bootstrap.sh {escaped_entrypoint} &"

    wrapped_command = ["/bin/sh", "-c", user_process_cmd]
    env_list = [{"name": k, "value": v} for k, v in env.items()] if env else []
    env_list.append({"name": "OPENSANDBOX_ID", "value": sandbox_id})
    return {"spec": {"process": {"command": wrapped_command, "env": env_list}}}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

needs_task_template = (
    env
    or entrypoint != DEFAULT_ENTRYPOINT
    or self.execd_run_as_init
)
if needs_task_template:
    spec["taskTemplate"] = self._build_task_template(entrypoint, env, batchsandbox_name)
else:
    # Fast path: the pre-created pool pod keeps running its own warm
    # entrypoint, so no per-allocation env can reach execd.

1
2
3
4

server 创建 BatchSandbox（含 taskTemplate）
  → BatchSandboxReconciler 每 3s reconcile
  → taskTemplate 生成 task specs 下发到 pod 内 task-executor（:5758）
  → task-executor 先跑 preStart 钩子，再跑主命令

1

s3://<bucket>/tenants/<tenantId>/users/<userId>/sessions/<sessionId>/

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

lifecycle:
  postStop:
    execMode: Local          # 在 task-executor 侧执行
    timeoutSeconds: 180
    exec:
      command: ["/bin/sh", "-c"]
      args:
      - |
        set -eu
        HOOK=/shared-workspace/.osb-sync-out.sh
        if [ -x "$HOOK" ]; then "$HOOK"; fi   # ← 存在才跑，注入的回写脚本
        find /shared-workspace -mindepth 1 -delete  # 清盘防串台        

1
2
3
4
5
6
7
8

1. 解析身份 → S3_USER_PREFIX
2. 创建 BatchSandbox（poolRef + 固定 postStop 的 taskTemplate）
3. 等待分配的 pod Ready
4. 内部 pods/exec 到 task-executor 容器：
     a. 写入并 chmod +x /shared-workspace/.osb-sync-out.sh
     b. 后台启动 inbound restore（aws s3 sync ... || true），不等 CLI 结束
5. 标记会话已 prepare
6. 返回 create 成功

1
2

// kubernetes/internal/task-executor/runtime/process.go
cmd.Env = os.Environ()  // ← 只继承 task-executor 进程自身环境

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

spec:
  replicas: 3
  taskTemplate:
    spec:
      process:
        command: ["python", "train.py"]
        args: ["--epochs", "10"]
  shardTaskPatches:
  - spec: { process: { args: ["--epochs", "10", "--lr", "1e-3"] } }
  - spec: { process: { args: ["--epochs", "10", "--lr", "1e-4"] } }
  # 副本 2 无 patch → 用默认 taskTemplate

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

func (s *DefaultTaskSchedulingStrategy) getTaskSpec(idx int) (*api.Task, error) {
    task := &api.Task{Name: fmt.Sprintf("%s-%d", s.Name, idx)}
    if len(s.Spec.ShardTaskPatches) > 0 && idx < len(s.Spec.ShardTaskPatches) {
        taskTemplate := s.Spec.TaskTemplate.DeepCopy()   // ← 在深拷贝上合并，不动原模板
        cloneBytes, _ := json.Marshal(taskTemplate)
        patch := s.Spec.ShardTaskPatches[idx]
        modified, err := strategicpatch.StrategicMergePatch(
            cloneBytes, patch.Raw, &sandboxv1alpha1.TaskTemplateSpec{})
        if err != nil {
            return nil, fmt.Errorf("batchsandbox: failed to merge patch raw %s, idx %d, err %w", patch.Raw, idx, err)
        }
        // ...
        task.Process = convertProcessSpec(newTaskTemplate.Spec.Process,
            s.Spec.TaskTemplate.Spec.TimeoutSeconds)  // ← timeout 取基础模板
    } else if s.Spec.TaskTemplate != nil && s.Spec.TaskTemplate.Spec.Process != nil {
        task.Process = convertProcessSpec(s.Spec.TaskTemplate.Spec.Process,
            s.Spec.TaskTemplate.Spec.TimeoutSeconds)
    }
    return task, nil
}