From 9d77ae55a745cb157a636f523ce007b53c916e69 Mon Sep 17 00:00:00 2001 From: host Date: Sat, 7 Nov 2026 10:40:12 +0000 Subject: [PATCH] =?UTF-8?q?feat(chv):=20=E9=98=B6=E6=AE=B5=202=20=E5=AE=A2?= =?UTF-8?q?=E6=9C=BA=E8=BF=90=E8=A1=8C=E6=97=B6=E4=B8=89=E5=88=86=20daemon?= =?UTF-8?q?/execd/syncd=20+=20restore=20=E7=95=99=20seed.iso=20+=20tap=20?= =?UTF-8?q?=E9=80=80=E9=81=BF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 客机三分（chv-guest/chv-daemon.py、chv-execd.py、chv-syncd.py， 仅标准库；systemd Restart=always；build.sh 拷入 + 单元 + enable； buildinfo/check/generate lockfile 登记）： daemon=心跳/资源上报/快照配合（prepare-snapshot 问 execd， 有运行中任务或联系不上即拒绝）；execd=shell/文件读写/codex 托管（chv-cell 起 host.py，setsid 脱离，任务登记落盘， 重启按 pid 认领；SO_PEERCRED 只认 uid 0/1000；codex env 白名单）；syncd=常驻循环只调 chv-sync pull/push，规则一条 不改（先 pull --ff-only 再 push，分叉即停、不 --force、 不 rebase；仓不存在记 needs-boot 不擅自 clone）。 host.py 仍是唯一任务/codex 引擎，broker/browser/cell 分工 不变，不另起第二套 agent。 - 阶段 1 遗留修复：restore_vm 失败路径 stop_vm(keep_disk=True)， seed.iso 与任务盘都留，重试有输入；vm.boot/restore 的 tap 触点遇 Resource busy 等 tap 空闲后重试一次（_wait_tap_idle 看 ip link state，_ch_remote 报错带 stderr 尾巴以便识别）； 诊断不用 pgrep -x（15 字符上限）。 - 单测 cloud/tests/test_chv_opt_stage2.py 25 项；P2 干跑夹具 与 enable 行断言随构建接线更新。 - 全量 cloud/tests 2001 全过（skipped=8）。 --- chv-guest/build.sh | 17 +- chv-guest/buildinfo.py | 2 + chv-guest/check_lockfile.py | 5 + chv-guest/chv-daemon.py | 315 +++++++++++++ chv-guest/chv-execd.py | 485 ++++++++++++++++++++ chv-guest/chv-syncd.py | 273 ++++++++++++ chv-guest/generate_lockfile.py | 5 + cloud/chv/executor.py | 145 +++++- cloud/tests/test_chv_muse_p2.py | 8 +- cloud/tests/test_chv_opt_stage2.py | 691 +++++++++++++++++++++++++++++ 10 files changed, 1932 insertions(+), 14 deletions(-) create mode 100644 chv-guest/chv-daemon.py create mode 100644 chv-guest/chv-execd.py create mode 100644 chv-guest/chv-syncd.py create mode 100644 cloud/tests/test_chv_opt_stage2.py diff --git a/chv-guest/build.sh b/chv-guest/build.sh index 6dbc7ef..ccf9398 100755 --- a/chv-guest/build.sh +++ b/chv-guest/build.sh @@ -186,6 +186,14 @@ for p2_script in chv-cell.py chv-broker.py chv-browser.py; do exit 1 fi done +# feat-chv-opt-landing（阶段 2）：客机运行时三分——daemon / +# execd / syncd 三脚本缺一即失败（与 chv-sync.sh 同款门）。 +for stage2_script in chv-daemon.py chv-execd.py chv-syncd.py; do + if [ ! -f "$SCRIPT_DIR/$stage2_script" ]; then + echo "缺少 $stage2_script（阶段 2 客机服务）：$SCRIPT_DIR" >&2 + exit 1 + fi +done if [ "$DRY_RUN" = "0" ]; then if [ ! -e /dev/kvm ]; then echo "CHV_GUEST_BUILD needs a KVM host: /dev/kvm 不存在" >&2 @@ -294,6 +302,10 @@ VC_ARGS=(-a "$OUT/disk.raw" --copy-in "$SCRIPT_DIR/chv-broker.py:/usr/local/bin/" --copy-in "$SCRIPT_DIR/chv-browser.py:/usr/local/bin/" --run-command 'mv /usr/local/bin/chv-cell.py /usr/local/bin/chv-cell && mv /usr/local/bin/chv-broker.py /usr/local/bin/chv-broker && mv /usr/local/bin/chv-browser.py /usr/local/bin/chv-browser && chmod +x /usr/local/bin/chv-cell /usr/local/bin/chv-broker /usr/local/bin/chv-browser' + --copy-in "$SCRIPT_DIR/chv-daemon.py:/usr/local/bin/" + --copy-in "$SCRIPT_DIR/chv-execd.py:/usr/local/bin/" + --copy-in "$SCRIPT_DIR/chv-syncd.py:/usr/local/bin/" + --run-command 'mv /usr/local/bin/chv-daemon.py /usr/local/bin/chv-daemon && mv /usr/local/bin/chv-execd.py /usr/local/bin/chv-execd && mv /usr/local/bin/chv-syncd.py /usr/local/bin/chv-syncd && chmod +x /usr/local/bin/chv-daemon /usr/local/bin/chv-execd /usr/local/bin/chv-syncd' --run-command 'printf "runcmd:\n - systemctl restart ssh || service ssh restart || true\n - test -f /etc/chv/guest-env && chown chv:chv /etc/chv/guest-env || true\n - /usr/local/bin/chv-data-mount >>/var/log/chv-data-mount.log 2>&1 || true\n - /usr/local/bin/chv-sync boot >>/var/log/chv-sync.log 2>&1 || true\n" > /etc/cloud/cloud.cfg.d/99-chv-sync.cfg' --run-command 'git config --system user.name chv && git config --system user.email chv@localhost && git config --system credential.helper store && git config --system init.defaultBranch main' --run-command "$GT_INSTALL_CMD" @@ -304,7 +316,10 @@ VC_ARGS=(-a "$OUT/disk.raw" --run-command 'useradd --system --home-dir /var/lib/chv-browser --shell /usr/sbin/nologin chv-browser && mkdir -p /var/lib/chv-browser /run/chv/shots && chown chv-browser:chv-browser /var/lib/chv-browser /run/chv/shots && chmod 0755 /run/chv /run/chv/shots' --run-command 'printf "[Unit]\nDescription=CHV cell broker (Muse P2)\nAfter=network.target\n\n[Service]\nType=simple\nExecStart=/usr/bin/python3 /usr/local/bin/chv-broker\nRestart=on-failure\nRestartSec=2\n\n[Install]\nWantedBy=multi-user.target\n" > /etc/systemd/system/chv-broker.service' --run-command 'printf "[Unit]\nDescription=CHV browser sub-agent (Muse P2)\nAfter=network.target\n\n[Service]\nType=simple\nUser=chv-browser\nGroup=chv-browser\nExecStart=/usr/bin/python3 /usr/local/bin/chv-browser\nRestart=on-failure\nRestartSec=2\n\n[Install]\nWantedBy=multi-user.target\n" > /etc/systemd/system/chv-browser.service' - --run-command 'printf "d /run/chv 0755 root root -\nd /run/chv/shots 0755 chv-browser chv-browser -\n" > /etc/tmpfiles.d/chv.conf && systemctl enable chv-broker chv-browser || true') + --run-command 'printf "[Unit]\nDescription=CHV guest runtime daemon (stage 2: heartbeat/resources/snapshot)\nAfter=network.target\n\n[Service]\nType=simple\nExecStart=/usr/bin/python3 /usr/local/bin/chv-daemon\nRestart=always\nRestartSec=2\n\n[Install]\nWantedBy=multi-user.target\n" > /etc/systemd/system/chv-daemon.service' + --run-command 'printf "[Unit]\nDescription=CHV guest exec daemon (stage 2: shell/files/codex hosting)\nAfter=network.target\n\n[Service]\nType=simple\nExecStart=/usr/bin/python3 /usr/local/bin/chv-execd\nRestart=always\nRestartSec=2\n\n[Install]\nWantedBy=multi-user.target\n" > /etc/systemd/system/chv-execd.service' + --run-command 'printf "[Unit]\nDescription=CHV Gitea sync daemon (stage 2: resident pull/push loop)\nAfter=network.target\n\n[Service]\nType=simple\nUser=chv\nGroup=chv\nExecStart=/usr/bin/python3 /usr/local/bin/chv-syncd\nRestart=always\nRestartSec=2\n\n[Install]\nWantedBy=multi-user.target\n" > /etc/systemd/system/chv-syncd.service' + --run-command 'printf "d /run/chv 0755 root root -\nd /run/chv/shots 0755 chv-browser chv-browser -\nd /run/chv/tasks 0755 root root -\n" > /etc/tmpfiles.d/chv.conf && systemctl enable chv-broker chv-browser chv-daemon chv-execd chv-syncd || true') # P1 交接（Muse P1 的 vsock_notify.py）：文件在仓库里存在时拷入 # 镜像（P1 分支才有，本分支尚无）；不存在时只警告、不失败—— # 合并后自动生效，P2 不改该脚本内容。 diff --git a/chv-guest/buildinfo.py b/chv-guest/buildinfo.py index c56afa0..e2b7fb9 100644 --- a/chv-guest/buildinfo.py +++ b/chv-guest/buildinfo.py @@ -41,6 +41,8 @@ BASE_CONTENTS = [ "systemd-container", "chv-cell", "chv-broker", "chv-browser", # feat-chv-opt-landing（阶段 1）：数据盘挂载脚本。 "chv-data-mount", + # feat-chv-opt-landing（阶段 2）：客机运行时三分服务。 + "chv-daemon", "chv-execd", "chv-syncd", ] # 执行器默认内核 cmdline（config.py），构建产物须与它一起记录（B-12）， diff --git a/chv-guest/check_lockfile.py b/chv-guest/check_lockfile.py index 297cbaf..8ec8f0d 100644 --- a/chv-guest/check_lockfile.py +++ b/chv-guest/check_lockfile.py @@ -33,6 +33,11 @@ DECLARED_SOURCES = { # feat-chv-opt-landing（阶段 1）：数据盘挂载脚本，版本随 # repo_commit（与 chv-sync 同款）。 "chv-data-mount": "repo_commit", + # feat-chv-opt-landing（阶段 2）：客机运行时三分服务，版本 + # 随 repo_commit（与 chv-sync 同款）。 + "chv-daemon": "repo_commit", + "chv-execd": "repo_commit", + "chv-syncd": "repo_commit", } diff --git a/chv-guest/chv-daemon.py b/chv-guest/chv-daemon.py new file mode 100644 index 0000000..9435515 --- /dev/null +++ b/chv-guest/chv-daemon.py @@ -0,0 +1,315 @@ +#!/usr/bin/env python3 +"""chv-daemon：客机运行时 daemon（阶段 2 三分之一）。 + +职责（只做这三件，不碰任务执行与同步）： + 1. 生命周期/心跳：每 10 秒写 /run/chv/heartbeat.json（atomic）， + 含 pid、启动时间、seq、uptime、loadavg、内存、workspace + 余量、execd 任务数、sync 最近状态摘要。seq 跨重启连续 + （启动时读旧文件续号），宿主侧据此判断“心跳丢了还是 + daemon 重启了”。 + 2. 资源上报：同上，心跳文件即上报载体（宿主经 SSH 读文件， + 不另开通道；控制通道仍是 SSH，见 executor.py 定稿）。 + 3. 快照配合：Unix socket /run/chv/daemon.sock，op + prepare-snapshot——问 execd 是否有运行中任务；有则 + 拒绝（{ok:false, reason:task-running}），无则写 + /run/chv/snapshot-ready.json 标记并返回 ok；execd + 联系不上时同样拒绝（fail-closed，不写标记）。 + 宿主侧快照流程暂未调它（阶段 2 不改 executor 快照流）， + 接口先行、供宿主侧后续接线与本阶段 live 演练。 + +与既有分工的映射（不要第二套 agent）： + - host.py 仍是唯一的任务/ codex 执行引擎（CLI，经 SSH 或 + execd 的 codex op 调）；daemon 从不跑 host.py。 + - chv-broker/chv-browser 仍是单元内外 IPC 与浏览器子代理， + daemon 不碰它们的 socket。 + - sync 仍是 chv-sync.sh + chv-syncd 常驻循环，daemon 只读 + sync-status.json 作摘要，不调 git。 + +只用标准库。以 root 运行（读 /proc 与各服务状态文件无权限 +问题；socket 0600 root:root，只接受本机 root 客户）。 +""" +import json +import os +import signal +import socket +import sys +import threading +import time + +HEARTBEAT_PATH = "/run/chv/heartbeat.json" +DAEMON_SOCK = "/run/chv/daemon.sock" +SNAPSHOT_READY_PATH = "/run/chv/snapshot-ready.json" +EXECD_SOCK = "/run/chv/execd.sock" +EXECD_TASKS = "/run/chv/execd-tasks.json" +SYNC_STATUS = "/run/chv/sync-status.json" +WORKSPACE = "/home/chv/workspace" +HEARTBEAT_INTERVAL = 10.0 +MAX_REQUEST_BYTES = 65536 + + +def _atomic_write(path, data: bytes): + tmp = "%s.tmp.%d" % (path, os.getpid()) + with open(tmp, "wb") as fh: + fh.write(data) + fh.flush() + os.fsync(fh.fileno()) + os.rename(tmp, path) + + +def _read_json(path): + try: + with open(path) as fh: + return json.load(fh) + except (OSError, ValueError): + return None + + +def _loadavg(): + try: + with open("/proc/loadavg") as fh: + parts = fh.read().split() + return [float(parts[0]), float(parts[1]), float(parts[2])] + except (OSError, ValueError, IndexError): + return None + + +def _meminfo(): + mem = {} + try: + with open("/proc/meminfo") as fh: + for line in fh: + if ":" not in line: + continue + key, val = line.split(":", 1) + val = val.strip().split() + if val and val[0].isdigit(): + mem[key.strip()] = int(val[0]) + except OSError: + return None + if not mem: + return None + return {"total_kb": mem.get("MemTotal"), + "available_kb": mem.get("MemAvailable")} + + +def _workspace_free(): + try: + st = os.statvfs(WORKSPACE) + return st.f_bavail * st.f_frsize + except OSError: + return None + + +def _execd_task_count(): + data = _read_json(EXECD_TASKS) + if not isinstance(data, dict): + return None + tasks = data.get("tasks") + if not isinstance(tasks, list): + return None + running = [t for t in tasks + if isinstance(t, dict) and t.get("state") == "running"] + return {"total": len(tasks), "running": len(running)} + + +def _sync_summary(): + data = _read_json(SYNC_STATUS) + if not isinstance(data, dict): + return None + return {"last": data.get("last"), "at": data.get("at"), + "head": data.get("head"), + "error": data.get("error")} + + +def build_heartbeat(pid, started_at, seq): + """组装一次心跳 dict（纯函数，便于单测与 --once）。""" + now = time.time() + return {"service": "chv-daemon", "pid": pid, + "started_at": started_at, "seq": seq, + "at": now, "uptime_s": round(now - started_at, 1), + "loadavg": _loadavg(), "mem": _meminfo(), + "workspace_free_bytes": _workspace_free(), + "execd_tasks": _execd_task_count(), + "sync": _sync_summary()} + + +def _ask_execd_status(timeout=2.0): + """问 execd 要任务状态；联系不上返回 (None, reason)。""" + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + try: + sock.settimeout(timeout) + sock.connect(EXECD_SOCK) + sock.sendall(json.dumps({"op": "status"}).encode() + b"\n") + buf = b"" + while len(buf) < MAX_REQUEST_BYTES: + chunk = sock.recv(4096) + if not chunk: + break + buf += chunk + if buf.endswith(b"\n"): + break + data = json.loads(buf.decode()) + if not data.get("ok"): + return None, "execd-status-not-ok" + return data, None + except (OSError, ValueError) as exc: + return None, "execd-unreachable: %s" % exc + finally: + sock.close() + + +def prepare_snapshot(): + """快照配合：execd 有运行中任务或联系不上即拒绝。""" + data, reason = _ask_execd_status() + if data is None: + return {"ok": False, "reason": reason} + tasks = data.get("tasks") or [] + running = [t for t in tasks + if isinstance(t, dict) and t.get("state") == "running"] + if running: + return {"ok": False, "reason": "task-running", + "running": [t.get("id") for t in running]} + marker = {"at": time.time(), "pid": os.getpid(), + "note": "no running execd task"} + _atomic_write(SNAPSHOT_READY_PATH, + (json.dumps(marker) + "\n").encode()) + return {"ok": True, "marker": SNAPSHOT_READY_PATH} + + +def handle_request(req): + """socket 请求分发（纯函数，便于单测）。""" + if not isinstance(req, dict): + return {"ok": False, "error": "request must be an object"} + op = req.get("op") + if op == "ping": + return {"ok": True, "service": "chv-daemon", + "pid": os.getpid()} + if op == "status": + hb = _read_json(HEARTBEAT_PATH) + return {"ok": True, "heartbeat": hb} + if op == "prepare-snapshot": + try: + return prepare_snapshot() + except OSError as exc: + return {"ok": False, "reason": "marker-write-failed: %s" + % exc} + return {"ok": False, "error": "unknown op: %s" % (op,)} + + +class Daemon: + def __init__(self, interval=HEARTBEAT_INTERVAL): + self.interval = interval + self.pid = os.getpid() + self.started_at = time.time() + self.seq = 0 + old = _read_json(HEARTBEAT_PATH) + if isinstance(old, dict) and isinstance(old.get("seq"), int): + # 续号：宿主据 seq 连续判断心跳丢包还是 daemon 重启。 + self.seq = old["seq"] + self._stop = threading.Event() + self._server = None + + def write_heartbeat(self): + self.seq += 1 + hb = build_heartbeat(self.pid, self.started_at, self.seq) + _atomic_write(HEARTBEAT_PATH, + (json.dumps(hb) + "\n").encode()) + return hb + + def _serve(self): + if os.path.exists(DAEMON_SOCK): + os.remove(DAEMON_SOCK) + srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + srv.bind(DAEMON_SOCK) + os.chmod(DAEMON_SOCK, 0o600) + srv.listen(8) + srv.settimeout(0.5) + self._server = srv + while not self._stop.is_set(): + try: + conn, _ = srv.accept() + except socket.timeout: + continue + except OSError: + break + with conn: + try: + conn.settimeout(5.0) + buf = b"" + while len(buf) < MAX_REQUEST_BYTES: + chunk = conn.recv(4096) + if not chunk: + break + buf += chunk + if buf.endswith(b"\n"): + break + req = json.loads(buf.decode()) + resp = handle_request(req) + except (OSError, ValueError) as exc: + resp = {"ok": False, "error": str(exc)} + try: + conn.sendall((json.dumps(resp) + "\n").encode()) + except OSError: + pass + try: + srv.close() + except OSError: + pass + try: + os.remove(DAEMON_SOCK) + except OSError: + pass + + def run(self): + thread = threading.Thread(target=self._serve, daemon=True) + thread.start() + try: + while not self._stop.is_set(): + self.write_heartbeat() + self._stop.wait(self.interval) + finally: + self._stop.set() + thread.join(timeout=5) + + def stop(self): + self._stop.set() + + +def main(argv=None): + argv = list(sys.argv[1:] if argv is None else argv) + interval = HEARTBEAT_INTERVAL + once = False + i = 0 + while i < len(argv): + if argv[i] == "--interval" and i + 1 < len(argv): + interval = float(argv[i + 1]) + i += 2 + elif argv[i] == "--once": + once = True + i += 1 + else: + print("用法：chv-daemon [--interval 秒] [--once]", + file=sys.stderr) + return 2 + daemon = Daemon(interval=interval) + if once: + daemon.write_heartbeat() + print(json.dumps({"ok": True, + "heartbeat": HEARTBEAT_PATH, + "seq": daemon.seq})) + return 0 + print("chv-daemon pid=%d heartbeat=%s" % (daemon.pid, + HEARTBEAT_PATH), + flush=True) + + def _on_signal(signum, _frame): + daemon.stop() + + signal.signal(signal.SIGTERM, _on_signal) + signal.signal(signal.SIGINT, _on_signal) + daemon.run() + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/chv-guest/chv-execd.py b/chv-guest/chv-execd.py new file mode 100644 index 0000000..3900cb1 --- /dev/null +++ b/chv-guest/chv-execd.py @@ -0,0 +1,485 @@ +#!/usr/bin/env python3 +"""chv-execd：客机运行时 execd（阶段 2 三分之一）。 + +职责（只做这三件）： + 1. shell 命令：op exec（同步跑完返回）/ spawn（脱离式后台跑， + 立即返回 id）。一律 list argv、绝不用 shell=True；工作区 + 内命令以 chv 身份跑（runuser -u chv），codex 除外。 + 2. 文件读写：op read/write，路径只允许 /home/chv/workspace + 之内（realpath + commonpath 校验），越界即拒绝。 + 3. codex 进程托管：op codex——经 chv-cell run 起 + `python3 /app/host.py `（单元隔离口径不变，不绕 + cell 直跑 host.py）；子进程 setsid 脱离，execd 崩溃不 + 带走它；任务登记落盘 /run/chv/execd-tasks.json，execd + 重启后按 pid 重新认领（reattach），任务状态不丢。 + +客户端鉴权：SO_PEERCRED 的 uid 须是 0 或 1000（root / chv）， +其余一律拒绝。socket /run/chv/execd.sock 0600 root:root。 +调用方是宿主执行器经 SSH（root）或客机内 root 服务；单元内 +codex 不直接调 execd（走 chv-broker 白名单）。 + +任务状态机：running → completed（rc 已收）/ failed（spawn +失败）/ killed / unknown（execd 重启后 pid 已死且退出码 +不可考——init 已收走，如实记 unknown，不编 rc）。 +spawn 输出进 /run/chv/tasks/.log（runuser 以 chv 写， +codex 以 root 写，均 0644）。 + +只用标准库。以 root 运行（要起 chv-cell + runuser）；单租户 +灰度口径——与今天“执行器经 SSH 以 root 跑 host.py”同权， +多租户收紧待安全再审翻绿后另议。 +""" +import json +import os +import signal +import socket +import struct +import subprocess +import sys +import threading +import time +import uuid + +EXECD_SOCK = "/run/chv/execd.sock" +TASKS_PATH = "/run/chv/execd-tasks.json" +TASKS_LOG_DIR = "/run/chv/tasks" +WORKSPACE = "/home/chv/workspace" +HOST_PY = "/app/host.py" +CELL_BIN = "/usr/local/bin/chv-cell" +ALLOWED_UIDS = (0, 1000) +MAX_REQUEST_BYTES = 65536 +MAX_OUTPUT_BYTES = 65536 +MAX_FILE_BYTES = 1 << 20 +DEFAULT_TIMEOUT = 120.0 + +# codex 子进程可透传的环境变量白名单（前缀匹配；调用方 +# 传什么别的都丢弃。模型网关 key 走这个口径，不走 argv， +# 绝不进任务登记文件与日志）。 +CODEX_ENV_PREFIXES = ("OPENAI_", "CODEX_", "CHV_MODEL_", + "PLAYWRIGHT_") + + +def _atomic_write(path, data: bytes): + tmp = "%s.tmp.%d" % (path, os.getpid()) + with open(tmp, "wb") as fh: + fh.write(data) + fh.flush() + os.fsync(fh.fileno()) + os.rename(tmp, path) + + +def _read_json(path): + try: + with open(path) as fh: + return json.load(fh) + except (OSError, ValueError): + return None + + +def _peer_uid(conn): + """SO_PEERCRED 取对端 uid；取不到即拒绝（不降级）。""" + try: + creds = conn.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, + struct.calcsize("3i")) + _pid, uid, _gid = struct.unpack("3i", creds) + return uid + except (OSError, struct.error): + return None + + +def _within_workspace(path): + """路径是否在 WORKSPACE 内（realpath 后 commonpath 校验）。""" + try: + real = os.path.realpath(path) + except OSError: + return False + return os.path.commonpath([real, WORKSPACE]) == WORKSPACE + + +def _truncate(text, limit=MAX_OUTPUT_BYTES): + if len(text) > limit: + return text[:limit] + "\n..." % ( + len(text) - limit) + return text + + +def _codex_env(extra): + """调用方 env 过白名单；只留前缀命中的项。""" + env = {"PATH": os.environ.get("PATH", "/usr/local/bin:/usr/bin:/bin"), + "HOME": "/home/chv", + "PLAYWRIGHT_BROWSERS_PATH": + "/usr/local/share/ms-playwright"} + if isinstance(extra, dict): + for key, val in extra.items(): + if isinstance(key, str) and isinstance(val, str) \ + and key.startswith(CODEX_ENV_PREFIXES): + env[key] = val + return env + + +class TaskStore: + """任务登记（落盘 JSON，execd 重启后重读认领）。""" + + def __init__(self, path=TASKS_PATH): + self.path = path + self._lock = threading.Lock() + self.tasks = {} + data = _read_json(path) + if isinstance(data, dict) and isinstance( + data.get("tasks"), list): + for item in data["tasks"]: + if isinstance(item, dict) and item.get("id"): + self.tasks[item["id"]] = item + self._reattach() + + def _reattach(self): + """重启后认领：running 任务按 pid 查存活；已死记 + unknown（退出码不可考，不编），活着的保持 running。""" + for task in self.tasks.values(): + if task.get("state") != "running": + continue + pid = task.get("pid") + alive = False + if isinstance(pid, int): + try: + os.kill(pid, 0) + alive = True + except (OSError, ProcessLookupError): + alive = False + except PermissionError: + alive = True + if alive: + task["note"] = "reattached after execd restart" + else: + task["state"] = "unknown" + task["note"] = ("execd restarted, exit code " + "unrecoverable (reaped by init)") + self._save_locked() + + def _save_locked(self): + data = {"tasks": [self.tasks[k] for k in sorted(self.tasks)]} + _atomic_write(self.path, + (json.dumps(data) + "\n").encode()) + + def add(self, task): + with self._lock: + self.tasks[task["id"]] = task + self._save_locked() + return task + + def update(self, task_id, **fields): + with self._lock: + task = self.tasks.get(task_id) + if task is None: + return None + task.update(fields) + self._save_locked() + return task + + def get(self, task_id): + with self._lock: + task = self.tasks.get(task_id) + return dict(task) if task else None + + def list(self): + with self._lock: + return [dict(self.tasks[k]) for k in sorted(self.tasks)] + + +class Execd: + def __init__(self, tasks_path=TASKS_PATH): + self.store = TaskStore(tasks_path) + self._stop = threading.Event() + self._procs = {} + self._procs_lock = threading.Lock() + self._reaper = threading.Thread(target=self._reap_loop, + daemon=True) + + def _register_proc(self, task_id, proc): + with self._procs_lock: + self._procs[task_id] = proc + + def _reap_loop(self): + """收割已退出的子进程，登记 rc；未知 pid 的直接丢弃。 + 重启后认领的 running 任务不在 _procs 里：轮询 pid + 存活，死了记 unknown（退出码不可考，不编）。""" + while not self._stop.is_set(): + with self._procs_lock: + items = list(self._procs.items()) + tracked = set() + for task_id, proc in items: + tracked.add(task_id) + rc = proc.poll() + if rc is None: + continue + with self._procs_lock: + self._procs.pop(task_id, None) + task = self.store.get(task_id) + if task and task.get("state") == "running": + self.store.update(task_id, state="completed", + rc=rc, ended_at=time.time()) + for task in self.store.list(): + if task.get("state") != "running" \ + or task.get("id") in tracked: + continue + pid = task.get("pid") + alive = False + if isinstance(pid, int): + try: + os.kill(pid, 0) + alive = True + except OSError: + alive = False + if not alive: + self.store.update( + task["id"], state="unknown", + note=("execd restarted, exit code " + "unrecoverable (reaped by init)"), + ended_at=time.time()) + self._stop.wait(1.0) + + # -- op 实现 --------------------------------------------------------- + def op_exec(self, req): + argv = req.get("argv") + if not isinstance(argv, list) or not argv \ + or not all(isinstance(a, str) for a in argv): + return {"ok": False, "error": "argv must be a non-empty list of strings"} + cwd = req.get("cwd") or WORKSPACE + if not _within_workspace(cwd) and cwd != "/home/chv": + return {"ok": False, "error": "cwd outside workspace: %s" % cwd} + try: + timeout = float(req.get("timeout") or DEFAULT_TIMEOUT) + except (TypeError, ValueError): + return {"ok": False, "error": "bad timeout"} + timeout = min(max(timeout, 1.0), 600.0) + cmd = ["runuser", "-u", "chv", "--"] + argv + try: + proc = subprocess.run(cmd, cwd=cwd, timeout=timeout, + capture_output=True, text=True, + errors="replace") + except subprocess.TimeoutExpired as exc: + return {"ok": False, "error": "timeout after %.0fs" % timeout, + "stdout": _truncate(exc.stdout or ""), + "stderr": _truncate(exc.stderr or "")} + except OSError as exc: + return {"ok": False, "error": "spawn failed: %s" % exc} + return {"ok": True, "rc": proc.returncode, + "stdout": _truncate(proc.stdout or ""), + "stderr": _truncate(proc.stderr or "")} + + def op_spawn(self, req, kind="exec"): + argv = req.get("argv") + if not isinstance(argv, list) or not argv \ + or not all(isinstance(a, str) for a in argv): + return {"ok": False, "error": "argv must be a non-empty list of strings"} + cwd = req.get("cwd") or WORKSPACE + if not _within_workspace(cwd) and cwd != "/home/chv": + return {"ok": False, "error": "cwd outside workspace: %s" % cwd} + if kind == "codex": + if not os.path.exists(CELL_BIN): + return {"ok": False, "error": "chv-cell not found"} + full = [CELL_BIN, "run", "--", "python3", HOST_PY] + argv + env = _codex_env(req.get("env")) + as_user = None + else: + full = ["runuser", "-u", "chv", "--"] + argv + env = None + as_user = "chv" + task_id = uuid.uuid4().hex[:12] + os.makedirs(TASKS_LOG_DIR, exist_ok=True) + log_path = os.path.join(TASKS_LOG_DIR, task_id + ".log") + try: + log_fh = open(log_path, "ab", buffering=0) + except OSError as exc: + return {"ok": False, "error": "log open failed: %s" % exc} + try: + proc = subprocess.Popen( + full, cwd=cwd, stdout=log_fh, stderr=log_fh, + stdin=subprocess.DEVNULL, start_new_session=True, + env=env, close_fds=True) + except OSError as exc: + log_fh.close() + return {"ok": False, "error": "spawn failed: %s" % exc} + task = {"id": task_id, "kind": kind, + "argv_preview": argv[:4], "pid": proc.pid, + "state": "running", "started_at": time.time(), + "log": log_path, + "as_user": as_user or "root-via-cell"} + self.store.add(task) + self._register_proc(task_id, proc) + return {"ok": True, "id": task_id, "pid": proc.pid, + "log": log_path} + + def op_read(self, req): + path = req.get("path") + if not isinstance(path, str) or not path: + return {"ok": False, "error": "path required"} + if not _within_workspace(path): + return {"ok": False, "error": "path outside workspace"} + try: + with open(path, "rb") as fh: + data = fh.read(MAX_FILE_BYTES + 1) + except OSError as exc: + return {"ok": False, "error": "read failed: %s" % exc} + truncated = len(data) > MAX_FILE_BYTES + return {"ok": True, "path": path, + "content": data[:MAX_FILE_BYTES].decode( + "utf-8", errors="replace"), + "truncated": truncated} + + def op_write(self, req): + path = req.get("path") + content = req.get("content") + if not isinstance(path, str) or not path: + return {"ok": False, "error": "path required"} + if not isinstance(content, str): + return {"ok": False, "error": "content must be a string"} + if not _within_workspace(path): + return {"ok": False, "error": "path outside workspace"} + try: + parent = os.path.dirname(path) + if parent: + os.makedirs(parent, exist_ok=True) + with open(path, "w") as fh: + fh.write(content) + except OSError as exc: + return {"ok": False, "error": "write failed: %s" % exc} + return {"ok": True, "path": path, "bytes": len(content)} + + def op_task_status(self, req): + task_id = req.get("id") + if task_id: + task = self.store.get(task_id) + if task is None: + return {"ok": False, "error": "no such task: %s" % task_id} + return {"ok": True, "task": task} + return {"ok": True, "tasks": self.store.list()} + + def op_task_kill(self, req): + task_id = req.get("id") + if not task_id: + return {"ok": False, "error": "id required"} + task = self.store.get(task_id) + if task is None: + return {"ok": False, "error": "no such task: %s" % task_id} + if task.get("state") != "running": + return {"ok": False, + "error": "task not running: %s" % task.get("state")} + pid = task.get("pid") + sig = signal.SIGTERM + if req.get("signal") == "KILL": + sig = signal.SIGKILL + try: + os.kill(pid, sig) + except OSError as exc: + return {"ok": False, "error": "kill failed: %s" % exc} + self.store.update(task_id, state="killed", + ended_at=time.time()) + return {"ok": True, "id": task_id, "pid": pid} + + def handle(self, req): + if not isinstance(req, dict): + return {"ok": False, "error": "request must be an object"} + op = req.get("op") + if op == "ping": + return {"ok": True, "service": "chv-execd", + "pid": os.getpid()} + if op == "status": + return {"ok": True, "tasks": self.store.list()} + if op == "exec": + return self.op_exec(req) + if op == "spawn": + return self.op_spawn(req, kind="exec") + if op == "codex": + return self.op_spawn(req, kind="codex") + if op == "read": + return self.op_read(req) + if op == "write": + return self.op_write(req) + if op == "task-status": + return self.op_task_status(req) + if op == "task-kill": + return self.op_task_kill(req) + return {"ok": False, "error": "unknown op: %s" % (op,)} + + # -- socket 服务 ------------------------------------------------------ + def serve(self): + if os.path.exists(EXECD_SOCK): + os.remove(EXECD_SOCK) + srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + srv.bind(EXECD_SOCK) + os.chmod(EXECD_SOCK, 0o600) + srv.listen(8) + srv.settimeout(0.5) + self._reaper.start() + print("chv-execd pid=%d sock=%s" % (os.getpid(), EXECD_SOCK), + flush=True) + try: + while not self._stop.is_set(): + try: + conn, _ = srv.accept() + except socket.timeout: + continue + except OSError: + break + with conn: + uid = _peer_uid(conn) + if uid not in ALLOWED_UIDS: + resp = {"ok": False, + "error": "uid %s not allowed" % (uid,)} + try: + conn.sendall( + (json.dumps(resp) + "\n").encode()) + except OSError: + pass + continue + try: + conn.settimeout(10.0) + buf = b"" + while len(buf) < MAX_REQUEST_BYTES: + chunk = conn.recv(4096) + if not chunk: + break + buf += chunk + if buf.endswith(b"\n"): + break + req = json.loads(buf.decode()) + resp = self.handle(req) + except (OSError, ValueError) as exc: + resp = {"ok": False, "error": str(exc)} + try: + conn.sendall((json.dumps(resp) + "\n").encode()) + except OSError: + pass + finally: + self._stop.set() + try: + srv.close() + except OSError: + pass + try: + os.remove(EXECD_SOCK) + except OSError: + pass + + def stop(self): + self._stop.set() + + +def main(argv=None): + argv = list(sys.argv[1:] if argv is None else argv) + if argv: + print("用法：chv-execd（无参数）", file=sys.stderr) + return 2 + execd = Execd() + + def _on_signal(signum, _frame): + execd.stop() + + signal.signal(signal.SIGTERM, _on_signal) + signal.signal(signal.SIGINT, _on_signal) + execd.serve() + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/chv-guest/chv-syncd.py b/chv-guest/chv-syncd.py new file mode 100644 index 0000000..9555f2a --- /dev/null +++ b/chv-guest/chv-syncd.py @@ -0,0 +1,273 @@ +#!/usr/bin/env python3 +"""chv-syncd：客机运行时 sync 常驻（阶段 2 三分之一）。 + +职责：常驻循环调用 /usr/local/bin/chv-sync 做 Gitea 双向同步， +状态落盘 /run/chv/sync-status.json，socket /run/chv/syncd.sock +接受 ping/status/sync-now。 + +同步规则一条不改（与 chv-sync.sh 同款，复用它不重写 git 逻辑）： + - Gitea https://gitea.tose.top 是权威远端；启动 clone 选定仓 + 最新 HEAD（仍由 cloud-init runcmd 的 chv-sync boot 做； + syncd 只在仓已 clone 的前提下循环，仓不存在记 needs-boot + 不擅自 clone——boot 须 root 写 credential，syncd 以 chv + 身份跑，无权也不该碰）。 + - 每轮先 pull --ff-only 再 push；pull 分叉即停住、跳过 push， + 状态记 error、绝不 --force、不自动 rebase、不覆盖任何一方。 + - 工作树脏时 pull 拒绝，同样记 error 等下一轮，不丢改动。 + - push 只推已提交的提交；未提交改动不参与同步。 + - token 只走 /home/chv/.git-credentials（0600），syncd + 自身不读 guest-env、不打印、不记入状态文件与日志。 + +以 chv 身份运行。默认每 300 秒一轮（CHV_SYNC_INTERVAL， +下限 60 秒）；sync-now 立即触发一轮（轮内重复触发跳过）。 +""" +import json +import os +import signal +import socket +import subprocess +import sys +import threading +import time + +SYNC_SOCK = "/run/chv/syncd.sock" +STATUS_PATH = "/run/chv/sync-status.json" +WORKSPACE = "/home/chv/workspace" +SYNC_BIN = "/usr/local/bin/chv-sync" +DEFAULT_INTERVAL = 300.0 +MIN_INTERVAL = 60.0 +MAX_REQUEST_BYTES = 65536 + + +def _atomic_write(path, data: bytes): + tmp = "%s.tmp.%d" % (path, os.getpid()) + with open(tmp, "wb") as fh: + fh.write(data) + fh.flush() + os.fsync(fh.fileno()) + os.rename(tmp, path) + + +def _read_json(path): + try: + with open(path) as fh: + return json.load(fh) + except (OSError, ValueError): + return None + + +def resolve_repo(workspace=WORKSPACE): + """WORKSPACE 下唯一的 git 仓库；多个/零个返回 (None, reason)。""" + try: + names = sorted(os.listdir(workspace)) + except OSError as exc: + return None, "workspace unreadable: %s" % exc + found = [os.path.join(workspace, name) for name in names + if os.path.isdir(os.path.join(workspace, name, ".git"))] + if len(found) == 1: + return found[0], None + if not found: + return None, "needs-boot: no cloned repo in workspace" + return None, "ambiguous: %d repos, explicit repo required" % len(found) + + +def _run_sync(action, repo): + """跑 chv-sync pull/push；返回 (ok, tail)。tail 只取末行 + （git 报错不含 token：remote URL 是干净 https，credential + 走 store 文件），截断 500 字符。""" + try: + proc = subprocess.run( + [SYNC_BIN, action, repo], capture_output=True, + text=True, errors="replace", timeout=300) + except OSError as exc: + return False, "sync binary failed: %s" % exc + except subprocess.TimeoutExpired: + return False, "sync %s timeout after 300s" % action + tail = ((proc.stdout or "") + (proc.stderr or "")).strip() + tail = tail.splitlines()[-1] if tail else "" + return proc.returncode == 0, tail[:500] + + +def _git_head(repo): + try: + proc = subprocess.run( + ["git", "-C", repo, "rev-parse", "HEAD"], + capture_output=True, text=True, errors="replace", + timeout=30) + except (OSError, subprocess.TimeoutExpired): + return None + if proc.returncode != 0: + return None + return proc.stdout.strip() + + +class Syncd: + def __init__(self, interval=DEFAULT_INTERVAL, + status_path=STATUS_PATH, workspace=WORKSPACE): + self.interval = max(interval, MIN_INTERVAL) + self.status_path = status_path + self.workspace = workspace + self._stop = threading.Event() + self._cycle_lock = threading.Lock() + self._cycle_running = False + self._status = _read_json(status_path) or { + "service": "chv-syncd", "last": "never", + "at": None, "head": None, "error": None, + "consecutive_errors": 0, "cycles": 0} + + def _write_status(self): + _atomic_write(self.status_path, + (json.dumps(self._status) + "\n").encode()) + + def run_cycle(self): + """一轮同步：先 pull 再 push；任一步失败即停住记 error。""" + if not self._cycle_lock.acquire(blocking=False): + return {"ok": False, "error": "cycle already running"} + try: + self._cycle_running = True + return self._cycle() + finally: + self._cycle_running = False + self._cycle_lock.release() + + def _cycle(self): + repo, reason = resolve_repo(self.workspace) + now = time.time() + self._status["cycles"] = self._status.get("cycles", 0) + 1 + if repo is None: + self._status.update( + {"last": "error", "at": now, "error": reason, + "consecutive_errors": + self._status.get("consecutive_errors", 0) + 1}) + self._write_status() + return {"ok": False, "error": reason} + head_before = _git_head(repo) + ok, tail = _run_sync("pull", repo) + if not ok: + self._status.update( + {"last": "error", "at": now, "repo": repo, + "head": head_before, + "error": "pull failed: %s" % tail, + "consecutive_errors": + self._status.get("consecutive_errors", 0) + 1}) + self._write_status() + return {"ok": False, "error": "pull failed: %s" % tail} + ok, tail = _run_sync("push", repo) + head_after = _git_head(repo) + if not ok: + self._status.update( + {"last": "error", "at": now, "repo": repo, + "head": head_after, + "error": "push failed: %s" % tail, + "consecutive_errors": + self._status.get("consecutive_errors", 0) + 1}) + self._write_status() + return {"ok": False, "error": "push failed: %s" % tail} + self._status.update( + {"last": "ok", "at": now, "repo": repo, + "head": head_after, "head_before": head_before, + "error": None, "consecutive_errors": 0}) + self._write_status() + return {"ok": True, "repo": repo, "head": head_after} + + def handle(self, req): + if not isinstance(req, dict): + return {"ok": False, "error": "request must be an object"} + op = req.get("op") + if op == "ping": + return {"ok": True, "service": "chv-syncd", + "pid": os.getpid()} + if op == "status": + return {"ok": True, "status": self._status} + if op == "sync-now": + return self.run_cycle() + return {"ok": False, "error": "unknown op: %s" % (op,)} + + def _serve(self): + if os.path.exists(SYNC_SOCK): + os.remove(SYNC_SOCK) + srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + srv.bind(SYNC_SOCK) + os.chmod(SYNC_SOCK, 0o600) + srv.listen(8) + srv.settimeout(0.5) + try: + while not self._stop.is_set(): + try: + conn, _ = srv.accept() + except socket.timeout: + continue + except OSError: + break + with conn: + try: + conn.settimeout(10.0) + buf = b"" + while len(buf) < MAX_REQUEST_BYTES: + chunk = conn.recv(4096) + if not chunk: + break + buf += chunk + if buf.endswith(b"\n"): + break + req = json.loads(buf.decode()) + resp = self.handle(req) + except (OSError, ValueError) as exc: + resp = {"ok": False, "error": str(exc)} + try: + conn.sendall((json.dumps(resp) + "\n").encode()) + except OSError: + pass + finally: + try: + srv.close() + except OSError: + pass + try: + os.remove(SYNC_SOCK) + except OSError: + pass + + def run(self): + thread = threading.Thread(target=self._serve, daemon=True) + thread.start() + print("chv-syncd pid=%d interval=%.0fs status=%s" + % (os.getpid(), self.interval, self.status_path), + flush=True) + try: + while not self._stop.is_set(): + self.run_cycle() + self._stop.wait(self.interval) + finally: + self._stop.set() + thread.join(timeout=5) + + def stop(self): + self._stop.set() + + +def main(argv=None): + argv = list(sys.argv[1:] if argv is None else argv) + interval = float(os.environ.get("CHV_SYNC_INTERVAL") + or DEFAULT_INTERVAL) + i = 0 + while i < len(argv): + if argv[i] == "--interval" and i + 1 < len(argv): + interval = float(argv[i + 1]) + i += 2 + else: + print("用法：chv-syncd [--interval 秒]", file=sys.stderr) + return 2 + syncd = Syncd(interval=interval) + + def _on_signal(signum, _frame): + syncd.stop() + + signal.signal(signal.SIGTERM, _on_signal) + signal.signal(signal.SIGINT, _on_signal) + syncd.run() + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/chv-guest/generate_lockfile.py b/chv-guest/generate_lockfile.py index c72ee64..0d70611 100644 --- a/chv-guest/generate_lockfile.py +++ b/chv-guest/generate_lockfile.py @@ -54,6 +54,11 @@ DECLARED_SOURCES = { # feat-chv-opt-landing（阶段 1）：数据盘挂载脚本，版本随 # repo_commit（与 chv-sync 同款）。 "chv-data-mount": "repo_commit", + # feat-chv-opt-landing（阶段 2）：客机运行时三分服务，版本 + # 随 repo_commit（与 chv-sync 同款）。 + "chv-daemon": "repo_commit", + "chv-execd": "repo_commit", + "chv-syncd": "repo_commit", } diff --git a/cloud/chv/executor.py b/cloud/chv/executor.py index 85a5749..c9920a8 100644 --- a/cloud/chv/executor.py +++ b/cloud/chv/executor.py @@ -1630,11 +1630,109 @@ class CloudHypervisorExecutor(Executor): "--api-socket=%s" % socket_path] + list(args)) if getattr(result, "returncode", 0) not in (0, None): + # feat-chv-opt-landing（阶段 2）：stderr 尾巴进报错， + # 否则 tap-busy（Resource busy / os error 16）这类 + # 原因只剩 returncode，调用方无法区分瞬时竞态与 + # 真失败、重试 wrapper 无从下手。截断 300 字符。 + tail = (getattr(result, "stderr", "") or "").strip() + tail = tail.splitlines()[-1] if tail else "" raise RuntimeError( - "ch-remote %s 失败：returncode=%s" - % (" ".join(args), getattr(result, "returncode"))) + "ch-remote %s 失败：returncode=%s%s" + % (" ".join(args), getattr(result, "returncode"), + (" stderr=%s" % tail[:300]) if tail else "")) return result + # -- tap 空闲轮询与退避重试（阶段 2，§6.1/§6.2 遗留） ------------ + def _tap_wait_seconds(self): + try: + return max(0.0, float( + self._env.get("CHV_TAP_WAIT_TIMEOUT", 30))) + except (TypeError, ValueError): + return 30.0 + + def _tap_wait_interval(self): + try: + return max(0.5, float( + self._env.get("CHV_TAP_WAIT_INTERVAL", 2))) + except (TypeError, ValueError): + return 2.0 + + @staticmethod + def _is_tap_busy_error(exc): + text = str(exc) + lowered = text.lower() + return ("resource busy" in lowered + or "os error 16" in lowered + or ("tap" in lowered and "busy" in lowered)) + + def _tap_names_for_task(self): + """该任务 net 条目的 tap 名（去重去空）。""" + if not self.net_json: + return [] + try: + entries = parse_net_entries(self.net_json) + except Exception: + return [] + taps = [] + for entry in entries: + tap = (entry.get("tap") or "").strip() + if tap and tap not in taps: + taps.append(tap) + return taps + + def _wait_tap_idle(self, taps, timeout=None): + """等 tap 空闲：`ip link show ` 不存在即空闲； + state DOWN / NO-CARRIER 即空闲；否则轮询到超时。 + 无 tap 配置、ip 不可用、探查抛错、输出为空时返回 True + （按旧行为直接尝试，重试 wrapper 仍会接住 busy 再 + 等一次）。诊断一律不用 pgrep -x cloud-hypervisor + （15 字符上限永远匹配不到），只看 tap 设备状态， + 或 ps -eo pid,cmd / 直接读 /proc。""" + if not taps: + return True + if timeout is None: + timeout = self._tap_wait_seconds() + interval = self._tap_wait_interval() + deadline = time.monotonic() + max(0.0, timeout) + while True: + idle = True + for tap in taps: + try: + result = self._call_runner( + ["ip", "link", "show", tap], timeout=10) + except Exception: + continue + if getattr(result, "returncode", 0) != 0: + continue # tap 不存在 = 空闲 + out = getattr(result, "stdout", "") or "" + if not out.strip(): + continue # 无输出 = 探查无果，按空闲处理 + if "state DOWN" in out or "NO-CARRIER" in out: + continue + idle = False + if idle: + return True + if time.monotonic() >= deadline: + return False + self.sleeper(interval) + + def _retry_on_tap_busy(self, task, fn, what): + """tap-busy 退避重试：fn 抛含 Resource busy 的错时， + 等 tap 空闲（CHV_TAP_WAIT_TIMEOUT，默认 30 秒）后 + 重试一次；第二次仍 busy 即原样抛（调用方失败路径 + 处理，restore 失败路径现已保留 seed.iso 可再试）。""" + try: + return fn() + except Exception as exc: + if not self._is_tap_busy_error(exc): + raise + taps = self._tap_names_for_task() + self._emit("vm_tap_busy_retry", task, + "vm=%s %s tap busy，等 tap 空闲后重试一次" + % (self.vm_name(task), what)) + self._wait_tap_idle(taps) + return fn() + def _registered_net_entries(self, exclude_vm=None): """已登记 VM 的 net 条目（H7）：扫描 state_dir/*/net.json （wake 成功后写，stop 时删）。文件损坏/缺失即跳过该 @@ -1858,7 +1956,7 @@ class CloudHypervisorExecutor(Executor): self._launch_api_only(task) self.wait_api_ready(task) self._check_api_socket_permissions(socket_path) - status, _ = self._api_call( + status, body = self._api_call( socket_path, "PUT", "/vm.create", self.build_vm_config( task_disk, seed_disk=seed_disk, @@ -1877,11 +1975,21 @@ class CloudHypervisorExecutor(Executor): "net_registry_write_failed", "vm=%s net 登记写失败，拒绝启动（缺登记会" "漏判 tap/mac/ip 冲突）" % name) - status, _ = self._api_call(socket_path, "PUT", "/vm.boot") - if status not in (200, 204): - raise ChvFailure( - "vm_boot_failed", - "vm.boot 失败：status=%s" % status) + # feat-chv-opt-landing（阶段 2）：vm.boot 触 tap，停机 + # →再启窗口的 tap 释放竞态在此处报 Resource busy； + # 响应体进报错（否则只剩 status，分不清瞬时竞态）， + # busy 时等 tap 空闲后重试一次，不立即失败。 + def _do_boot(): + _status, _body = self._api_call( + socket_path, "PUT", "/vm.boot") + if _status not in (200, 204): + raise ChvFailure( + "vm_boot_failed", + "vm.boot 失败：status=%s body=%s" + % (_status, _body)) + return _status + + self._retry_on_tap_busy(task, _do_boot, "vm.boot") self._emit("vm_boot", task, "vm=%s" % name) self.wait_guest_ready(task) # V-9：密钥已投递时等 guest-env 对 SSH 用户可读再报 @@ -2623,8 +2731,14 @@ class CloudHypervisorExecutor(Executor): self._launch_api_only(task) self.wait_api_ready(task) self._check_api_socket_permissions(self.api_socket(task)) - self._ch_remote(task, "restore", - "source_url=%s" % snapshot_url) + # feat-chv-opt-landing（阶段 2）：restore 触 tap，停机 + # →恢复窗口的 tap 释放竞态在此处报 Resource busy； + # 等 tap 空闲后重试一次，不立即失败（见 _retry_on_tap_busy）。 + self._retry_on_tap_busy( + task, + lambda: self._ch_remote(task, "restore", + "source_url=%s" % snapshot_url), + "restore") # feat-chv-cover100：与 _boot_vm 同款——恢复成功后登记 # net.json，登记失败即拒恢复（P2-19 语义：审计 + # 返回 False + 调用方 ChvFailure）；restore_vm 的 @@ -2656,7 +2770,16 @@ class CloudHypervisorExecutor(Executor): self._restore_boot(task, snapshot_url, net_entries) except Exception: try: - self.stop_vm(task) + # feat-chv-opt-landing（阶段 2，§6.2 遗留缺陷）： + # 失败路径原先 stop_vm(task)（keep_disk=False） + # 会删 seed.iso，而快照 config.json 的 disks + # 同时引用 disk.raw 与 seed.iso，删了之后任何 + # 重试 restore 都报 seed.iso ENOENT——失败清理 + # 毁掉了恢复输入。现改 keep_disk=True：任务盘 + # 与 seed.iso 都留，重试成为可能；残留盘由 + # GC（有登记快照的受保护、无登记的按孤儿清） + # 与显式回收接手，不长期滞留。 + self.stop_vm(task, keep_disk=True) except Exception: pass raise diff --git a/cloud/tests/test_chv_muse_p2.py b/cloud/tests/test_chv_muse_p2.py index 7c7559e..ee7d340 100644 --- a/cloud/tests/test_chv_muse_p2.py +++ b/cloud/tests/test_chv_muse_p2.py @@ -569,7 +569,8 @@ class BuildDryRunTest(unittest.TestCase): for name in ("build.sh", "buildinfo.py", "check_lockfile.py", "generate_lockfile.py", "chv-sync.sh", "chv-cell.py", "chv-broker.py", "chv-browser.py", - "chv-data-mount.sh"): + "chv-data-mount.sh", "chv-daemon.py", + "chv-execd.py", "chv-syncd.py"): shutil.copy2(ROOT / "chv-guest" / name, os.path.join(guest, name)) shutil.copy2(ROOT / "host.py", os.path.join(repo, "host.py")) @@ -607,7 +608,10 @@ class BuildDryRunTest(unittest.TestCase): for needle in ("systemd-container", "chv-cell", "chv-broker", "chv-browser", "chv-browser.service", "chv-broker.service", "useradd --system", - "systemctl enable chv-broker chv-browser"): + "systemctl enable chv-broker chv-browser " + "chv-daemon chv-execd chv-syncd", + "chv-daemon.service", "chv-execd.service", + "chv-syncd.service"): self.assertIn(needle, out, needle) # P1 文件不在本分支：干跑只警告、不失败。 self.assertIn("vsock_notify.py 不在仓库里", out) diff --git a/cloud/tests/test_chv_opt_stage2.py b/cloud/tests/test_chv_opt_stage2.py new file mode 100644 index 0000000..a4f073f --- /dev/null +++ b/cloud/tests/test_chv_opt_stage2.py @@ -0,0 +1,691 @@ +"""Cloud Hypervisor 优化落地阶段 2（feat-chv-opt-landing）测试。 + +只用标准库与 mock runner/launcher/api_client，绝不启动真实 +VM（与 test_chv_opt_stage1.py 同款边界）。覆盖： +- 客机运行时三分：daemon 心跳/快照配合、execd 任务登记与 + 重启认领、syncd 只调 chv-sync pull/push（规则一条不改） +- 阶段 1 遗留修复：restore 失败后 seed.iso 仍在；tap 空闲 + 轮询 + vm.boot/restore 的 tap-busy 退避重试一次 +- 构建接线：build.sh 拷入三服务 + systemd 单元 + （Restart=always）+ enable；lockfile/buildinfo 登记三服务； + 99-chv-sync.cfg 的 runcmd 顺序不变（restart ssh → chown + guest-env → data-mount → chv-sync boot） +""" + +import importlib.util +import json +import os +import socket +import sys +import tempfile +import threading +import time +import unittest +from pathlib import Path +from unittest import mock + +try: + from cloud.chv import ChvFailure + from cloud.chv import CloudHypervisorExecutor + from cloud.chv.disk import record_snapshot +except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.chv import ChvFailure + from cloud.chv import CloudHypervisorExecutor + from cloud.chv.disk import record_snapshot + + +def _load_guest_script(name): + path = os.path.join(Path(__file__).resolve().parents[2], + "chv-guest", name) + spec = importlib.util.spec_from_file_location( + "stage2_" + name[:-3], path) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + +class FakeResult: + def __init__(self, returncode=0, stdout="", stderr=""): + self.returncode = returncode + self.stdout = stdout + self.stderr = stderr + + +class FakeProc: + def __init__(self, alive=True): + self._alive = alive + + def poll(self): + return None if self._alive else 1 + + def terminate(self): + self._alive = False + + def wait(self, timeout=None): + return 0 + + def kill(self): + self._alive = False + + +def _write(path, content=b"x"): + with open(path, "wb") as fh: + fh.write(content) + os.chmod(path, 0o755) + return path + + +def _repo(*parts): + return os.path.join(Path(__file__).resolve().parents[2], *parts) + + +# -- daemon：心跳与快照配合 --------------------------------------------------- +class DaemonTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.mod = _load_guest_script("chv-daemon.py") + self.mod.HEARTBEAT_PATH = os.path.join( + self.tmp.name, "heartbeat.json") + self.mod.SNAPSHOT_READY_PATH = os.path.join( + self.tmp.name, "snapshot-ready.json") + self.mod.EXECD_TASKS = os.path.join( + self.tmp.name, "execd-tasks.json") + self.mod.SYNC_STATUS = os.path.join( + self.tmp.name, "sync-status.json") + self.mod.WORKSPACE = self.tmp.name + + def tearDown(self): + self.tmp.cleanup() + + def test_heartbeat_fields_and_seq_continues(self): + daemon = self.mod.Daemon(interval=10) + first = daemon.write_heartbeat() + self.assertEqual(first["service"], "chv-daemon") + self.assertEqual(first["pid"], os.getpid()) + self.assertIn("loadavg", first) + self.assertIn("mem", first) + self.assertIn("workspace_free_bytes", first) + self.assertEqual(first["seq"], 1) + daemon.write_heartbeat() + # 新 daemon 启动读旧文件续号，不是从 1 重来。 + daemon2 = self.mod.Daemon(interval=10) + third = daemon2.write_heartbeat() + self.assertEqual(third["seq"], 3) + + def test_handle_ping_status_unknown(self): + self.assertTrue(self.mod.handle_request( + {"op": "ping"})["ok"]) + status = self.mod.handle_request({"op": "status"}) + self.assertTrue(status["ok"]) + self.assertIsNone(status["heartbeat"]) + self.assertFalse(self.mod.handle_request( + {"op": "bogus"})["ok"]) + self.assertFalse(self.mod.handle_request( + "not-a-dict")["ok"]) + + def test_prepare_snapshot_refuses_when_execd_unreachable(self): + with mock.patch.object( + self.mod, "_ask_execd_status", + return_value=(None, "execd-unreachable: refused")): + resp = self.mod.handle_request({"op": "prepare-snapshot"}) + self.assertFalse(resp["ok"]) + self.assertIn("execd-unreachable", resp["reason"]) + self.assertFalse(os.path.exists( + self.mod.SNAPSHOT_READY_PATH)) + + def test_prepare_snapshot_refuses_running_task(self): + with mock.patch.object( + self.mod, "_ask_execd_status", + return_value=({"ok": True, "tasks": [ + {"id": "abc", "state": "running"}]}, None)): + resp = self.mod.handle_request({"op": "prepare-snapshot"}) + self.assertFalse(resp["ok"]) + self.assertEqual(resp["reason"], "task-running") + self.assertEqual(resp["running"], ["abc"]) + + def test_prepare_snapshot_ok_writes_marker(self): + with mock.patch.object( + self.mod, "_ask_execd_status", + return_value=({"ok": True, "tasks": [ + {"id": "abc", "state": "completed", "rc": 0}]}, + None)): + resp = self.mod.handle_request({"op": "prepare-snapshot"}) + self.assertTrue(resp["ok"]) + marker = json.load(open(self.mod.SNAPSHOT_READY_PATH)) + self.assertIn("at", marker) + + +# -- execd：任务登记、路径 confinement、codex 环境白名单 ---------------------- +class ExecdTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.dir = self.tmp.name + self.mod = _load_guest_script("chv-execd.py") + self.mod.TASKS_PATH = os.path.join(self.dir, "tasks.json") + self.mod.TASKS_LOG_DIR = os.path.join(self.dir, "tasks") + self.mod.WORKSPACE = os.path.join(self.dir, "workspace") + os.makedirs(self.mod.WORKSPACE, exist_ok=True) + + def tearDown(self): + self.tmp.cleanup() + + def _write_tasks(self, tasks): + with open(self.mod.TASKS_PATH, "w") as fh: + json.dump({"tasks": tasks}, fh) + + def test_taskstore_reattach_dead_pid_unknown(self): + self._write_tasks([{ + "id": "dead1", "kind": "exec", "pid": 999999, + "state": "running", "started_at": 1.0}]) + store = self.mod.TaskStore(self.mod.TASKS_PATH) + task = store.get("dead1") + self.assertEqual(task["state"], "unknown") + self.assertIn("unrecoverable", task["note"]) + + def test_taskstore_reattach_live_pid_stays_running(self): + self._write_tasks([{ + "id": "live1", "kind": "exec", "pid": os.getpid(), + "state": "running", "started_at": 1.0}]) + store = self.mod.TaskStore(self.mod.TASKS_PATH) + task = store.get("live1") + self.assertEqual(task["state"], "running") + self.assertIn("reattached", task["note"]) + + def test_read_write_roundtrip_and_confinement(self): + execd = self.mod.Execd(tasks_path=self.mod.TASKS_PATH) + inside = os.path.join(self.mod.WORKSPACE, "sub", "a.txt") + out = execd.op_write({"path": inside, "content": "hello-stage2"}) + self.assertTrue(out["ok"]) + back = execd.op_read({"path": inside}) + self.assertEqual(back["content"], "hello-stage2") + # 越界：工作区外、.. 穿透都拒绝。 + self.assertFalse(execd.op_write( + {"path": "/etc/passwd", "content": "x"})["ok"]) + self.assertFalse(execd.op_read( + {"path": os.path.join( + self.mod.WORKSPACE, "..", "tasks.json")})["ok"]) + self.assertFalse(execd.op_write( + {"path": inside, "content": 123})["ok"]) + + def test_exec_validates_argv_and_cwd(self): + execd = self.mod.Execd(tasks_path=self.mod.TASKS_PATH) + self.assertFalse(execd.op_exec({"argv": []})["ok"]) + self.assertFalse(execd.op_exec({"argv": "id"})["ok"]) + self.assertFalse(execd.op_exec( + {"argv": ["id"], "cwd": "/etc"})["ok"]) + + def test_codex_env_allowlist(self): + env = self.mod._codex_env({ + "OPENAI_API_KEY": "sk-test", + "CODEX_FOO": "yes", + "GITEA_TOKEN": "must-not-pass", + "HOME": "/root", + "CHV_MODEL_BASE": "http://x", + "PLAYWRIGHT_BROWSERS_PATH": "/custom"}) + self.assertEqual(env["OPENAI_API_KEY"], "sk-test") + self.assertEqual(env["CODEX_FOO"], "yes") + self.assertEqual(env["CHV_MODEL_BASE"], "http://x") + self.assertNotIn("GITEA_TOKEN", env) + # HOME 固定为 chv 家目录，不接受调用方传入。 + self.assertEqual(env["HOME"], "/home/chv") + + def test_socket_ping_and_read_write(self): + sock_path = os.path.join(self.dir, "execd.sock") + self.mod.EXECD_SOCK = sock_path + execd = self.mod.Execd(tasks_path=self.mod.TASKS_PATH) + thread = threading.Thread(target=execd.serve, daemon=True) + thread.start() + try: + for _ in range(50): + if os.path.exists(sock_path): + break + time.sleep(0.1) + self.assertTrue(os.path.exists(sock_path)) + + def call(req): + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + sock.settimeout(5) + sock.connect(sock_path) + sock.sendall((json.dumps(req) + "\n").encode()) + buf = b"" + while not buf.endswith(b"\n"): + chunk = sock.recv(4096) + if not chunk: + break + buf += chunk + sock.close() + return json.loads(buf.decode()) + + self.assertTrue(call({"op": "ping"})["ok"]) + path = os.path.join(self.mod.WORKSPACE, "via-sock.txt") + self.assertTrue(call({"op": "write", "path": path, + "content": "sock-ok"})["ok"]) + back = call({"op": "read", "path": path}) + self.assertEqual(back["content"], "sock-ok") + self.assertFalse(call({"op": "bogus"})["ok"]) + status = call({"op": "status"}) + self.assertTrue(status["ok"]) + finally: + execd.stop() + thread.join(timeout=5) + + +# -- syncd：只调 chv-sync，规则不改 ------------------------------------------- +class SyncdTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.dir = self.tmp.name + self.mod = _load_guest_script("chv-syncd.py") + self.mod.WORKSPACE = os.path.join(self.dir, "workspace") + os.makedirs(self.mod.WORKSPACE, exist_ok=True) + self.status_path = os.path.join(self.dir, "sync-status.json") + + def tearDown(self): + self.tmp.cleanup() + + def _repo_dir(self, name="chv-sync-test"): + path = os.path.join(self.mod.WORKSPACE, name) + os.makedirs(os.path.join(path, ".git"), exist_ok=True) + return path + + def test_resolve_repo_zero_one_many(self): + repo, reason = self.mod.resolve_repo(self.mod.WORKSPACE) + self.assertIsNone(repo) + self.assertIn("needs-boot", reason) + only = self._repo_dir() + repo, _ = self.mod.resolve_repo(self.mod.WORKSPACE) + self.assertEqual(repo, only) + self._repo_dir("other") + repo, reason = self.mod.resolve_repo(self.mod.WORKSPACE) + self.assertIsNone(repo) + self.assertIn("ambiguous", reason) + + def test_cycle_pull_fail_stops_before_push(self): + repo = self._repo_dir() + syncd = self.mod.Syncd(interval=300, + status_path=self.status_path, + workspace=self.mod.WORKSPACE) + calls = [] + with mock.patch.object( + self.mod, "_run_sync", + side_effect=lambda a, r: ( + calls.append(a), + (False, "non-fast-forward"))[1]): + with mock.patch.object( + self.mod, "_git_head", return_value="abc123"): + out = syncd.run_cycle() + self.assertFalse(out["ok"]) + self.assertEqual(calls, ["pull"]) # pull 失败即停，不 push + with open(self.status_path) as fh: + status = json.load(fh) + self.assertEqual(status["last"], "error") + self.assertIn("pull failed", status["error"]) + self.assertEqual(status["head"], "abc123") + self.assertEqual(status["consecutive_errors"], 1) + + def test_cycle_pull_push_ok(self): + repo = self._repo_dir() + syncd = self.mod.Syncd(interval=300, + status_path=self.status_path, + workspace=self.mod.WORKSPACE) + calls = [] + with mock.patch.object( + self.mod, "_run_sync", + side_effect=lambda a, r: ( + calls.append(a), (True, ""))[1]): + with mock.patch.object( + self.mod, "_git_head", + side_effect=["before", "after"]): + out = syncd.run_cycle() + self.assertTrue(out["ok"]) + self.assertEqual(calls, ["pull", "push"]) + with open(self.status_path) as fh: + status = json.load(fh) + self.assertEqual(status["last"], "ok") + self.assertEqual(status["head"], "after") + self.assertEqual(status["consecutive_errors"], 0) + + def test_syncd_source_never_forces(self): + with open(_repo("chv-guest", "chv-syncd.py")) as fh: + text = fh.read() + # 只查可执行代码里的字符串字面量与命令（文档里“绝不 + # --force / 不自动 rebase”是规则说明，不是命令）。 + for banned in ('"--force"', '"rebase"', "'--force'", + "reset --hard", '"GITEA_TOKEN"'): + self.assertNotIn(banned, text) + # syncd 只调 chv-sync（pull/push）与 git rev-parse， + # 不直接跑 git pull/push/clone。 + self.assertIn('[SYNC_BIN, action, repo]', text) + self.assertNotIn('"pull"', text.replace( + '"pull", repo', '')) + self.assertNotIn('["git", "-C", repo, "pull"', text) + self.assertNotIn('["git", "-C", repo, "push"', text) + + def test_chv_sync_rules_unchanged(self): + with open(_repo("chv-guest", "chv-sync.sh")) as fh: + lines = fh.readlines() + # 只查非注释行里的 git 命令：含 --force / rebase 的行 + # 必须是 die/echo 文案（“绝不 --force”是规则说明）， + # 不能是真正的 git 调用。 + code = [ln for ln in lines + if not ln.strip().startswith("#")] + joined = "".join(code) + self.assertIn("pull --ff-only", joined) + offending = [ln.strip() for ln in code + if ("--force" in ln or "rebase" in ln) + and "die " not in ln and "echo " not in ln] + self.assertEqual(offending, []) + self.assertIn("不自动 rebase", "".join(lines)) + + +# -- 构建接线 ----------------------------------------------------------------- +class BuildWiringTest(unittest.TestCase): + def test_build_sh_wires_three_services(self): + with open(_repo("chv-guest", "build.sh")) as fh: + text = fh.read() + for name in ("chv-daemon", "chv-execd", "chv-syncd"): + self.assertIn("%s.py:/usr/local/bin/" % name, text) + self.assertIn("/usr/local/bin/%s" % name, text) + self.assertIn("%s.service" % name, text) + # Restart=always 恰好 3 处（三个阶段 2 服务；broker / + # browser 仍是 on-failure，不动它们），kill -9 / + # kill -TERM 都自动拉起，只有 systemctl stop 才干净停。 + self.assertEqual(text.count("Restart=always"), 3) + self.assertEqual(text.count("Restart=on-failure"), 2) + for name, desc in ( + ("chv-daemon", "CHV guest runtime daemon"), + ("chv-execd", "CHV guest exec daemon"), + ("chv-syncd", "CHV Gitea sync daemon")): + self.assertIn(desc, text) + self.assertIn("%s.service" % name, text) + self.assertIn( + "systemctl enable chv-broker chv-browser " + "chv-daemon chv-execd chv-syncd", text) + # 99-chv-sync.cfg 的 runcmd 顺序一条不改。 + cfg_line = next( + line for line in text.splitlines() + if "99-chv-sync.cfg" in line and "runcmd" in line) + self.assertLess(cfg_line.index("systemctl restart ssh"), + cfg_line.index("chown chv:chv /etc/chv/guest-env")) + self.assertLess(cfg_line.index("chv-data-mount >>"), + cfg_line.index("chv-sync boot")) + + def test_lockfile_and_buildinfo_declare_services(self): + spec = importlib.util.spec_from_file_location( + "stage2_buildinfo", + _repo("chv-guest", "buildinfo.py")) + buildinfo = importlib.util.module_from_spec(spec) + spec.loader.exec_module(buildinfo) + for name in ("chv-daemon", "chv-execd", "chv-syncd"): + self.assertIn(name, buildinfo.BASE_CONTENTS) + for name in ("check_lockfile.py", "generate_lockfile.py"): + spec = importlib.util.spec_from_file_location( + "stage2_" + name[:-3], _repo("chv-guest", name)) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + for svc in ("chv-daemon", "chv-execd", "chv-syncd"): + self.assertEqual(mod.DECLARED_SOURCES[svc], + "repo_commit") + + +# -- executor：restore 失败留 seed.iso + tap 退避 ------------------------------ +class ChvOpt2Case(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.dir = self.tmp.name + self.binary = _write(os.path.join(self.dir, "cloud-hypervisor")) + self.ch_remote = _write(os.path.join(self.dir, "ch-remote")) + self.kernel = _write(os.path.join(self.dir, "vmlinux")) + self.rootfs = _write(os.path.join(self.dir, "base.qcow2"), + b"R" * 100) + self.ssh_key = _write(os.path.join(self.dir, "id_chv")) + self.kvm = _write(os.path.join(self.dir, "kvm")) + self.cpuinfo = os.path.join(self.dir, "cpuinfo") + with open(self.cpuinfo, "w") as fh: + fh.write("flags : fpu vmx sse\n") + self.state_dir = os.path.join(self.dir, "state") + self.calls = [] + self.api_calls = [] + self.launches = [] + self.vm_states = {} + + def tearDown(self): + self.tmp.cleanup() + + def make_exec(self, runner=None, api_client=None, launcher=None, + env_extra=None, **overrides): + env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888", + "CHV_HOST_KEY_MODE": "keyscan", + "CHV_SSH_FORWARD": "1", + "CHV_NET_JSON": json.dumps([{ + "tap": "tap-chv", "mac": "02:00:00:00:00:11"}]), + "CHV_TAP_WAIT_TIMEOUT": "0", + "CHV_TAP_WAIT_INTERVAL": "0.5"} + if env_extra: + env.update(env_extra) + if runner is None: + runner = self._default_runner + if api_client is None: + api_client = self._default_api + if launcher is None: + launcher = self._default_launcher + kwargs = dict( + binary=self.binary, ch_remote=self.ch_remote, + kernel=self.kernel, rootfs=self.rootfs, + ssh_key=self.ssh_key, kvm_path=self.kvm, + cpuinfo_path=self.cpuinfo, state_dir=self.state_dir, + runner=runner, launcher=launcher, api_client=api_client, + sleeper=lambda s: None, env=env) + kwargs.update(overrides) + return CloudHypervisorExecutor(**kwargs) + + def _default_runner(self, cmd, timeout=None): + self.calls.append(list(cmd)) + if cmd[0] == "ssh-keyscan": + return FakeResult( + 0, "[127.0.0.1]:2222 ssh-ed25519 AAAAB3test\n", "") + if cmd[0] == "ssh": + return FakeResult(0, "", "") + if cmd[:3] == ["ip", "link", "show"]: + # tap 不存在（rc=1）= 空闲，立即返回不等待。 + return FakeResult(1, "", "") + return FakeResult(0, "", "") + + def _default_api(self, socket_path, method, path, payload=None): + self.api_calls.append((method, path, payload)) + if path == "/vm.info": + state = self.vm_states.get( + os.path.basename(socket_path)[:-len(".sock")]) + if state: + return (200, {"state": state}) + return (200, {}) + return (200, {}) + + def _default_launcher(self, cmd): + self.launches.append(list(cmd)) + return FakeProc(alive=True) + + def _seed_vm_files(self, vm): + vm_dir = os.path.join(self.state_dir, vm) + os.makedirs(vm_dir, exist_ok=True) + seed = os.path.join(vm_dir, "seed.iso") + disk = os.path.join(vm_dir, "disk.raw") + with open(seed, "wb") as fh: + fh.write(b"SEED") + with open(disk, "wb") as fh: + fh.write(b"DISK") + return seed, disk + + def _snap_url(self, vm, name="snap1"): + snap_dir = os.path.join(self.state_dir, vm, name) + os.makedirs(snap_dir, exist_ok=True) + record_snapshot(self.state_dir, vm, + "file://" + snap_dir, 20.0, + tenant_id="tenant-a") + return "file://" + snap_dir + + +class RestoreKeepsSeedTest(ChvOpt2Case): + def test_restore_failure_keeps_seed_iso(self): + url = self._snap_url("chv-a") + seed, disk = self._seed_vm_files("chv-a") + ex = self.make_exec() + calls = {"n": 0} + + def failing_restore(task_or_name, *args): + calls["n"] += 1 + raise RuntimeError( + "ch-remote restore 失败：returncode=1 " + "stderr=Cannot open disk path") + with mock.patch.object(ex, "_ch_remote", + side_effect=failing_restore): + with self.assertRaises(RuntimeError): + ex.restore_vm({"id": "a", "project_id": "tenant-a"}, + url) + # 非 tap-busy 失败：不重试，但 seed.iso 与任务盘都在， + # 重试 restore 仍有输入。 + self.assertEqual(calls["n"], 1) + self.assertTrue(os.path.exists(seed)) + self.assertTrue(os.path.exists(disk)) + + def test_restore_tap_busy_retries_once_then_keeps_seed(self): + url = self._snap_url("chv-a") + seed, disk = self._seed_vm_files("chv-a") + ex = self.make_exec() + events = [] + ex.audit_callback = lambda e, t, d, k: events.append(e) + calls = {"n": 0} + + def busy_restore(task_or_name, *args): + calls["n"] += 1 + raise RuntimeError( + "ch-remote restore 失败：returncode=1 " + "stderr=Resource busy (os error 16)") + with mock.patch.object(ex, "_ch_remote", + side_effect=busy_restore): + with self.assertRaises(RuntimeError): + ex.restore_vm({"id": "a", "project_id": "tenant-a"}, + url) + # tap-busy：等空闲后重试一次（共 2 次），仍失败才抛； + # seed.iso 与盘仍在，可再试。 + self.assertEqual(calls["n"], 2) + self.assertIn("vm_tap_busy_retry", events) + self.assertTrue(os.path.exists(seed)) + self.assertTrue(os.path.exists(disk)) + + def test_restore_tap_busy_second_try_succeeds(self): + url = self._snap_url("chv-a") + self._seed_vm_files("chv-a") + ex = self.make_exec() + calls = {"n": 0} + + def flaky_restore(task_or_name, *args): + calls["n"] += 1 + if calls["n"] == 1: + raise RuntimeError( + "ch-remote restore 失败：returncode=1 " + "stderr=Resource busy (os error 16)") + return FakeResult(0, "", "") + with mock.patch.object(ex, "_ch_remote", + side_effect=flaky_restore): + with mock.patch.object(ex, "resume_vm", + return_value=None), \ + mock.patch.object(ex, "wait_guest_ready", + return_value=None): + out = ex.restore_vm( + {"id": "a", "project_id": "tenant-a"}, url) + self.assertEqual(calls["n"], 2) + self.assertEqual(out["mode"], "restored") + + +class TapWaitTest(ChvOpt2Case): + def test_is_tap_busy_error(self): + ex = self.make_exec() + self.assertTrue(ex._is_tap_busy_error(RuntimeError( + "Resource busy (os error 16)"))) + self.assertTrue(ex._is_tap_busy_error(ChvFailure( + "vm_boot_failed", "vm.boot 失败：status=500 " + "body=Resource busy (os error 16)"))) + self.assertFalse(ex._is_tap_busy_error(RuntimeError( + "Cannot open disk path ... No such file"))) + self.assertFalse(ex._is_tap_busy_error(ChvFailure( + "vm_boot_failed", "vm.boot 失败：status=500 body=boom"))) + + def test_wait_tap_idle_states(self): + ex = self.make_exec() + + def down_runner(cmd, timeout=None): + if cmd[:3] == ["ip", "link", "show"]: + return FakeResult( + 0, "3: tap-chv: " + "mtu 1500 state DOWN mode DEFAULT", "") + return FakeResult(0, "", "") + ex.runner = down_runner + self.assertTrue(ex._wait_tap_idle(["tap-chv"], timeout=0)) + + def up_runner(cmd, timeout=None): + if cmd[:3] == ["ip", "link", "show"]: + return FakeResult( + 0, "3: tap-chv: " + "mtu 1500 state UP mode DEFAULT", "") + return FakeResult(0, "", "") + ex.runner = up_runner + self.assertFalse(ex._wait_tap_idle(["tap-chv"], timeout=0)) + + def absent_runner(cmd, timeout=None): + if cmd[:3] == ["ip", "link", "show"]: + return FakeResult(1, "", "") + return FakeResult(0, "", "") + ex.runner = absent_runner + self.assertTrue(ex._wait_tap_idle(["tap-chv"], timeout=0)) + self.assertTrue(ex._wait_tap_idle([], timeout=0)) + + def test_boot_tap_busy_retries_once(self): + ex = self.make_exec() + boots = {"n": 0} + + def flaky_api(socket_path, method, path, payload=None): + self.api_calls.append((method, path, payload)) + if path == "/vm.boot": + boots["n"] += 1 + if boots["n"] == 1: + return (500, "Resource busy (os error 16)") + return (204, {}) + return (200, {}) + ex.api_client = flaky_api + with mock.patch.object(ex, "wait_guest_ready", + return_value=None), \ + mock.patch.object(ex, "_wait_guest_env_usable", + return_value=None): + ex._boot_vm({"id": "a"}, "chv-a", + os.path.join(self.state_dir, "chv-a", + "disk.raw"), + None, []) + self.assertEqual(boots["n"], 2) + + def test_boot_non_busy_failure_no_retry(self): + ex = self.make_exec() + boots = {"n": 0} + + def failing_api(socket_path, method, path, payload=None): + if path == "/vm.boot": + boots["n"] += 1 + return (500, "Cannot open disk path") + return (200, {}) + ex.api_client = failing_api + with mock.patch.object(ex, "wait_guest_ready", + return_value=None), \ + mock.patch.object(ex, "_wait_guest_env_usable", + return_value=None): + with self.assertRaises(ChvFailure) as ctx: + ex._boot_vm({"id": "a"}, "chv-a", + os.path.join(self.state_dir, "chv-a", + "disk.raw"), + None, []) + self.assertEqual(ctx.exception.kind, "vm_boot_failed") + self.assertEqual(boots["n"], 1) + + +if __name__ == "__main__": + unittest.main()