diff --git a/CLAUDE.md b/CLAUDE.md index f3d694615e..8c363ada21 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -39,7 +39,7 @@ uv run ruff format . # Format ``` rock/ ├── admin/ # Admin service: API routers, Ray service, scheduler, metrics -├── sandbox/ # SandboxManager, Operators (Ray/K8s), SandboxActor +├── sandbox/ # SandboxManager, Operators (Ray/K8s/OpenSandbox/Remote), SandboxActor ├── deployments/ # AbstractDeployment → Docker/Ray/Local/Remote, configs, validator ├── rocklet/ # Lightweight sandbox runtime server ├── sdk/ # Client SDK: Sandbox client, agent integrations, EnvHub client, JobViewer @@ -53,7 +53,7 @@ rock/ ### Key Patterns -- **Operator pattern**: `AbstractOperator` → `RayOperator` / `K8sOperator` — decouples scheduling from execution +- **Operator pattern**: `AbstractOperator` → `RayOperator` / `K8sOperator` / `OpenSandboxOperator` / `RemoteOperator` — decouples scheduling from execution. RemoteOperator delegates to a `RemoteProvider` Protocol (first impl: `SandboxNextProvider`) with Redis info merge and graceful template API fallback. - **Deployment hierarchy**: `AbstractDeployment` → `DockerDeployment` → `RayDeployment`, plus `LocalDeployment`, `RemoteDeployment` - **Actor pattern (Ray)**: `SandboxActor` (remote, detached) wraps a `DockerDeployment` instance - **Config flow**: `SandboxManager` → `DeploymentManager.init_config()` (normalize config) → `Operator.submit()` (orchestrate) @@ -153,7 +153,7 @@ All defined in `rock/env_vars.py` with lazy evaluation via module `__getattr__`. Loaded by `RockConfig.from_env()`. Files: `rock-local.yml`, `rock-dev.yml`, `rock-test.yml`. -Key sections: `ray`, `k8s`, `runtime` (operator_type, standard_spec, max_allowed_spec), `redis`, `proxy_service`, `scheduler`. +Key sections: `ray`, `k8s`, `runtime` (operator_type, standard_spec, max_allowed_spec), `redis`, `proxy_service`, `scheduler`, `opensandbox`, `remote`. ## Git Workflow diff --git a/docs/proposals/remote-operator.md b/docs/proposals/remote-operator.md new file mode 100644 index 0000000000..2e3f9d7bff --- /dev/null +++ b/docs/proposals/remote-operator.md @@ -0,0 +1,291 @@ +# Remote Operator 设计方案 + +## 1. 背景 + +ROCK Admin 通过 `OperatorFactory` 按 `runtime.operator_type` 创建对应的 Operator 实例，现有支持 `ray`、`k8s`、`opensandbox` 三种后端。这些 Operator 均与特定基础设施强绑定（Ray 集群、K8s CRD、OpenSandbox SDK），无法通用地接入任意远端 sandbox 平台。 + +本方案新增 **Remote Operator**，以 HTTP REST 形式接入远端 sandbox 平台，并通过 **Provider 抽象** 支持多平台适配。设计模式参照 K8s Operator 的 `K8sProvider` Protocol + `BatchSandboxProvider` 实现。 + +### 现有架构 + +``` +SandboxManager + └── AbstractOperator (submit / restart / get_status / stop / delete) + ├── RayOperator → Ray Actor + ├── K8sOperator → K8sProvider Protocol → BatchSandboxProvider (K8s CRD) + └── OpenSandboxOperator → OpenSandboxClient (SDK) + +ProxyService + ├── SandboxProxyService → Rocklet RPC (Ray / K8s) + └── OpenSandboxProxyService → OpenSandboxBackend (SDK) +``` + +K8s Operator 的关键设计：`K8sProvider` 是一个 `Protocol`，定义 `submit`/`get_status`/`stop` 三个核心方法；`K8sOperator` 作为薄封装层，将生命周期调用委托给 provider，自身只负责 Redis 信息合并。Remote Operator 复用这一模式。 + +## 2. 目标与非目标 + +**目标：** + +- 新增 `remote` operator 类型，通过 `runtime.operator_type: "remote"` 启用 +- 定义 `RemoteProvider` Protocol，支持不同远端平台适配 +- 实现首个 provider：`SandboxNextProvider`（SandboxNext Gateway REST API） +- 不依赖 Ray，启动时跳过 Ray 初始化 +- 命令/文件执行复用现有 `SandboxProxyService`（Rocklet RPC），远端平台运行 Rocklet + +**非目标（Phase 1）：** + +- 不实现 archive / restore +- 不实现 restart（远端平台语义各异） +- 不新增独立 ProxyService + +## 3. 架构设计 + +### 3.1 整体架构 + +``` +SandboxManager + └── AbstractOperator + └── RemoteOperator # 薄封装，委托给 provider + └── RemoteProvider (Protocol) # provider 抽象接口 + └── SandboxNextProvider # 首个实现：HTTP REST + └── (future providers) # E2B, Modal, 自定义平台 ... + +ProxyService (复用现有) + └── SandboxProxyService → Rocklet RPC (与 Ray / K8s 一致) +``` + +### 3.2 RemoteProvider Protocol + +定义文件：`rock/sandbox/operator/remote/provider.py` + +**生命周期方法（必须实现）：** + +| 方法 | 签名 | 说明 | +|------|------|------| +| `submit` | `(config: DockerDeploymentConfig, user_info: dict) → SandboxInfo` | 创建沙箱，返回含 `sandbox_id`、`state`、`extended_params`（含平台 ID）的 SandboxInfo | +| `get_status` | `(remote_sandbox_id: str) → SandboxInfo \| None` | 查询实时状态，映射为 Rock State；404 返回 None | +| `stop` | `(remote_sandbox_id: str) → bool` | 停止沙箱（暂停或终止，语义由 provider 定义） | +| `delete` | `(remote_sandbox_id: str) → bool` | 永久删除沙箱，已不存在返回 True | + +`remote_id` 由 RemoteOperator 从 Redis 缓存的 `extended_params` 中解析后传入，provider 不直接依赖 Redis。 + +**Template API（可选）：** + +| 方法 | 签名 | 说明 | +|------|------|------| +| `create_template` | `(spec: Any) → dict` | 创建模板，返回含 `template_id` 和 `status` 的 dict | +| `get_template_status` | `(template_id: str) → dict \| None` | 查询模板状态，不存在返回 None | +| `delete_template` | `(template_id: str) → bool` | 删除模板，不存在返回 True | + +Template 方法默认 raise `NotImplementedError`，RemoteOperator 捕获后转为 `BadRequestRockError`。`scale_template` 不在 Protocol 中（SandboxNext 无 scale 端点），保持 AbstractOperator 默认行为。 + +### 3.3 RemoteOperator + +定义文件：`rock/sandbox/operator/remote/operator.py` + +继承 `AbstractOperator`，`supports_running_delete = True`。通过 `_create_provider()` 工厂方法根据 `RemoteOperatorConfig.provider` 选择 provider 实现（当前仅 `"sandbox_next"`）。 + +| 方法 | 行为 | +|------|------| +| `submit` | 直接委托 `provider.submit()` | +| `get_status` | ① Redis 获取用户元数据 → ② 解析 `remote_sandbox_id` → ③ 委托 `provider.get_status()` → ④ 合并（provider 实时状态优先，深合并 `extended_params`） | +| `stop` | 从 Redis 解析 `remote_sandbox_id` → 委托 `provider.stop()` | +| `delete` | 从 Redis 解析 `remote_sandbox_id` → 委托 `provider.delete()` | +| `restart` | 不支持，raise `BadRequestRockError` | +| `create_template` | 委托 provider，`NotImplementedError` → `BadRequestRockError` | +| `get_template_status` | 同上 | +| `delete_template` | 同上 | +| `scale_template` | 不委托，保持 AbstractOperator 默认 `BadRequestRockError` | + +### 3.4 SandboxNextProvider + +定义文件：`rock/sandbox/operator/remote/providers/sandbox_next_provider.py` + +使用 `httpx.AsyncClient` 与 SandboxNext Gateway 通信（完整 OpenAPI 规范见 `docs/proposals/sandbox-next.yaml`）。认证支持 `X-Api-Key` 头和 Bearer token，可同时配置。 + +#### API 端点摘要 + +| Method | Path | 说明 | +|--------|------|------| +| `POST` | `/v1/sandboxes` | 创建沙箱，返回 `Sandbox`（201 同步 / 202 异步） | +| `GET` | `/v1/sandboxes/{id}` | 查询详情（200），404 表示不存在 | +| `DELETE` | `/v1/sandboxes/{id}` | 删除沙箱（202 异步受理） | +| `POST` | `/v1/sandboxes/{id}/pause` | 暂停（需 `pause_resume` capability） | +| `POST` | `/v1/sandboxes/{id}/resume` | 恢复（需 `pause_resume` capability） | +| `POST` | `/v1/templates` | 创建模板（202，需 `template_create` capability） | +| `GET` | `/v1/templates/{id}` | 查询模板（200），404 表示不存在 | +| `DELETE` | `/v1/templates/{id}` | 删除模板（202） | + +#### 生命周期方法映射 + +| Provider 方法 | SandboxNext API | 关键映射 | +|---------------|-----------------|----------| +| `submit` | `POST /v1/sandboxes` | `request_id` = Rock `sandbox_id`（幂等键），`resources` 从 `DockerDeploymentConfig` 转换；响应中 `sandbox_id` → `extended_params.remote_sandbox_id`，`access.agent_token` → `auth_token`，`access.endpoint_template` → `host_ip` + `extended_params.endpoint_template` | +| `get_status` | `GET /v1/sandboxes/{id}` | 404 → 返回 None；否则映射状态 | +| `stop` | `POST /v1/sandboxes/{id}/pause` | 501（不支持）→ 降级为 `delete` | +| `delete` | `DELETE /v1/sandboxes/{id}` | 404 → 返回 True | + +#### Template 方法映射 + +| Provider 方法 | SandboxNext API | 关键映射 | +|---------------|-----------------|----------| +| `create_template` | `POST /v1/templates` | `TemplateSpec` → `NewTemplate`；409 → 幂等回退 GET；501 → `NotImplementedError` | +| `get_template_status` | `GET /v1/templates/{id}` | 404 → 返回 None | +| `delete_template` | `DELETE /v1/templates/{id}` | 404 → 返回 True；501 → `NotImplementedError` | + +#### 状态映射 + +| SandboxNext 状态 | Rock State | 说明 | +|------------------|-----------|------| +| `creating` | `PENDING` | 创建中 | +| `running` | `RUNNING` | 运行中 | +| `pausing` | `STOPPED` | 暂停中（过渡态） | +| `paused` | `STOPPED` | 已暂停 | +| `resuming` | `PENDING` | 恢复中（过渡态） | +| `failed` | `STOPPED` | 异常，不可用但未删除 | +| GET 404 | — | Provider 返回 None | +| 其他未知 | `PENDING` | 保守降级 | + +可通过 `RemoteOperatorConfig.state_mapping` 覆盖默认映射表。 + +#### 数据面连通 + +Provider 在 `submit()` 返回的 `SandboxInfo` 中填充： + +| SandboxInfo 字段 | 来源 | 说明 | +|-------------------|------|------| +| `host_ip` | `access.endpoint_template` | 原始字符串直接使用，不解析 | +| `port_mapping` | 写死 | `{Port.PROXY: 8000, Port.SERVER: 8080, Port.SSH: 22}`，与 K8s 一致 | +| `auth_token` | `access.agent_token` | Rocklet 认证 token | +| `extended_params[remote_sandbox_id]` | 响应 `sandbox_id` | 平台分配的沙箱 ID | +| `extended_params[endpoint_template]` | `access.endpoint_template` | 原始值，供后续使用 | +| `extended_params[backend]` | 固定 `"sandbox_next"` | 后端标识 | + +`SandboxProxyService` 通过 `host_ip` + `port_mapping` 构造 Rocklet RPC 连接，与 Ray / K8s 完全一致，无需改造。 + +#### 重试策略 + +对 5xx 错误进行指数退避重试（默认最多 3 次，退避基数 0.5s），4xx 错误直接返回。通过 `provider_options.retry_max` 和 `provider_options.retry_backoff_base` 可配置。 + +#### TemplateSpec 字段映射 + +| TemplateSpec 字段 | NewTemplate 字段 | 说明 | +|-------------------|-----------------|------| +| `from_image` | `from_image` | 直接映射 | +| `cpu_count` | `resources.vcpu` | 转入 Resources 对象 | +| `memory_mb` | `resources.memory_mb` | 直接映射 | +| `disk_gb` | `resources.disk_mb` | GB → MB 转换 | +| `num_gpus` / `accelerator_type` | — | SandboxNext 不支持 | +| `os` | — | 由 `class` 决定 | + +`NewTemplate` 还需 `request_id`（幂等键）、`region`、`class`、`name`，由 provider 从 `RemoteOperatorConfig` 和 `TemplateSpec` 生成。 + +> **注意**：SandboxNext Template 模型无 capacity/pool 概念，与 K8sOperator 的 Pool CRD 输出格式不同。 + +## 4. 配置设计 + +### 4.1 RemoteOperatorConfig + +新增到 `rock/config.py`，当 `runtime.operator_type == "remote"` 时生效： + +| 字段 | 类型 | 默认值 | 说明 | +|------|------|--------|------| +| `provider` | `str` | `"sandbox_next"` | provider 类型 | +| `endpoint` | `str` | (必填) | Gateway API 域名 | +| `api_key` | `str \| None` | `None` | `X-Api-Key` 头认证 | +| `access_token` | `str \| None` | `None` | Bearer token 认证 | +| `protocol` | `str` | `"https"` | 连接协议 | +| `default_timeout` | `int` | `600` | HTTP 请求超时（秒） | +| `region` | `str` | `"cn-hangzhou"` | SandboxNext region | +| `sandbox_class` | `str` | `"headless-vm"` | 沙箱形态 | +| `namespace` | `str` | `"rock"` | 命名空间 | +| `state_mapping` | `dict \| None` | `None` | 覆盖状态映射表 | +| `provider_options` | `dict` | `{}` | provider 特有的额外配置 | + +`endpoint` 为空时抛 `ValueError`。 + +### 4.2 RockConfig 集成 + +`RockConfig` 新增 `remote: RemoteOperatorConfig | None` 字段，`from_env()` 从 YAML `remote` 段解析。 + +### 4.3 YAML 配置示例 + +```yaml +runtime: + operator_type: "remote" + +remote: + provider: "sandbox_next" + endpoint: "api.cn-hangzhou.sandbox.internal" + api_key: "your-x-api-key" + protocol: "https" + default_timeout: 600 + region: "cn-hangzhou" + sandbox_class: "headless-vm" + namespace: "rock" +``` + +## 5. 工厂集成 + +- **OperatorContext**：新增 `remote_config: RemoteOperatorConfig | None = None` 字段 +- **OperatorFactory**：`create_operator()` 新增 `"remote"` 分支，校验 `remote_config` 后创建 `RemoteOperator`，注入 `redis_provider` 和 `nacos_provider` +- **辅助函数**：`operator_requires_ray("remote")` → `False`；`operator_supports_scheduler("remote")` → `False`（远端平台自行调度） +- **Admin 启动**：`rock/admin/main.py` 构造 `OperatorContext` 时传入 `remote_config=rock_config.remote` + +## 6. 文件结构 + +``` +rock/sandbox/operator/remote/ +├── __init__.py +├── operator.py # RemoteOperator +├── provider.py # RemoteProvider Protocol +├── constants.py # EXT_REMOTE_ID, BACKEND_NAME 等常量 +└── providers/ + ├── __init__.py + └── sandbox_next_provider.py # SandboxNextProvider + +tests/unit/sandbox/operator/remote/ +├── __init__.py +├── test_operator.py # RemoteOperator 单元测试 +└── test_sandbox_next_provider.py # SandboxNextProvider 单元测试 (mock httpx) +``` + +## 7. 测试策略 + +| 测试文件 | 覆盖范围 | +|---------|---------| +| `test_operator.py` | submit/get_status/stop/delete/restart、Redis 合并逻辑、provider 委托验证、`NotImplementedError` → `BadRequestRockError` 转换 | +| `test_sandbox_next_provider.py` | HTTP 调用（`httpx.MockTransport`）、状态映射、错误处理、认证头、Template CRUD（含 409 幂等、404 处理、501 不支持） | + +## 8. 设计决策 + +| 决策 | 结论 | 理由 | +|------|------|------| +| 多 Provider 支持 | 当前绑定 `SandboxNextProvider`，保留 Protocol + 工厂方法扩展点 | 暂无多平台需求，但抽象层不删 | +| 探活机制 | 与 K8s/Ray 一致，复用现有逻辑 | 统一运维 | +| 重试策略 | 5xx 指数退避（最多 3 次），4xx 直接返回 | 有限重试，避免无限等待 | +| 数据面连通 | `endpoint_template` 直接作为 `host_ip`，`port_mapping` 写死 | 复用现有 proxy 链路，与 K8s 一致 | +| stop 语义 | 不支持 `pause_resume`（501）时降级为 `delete` | 保证 stop 语义可达 | +| 租约管理 | 不设 `timeout_seconds`，使用平台默认值；不实现 renew | Rock 自身的 `auto_archive_seconds` / `auto_delete_seconds` 控制生命周期 | +| Region / Class | 全局配置，所有沙箱使用同一值 | 简化 Phase 1 | + +## 9. 与现有 Operator 对比 + +| 维度 | RayOperator | K8sOperator | OpenSandboxOperator | RemoteOperator | +|------|------------|------------|--------------------|---------------| +| 后端 | Ray Actor | K8s CRD | OpenSandbox SDK | HTTP REST API | +| Provider 抽象 | 无 | K8sProvider Protocol | 无 | RemoteProvider Protocol | +| 需要 Ray | 是 | 否 | 否 | 否 | +| 支持 Scheduler | 是 | 是 | 否 | 否 | +| supports_running_delete | False | False | True | True | +| restart | 支持 | 不支持 | 不支持 | 不支持 | +| Template API | 不支持 | 支持 (Pool CRD) | 不支持 | 支持 (HTTP REST，可选) | +| Proxy 层 | Rocklet RPC | Rocklet RPC | OpenSandboxBackend | Rocklet RPC (复用) | + +## 10. 后续扩展 + +当前 `RemoteProvider` Protocol 和 `_create_provider()` 工厂方法已作为扩展点保留。后续如需接入其他平台： + +1. 新增 Provider 实现 `RemoteProvider` Protocol +2. 在 `_create_provider()` 中新增分支 +3. 可有独立的配置子结构 diff --git a/docs/proposals/sandbox-next.yaml b/docs/proposals/sandbox-next.yaml new file mode 100644 index 0000000000..88e6c39088 --- /dev/null +++ b/docs/proposals/sandbox-next.yaml @@ -0,0 +1,1319 @@ +openapi: 3.0.0 +info: + version: 1.0.0 + title: Sandbox Gateway API (native /v1) + description: > + infra 层统一控制面入口的原生 API。对平台层屏蔽物理集群与底座实现， + 覆盖沙箱、模板、快照的生命周期。数据面流量不经过 Gateway，响应只返回 + 数据面坐标与访问凭证。设计详见 docs/proposals/20260813-gateway-api.md。 + + E2B 兼容表面（/sandboxes、/v2、/v3）不在本 spec 内：它跟随上游协议、 + 走独立监听与独立域名，直接以上游 openapi 为准。 + +servers: + - url: https://api.{region}.sandbox.internal + description: 每个 region 独立 endpoint；region 由部署边界确定，不编码进 ID + variables: + region: + default: cn-hangzhou + +security: + - ApiKeyAuth: [] + - AccessTokenAuth: [] + +tags: + - name: sandboxes + description: 沙箱生命周期 + - name: templates + description: 模板（可用于创建沙箱的启动配置引用；不暴露 build 概念） + - name: snapshots + description: 从运行沙箱导出的可复用产物 + - name: regions + description: region 与 class 能力发现 + - name: passthrough + description: 长尾场景的受控透传（非稳定契约） + +paths: + /v1/sandboxes: + post: + operationId: createSandbox + tags: [sandboxes] + summary: 创建沙箱 + description: > + 幂等：同一 tenant + request_id 重复请求返回同一对象与相同状态码。 + 同步底座（capability sync_create）返回 201 且 state=running； + 声明式底座返回 202 且 state=creating，需轮询 GET 至 running。 + requestBody: + required: true + content: + application/json: + schema: + $ref: "#/components/schemas/NewSandbox" + responses: + "201": + description: 已创建并就绪（同步底座） + content: + application/json: + schema: + $ref: "#/components/schemas/Sandbox" + "202": + description: 已受理，尚未就绪（声明式底座） + content: + application/json: + schema: + $ref: "#/components/schemas/Sandbox" + "400": + $ref: "#/components/responses/BadRequest" + "401": + $ref: "#/components/responses/Unauthenticated" + "403": + $ref: "#/components/responses/PermissionDenied" + "409": + $ref: "#/components/responses/Conflict" + "429": + $ref: "#/components/responses/RateLimited" + "501": + $ref: "#/components/responses/NotImplemented" + "503": + $ref: "#/components/responses/Unavailable" + "500": + $ref: "#/components/responses/Internal" + get: + operationId: listSandboxes + tags: [sandboxes] + summary: 列举沙箱 + description: 省略 region 时在凭证允许范围内跨 region 扇出；部分不可用返回 partial=true。 + parameters: + - $ref: "#/components/parameters/regionQuery" + - name: state + in: query + description: 按状态过滤，可重复 + schema: + $ref: "#/components/schemas/SandboxState" + - name: class + in: query + schema: + type: string + - name: metadata + in: query + description: 形如 metadata.key=value 的元数据过滤，可重复 + schema: + type: string + - $ref: "#/components/parameters/pageSize" + - $ref: "#/components/parameters/pageToken" + responses: + "200": + description: 沙箱列表 + content: + application/json: + schema: + $ref: "#/components/schemas/SandboxList" + "401": + $ref: "#/components/responses/Unauthenticated" + "429": + $ref: "#/components/responses/RateLimited" + "500": + $ref: "#/components/responses/Internal" + + /v1/sandboxes/{sandbox_id}: + parameters: + - $ref: "#/components/parameters/sandboxId" + get: + operationId: getSandbox + tags: [sandboxes] + summary: 查询沙箱详情 + responses: + "200": + description: 沙箱详情 + content: + application/json: + schema: + $ref: "#/components/schemas/SandboxDetail" + "401": + $ref: "#/components/responses/Unauthenticated" + "404": + $ref: "#/components/responses/NotFound" + "429": + $ref: "#/components/responses/RateLimited" + "503": + $ref: "#/components/responses/Unavailable" + "500": + $ref: "#/components/responses/Internal" + delete: + operationId: deleteSandbox + tags: [sandboxes] + summary: 删除沙箱（受理即返回） + responses: + "202": + description: 删除意图已受理 + "401": + $ref: "#/components/responses/Unauthenticated" + "404": + $ref: "#/components/responses/NotFound" + "429": + $ref: "#/components/responses/RateLimited" + "500": + $ref: "#/components/responses/Internal" + + /v1/sandboxes/{sandbox_id}/pause: + parameters: + - $ref: "#/components/parameters/sandboxId" + post: + operationId: pauseSandbox + tags: [sandboxes] + summary: 暂停沙箱 + description: 需 class 具备 pause_resume capability，否则 501。受理即返回，终态经 GET 轮询。 + responses: + "202": + description: 暂停意图已受理，state=pausing + content: + application/json: + schema: + $ref: "#/components/schemas/SandboxDetail" + "401": + $ref: "#/components/responses/Unauthenticated" + "404": + $ref: "#/components/responses/NotFound" + "409": + $ref: "#/components/responses/Conflict" + "429": + $ref: "#/components/responses/RateLimited" + "501": + $ref: "#/components/responses/NotImplemented" + "500": + $ref: "#/components/responses/Internal" + + /v1/sandboxes/{sandbox_id}/resume: + parameters: + - $ref: "#/components/parameters/sandboxId" + post: + operationId: resumeSandbox + tags: [sandboxes] + summary: 从暂停态恢复 + requestBody: + required: false + content: + application/json: + schema: + $ref: "#/components/schemas/ResumeRequest" + responses: + "200": + description: 已恢复（同步底座） + content: + application/json: + schema: + $ref: "#/components/schemas/SandboxDetail" + "202": + description: 恢复意图已受理，state=resuming（声明式底座） + content: + application/json: + schema: + $ref: "#/components/schemas/SandboxDetail" + "401": + $ref: "#/components/responses/Unauthenticated" + "404": + $ref: "#/components/responses/NotFound" + "409": + $ref: "#/components/responses/Conflict" + "429": + $ref: "#/components/responses/RateLimited" + "501": + $ref: "#/components/responses/NotImplemented" + "500": + $ref: "#/components/responses/Internal" + + /v1/sandboxes/{sandbox_id}/connect: + parameters: + - $ref: "#/components/parameters/sandboxId" + post: + operationId: connectSandbox + tags: [sandboxes] + summary: 获取连接坐标，必要时恢复并续租 + description: > + 复合操作：running 则刷新租约并返回坐标；paused 且 auto_resume=true + 且具备 pause_resume 则先恢复再返回；paused 且 auto_resume=false 返回 409。 + requestBody: + required: true + content: + application/json: + schema: + $ref: "#/components/schemas/ConnectRequest" + responses: + "200": + description: 已可连接 + content: + application/json: + schema: + $ref: "#/components/schemas/Sandbox" + "401": + $ref: "#/components/responses/Unauthenticated" + "404": + $ref: "#/components/responses/NotFound" + "409": + $ref: "#/components/responses/Conflict" + "429": + $ref: "#/components/responses/RateLimited" + "500": + $ref: "#/components/responses/Internal" + + /v1/sandboxes/{sandbox_id}/renew: + parameters: + - $ref: "#/components/parameters/sandboxId" + post: + operationId: renewSandbox + tags: [sandboxes] + summary: 延长租约（只向后推进） + description: candidate = now + timeout_seconds；不晚于当前 expire_at 时幂等返回。 + requestBody: + required: true + content: + application/json: + schema: + $ref: "#/components/schemas/RenewRequest" + responses: + "200": + description: 已续租 + content: + application/json: + schema: + $ref: "#/components/schemas/SandboxDetail" + "401": + $ref: "#/components/responses/Unauthenticated" + "404": + $ref: "#/components/responses/NotFound" + "429": + $ref: "#/components/responses/RateLimited" + "500": + $ref: "#/components/responses/Internal" + + /v1/sandboxes/{sandbox_id}/timeout: + parameters: + - $ref: "#/components/parameters/sandboxId" + post: + operationId: setSandboxTimeout + tags: [sandboxes] + summary: 绝对设置到期时间（可缩短） + description: 与 renew 不同，timeout 是绝对覆盖，允许把 expire_at 提前。 + requestBody: + required: true + content: + application/json: + schema: + $ref: "#/components/schemas/TimeoutRequest" + responses: + "200": + description: 已设置 + content: + application/json: + schema: + $ref: "#/components/schemas/SandboxDetail" + "401": + $ref: "#/components/responses/Unauthenticated" + "404": + $ref: "#/components/responses/NotFound" + "429": + $ref: "#/components/responses/RateLimited" + "500": + $ref: "#/components/responses/Internal" + + /v1/sandboxes/{sandbox_id}/snapshots: + parameters: + - $ref: "#/components/parameters/sandboxId" + post: + operationId: createSnapshot + tags: [snapshots] + summary: 从运行实例导出快照 + description: > + 需 class 具备 snapshot capability。快照不保留进程内存态（内存态保留是 + pause_resume 的语义）。受理即返回，就绪经 GET /v1/snapshots/{id} 轮询。 + requestBody: + required: false + content: + application/json: + schema: + $ref: "#/components/schemas/CreateSnapshotRequest" + responses: + "202": + description: 快照意图已受理 + content: + application/json: + schema: + $ref: "#/components/schemas/Snapshot" + "401": + $ref: "#/components/responses/Unauthenticated" + "404": + $ref: "#/components/responses/NotFound" + "409": + $ref: "#/components/responses/Conflict" + "429": + $ref: "#/components/responses/RateLimited" + "501": + $ref: "#/components/responses/NotImplemented" + "500": + $ref: "#/components/responses/Internal" + + /v1/sandboxes/{sandbox_id}/logs: + parameters: + - $ref: "#/components/parameters/sandboxId" + get: + operationId: getSandboxLogs + tags: [sandboxes] + summary: 查询沙箱日志 + description: 依赖 class 的 logs capability。 + parameters: + - name: start + in: query + description: 起始时间（unix 秒） + schema: + type: integer + format: int64 + - name: end + in: query + description: 结束时间（unix 秒） + schema: + type: integer + format: int64 + - name: level + in: query + schema: + $ref: "#/components/schemas/LogLevel" + - $ref: "#/components/parameters/pageSize" + - $ref: "#/components/parameters/pageToken" + responses: + "200": + description: 日志条目 + content: + application/json: + schema: + $ref: "#/components/schemas/LogList" + "401": + $ref: "#/components/responses/Unauthenticated" + "404": + $ref: "#/components/responses/NotFound" + "429": + $ref: "#/components/responses/RateLimited" + "501": + $ref: "#/components/responses/NotImplemented" + "500": + $ref: "#/components/responses/Internal" + + /v1/sandboxes/{sandbox_id}/metrics: + parameters: + - $ref: "#/components/parameters/sandboxId" + get: + operationId: getSandboxMetrics + tags: [sandboxes] + summary: 查询沙箱指标 + description: 依赖 class 的 metrics capability。 + parameters: + - name: start + in: query + schema: + type: integer + format: int64 + - name: end + in: query + schema: + type: integer + format: int64 + responses: + "200": + description: 时间序列指标 + content: + application/json: + schema: + type: array + items: + $ref: "#/components/schemas/SandboxMetric" + "401": + $ref: "#/components/responses/Unauthenticated" + "404": + $ref: "#/components/responses/NotFound" + "429": + $ref: "#/components/responses/RateLimited" + "501": + $ref: "#/components/responses/NotImplemented" + "500": + $ref: "#/components/responses/Internal" + + /v1/templates: + post: + operationId: createTemplate + tags: [templates] + summary: 创建模板 + description: > + 需 class 具备 template_create capability，否则 501。受理即返回， + 通过 GET 轮询 status 至 ready。Gateway 不暴露 build 概念。 + requestBody: + required: true + content: + application/json: + schema: + $ref: "#/components/schemas/NewTemplate" + responses: + "202": + description: 模板创建已受理 + content: + application/json: + schema: + $ref: "#/components/schemas/Template" + "400": + $ref: "#/components/responses/BadRequest" + "401": + $ref: "#/components/responses/Unauthenticated" + "403": + $ref: "#/components/responses/PermissionDenied" + "409": + $ref: "#/components/responses/Conflict" + "429": + $ref: "#/components/responses/RateLimited" + "501": + $ref: "#/components/responses/NotImplemented" + "500": + $ref: "#/components/responses/Internal" + get: + operationId: listTemplates + tags: [templates] + summary: 列举模板 + parameters: + - $ref: "#/components/parameters/regionQuery" + - $ref: "#/components/parameters/pageSize" + - $ref: "#/components/parameters/pageToken" + responses: + "200": + description: 模板列表 + content: + application/json: + schema: + $ref: "#/components/schemas/TemplateList" + "401": + $ref: "#/components/responses/Unauthenticated" + "429": + $ref: "#/components/responses/RateLimited" + "500": + $ref: "#/components/responses/Internal" + + /v1/templates/{template_id}: + parameters: + - $ref: "#/components/parameters/templateId" + get: + operationId: getTemplate + tags: [templates] + summary: 查询模板 + description: 查询路由回模板的 origin cluster。 + responses: + "200": + description: 模板详情 + content: + application/json: + schema: + $ref: "#/components/schemas/Template" + "401": + $ref: "#/components/responses/Unauthenticated" + "404": + $ref: "#/components/responses/NotFound" + "429": + $ref: "#/components/responses/RateLimited" + "503": + $ref: "#/components/responses/Unavailable" + "500": + $ref: "#/components/responses/Internal" + delete: + operationId: deleteTemplate + tags: [templates] + summary: 删除模板 + description: 需 template_create capability，否则 501。 + responses: + "202": + description: 删除意图已受理 + "401": + $ref: "#/components/responses/Unauthenticated" + "404": + $ref: "#/components/responses/NotFound" + "429": + $ref: "#/components/responses/RateLimited" + "501": + $ref: "#/components/responses/NotImplemented" + "500": + $ref: "#/components/responses/Internal" + + /v1/snapshots: + get: + operationId: listSnapshots + tags: [snapshots] + summary: 列举快照 + parameters: + - name: sandbox_id + in: query + description: 按来源沙箱过滤 + schema: + type: string + - $ref: "#/components/parameters/pageSize" + - $ref: "#/components/parameters/pageToken" + responses: + "200": + description: 快照列表 + content: + application/json: + schema: + $ref: "#/components/schemas/SnapshotList" + "401": + $ref: "#/components/responses/Unauthenticated" + "429": + $ref: "#/components/responses/RateLimited" + "500": + $ref: "#/components/responses/Internal" + + /v1/snapshots/{snapshot_id}: + parameters: + - $ref: "#/components/parameters/snapshotId" + get: + operationId: getSnapshot + tags: [snapshots] + summary: 查询快照状态 + responses: + "200": + description: 快照详情 + content: + application/json: + schema: + $ref: "#/components/schemas/Snapshot" + "401": + $ref: "#/components/responses/Unauthenticated" + "404": + $ref: "#/components/responses/NotFound" + "429": + $ref: "#/components/responses/RateLimited" + "500": + $ref: "#/components/responses/Internal" + delete: + operationId: deleteSnapshot + tags: [snapshots] + summary: 删除快照 + responses: + "202": + description: 删除意图已受理 + "401": + $ref: "#/components/responses/Unauthenticated" + "404": + $ref: "#/components/responses/NotFound" + "429": + $ref: "#/components/responses/RateLimited" + "500": + $ref: "#/components/responses/Internal" + + /v1/regions: + get: + operationId: listRegions + tags: [regions] + summary: region 与 class 能力发现 + description: 平台层据此做灰度与降级，不应硬编码 class 列表或靠试错发现能力缺失。 + responses: + "200": + description: region 能力列表 + content: + application/json: + schema: + $ref: "#/components/schemas/RegionList" + "401": + $ref: "#/components/responses/Unauthenticated" + "500": + $ref: "#/components/responses/Internal" + + /v1/passthrough/{route_name}/{proxy_path}: + parameters: + - name: route_name + in: path + required: true + description: 必须在 Gateway 配置的白名单中 + schema: + type: string + - name: proxy_path + in: path + required: true + description: 透传到底座的剩余路径（多段） + schema: + type: string + post: + operationId: passthrough + tags: [passthrough] + summary: 受控透传（非稳定契约） + description: > + 用于抽象未覆盖的长尾场景。Gateway 只做鉴权、限流、路由与 trace_id 注入， + 不转换 body。响应恒带 X-Gateway-Unstable-Contract: true，且不做 schema 承诺。 + x-gateway-unstable-contract: true + requestBody: + required: false + content: + application/octet-stream: + schema: + type: string + format: binary + responses: + "200": + description: 透传响应（形状由目标底座决定，不做承诺） + "401": + $ref: "#/components/responses/Unauthenticated" + "403": + $ref: "#/components/responses/PermissionDenied" + "404": + $ref: "#/components/responses/NotFound" + "429": + $ref: "#/components/responses/RateLimited" + "503": + $ref: "#/components/responses/Unavailable" + +components: + securitySchemes: + ApiKeyAuth: + type: apiKey + in: header + name: X-Api-Key + AccessTokenAuth: + type: http + scheme: bearer + + parameters: + sandboxId: + name: sandbox_id + in: path + required: true + description: 不透明字符串，客户端不得解析其内部结构 + schema: + type: string + templateId: + name: template_id + in: path + required: true + schema: + type: string + snapshotId: + name: snapshot_id + in: path + required: true + schema: + type: string + regionQuery: + name: region + in: query + description: 省略时在凭证允许范围内跨 region 扇出 + schema: + type: string + pageSize: + name: page_size + in: query + schema: + type: integer + format: int32 + default: 100 + minimum: 1 + maximum: 1000 + pageToken: + name: page_token + in: query + description: 不透明分页游标，禁止客户端构造 + schema: + type: string + + responses: + BadRequest: + description: 请求非法（字段缺失、取值非法、超出预算） + content: + application/json: + schema: + $ref: "#/components/schemas/Error" + Unauthenticated: + description: 凭证缺失或无效 + content: + application/json: + schema: + $ref: "#/components/schemas/Error" + PermissionDenied: + description: 凭证有效但无权访问该 region / class / 对象 + content: + application/json: + schema: + $ref: "#/components/schemas/Error" + NotFound: + description: 对象不存在或已删除 + content: + application/json: + schema: + $ref: "#/components/schemas/Error" + Conflict: + description: 幂等键冲突或当前状态不允许该操作 + content: + application/json: + schema: + $ref: "#/components/schemas/Error" + RateLimited: + description: 实例级限流（resource_exhausted）或底座配额不足（quota_exceeded） + headers: + Retry-After: + description: 建议重试等待秒数（仅 resource_exhausted 附带） + schema: + type: integer + content: + application/json: + schema: + $ref: "#/components/schemas/Error" + NotImplemented: + description: 目标 class 不具备该 capability + content: + application/json: + schema: + $ref: "#/components/schemas/Error" + Unavailable: + description: 底座或 region 暂时不可达 + content: + application/json: + schema: + $ref: "#/components/schemas/Error" + Internal: + description: 未分类内部错误 + content: + application/json: + schema: + $ref: "#/components/schemas/Error" + + schemas: + Error: + type: object + required: + - code + - message + - retryable + properties: + code: + type: string + description: 稳定字符串错误码，调用方唯一可据此分支的字段 + enum: + - invalid_argument + - unauthenticated + - permission_denied + - not_found + - already_exists + - failed_precondition + - resource_exhausted + - quota_exceeded + - not_implemented + - backend_unavailable + - region_unavailable + - deadline_exceeded + - internal + message: + type: string + description: 面向人的描述，措辞可变，不得据此判定 + request_id: + type: string + trace_id: + type: string + retryable: + type: boolean + description: 同样的请求原样重试是否有意义 + + Resources: + type: object + properties: + vcpu: + type: integer + format: int32 + minimum: 1 + memory_mb: + type: integer + format: int32 + minimum: 128 + disk_mb: + type: integer + format: int32 + minimum: 0 + + SandboxState: + type: string + description: 归一化后的统一状态词表 + enum: + - creating + - running + - pausing + - paused + - resuming + - failed + + OnTimeout: + type: string + enum: + - delete + - pause + default: delete + + Capability: + type: string + enum: + - pause_resume + - snapshot + - template_create + - prewarm_pool + - sync_create + - metrics + - logs + + SandboxAccess: + type: object + description: 数据面坐标与访问凭证；Gateway 不代理数据面流量 + properties: + endpoint_template: + type: string + description: 唯一数据面契约，客户端把 {port} 替换为目标端口 + example: https://{port}-s485cf4a26daa06baf87e7636e63ca8873f.cn-hangzhou.sandbox.example.com + agent_token: + type: string + description: 访问沙箱内 agent 的凭证；未启用时为空串 + traffic_token: + type: string + nullable: true + description: 经代理访问沙箱的凭证；未启用时为 null + route_hint: + type: string + description: 不透明首访路由提示，可原样回传数据面 + + NewSandbox: + type: object + required: + - request_id + - region + - class + properties: + request_id: + type: string + description: 幂等键，参与 uid 派生与集群选择 + region: + type: string + class: + type: string + description: 沙箱形态，如 headless-vm / gui-pod + template_id: + type: string + description: 与 snapshot_id 二选一 + snapshot_id: + type: string + description: 与 template_id 二选一 + resources: + $ref: "#/components/schemas/Resources" + timeout_seconds: + type: integer + format: int32 + description: 缺省用 class 的 default_timeout_seconds + on_timeout: + $ref: "#/components/schemas/OnTimeout" + env_vars: + type: object + additionalProperties: + type: string + metadata: + type: object + additionalProperties: + type: string + + Sandbox: + type: object + description: 创建 / 连接的响应 + required: + - sandbox_id + - region + - class + - state + properties: + sandbox_id: + type: string + region: + type: string + class: + type: string + template_id: + type: string + state: + $ref: "#/components/schemas/SandboxState" + state_uncertain: + type: boolean + description: 控制面无法确认节点事实时为 true（呈现最近稳态） + resources: + $ref: "#/components/schemas/Resources" + created_at: + type: string + format: date-time + expire_at: + type: string + format: date-time + access: + $ref: "#/components/schemas/SandboxAccess" + capabilities: + type: array + items: + $ref: "#/components/schemas/Capability" + + SandboxDetail: + type: object + description: 查询详情；列表项是其子集（不含 access 凭证） + required: + - sandbox_id + - region + - class + - state + properties: + sandbox_id: + type: string + region: + type: string + class: + type: string + template_id: + type: string + state: + $ref: "#/components/schemas/SandboxState" + state_uncertain: + type: boolean + resources: + $ref: "#/components/schemas/Resources" + on_timeout: + $ref: "#/components/schemas/OnTimeout" + metadata: + type: object + additionalProperties: + type: string + created_at: + type: string + format: date-time + expire_at: + type: string + format: date-time + capabilities: + type: array + items: + $ref: "#/components/schemas/Capability" + + SandboxList: + type: object + required: + - sandboxes + - partial + properties: + sandboxes: + type: array + items: + $ref: "#/components/schemas/SandboxDetail" + next_page_token: + type: string + description: 空串表示结束 + partial: + type: boolean + description: 部分集群不可用时为 true + unavailable_scopes: + type: array + description: 不可用范围的不透明标识 + items: + type: string + + ConnectRequest: + type: object + required: + - timeout_seconds + properties: + timeout_seconds: + type: integer + format: int32 + auto_resume: + type: boolean + default: true + + ResumeRequest: + type: object + properties: + timeout_seconds: + type: integer + format: int32 + + RenewRequest: + type: object + required: + - timeout_seconds + properties: + timeout_seconds: + type: integer + format: int32 + minimum: 1 + + TimeoutRequest: + type: object + required: + - timeout_seconds + properties: + timeout_seconds: + type: integer + format: int32 + description: 绝对到期 = now + timeout_seconds，允许缩短 + + CreateSnapshotRequest: + type: object + properties: + name: + type: string + + SnapshotStatus: + type: string + enum: + - pending + - running + - ready + - error + + Snapshot: + type: object + required: + - snapshot_id + - status + properties: + snapshot_id: + type: string + sandbox_id: + type: string + description: 来源沙箱 + status: + $ref: "#/components/schemas/SnapshotStatus" + failure: + $ref: "#/components/schemas/Failure" + created_at: + type: string + format: date-time + ready_at: + type: string + format: date-time + nullable: true + + SnapshotList: + type: object + required: + - snapshots + properties: + snapshots: + type: array + items: + $ref: "#/components/schemas/Snapshot" + next_page_token: + type: string + + Failure: + type: object + nullable: true + properties: + message: + type: string + + NewTemplate: + type: object + required: + - request_id + - region + - class + - name + properties: + request_id: + type: string + region: + type: string + class: + type: string + name: + type: string + maxLength: 128 + resources: + $ref: "#/components/schemas/Resources" + from_image: + type: string + description: 仅 template_create capability 的 class 支持 + env_vars: + type: object + additionalProperties: + type: string + + TemplateStatus: + type: string + enum: + - pending + - preparing + - ready + - failed + + Template: + type: object + required: + - template_id + - region + - class + - status + properties: + template_id: + type: string + name: + type: string + region: + type: string + class: + type: string + resources: + $ref: "#/components/schemas/Resources" + status: + $ref: "#/components/schemas/TemplateStatus" + failure: + $ref: "#/components/schemas/Failure" + created_at: + type: string + format: date-time + updated_at: + type: string + format: date-time + ready_at: + type: string + format: date-time + nullable: true + + TemplateList: + type: object + required: + - templates + properties: + templates: + type: array + items: + $ref: "#/components/schemas/Template" + next_page_token: + type: string + + ResourceLimits: + type: object + properties: + max_vcpu: + type: integer + format: int32 + max_memory_mb: + type: integer + format: int32 + max_disk_mb: + type: integer + format: int32 + + SandboxClass: + type: object + required: + - class + - status + - capabilities + properties: + class: + type: string + status: + $ref: "#/components/schemas/AvailabilityStatus" + capabilities: + type: array + items: + $ref: "#/components/schemas/Capability" + resource_limits: + $ref: "#/components/schemas/ResourceLimits" + default_timeout_seconds: + type: integer + format: int32 + max_timeout_seconds: + type: integer + format: int32 + + AvailabilityStatus: + type: string + enum: + - available + - degraded + - unavailable + + Region: + type: object + required: + - region + - status + - classes + properties: + region: + type: string + status: + $ref: "#/components/schemas/AvailabilityStatus" + classes: + type: array + items: + $ref: "#/components/schemas/SandboxClass" + + RegionList: + type: object + required: + - regions + properties: + regions: + type: array + items: + $ref: "#/components/schemas/Region" + + LogLevel: + type: string + enum: + - debug + - info + - warn + - error + + LogEntry: + type: object + properties: + timestamp: + type: string + format: date-time + level: + $ref: "#/components/schemas/LogLevel" + message: + type: string + fields: + type: object + additionalProperties: + type: string + + LogList: + type: object + required: + - logs + properties: + logs: + type: array + items: + $ref: "#/components/schemas/LogEntry" + next_page_token: + type: string + + SandboxMetric: + type: object + properties: + timestamp_unix: + type: integer + format: int64 + cpu_count: + type: integer + format: int32 + cpu_used_pct: + type: number + format: float + mem_used: + type: integer + format: int64 + mem_total: + type: integer + format: int64 + disk_used: + type: integer + format: int64 + disk_total: + type: integer + format: int64 diff --git a/rock/admin/main.py b/rock/admin/main.py index 72d61b1bf2..4dcae674a0 100644 --- a/rock/admin/main.py +++ b/rock/admin/main.py @@ -214,6 +214,7 @@ async def lifespan(app: FastAPI): template_table=template_table, k8s_config=rock_config.k8s, opensandbox_config=rock_config.opensandbox, + remote_config=rock_config.remote, ) operator = OperatorFactory.create_operator(operator_context) diff --git a/rock/config.py b/rock/config.py index 226934647b..86cf1331dd 100644 --- a/rock/config.py +++ b/rock/config.py @@ -442,6 +442,54 @@ class OpenSandboxConfig: """Default request timeout (seconds) for SDK calls.""" +@dataclass +class RemoteOperatorConfig: + """Configuration for the Remote operator backend. + + Used when ``runtime.operator_type == "remote"``. Rock delegates sandbox + lifecycle to a remote platform (SandboxNext) via HTTP REST API. The first + and currently only provider is ``SandboxNextProvider``. + See docs/proposals/remote-operator.md. + """ + + provider: str = "sandbox_next" + """Provider type. Currently only "sandbox_next" is supported.""" + + endpoint: str = "" + """SandboxNext Gateway API domain.""" + + api_key: str | None = None + """X-Api-Key header authentication.""" + + access_token: str | None = None + """Bearer token authentication.""" + + protocol: str = "https" + """Connection protocol: "http" or "https".""" + + default_timeout: int = 600 + """HTTP request timeout in seconds.""" + + region: str = "cn-hangzhou" + """SandboxNext region.""" + + sandbox_class: str = "headless-vm" + """SandboxNext class (sandbox form factor).""" + + namespace: str = "rock" + """Namespace/label scope for sandboxes created by this Rock instance.""" + + state_mapping: dict | None = None + """Override the default SandboxNext state -> Rock State mapping.""" + + provider_options: dict = field(default_factory=dict) + """Provider-specific extra configuration (e.g. retry settings).""" + + def __post_init__(self): + if not self.endpoint: + raise ValueError("RemoteOperatorConfig.endpoint is required") + + @dataclass class RuntimeConfig: enable_auto_clear: bool = False @@ -561,6 +609,7 @@ class RockConfig: lifecycle: SandboxLifecycleConfig = field(default_factory=SandboxLifecycleConfig) runtime: RuntimeConfig = field(default_factory=RuntimeConfig) opensandbox: OpenSandboxConfig = field(default_factory=OpenSandboxConfig) + remote: RemoteOperatorConfig | None = None proxy_service: ProxyServiceConfig = field(default_factory=ProxyServiceConfig) aes_encrypt_key: str | None = None scheduler: SchedulerConfig = field(default_factory=SchedulerConfig) @@ -630,6 +679,8 @@ def from_env(cls, config_path: str | None = None): kwargs["runtime"] = RuntimeConfig(**config["runtime"]) if "opensandbox" in config: kwargs["opensandbox"] = OpenSandboxConfig(**config["opensandbox"]) + if "remote" in config: + kwargs["remote"] = RemoteOperatorConfig(**config["remote"]) if "proxy_service" in config: kwargs["proxy_service"] = ProxyServiceConfig(**config["proxy_service"]) if "aes_encrypt_key" in config: diff --git a/rock/sandbox/operator/factory.py b/rock/sandbox/operator/factory.py index aced0c95d7..37e64db8e3 100644 --- a/rock/sandbox/operator/factory.py +++ b/rock/sandbox/operator/factory.py @@ -4,13 +4,14 @@ from typing import Any from rock.admin.core.ray_service import RayService -from rock.config import K8sConfig, OpenSandboxConfig, RuntimeConfig +from rock.config import K8sConfig, OpenSandboxConfig, RemoteOperatorConfig, RuntimeConfig from rock.logger import init_logger from rock.sandbox.operator.abstract import AbstractOperator from rock.sandbox.operator.k8s.operator import K8sOperator from rock.sandbox.operator.k8s.provider import TemplateFiberPoolLookup from rock.sandbox.operator.opensandbox.operator import OpenSandboxOperator from rock.sandbox.operator.ray import RayOperator +from rock.sandbox.operator.remote.operator import RemoteOperator from rock.utils.providers.nacos_provider import NacosConfigProvider from rock.utils.providers.redis_provider import RedisProvider @@ -34,7 +35,7 @@ def operator_supports_scheduler(operator_type: str) -> bool: the scheduler discovers Ray workers and dispatches every task through their Rocklet endpoints. Other operators retain their existing behavior. """ - return operator_type.lower() != "opensandbox" + return operator_type.lower() not in ("opensandbox", "remote") @dataclass @@ -55,6 +56,8 @@ class OperatorContext: template_table: TemplateFiberPoolLookup | None = None # OpenSandbox operator dependencies opensandbox_config: OpenSandboxConfig | None = None + # Remote operator dependencies + remote_config: RemoteOperatorConfig | None = None # Future operator dependencies can be added here without breaking existing code extra_params: dict[str, Any] = field(default_factory=dict) @@ -111,5 +114,15 @@ def create_operator(context: OperatorContext) -> AbstractOperator: if context.nacos_provider is not None: opensandbox_operator.set_nacos_provider(context.nacos_provider) return opensandbox_operator + elif operator_type == "remote": + if context.remote_config is None: + raise ValueError("RemoteOperatorConfig is required for RemoteOperator") + logger.info("Creating RemoteOperator") + remote_operator = RemoteOperator(remote_config=context.remote_config) + if context.redis_provider is not None: + remote_operator.set_redis_provider(context.redis_provider) + if context.nacos_provider is not None: + remote_operator.set_nacos_provider(context.nacos_provider) + return remote_operator else: - raise ValueError(f"Unsupported operator type: {operator_type}. Supported types: ray, k8s, opensandbox") + raise ValueError(f"Unsupported operator type: {operator_type}. Supported types: ray, k8s, opensandbox, remote") diff --git a/rock/sandbox/operator/remote/__init__.py b/rock/sandbox/operator/remote/__init__.py new file mode 100644 index 0000000000..e08f031e6f --- /dev/null +++ b/rock/sandbox/operator/remote/__init__.py @@ -0,0 +1 @@ +"""Remote operator package — delegates sandbox lifecycle to a remote platform.""" diff --git a/rock/sandbox/operator/remote/constants.py b/rock/sandbox/operator/remote/constants.py new file mode 100644 index 0000000000..eb9ad55013 --- /dev/null +++ b/rock/sandbox/operator/remote/constants.py @@ -0,0 +1,8 @@ +"""Constants for the Remote operator.""" + +BACKEND_NAME = "sandbox_next" + +# extended_params keys +EXT_REMOTE_ID = "remote_sandbox_id" +EXT_ENDPOINT = "endpoint_template" +EXT_BACKEND = "backend" diff --git a/rock/sandbox/operator/remote/operator.py b/rock/sandbox/operator/remote/operator.py new file mode 100644 index 0000000000..acf191474d --- /dev/null +++ b/rock/sandbox/operator/remote/operator.py @@ -0,0 +1,114 @@ +"""RemoteOperator — manages sandboxes on a remote platform via a Provider. + +Delegates lifecycle calls to a RemoteProvider (SandboxNextProvider by default) +and handles Redis metadata merging, template API graceful fallback, and +extended_params bookkeeping. + +See docs/proposals/remote-operator.md for the full design. +""" + +from __future__ import annotations + +from typing import Any + +from rock.actions.sandbox.sandbox_info import SandboxInfo +from rock.common.constants import StopReason +from rock.config import RemoteOperatorConfig +from rock.deployments.config import DockerDeploymentConfig +from rock.logger import init_logger +from rock.sandbox.operator.abstract import AbstractOperator +from rock.sandbox.operator.remote.constants import EXT_REMOTE_ID +from rock.sandbox.operator.remote.providers.sandbox_next_provider import SandboxNextProvider +from rock.sdk.common.exceptions import BadRequestRockError + +logger = init_logger(__name__) + + +class RemoteOperator(AbstractOperator): + """Operator that manages sandboxes on a remote platform via HTTP REST API.""" + + supports_running_delete = True + + def __init__(self, remote_config: RemoteOperatorConfig): + self._config = remote_config + self._provider = self._create_provider(remote_config) + logger.info("Initialized RemoteOperator (provider=%s, endpoint=%s)", remote_config.provider, remote_config.endpoint) + + @staticmethod + def _create_provider(config: RemoteOperatorConfig): + """Factory method — the single extension point for new providers.""" + if config.provider == "sandbox_next": + return SandboxNextProvider(config) + raise ValueError(f"Unsupported remote provider: {config.provider}. Supported: sandbox_next") + + async def _resolve_remote_id(self, sandbox_id: str) -> str | None: + """Read the platform-assigned ID from Redis extended_params.""" + info = await self.get_sandbox_info_from_redis(sandbox_id) + if not info: + return None + return (info.get("extended_params") or {}).get(EXT_REMOTE_ID) + + async def submit(self, config: DockerDeploymentConfig, user_info: dict = {}) -> SandboxInfo: + return await self._provider.submit(config, user_info) + + async def restart(self, config: DockerDeploymentConfig, host_ip: str | None = None) -> SandboxInfo: + raise BadRequestRockError("RemoteOperator does not support container-reuse restart") + + async def get_status(self, sandbox_id: str) -> SandboxInfo | None: + redis_info = await self.get_sandbox_info_from_redis(sandbox_id) + if not redis_info: + return None + remote_id = (redis_info.get("extended_params") or {}).get(EXT_REMOTE_ID) + if not remote_id: + logger.warning("[%s] no remote_sandbox_id in cached info", sandbox_id) + return None + provider_info = await self._provider.get_status(remote_id) + if provider_info is None: + # Sandbox no longer exists on the remote platform + redis_info["state"] = "deleted" + return redis_info + # Merge: provider real-time status overrides Redis base fields; + # Redis user metadata is preserved (deep merge extended_params). + merged = dict(redis_info) + merged.update(provider_info) + merged_extended = dict(redis_info.get("extended_params") or {}) + merged_extended.update(provider_info.get("extended_params") or {}) + merged["extended_params"] = merged_extended + return merged + + async def stop(self, sandbox_id: str, reason: StopReason = StopReason.MANUAL) -> bool: + remote_id = await self._resolve_remote_id(sandbox_id) + if not remote_id: + raise BadRequestRockError(f"cannot resolve remote_sandbox_id for sandbox {sandbox_id}") + logger.info("[%s] remote stop -> pause (reason=%s)", sandbox_id, reason.value) + return await self._provider.stop(remote_id) + + async def delete(self, config: DockerDeploymentConfig, host_ip: str | None = None) -> bool: + sandbox_id = config.container_name + remote_id = (config.extended_params or {}).get(EXT_REMOTE_ID) or await self._resolve_remote_id(sandbox_id) + if not remote_id: + raise BadRequestRockError(f"cannot resolve remote_sandbox_id for sandbox {sandbox_id}") + logger.info("[%s] remote delete", sandbox_id) + return await self._provider.delete(remote_id) + + # ======================================================================== + # Template API — delegate to provider with graceful fallback + # ======================================================================== + + async def create_template(self, spec: Any) -> dict: + try: + return await self._provider.create_template(spec) + except NotImplementedError: + raise BadRequestRockError(f"template not supported on {type(self).__name__}") + + async def get_template_status(self, template_id: str) -> dict | None: + try: + return await self._provider.get_template_status(template_id) + except NotImplementedError: + raise BadRequestRockError(f"template not supported on {type(self).__name__}") + + async def delete_template(self, template_id: str) -> bool: + try: + return await self._provider.delete_template(template_id) + except NotImplementedError: + raise BadRequestRockError(f"template not supported on {type(self).__name__}") diff --git a/rock/sandbox/operator/remote/provider.py b/rock/sandbox/operator/remote/provider.py new file mode 100644 index 0000000000..82c87698ea --- /dev/null +++ b/rock/sandbox/operator/remote/provider.py @@ -0,0 +1,63 @@ +"""RemoteProvider Protocol — abstraction for remote sandbox platform providers. + +The Protocol decouples RemoteOperator from any specific platform SDK. +The first (and currently only) implementation is ``SandboxNextProvider``, +which talks to the SandboxNext Gateway REST API. + +See docs/proposals/remote-operator.md for the full design. +""" + +from __future__ import annotations + +from typing import Any, Protocol, runtime_checkable + +from rock.actions.sandbox.sandbox_info import SandboxInfo +from rock.deployments.config import DockerDeploymentConfig + + +@runtime_checkable +class RemoteProvider(Protocol): + """Protocol that every remote platform provider must implement. + + The operator delegates lifecycle calls to the provider and handles + Redis metadata merging itself, so the provider stays pure (no Redis). + """ + + # --- Lifecycle --- + + async def submit(self, config: DockerDeploymentConfig, user_info: dict) -> SandboxInfo: + """Create a sandbox on the remote platform. + + Returns a SandboxInfo with at least sandbox_id, host_ip, port_mapping, + auth_token, and extended_params (including the platform-assigned ID). + """ + ... # pragma: no cover + + async def get_status(self, remote_sandbox_id: str) -> SandboxInfo | None: + """Query the remote platform for current sandbox status. + + Returns None when the sandbox no longer exists (404). + """ + ... # pragma: no cover + + async def stop(self, remote_sandbox_id: str) -> bool: + """Pause the sandbox. May fall back to delete if pause is unsupported.""" + ... # pragma: no cover + + async def delete(self, remote_sandbox_id: str) -> bool: + """Delete the sandbox. Returns True on success or already-gone (404).""" + ... # pragma: no cover + + # --- Template API (optional; raise NotImplementedError if unsupported) --- + + async def create_template(self, spec: Any) -> dict: + """Create or reuse a template. 409 → idempotent GET fallback.""" + ... # pragma: no cover + + async def get_template_status(self, template_id: str) -> dict | None: + """Get template status. 404 → None.""" + ... # pragma: no cover + + async def delete_template(self, template_id: str) -> bool: + """Delete template. 404 → True (already gone).""" + ... # pragma: no cover diff --git a/rock/sandbox/operator/remote/providers/__init__.py b/rock/sandbox/operator/remote/providers/__init__.py new file mode 100644 index 0000000000..23fb00b987 --- /dev/null +++ b/rock/sandbox/operator/remote/providers/__init__.py @@ -0,0 +1 @@ +"""Remote platform provider implementations.""" diff --git a/rock/sandbox/operator/remote/providers/sandbox_next_provider.py b/rock/sandbox/operator/remote/providers/sandbox_next_provider.py new file mode 100644 index 0000000000..42e7093d58 --- /dev/null +++ b/rock/sandbox/operator/remote/providers/sandbox_next_provider.py @@ -0,0 +1,287 @@ +"""SandboxNext provider — talks to the SandboxNext Gateway REST API. + +Implements the RemoteProvider Protocol using httpx.AsyncClient. +See docs/proposals/sandbox-next.yaml for the OpenAPI spec. +""" + +from __future__ import annotations + +from typing import Any + +import httpx + +from rock.actions.sandbox.response import State +from rock.actions.sandbox.sandbox_info import SandboxInfo +from rock.config import RemoteOperatorConfig +from rock.deployments.config import DockerDeploymentConfig +from rock.deployments.constants import Port +from rock.logger import init_logger +from rock.sandbox.operator.remote.constants import EXT_BACKEND, EXT_ENDPOINT, EXT_REMOTE_ID, BACKEND_NAME + +logger = init_logger(__name__) + +# --- SandboxNext SandboxState -> Rock State --- + +_DEFAULT_STATE_MAP: dict[str, State] = { + "creating": State.PENDING, + "running": State.RUNNING, + "pausing": State.STOPPED, + "paused": State.STOPPED, + "resuming": State.PENDING, + "failed": State.STOPPED, +} + + +def _map_state(sn_state: str | None, state_map: dict[str, State] | None = None) -> State: + table = state_map or _DEFAULT_STATE_MAP + return table.get(sn_state or "", State.PENDING) + + +def _parse_mem_to_mb(mem: str) -> int: + """Convert docker-style memory string (``8g``/``4096m``/``2048``) to MB.""" + s = mem.strip().lower() + if not s: + return 0 + if s.endswith("g"): + return int(float(s[:-1]) * 1024) + if s.endswith("m"): + return int(float(s[:-1])) + return int(float(s)) + + +def _parse_disk_to_mb(disk: str | None) -> int: + """Convert docker-style disk string (``50G``/``51200M``) to MB.""" + if not disk: + return 0 + return _parse_mem_to_mb(disk) + + +class SandboxNextProvider: + """Provider that talks to the SandboxNext Gateway REST API.""" + + def __init__(self, config: RemoteOperatorConfig, *, client: httpx.AsyncClient | None = None): + self._config = config + self._state_map = config.state_mapping or _DEFAULT_STATE_MAP + self._retry_max = config.provider_options.get("retry_max", 3) + self._retry_backoff = config.provider_options.get("retry_backoff_base", 0.5) + + base_url = f"{config.protocol}://{config.endpoint}" + headers: dict[str, str] = {} + if config.api_key: + headers["X-Api-Key"] = config.api_key + if config.access_token: + headers["Authorization"] = f"Bearer {config.access_token}" + + self._client = client or httpx.AsyncClient( + base_url=base_url, + headers=headers, + timeout=config.default_timeout, + ) + logger.info("Initialized SandboxNextProvider (endpoint=%s, region=%s)", config.endpoint, config.region) + + # --- HTTP helpers --- + + async def _request(self, method: str, path: str, **kwargs) -> httpx.Response: + """Send an HTTP request with limited retry on 5xx errors.""" + response = await self._client.request(method, path, **kwargs) + retry_count = 0 + while response.status_code >= 500 and retry_count < self._retry_max: + retry_count += 1 + import asyncio + + await asyncio.sleep(self._retry_backoff * (2 ** (retry_count - 1))) + response = await self._client.request(method, path, **kwargs) + return response + + # --- Lifecycle --- + + async def submit(self, config: DockerDeploymentConfig, user_info: dict) -> SandboxInfo: + sandbox_id = config.container_name + user_id = user_info.get("user_id", "default") + experiment_id = user_info.get("experiment_id", "default") + namespace = user_info.get("namespace", "default") + + body: dict[str, Any] = { + "request_id": sandbox_id, + "region": self._config.region, + "class": self._config.sandbox_class, + "resources": { + "vcpu": int(config.cpus), + "memory_mb": _parse_mem_to_mb(config.memory), + "disk_mb": _parse_disk_to_mb(config.disk), + }, + "metadata": { + "rock_sandbox_id": sandbox_id or "", + "user_id": user_id, + "experiment_id": experiment_id, + "namespace": namespace, + }, + } + if config.env_vars: + body["env_vars"] = config.env_vars + + response = await self._request("POST", "/v1/sandboxes", json=body) + response.raise_for_status() + data = response.json() + + sn_id = data["sandbox_id"] + sn_state = data.get("state") + access = data.get("access") or {} + endpoint_template = access.get("endpoint_template", "") + agent_token = access.get("agent_token", "") + + logger.info("[%s] sandbox_next submitted, remote_id=%s, state=%s", sandbox_id, sn_id, sn_state) + + info: SandboxInfo = { + "sandbox_id": sandbox_id, + "image": config.image, + "cpus": config.cpus, + "memory": config.memory, + "user_id": user_id, + "experiment_id": experiment_id, + "namespace": namespace, + "state": _map_state(sn_state, self._state_map), + "host_ip": endpoint_template, + "port_mapping": { + Port.PROXY: 8000, + Port.SERVER: 8080, + Port.SSH: 22, + }, + "auth_token": agent_token, + "extended_params": { + EXT_BACKEND: BACKEND_NAME, + EXT_REMOTE_ID: sn_id, + EXT_ENDPOINT: endpoint_template, + }, + } + return info + + async def get_status(self, remote_sandbox_id: str) -> SandboxInfo | None: + response = await self._request("GET", f"/v1/sandboxes/{remote_sandbox_id}") + if response.status_code == 404: + return None + response.raise_for_status() + data = response.json() + + sn_state = data.get("state") + access = data.get("access") or {} + endpoint_template = access.get("endpoint_template", "") + agent_token = access.get("agent_token", "") + + info: SandboxInfo = { + "sandbox_id": remote_sandbox_id, + "state": _map_state(sn_state, self._state_map), + "host_ip": endpoint_template, + "port_mapping": { + Port.PROXY: 8000, + Port.SERVER: 8080, + Port.SSH: 22, + }, + "auth_token": agent_token, + "extended_params": { + EXT_BACKEND: BACKEND_NAME, + EXT_REMOTE_ID: remote_sandbox_id, + EXT_ENDPOINT: endpoint_template, + }, + } + return info + + async def stop(self, remote_sandbox_id: str) -> bool: + """Pause the sandbox. Falls back to delete if pause is unsupported (501).""" + response = await self._request("POST", f"/v1/sandboxes/{remote_sandbox_id}/pause") + if response.status_code == 501: + logger.info("[%s] pause not supported (501), falling back to delete", remote_sandbox_id) + return await self.delete(remote_sandbox_id) + response.raise_for_status() + return True + + async def delete(self, remote_sandbox_id: str) -> bool: + response = await self._request("DELETE", f"/v1/sandboxes/{remote_sandbox_id}") + if response.status_code == 404: + return True + response.raise_for_status() + return True + + # --- Template API --- + + async def create_template(self, spec: Any) -> dict: + body = self._template_spec_to_new(spec) + response = await self._request("POST", "/v1/templates", json=body) + if response.status_code == 409: + # Idempotent: fetch existing by request_id + request_id = body.get("request_id", "") + if request_id: + get_resp = await self._request("GET", f"/v1/templates/{request_id}") + if get_resp.status_code == 200: + return self._template_to_dict(get_resp.json()) + response.raise_for_status() + if response.status_code == 501: + raise NotImplementedError("template_create not supported on this class") + response.raise_for_status() + return self._template_to_dict(response.json()) + + async def get_template_status(self, template_id: str) -> dict | None: + response = await self._request("GET", f"/v1/templates/{template_id}") + if response.status_code == 404: + return None + response.raise_for_status() + return self._template_to_dict(response.json()) + + async def delete_template(self, template_id: str) -> bool: + response = await self._request("DELETE", f"/v1/templates/{template_id}") + if response.status_code == 404: + return True + if response.status_code == 501: + raise NotImplementedError("template_create not supported on this class") + response.raise_for_status() + return True + + # --- Template mapping helpers --- + + def _template_spec_to_new(self, spec: Any) -> dict: + """Convert a Rock TemplateSpec-like object to SandboxNext NewTemplate.""" + # Accept both dict and dataclass/pydantic model + if hasattr(spec, "model_dump"): + spec = spec.model_dump() + elif hasattr(spec, "__dict__"): + spec = {k: v for k, v in vars(spec).items() if not k.startswith("_")} + elif not isinstance(spec, dict): + spec = dict(spec) + + body: dict[str, Any] = { + "request_id": spec.get("template_id") or spec.get("request_id", ""), + "region": spec.get("region", self._config.region), + "class": spec.get("sandbox_class") or spec.get("class") or self._config.sandbox_class, + "name": spec.get("name", "default"), + } + resources = spec.get("resources") + if resources: + body["resources"] = resources + else: + cpus = spec.get("cpus") + memory = spec.get("memory") + disk = spec.get("disk") + res: dict[str, int] = {} + if cpus is not None: + res["vcpu"] = int(cpus) + if memory is not None: + res["memory_mb"] = _parse_mem_to_mb(memory) + if disk is not None: + res["disk_mb"] = _parse_disk_to_mb(disk) + if res: + body["resources"] = res + if spec.get("image"): + body["from_image"] = spec["image"] + if spec.get("env_vars"): + body["env_vars"] = spec["env_vars"] + return body + + def _template_to_dict(self, data: dict) -> dict: + """Convert SandboxNext Template response to Rock template status dict.""" + return { + "template_id": data.get("template_id", ""), + "name": data.get("name", ""), + "status": data.get("status", "pending"), + "resources": data.get("resources", {}), + "failure": data.get("failure"), + } diff --git a/tests/unit/sandbox/operator/remote/__init__.py b/tests/unit/sandbox/operator/remote/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/unit/sandbox/operator/remote/test_operator.py b/tests/unit/sandbox/operator/remote/test_operator.py new file mode 100644 index 0000000000..59a7c72d0b --- /dev/null +++ b/tests/unit/sandbox/operator/remote/test_operator.py @@ -0,0 +1,216 @@ +"""Unit tests for RemoteOperator — provider delegation + Redis merge.""" + +import pytest +from unittest.mock import AsyncMock + +from rock.actions.sandbox.response import State +from rock.actions.sandbox.sandbox_info import SandboxInfo +from rock.common.constants import StopReason +from rock.config import RemoteOperatorConfig +from rock.deployments.config import DockerDeploymentConfig +from rock.sandbox.operator.remote.constants import EXT_REMOTE_ID, EXT_BACKEND, BACKEND_NAME +from rock.sandbox.operator.remote.operator import RemoteOperator + + +def _make_config(**overrides) -> RemoteOperatorConfig: + defaults = {"endpoint": "api.sandbox.test", "api_key": "test-key"} + defaults.update(overrides) + return RemoteOperatorConfig(**defaults) + + +def _make_docker_config(**overrides) -> DockerDeploymentConfig: + defaults = { + "image": "python:3.11", + "cpus": 2.0, + "memory": "8g", + "disk": "50G", + "container_name": "sb-test-001", + } + defaults.update(overrides) + return DockerDeploymentConfig(**defaults) + + +class TestRemoteOperatorInit: + def test_default_provider_is_sandbox_next(self): + op = RemoteOperator(_make_config()) + from rock.sandbox.operator.remote.providers.sandbox_next_provider import SandboxNextProvider + + assert isinstance(op._provider, SandboxNextProvider) + + def test_unsupported_provider_raises(self): + with pytest.raises(ValueError, match="Unsupported remote provider"): + RemoteOperator(_make_config(provider="e2b")) + + def test_missing_endpoint_raises(self): + with pytest.raises(ValueError, match="endpoint is required"): + RemoteOperatorConfig(endpoint="") + + +class TestRemoteOperatorSubmit: + @pytest.mark.asyncio + async def test_submit_delegates_to_provider(self): + op = RemoteOperator(_make_config()) + op._provider = AsyncMock() + expected_info: SandboxInfo = {"sandbox_id": "sb-1", "state": State.PENDING} + op._provider.submit = AsyncMock(return_value=expected_info) + + docker_config = _make_docker_config() + result = await op.submit(docker_config, {"user_id": "u1"}) + assert result == expected_info + op._provider.submit.assert_awaited_once() + + +class TestRemoteOperatorGetStatus: + @pytest.mark.asyncio + async def test_no_redis_info_returns_none(self): + op = RemoteOperator(_make_config()) + op._redis_provider = None + # get_sandbox_info_from_redis will raise RuntimeError without provider + # but the method checks redis_info first; mock it + op.get_sandbox_info_from_redis = AsyncMock(return_value=None) + result = await op.get_status("sb-1") + assert result is None + + @pytest.mark.asyncio + async def test_no_remote_id_returns_none(self): + op = RemoteOperator(_make_config()) + op.get_sandbox_info_from_redis = AsyncMock(return_value={"sandbox_id": "sb-1"}) + result = await op.get_status("sb-1") + assert result is None + + @pytest.mark.asyncio + async def test_merge_redis_and_provider_info(self): + op = RemoteOperator(_make_config()) + op._provider = AsyncMock() + op._provider.get_status = AsyncMock(return_value={ + "sandbox_id": "sb-1", + "state": State.RUNNING, + "host_ip": "host.example.com", + "port_mapping": {22555: 443}, + "auth_token": "tok", + "extended_params": {EXT_REMOTE_ID: "sn-1", EXT_BACKEND: BACKEND_NAME}, + }) + op.get_sandbox_info_from_redis = AsyncMock(return_value={ + "sandbox_id": "sb-1", + "state": State.PENDING, + "user_id": "u1", + "image": "python:3.11", + "extended_params": {EXT_REMOTE_ID: "sn-1", "custom_key": "val"}, + }) + result = await op.get_status("sb-1") + assert result is not None + assert result["state"] == State.RUNNING + assert result["user_id"] == "u1" + assert result["host_ip"] == "host.example.com" + ext = result["extended_params"] + assert ext[EXT_REMOTE_ID] == "sn-1" + assert ext["custom_key"] == "val" # preserved from Redis + assert ext[EXT_BACKEND] == BACKEND_NAME # from provider + + @pytest.mark.asyncio + async def test_provider_404_marks_deleted(self): + op = RemoteOperator(_make_config()) + op._provider = AsyncMock() + op._provider.get_status = AsyncMock(return_value=None) + op.get_sandbox_info_from_redis = AsyncMock(return_value={ + "sandbox_id": "sb-1", + "state": State.RUNNING, + "extended_params": {EXT_REMOTE_ID: "sn-1"}, + }) + result = await op.get_status("sb-1") + assert result is not None + assert result["state"] == "deleted" + + +class TestRemoteOperatorStop: + @pytest.mark.asyncio + async def test_stop_delegates_to_provider(self): + op = RemoteOperator(_make_config()) + op._provider = AsyncMock() + op._provider.stop = AsyncMock(return_value=True) + op.get_sandbox_info_from_redis = AsyncMock(return_value={ + "sandbox_id": "sb-1", + "extended_params": {EXT_REMOTE_ID: "sn-1"}, + }) + result = await op.stop("sb-1", StopReason.MANUAL) + assert result is True + op._provider.stop.assert_awaited_once_with("sn-1") + + @pytest.mark.asyncio + async def test_stop_no_remote_id_raises(self): + op = RemoteOperator(_make_config()) + op.get_sandbox_info_from_redis = AsyncMock(return_value=None) + with pytest.raises(Exception, match="cannot resolve"): + await op.stop("sb-1") + + +class TestRemoteOperatorDelete: + @pytest.mark.asyncio + async def test_delete_from_config_extended_params(self): + op = RemoteOperator(_make_config()) + op._provider = AsyncMock() + op._provider.delete = AsyncMock(return_value=True) + docker_config = _make_docker_config() + docker_config.extended_params = {EXT_REMOTE_ID: "sn-1"} + result = await op.delete(docker_config) + assert result is True + op._provider.delete.assert_awaited_once_with("sn-1") + + @pytest.mark.asyncio + async def test_delete_fallback_to_redis(self): + op = RemoteOperator(_make_config()) + op._provider = AsyncMock() + op._provider.delete = AsyncMock(return_value=True) + op.get_sandbox_info_from_redis = AsyncMock(return_value={ + "sandbox_id": "sb-1", + "extended_params": {EXT_REMOTE_ID: "sn-redis"}, + }) + docker_config = _make_docker_config() + result = await op.delete(docker_config) + assert result is True + op._provider.delete.assert_awaited_once_with("sn-redis") + + +class TestRemoteOperatorRestart: + @pytest.mark.asyncio + async def test_restart_not_supported(self): + from rock.sdk.common.exceptions import BadRequestRockError + + op = RemoteOperator(_make_config()) + with pytest.raises(BadRequestRockError, match="restart"): + await op.restart(_make_docker_config()) + + +class TestRemoteOperatorTemplateAPI: + @pytest.mark.asyncio + async def test_create_template_delegates(self): + op = RemoteOperator(_make_config()) + op._provider = AsyncMock() + op._provider.create_template = AsyncMock(return_value={"template_id": "tpl-1", "status": "pending"}) + result = await op.create_template({"name": "test"}) + assert result["template_id"] == "tpl-1" + + @pytest.mark.asyncio + async def test_create_template_not_implemented_fallback(self): + from rock.sdk.common.exceptions import BadRequestRockError + + op = RemoteOperator(_make_config()) + op._provider = AsyncMock() + op._provider.create_template = AsyncMock(side_effect=NotImplementedError()) + with pytest.raises(BadRequestRockError, match="template not supported"): + await op.create_template({"name": "test"}) + + @pytest.mark.asyncio + async def test_get_template_status_404(self): + op = RemoteOperator(_make_config()) + op._provider = AsyncMock() + op._provider.get_template_status = AsyncMock(return_value=None) + result = await op.get_template_status("tpl-gone") + assert result is None + + @pytest.mark.asyncio + async def test_delete_template_success(self): + op = RemoteOperator(_make_config()) + op._provider = AsyncMock() + op._provider.delete_template = AsyncMock(return_value=True) + assert await op.delete_template("tpl-1") is True diff --git a/tests/unit/sandbox/operator/remote/test_sandbox_next_provider.py b/tests/unit/sandbox/operator/remote/test_sandbox_next_provider.py new file mode 100644 index 0000000000..597470a2c4 --- /dev/null +++ b/tests/unit/sandbox/operator/remote/test_sandbox_next_provider.py @@ -0,0 +1,284 @@ +"""Unit tests for SandboxNextProvider — mock httpx transport.""" + +import pytest +import httpx + +from rock.actions.sandbox.response import State +from rock.config import RemoteOperatorConfig +from rock.deployments.config import DockerDeploymentConfig +from rock.sandbox.operator.remote.constants import EXT_REMOTE_ID, EXT_ENDPOINT, EXT_BACKEND, BACKEND_NAME +from rock.sandbox.operator.remote.providers.sandbox_next_provider import ( + SandboxNextProvider, + _map_state, + _parse_mem_to_mb, + _parse_disk_to_mb, +) + + +# --- Config / fixture helpers --- + +def _make_config(**overrides) -> RemoteOperatorConfig: + defaults = {"endpoint": "api.sandbox.test", "api_key": "test-key"} + defaults.update(overrides) + return RemoteOperatorConfig(**defaults) + + +def _make_docker_config(**overrides) -> DockerDeploymentConfig: + defaults = { + "image": "python:3.11", + "cpus": 2.0, + "memory": "8g", + "disk": "50G", + "container_name": "sb-test-001", + } + defaults.update(overrides) + return DockerDeploymentConfig(**defaults) + + +def _make_client(handler) -> httpx.AsyncClient: + """Create an httpx.AsyncClient with a mock transport.""" + return httpx.AsyncClient( + base_url="https://api.sandbox.test", + transport=httpx.MockTransport(handler), + ) + + +# --- Utility tests --- + +class TestParseMemToMb: + def test_gigabytes(self): + assert _parse_mem_to_mb("8g") == 8192 + + def test_megabytes(self): + assert _parse_mem_to_mb("4096m") == 4096 + + def test_plain_number(self): + assert _parse_mem_to_mb("2048") == 2048 + + def test_empty(self): + assert _parse_mem_to_mb("") == 0 + + def test_uppercase(self): + assert _parse_mem_to_mb("4G") == 4096 + + +class TestParseDiskToMb: + def test_gigabytes(self): + assert _parse_disk_to_mb("50G") == 51200 + + def test_none(self): + assert _parse_disk_to_mb(None) == 0 + + +class TestMapState: + def test_creating(self): + assert _map_state("creating") == State.PENDING + + def test_running(self): + assert _map_state("running") == State.RUNNING + + def test_paused(self): + assert _map_state("paused") == State.STOPPED + + def test_failed(self): + assert _map_state("failed") == State.STOPPED + + def test_unknown(self): + assert _map_state("nonsense") == State.PENDING + + def test_none(self): + assert _map_state(None) == State.PENDING + + +# --- Provider lifecycle tests --- + +class TestSandboxNextProviderSubmit: + @pytest.mark.asyncio + async def test_submit_success(self): + def handler(request: httpx.Request) -> httpx.Response: + assert request.method == "POST" + assert "/v1/sandboxes" in str(request.url) + body = httpx.Response( + 201, + json={ + "sandbox_id": "sn-abc123", + "region": "cn-hangzhou", + "class": "headless-vm", + "state": "creating", + "access": { + "endpoint_template": "https://{port}-s485cf4a26daa06baf87e7636e63ca8873f.cn-hangzhou.sandbox.example.com", + "agent_token": "agent-token-xyz", + }, + }, + ) + return body + + config = _make_config() + client = _make_client(handler) + provider = SandboxNextProvider(config, client=client) + docker_config = _make_docker_config() + info = await provider.submit(docker_config, {"user_id": "u1", "experiment_id": "e1", "namespace": "ns"}) + + assert info["sandbox_id"] == "sb-test-001" + assert info["state"] == State.PENDING + assert info["auth_token"] == "agent-token-xyz" + assert info["host_ip"] == "https://{port}-s485cf4a26daa06baf87e7636e63ca8873f.cn-hangzhou.sandbox.example.com" + assert info["port_mapping"] == {22555: 8000, 8080: 8080, 22: 22} + ext = info["extended_params"] + assert ext[EXT_REMOTE_ID] == "sn-abc123" + assert ext[EXT_BACKEND] == BACKEND_NAME + assert EXT_ENDPOINT in ext + + @pytest.mark.asyncio + async def test_submit_with_env_vars(self): + def handler(request: httpx.Request) -> httpx.Response: + import json + + payload = json.loads(request.content) + assert payload["env_vars"] == {"FOO": "bar"} + return httpx.Response(202, json={ + "sandbox_id": "sn-2", + "state": "creating", + "access": {"endpoint_template": "http://{port}-x.test", "agent_token": ""}, + }) + + config = _make_config() + client = _make_client(handler) + provider = SandboxNextProvider(config, client=client) + docker_config = _make_docker_config(env_vars={"FOO": "bar"}) + info = await provider.submit(docker_config, {}) + assert info["extended_params"][EXT_REMOTE_ID] == "sn-2" + + +class TestSandboxNextProviderGetStatus: + @pytest.mark.asyncio + async def test_running(self): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, json={ + "sandbox_id": "sn-1", + "state": "running", + "access": {"endpoint_template": "https://{port}-x.test", "agent_token": "tok"}, + }) + + provider = SandboxNextProvider(_make_config(), client=_make_client(handler)) + info = await provider.get_status("sn-1") + assert info is not None + assert info["state"] == State.RUNNING + assert info["auth_token"] == "tok" + + @pytest.mark.asyncio + async def test_404_returns_none(self): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(404, json={"error": "not found"}) + + provider = SandboxNextProvider(_make_config(), client=_make_client(handler)) + info = await provider.get_status("sn-gone") + assert info is None + + +class TestSandboxNextProviderStop: + @pytest.mark.asyncio + async def test_pause_success(self): + def handler(request: httpx.Request) -> httpx.Response: + assert "/pause" in str(request.url) + return httpx.Response(202, json={"state": "pausing"}) + + provider = SandboxNextProvider(_make_config(), client=_make_client(handler)) + result = await provider.stop("sn-1") + assert result is True + + @pytest.mark.asyncio + async def test_pause_501_fallback_to_delete(self): + call_count = {"delete": 0} + + def handler(request: httpx.Request) -> httpx.Response: + if "/pause" in str(request.url): + return httpx.Response(501, json={"error": "not implemented"}) + if request.method == "DELETE": + call_count["delete"] += 1 + return httpx.Response(202) + return httpx.Response(500) + + provider = SandboxNextProvider(_make_config(), client=_make_client(handler)) + result = await provider.stop("sn-1") + assert result is True + assert call_count["delete"] == 1 + + +class TestSandboxNextProviderDelete: + @pytest.mark.asyncio + async def test_delete_success(self): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(202) + + provider = SandboxNextProvider(_make_config(), client=_make_client(handler)) + assert await provider.delete("sn-1") is True + + @pytest.mark.asyncio + async def test_delete_404(self): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(404) + + provider = SandboxNextProvider(_make_config(), client=_make_client(handler)) + assert await provider.delete("sn-gone") is True + + +# --- Template API tests --- + +class TestSandboxNextProviderTemplate: + @pytest.mark.asyncio + async def test_create_template_success(self): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(202, json={ + "template_id": "tpl-1", + "name": "py311", + "status": "pending", + "resources": {"vcpu": 2, "memory_mb": 8192}, + }) + + provider = SandboxNextProvider(_make_config(), client=_make_client(handler)) + result = await provider.create_template({"template_id": "tpl-1", "name": "py311", "cpus": 2, "memory": "8g"}) + assert result["template_id"] == "tpl-1" + assert result["status"] == "pending" + + @pytest.mark.asyncio + async def test_create_template_409_idempotent(self): + def handler(request: httpx.Request) -> httpx.Response: + if request.method == "POST": + return httpx.Response(409, json={"error": "conflict"}) + # GET fallback + return httpx.Response(200, json={ + "template_id": "tpl-1", + "name": "py311", + "status": "ready", + }) + + provider = SandboxNextProvider(_make_config(), client=_make_client(handler)) + result = await provider.create_template({"template_id": "tpl-1", "name": "py311"}) + assert result["template_id"] == "tpl-1" + assert result["status"] == "ready" + + @pytest.mark.asyncio + async def test_get_template_status_404(self): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(404) + + provider = SandboxNextProvider(_make_config(), client=_make_client(handler)) + assert await provider.get_template_status("tpl-gone") is None + + @pytest.mark.asyncio + async def test_delete_template_404(self): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(404) + + provider = SandboxNextProvider(_make_config(), client=_make_client(handler)) + assert await provider.delete_template("tpl-gone") is True + + @pytest.mark.asyncio + async def test_create_template_501_not_implemented(self): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(501, json={"error": "not implemented"}) + + provider = SandboxNextProvider(_make_config(), client=_make_client(handler)) + with pytest.raises(NotImplementedError): + await provider.create_template({"template_id": "tpl-1", "name": "test"})