diff --git a/cloud/chv/config.py b/cloud/chv/config.py index a9d2554..4d9c7bd 100644 --- a/cloud/chv/config.py +++ b/cloud/chv/config.py @@ -259,6 +259,60 @@ def _gitea_fragment(env: dict) -> dict: return {"gitea_sync_env": config_gitea_sync_env(env)} +def _to_switch(raw: object, name: str) -> bool | None: + """CHV_* 开关解析（P1 气球/vsock 用）：未设 → None（调用方 + 再定默认值）；"1" 开、"0" 关；其它值一律 ValueError，绝不 + 静默当关（与 CHV_SSH_FORWARD 的 `== "1"` 旧口径不同，本轨 + 新开关按差距表要求非法值即报错）。""" + if raw is None: + return None + if raw == "1": + return True + if raw == "0": + return False + raise ValueError("%s 只能是 0 或 1：%r" % (name, raw)) + + +def _balloon_vsock_fragment(env: dict) -> dict: + """气球/vsock 片段（P1，feat-chv-muse-p1）：virtio-balloon + + free-page-reporting 开关、vsock notify 通道开关与客 CID。 + + - CHV_BALLOON：默认 1（开）。只认 "1"/"0"，其它值 + ValueError。 + - CHV_VSOCK：默认 1（开）。只认 "1"/"0"，其它值 + ValueError。 + - CHV_VSOCK_CID：客机 vsock CID，可选，默认 3。必须是 + 整数且在 3-4294967295 之间：0/1/2 是 vsock 保留 CID + （Hypervisor/Loopback/Host，CHV 的 VmConfig 校验会 + 以 VsockSpecialCid 拒绝），-1（随机 CID）CHV 的 + vm.create 同样 400 拒绝（上游 discussion #7531）， + 所以这里不接受保留值/负值；显式设了非法 CID 即 + ValueError，哪怕 CHV_VSOCK=0（配错早炸，不留到起 VM + 才暴露）。CID 由运维指定、不是 CHV 分配的：同一宿主 + 上各 VM 的 vsock socket 文件不同，CID 相同（默认 3） + 并不冲突。""" + balloon = _to_switch(env.get("CHV_BALLOON"), "CHV_BALLOON") + if balloon is None: + balloon = True + vsock = _to_switch(env.get("CHV_VSOCK"), "CHV_VSOCK") + if vsock is None: + vsock = True + raw_cid = env.get("CHV_VSOCK_CID") + if raw_cid is None: + vsock_cid = 3 + else: + try: + vsock_cid = _to_int(raw_cid) + except ValueError: + raise ValueError("CHV_VSOCK_CID 必须是整数") + if not 3 <= vsock_cid <= 0xFFFFFFFF: + raise ValueError( + "CHV_VSOCK_CID 必须在 3-4294967295 之间：%s" + % vsock_cid) + return {"balloon": balloon, "vsock": vsock, + "vsock_cid": vsock_cid} + + def _seed_fragment(params: dict, env: dict) -> dict: """seed 附加项片段：只存宿主文件路径，不读文件内容 （读文件在 wake 时 fail-closed，见执行器 _seed_extras； @@ -295,17 +349,19 @@ def build_chv_config(params: dict, env: dict) -> "ChvConfig": """把显式构造参数与环境变量解析成 ChvConfig。 片段调用顺序即旧校验顺序（SSH → 配额 → 磁盘 → - 超时），只有一处可观测的顺序变化：warm_pool_size - 从旧超时片段移到配额片段，因此“CHV_WARM_POOL_SIZE - 非法 + 磁盘非法同时出现”时先报前者；单项非法时 - 报错与旧代码逐字一致。字段集合与取值与旧代码一致， - 由测试的字段集断言钉住。""" + 气球/vsock → 超时），只有两处可观测的顺序变化： + warm_pool_size 从旧超时片段移到配额片段，因此 + “CHV_WARM_POOL_SIZE 非法 + 磁盘非法同时出现”时先报 + 前者；单项非法时报错与旧代码逐字一致。字段集合 = + 旧代码字段 + P1 气球/vsock 三字段（balloon/vsock/ + vsock_cid），由测试的字段集断言钉住。""" env = env if env is not None else {} return _assemble_chv_config([ _base_fragment(params, env), _ssh_fragment(params, env), _quota_fragment(env), _disk_fragment(env), + _balloon_vsock_fragment(env), _network_fragment(params, env), _timeout_fragment(params, env), _gitea_fragment(env), diff --git a/cloud/chv/executor.py b/cloud/chv/executor.py index 07d37f1..5161c78 100644 --- a/cloud/chv/executor.py +++ b/cloud/chv/executor.py @@ -97,6 +97,10 @@ class CloudHypervisorExecutor(Executor): SSH 直接复用 OpenSSH 的密钥认证与端口转发，云镜像自带 sshd，无需在客内再写一个 vsock agent、也无需在宿主再写 一个 vsock 代理去翻译 host.py 协议。 + - P1（feat-chv-muse-p1）补充：vsock 只做最小 notify + （客机发 ready/心跳，见 cloud/chv/vsock.py 与 + cloud/chv/guest/vsock_notify.py），与 SSH 并存；SSH + 仍是唯一命令通道，host key 钉死不动。 - 放弃 vsock 的代价：需要运维提供从宿主到客机的 SSH 通路 （tap/桥接 + 端口转发，见下文“VM 生命周期”）。 - Host key 钉死（本轨 #216 后的加固，默认行为变化）： @@ -131,8 +135,8 @@ class CloudHypervisorExecutor(Executor): “true”命令，直到 CHV_READY_TIMEOUT（默认 120 秒， 轮询间隔 CHV_READY_INTERVAL，默认 5 秒）内成功才跑 host.py/任务；超时则关 VM 并报失败，绝不在未就绪时 - 就派任务。vsock 放弃理由同客内通道一节（需客内 agent - + 宿主代理两套新代码，v1 不拼半吊子）。 + 就派任务。P1 的 vsock ready 只做就绪辅助信号 + （cloud/chv/vsock.py），绝不取代 SSH 就绪判定。 - 暂停与快照唤醒是两条独立代码分支： pause_vm()/resume_vm() 走 ch-remote pause/resume； restore_vm() 先 ch-remote restore（source_url=...）， @@ -230,6 +234,12 @@ class CloudHypervisorExecutor(Executor): - CHV_API_TIMEOUT：等 vmm.ping 通的上限秒数（默认 30） - CHV_READY_TIMEOUT：等 guest SSH 就绪的上限秒数（默认 120） - CHV_READY_INTERVAL：SSH 就绪轮询间隔秒数（默认 5） + - CHV_BALLOON：virtio-balloon + free-page-reporting 开关 + （默认 1 开；只认 0/1，非法值 ValueError） + - CHV_VSOCK：vsock notify 通道开关（默认 1 开；只认 + 0/1，非法值 ValueError） + - CHV_VSOCK_CID：客机 vsock CID（默认 3；须在 + 3-4294967295 内，0/1/2 为保留 CID） 前置校验（preflight，执行入口必调，缺一即 fail-fast）： 收集全部缺失项一次报出具体缺哪一项，绝不笼统报错： @@ -282,6 +292,9 @@ class CloudHypervisorExecutor(Executor): task_timeout: float | None snapshot_url: str | None warm_pool_size: int + balloon: bool + vsock: bool + vsock_cid: int api_timeout: float ready_timeout: float ready_interval: float @@ -431,7 +444,8 @@ class CloudHypervisorExecutor(Executor): "snapshot_max_total_gb", "snapshot_retention_days", "snapshot_dir", "allow_remote_snapshot", "gitea_sync_env", "seed_user_data_extra_file", - "seed_network_config_file"): + "seed_network_config_file", "balloon", "vsock", + "vsock_cid"): setattr(self, _field, getattr(_cfg, _field)) # -- 名字 / 路径 --------------------------------------------------- @@ -1024,7 +1038,8 @@ class CloudHypervisorExecutor(Executor): "image_type": "Raw"}) return disks - def build_vm_config(self, task_disk: str | None = None, seed_disk: str | None = None) -> dict: + def build_vm_config(self, task_disk: str | None = None, seed_disk: str | None = None, + vsock_socket: str | None = None) -> dict: """API 模式的 VmConfig（vm.create 请求体，纯函数）： cpus{boot_vcpus,max_vcpus}、memory{size: 字节}、 payload{kernel, cmdline}、disks[{path: 每任务磁盘}]、 @@ -1037,7 +1052,11 @@ class CloudHypervisorExecutor(Executor): 旧版顶层 kernel{path}/cmdline{args} 会被服务端忽略， vm.create 照常 204、vm.boot 报 400 "No bootitem provided: neither firmware nor kernel"（2026-11-07 - WSL 真机 CHV 54.0.0 实测）。""" + WSL 真机 CHV 54.0.0 实测）。 + P1（feat-chv-muse-p1）：默认带 balloon/vsock 设备 + （见 _balloon_vsock_payload；CHV_BALLOON/CHV_VSOCK + 可关）。vsock_socket 显式给时用它，否则按任务盘 + 位置派生每 VM 唯一路径（_boot_vm 未传时即走派生）。""" try: cpus = int(self.cpus) except ValueError: @@ -1054,6 +1073,8 @@ class CloudHypervisorExecutor(Executor): "disks": self._vm_config_disks(task_disk, seed_disk), "rng": {"src": "/dev/urandom"}, } + config.update(self._balloon_vsock_payload( + vsock_socket=vsock_socket, task_disk=task_disk)) if self.net_json: try: net = json.loads(self.net_json) @@ -1062,6 +1083,67 @@ class CloudHypervisorExecutor(Executor): config["net"] = net if isinstance(net, list) else [net] return config + def _balloon_vsock_payload(self, vsock_socket=None, + task_disk=None) -> dict: + """P1（feat-chv-muse-p1）：VmConfig 的 balloon/vsock + 设备（CHV 54 API，VmConfig 顶层键；字段名以 CHV + OpenAPI 生成的客户端为准：BalloonConfig{size, + deflate_on_oom, free_page_reporting}、VsockConfig{ + cid, socket}；本机无 /dev/kvm，尚未真机验证，见下）。 + + balloon（默认开）：{"size": 0, "deflate_on_oom": true, + "free_page_reporting": true}。size 是目标气球尺寸 + （字节）：0 = 启动时不预扣客机内存，只靠 + free_page_reporting 让客机持续上报空闲页、宿主 + MADV_DONTNEED 回收——被动回收，不主动 inflate。 + deflate_on_oom 是安全网：日后若经 vm.resize 充气 + 后客机承压，气球自动放气而不 OOM。待真机验证： + size=0 是否被 CHV 54 接受（若拒绝，改 size 为 VM + 内存字节数并重新验证，见汇报）；气球实际省内存量。 + vsock（默认开）：{"cid": , + "socket": <宿主侧 Unix socket 路径>}。cid 是客机 + CID（0/1/2 保留，宿主永远是 CID 2，客机连宿主用 + CID 2；CHV 不分配 CID，由配置指定）。socket 路径 + 派生：显式给 vsock_socket 时用它；否则任务盘在 + state_dir 内时落任务盘同目录（state_dir// + vsock.sock，每 VM 唯一）；reuse/lab（任务盘即基础 + 镜像或未给）时落 state_dir/vsock.sock——单 VM 可 + 用，多 VM 并发会在 vm.create 撞路径（大声失败， + 不静默），_boot_vm 后续接线应显式传每 VM 路径。 + 宿主侧监听在 _ 上（port 见 + cloud/chv/vsock.py 的 VSOCK_NOTIFY_PORT），客机 + 连 CID 2 发 ready/心跳（cloud/chv/guest/ + vsock_notify.py）。开关关时对应键整个不出现 + （不是发 false），旧行为（无设备）原样保留。""" + extra: dict = {} + if getattr(self, "balloon", True): + extra["balloon"] = { + "size": 0, "deflate_on_oom": True, + "free_page_reporting": True, + } + if getattr(self, "vsock", True): + extra["vsock"] = { + "cid": getattr(self, "vsock_cid", 3), + "socket": self._resolve_vsock_socket( + vsock_socket, task_disk), + } + return extra + + def _resolve_vsock_socket(self, vsock_socket, task_disk): + """vsock socket 路径派生（_balloon_vsock_payload 的 + 子步，单拆只为 50 行规矩，行为不动）：显式给定时 + 用它；否则任务盘在 state_dir 内时落任务盘同目录 + （每 VM 唯一），reuse/lab 时落 state_dir/vsock.sock + （多 VM 并发会撞路径，大声失败）。""" + if vsock_socket is not None: + return vsock_socket + if task_disk and os.path.realpath( + task_disk).startswith( + os.path.realpath(self.state_dir) + os.sep): + return os.path.join( + os.path.dirname(task_disk), "vsock.sock") + return os.path.join(self.state_dir, "vsock.sock") + def _api_call(self, socket_path, method, path, payload=None): """API 调用归一化（实现在 cloud/chv/api_client.py）。""" return normalize_api_result( @@ -2245,27 +2327,118 @@ class CloudHypervisorExecutor(Executor): "interrupted_reason": reason, } - def _run_with_vm(self, task, inner_args, goal=False, extra_env=None, - **kwargs): - """每任务 VM 生命周期（冷启动正式路径，feat-chv-arch95 - 定案）：wake_vm（API 模式 + SSH 就绪）→ 跑客内命令 → - finally shutdown_vm。本方法绝不查 WarmPool：预热池是 - 实验性的（见 experimental_pool.py 头定案与 ADR-5），dispatch 永远 - 冷启动，warm_pool_wired_to_dispatch 恒 False。 - wake 失败时不跑 - 任务、直接抛分类失败。本方法不做内部重试：重试是 - worker 的 max_retries 语义，执行器内部重试会把一次 - 失败放大成多次计费与多次 VM 启动。 + def _resident_enabled(self): + """常驻默认开；CHV_RESIDENT_ENABLED=0/false/no/off 关。 + 优先认执行器上已有的 resident_enabled 属性（P1-B 若 + 经 config 落属性，直接复用），否则惰性读 env；__init__ + 不动，避免与 P1-B 的 __init__ 改动撞车。""" + flag = getattr(self, "resident_enabled", None) + if flag is None: + env = getattr(self, "_env", None) + if env is None: + env = os.environ + flag = env.get("CHV_RESIDENT_ENABLED", "1") + if isinstance(flag, str): + return flag.strip().lower() not in ( + "0", "false", "no", "off") + return bool(flag) - 中途死亡分类（本轨新增）：客内命令抛传输异常、或 - 非 goal 结果已 failed 时，若登记的 VMM 进程已死， - 判 interrupted（断点可续跑）而不是笼统 failed 或 - 原样抛异常；VM 还活着时的异常/失败原样处理，不 - 误判。goal 结果本身已有 interrupted 分类（见 - parse_goal_result），这里不再二次改写 goal 结论。 - feat-chv-secfix（H2）：extra_env 透传给 wake_vm（密钥 - 子集在 provisioned 模式下随 seed 投递），再透传给 - _run_guest（远程命令 source 客内密钥文件）。""" + def _resident_pool_instance(self): + """惰性建唯一常驻池并缓存（方法内 import：执行器顶层 + 绝不 import 池/orphans，见 ADR-12 的 AST 钉子）。 + 同一 state_dir 只活一个池实例：registry 只在构造时 + 读一次，重启后条目一律 orphan（见 resident_pool）。""" + pool = getattr(self, "_resident_pool", None) + if pool is not None: + return pool + with self._vms_lock: + pool = getattr(self, "_resident_pool", None) + if pool is None: + from cloud.chv.resident_pool import ResidentPool + pool = ResidentPool(self) + self._resident_pool = pool + return pool + + def _resident_acquire(self, task, extra_env=None): + """常驻获取（best-effort，一律不抛，失败回 None）： + 未启用/带密钥 env → None（走冷启动；密钥须随 seed + 投递，常驻 VM 的 seed 是建池时定的，带密钥任务只能 + 冷启动）。有 paused 条目 → resume + SSH 就绪；池空/ + 失败 → 惰性 provision 一次再取，仍无 → None。返回 + {"task": 常驻伪任务, "pool": 池, "user_id": 用户}。""" + try: + if not self._resident_enabled(): + return None + secrets, _plain = split_secret_env(extra_env) + if secrets: + return None + pool = self._resident_pool_instance() + user_id = task.get("user_id") \ + if isinstance(task, dict) else None + acquired = pool.acquire(user_id) + if acquired.get("mode") == "resident": + return {"task": acquired["task"], "pool": pool, + "user_id": user_id} + provisioned = pool.provision(user_id) + if provisioned.get("mode") != "resident": + return None + acquired = pool.acquire(user_id) + if acquired.get("mode") != "resident": + return None + return {"task": acquired["task"], "pool": pool, + "user_id": user_id} + except Exception: + return None + + def _run_on_resident(self, task, resident_task, inner_args, + goal=False, extra_env=None, **kwargs): + """常驻路径执行（_run_with_vm 的常驻分支本体）： + _run_guest/_run_guest_goal 一律走常驻伪任务传参（SSH + known_hosts、api-socket、密钥投递记录、VMM 登记都按 + 常驻 VM 名算），返回后把 task_id 改回真实任务 id； + VM 中途死亡判据同样按常驻伪任务查登记进程，死亡 + 结论里的任务 id 用真实任务。finally 的 pause 在调 + 用方（_run_with_vm）做，这里只跑命令。""" + try: + if goal: + result = self._run_guest_goal( + resident_task, inner_args, extra_env=extra_env, + **kwargs) + else: + result = self._run_guest( + resident_task, inner_args, extra_env=extra_env, + **kwargs) + except Exception as exc: + if self._vm_proc_dead(resident_task): + return self._vm_death_result( + task, goal, str(exc)[:500]) + raise + if not goal and result.get("status") == "failed" \ + and self._vm_proc_dead(resident_task): + result = dict(result) + result["status"] = "interrupted" + result["vm_dead"] = True + result["interrupted_reason"] = ( + "VM 中途死亡，任务未完成、断点可续跑：" + "客内命令已 failed 且登记的 VMM 进程已死") + if isinstance(result, dict): + result = dict(result) + result["task_id"] = task["id"] + return result + + def _run_cold(self, task, inner_args, goal=False, extra_env=None, + **kwargs): + """冷启动路径（原 _run_with_vm 本体，逐字搬运）： + wake_vm（API 模式 + SSH 就绪）→ 跑客内命令 → + finally shutdown_vm。wake 失败时不跑任务、直接抛 + 分类失败；本方法不做内部重试（重试是 worker 的 + max_retries 语义）。中途死亡分类：客内命令抛传输 + 异常、或非 goal 结果已 failed 时，若登记的 VMM 进程 + 已死，判 interrupted（断点可续跑）；VM 还活着时的 + 异常/失败原样处理。feat-chv-secfix（H2）：extra_env + 透传给 wake_vm（密钥子集在 provisioned 模式下随 seed + 投递），再透传给 _run_guest（远程命令 source 客内 + 密钥文件）。""" self.wake_vm(task, extra_env=extra_env) try: try: @@ -2294,6 +2467,35 @@ class CloudHypervisorExecutor(Executor): except Exception: pass + def _run_with_vm(self, task, inner_args, goal=False, extra_env=None, + **kwargs): + """任务 VM 生命周期（feat-chv-muse-p1，见 ADR-17）： + 常驻启用时先 _resident_acquire——paused 常驻 VM 经 + resume + SSH 就绪即跑，跑完 finally 里 release（pause， + 不 shutdown）；池空/失败/未启用/带密钥 env 时走冷启动 + （_run_cold：wake_vm → 跑客内命令 → finally shutdown）。 + 常驻失败一律回落冷启动，绝不因池坏而丢任务；WarmPool + 仍绝不查（实验模块，wired_to_dispatch 恒 False）。 + wake 失败时不跑任务、直接抛分类失败；本方法不做内部 + 重试（重试是 worker 的 max_retries 语义）。中途死亡 + 判 interrupted（断点可续跑），见 _run_cold 与 + _run_on_resident。dispatch/run/goal/goal_resume/断点 + 续跑都走本方法，签名不动。""" + resident = self._resident_acquire(task, extra_env=extra_env) + if resident is not None: + pool = resident["pool"] + try: + return self._run_on_resident( + task, resident["task"], inner_args, goal=goal, + extra_env=extra_env, **kwargs) + finally: + try: + pool.release(resident["user_id"]) + except Exception: + pass + return self._run_cold(task, inner_args, goal=goal, + extra_env=extra_env, **kwargs) + def execute(self, task: dict) -> dict: """只读执行：客内 host.py status，不直接跑模型回合 （与 Local/Docker 的 execute 同款语义）。 diff --git a/cloud/chv/guest/vsock_notify.py b/cloud/chv/guest/vsock_notify.py new file mode 100755 index 0000000..d7e3cf5 --- /dev/null +++ b/cloud/chv/guest/vsock_notify.py @@ -0,0 +1,136 @@ +#!/usr/bin/env python3 +"""客内 vsock notify 脚本（P1，feat-chv-muse-p1）。 + +启动后连宿主 CID 2 的 1024 端口发 ready，之后每 30 秒发一次 +heartbeat。消息是 JSON 行： + + {"type": "ready"|"heartbeat", "vm": "", "ts": } + +宿主在 _1024 这个 Unix socket 上监听接收 +（cloud/chv/vsock.py 的 wait_ready/listen_notify）。 + +失败一律静默退出（exit 0，不打日志、不留 traceback）：vsock +只是就绪辅助信号，SSH 仍是唯一命令通道与就绪判定——本脚本 +绝不能因为连不上、发失败而影响 sshd 或任务。调试时设 +VSOCK_NOTIFY_DEBUG=1 可在 stderr 看到异常。 + +================ P2 镜像集成接入说明（本轨只写脚本，不改构建） ================ +本文件不由 chv-guest/build.sh 安装，镜像集成是 P2 的范围。 +P2 接入时做这三件： + +1. 把本文件拷进镜像，例如 /app/vsock_notify.py（须 python3， + 客机镜像已有 Python；若最小镜像无 python3，P2 须先补）； +2. 加 systemd 单元，例如 /etc/systemd/system/vsock-notify.service： + + [Unit] + Description=vsock notify to host (ready/heartbeat) + After=multi-user.target + [Service] + Type=simple + ExecStart=/usr/bin/python3 /app/vsock_notify.py %H + Restart=no + [Install] + WantedBy=multi-user.target + + （%H 是客机 hostname；也可直接写 VM 名，或由 cloud-init + 写进环境变量 CHV_VM_NAME。Restart=no：本脚本失败即静默 + 退出，不由 systemd 反复拉起打扰宿主监听。） +3. 前置条件（缺一则本脚本静默退出，任务仍走 SSH，不报错）： + - 客机内核须含 CONFIG_VIRTIO_VSOCKETS； + - 起 VM 时 CHV_VSOCK=1（默认开），VmConfig 带 vsock 设备 + （cloud/chv/executor.py 的 _balloon_vsock_payload）； + - 宿主侧须在 VM boot 前后监听 _1024（否则客机 + 首次 connect 失败，本脚本在 30 秒内重试，仍失败即退出）。 + +端口 1024 与宿主侧 cloud/chv/vsock.py 的 VSOCK_NOTIFY_PORT +须一致；改一处必须改另一处。 +=============================================================================== +""" + +import json +import os +import socket +import sys +import time + +# 宿主 CID（vsock 众所周知值，CHV 官方 docs/vsock.md）。 +HOST_CID = 2 +# notify 端口（与宿主侧 VSOCK_NOTIFY_PORT 一致）。 +NOTIFY_PORT = 1024 +# 首次 connect 重试窗口（秒）：宿主监听可能比客机启动晚到。 +CONNECT_RETRY_SECONDS = 30 +# 心跳间隔（秒）。 +HEARTBEAT_INTERVAL = 30 + + +def _vm_name(argv) -> str: + """VM 名：argv[1] > CHV_VM_NAME > hostname。""" + if len(argv) > 1 and argv[1]: + return argv[1] + env_name = os.environ.get("CHV_VM_NAME") or os.environ.get( + "VSOCK_VM_NAME") + if env_name: + return env_name + try: + return socket.gethostname() + except OSError: + return "unknown" + + +def _send(sock, message_type: str, vm: str) -> None: + line = json.dumps({"type": message_type, "vm": vm, + "ts": time.time()}, + separators=(",", ":")) + sock.sendall((line + "\n").encode("utf-8")) + + +def _connect(deadline: float): + """在 deadline 前反复连 (HOST_CID, NOTIFY_PORT)，成功返回 + socket，超时返回 None。""" + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + return None + sock = socket.socket(socket.AF_VSOCK, socket.SOCK_STREAM) + try: + sock.settimeout(min(remaining, 5)) + sock.connect((HOST_CID, NOTIFY_PORT)) + sock.settimeout(None) + return sock + except OSError as exc: + try: + sock.close() + except OSError: + pass + if os.environ.get("VSOCK_NOTIFY_DEBUG") == "1": + print("vsock_notify connect 失败：%r" % (exc,), + file=sys.stderr, flush=True) + time.sleep(1) + + +def main(argv) -> int: + vm = _vm_name(argv) + deadline = time.monotonic() + CONNECT_RETRY_SECONDS + sock = _connect(deadline) + if sock is None: + return 0 + try: + _send(sock, "ready", vm) + while True: + time.sleep(HEARTBEAT_INTERVAL) + _send(sock, "heartbeat", vm) + except (OSError, ValueError): + # 任何发送失败都静默退出，不影响 sshd/任务。 + return 0 + except KeyboardInterrupt: + return 0 + finally: + try: + sock.close() + except OSError: + pass + return 0 + + +if __name__ == "__main__": + sys.exit(main(sys.argv)) diff --git a/cloud/chv/resident_pool.py b/cloud/chv/resident_pool.py new file mode 100644 index 0000000..1041318 --- /dev/null +++ b/cloud/chv/resident_pool.py @@ -0,0 +1,515 @@ +"""每用户常驻 VM 池（生产模块，P1 对齐 Muse，见 ADR-17）。 + +每用户一台常驻 VM：任务到时 resume 已暂停的 VM，跑完 +再 pause，不销毁；池空或任何一步失败都回落冷启动， +绝不因池坏而丢任务。常驻 VM 的盘是每用户持久工作区盘 +（state_dir/chv-resident-<用户>/disk.raw），不是每任务 +新盘——隔离单元在常驻路径上从“每任务”变成“每用户” +（单租户可接受，多租户重审前不得把常驻当租户隔离）。 + +状态机：cold → provisioning → paused → running → paused。 +acquire(user)：有 paused 条目时 resume_vm + 等 SSH 就绪 +（复用执行器 wait_guest_ready，不另写判定），返回 +mode=resident；无可用/失败时返回 mode=cold。release(user)： +pause_vm，不销毁；pause 失败时条目留在 running（不删 +不销毁，worker 关闭时 reconcile 清理）。provision(user)： +wake_vm + pause_vm 建 paused 条目（惰性建池，失败回落）。 +reclaim_idle()：paused 超 CHV_RESIDENT_IDLE_MINUTES（默认 +0=不自动关机）时 hibernate（pause + snapshot + shutdown） +并删条目，快照文件留作恢复物，restore 是后续工作。 + +Registry（state_dir/resident-pool.json）：文件 IO 与损坏 +语义复用 cloud.chv.orphans（原子写、损坏 quarantine、 +单条目非法只丢该条）；重启后读到的条目一律 orphan，不 +复用，只能 reconcile 清理。条目按用户键隔离：release +绝不把用过的 VM 放回给另一用户；同一用户同一时间只跑 +一个常驻任务，并发第二个任务回落冷启动。 + +诚实边界：resume 失败时 wait_guest_ready 的失败路径会 +stop_vm（含删工作区盘），条目随之丢弃——该路径的工作区 +以 Gitea 同步为准；带密钥 extra_env 的任务不走常驻 +（seed 是建池时定的，密钥只能冷启动随 seed 投递）。 +""" + +import logging +import os +import re +import threading +import time + +from cloud.chv.orphans import _read_registry, _write_registry + +_LOGGER = logging.getLogger("cloud.cloud_hypervisor") + +_REGISTRY_FILENAME = "resident-pool.json" +_RESIDENT_STATES = ("paused", "running", "provisioning") +_DEFAULT_USER = "default" + + +def resident_user_key(user_id: str | None) -> str: + """用户键：非空即原样消毒（非法字符变 -），空/缺省 + 即 "default"（当前单租户：任务 dict 无 user_id 时全 + 部落到同一常驻 VM；多租户重审前不得当租户边界用）。""" + raw = _DEFAULT_USER if user_id is None else str(user_id) + key = re.sub(r"[^A-Za-z0-9_.-]", "-", raw).strip(".-") + return key or _DEFAULT_USER + + +def resident_task_for_user(user_id: str | None) -> dict: + """常驻伪任务：执行器按它算 VM 名/known_hosts/socket。 + vm_name({"id": "resident-alice"}) = "chv-resident-alice"。""" + return {"id": "resident-" + resident_user_key(user_id)} + + +def resident_registry_path_for_executor(executor: object) -> str | None: + """常驻 registry 路径：state_dir/resident-pool.json。 + 无 state_dir 时返回 None（调用方按“无残留”处理）。""" + state_dir = getattr(executor, "state_dir", None) + if not state_dir: + return None + return os.path.join(state_dir, _REGISTRY_FILENAME) + + +def reconcile_resident_orphans_for_executor(executor: object) -> list[str]: + """生产路径的常驻残留清理入口：返回已清理的 VM 名。 + 只做清理：读 resident-pool.json → 每个上一世条目尽力 + shutdown_vm → 落盘为空 → best-effort 审计。绝不构造 + ResidentPool、绝不把任何条目发给任务。无 registry 路 + 径时返回 []。文件 IO 与损坏语义复用 orphans。""" + path = resident_registry_path_for_executor(executor) + if not path: + return [] + data, _corrupt = _read_registry(path) + if data is None: + return [] + cleaned: list[str] = [] + for entry in data.values(): + if not isinstance(entry, dict): + continue + if not entry.get("vm"): + continue + if entry.get("state") not in _RESIDENT_STATES: + continue + vm = str(entry["vm"]) + try: + executor.shutdown_vm(vm) # type: ignore[attr-defined] + except Exception: + pass + cleaned.append(vm) + if cleaned: + _write_registry(path, {}) + try: + executor._emit( # type: ignore[attr-defined] + "resident_pool_reconcile", {"id": None}, + "cleaned=%s" % ",".join(cleaned)) + except Exception: + pass + return cleaned + + +class ResidentPool: + """每用户常驻 VM 池（生产路径，见模块 docstring）。 + + 同一 state_dir 只活一个池实例：registry 只在构造时 + 读一次，重启后条目一律 orphan（见 _load_registry）。 + 公共方法全部加锁：并发 acquire/release/provision 只 + 放行一路，其余回 cold/False，不把同一 VM 发给两个任务。 + """ + + def __init__(self, executor: object, idle_minutes: int | None = None, + registry_path: str | None = None) -> None: + """建池并读 registry：读到的条目一律 orphan（见模块 + docstring）。idle_minutes 未给时读执行器属性或 + CHV_RESIDENT_IDLE_MINUTES，非法/负数即 0（不自动关机）。""" + self.executor = executor + self._lock = threading.Lock() + self._busy: set[str] = set() + self._entries: dict = {} + self.registry_path = registry_path + if self.registry_path is None: + self.registry_path = \ + resident_registry_path_for_executor(executor) + if idle_minutes is None: + idle_minutes = self._idle_minutes_from_env() + self.idle_minutes = idle_minutes if idle_minutes > 0 else 0 + self.registry_corrupt = False + self.registry_corrupt_entries = 0 + self._load_registry() + + # -- registry ------------------------------------------------------ + def _load_registry(self) -> None: + """载入 registry：读到的条目一律标 orphan（绝不直接 + 当 paused 发给任务，旧 VMM 进程已随旧 worker 消失）。 + 无文件时为空池；损坏时 quarantine 并从空开始；单条 + 目非法只丢该条并计数，不连坐其它条目。""" + if not self.registry_path: + return + data, corrupt = _read_registry(self.registry_path) + if corrupt: + self.registry_corrupt = True + return + if data is None: + return + for key, entry in data.items(): + if not isinstance(entry, dict) \ + or not entry.get("vm") \ + or entry.get("state") not in _RESIDENT_STATES: + self.registry_corrupt_entries += 1 + continue + self._entries[key] = { + "state": "orphan", + "vm": entry["vm"], + "user_id": entry.get("user_id") or key, + "paused_at": None, + "last_used": None, + } + + def _save_registry(self) -> None: + """落盘当前条目（只写 paused/running/provisioning； + orphan 是待清理的上一世条目，reconcile 后即删，不写 + 回）。调用方须已持锁；写失败只记日志，不炸任务流。""" + if not self.registry_path: + return + data = {} + for key, entry in self._entries.items(): + if entry["state"] not in _RESIDENT_STATES: + continue + data[key] = { + "state": entry["state"], + "vm": entry.get("vm"), + "user_id": entry.get("user_id"), + "paused_at": entry.get("paused_at"), + "last_used": entry.get("last_used"), + } + _write_registry(self.registry_path, data) + + def _emit(self, event: str, task: dict, detail: str) -> None: + """best-effort 审计：执行器无 _emit 或抛错都不炸 + 池操作（与 WarmPool 同款）。""" + emit = getattr(self.executor, "_emit", None) + if not callable(emit): + return + try: + emit(event, task, detail) + except Exception: + pass + + def _idle_minutes_from_env(self) -> int: + """闲置阈值：执行器属性优先（P1-B 若经 config 落 + 属性，直接复用），否则读 CHV_RESIDENT_IDLE_MINUTES； + 非数字/负数一律 0（不自动关机，不炸建池）。""" + raw = getattr(self.executor, "resident_idle_minutes", None) + if raw is None: + env = getattr(self.executor, "_env", None) + if env is None: + env = os.environ + raw = env.get("CHV_RESIDENT_IDLE_MINUTES", 0) + if raw is None: + return 0 + try: + minutes = int(raw) + except (TypeError, ValueError): + return 0 + return minutes if minutes > 0 else 0 + + def _vm_name(self, resident_task: dict) -> str: + """常驻 VM 名：优先执行器 vm_name（与冷启动同款派 + 生），鸭子型执行器无该方法时退回 chv- 前缀拼法。""" + vm_name = getattr(self.executor, "vm_name", None) + if callable(vm_name): + try: + return str(vm_name(resident_task)) + except Exception: + pass + return "chv-" + str(resident_task.get("id")) + + # -- 计数 ---------------------------------------------------------- + def paused_count(self) -> int: + """当前 paused（可分配）的常驻条目数。""" + with self._lock: + return sum(1 for e in self._entries.values() + if e.get("state") == "paused") + + def orphan_count(self) -> int: + """上一世残留条目数（待 reconcile，不可分配）。""" + with self._lock: + return sum(1 for e in self._entries.values() + if e.get("state") == "orphan") + + # -- acquire / release ---------------------------------------------- + def _claim_paused(self, user_key: str) -> dict | None: + """占一条 paused 条目：加忙标记并返回其快照；无条目/ + 非 paused/正忙时返回 None（调用方回落冷启动）。""" + with self._lock: + entry = self._entries.get(user_key) + if entry is None or entry.get("state") != "paused": + return None + if user_key in self._busy: + return None + self._busy.add(user_key) + return dict(entry) + + def _resume_and_wait(self, resident_task: dict) -> bool: + """resume + 等 SSH 就绪（复用执行器判定）。 + resume 缺失的鸭子型执行器跳过 resume（与 WarmPool + 同款宽容）；SSH 未就绪即 False（wait_guest_ready + 的失败路径已自行 stop_vm，调用方只须删条目）。""" + resume = getattr(self.executor, "resume_vm", None) + if callable(resume): + try: + resume(resident_task) + except Exception: + return False + wait = getattr(self.executor, "wait_guest_ready", None) + if not callable(wait): + return True + try: + return bool(wait(resident_task)) + except Exception: + return False + + def _settle_acquire(self, user_key: str, ok: bool) -> None: + """acquire 收尾：成功标 running，失败删条目；都清 + 忙标记并落盘。""" + with self._lock: + self._busy.discard(user_key) + entry = self._entries.get(user_key) + if entry is None: + return + if ok: + entry["state"] = "running" + entry["last_used"] = time.time() + else: + self._entries.pop(user_key, None) + self._save_registry() + + def acquire(self, user_id: str | None) -> dict: + """取常驻 VM：paused 条目 → resume + SSH 就绪 → + mode=resident；无条目/正忙/失败 → mode=cold（调用 + 方回落冷启动）。失败条目直接丢弃（VM 本身已被 + wait_guest_ready 的失败路径关掉，或 resume 已拒）。""" + user_key = resident_user_key(user_id) + entry = self._claim_paused(user_key) + if entry is None: + return {"mode": "cold", "vm": None, "task": None} + resident_task = resident_task_for_user(user_id) + ok = self._resume_and_wait(resident_task) + self._settle_acquire(user_key, ok) + if not ok: + return {"mode": "cold", "vm": None, "task": None} + self._emit("resident_pool_acquire", resident_task, + "vm=%s user=%s" % (entry.get("vm"), user_key)) + return {"mode": "resident", "vm": entry.get("vm"), + "task": resident_task} + + def _pause(self, resident_task: dict) -> bool: + """pause_vm 尽力而为：失败返回 False，不抛。""" + pause = getattr(self.executor, "pause_vm", None) + if not callable(pause): + return False + try: + pause(resident_task) + return True + except Exception: + return False + + def release(self, user_id: str | None) -> bool: + """还常驻 VM：running 条目 → pause_vm，不销毁。 + 无条目/非 running/正忙 → False；pause 失败时条目 + 留在 running（不删不销毁，worker 关闭时 reconcile + 清理），返回 False。只动本用户的条目。""" + user_key = resident_user_key(user_id) + resident_task = resident_task_for_user(user_id) + with self._lock: + entry = self._entries.get(user_key) + if entry is None or entry.get("state") != "running": + return False + if user_key in self._busy: + return False + self._busy.add(user_key) + paused = self._pause(resident_task) + with self._lock: + self._busy.discard(user_key) + entry = self._entries.get(user_key) + if entry is None or not paused: + return False + entry["state"] = "paused" + entry["paused_at"] = time.time() + self._save_registry() + self._emit("resident_pool_release", resident_task, + "vm=%s user=%s" % (entry.get("vm"), user_key)) + return True + + # -- provision ------------------------------------------------------- + def _wake_and_pause(self, resident_task: dict) -> bool: + """建池：wake_vm（冷启动全套，含 SSH 就绪）再 + pause_vm。任一步失败都尽力 shutdown，不留半醒的 + VM；调用方随后删条目回落冷启动。""" + try: + wake = getattr(self.executor, "wake_vm", None) + pause = getattr(self.executor, "pause_vm", None) + if not callable(wake) or not callable(pause): + return False + wake(resident_task) + pause(resident_task) + return True + except Exception: + shutdown = getattr(self.executor, "shutdown_vm", None) + if callable(shutdown): + try: + shutdown(resident_task) + except Exception: + pass + return False + + def _finish_provision(self, user_key: str, ok: bool) -> str | None: + """建池收尾：成功标 paused 并记时间，失败删条目； + 都清忙标记并落盘。返回 VM 名或 None。""" + with self._lock: + self._busy.discard(user_key) + entry = self._entries.get(user_key) + if entry is None: + return None + if not ok: + self._entries.pop(user_key, None) + self._save_registry() + return None + entry["state"] = "paused" + entry["paused_at"] = time.time() + self._save_registry() + return str(entry.get("vm")) + + def provision(self, user_id: str | None) -> dict: + """建常驻 VM：已有 paused/running 条目直接返回（不 + 重 wake）；否则 wake + pause，成功落 paused 条目， + 失败删条目回 cold。并发建池只放行一路，其余回 cold。""" + user_key = resident_user_key(user_id) + resident_task = resident_task_for_user(user_id) + with self._lock: + entry = self._entries.get(user_key) + if entry is not None and entry.get("state") in ( + "paused", "running"): + return {"mode": "resident", "vm": entry.get("vm"), + "task": resident_task} + if user_key in self._busy: + return {"mode": "cold", "vm": None, "task": None} + self._busy.add(user_key) + self._entries[user_key] = { + "state": "provisioning", + "vm": self._vm_name(resident_task), + "user_id": user_id if user_id is not None + else _DEFAULT_USER, + "paused_at": None, "last_used": None} + self._save_registry() + ok = self._wake_and_pause(resident_task) + vm = self._finish_provision(user_key, ok) + if vm is None: + return {"mode": "cold", "vm": None, "task": None} + self._emit("resident_pool_provision", resident_task, + "vm=%s user=%s" % (vm, user_key)) + return {"mode": "resident", "vm": vm, "task": resident_task} + + # -- 空闲回收 -------------------------------------------------------- + def _hibernate_one(self, user_key: str) -> str | None: + """单条目休眠：hibernate_vm（快照到 state_dir 内 + resident.snapshot，成功后删条目）。失败留条目下次 + 再试，返回 None；无 state_dir 或无 hibernate 入口 + 同样留条目。""" + with self._lock: + entry = self._entries.get(user_key) + if entry is None or entry.get("state") != "paused": + return None + if user_key in self._busy: + return None + self._busy.add(user_key) + resident_task = resident_task_for_user( + entry.get("user_id")) + vm = str(entry.get("vm")) + state_dir = getattr(self.executor, "state_dir", None) + hibernate = getattr(self.executor, "hibernate_vm", None) + if not state_dir or not callable(hibernate): + with self._lock: + self._busy.discard(user_key) + return None + snapshot_url = "file://" + os.path.join( + state_dir, vm, "resident.snapshot") + try: + hibernate(resident_task, snapshot_url) + except Exception: + with self._lock: + self._busy.discard(user_key) + return None + with self._lock: + self._busy.discard(user_key) + self._entries.pop(user_key, None) + self._save_registry() + self._emit("resident_pool_reclaim", resident_task, + "vm=%s snapshot=%s" % (vm, snapshot_url)) + return vm + + def reclaim_idle(self, now: float | None = None) -> list[str]: + """空闲回收：paused 超阈值的条目 hibernate（pause + + snapshot + shutdown）并删条目。阈值 0 即关闭回收； + 未到阈值/正忙/无 paused_at 的条目不动。返回回收的 + VM 名。""" + if self.idle_minutes <= 0: + return [] + if now is None: + now = time.time() + due: list[str] = [] + with self._lock: + for key, entry in self._entries.items(): + if entry.get("state") != "paused": + continue + if key in self._busy: + continue + paused_at = entry.get("paused_at") + if paused_at is None: + continue + if now - paused_at > self.idle_minutes * 60: + due.append(key) + reclaimed: list[str] = [] + for key in due: + vm = self._hibernate_one(key) + if vm is not None: + reclaimed.append(vm) + return reclaimed + + # -- 孤儿清理 -------------------------------------------------------- + def _shutdown_one(self, user_key: str) -> str | None: + """单条目尽力 shutdown 并从内存删掉（失败仍计入 + 已清理，与 orphans 旧语义一致：不因单条卡住整批）。 + 返回 VM 名或 None（条目已不在）。""" + with self._lock: + entry = self._entries.get(user_key) + if entry is None: + return None + vm = str(entry.get("vm")) + shutdown = getattr(self.executor, "shutdown_vm", None) + if callable(shutdown): + try: + shutdown(vm) + except Exception: + pass + with self._lock: + self._entries.pop(user_key, None) + return vm + + def reconcile_orphans(self) -> list[str]: + """清理本池内存里的 orphan 条目（重启后读到的上一 + 世条目）：每个尽力 shutdown_vm，条目清掉并落盘。 + 生产启动/关闭不走本方法，走模块级 facade。""" + with self._lock: + orphans = [k for k, e in self._entries.items() + if e.get("state") == "orphan"] + cleaned: list[str] = [] + for key in orphans: + vm = self._shutdown_one(key) + if vm is not None: + cleaned.append(vm) + if cleaned: + with self._lock: + self._save_registry() + self._emit("resident_pool_reconcile", {"id": None}, + "cleaned=%s" % ",".join(cleaned)) + return cleaned diff --git a/cloud/chv/vsock.py b/cloud/chv/vsock.py new file mode 100644 index 0000000..b07b9ba --- /dev/null +++ b/cloud/chv/vsock.py @@ -0,0 +1,223 @@ +"""宿主侧 vsock notify 最小实现（P1，feat-chv-muse-p1）。 + +方向只一个：客机 → 宿主。客机连宿主 CID 2 的 +VSOCK_NOTIFY_PORT 端口发 ready/心跳（客内脚本 +cloud/chv/guest/vsock_notify.py）；宿主在 +``_`` 这个 Unix socket 上监听接收。 + +形态依据（CHV 官方 docs/vsock.md，upstream 原文）： +- ``--vsock cid=3,socket=/tmp/ch.vsock``：cid 是客机 CID， + socket 是宿主侧 Unix socket 路径； +- 客 → 宿主：宿主须先在 ``_`` 上监听 + （如 ``/tmp/ch.vsock_1234``），客机连众所周知的 CID 2 + （Host）； +- 只支持 stream socket。 + +只做就绪辅助：wait_ready() 收到 ready 即返回，供 +_run_with_vm/常驻路径在 SSH 轮询之外多一个早期信号； +SSH 仍是唯一命令通道与就绪判定——vsock 超时、无消息、 +监听失败一律返回 None，绝不因此判任务失败、绝不抛到 +调用方。消息是 JSON 行： + + {"type": "ready"|"heartbeat", "vm": "", "ts": } + +不做宿主 → 客机方向（客机不监听，host.py 协议仍走 SSH）。 +""" + +import json +import logging +import os +import socket +import time + +_LOGGER = logging.getLogger("cloud.cloud_hypervisor") + +# 宿主 CID（vsock 众所周知值，CHV 官方 docs/vsock.md）。 +HOST_CID = 2 +# 客机 CID 默认值（与 CHV_VSOCK_CID 默认值一致；0/1/2 保留）。 +DEFAULT_GUEST_CID = 3 +# notify 端口：客机连 (CID 2, 本端口) 发 ready/心跳； +# 宿主监听 _<本端口>。与客内脚本的端口须一致。 +VSOCK_NOTIFY_PORT = 1024 +# 单行上限：客机发来的行超长即丢弃该连接（防脏数据撑内存）。 +MAX_LINE_BYTES = 65536 +# 合法消息类型。 +VALID_TYPES = ("ready", "heartbeat") + + +def vsock_socket_path(state_dir: str, vm_name: str) -> str: + """该 VM 的 vsock socket 路径（放进 VmConfig 的值）。 + + 与 build_vm_config 的派生一致：任务盘同目录的 + ``state_dir//vsock.sock``。vm_name 须与 + api_socket/任务盘用的名字相同（executor.vm_name(task) + 的返回值，含 chv- 前缀），这里不二次加前缀。""" + return os.path.join(state_dir, vm_name, "vsock.sock") + + +def notify_socket_path(socket_path: str, + port: int = VSOCK_NOTIFY_PORT) -> str: + """宿主侧监听路径：``_``（CHV 官方约定）。""" + return "%s_%d" % (socket_path, port) + + +def encode_notify(message_type: str, vm: str, + ts: float | None = None) -> bytes: + """编码一条 notify 消息为 JSON 行（含换行）。 + + type 非法或 vm 为空即 ValueError（调用方是可信的客内 + 脚本/宿主代码，编码错误是编程错误，早炸）。""" + if message_type not in VALID_TYPES: + raise ValueError("vsock notify 类型非法：%r" % (message_type,)) + if not vm: + raise ValueError("vsock notify vm 不能为空") + if ts is None: + ts = time.time() + line = json.dumps({"type": message_type, "vm": vm, "ts": ts}, + separators=(",", ":")) + return (line + "\n").encode("utf-8") + + +def decode_notify(line: bytes | str) -> dict | None: + """解码一行 notify 消息：合法返回 dict，否则返回 None。 + + 客机字节不可信，任何畸形（非 JSON、非对象、类型不在 + ready/heartbeat 内、vm 空/非字符串、ts 非数字）一律 + None，绝不抛异常。""" + try: + if isinstance(line, bytes): + line = line.decode("utf-8") + obj = json.loads(line) + except (UnicodeDecodeError, ValueError): + return None + if not isinstance(obj, dict): + return None + if obj.get("type") not in VALID_TYPES: + return None + if not isinstance(obj.get("vm"), str) or not obj.get("vm"): + return None + if not isinstance(obj.get("ts"), (int, float)) \ + or isinstance(obj.get("ts"), bool): + return None + return obj + + +def _bind_notify_listener(listener, notify_path: str) -> bool: + """bind notify socket：残留文件先删一次再重试一次； + 两次都失败即 False（调用方记 warning 后返回 None， + 与原来直抛到外层 except 的行为一致）。""" + try: + listener.bind(notify_path) + return True + except OSError: + pass + try: + os.unlink(notify_path) + except OSError: + pass + try: + listener.bind(notify_path) + return True + except OSError: + return False + + +def listen_notify(notify_path: str, timeout: float, + want_type: str | None = None) -> dict | None: + """在 notify_path（Unix socket）上监听，等一条 notify。 + + 收到首条合法消息即返回（want_type 给定时跳过其它类型 + 继续等，直到超时）；超时、客机断开未发合法消息、监听 + 失败一律返回 None。监听用的 socket 文件用完即删（尽力 + 而为）；bind 前已存在的同名残留文件先删一次再 bind。 + timeout <= 0 即不等，直接返回 None。""" + if timeout is None or timeout <= 0: + return None + deadline = time.monotonic() + timeout + listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + try: + if not _bind_notify_listener(listener, notify_path): + _LOGGER.warning("vsock notify 监听失败：%s", notify_path) + return None + listener.listen(1) + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + return None + listener.settimeout(remaining) + try: + conn, _ = listener.accept() + except socket.timeout: + return None + except OSError: + _LOGGER.warning("vsock notify accept 失败：%s", + notify_path) + return None + try: + return _read_notify(conn, deadline, want_type) + finally: + try: + conn.close() + except OSError: + pass + except OSError: + _LOGGER.warning("vsock notify 监听失败：%s", notify_path) + return None + finally: + try: + listener.close() + except OSError: + pass + try: + os.unlink(notify_path) + except OSError: + pass + + +def _read_notify(conn, deadline: float, + want_type: str | None) -> dict | None: + """从一条已接受的连接上读 JSON 行，直到合法消息或超时/ + 断开。want_type 给定时跳过其它合法类型继续读。""" + try: + stream = conn.makefile("rb") + except OSError: + return None + try: + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + return None + try: + conn.settimeout(remaining) + line = stream.readline(MAX_LINE_BYTES + 1) + except (OSError, ValueError): + return None + if not line: + return None + if len(line) > MAX_LINE_BYTES: + return None + msg = decode_notify(line) + if msg is None: + continue + if want_type is not None and msg["type"] != want_type: + continue + return msg + finally: + try: + stream.close() + except OSError: + pass + + +def wait_ready(state_dir: str, vm_name: str, + timeout: float = 30, + port: int = VSOCK_NOTIFY_PORT) -> dict | None: + """等一台 VM 的 vsock ready（就绪辅助，非 SSH 判定）。 + + 在 ``state_dir//vsock.sock_`` 上监听，直到 + 收到 {"type": "ready"} 或超时。返回 ready 消息 dict， + 超时/失败返回 None。调用方（_run_with_vm/常驻路径） + 拿到 None 时继续走 SSH 就绪轮询，绝不因此失败。""" + path = notify_socket_path( + vsock_socket_path(state_dir, vm_name), port) + return listen_notify(path, timeout, want_type="ready") diff --git a/cloud/chv/worker_adapter.py b/cloud/chv/worker_adapter.py index b7b32e5..efdd1c8 100644 --- a/cloud/chv/worker_adapter.py +++ b/cloud/chv/worker_adapter.py @@ -12,6 +12,9 @@ cloud.logging_config，不 import cloud.worker，绝无循环。 （原 Worker.__init__ 里的闭包本体）。 - reconcile_warm_pool：生产残留清理 facade 调用 + 审计 （原 Worker._reconcile_warm_pool 的本体）。 +- reconcile_resident_pool：常驻池残留清理 facade 调用 + + 审计（新，feat-chv-muse-p1；调 resident_pool 的模块级 + facade，不构造 ResidentPool、不碰实验接口）。 - cancel_watchdog_loop / quota_watchdog_loop：同步路径 两条看门狗循环（原 Worker._chv_*_watchdog 的本体， self 依赖全部改成回调参数）。 @@ -115,6 +118,38 @@ def reconcile_warm_pool(executor: object, audit_fn: Callable, return cleaned +def reconcile_resident_pool(executor: object, audit_fn: Callable, + when: str) -> list: + """显式调常驻残留清理 facade，返回已清理 VM 名。 + + 仅 CHV worker 有效（其它执行器返回 []）。实现是 + cloud.chv.resident_pool.reconcile_resident_orphans_ + for_executor()：直接读 resident-pool.json 做清理，不 + 构造 ResidentPool、不碰实验接口。调用失败不炸主流程， + 但进审计（失败本身也审计，不静默丢弃）。orphan 条目 + 绝不发给任务，reconcile 只做清理。""" + if not isinstance(executor, CloudHypervisorExecutor): + return [] + try: + from cloud.chv.resident_pool import ( + reconcile_resident_orphans_for_executor as _reconcile) + cleaned = _reconcile(executor) or [] + except Exception as exc: + try: + audit_fn(None, "chv_resident_pool_reconcile_failed", + "when=%s error=%s" % (when, exc)) + except Exception: + pass + return [] + if cleaned: + try: + audit_fn(None, "chv_resident_pool_reconcile", + "when=%s cleaned=%s" % (when, ",".join(cleaned))) + except Exception: + pass + return cleaned + + def cancel_watchdog_loop(task_id: str, vm_name: str | None, stop_event: threading.Event, cancel_requested: set, diff --git a/cloud/tests/test_chv_arch_maint_fe.py b/cloud/tests/test_chv_arch_maint_fe.py index 56a8a78..dbbc8cf 100644 --- a/cloud/tests/test_chv_arch_maint_fe.py +++ b/cloud/tests/test_chv_arch_maint_fe.py @@ -31,6 +31,7 @@ try: from cloud.chv.compat215 import REPLACEMENT_POINTS from cloud.chv.compat215 import resolve_inside, tmpfs_isolated from cloud.chv.config import _assemble_chv_config + from cloud.chv.config import _balloon_vsock_fragment from cloud.chv.config import _base_fragment from cloud.chv.config import _disk_fragment from cloud.chv.config import _network_fragment @@ -48,6 +49,7 @@ except ImportError: from cloud.chv.compat215 import REPLACEMENT_POINTS from cloud.chv.compat215 import resolve_inside, tmpfs_isolated from cloud.chv.config import _assemble_chv_config + from cloud.chv.config import _balloon_vsock_fragment from cloud.chv.config import _base_fragment from cloud.chv.config import _disk_fragment from cloud.chv.config import _network_fragment @@ -76,6 +78,8 @@ EXPECTED_FIELDS = { "snapshot_max_total_gb", "snapshot_retention_days", "snapshot_dir", "allow_remote_snapshot", "snapshot_url", "task_timeout", "api_timeout", "ready_timeout", "ready_interval", + # P1-B（feat-chv-muse-p1）：气球/vsock 三字段。 + "balloon", "vsock", "vsock_cid", } @@ -249,13 +253,15 @@ class ConfigFragmentsTest(unittest.TestCase): def test_fragment_keys_disjoint_and_union_is_field_set(self): frags = [_base_fragment({}, {}), _ssh_fragment({}, {}), _network_fragment({}, {}), _quota_fragment({}), - _disk_fragment({}), _timeout_fragment({}, {})] + _disk_fragment({}), _timeout_fragment({}, {}), + # P1-B：新片段加入并集断言（字段 38 → 41）。 + _balloon_vsock_fragment({})] seen = set() for frag in frags: self.assertFalse(seen & set(frag), "片段键重叠") seen |= set(frag) self.assertEqual(seen, EXPECTED_FIELDS) - self.assertEqual(len(EXPECTED_FIELDS), 38) + self.assertEqual(len(EXPECTED_FIELDS), 41) def test_defaults_and_overrides(self): cfg = build_chv_config({}, {}) diff --git a/cloud/tests/test_chv_closeup.py b/cloud/tests/test_chv_closeup.py index 52b4741..9cd94af 100644 --- a/cloud/tests/test_chv_closeup.py +++ b/cloud/tests/test_chv_closeup.py @@ -598,8 +598,16 @@ class VmDeathTest(CloseupCase): def runner(cmd, timeout=None): self.calls.append(list(cmd)) if cmd[0] == "ssh" and "dispatch" in cmd: - proc = ex._vms.get("chv-abc123") - if proc is not None: + # 死亡模拟须杀当时已登记的 VM（即实际跑命令 + # 的那台）：P1 常驻路由下 dispatch 跑在常驻 VM + #（chv-resident-*）上，任务自己的 chv-abc123 + # 从未 wake、登记里根本没有，盯着它杀是空操 + # 作、模拟落空（探针实证：登记里只有 + # chv-resident-default 且活着，异常直抛）。 + # 常驻 VM 真死时 _run_on_resident 的判定已验 + # 证正确（两态皆 interrupted），这里只修模拟 + # 器的靶子，不动断言与判定逻辑。 + for proc in list(ex._vms.values()): proc._alive = False if mode == "raise": raise RuntimeError("ssh broken") diff --git a/cloud/tests/test_chv_muse_p1_balloon_vsock.py b/cloud/tests/test_chv_muse_p1_balloon_vsock.py new file mode 100644 index 0000000..5078336 --- /dev/null +++ b/cloud/tests/test_chv_muse_p1_balloon_vsock.py @@ -0,0 +1,423 @@ +"""P1-B（feat-chv-muse-p1）：气球/vsock 配置、payload 与 vsock 编解码测试。 + +覆盖三件： + +- config：CHV_BALLOON/CHV_VSOCK 开关默认开、只认 0/1， + CHV_VSOCK_CID 默认 3 且拒绝保留 CID（0/1/2）与非法值； +- executor：build_vm_config 默认带 balloon/vsock 设备，开关 + 关时对应键整个不出现，vsock socket 路径按任务盘派生； +- vsock（宿主侧）与客内脚本：JSON 行编解码、监听/超时、 + 客内脚本静默失败（连不上宿主即 exit 0，不影响 sshd/任务）。 + +全部用假执行器与本地 Unix socket，绝不连真实网络、不起 +真实 VM。CHV 54 的 balloon size=0 与 vsock 端到端收发尚未 +真机验证（本机无 /dev/kvm，第三棒占着 WSL 测试机），见 +汇报的待验证项——这里只钉住配置层 + payload 形态。 +""" + +import ast +import importlib.util +import json +import os +import socket +import sys +import tempfile +import threading +import time +import unittest +from pathlib import Path + +from cloud.chv import vsock as chv_vsock +from cloud.chv.config import _balloon_vsock_fragment +from cloud.chv.config import _base_fragment +from cloud.chv.config import _disk_fragment +from cloud.chv.config import _network_fragment +from cloud.chv.config import _quota_fragment +from cloud.chv.config import _ssh_fragment +from cloud.chv.config import _timeout_fragment +from cloud.chv.config import build_chv_config +from cloud.chv.executor import CloudHypervisorExecutor + +ROOT = Path(__file__).resolve().parents[2] +GUEST_SCRIPT = ROOT / "cloud" / "chv" / "guest" / "vsock_notify.py" + + +def _load_guest_module(): + spec = importlib.util.spec_from_file_location( + "vsock_notify_guest", str(GUEST_SCRIPT)) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def _stdlib_only(path): + tree = ast.parse(Path(path).read_text()) + mods = set() + for node in ast.walk(tree): + if isinstance(node, ast.Import): + mods.update(a.name.split(".")[0] for a in node.names) + elif isinstance(node, ast.ImportFrom) and node.module: + mods.add(node.module.split(".")[0]) + return mods <= set(sys.stdlib_module_names) + + +class BalloonVsockConfigTest(unittest.TestCase): + def test_defaults_on(self): + cfg = build_chv_config({}, {}) + self.assertTrue(cfg.balloon) + self.assertTrue(cfg.vsock) + self.assertEqual(cfg.vsock_cid, 3) + + def test_switches_off(self): + cfg = build_chv_config( + {}, {"CHV_BALLOON": "0", "CHV_VSOCK": "0"}) + self.assertFalse(cfg.balloon) + self.assertFalse(cfg.vsock) + self.assertEqual(cfg.vsock_cid, 3) + + def test_switches_on_explicit(self): + cfg = build_chv_config( + {}, {"CHV_BALLOON": "1", "CHV_VSOCK": "1"}) + self.assertTrue(cfg.balloon) + self.assertTrue(cfg.vsock) + + def test_illegal_switch_values_raise(self): + for env in ({"CHV_BALLOON": "2"}, {"CHV_BALLOON": "yes"}, + {"CHV_BALLOON": "true"}, {"CHV_VSOCK": "on"}, + {"CHV_VSOCK": ""}): + with self.assertRaises(ValueError): + build_chv_config({}, env) + + def test_vsock_cid_override(self): + cfg = build_chv_config({}, {"CHV_VSOCK_CID": "7"}) + self.assertEqual(cfg.vsock_cid, 7) + cfg2 = build_chv_config({}, {"CHV_VSOCK_CID": "4294967295"}) + self.assertEqual(cfg2.vsock_cid, 4294967295) + + def test_vsock_cid_reserved_and_illegal_raise(self): + # 0/1/2 是 vsock 保留 CID（Hypervisor/Loopback/Host）， + # -1 随机 CID 也被 CHV vm.create 400 拒绝。 + for cid in ("0", "1", "2", "-1", "4294967296", "abc", ""): + with self.assertRaises(ValueError, msg=cid): + build_chv_config({}, {"CHV_VSOCK_CID": cid}) + + def test_fragment_keys_disjoint(self): + frag = _balloon_vsock_fragment({}) + self.assertEqual(set(frag), {"balloon", "vsock", "vsock_cid"}) + others = [_base_fragment({}, {}), _ssh_fragment({}, {}), + _network_fragment({}, {}), _quota_fragment({}), + _disk_fragment({}), _timeout_fragment({}, {})] + seen = set() + for other in others: + seen |= set(other) + self.assertFalse(seen & set(frag), "新片段键与旧片段冲突") + + +class _PayloadCase(unittest.TestCase): + def setUp(self): + self.dir = tempfile.mkdtemp(prefix="chv-p1b-") + self.state_dir = os.path.join(self.dir, "state") + os.makedirs(self.state_dir) + + def tearDown(self): + import shutil + shutil.rmtree(self.dir, ignore_errors=True) + + def make_exec(self, env_extra=None): + env = dict(env_extra or {}) + return CloudHypervisorExecutor( + kernel=os.path.join(self.dir, "vmlinux"), + rootfs=os.path.join(self.dir, "root.raw"), + ssh_key=os.path.join(self.dir, "id_ed25519"), + state_dir=self.state_dir, + runner=lambda cmd, timeout=None: None, + launcher=lambda cmd: None, + api_client=lambda *a: (200, {}), + sleeper=lambda s: None, env=env) + + +class BalloonVsockPayloadTest(_PayloadCase): + def test_default_payload_has_balloon_and_vsock(self): + ex = self.make_exec() + cfg = ex.build_vm_config() + # CHV 54 BalloonConfig：size 是目标气球尺寸（字节）， + # 0 = 启动不预扣，只靠 free_page_reporting 被动回收。 + self.assertEqual(cfg["balloon"], { + "size": 0, "deflate_on_oom": True, + "free_page_reporting": True}) + self.assertEqual(cfg["vsock"], { + "cid": 3, + "socket": os.path.join(self.state_dir, "vsock.sock")}) + + def test_switches_off_drop_keys(self): + ex = self.make_exec({"CHV_BALLOON": "0", "CHV_VSOCK": "0"}) + cfg = ex.build_vm_config() + self.assertNotIn("balloon", cfg) + self.assertNotIn("vsock", cfg) + # 旧键原样保留。 + self.assertIn("cpus", cfg) + self.assertIn("memory", cfg) + self.assertIn("payload", cfg) + self.assertIn("disks", cfg) + self.assertIn("rng", cfg) + + def test_balloon_off_only(self): + ex = self.make_exec({"CHV_BALLOON": "0"}) + cfg = ex.build_vm_config() + self.assertNotIn("balloon", cfg) + self.assertIn("vsock", cfg) + + def test_vsock_cid_honored(self): + ex = self.make_exec({"CHV_VSOCK_CID": "9"}) + cfg = ex.build_vm_config() + self.assertEqual(cfg["vsock"]["cid"], 9) + + def test_vsock_socket_derived_from_task_disk_dir(self): + ex = self.make_exec() + disk = os.path.join(self.state_dir, "chv-t1", "disk.raw") + cfg = ex.build_vm_config(disk) + self.assertEqual( + cfg["vsock"]["socket"], + os.path.join(self.state_dir, "chv-t1", "vsock.sock")) + + def test_vsock_socket_falls_back_when_disk_outside_state(self): + # reuse/lab：任务盘即基础镜像（state_dir 外），socket + # 落 state_dir/vsock.sock（单 VM 可用，见 helper 注释）。 + ex = self.make_exec() + cfg = ex.build_vm_config() + self.assertEqual( + cfg["vsock"]["socket"], + os.path.join(self.state_dir, "vsock.sock")) + + def test_explicit_vsock_socket_wins(self): + ex = self.make_exec() + cfg = ex.build_vm_config( + vsock_socket="/tmp/per-vm/v.sock") + self.assertEqual(cfg["vsock"]["socket"], "/tmp/per-vm/v.sock") + + def test_helper_direct(self): + ex = self.make_exec({"CHV_BALLOON": "0"}) + extra = ex._balloon_vsock_payload( + vsock_socket="/tmp/s.sock") + self.assertEqual(extra, {"vsock": {"cid": 3, + "socket": "/tmp/s.sock"}}) + + +class VsockCodecTest(unittest.TestCase): + def test_encode_decode_round_trip(self): + raw = chv_vsock.encode_notify("ready", "chv-t1", ts=1234.5) + self.assertTrue(raw.endswith(b"\n")) + msg = chv_vsock.decode_notify(raw) + self.assertEqual(msg, {"type": "ready", "vm": "chv-t1", + "ts": 1234.5}) + + def test_encode_rejects_bad_type_or_vm(self): + with self.assertRaises(ValueError): + chv_vsock.encode_notify("bogus", "chv-t1") + with self.assertRaises(ValueError): + chv_vsock.encode_notify("ready", "") + + def test_decode_rejects_garbage(self): + cases = [b"not json\n", b"[1, 2]\n", b"123\n", b"\n", + b'{"type": "bogus", "vm": "v", "ts": 1}\n', + b'{"type": "ready", "vm": "", "ts": 1}\n', + b'{"type": "ready", "vm": 7, "ts": 1}\n', + b'{"type": "ready", "vm": "v"}\n', + b'{"type": "ready", "vm": "v", "ts": "x"}\n', + b'{"type": "ready", "vm": "v", "ts": true}\n', + b'\xff\xfe bad utf8\n'] + for line in cases: + self.assertIsNone(chv_vsock.decode_notify(line), line) + + def test_decode_accepts_int_ts(self): + msg = chv_vsock.decode_notify( + b'{"type": "heartbeat", "vm": "v", "ts": 7}') + self.assertEqual(msg["ts"], 7) + + def test_paths(self): + self.assertEqual( + chv_vsock.vsock_socket_path("/s", "chv-t1"), + "/s/chv-t1/vsock.sock") + self.assertEqual( + chv_vsock.notify_socket_path("/s/chv-t1/vsock.sock"), + "/s/chv-t1/vsock.sock_1024") + self.assertEqual(chv_vsock.HOST_CID, 2) + self.assertEqual(chv_vsock.VSOCK_NOTIFY_PORT, 1024) + + +def _wait_for_socket(path, timeout=2.0): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if os.path.exists(path): + return True + time.sleep(0.02) + return False + + +class VsockListenTest(unittest.TestCase): + def setUp(self): + self.dir = tempfile.mkdtemp(prefix="chv-p1b-vsock-") + + def tearDown(self): + import shutil + shutil.rmtree(self.dir, ignore_errors=True) + + def _client(self, path, lines, delay=0.05): + """后台线程：等 socket 出现，连上发 lines 后关闭。 + + connect 带重试：监听方 bind 与 listen 之间、以及删 + 残留文件重 bind 的窗口里，单次 connect 可能被拒； + 须重试到成功或超时，否则测试会偶发失败。""" + def run(): + if not _wait_for_socket(path): + return + deadline = time.monotonic() + 5 + sock = None + while time.monotonic() < deadline: + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + try: + sock.connect(path) + break + except OSError: + try: + sock.close() + except OSError: + pass + sock = None + time.sleep(0.02) + if sock is None: + return + try: + time.sleep(delay) + for line in lines: + sock.sendall(line) + except OSError: + pass + finally: + try: + sock.close() + except OSError: + pass + thread = threading.Thread(target=run, daemon=True) + thread.start() + return thread + + def test_listen_returns_ready(self): + path = os.path.join(self.dir, "n.sock_1024") + thread = self._client( + path, [chv_vsock.encode_notify("ready", "chv-t1")]) + msg = chv_vsock.listen_notify(path, timeout=5) + thread.join(timeout=5) + self.assertIsNotNone(msg) + self.assertEqual(msg["type"], "ready") + self.assertEqual(msg["vm"], "chv-t1") + self.assertFalse(os.path.exists(path), "监听完须清 socket 文件") + + def test_listen_timeout_returns_none(self): + path = os.path.join(self.dir, "t.sock_1024") + start = time.monotonic() + self.assertIsNone(chv_vsock.listen_notify(path, timeout=0.3)) + self.assertLess(time.monotonic() - start, 2) + self.assertFalse(os.path.exists(path)) + + def test_listen_skips_invalid_lines_and_wrong_type(self): + path = os.path.join(self.dir, "w.sock_1024") + thread = self._client(path, [ + b"not json\n", + chv_vsock.encode_notify("heartbeat", "chv-t1"), + chv_vsock.encode_notify("ready", "chv-t1"), + ]) + msg = chv_vsock.listen_notify(path, timeout=5, + want_type="ready") + thread.join(timeout=5) + self.assertIsNotNone(msg) + self.assertEqual(msg["type"], "ready") + + def test_listen_survives_stale_socket_file(self): + path = os.path.join(self.dir, "s.sock_1024") + with open(path, "w") as fh: + fh.write("stale") + thread = self._client( + path, [chv_vsock.encode_notify("ready", "chv-t1")]) + msg = chv_vsock.listen_notify(path, timeout=5) + thread.join(timeout=5) + self.assertIsNotNone(msg) + self.assertEqual(msg["type"], "ready") + + def test_wait_ready_end_to_end(self): + vm_dir = os.path.join(self.dir, "chv-t1") + os.makedirs(vm_dir) + path = chv_vsock.notify_socket_path( + chv_vsock.vsock_socket_path(self.dir, "chv-t1")) + thread = self._client( + path, [chv_vsock.encode_notify("ready", "chv-t1")]) + msg = chv_vsock.wait_ready(self.dir, "chv-t1", timeout=5) + thread.join(timeout=5) + self.assertIsNotNone(msg) + self.assertEqual(msg["type"], "ready") + self.assertEqual(msg["vm"], "chv-t1") + + def test_wait_ready_timeout_returns_none(self): + vm_dir = os.path.join(self.dir, "chv-t2") + os.makedirs(vm_dir) + self.assertIsNone( + chv_vsock.wait_ready(self.dir, "chv-t2", timeout=0.3)) + + +class GuestNotifyScriptTest(unittest.TestCase): + def test_script_exists_and_stdlib_only(self): + self.assertTrue(GUEST_SCRIPT.is_file()) + self.assertTrue(_stdlib_only(GUEST_SCRIPT)) + self.assertTrue( + _stdlib_only(ROOT / "cloud" / "chv" / "vsock.py")) + + def test_vm_name_sources(self): + mod = _load_guest_module() + self.assertEqual(mod._vm_name(["prog", "chv-t1"]), "chv-t1") + old = os.environ.pop("CHV_VM_NAME", None) + try: + os.environ["CHV_VM_NAME"] = "chv-env" + self.assertEqual(mod._vm_name(["prog"]), "chv-env") + finally: + if old is None: + os.environ.pop("CHV_VM_NAME", None) + else: + os.environ["CHV_VM_NAME"] = old + name = mod._vm_name(["prog"]) + self.assertIsInstance(name, str) + self.assertTrue(name) + + def test_send_writes_json_line(self): + mod = _load_guest_module() + client, server = socket.socketpair() + try: + mod._send(client, "heartbeat", "chv-t1") + server.settimeout(2) + data = server.recv(4096) + msg = json.loads(data.decode("utf-8")) + self.assertEqual(msg["type"], "heartbeat") + self.assertEqual(msg["vm"], "chv-t1") + self.assertIsInstance(msg["ts"], float) + finally: + client.close() + server.close() + + def test_main_exits_silently_when_host_unreachable(self): + # 本机无 vsock 监听：main 应在重试窗口内静默返回 0， + # 绝不抛异常、不影响调用方。 + mod = _load_guest_module() + old = mod.CONNECT_RETRY_SECONDS + mod.CONNECT_RETRY_SECONDS = 0 + try: + self.assertEqual(mod.main(["prog", "chv-t1"]), 0) + finally: + mod.CONNECT_RETRY_SECONDS = old + + def test_ports_agree(self): + mod = _load_guest_module() + self.assertEqual(mod.NOTIFY_PORT, chv_vsock.VSOCK_NOTIFY_PORT) + self.assertEqual(mod.HOST_CID, chv_vsock.HOST_CID) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_chv_muse_p1_resident.py b/cloud/tests/test_chv_muse_p1_resident.py new file mode 100644 index 0000000..8a3b260 --- /dev/null +++ b/cloud/tests/test_chv_muse_p1_resident.py @@ -0,0 +1,730 @@ +"""CHV 常驻池（feat-chv-muse-p1）测试。 + +覆盖 cloud/chv/resident_pool.py、executor 的常驻接线 +（_run_with_vm 常驻/冷启动两路、finally 语义）与 +worker_adapter.reconcile_resident_pool facade。全部用假 +执行器/假 VMM，绝不起真实 VM、不连真实网络。 + +断言点（按任务书）： +- acquire/resume 失败回落冷启动，失败条目丢弃； +- release 只 pause 不销毁，pause 失败条目留 running； +- registry 损坏 quarantine、单坏条目不连坐； +- 重启后条目一律 orphan 不复用，只能 reconcile 清理； +- _run_with_vm 常驻命中跑完 pause、冷启动跑完 shutdown， + 客内抛错时两路 finally 都执行； +- 带密钥 env / 未启用 / 建池失败时回落冷启动，任务不丢； +- 每用户隔离：release 只动本用户条目； +- 空闲回收只动超阈值 paused 条目，阈值 0 即关闭； +- WarmPool 实验模块不动（wired_to_dispatch 仍 False）。 +""" + +import json +import os +import re +import tempfile +import unittest +from types import SimpleNamespace +from unittest import mock + +from cloud.chv import worker_adapter +from cloud.chv.errors import ChvFailure +from cloud.chv.executor import CloudHypervisorExecutor +from cloud.chv.experimental_pool import WarmPool +from cloud.chv.resident_pool import ( + ResidentPool, + reconcile_resident_orphans_for_executor, + resident_task_for_user, + resident_user_key, +) + + +class _FakeExecutor: + """ResidentPool 用的鸭子型执行器：记录调用，可注入失败。 + + wait 失败时模拟真实语义：先自行 shutdown 再抛错 + （wait_guest_ready 的失败路径就是 stop_vm 后 raise）。 + """ + + def __init__(self, state_dir, env=None): + self.state_dir = state_dir + self._env = env or {} + self.resumes = [] + self.pauses = [] + self.wakes = [] + self.waits = [] + self.shutdowns = [] + self.hibernates = [] + self.emits = [] + self.resume_fail = False + self.wait_fail = False + self.wake_fail = False + self.pause_fail = False + self.hibernate_fail = False + self.shutdown_fail = False + + def vm_name(self, task_or_name): + raw = task_or_name.get("id") if isinstance( + task_or_name, dict) else str(task_or_name) + safe = re.sub(r"[^A-Za-z0-9_.-]", "-", raw) + if safe.startswith("chv-"): + return safe + return "chv-" + safe + + def resume_vm(self, task_or_name): + self.resumes.append(self.vm_name(task_or_name)) + if self.resume_fail: + raise RuntimeError("resume boom") + return True + + def pause_vm(self, task_or_name): + self.pauses.append(self.vm_name(task_or_name)) + if self.pause_fail: + raise RuntimeError("pause boom") + return True + + def wake_vm(self, task_or_name): + self.wakes.append(self.vm_name(task_or_name)) + if self.wake_fail: + raise RuntimeError("wake boom") + return {"vm": self.vm_name(task_or_name), "mode": "cold"} + + def wait_guest_ready(self, task): + self.waits.append(self.vm_name(task)) + if self.wait_fail: + self.shutdown_vm(task) + raise ChvFailure("guest_not_ready", "ssh 未就绪") + return True + + def shutdown_vm(self, task_or_name): + self.shutdowns.append(self.vm_name(task_or_name)) + if self.shutdown_fail: + raise RuntimeError("shutdown boom") + return True + + def hibernate_vm(self, task_or_name, snapshot_url): + self.hibernates.append( + (self.vm_name(task_or_name), snapshot_url)) + if self.hibernate_fail: + raise RuntimeError("hibernate boom") + return {"vm": self.vm_name(task_or_name), + "state": "hibernated", "snapshot": snapshot_url} + + def _emit(self, event, task, detail="", kind=None): + self.emits.append(event) + + +def _write(path, content): + with open(path, "w") as fh: + fh.write(content) + return path + + +def _registry_path(state_dir): + return os.path.join(state_dir, "resident-pool.json") + + +class ResidentKeyTest(unittest.TestCase): + def test_user_key_sanitizes_and_defaults(self): + self.assertEqual(resident_user_key(None), "default") + self.assertEqual(resident_user_key(""), "default") + self.assertEqual(resident_user_key("alice"), "alice") + self.assertEqual(resident_user_key("a/b c"), "a-b-c") + self.assertEqual(resident_task_for_user("alice"), + {"id": "resident-alice"}) + self.assertEqual(resident_task_for_user(None), + {"id": "resident-default"}) + + +class ResidentPoolAcquireTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.fake = _FakeExecutor(self.tmp.name) + + def tearDown(self): + self.tmp.cleanup() + + def _provisioned_pool(self, user="alice"): + pool = ResidentPool(self.fake) + got = pool.provision(user) + self.assertEqual(got["mode"], "resident") + return pool + + def test_provision_then_acquire_then_release(self): + pool = self._provisioned_pool() + self.assertEqual(self.fake.wakes, ["chv-resident-alice"]) + self.assertEqual(self.fake.pauses, ["chv-resident-alice"]) + self.assertEqual(pool.paused_count(), 1) + got = pool.acquire("alice") + self.assertEqual(got["mode"], "resident") + self.assertEqual(got["vm"], "chv-resident-alice") + self.assertEqual(got["task"], {"id": "resident-alice"}) + self.assertEqual(self.fake.resumes, ["chv-resident-alice"]) + self.assertEqual(self.fake.waits, ["chv-resident-alice"]) + self.assertEqual(pool.paused_count(), 0) + # running 中的条目不接受第二个 acquire（回落冷启动）。 + self.assertEqual(pool.acquire("alice")["mode"], "cold") + self.assertEqual(len(self.fake.resumes), 1) + # release 只 pause，不销毁。 + self.assertTrue(pool.release("alice")) + self.assertEqual( + self.fake.pauses, + ["chv-resident-alice", "chv-resident-alice"]) + self.assertEqual(self.fake.shutdowns, []) + self.assertEqual(pool.paused_count(), 1) + # 再取仍是同一台 VM。 + got = pool.acquire("alice") + self.assertEqual(got["vm"], "chv-resident-alice") + + def test_resume_failure_falls_back_to_cold_and_drops_entry(self): + pool = self._provisioned_pool() + self.fake.resume_fail = True + got = pool.acquire("alice") + self.assertEqual(got, {"mode": "cold", "vm": None, + "task": None}) + self.assertEqual(pool.paused_count(), 0) + with open(_registry_path(self.tmp.name)) as fh: + self.assertEqual(json.load(fh), {}) + + def test_wait_failure_falls_back_to_cold_and_drops_entry(self): + pool = self._provisioned_pool() + self.fake.wait_fail = True + got = pool.acquire("alice") + self.assertEqual(got["mode"], "cold") + self.assertEqual(pool.paused_count(), 0) + # wait 的失败路径已自行 stop_vm（模拟真实语义）。 + self.assertEqual(self.fake.shutdowns, + ["chv-resident-alice"]) + with open(_registry_path(self.tmp.name)) as fh: + self.assertEqual(json.load(fh), {}) + + def test_release_without_running_entry_is_false(self): + pool = self._provisioned_pool() + self.assertFalse(pool.release("alice")) + self.assertFalse(pool.release("nobody")) + self.assertEqual(self.fake.pauses, ["chv-resident-alice"]) + + def test_release_pause_failure_keeps_entry_running(self): + pool = self._provisioned_pool() + pool.acquire("alice") + self.fake.pause_fail = True + self.assertFalse(pool.release("alice")) + # 不删条目、不销毁：下次 acquire 仍是 running→cold， + # VM 本身没被 shutdown，留待 worker 关闭时清理。 + self.assertEqual(self.fake.shutdowns, []) + self.assertEqual(pool.acquire("alice")["mode"], "cold") + + def test_provision_is_idempotent_and_per_user_isolated(self): + pool = ResidentPool(self.fake) + self.assertEqual(pool.provision("alice")["mode"], "resident") + self.assertEqual(pool.provision("bob")["mode"], "resident") + self.assertEqual(pool.provision("alice")["mode"], "resident") + self.assertEqual(self.fake.wakes.count("chv-resident-alice"), 1) + self.assertEqual(self.fake.wakes.count("chv-resident-bob"), 1) + self.assertEqual(pool.paused_count(), 2) + # release 只动本用户条目。 + self.assertFalse(pool.release("bob")) # bob 未 running + pool.acquire("alice") + self.assertTrue(pool.release("alice")) + self.assertEqual(pool.paused_count(), 2) + got = pool.acquire("bob") + self.assertEqual(got["vm"], "chv-resident-bob") + self.assertEqual(pool.paused_count(), 1) + self.assertEqual(self.fake.resumes, ["chv-resident-alice", + "chv-resident-bob"]) + + def test_provision_wake_failure_falls_back_and_cleans_up(self): + pool = ResidentPool(self.fake) + self.fake.wake_fail = True + got = pool.provision("alice") + self.assertEqual(got, {"mode": "cold", "vm": None, + "task": None}) + # 建池失败尽力 shutdown，不留半醒 VM 与条目。 + self.assertEqual(self.fake.shutdowns, + ["chv-resident-alice"]) + self.assertEqual(pool.paused_count(), 0) + with open(_registry_path(self.tmp.name)) as fh: + self.assertEqual(json.load(fh), {}) + + +class ResidentPoolRegistryTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.fake = _FakeExecutor(self.tmp.name) + + def tearDown(self): + self.tmp.cleanup() + + def test_corrupt_registry_is_quarantined(self): + path = _write(_registry_path(self.tmp.name), "{not json") + pool = ResidentPool(self.fake) + self.assertTrue(pool.registry_corrupt) + self.assertEqual(pool.paused_count(), 0) + self.assertEqual(pool.orphan_count(), 0) + self.assertTrue(os.path.exists(path + ".corrupt")) + self.assertFalse(os.path.exists(path)) + + def test_top_level_non_dict_is_corrupt(self): + path = _write(_registry_path(self.tmp.name), "[1, 2]") + pool = ResidentPool(self.fake) + self.assertTrue(pool.registry_corrupt) + self.assertTrue(os.path.exists(path + ".corrupt")) + + def test_bad_entries_dropped_without_tainting_good_ones(self): + _write(_registry_path(self.tmp.name), json.dumps({ + "good": {"state": "paused", "vm": "chv-resident-good", + "user_id": "good"}, + "no-vm": {"state": "paused"}, + "not-a-dict": "x", + "wrong-state": {"state": "flying", + "vm": "chv-resident-x"}, + })) + pool = ResidentPool(self.fake) + self.assertEqual(pool.registry_corrupt_entries, 3) + self.assertEqual(pool.orphan_count(), 1) + self.assertEqual(pool.paused_count(), 0) + + def test_restart_never_reuses_entries(self): + pool1 = ResidentPool(self.fake) + pool1.provision("alice") + pool2 = ResidentPool(self.fake) + self.assertEqual(pool2.orphan_count(), 1) + self.assertEqual(pool2.paused_count(), 0) + # 新池绝不把上一世条目发给任务。 + self.assertEqual(pool2.acquire("alice")["mode"], "cold") + self.assertEqual(self.fake.resumes, []) + cleaned = pool2.reconcile_orphans() + self.assertEqual(cleaned, ["chv-resident-alice"]) + self.assertEqual(self.fake.shutdowns, + ["chv-resident-alice"]) + with open(_registry_path(self.tmp.name)) as fh: + self.assertEqual(json.load(fh), {}) + + def test_idle_minutes_env_parsing(self): + self.assertEqual( + ResidentPool(self.fake).idle_minutes, 0) + fake = _FakeExecutor( + self.tmp.name, env={"CHV_RESIDENT_IDLE_MINUTES": "45"}) + self.assertEqual(ResidentPool(fake).idle_minutes, 45) + fake = _FakeExecutor( + self.tmp.name, env={"CHV_RESIDENT_IDLE_MINUTES": "junk"}) + self.assertEqual(ResidentPool(fake).idle_minutes, 0) + fake = _FakeExecutor( + self.tmp.name, env={"CHV_RESIDENT_IDLE_MINUTES": "-5"}) + self.assertEqual(ResidentPool(fake).idle_minutes, 0) + pool = ResidentPool(self.fake, idle_minutes=10) + self.assertEqual(pool.idle_minutes, 10) + + +class ResidentPoolIdleTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.fake = _FakeExecutor(self.tmp.name) + + def tearDown(self): + self.tmp.cleanup() + + def test_idle_disabled_by_default(self): + pool = ResidentPool(self.fake) + pool.provision("alice") + import time + self.assertEqual( + pool.reclaim_idle(now=time.time() + 10 ** 9), []) + self.assertEqual(self.fake.hibernates, []) + + def test_overdue_paused_entry_is_hibernated(self): + import time + pool = ResidentPool(self.fake, idle_minutes=30) + pool.provision("alice") + pool.provision("bob") + now = time.time() + 31 * 60 + reclaimed = pool.reclaim_idle(now=now) + self.assertEqual(reclaimed, ["chv-resident-alice", + "chv-resident-bob"]) + self.assertEqual(len(self.fake.hibernates), 2) + for vm, url in self.fake.hibernates: + self.assertTrue(url.startswith("file://")) + self.assertIn(self.tmp.name, url) + self.assertTrue(url.endswith( + os.path.join(vm, "resident.snapshot"))) + self.assertEqual(pool.paused_count(), 0) + with open(_registry_path(self.tmp.name)) as fh: + self.assertEqual(json.load(fh), {}) + + def test_not_yet_idle_entry_is_kept(self): + import time + pool = ResidentPool(self.fake, idle_minutes=30) + pool.provision("alice") + self.assertEqual( + pool.reclaim_idle(now=time.time() + 10 * 60), []) + self.assertEqual(pool.paused_count(), 1) + + def test_hibernate_failure_keeps_entry_for_retry(self): + import time + pool = ResidentPool(self.fake, idle_minutes=30) + pool.provision("alice") + self.fake.hibernate_fail = True + self.assertEqual( + pool.reclaim_idle(now=time.time() + 31 * 60), []) + self.assertEqual(pool.paused_count(), 1) + + +class ResidentFacadeTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.fake = _FakeExecutor(self.tmp.name) + + def tearDown(self): + self.tmp.cleanup() + + def test_module_facade_cleans_and_empties_registry(self): + _write(_registry_path(self.tmp.name), json.dumps({ + "a": {"state": "paused", "vm": "chv-resident-a"}, + "b": {"state": "running", "vm": "chv-resident-b"}, + "c": {"state": "provisioning", "vm": "chv-resident-c"}, + "d": {"state": "warm", "vm": "chv-resident-d"}, + "e": {"state": "paused"}, + "f": "not-a-dict", + })) + cleaned = reconcile_resident_orphans_for_executor(self.fake) + self.assertEqual(cleaned, ["chv-resident-a", "chv-resident-b", + "chv-resident-c"]) + self.assertEqual(self.fake.shutdowns, cleaned) + with open(_registry_path(self.tmp.name)) as fh: + self.assertEqual(json.load(fh), {}) + self.assertIn("resident_pool_reconcile", self.fake.emits) + + def test_module_facade_without_registry_is_empty(self): + self.assertEqual( + reconcile_resident_orphans_for_executor(self.fake), []) + self.assertEqual(self.fake.shutdowns, []) + + def test_module_facade_without_state_dir_is_empty(self): + fake = _FakeExecutor(None) + fake.state_dir = None + self.assertEqual( + reconcile_resident_orphans_for_executor(fake), []) + +class WorkerAdapterFacadeTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.fake = _FakeExecutor(self.tmp.name) + self.audits = [] + + def tearDown(self): + self.tmp.cleanup() + + def _audit(self, project_id, event, detail): + self.audits.append((project_id, event, detail)) + + def test_facade_cleans_resident_registry(self): + # worker_adapter 的 isinstance 门要求真执行器：用假 + # 的 shutdown 替身包一层真执行器实例（只借类型）。 + ex = _RealExecCase.make_executor(self.tmp.name, self.tmp.name) + ex.shutdown_vm = self.fake.shutdown_vm + ex._emit = self.fake._emit + _write(_registry_path(self.tmp.name), json.dumps({ + "a": {"state": "paused", "vm": "chv-resident-a"}, + "b": {"state": "running", "vm": "chv-resident-b"}, + })) + cleaned = worker_adapter.reconcile_resident_pool( + ex, self._audit, "startup") + self.assertEqual(cleaned, ["chv-resident-a", "chv-resident-b"]) + self.assertEqual(self.fake.shutdowns, cleaned) + self.assertTrue(any( + event == "chv_resident_pool_reconcile" + for _p, event, _d in self.audits)) + with open(_registry_path(self.tmp.name)) as fh: + self.assertEqual(json.load(fh), {}) + + def test_facade_never_constructs_resident_pool(self): + ex = _RealExecCase.make_executor(self.tmp.name, self.tmp.name) + ex.shutdown_vm = self.fake.shutdown_vm + ex._emit = self.fake._emit + _write(_registry_path(self.tmp.name), json.dumps({ + "a": {"state": "paused", "vm": "chv-resident-a"}, + })) + with mock.patch( + "cloud.chv.resident_pool.ResidentPool", + side_effect=AssertionError("must not construct")): + cleaned = worker_adapter.reconcile_resident_pool( + ex, self._audit, "shutdown") + self.assertEqual(cleaned, ["chv-resident-a"]) + + def test_facade_ignores_non_chv_executor(self): + cleaned = worker_adapter.reconcile_resident_pool( + object(), self._audit, "startup") + self.assertEqual(cleaned, []) + self.assertEqual(self.audits, []) + + def test_facade_does_not_touch_warm_pool_reconcile(self): + # reconcile_warm_pool 仍走 warm-pool.json，与常驻无关。 + ex = _RealExecCase.make_executor(self.tmp.name, self.tmp.name) + ex.shutdown_vm = self.fake.shutdown_vm + ex._emit = self.fake._emit + warm_path = os.path.join(self.tmp.name, "warm-pool.json") + _write(warm_path, json.dumps({ + "w": {"state": "warm", "vm": "chv-warm-1", + "task_id": "w"}, + })) + cleaned = worker_adapter.reconcile_warm_pool( + ex, self._audit, "startup") + self.assertEqual(cleaned, ["chv-warm-1"]) + self.assertFalse(os.path.exists(_registry_path(self.tmp.name))) + + +def _write_file(path, content=b"x"): + with open(path, "wb") as fh: + fh.write(content) + os.chmod(path, 0o755) + return path + + +class _RealExecCase(unittest.TestCase): + """真执行器 + 全方法替身：测 _run_with_vm 的接线本身， + 不测 VMM。""" + + @staticmethod + def make_executor(state_dir, base_dir, env_extra=None): + env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888", + "CHV_HOST_KEY_MODE": "keyscan", + "CHV_SSH_FORWARD": "1"} + if env_extra: + env.update(env_extra) + cpuinfo = os.path.join(base_dir, "cpuinfo") + with open(cpuinfo, "w") as fh: + fh.write("flags : fpu vmx sse\n") + ex = CloudHypervisorExecutor( + binary=_write_file(os.path.join(base_dir, + "cloud-hypervisor")), + ch_remote=_write_file(os.path.join(base_dir, "ch-remote")), + kernel=_write_file(os.path.join(base_dir, "vmlinux")), + rootfs=_write_file(os.path.join(base_dir, "disk.raw"), + b"R" * 100), + ssh_key=_write_file(os.path.join(base_dir, "id_chv")), + kvm_path=_write_file(os.path.join(base_dir, "kvm")), + cpuinfo_path=cpuinfo, + state_dir=state_dir, + sleeper=lambda s: None, env=env) + ex.audit_callback = lambda e, t, d, k=None: None + return ex + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.state_dir = os.path.join(self.tmp.name, "state") + self.calls = [] + + def tearDown(self): + self.tmp.cleanup() + + def _stubbed_exec(self, env_extra=None, wake_fail_resident=False): + ex = self.make_executor(self.state_dir, self.tmp.name, env_extra) + calls = self.calls + + def wake(task, extra_env=None): + calls.append(("wake", ex.vm_name(task))) + if wake_fail_resident and ex.vm_name(task).startswith( + "chv-resident-"): + raise RuntimeError("provision boom") + return {"vm": ex.vm_name(task), "mode": "cold"} + + def pause(task_or_name): + calls.append(("pause", ex.vm_name(task_or_name))) + return True + + def resume(task_or_name): + calls.append(("resume", ex.vm_name(task_or_name))) + return True + + def wait(task): + calls.append(("wait", ex.vm_name(task))) + return True + + def shutdown(task_or_name): + calls.append(("shutdown", ex.vm_name(task_or_name))) + return True + + def run_guest(task, inner_args, extra_env=None, **kwargs): + calls.append(("guest", task["id"], list(inner_args))) + return {"task_id": task["id"], "status": "completed", + "stdout": "ok", "vm": ex.vm_name(task)} + + def run_guest_goal(task, inner_args, extra_env=None, **kwargs): + calls.append(("guest-goal", task["id"], list(inner_args))) + return {"task_id": task["id"], "status": "completed", + "goal_met": True, "vm": ex.vm_name(task)} + + ex.wake_vm = wake + ex.pause_vm = pause + ex.resume_vm = resume + ex.wait_guest_ready = wait + ex.shutdown_vm = shutdown + ex._run_guest = run_guest + ex._run_guest_goal = run_guest_goal + return ex + + +class RunWithVmResidentTest(_RealExecCase): + def test_first_task_provisions_and_second_reuses(self): + ex = self._stubbed_exec() + task = {"id": "task-1", "prompt": "hi"} + result = ex._run_with_vm(task, ["dispatch", "--task", "hi"]) + self.assertEqual(result["status"], "completed") + # 常驻伪任务跑命令，但 task_id 改回真实任务。 + self.assertEqual(result["task_id"], "task-1") + self.assertEqual(result["vm"], "chv-resident-default") + kinds = [c[0] for c in self.calls] + self.assertEqual(kinds, ["wake", "pause", "resume", "wait", + "guest", "pause"]) + self.assertEqual(self.calls[0][1], "chv-resident-default") + self.assertEqual(self.calls[4][1], "resident-default") + # 第二个任务复用同一台常驻 VM，不再 wake。 + self.calls.clear() + task2 = {"id": "task-2", "prompt": "hi2"} + result2 = ex._run_with_vm(task2, ["dispatch", "--task", "hi2"]) + self.assertEqual(result2["task_id"], "task-2") + self.assertEqual(result2["vm"], "chv-resident-default") + self.assertEqual([c[0] for c in self.calls], + ["resume", "wait", "guest", "pause"]) + # 落盘条目是 paused，可供下次 acquire。 + with open(_registry_path(self.state_dir)) as fh: + data = json.load(fh) + self.assertEqual(data["default"]["state"], "paused") + self.assertEqual(data["default"]["vm"], "chv-resident-default") + + def test_goal_path_runs_on_resident_and_rewrites_task_id(self): + ex = self._stubbed_exec() + task = {"id": "goal-1", "prompt": "g"} + result = ex._run_with_vm(task, ["goal", "--task", "g"], + goal=True) + self.assertEqual(result["status"], "completed") + self.assertEqual(result["task_id"], "goal-1") + self.assertTrue(result["goal_met"]) + self.assertIn(("guest-goal", "resident-default", + ["goal", "--task", "g"]), self.calls) + + def test_dispatch_task_flows_through_resident(self): + ex = self._stubbed_exec() + result = ex.dispatch_task({"id": "d-1", "prompt": "hello"}) + self.assertEqual(result["status"], "completed") + self.assertEqual(result["task_id"], "d-1") + self.assertEqual(result["vm"], "chv-resident-default") + + def test_per_user_vms_are_isolated(self): + ex = self._stubbed_exec() + ex._run_with_vm({"id": "t-a", "user_id": "alice", + "prompt": "x"}, ["dispatch"]) + ex._run_with_vm({"id": "t-b", "user_id": "bob", + "prompt": "x"}, ["dispatch"]) + wakes = [c[1] for c in self.calls if c[0] == "wake"] + self.assertEqual(wakes, ["chv-resident-alice", + "chv-resident-bob"]) + with open(_registry_path(self.state_dir)) as fh: + data = json.load(fh) + self.assertEqual(set(data), {"alice", "bob"}) + + def test_disabled_flag_goes_cold(self): + ex = self._stubbed_exec( + env_extra={"CHV_RESIDENT_ENABLED": "0"}) + result = ex._run_with_vm({"id": "task-1", "prompt": "hi"}, + ["dispatch"]) + self.assertEqual(result["status"], "completed") + self.assertEqual(result["task_id"], "task-1") + kinds = [c[0] for c in self.calls] + self.assertEqual(kinds, ["wake", "guest", "shutdown"]) + self.assertEqual(self.calls[0][1], "chv-task-1") + self.assertIsNone(getattr(ex, "_resident_pool", None)) + self.assertFalse(os.path.exists( + _registry_path(self.state_dir))) + + def test_secret_env_goes_cold(self): + ex = self._stubbed_exec() + result = ex._run_with_vm( + {"id": "task-1", "prompt": "hi"}, ["dispatch"], + extra_env={"SUB2API_API_KEY": "sekret"}) + self.assertEqual(result["status"], "completed") + kinds = [c[0] for c in self.calls] + self.assertEqual(kinds, ["wake", "guest", "shutdown"]) + self.assertFalse(os.path.exists( + _registry_path(self.state_dir))) + + def test_provision_failure_falls_back_to_cold(self): + ex = self._stubbed_exec(wake_fail_resident=True) + result = ex._run_with_vm({"id": "task-1", "prompt": "hi"}, + ["dispatch"]) + self.assertEqual(result["status"], "completed") + self.assertEqual(result["task_id"], "task-1") + kinds = [c[0] for c in self.calls] + # 建池 wake 失败（尽力 shutdown）→ 冷启动照常跑任务。 + self.assertEqual(kinds, ["wake", "shutdown", "wake", + "guest", "shutdown"]) + self.assertEqual(self.calls[1][1], "chv-resident-default") + self.assertEqual(self.calls[2][1], "chv-task-1") + + def test_guest_error_still_pauses_resident_in_finally(self): + ex = self._stubbed_exec() + + def boom(task, inner_args, extra_env=None, **kwargs): + raise RuntimeError("guest boom") + + ex._run_guest = boom + with self.assertRaises(RuntimeError): + ex._run_with_vm({"id": "task-1", "prompt": "hi"}, + ["dispatch"]) + kinds = [c[0] for c in self.calls] + self.assertEqual(kinds, ["wake", "pause", "resume", "wait", + "pause"]) + self.assertEqual(self.calls[-1][1], "chv-resident-default") + + def test_guest_error_still_shuts_down_cold_in_finally(self): + ex = self._stubbed_exec( + env_extra={"CHV_RESIDENT_ENABLED": "0"}) + + def boom(task, inner_args, extra_env=None, **kwargs): + raise RuntimeError("guest boom") + + ex._run_guest = boom + with self.assertRaises(RuntimeError): + ex._run_with_vm({"id": "task-1", "prompt": "hi"}, + ["dispatch"]) + self.assertEqual([c[0] for c in self.calls], + ["wake", "shutdown"]) + + def test_wake_failure_propagates_without_shutdown(self): + ex = self._stubbed_exec( + env_extra={"CHV_RESIDENT_ENABLED": "0"}) + + def wake_fail(task, extra_env=None): + raise ChvFailure("vm_create_failed", "no vm") + + ex.wake_vm = wake_fail + with self.assertRaises(ChvFailure): + ex._run_with_vm({"id": "task-1", "prompt": "hi"}, + ["dispatch"]) + self.assertEqual(self.calls, []) + + +class WarmPoolUntouchedTest(unittest.TestCase): + def test_wired_to_dispatch_still_false(self): + self.assertFalse(WarmPool.wired_to_dispatch) + + def test_executor_source_never_imports_experimental(self): + import ast + from pathlib import Path + tree = ast.parse((Path(__file__).resolve().parents[2] + / "cloud" / "chv" / "executor.py").read_text()) + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + self.assertNotIn(alias.name, ( + "cloud.chv.pool", + "cloud.chv.experimental_pool", + "cloud.chv.orphans")) + if isinstance(node, ast.ImportFrom): + self.assertNotIn(node.module, ( + "cloud.chv.pool", + "cloud.chv.experimental_pool", + "cloud.chv.orphans")) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_cloud_hypervisor.py b/cloud/tests/test_cloud_hypervisor.py index 65efe9b..12389cf 100644 --- a/cloud/tests/test_cloud_hypervisor.py +++ b/cloud/tests/test_cloud_hypervisor.py @@ -227,7 +227,15 @@ class PreflightTest(ChvCase): self.assertIn("内核", text) def test_dispatch_missing_dependency_does_not_call_runner(self): - ex = self.make_exec(binary="/tmp/no-chv") + # P1 常驻默认开：这里显式关常驻，测的仍是冷启动 + # preflight 路径（缺依赖 → 零 runner 调用）。不开 + # 关的话，_resident_acquire 会先试建池，wake 失败 + # 后的清理 shutdown 会多一次 ch-remote 调用——那是 + # 建池失败清理的既定行为（见 P1 的 provision 失败 + # 回落测试），不是本测试要守的“缺依赖零调用”。 + ex = self.make_exec( + binary="/tmp/no-chv", + env_extra={"CHV_RESIDENT_ENABLED": "0"}) with self.assertRaises(ChvFailure): ex.dispatch_task(self._task()) self.assertEqual(self.calls, []) @@ -538,7 +546,12 @@ class WarmPoolTest(ChvCase): # -- dispatch / goal（wake → 跑 → finally 关） --------------------------- class DispatchTest(ChvCase): def test_dispatch_wakes_runs_shuts_down(self): - ex = self.make_exec() + # 本测试钉的是冷启动 dispatch 序列（wake chv-abc123 + # → 跑 → finally shutdown）；P1 起常驻默认开，须显 + # 式关常驻才走这条路。常驻 dispatch 由 + # test_chv_muse_p1_resident 覆盖。 + ex = self.make_exec( + env_extra={"CHV_RESIDENT_ENABLED": "0"}) out = ex.dispatch_task(self._task()) self.assertEqual(out["executor"], "cloud-hypervisor") self.assertEqual(out["status"], "completed") diff --git a/docs/chv-architecture.md b/docs/chv-architecture.md index ed06307..526da16 100644 --- a/docs/chv-architecture.md +++ b/docs/chv-architecture.md @@ -57,6 +57,60 @@ - 重启条件：将来要重新接线，必须先实现任务盘重供给并补 “任务盘来源”不变量测试，否则不得把 wired 改成 True。 +## ADR-17 常驻 VM 池取代 WarmPool 接线（feat-chv-muse-p1） + +- 决策：不再接线 WarmPool，改走新建的生产模块 + `cloud/chv/resident_pool.py`（ResidentPool）：每用户一台 + 常驻 VM，任务到时 resume 已暂停的 VM，跑完 pause，不 + 销毁；池空或任何一步失败都回落冷启动。WarmPool 实验模块 + 一字不动，`wired_to_dispatch` 仍恒 False，ADR-5 的定案 + （预热盘是干净快照盘、不是任务盘）仍然成立、只是不再 + 是接线障碍——常驻 VM 的盘本来就是每用户持久工作区盘， + 不是每任务新盘。 +- 隔离单元变了：常驻路径下隔离单元从“每任务”变成“每用户” + ——同一用户的任务共享同一台常驻 VM 与同一块工作区盘， + 每任务新盘的不变量在常驻路径上不再成立。冷启动路径 + （每任务新盘）保留作回落与隔离严格场景。当前是单租户， + 这个取舍可接受；多租户重审前不得把常驻路径当多租户隔离。 + 任务 dict 目前无 user_id，全落到 "default" 键；user_id + 出现时按键隔离，release 绝不把用过的 VM 放回给另一用户。 +- 接线：`_run_with_vm` 先常驻 acquire，命中走常驻路径 + （finally 里 release=pause），失败/未启用走原冷启动路径 + （finally shutdown 不变，冷启动本体逐字移到 `_run_cold`）； + dispatch/run/goal/goal_resume/断点续跑都走 `_run_with_vm`， + 签名不动。常驻默认开，`CHV_RESIDENT_ENABLED=0` 可关； + 开关用 getattr/env 惰性读，`__init__` 不动。池空时惰性 + provision 一次（wake + pause）再取，仍无则冷启动——首个 + 任务自举常驻 VM，不依赖 worker 启动时预热。 +- 常驻执行细节：`_run_guest`/`_run_guest_goal` 一律走常驻伪 + 任务 `{"id": "resident-<用户>"}` 传参（SSH known_hosts、 + api-socket、密钥投递记录、VMM 登记都按常驻 VM 名算）， + 返回后把 `task_id` 改回真实任务 id；中途死亡判据同样按 + 常驻伪任务查登记进程。带密钥 extra_env 的任务不走常驻 + （seed 是建池时定的，密钥只能冷启动随 seed 投递），直接 + 冷启动。同一用户同一时间只跑一个常驻任务，并发第二个 + 任务回落冷启动。 +- Registry：`state_dir/resident-pool.json`，文件 IO 与损坏 + 语义复用 orphans（原子写、损坏 quarantine、单条目非法 + 只丢该条）；重启后读到的条目一律 orphan 不复用、只清理。 + 空闲回收：paused 超 `CHV_RESIDENT_IDLE_MINUTES`（默认 0= + 不自动关机）时 hibernate（pause + snapshot + shutdown） + 并删条目，快照文件留作恢复物，restore 是后续工作。 +- 依赖方向（ADR-12 的禁令全部保留）：executor 仍绝不 import + orphans/experimental_pool（常驻池用方法内 import 拿 + ResidentPool）；resident_pool → orphans（只借 registry + 文件 IO，与 experimental_pool 同款方向）；worker 启动/ + 关闭调 `worker_adapter.reconcile_resident_pool` facade， + 它只读 resident-pool.json 做清理，不构造 ResidentPool。 + `reconcile_warm_pool` 不动。 +- 诚实边界（未验证，不宣称）：resume 耗时、到 SSH 5 秒内、 + 气球压内存——无真机验证（第三棒占着 WSL 测试机），一律 + 记入真机待验证清单；resume 失败时 `wait_guest_ready` 的 + 失败路径会 stop_vm（含删工作区盘），该路径的工作区以 + Gitea 同步为准；pause 失败时条目留在 running（不删不销 + 毁），worker 关闭时 reconcile 清理，在此之前该用户只走 + 冷启动。 + ## ADR-6 Docker goal 解析分叉：共享正典 + 同步守卫（本轨定案） - 决策：Local/CHV 共用 parse_goal_result 正典；Docker goal_task diff --git a/docs/chv-muse-p1-gap.md b/docs/chv-muse-p1-gap.md new file mode 100644 index 0000000..3820292 --- /dev/null +++ b/docs/chv-muse-p1-gap.md @@ -0,0 +1,90 @@ +# 对齐 Muse P1 差距表：常驻 VM + WarmPool + 气球 + vsock + +依据：`~/workspace/research_notes/meta-muse-ai-architecture-20261101-0755/report.md` +（官方以 Meta security.muse.ai 2026-09-08 为准；第三方拆解只作参考， +逆向细节不得写成官方架构）。Muse 模型一句话：每用户一台常驻 Secure VM， +闲置靠气球压内存，宿主–客机走 vsock 通知通道。 + +本轨范围只限 P1：常驻 VM、WarmPool 扶正、内存气球、vsock 通道。 +不碰 P2（客机 nspawn 单元、浏览器子代理独立）、不碰 P3（authd 替身凭证、 +Sentinel 式出口审批）。分支 `feat-chv-muse-p1`（从 `feat-chv-cover100` +分出，基线 HEAD `06c3e17`）。 + +## 1. 现状 vs Muse + +| 项 | 现状（实测/代码为准） | Muse 目标 | 本轨做法 | +|---|---|---|---| +| VM 生命周期 | 1 任务 = 1 VM，用完销毁；冷启动到 SSH 实测 11 秒 | 每用户一台常驻 Secure VM（hatchling），任务来了直接用 | 新增常驻服务：每用户一台常驻 VM（暂停态），任务到时 resume；池空回落冷启动（11 秒路径不动） | +| 派发查池 | `_run_with_vm` 绝不查池，`warm_pool_wired_to_dispatch` 恒 False；worker 派发前查池未接线 | 常驻 VM 就是派发目标，无冷启动 | `_run_with_vm` 接常驻路径：先 acquire 常驻 VM（resume + 等 SSH 就绪），无可用时回落 `wake_vm` | +| 预热池状态 | `experimental_pool.WarmPool` 纯实验，生产禁 import（ADR-12）；ADR-5 定案：预热盘是干净快照盘、不是任务盘，接线须先实现任务盘重供给 | —（Muse 无此概念：常驻 VM 的盘就是用户工作区盘） | WarmPool 实验模块不动（仍禁生产 import）；新建生产模块 `resident_pool.py`。常驻 VM 的盘是**每用户持久工作区盘**，不是每任务新盘 | +| 内存 | 固定 2G，无气球 | 气球 + 空闲页上报，闲置 8G 压到约 2G 工作集 | CHV 配置加 virtio-balloon + free-page-reporting（默认开，可关） | +| 宿主–客机通道 | 只 SSH；executor 文档明确“只选 SSH，不做 vsock 第二套”（旧决策） | vsock 通知通道（另有 Unix socket + SO_PEERCRED 的单元内 IPC，属 P2/P3 层） | vsock 只做最小 notify（就绪/心跳上报），与 SSH 并存；SSH 仍是命令通道与就绪判定，host key 钉死不动 | +| 池持久化/孤儿 | WarmPool registry `state_dir/warm-pool.json` 已有（原子写、损坏 quarantine、重启后条目一律 orphan 不复用） | — | 常驻池复用同款 registry 语义，另建 `resident-pool.json`；worker 启动/关闭走 `orphans` facade 清理，不自动复用上一世条目 | + +## 2. 改了什么（本轨交付物） + +- `cloud/chv/resident_pool.py`（新）：常驻池。状态机 cold → provisioning → + paused → running → paused；`acquire(user)` resume + SSH 就绪， + `release(user)` pause 不销毁；空闲回收策略（pause 后常驻，超闲置阈值 + 关机+快照，阈值可配，默认不开自动关机）；registry 原子落盘、损坏 + quarantine、重启后不复用。 +- `cloud/chv/executor.py`：`_run_with_vm` 接常驻路径（默认开，可配关； + 常驻失败一律回落冷启动，绝不因池坏而丢任务）；`_create` payload 加 + balloon/vsock 设备。 +- `cloud/chv/config.py`：`CHV_RESIDENT_IDLE_MINUTES`、`CHV_BALLOON`、 + `CHV_VSOCK`（开关 + CID），非法值 ValueError，不静默 fallback。 +- `cloud/chv/worker_adapter.py`：常驻池 reconcile facade（接 `orphans`）。 +- `cloud/chv/vsock.py`（新，宿主侧）：vsock notify 客户端/监听最小实现。 +- `cloud/chv/guest/vsock_notify.py`（新，客内脚本）：连宿主 CID 2 发 + ready/心跳。**只写脚本文件，不改 `chv-guest/build.sh`**——镜像集成 + 是 P2 的范围，本轨只留接入说明。 +- `cloud/tests/test_chv_muse_p1_*.py`（新）：常驻池、气球/vsock 配置与 + payload、回落语义的单测；全量 cloud 套件全过。 +- `docs/chv-architecture.md`：ADR-5/ADR-12 补充新决策（常驻路径取代 + WarmPool 接线问题），旧定案不删、只追加。 + +## 3. 诚实边界（本轮不宣称的东西） + +1. **隔离单元变了**：常驻路径下隔离单元从“每任务”变成“每用户”——同一 + 用户的任务共享同一台常驻 VM 与同一块工作区盘，每任务新盘的不变量在 + 常驻路径上不再成立。冷启动路径（每任务新盘）保留作回落与隔离严格场景。 + 当前是单租户，这个取舍可接受；多租户重审前不得把常驻路径当多租户隔离。 +2. **性能数字全是待验证目标，不是结果**：resume 1–2 秒、到 SSH 5 秒内、 + 气球压内存——本轮无真机验证（第三棒 V-1～V-9 占着 WSL 测试机），一律 + 记入“真机待验证清单”，不在代码/文档里写成已测得。 +3. **气球 API 形态须核对**：CHV 54 的 balloon/vsock payload 字段以 + Cloud Hypervisor 官方文档/二进制为准；本机无 /dev/kvm，无法本地起 + VM 校验，只做配置层 + payload 单测。 +4. **vsock 端到端未通**：本轮只做到宿主侧 + 客内脚本；客内脚本进镜像、 + 真机收发验证都待 P2 集成后、测试机空闲时再做。SSH 不动，vsock 坏了 + 不影响任务。 +5. **WarmPool 实验模块不动**：不把它翻成生产模块，不改 + `wired_to_dispatch`；常驻是新模块、新路径，旧实验代码与测试原样保留。 + +## 4. 恢复收尾（分离后在本分支修掉的回归） + +分离时 chv 子集 3 失败 + 1 错误，全量另有 2 失败，均已修 +掉、未删测试、未改断言： + +- **VmDeathTest 两处**：根因不是生产判定错——探针实证， + 常驻 VM 真死时 `_run_on_resident` 两态都正确判 + interrupted；错的是旧测试的死亡模拟：它杀 + `ex._vms["chv-abc123"]`，P1 常驻路由下该 VM 从未 wake、 + 登记里只有 `chv-resident-default`，模拟是空操作。修 + 法：模拟器改杀 dispatch 时已登记的 VM（即实际跑命令 + 的那台），断言不动。 +- **test_cloud_hypervisor 两处冷路径测试** + （`test_dispatch_wakes_runs_shuts_down`、 + `test_dispatch_missing_dependency_does_not_call_runner`）： + P1 把常驻改成默认开后，它们测的冷启动序列不再是默认 + 路径；修法：两测试显式 `CHV_RESIDENT_ENABLED=0`， + 继续钉冷启动行为，断言不动。常驻 dispatch 由 P1 新测 + 试覆盖。附带说明：建池 wake 失败后的清理 shutdown 是 + 既定行为（P1 回落测试钉住），不是回归。 +- **超 50 行**：`_balloon_vsock_payload`（54）拆出 + `_resolve_vsock_socket`；`vsock.listen_notify`（55） + 拆出 `_bind_notify_listener`；行为不动。 +- **mypy**：`resident_pool.py:208` 加 None 守卫后干净； + `executor.py` 的 `out` 未注解错与 `marker_tools.py:142` + 是基线 06c3e17 已有（`git show` 核对，P1 diff 零行碰 + `out`），如实记录、未动。