From 7a6fae79933894860bf9c14e2b0f6b3d9eb0e3f0 Mon Sep 17 00:00:00 2001 From: host Date: Sat, 7 Nov 2026 05:46:52 +0000 Subject: [PATCH] =?UTF-8?q?feat(chv):=20=E9=98=B6=E6=AE=B5=201=20=E6=8C=81?= =?UTF-8?q?=E4=B9=85=20VM=E3=80=81overlay=20=E5=88=86=E5=B1=82=E4=B8=8E?= =?UTF-8?q?=E9=A2=84=E7=83=AD=E5=BF=AB=E7=85=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - disk.py：overlay 模式（qcow2 backing file 指只读 base， 建盘/复用/重建校验 + qemu-img fail-closed）、数据盘 data.qcow2 独立供给（不进快照/GC/stop）、配额增量口径 （overlay 本次增量 0）、快照登记 tenant 字段、每 VM tenant.json、持久登记 persistent/.json、快照 登记按 URL 查找。 - config.py：CHV_DISK_MODE 白名单加 overlay（默认仍 copy； 错误文案保留旧子串，防回归测试钉死）。 - executor.py：VmConfig image_type 按后缀（.qcow2→Qcow2， 其余 Raw）+ 数据盘第三块盘；overlay 供给写 tenant 标记； snapshot_vm 登记带 tenant，_validate_restore 归属校验 （跨用户 snapshot_tenant_mismatch 拒绝 + 审计，legacy 未标记/未登记放行并警告）；stop/GC 在 overlay 模式下 保护持久登记/快照认领的盘；持久 VM（prewarm/acquire/ suspend/dispatch_task_persistent/recycle_persistent_vm， CHV_WARM_POOL_SIZE>=1 启用，=0 退化冷启旧语义）， resume 只认 vm.info=Paused + socket 活性 + 归属一致。 - tests：cloud/tests/test_chv_opt_stage1.py 26 项（overlay、 tenant、持久池、跨用户拒绝、两连任务、显式回收）。 回归：discover test_chv_*.py 与 test_cloud_hypervisor.py 全过（结构契约：函数 ≤50 行、公开 API 全注解、文案钉死）。 不推、不合、不部署；只本地分支。 --- cloud/chv/config.py | 5 +- cloud/chv/disk.py | 368 +++++++++++++++- cloud/chv/executor.py | 678 +++++++++++++++++++++++++++-- cloud/tests/test_chv_opt_stage1.py | 657 ++++++++++++++++++++++++++++ 4 files changed, 1643 insertions(+), 65 deletions(-) create mode 100644 cloud/tests/test_chv_opt_stage1.py diff --git a/cloud/chv/config.py b/cloud/chv/config.py index a9d2554..68c6001 100644 --- a/cloud/chv/config.py +++ b/cloud/chv/config.py @@ -177,9 +177,10 @@ def _disk_fragment(env: dict) -> dict: 文案与旧 quota 片段逐字一致（disk_mode 白名单、 disk_gb > 0、快照数非负）。""" disk_mode = env.get("CHV_DISK_MODE") or "copy" - if disk_mode not in ("copy", "reuse"): + if disk_mode not in ("copy", "reuse", "overlay"): raise ValueError( - "CHV_DISK_MODE 只能是 copy 或 reuse：%s" % disk_mode) + "CHV_DISK_MODE 只能是 copy 或 reuse 或 overlay：%s" + % disk_mode) try: disk_gb = _to_float(env.get("CHV_DISK_GB")) \ if env.get("CHV_DISK_GB") else 50.0 diff --git a/cloud/chv/disk.py b/cloud/chv/disk.py index 7ed0ac4..81677a6 100644 --- a/cloud/chv/disk.py +++ b/cloud/chv/disk.py @@ -14,10 +14,19 @@ check_disk_quota 改为三项：state_dir 已用总量 + 本次 是直接调用者的第二道门）。provision_task_disk 对已存在 磁盘校验大小：与源不一致即重拷；源不可读即报失败， 绝不复用脏磁盘。 + +feat-chv-opt-landing（阶段 1）：黄金镜像只读 base + +qcow2 overlay 分层（disk_mode="overlay"）、数据盘独立 +持久（data.qcow2，绝不进快照/GC）、快照与 overlay 的 +tenant 归属标记（登记条目 tenant 字段 + 每 VM tenant.json ++ 持久 VM 登记 persistent/.json）。overlay 与 +数据盘只在 overlay 模式下供给；copy/reuse 语义不变。 """ import os import shutil +import subprocess +from collections.abc import Callable from cloud.chv.errors import ChvFailure @@ -33,7 +42,8 @@ def _dir_total_bytes(path): return total -def check_disk_quota(rootfs: str, disk_gb: float, state_dir: str | None = None) -> None: +def check_disk_quota(rootfs: str, disk_gb: float, state_dir: str | None = None, + increment_bytes: int | None = None) -> None: """磁盘配额门（feat-chv-fix 口径）： - disk_gb 未设时不检查； @@ -42,6 +52,11 @@ def check_disk_quota(rootfs: str, disk_gb: float, state_dir: str | None = None) - state_dir 已给时：已用总量 + 本次拷贝超过 disk_gb 即 disk_quota_exceeded；宿主剩余空间不足以容纳本次 拷贝即 disk_quota_exceeded（宿主余量校验）。 + - increment_bytes（feat-chv-opt-landing）：本次实际新增 + 的磁盘字节数，overlay 模式传 0（只建空 overlay，不拷 + base，base 本身不计入增量）；不给时默认等于镜像大小 + （copy 模式口径不变）。镜像大小超过 disk_gb 的硬帽 + 仍按镜像本身算，与增量无关。 """ if disk_gb is None or not rootfs: return @@ -55,33 +70,46 @@ def check_disk_quota(rootfs: str, disk_gb: float, state_dir: str | None = None) raise ChvFailure( "disk_quota_exceeded", "磁盘镜像 %d 字节超过配额 %.1f GB" % (size, disk_gb)) + increment = size if increment_bytes is None else increment_bytes if state_dir and os.path.isdir(state_dir): used = _dir_total_bytes(state_dir) - if used + size > disk_gb * 1024 ** 3: + if used + increment > disk_gb * 1024 ** 3: raise ChvFailure( "disk_quota_exceeded", "磁盘总量 %d 字节 + 本次 %d 字节超过配额 %.1f GB" - % (used, size, disk_gb)) + % (used, increment, disk_gb)) try: free = shutil.disk_usage(state_dir).free except OSError: free = None - if free is not None and free < size: + if free is not None and free < increment: raise ChvFailure( "disk_quota_exceeded", "宿主剩余空间 %d 字节不足以容纳本次磁盘拷贝 %d 字节" - % (free, size)) + % (free, increment)) -def provision_task_disk(state_dir: str, vm_name: str, rootfs: str, disk_mode: str) -> str: +def provision_task_disk(state_dir: str, vm_name: str, rootfs: str, disk_mode: str, + runner: Callable[..., object] | None = None) -> str: """每任务磁盘路径：copy 模式复制到 - state_dir//disk.raw；reuse 模式直接返回基础镜像路径。 + state_dir//disk.raw；reuse 模式直接返回基础镜像路径； + overlay 模式（feat-chv-opt-landing，阶段 1）在 + state_dir//disk.qcow2 建空 qcow2 overlay，backing + file 指向只读 base（base 本身一字节都不改、不拷）。 feat-chv-fix：已存在磁盘须校验——大小与源不一致即 重拷（脏磁盘不复用）；源不可读即 ChvFailure，绝不 - 静默返回旧磁盘。""" + 静默返回旧磁盘。overlay 模式已存在 overlay 时校验 + backing file 是否仍指向同一 base：是即复用，否即删 + 了重建（旧 overlay 的 base 已换，复用等于挂错盘）。""" if disk_mode == "reuse": return rootfs + if disk_mode == "overlay": + task_dir = os.path.join(state_dir, vm_name) + os.makedirs(task_dir, exist_ok=True) + return create_overlay_disk( + rootfs, overlay_disk_path(state_dir, vm_name), + runner=runner) task_dir = os.path.join(state_dir, vm_name) os.makedirs(task_dir, exist_ok=True) task_disk = os.path.join(task_dir, "disk.raw") @@ -113,23 +141,325 @@ def provision_task_disk(state_dir: str, vm_name: str, rootfs: str, disk_mode: st def task_disk_path(state_dir: str, vm_name: str) -> str: - """该 VM 的任务磁盘路径：state_dir//disk.raw。""" + """该 VM 的任务磁盘路径：state_dir//disk.raw + （copy 模式；overlay 模式见 overlay_disk_path）。""" return os.path.join(state_dir, vm_name, "disk.raw") +def task_disk_paths(state_dir: str, vm_name: str) -> list: + """该 VM 现存的任务磁盘文件（copy 的 disk.raw 与/或 + overlay 的 disk.qcow2）：GC 与清理按此清单查，不猜 + 文件名；数据盘 data.qcow2 不在内——用户数据盘永不 + 进任务盘 GC，见 delete_data_disk。""" + out = [] + for name in ("disk.raw", OVERLAY_DISK_NAME): + path = os.path.join(state_dir, vm_name, name) + if os.path.islink(path) or os.path.exists(path): + out.append(path) + return out + + def delete_task_disk(state_dir: str, vm_name: str) -> bool: """删任务磁盘（feat-chv-secfix，H3）：stop/recycle/ 清理路径都调它。返回 True 表示确实删掉了一个文件， False 表示本来就没有；删失败抛 OSError（调用方记审计， 绝不静默——残留盘由 gc_stale_disks 下次再清）。 - 符号链接只删链接本身，不跟随。""" - path = task_disk_path(state_dir, vm_name) + 符号链接只删链接本身，不跟随。 + feat-chv-opt-landing：disk.raw 与 disk.qcow2 都删 + （任一存在即删）；数据盘 data.qcow2 绝不在此删—— + 它是用户数据，删它须走 delete_data_disk（显式回收）。""" + removed = False + for path in task_disk_paths(state_dir, vm_name): + os.remove(path) + removed = True + return removed + + +# -- overlay 分层与数据盘（feat-chv-opt-landing，阶段 1）------ +# 黄金镜像只读 base（qcow2，基线 773 MiB）绝不被写：每台 +# 用户 VM 只有一个空 overlay（disk.qcow2，backing file 指 +# 向 base）；数据盘（data.qcow2，工作区 + 浏览器 profile +# 的载体）是第二块独立盘——快照登记与 GC 只认任务盘， +# 数据盘不进快照、不进 GC、不随 stop 删（删它须显式回收）。 +# 客内挂载见镜像 99-chv-sync.cfg runcmd 的 chv-data-mount +#（按 LABEL=CHVDATA 找盘，不依赖 /dev/vd? 固定名）。 + +OVERLAY_DISK_NAME = "disk.qcow2" +DATA_DISK_NAME = "data.qcow2" +DATA_DISK_GB = 10.0 +DATA_DISK_LABEL = "CHVDATA" +TENANT_FILE_NAME = "tenant.json" +PERSISTENT_DIR_NAME = "persistent" + + +def overlay_disk_path(state_dir: str, vm_name: str) -> str: + """该 VM 的 overlay 磁盘路径：state_dir//disk.qcow2。""" + return os.path.join(state_dir, vm_name, OVERLAY_DISK_NAME) + + +def data_disk_path(state_dir: str, vm_name: str) -> str: + """该 VM 的数据盘路径：state_dir//data.qcow2。""" + return os.path.join(state_dir, vm_name, DATA_DISK_NAME) + + +def _default_qemu_runner(cmd, **kwargs): + """qemu-img 的默认执行器：subprocess 直调（宿主命令， + 与执行器的 SSH/ch-remote runner 分开，不混用）。""" + return subprocess.run(cmd, capture_output=True, text=True, + timeout=kwargs.get("timeout", 120)) + + +def _qemu_img_cmd(): + """qemu-img 可执行路径：PATH 里找不到即报 + disk_overlay_unsupported（调用方 fail-fast，不静默 + 回落 copy——回落等于悄悄拷 773 MiB base，违背分层）。""" + import shutil as _shutil + path = _shutil.which("qemu-img") + if not path: + raise ChvFailure( + "disk_overlay_unsupported", + "qemu-img 不可用，overlay 模式无法建盘") + return path + + +def _base_format(base_path: str) -> str: + """backing file 的 -F 口径：按 base 后缀定（.qcow2 / + .raw）；其它后缀即 disk_overlay_unsupported，不猜 + （格式猜错会建出读不了 base 的 overlay，启动才炸）。""" + if base_path.endswith(".qcow2"): + return "qcow2" + if base_path.endswith(".raw"): + return "raw" + raise ChvFailure( + "disk_overlay_unsupported", + "overlay 的 base 须是 .qcow2 或 .raw：%s" % base_path) + + +def _overlay_backing_matches(overlay_path: str, base_path: str, + runner, qemu: str = "qemu-img") -> bool: + """已存在 overlay 的 backing file 是否仍指向同一 base： + qemu-img info --output=json 的 backing-filename 与 + base 绝对路径比对；查不到/解析失败即 False（不复用， + 调用方删了重建，不猜）。""" + try: + result = runner([qemu, "info", + "--output=json", overlay_path]) + except Exception: + return False + if getattr(result, "returncode", 1) != 0: + return False + try: + import json as _json + info = _json.loads(getattr(result, "stdout", "") or "") + except ValueError: + return False + if not isinstance(info, dict): + return False + backing = info.get("backing-filename") + return isinstance(backing, str) \ + and os.path.abspath(backing) == os.path.abspath(base_path) + + +def create_overlay_disk(base_path: str, overlay_path: str, + runner: Callable[..., object] | None = None) -> str: + """建空 overlay（qemu-img create -f qcow2 -b + -F ）：base 只读，增量只写 overlay。 + 已存在且 backing 仍指向同一 base 即复用；backing 不符 + 即删了重建；overlay 是符号链接时先删链接本身（L1 同款， + 绝不跟随链接写目标）。建完以 qemu-img info 核对 backing， + 对不上即 ChvFailure disk_overlay_invalid（建了坏盘不 + 交给执行器去启动才炸）。runner 可注入（单测 fake）； + 不给即 subprocess 直调。""" + if not os.path.isfile(base_path) \ + or not os.access(base_path, os.R_OK): + raise ChvFailure( + "disk_overlay_unsupported", + "overlay 的 base 镜像不可读：%s" % base_path) + # runner 注入时（单测 fake）不查本机 qemu-img 是否存在， + # 命令名只作 fake 的分发标记；默认 runner 才查真实二进制。 + run = runner if runner is not None else _default_qemu_runner + qemu = "qemu-img" if runner is not None else _qemu_img_cmd() + if os.path.islink(overlay_path): + os.remove(overlay_path) + if os.path.exists(overlay_path): + if _overlay_backing_matches(overlay_path, base_path, run, + qemu=qemu): + return overlay_path + os.remove(overlay_path) + result = run([qemu, "create", "-f", "qcow2", + "-b", os.path.abspath(base_path), + "-F", _base_format(base_path), overlay_path]) + if getattr(result, "returncode", 1) != 0: + raise ChvFailure( + "disk_overlay_invalid", + "overlay 建盘失败：%s：%s" + % (overlay_path, getattr(result, "stderr", "") or "")) + if not os.path.exists(overlay_path): + raise ChvFailure( + "disk_overlay_invalid", + "overlay 建盘后文件不存在：%s" % overlay_path) + if not _overlay_backing_matches(overlay_path, base_path, run, + qemu=qemu): + raise ChvFailure( + "disk_overlay_invalid", + "overlay 建盘后 backing file 与 base 不符：%s" + % overlay_path) + return overlay_path + + +def provision_data_disk(state_dir: str, vm_name: str, + size_gb: float = DATA_DISK_GB, + runner: Callable[..., object] | None = None) -> str: + """建数据盘（空 qcow2 稀疏盘，虚大小默认 10 GB，实际 + 占用随写入增长）：已存在即复用（不 resize、不重建， + 旧盘里可能已有用户数据）；不存在即 qemu-img create。 + 失败即 ChvFailure disk_overlay_invalid（数据盘建不出 + 来时 overlay 模式拒绝启动，不让任务在无数据盘的 + VM 里把工作区写进系统盘——静默回落等于丢“独立持久” + 的保证）。""" + run = runner if runner is not None else _default_qemu_runner + qemu = "qemu-img" if runner is not None else _qemu_img_cmd() + path = data_disk_path(state_dir, vm_name) + os.makedirs(os.path.dirname(path), exist_ok=True) + if os.path.islink(path): + os.remove(path) + if os.path.exists(path): + return path + result = run([qemu, "create", "-f", "qcow2", + path, "%gG" % size_gb]) + if getattr(result, "returncode", 1) != 0: + raise ChvFailure( + "disk_overlay_invalid", + "数据盘建盘失败：%s：%s" + % (path, getattr(result, "stderr", "") or "")) + if not os.path.exists(path): + raise ChvFailure( + "disk_overlay_invalid", + "数据盘建盘后文件不存在：%s" % path) + return path + + +def delete_data_disk(state_dir: str, vm_name: str) -> bool: + """显式删数据盘（仅回收路径调）：返回 True 表示确实 + 删掉了一个文件。任务盘 GC、stop、hibernate 一律不调 + 它——用户数据不自动删，删须是显式回收决定。""" + path = data_disk_path(state_dir, vm_name) if os.path.islink(path) or os.path.exists(path): os.remove(path) return True return False +# -- tenant 归属标记（feat-chv-opt-landing，阶段 1）------------ +# 三处标记，缺一不可互相印证： +# ① 每 VM tenant.json：overlay/数据盘的归属（供给时写）； +# ② 快照登记条目 tenant 字段（record_snapshot 写）； +# ③ 持久 VM 登记 persistent/.json（预热/暂停时写）。 +# 归属校验只认“有标记且一致”；无标记是阶段 1 之前的旧 +# 登记（legacy），恢复时放行但发审计警告，不静默当作已验。 + +def _tenant_file_path(state_dir, vm_name): + return os.path.join(state_dir, vm_name, TENANT_FILE_NAME) + + +def write_vm_tenant(state_dir: str, vm_name: str, + tenant_id: str | None, + disk: str | None = None) -> str: + """写该 VM 的 tenant 归属标记（atomic 写 + fsync， + 与 snapshots.json 同款）；tenant_id 为空即不写（无 + 归属可标时不伪造标记，调用方按 legacy 处理）。""" + import time + if not tenant_id: + return "" + from cloud.chv.atomic_file import atomic_write_text + import json + path = _tenant_file_path(state_dir, vm_name) + os.makedirs(os.path.dirname(path), exist_ok=True) + atomic_write_text(path, json.dumps({ + "tenant_id": tenant_id, "vm": vm_name, + "disk": disk, "created_at": time.time()})) + return path + + +def read_vm_tenant(state_dir: str, vm_name: str) -> str | None: + """读该 VM 的 tenant 归属标记：无文件/损坏即 None + （legacy，未标记；调用方不得把 None 当成“某租户”）。""" + import json + try: + with open(_tenant_file_path(state_dir, vm_name), "r") as fh: + data = json.load(fh) + except (OSError, ValueError): + return None + if not isinstance(data, dict): + return None + tenant = data.get("tenant_id") + return tenant if tenant else None + + +def find_snapshot_entry(state_dir: str, snapshot_url: str) -> tuple[str | None, dict | None]: + """按 URL 在全部 VM 的快照登记里找条目：返回 + (vm_name, entry)；找不到即 (None, None)（未登记的 + 快照：归属无从核对，调用方按 legacy/未登记处理， + 不在此处捏造归属）。URL 精确比对，不做前缀猜测。""" + try: + entries = os.listdir(state_dir) + except OSError: + return None, None + for entry in entries: + if not entry.startswith("chv-"): + continue + for item in _load_snapshots(state_dir, entry): + if isinstance(item, dict) \ + and item.get("url") == snapshot_url: + return entry, item + return None, None + + +def _persistent_file_path(state_dir, tenant_id): + import re + safe = re.sub(r"[^A-Za-z0-9_.-]", "-", str(tenant_id or "default")) + return os.path.join(state_dir, PERSISTENT_DIR_NAME, + safe + ".json") + + +def write_persistent_vm(state_dir: str, tenant_id: str | None, record: dict) -> str: + """写持久 VM 登记（atomic 写）：record 至少含 + {vm, tenant_id, state: paused|running, origin: + prewarm|task, gitea_synced}；调用方负责字段完整。""" + import json + from cloud.chv.atomic_file import atomic_write_text + path = _persistent_file_path(state_dir, tenant_id) + os.makedirs(os.path.dirname(path), exist_ok=True) + atomic_write_text(path, json.dumps(record)) + return path + + +def read_persistent_vm(state_dir: str, tenant_id: str | None) -> dict | None: + """读持久 VM 登记：无文件/损坏/字段不全即 None + （无可用持久 VM，调用方回落冷启，不猜）。""" + import json + try: + with open(_persistent_file_path(state_dir, tenant_id), + "r") as fh: + data = json.load(fh) + except (OSError, ValueError): + return None + if not isinstance(data, dict) or not data.get("vm") \ + or data.get("tenant_id") != tenant_id: + return None + return data + + +def clear_persistent_vm(state_dir: str, tenant_id: str | None) -> bool: + """清持久 VM 登记（VM 已死/已回收时调）：返回 True + 表示确实清掉了一个登记。""" + path = _persistent_file_path(state_dir, tenant_id) + if os.path.exists(path): + os.remove(path) + return True + return False + + # -- 快照登记与上限（feat-chv-secfix，M3）------------------------- # 登记文件：state_dir//snapshots.json，内容为 # [{url, path（file:// 才有）, size, created_at}]。登记是 @@ -256,11 +586,16 @@ def check_snapshot_limits(state_dir: str, vm_name: str, max_count: int, return kept -def record_snapshot(state_dir: str, vm_name: str, snapshot_url: str, max_total_gb: float) -> tuple[list, list]: +def record_snapshot(state_dir: str, vm_name: str, snapshot_url: str, max_total_gb: float, + tenant_id: str | None = None) -> tuple[list, list]: """快照成功后登记；file:// 目标记实际大小。登记后若 总量超上限，按最旧优先删登记外的旧快照文件（总量硬 帽，删动作由调用方审计——本函数返回被删的旧条目）。 - 返回（全部条目，被总量帽删掉的旧条目）。""" + 返回（全部条目，被总量帽删掉的旧条目）。 + feat-chv-opt-landing：tenant_id 给出时条目带 tenant + 字段（归属标记）；不给时不写该字段——缺字段即 legacy + 未标记，恢复校验时按 legacy 处理，绝不事后补一个 + 猜出来的 tenant。""" import time path = None size = None @@ -268,8 +603,11 @@ def record_snapshot(state_dir: str, vm_name: str, snapshot_url: str, max_total_g path = snapshot_url[len("file://"):] size = _snapshot_path_size(path) entries = _load_snapshots(state_dir, vm_name) - entries.append({"url": snapshot_url, "path": path, - "size": size, "created_at": time.time()}) + entry = {"url": snapshot_url, "path": path, + "size": size, "created_at": time.time()} + if tenant_id: + entry["tenant"] = tenant_id + entries.append(entry) _save_snapshots(state_dir, vm_name, entries) removed = [] if max_total_gb is not None: diff --git a/cloud/chv/executor.py b/cloud/chv/executor.py index b51ed3e..3ec96ae 100644 --- a/cloud/chv/executor.py +++ b/cloud/chv/executor.py @@ -42,11 +42,22 @@ from cloud.chv.config import _parse_memory_bytes, build_chv_config from cloud.chv.disk import ( check_disk_quota, check_snapshot_limits, + clear_persistent_vm, + data_disk_path, + delete_data_disk, delete_task_disk, + find_snapshot_entry, has_recorded_snapshots, + overlay_disk_path, + provision_data_disk, provision_task_disk, + read_persistent_vm, + read_vm_tenant, record_snapshot, task_disk_path, + task_disk_paths, + write_persistent_vm, + write_vm_tenant, ) from cloud.chv.errors import ChvFailure from cloud.chv.hardening import ( @@ -211,6 +222,13 @@ def _ensure_snapshot_dir(snapshot_url, snapshot_path) -> None: "快照目录无法创建：%s：%s" % (snapshot_url, exc)) +def _disk_image_type(path: str) -> str: + """VmConfig 磁盘 image_type 枚举原名（feat-chv-opt-landing）： + .qcow2 后缀即 Qcow2，其余即 Raw（raw 无魔数、JSON 里写 + 小写 raw 会报 unknown variant，见 _vm_config_disks）。""" + return "Qcow2" if str(path).endswith(".qcow2") else "Raw" + + def _default_popen_launcher(cmd): """默认 VMM 启动器：独立会话子进程，输出丢弃（纯移动）。""" return subprocess.Popen( @@ -304,11 +322,21 @@ class CloudHypervisorExecutor(Executor): - 单任务单 VM：wake → 跑 → finally 关，同一任务不共享 VM； 每任务磁盘 copy 模式独立，租户无共享路径、无东西向互访； VM 内 root 只能影响本 VM。 - - 快速启动：WarmPool（cloud/chv/experimental_pool.py）预热干净快照 VM， - acquire 消耗、release 销毁，池空回落冷启动，绝不降级； - 池状态另落 state_dir/warm-pool.json registry，worker - 重启后载入的条目一律当孤儿、只能 reconcile_orphans() - 清理，绝不重新发给任务（见 orphans.py）。 + - 快速启动：持久 VM + 预热池（feat-chv-opt-landing， + 阶段 1，生产路径，见 pause/resume 与 + dispatch_task_persistent/prewarm_persistent_vm）： + 同一租户的任务共用一台持久 VM（overlay 系统盘 + + 独立数据盘），任务间隙 pause、到达时 resume； + CHV_WARM_POOL_SIZE>=1 时控制面常备 1 台已开机、已 + 完成 Gitea 首次同步的 paused 持久 VM，任务到达先 + 领它、结束暂停回池，池空回落冷启不报错；为 0（默认） + 时全部退化为冷启旧语义。WarmPool（experimental_pool.py） + 仍是纯实验（ADR-5/ADR-12），生产 dispatch 绝不查它， + 阶段 1 的池不经过它；池状态另落 + state_dir/persistent/.json 登记，worker + 重启后载入的条目须经 vm.info=Paused + socket 活性 + + tenant 归属三重核对才认领，否则只当孤儿清理， + 绝不直接发给任务（见 orphans.py 与 acquire_for_task）。 - 长任务中途死亡：_run_with_vm 在客内命令抛传输异常、 或非 goal 结果已 failed 且 VM 登记进程已死时，改判 interrupted（断点可续跑）而不是笼统 failed；VM 还活 @@ -603,6 +631,25 @@ class CloudHypervisorExecutor(Executor): return os.path.join(self.state_dir, self.vm_name(task_or_name) + ".sock") + def _tenant_id(self, task) -> str | None: + """归属租户 id（feat-chv-opt-landing）：任务 dict 的 + tenant_id > project_id > 单租户声明 + CHV_SINGLE_TENANT_PROJECT；三者皆无即 None（无归属 + 可标，快照/overlay 按 legacy 处理，不伪造）。""" + if isinstance(task, dict): + tenant = task.get("tenant_id") or task.get("project_id") + if tenant: + return str(tenant) + return getattr(self, "single_tenant_project", None) + + def persistent_vm_name(self, tenant_id: str | None) -> str: + """该租户的持久 VM 名：chv-persist-<安全字符租户> + （阶段 1：同一租户的连续任务共用这一台 VM，任务间 + 暂停不断电、不删盘；租户为空即 default 单租户）。""" + safe = re.sub(r"[^A-Za-z0-9_.-]", "-", + str(tenant_id or "default")) + return "chv-persist-" + safe + # -- 前置校验 ------------------------------------------------------ @staticmethod def _binary_ok(path): @@ -1138,22 +1185,30 @@ class CloudHypervisorExecutor(Executor): # -- API 模式：VmConfig / API 调用 / 就绪轮询 ----------------------- def _vm_config_disks(self, task_disk: str | None, - seed_disk: str | None) -> list: + seed_disk: str | None, + data_disk: str | None = None) -> list: """VmConfig 磁盘列表（to10 提取，含 L2 白名单校验）。 任务盘 + 可选 seed 只读盘（CIDATA ISO，cloud-init - NoCloud 数据源）；显式给的盘若不是基础镜像本身， - 须在 state_dir 内，否则 VmConfig 不能挂进宿主任意文件。 - image_type 必须显式给 Raw：CHV 54 的磁盘类型靠文件 + NoCloud 数据源）+ 可选数据盘（overlay 模式第二块盘， + 工作区/浏览器 profile 载体，客内按 LABEL=CHVDATA 挂 + 载到 /home/chv/workspace）；显式给的盘若不是基础 + 镜像本身，须在 state_dir 内，否则 VmConfig 不能挂 + 进宿主任意文件。 + image_type 必须显式给：CHV 54 的磁盘类型靠文件 魔数探测，raw 盘无魔数、探测失败即 vm.boot 400 "Image type required for disk"（2026-11-07 WSL 真机 CHV 54.0.0 实测）；值须是枚举原名 Raw（CLI 的小写 raw 是另一层映射，JSON 里写 raw 会报 unknown variant）； CLI 各轮一直传 image_type=raw（seed ISO 亦然，客内 - 仍以 /dev/vdb 读到 CIDATA）。""" + 仍以 /dev/vdb 读到 CIDATA）。 + feat-chv-opt-landing：.qcow2 后缀的盘（overlay 与 + 数据盘）image_type 为 Qcow2（qcow2 有魔数可探测， + 但与 raw 同款必须显式给枚举原名，不依赖探测）。""" # 旧文案逐字保留：任务盘“任务磁盘须在…”，seed 盘 # “seed 磁盘须在…”（label 本身含“磁盘”与空格）。 for path, label in ((task_disk, "任务磁盘"), - (seed_disk, "seed 磁盘")): + (seed_disk, "seed 磁盘"), + (data_disk, "数据磁盘")): if path and path != self.rootfs: if not os.path.realpath(path).startswith( os.path.realpath(self.state_dir) + os.sep): @@ -1161,13 +1216,18 @@ class CloudHypervisorExecutor(Executor): "disk_path_not_allowed", "%s须在 state_dir 内：%s" % (label, path)) disks: list = [{"path": task_disk or self.rootfs, - "image_type": "Raw"}] + "image_type": _disk_image_type( + task_disk or self.rootfs)}] if seed_disk: disks.append({"path": seed_disk, "readonly": True, "image_type": "Raw"}) + if data_disk: + disks.append({"path": data_disk, + "image_type": _disk_image_type(data_disk)}) 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, + data_disk: str | None = None) -> dict: """API 模式的 VmConfig（vm.create 请求体，纯函数）： cpus{boot_vcpus,max_vcpus}、memory{size: 字节}、 payload{kernel, cmdline}、disks[{path: 每任务磁盘}]、 @@ -1194,7 +1254,8 @@ class CloudHypervisorExecutor(Executor): "cpus": {"boot_vcpus": cpus, "max_vcpus": cpus}, "memory": {"size": _parse_memory_bytes(self.memory)}, "payload": payload, - "disks": self._vm_config_disks(task_disk, seed_disk), + "disks": self._vm_config_disks(task_disk, seed_disk, + data_disk), "rng": {"src": "/dev/urandom"}, } if self.net_json: @@ -1317,9 +1378,14 @@ class CloudHypervisorExecutor(Executor): def _check_disk_quota(self): """磁盘配额门（实现在 cloud/chv/disk.py，feat-chv-fix - 口径：总量 + 本次 + 宿主余量，镜像不可读不静默跳过）。""" + 口径：总量 + 本次 + 宿主余量，镜像不可读不静默跳过）。 + feat-chv-opt-landing：overlay 模式本次增量为 0（只 + 建空 overlay，不拷 base），配额仍按 base 大小硬帽。""" + increment = 0 if getattr(self, "disk_mode", "copy") \ + == "overlay" else None check_disk_quota(self.rootfs, self.disk_gb, - state_dir=self.state_dir) + state_dir=self.state_dir, + increment_bytes=increment) def _provision_task_disk(self, task): """每任务磁盘供给（实现在 cloud/chv/disk.py）： @@ -1329,10 +1395,29 @@ class CloudHypervisorExecutor(Executor): - reuse（仅单租户实验）：直接返回基础镜像路径，磁盘 不独立；preflight 拦截（须显式实验二次确认并审计 放行，见 preflight，feat-chv-fix）。 + - overlay（feat-chv-opt-landing，阶段 1）：在 + state_dir//disk.qcow2 建空 overlay，backing + file 指向只读 base；同时供给独立数据盘 + state_dir//data.qcow2（工作区/浏览器 profile + 载体，不进快照）；供给后写 tenant 归属标记 + tenant.json（reuse 不写——盘就是共享 base，标给 + 某租户是错的）。 返回每任务磁盘路径。""" - return provision_task_disk( + disk = provision_task_disk( self.state_dir, self.vm_name(task), self.rootfs, self.disk_mode) + if getattr(self, "disk_mode", "copy") == "overlay": + provision_data_disk(self.state_dir, self.vm_name(task)) + write_vm_tenant(self.state_dir, self.vm_name(task), + self._tenant_id(task), disk=disk) + return disk + + def _provision_data_disk_path(self, task): + """overlay 模式的数据盘路径（VmConfig 第三块盘）： + 非 overlay 模式即 None（不挂数据盘，行为与旧版一致）。""" + if getattr(self, "disk_mode", "copy") != "overlay": + return None + return data_disk_path(self.state_dir, self.vm_name(task)) def wait_api_ready(self, task: dict) -> bool: """等 vmm.ping 通：按次数轮询（api_timeout/1 秒取整， @@ -1768,7 +1853,9 @@ class CloudHypervisorExecutor(Executor): self._check_api_socket_permissions(socket_path) status, _ = self._api_call( socket_path, "PUT", "/vm.create", - self.build_vm_config(task_disk, seed_disk=seed_disk)) + self.build_vm_config( + task_disk, seed_disk=seed_disk, + data_disk=self._provision_data_disk_path(task))) if status not in (200, 204): raise ChvFailure( "vm_create_failed", @@ -1910,6 +1997,392 @@ class CloudHypervisorExecutor(Executor): "vm=%s" % self.vm_name(task_or_name)) return True + # -- 持久 VM 与预热池（feat-chv-opt-landing，阶段 1）----------- + # 模型（单租户灰度，CHV_WARM_POOL_SIZE=1 时启用）： + # 同一租户的任务共用一台持久 VM（chv-persist-， + # overlay 系统盘 + 独立数据盘）；任务间隙 pause 存盘， + # 下个任务 resume，不走完整开机。预热池 = 常备 1 台已 + # 开机、已完成 Gitea 首次同步、处于 paused 的持久 VM： + # 任务到达时先领它顶上，任务结束暂停回池（即补建回池）； + # 池空（无 paused 登记）自动回落冷启，不报错。 + # 与 experimental_pool.WarmPool 的区别（ADR-5 的缺口 + # 在此补上）：预热 VM 的盘从生下来就是该租户的持久盘 + #（overlay + 数据盘 + tenant 标记），任务直接跑在它 + # 自己的盘上，不存在“干净快照盘当任务盘用”的隔离击穿， + # 故 acquire 不需要热换盘。WarmPool 本身仍是实验模块， + # 生产路径不 import 它（ADR-12）；本节是生产路径。 + # 开关：CHV_WARM_POOL_SIZE（默认 0）——为 0 时本节全部 + # 入口退化为冷启旧语义（dispatch_task_persistent 直接 + # 调 dispatch_task；prewarm/refill 为 no-op），绝不硬 + # 编码改默认值偷换全局行为；阶段 1 验证用 + # CHV_WARM_POOL_SIZE=1。 + + def _vm_info_state(self, task_or_name) -> str | None: + """GET /vm.info 取 VM 状态（Running/Paused/...）： + 取不到/无 state 字段即 None（调用方按未知处理，绝 + 不猜它是 paused 还是 running）。""" + try: + status, body = self._api_call( + self.api_socket(task_or_name), "GET", "/vm.info") + except Exception: + return None + if status != 200 or not isinstance(body, dict): + return None + state = body.get("state") + return state if isinstance(state, str) else None + + def _persistent_task(self, tenant_id): + """持久 VM 的合成任务 dict：id 固定为 persist-<租户>， + 故 vm 名恒为 chv-persist-<租户>；gitea_sync 等任务 + 级同步变量由调用方经 extra_env 给，不进这里。""" + return {"id": "persist-%s" % (tenant_id or "default"), + "project_id": tenant_id} + + def _wait_sync_workspace(self, vm_task, sync_workspace) -> bool: + """等 Gitea 首次同步完成：轮询客内 test -d + /.git（clone 落盘的硬证据）；每轮 + 间隔 ready_interval、上限 ready_timeout；超时即 + False（调用方 fail-closed，不把未同步的 VM 当预热 + 机发给任务）。""" + attempts = max( + 1, int(self.ready_timeout / self.ready_interval)) + for attempt in range(attempts): + try: + result = self._call_runner(self.build_ssh_command( + vm_task, ["test", "-d", + sync_workspace + "/.git"])) + if getattr(result, "returncode", 1) == 0: + return True + except Exception: + pass + if attempt + 1 < attempts: + self.sleeper(self.ready_interval) + return False + + def _prewarm_build_new(self, tenant_id, extra_env, sync_workspace): + """预热新建段（prewarm_persistent_vm 的后半段）：冷启 + 一台持久 VM；sync_workspace 给出时等首次同步完成， + 等不到即关 VM 报 warm_pool_sync_incomplete（绝不把 + 未同步的 VM 记成预热机）。返回 (vm_task, gitea_synced)。""" + vm_task = self._persistent_task(tenant_id) + self.wake_vm(vm_task, extra_env=extra_env) + synced = False + if sync_workspace: + synced = self._wait_sync_workspace( + vm_task, sync_workspace) + if not synced: + try: + self.stop_vm(vm_task) + except Exception: + pass + raise ChvFailure( + "warm_pool_sync_incomplete", + "预热 VM %s 在 %s 秒内未完成 Gitea 首次同步" + "（%s/.git 未出现），已关 VM，不进池" + % (self.vm_name(vm_task), self.ready_timeout, + sync_workspace)) + else: + synced = bool(getattr(self, "gitea_sync_env", None)) + return vm_task, synced + + def prewarm_persistent_vm(self, tenant_id: str | None, + extra_env: dict | None = None, + sync_workspace: str | None = None) -> dict: + """预热池进池（幂等）：CHV_WARM_POOL_SIZE>=1 才生效， + 为 0 即 {pooled: False, reason: warm_pool_disabled}。 + 已有 paused 登记且 socket 活着即返回它（already-warm， + 不重建）；否则冷启一台持久 VM（overlay 盘 + Gitea + 同步 env），sync_workspace 给出时等首次同步完成 + （等不到即关 VM 报 warm_pool_sync_incomplete，绝不 + 把未同步的 VM 记成预热机），再 pause、写登记。 + 返回 {pooled, vm, mode, gitea_synced}。""" + if int(getattr(self, "warm_pool_size", 0)) <= 0: + return {"pooled": False, "vm": None, + "reason": "warm_pool_disabled"} + vm = self.persistent_vm_name(tenant_id) + record = read_persistent_vm(self.state_dir, tenant_id) + if record is not None and record.get("state") == "paused" \ + and record.get("vm") == vm \ + and os.path.exists(self.api_socket({"id": vm})): + return {"pooled": True, "vm": vm, + "mode": "already-warm", + "gitea_synced": bool( + record.get("gitea_synced"))} + vm_task, synced = self._prewarm_build_new( + tenant_id, extra_env, sync_workspace) + self.pause_vm(vm_task) + import time as _time + write_persistent_vm(self.state_dir, tenant_id, { + "vm": vm, "tenant_id": tenant_id, + "state": "paused", "origin": "prewarm", + "gitea_synced": synced, "updated_at": _time.time()}) + self._emit("warm_pool_prewarm", vm_task, + "vm=%s gitea_synced=%s" % (vm, synced)) + return {"pooled": True, "vm": vm, "mode": "prewarmed", + "gitea_synced": synced} + + def refill_warm_pool(self, tenant_id: str | None, + extra_env: dict | None = None, + sync_workspace: str | None = None) -> dict: + """后台补建回池（调用方自行起线程跑，本方法本身 + 同步）：池里已有 paused 登记即 no-op；池空即 + prewarm 一台。warm_pool_size==0 时 no-op。""" + record = read_persistent_vm(self.state_dir, tenant_id) + if record is not None and record.get("state") == "paused" \ + and os.path.exists(self.api_socket( + {"id": record.get("vm") or ""})): + return {"pooled": True, "vm": record.get("vm"), + "mode": "already-warm", + "gitea_synced": bool(record.get("gitea_synced"))} + return self.prewarm_persistent_vm( + tenant_id, extra_env=extra_env, + sync_workspace=sync_workspace) + + def _acquire_paused_candidate(self, tenant): + """认领的登记核对（acquire_for_task 的前半段）：登记 + 是 paused、socket 存在、tenant.json 归属一致才返回 + (vm, record)；任一不符即清登记、返回 (None, None) + （调用方回落冷启）。归属不符与 socket 消失是登记 + 本身已坏，清掉；状态未知（vm.info 取不到）不在此 + 判——那是认领时的第二道门，见 acquire_for_task。""" + record = read_persistent_vm(self.state_dir, tenant) + if record is None or record.get("state") != "paused": + return None, None + vm = record.get("vm") + if not vm: + return None, None + owner = read_vm_tenant(self.state_dir, vm) + if owner is not None and owner != tenant: + # 盘的归属与请求方不一致：这台不是你的 VM，绝不 + # resume，清登记回落冷启（legacy 未标记盘 owner + # 为 None 时放行，见 _check_snapshot_tenant 同款）。 + clear_persistent_vm(self.state_dir, tenant) + return None, None + if not os.path.exists(self.api_socket({"id": vm})): + clear_persistent_vm(self.state_dir, tenant) + return None, None + return vm, record + + def acquire_for_task(self, task: dict, + extra_env: dict | None = None) -> dict: + """取本租户的可用 VM（任务到达时的认领）： + - 持久登记是 paused、socket 活着、vm.info 确认 + Paused、tenant.json 归属一致 → resume + 等 SSH + 就绪，返回 {mode: resumed, vm, resume_seconds, + warm_hit}（warm_hit 表示这台来自 prewarm， + 不是上个任务暂停留下的）。 + - 其它情形（无登记 / VM 已死 / 状态未知 / + resume 或 SSH 就绪失败）→ 丢弃该登记（若有）， + 返回 {mode: cold, vm: None}，调用方回落冷启， + 不报错。warm_pool_size==0 时直接 cold。 + resume_seconds 只计 resume+SSH 就绪，不含任务 + 自身启动。""" + if int(getattr(self, "warm_pool_size", 0)) <= 0: + return {"mode": "cold", "vm": None} + tenant = self._tenant_id(task) + vm, record = self._acquire_paused_candidate(tenant) + if vm is None: + return {"mode": "cold", "vm": None} + state = self._vm_info_state({"id": vm}) + if state != "Paused": + # 状态未知或非 paused 时不猜：不是 paused 就不 + # 认领（running 的 VM resume 会错、死了的 VM + # resume 也会错），回落冷启，登记留给清理。 + return {"mode": "cold", "vm": None} + return self._acquire_resume_recorded(task, tenant, vm, record) + + def _acquire_resume_recorded(self, task, tenant, vm, record): + """认领的 resume 段（acquire_for_task 的后半段）： + resume + 等 SSH 就绪，计时写回登记（state=running） + 并返回 resumed；resume/就绪失败即清登记、回落 cold。 + resume_seconds 只计 resume+SSH 就绪。""" + import time as _time + t0 = _time.time() + try: + self.resume_vm({"id": vm}) + self.wait_guest_ready({"id": vm}) + except Exception as exc: + self._emit("warm_pool_acquire_failed", task, + "vm=%s resume/就绪失败，回落冷启：%s" + % (vm, exc), "warm_pool_acquire_failed") + clear_persistent_vm(self.state_dir, tenant) + return {"mode": "cold", "vm": None} + resume_seconds = _time.time() - t0 + write_persistent_vm(self.state_dir, tenant, { + "vm": vm, "tenant_id": tenant, + "state": "running", + "origin": record.get("origin") or "task", + "gitea_synced": bool(record.get("gitea_synced")), + "updated_at": _time.time()}) + self._emit("warm_pool_acquire", task, + "vm=%s mode=resumed resume_seconds=%.2f" + " warm_hit=%s" + % (vm, resume_seconds, + record.get("origin") == "prewarm")) + return {"mode": "resumed", "vm": vm, + "resume_seconds": resume_seconds, + "warm_hit": record.get("origin") == "prewarm"} + + def suspend_after_task(self, task_or_name: dict | str, + tenant_id: str | None = None) -> bool: + """任务间隙暂停：pause + 写持久登记（state=paused， + origin=task）。pause 失败或 VM 已死即清登记并返回 + False（调用方不得把死 VM 记成可用预热机）。""" + vm = self.vm_name(task_or_name) + tenant = tenant_id or self._tenant_id( + task_or_name if isinstance(task_or_name, dict) + else {"id": task_or_name}) + try: + self.pause_vm(task_or_name) + except Exception as exc: + self._emit("vm_suspend_failed", + task_or_name if isinstance(task_or_name, dict) + else {"id": task_or_name}, + "vm=%s 暂停失败，清持久登记：%s" % (vm, exc), + "vm_suspend_failed") + if tenant: + clear_persistent_vm(self.state_dir, tenant) + return False + if tenant: + import time as _time + record = read_persistent_vm(self.state_dir, tenant) or {} + write_persistent_vm(self.state_dir, tenant, { + "vm": vm, "tenant_id": tenant, + "state": "paused", "origin": "task", + "gitea_synced": bool(record.get("gitea_synced")), + "updated_at": _time.time()}) + return True + + def _run_guest_on_vm(self, task, vm, inner_args, goal, + extra_env=None, **kwargs): + """在指定 VM 上跑客内命令并做中途死亡分类（ + _run_with_persistent_vm 的执行段）：登记进程已死即 + interrupted（断点可续跑），VM 活着时的异常/失败原 + 样处理，不误判。返回结果 dict（未附 vm_mode，调用 + 方贴）。""" + try: + if goal: + result = self._run_guest_goal( + task, inner_args, extra_env=extra_env, + vm_name=vm, **kwargs) + else: + result = self._run_guest( + task, inner_args, extra_env=extra_env, + vm_name=vm, **kwargs) + except Exception as exc: + if self._vm_proc_dead({"id": vm}): + return self._vm_death_result( + task, goal, str(exc)[:500]) + raise + if not goal and result.get("status") == "failed" \ + and self._vm_proc_dead({"id": vm}): + result = dict(result) + result["status"] = "interrupted" + result["vm_dead"] = True + result["interrupted_reason"] = ( + "VM 中途死亡，任务未完成、断点可续跑：" + "客内命令已 failed 且登记的 VMM 进程已死") + return result + + def _run_with_persistent_vm(self, task, inner_args, goal=False, + extra_env=None, **kwargs): + """持久 VM 任务生命周期（阶段 1，与 _run_with_vm 对 + 照）：acquire（resumed）或冷启持久 VM → 跑客内命令 + → finally 暂停回池（suspend_after_task），不关机、 + 不删盘。warm_pool_size==0 时本方法不被调（调用方 + 直接走 _run_with_vm）。冷启路径：wake 持久 VM + （合成任务 persist-<租户>），任务本身的 host 任务 + 名仍按真实任务 id 派生（_run_guest 的 vm_name 参数 + 分离 VM 身份与任务身份）。中途死亡分类见 + _run_guest_on_vm（登记进程已死即 interrupted）。""" + tenant = self._tenant_id(task) + vm_task = self._persistent_task(tenant) + acquired = self.acquire_for_task(task, extra_env=extra_env) + vm = acquired.get("vm") or self.vm_name(vm_task) + mode = acquired["mode"] + resume_seconds = acquired.get("resume_seconds") + if mode == "cold": + self.wake_vm(vm_task, extra_env=extra_env) + try: + result = self._run_guest_on_vm( + task, vm, inner_args, goal, + extra_env=extra_env, **kwargs) + if isinstance(result, dict): + result = dict(result) + result["vm_mode"] = mode + if resume_seconds is not None: + result["resume_seconds"] = resume_seconds + return result + finally: + try: + if self._vm_proc_dead({"id": vm}): + # VM 已死：暂停无意义，清登记，免得下次 + # acquire 把死 VM 当可用预热机。 + if tenant: + clear_persistent_vm(self.state_dir, tenant) + else: + self.suspend_after_task({"id": vm}, + tenant_id=tenant) + except Exception: + pass + + def dispatch_task_persistent(self, task: dict, + extra_env: dict | None = None, + model_provider: str | None = None) -> dict: + """持久 VM 任务入口（阶段 1）：warm_pool_size>=1 时 + 走 _run_with_persistent_vm（resume/冷启持久 VM → + 跑 → 暂停回池）；==0 时退化为 dispatch_task（冷启 + → 跑 → 关机），与旧行为一致。结果 dict 附 vm_mode + （resumed/cold）与 resume_seconds（resumed 时）。""" + if not isinstance(task, dict) or task.get("id") is None: + raise ValueError("任务缺少 id") + if not task.get("prompt"): + raise ValueError("任务缺少 prompt") + if int(getattr(self, "warm_pool_size", 0)) <= 0: + return self.dispatch_task( + task, extra_env=extra_env, + model_provider=model_provider) + args = ["dispatch", "--name", self._host_name(task), + "--task", task["prompt"]] + repo = task.get("repo") + if repo: + args += ["--repo", repo] + return self._run_with_persistent_vm( + task, args, extra_env=extra_env, model=task.get("model"), + effort=task.get("effort"), approval=task.get("approval"), + model_provider=model_provider, + sandbox=("read-only" if task.get("plan") else None)) + + def recycle_persistent_vm(self, task_or_name: dict | str, + tenant_id: str | None = None) -> dict: + """显式回收持久 VM（唯一删用户盘的入口）：先 shutdown + （stop_vm，keep_disk=False），再删 overlay 与数据盘、 + 清 tenant 标记与持久登记。返回 {vm, state: recycled, + deleted_disks}。与 recycle_vm 的区别：后者只关 VM + 不删持久盘（持久盘须显式回收才删，用户数据不静默丢）。""" + vm = self.vm_name(task_or_name) + tenant = tenant_id or self._tenant_id( + task_or_name if isinstance(task_or_name, dict) + else {"id": task_or_name}) + self.shutdown_vm(task_or_name) + deleted = [] + if delete_task_disk(self.state_dir, vm): + deleted.append("task") + if delete_data_disk(self.state_dir, vm): + deleted.append("data") + _tenant_path = os.path.join(self.state_dir, vm, + "tenant.json") + if os.path.exists(_tenant_path): + os.remove(_tenant_path) + if tenant: + clear_persistent_vm(self.state_dir, tenant) + self._emit("vm_persistent_recycled", + task_or_name if isinstance(task_or_name, dict) + else {"id": task_or_name}, + "vm=%s deleted_disks=%s" % (vm, deleted)) + return {"vm": vm, "state": "recycled", + "deleted_disks": deleted} + def snapshot_vm(self, task_or_name: dict | str, snapshot_url: str) -> str: """给（已暂停的）VM 做快照：ch-remote snapshot <裸 URL>。 注意：snapshot 的参数是裸 URL，不是 source_url=... @@ -1960,21 +2433,70 @@ class CloudHypervisorExecutor(Executor): def _record_snapshot_and_emit(self, task_or_name, vm_name, snapshot_url): """快照成功后登记 + 总量超帽时最旧优先剪枝（删动作 - 审计）+ vm_snapshot 审计，返回 snapshot_url。""" + 审计）+ vm_snapshot 审计，返回 snapshot_url。 + feat-chv-opt-landing：登记条目带 tenant 字段（归属 + 标记，见 disk.record_snapshot）；审计 detail 同样 + 带 tenant，便于事后核对谁的快照。""" task = task_or_name if isinstance(task_or_name, dict) \ else {"id": task_or_name} + tenant = self._tenant_id(task) _entries, _removed = record_snapshot( self.state_dir, vm_name, snapshot_url, - getattr(self, "snapshot_max_total_gb", 20.0)) + getattr(self, "snapshot_max_total_gb", 20.0), + tenant_id=tenant) for _old in _removed: self._emit("vm_snapshot_pruned", task, "vm=%s pruned snapshot=%s（总量超帽，最旧优先）" % (vm_name, _old.get("url"))) self._emit("vm_snapshot", task, - "vm=%s snapshot=%s" - % (self.vm_name(task_or_name), snapshot_url)) + "vm=%s snapshot=%s tenant=%s" + % (self.vm_name(task_or_name), snapshot_url, tenant)) return snapshot_url + def _check_snapshot_tenant(self, task, snapshot_url) -> None: + """restore 的归属校验（feat-chv-opt-landing，阶段 1 + 必须补的代码项）：按 URL 在全部 VM 的快照登记里找 + 条目；条目带 tenant 且与请求方 tenant 不一致时， + 直接拒绝（ChvFailure snapshot_tenant_mismatch）并 + 发 vm_restore_tenant_denied 审计——A 的快照绝不能 + 恢复到 B 名下。条目无 tenant 是阶段 1 之前的 legacy + 登记：放行但发 vm_restore_tenant_unmarked 审计警告 + （单租户灰度下的旧快照，不静默当作已验）；条目根本 + 未登记：同样放行 + 警告（未登记快照无从核对归属， + 登记才是归属的唯一账本——生产须要求快照必登记， + 见报告残留）。请求方无 tenant（None）而条目有 + tenant 时同样拒绝（无法证明归属即 fail-closed）。""" + _owner_vm, entry = find_snapshot_entry( + self.state_dir, snapshot_url) + if entry is None: + self._emit("vm_restore_tenant_unmarked", task, + "vm=%s snapshot=%s 未在登记中找到，归属" + "无法核对（未登记快照，放行并警告）" + % (self.vm_name(task), snapshot_url), + "snapshot_tenant_unmarked") + return + marked = entry.get("tenant") + if marked is None: + self._emit("vm_restore_tenant_unmarked", task, + "vm=%s snapshot=%s 登记无 tenant 字段" + "（legacy 旧快照，放行并警告）" + % (self.vm_name(task), snapshot_url), + "snapshot_tenant_unmarked") + return + requester = self._tenant_id(task) + if requester != marked: + self._emit("vm_restore_tenant_denied", task, + "vm=%s snapshot=%s 归属 tenant=%s，请求方" + " tenant=%s，拒绝跨用户恢复" + % (self.vm_name(task), snapshot_url, + marked, requester), + "snapshot_tenant_mismatch") + raise ChvFailure( + "snapshot_tenant_mismatch", + "快照归属 tenant=%s，与请求方 tenant=%s 不一致，" + "拒绝跨用户恢复：%s" % (marked, requester, + snapshot_url)) + def _validate_restore(self, task, snapshot_url): """restore 前置校验：返回（归一化后的 snapshot_url， 归一化后的 net 条目）。 @@ -2002,6 +2524,7 @@ class CloudHypervisorExecutor(Executor): snapshot_url, self.state_dir, snapshot_dir=getattr(self, "snapshot_dir", None), allow_remote=getattr(self, "allow_remote_snapshot", False)) + self._check_snapshot_tenant(task, snapshot_url) self._check_snapshot_disk(snapshot_url) self.preflight() self.resolve_ssh_host() @@ -2010,11 +2533,16 @@ class CloudHypervisorExecutor(Executor): "host_key_not_pinned", "provisioned 模式下 restore 缺 known_hosts，拒绝 " "keyscan 回落：VM %s" % self.vm_name(task)) - net_entries = [] + return snapshot_url, self._validate_restore_net(task) + + def _validate_restore_net(self, task): + """restore 的 net 隔离段（_validate_restore 的尾段）： + 与 wake 的 _wake_prechecks 同款，冲突在起 VMM 前 + 拒绝（net_tap_conflict）；net 未配即空列表。""" if not self.net_json: - return snapshot_url, net_entries + return [] try: - net_entries = validate_net_entries( + return validate_net_entries( parse_net_entries(self.net_json), self._registered_net_entries( exclude_vm=self.vm_name(task))) @@ -2024,7 +2552,6 @@ class CloudHypervisorExecutor(Executor): % (self.vm_name(task), exc.kind), exc.kind) raise - return snapshot_url, net_entries def _check_snapshot_disk(self, snapshot_url): """restore 前的快照磁盘预检（feat-chv-v3，hibernate @@ -2199,6 +2726,49 @@ class CloudHypervisorExecutor(Executor): if os.path.exists(_p): os.remove(_p) + def _persistent_record_for_vm(self, name): + """该 VM 是否被某租户的持久登记认领：读 tenant.json + 得租户，再读该租户的持久登记、核对 vm 名一致。 + 返回登记 dict 或 None（未认领：冷路径 VM 或旧 VM， + stop/GC 按旧语义处理）。""" + tenant = read_vm_tenant(self.state_dir, name) + if not tenant: + return None + record = read_persistent_vm(self.state_dir, tenant) + if record is not None and record.get("vm") == name: + return record + return None + + def _overlay_disk_protected(self, name) -> bool: + """该 VM 的 overlay 盘是否受保护（stop/GC 不删）： + 有持久登记认领，或有登记快照（hibernate 的 restore + 输入）。copy 模式的 disk.raw 不受此保护（旧语义， + _cleanup_vm_files 只在 overlay 模式下查它——见下）， + 但本方法本身不分模式，调用方按模式决定是否查。 + 数据盘 data.qcow2 永远不受 GC/stop 删除（另有显式 + 回收入口），本方法不覆盖它。""" + if self._persistent_record_for_vm(name) is not None: + return True + return has_recorded_snapshots(self.state_dir, name) + + def _cleanup_vm_task_disk(self, name, task, keep_disk) -> bool: + """stop 的任务盘删除块（_cleanup_vm_files 的中段）： + keep_disk 或 overlay 受保护盘（持久登记认领/有登记 + 快照）即跳过；删掉返回 False（无失败），删失败记 + 审计并返回 True（脏标记由调用方写）。""" + _protected = getattr(self, "disk_mode", "copy") == "overlay" \ + and self._overlay_disk_protected(name) + try: + if not keep_disk and not _protected \ + and delete_task_disk(self.state_dir, name): + self._emit("vm_disk_deleted", task, + "vm=%s disk deleted on stop" % name) + except Exception as _disk_exc: + self._emit("vm_disk_delete_failed", task, + "vm=%s error=%s" % (name, _disk_exc)) + return True + return False + def _cleanup_vm_files(self, name, task, keep_disk=False): """stop 的三块文件清理：seed/私钥、任务磁盘、net 登记。 @@ -2209,7 +2779,12 @@ class CloudHypervisorExecutor(Executor): 与 seed.iso，任一删了 restore 即失败——见 hibernate_vm 的评估注释与 V-7/D2；seed 目录与宿主 私钥仍删，config.json 不引用它们，私钥明文已嵌进 - seed.iso、宿主侧不留）。""" + seed.iso、宿主侧不留）。 + feat-chv-opt-landing：overlay 模式下，任务盘是持久 + 用户盘——该 VM 有持久登记认领或有登记快照时，stop + 不删 overlay（删了等于销毁用户 VM / 毁掉快照输入）； + 删盘只走显式回收（recycle_persistent_vm）。无认领 + 又无快照的 overlay VM（冷路径残留）仍按旧语义删。""" from cloud.chv.hostkeys import dirty_marker_path # noqa: F401 failed = False try: @@ -2218,14 +2793,11 @@ class CloudHypervisorExecutor(Executor): failed = True self._emit("vm_key_cleanup_failed", task, "vm=%s error=%s" % (name, _cleanup_exc)) - try: - if not keep_disk and delete_task_disk(self.state_dir, name): - self._emit("vm_disk_deleted", task, - "vm=%s disk deleted on stop" % name) - except Exception as _disk_exc: + # feat-chv-opt-landing：overlay 模式下受保护的盘（持久 + # 登记认领或有登记快照）stop 不删；copy 模式一律旧语义。 + # 见 _cleanup_vm_task_disk。 + if self._cleanup_vm_task_disk(name, task, keep_disk): failed = True - self._emit("vm_disk_delete_failed", task, - "vm=%s error=%s" % (name, _disk_exc)) try: _net_path = os.path.join(self.state_dir, name, "net.json") if os.path.exists(_net_path): @@ -2321,12 +2893,15 @@ class CloudHypervisorExecutor(Executor): return had def gc_stale_disks(self) -> dict: - """残留盘 GC（H3）：扫描 state_dir/*/disk.raw，VM - 非 live（无 api-socket 文件且无登记活进程）即删盘 - 并审计；删失败审计 vm_disk_delete_failed 并计入 - failed。返回 {deleted, failed}。 + """残留盘 GC（H3）：扫描 state_dir/*/disk.raw 与 + disk.qcow2，VM 非 live（无 api-socket 文件且无登记 + 活进程）即删盘并审计；删失败审计 vm_disk_delete_failed + 并计入 failed。返回 {deleted, failed}。 feat-chv-v3：有登记快照的 VM 不 GC 盘——那是休眠 - VM 的盘，restore 还要读它（见 hibernate_vm）。""" + VM 的盘，restore 还要读它（见 hibernate_vm）。 + feat-chv-opt-landing：有持久登记认领的 VM 同样不 + GC（那是用户的持久盘）；数据盘 data.qcow2 永不进 + GC（用户数据，只走显式回收 recycle_persistent_vm）。""" deleted: list = [] failed: list = [] try: @@ -2336,11 +2911,12 @@ class CloudHypervisorExecutor(Executor): for entry in entries: if not entry.startswith("chv-"): continue - disk = task_disk_path(self.state_dir, entry) - if not (os.path.islink(disk) or os.path.exists(disk)): + if not task_disk_paths(self.state_dir, entry): continue if has_recorded_snapshots(self.state_dir, entry): continue + if self._persistent_record_for_vm(entry) is not None: + continue live = os.path.exists( os.path.join(self.state_dir, entry + ".sock")) with self._vms_lock: @@ -2523,15 +3099,20 @@ class CloudHypervisorExecutor(Executor): return cleaned # -- 客内执行 ------------------------------------------------------ - def _run_guest(self, task, inner_args, extra_env=None, **kwargs): + def _run_guest(self, task, inner_args, extra_env=None, vm_name=None, **kwargs): """SSH 调客内 host.py，返回 Local/Docker 同款结果 dict （executor 字段改写为 cloud-hypervisor，并附 vm 名与 api-socket）。runner 异常即原样传播，由 worker 侧统一 写回 failed——本类不在这里把异常吞成假成功。 - extra_env（P1-1）：经 SSH env 前缀进客机环境。""" + extra_env（P1-1）：经 SSH env 前缀进客机环境。 + feat-chv-opt-landing：vm_name 给出时 SSH 目标是该 + VM（持久 VM 路径：任务 id ≠ VM 名，host 任务名仍按 + 任务 id 派生）；不给时目标即 vm_name(task)，与旧 + 行为一致。""" self._validate() + vm_task = {"id": vm_name} if vm_name else task ssh_cmd = self.build_ssh_command( - task, self._guest_host_command(*inner_args, **kwargs), + vm_task, self._guest_host_command(*inner_args, **kwargs), extra_env=extra_env) result = self._call_ssh_with_transient_retry( task, ssh_cmd, timeout=self.task_timeout) @@ -2540,8 +3121,8 @@ class CloudHypervisorExecutor(Executor): parsed = _parse_host_result(task["id"], SimpleNamespaceShim(result)) parsed["executor"] = "cloud-hypervisor" - parsed["vm"] = self.vm_name(task) - parsed["api_socket"] = self.api_socket(task) + parsed["vm"] = self.vm_name(vm_task) + parsed["api_socket"] = self.api_socket(vm_task) return parsed def _goal_result(self, task_id, result): @@ -2562,10 +3143,11 @@ class CloudHypervisorExecutor(Executor): "cloud-hypervisor") def _run_guest_goal(self, task, inner_args, extra_env=None, - **kwargs): + vm_name=None, **kwargs): self._validate() + vm_task = {"id": vm_name} if vm_name else task ssh_cmd = self.build_ssh_command( - task, self._guest_host_command(*inner_args, **kwargs), + vm_task, self._guest_host_command(*inner_args, **kwargs), extra_env=extra_env) return self._goal_result( task["id"], diff --git a/cloud/tests/test_chv_opt_stage1.py b/cloud/tests/test_chv_opt_stage1.py new file mode 100644 index 0000000..dc3418d --- /dev/null +++ b/cloud/tests/test_chv_opt_stage1.py @@ -0,0 +1,657 @@ +"""Cloud Hypervisor 优化落地阶段 1（feat-chv-opt-landing）测试。 + +只用标准库与 mock runner/launcher/api_client，绝不启动真实 +VM（与 test_chv_optimize.py 同款边界）。覆盖： +- overlay 分层：qemu-img backing 建盘/复用/重建、配额增量 0、 + VmConfig image_type Qcow2、数据盘独立供给与永不 GC/随 stop 删 +- tenant 归属：快照登记 tenant 字段、overlay tenant.json、 + 持久登记、跨用户恢复拒绝（_validate_restore） +- 持久 VM 与预热池：prewarm 幂等、acquire resume/warm_hit、 + 池空回落冷启、warm_pool_size=0 退化旧语义、暂停回池、 + dispatch_task_persistent 两连任务、显式回收删盘 +""" + +import json +import os +import sys +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +try: + from cloud.chv import ChvFailure + from cloud.chv import CloudHypervisorExecutor + from cloud.chv import disk as chv_disk + from cloud.chv.disk import ( + clear_persistent_vm, + create_overlay_disk, + delete_data_disk, + delete_task_disk, + find_snapshot_entry, + provision_data_disk, + provision_task_disk, + read_persistent_vm, + read_vm_tenant, + record_snapshot, + write_persistent_vm, + write_vm_tenant, + ) +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 import disk as chv_disk + from cloud.chv.disk import ( + clear_persistent_vm, + create_overlay_disk, + delete_data_disk, + delete_task_disk, + find_snapshot_entry, + provision_data_disk, + provision_task_disk, + read_persistent_vm, + read_vm_tenant, + record_snapshot, + write_persistent_vm, + write_vm_tenant, + ) + + +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 _fake_qemu_runner_factory(files_to_create): + """fake qemu-img：create 写空文件，info 回 backing JSON。 + + files_to_create：overlay 路径 -> backing 绝对路径（info + 应答用）；create 时按命令写文件并登记 backing。""" + state = {"backing": dict(files_to_create)} + + def runner(cmd, **kwargs): + if cmd[1] == "create" and cmd[2] == "-f": + # overlay：... -b -F （目标在末尾）； + # 数据盘：... G（目标在倒数第二）。 + target = cmd[-1] if "-b" in cmd else cmd[4] + with open(target, "wb") as fh: + fh.write(b"QFI\xfb fake-qcow2") + if "-b" in cmd: + state["backing"][target] = cmd[cmd.index("-b") + 1] + return FakeResult(0, "", "") + if cmd[1] == "info": + target = cmd[-1] + backing = state["backing"].get(target) + payload = {"format": "qcow2"} + if backing: + payload["backing-filename"] = backing + return FakeResult(0, json.dumps(payload), "") + return FakeResult(0, "", "") + + runner.state = state + return runner + + +class ChvOpt1Case(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"} + 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": + if cmd[-1] == "true": + return FakeResult(0, "", "") + if "dispatch" in cmd: + return FakeResult(0, json.dumps( + {"status": "completed", "text": "ok-dispatch"}), "") + return FakeResult(0, "", "") + 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 _task(self, task_id="abc123", project="proj1"): + return {"id": task_id, "prompt": "做一个加法函数", + "project_id": project} + + +# -- overlay 分层 ------------------------------------------------------------ +class OverlayDiskTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.dir = self.tmp.name + self.base = os.path.join(self.dir, "base.qcow2") + with open(self.base, "wb") as fh: + fh.write(b"R" * 100) + self.state = os.path.join(self.dir, "state") + + def tearDown(self): + self.tmp.cleanup() + + def test_create_overlay_backing_and_reuse(self): + runner = _fake_qemu_runner_factory({}) + overlay = os.path.join(self.dir, "ov.qcow2") + out = create_overlay_disk(self.base, overlay, runner=runner) + self.assertEqual(out, overlay) + self.assertTrue(os.path.exists(overlay)) + create_args = runner.state["backing"] + self.assertEqual(create_args[overlay], + os.path.abspath(self.base)) + # 第二次：backing 一致即复用，不重建（文件 mtime 不变）。 + before = os.stat(overlay).st_mtime_ns + out2 = create_overlay_disk(self.base, overlay, runner=runner) + self.assertEqual(out2, overlay) + self.assertEqual(os.stat(overlay).st_mtime_ns, before) + + def test_create_overlay_backing_mismatch_rebuilds(self): + other = os.path.join(self.dir, "other.qcow2") + with open(other, "wb") as fh: + fh.write(b"O" * 100) + runner = _fake_qemu_runner_factory({}) + overlay = os.path.join(self.dir, "ov.qcow2") + create_overlay_disk(self.base, overlay, runner=runner) + # backing 被换成别的 base：下次建盘须删了重建。 + runner.state["backing"][overlay] = os.path.abspath(other) + out = create_overlay_disk(self.base, overlay, runner=runner) + self.assertEqual(out, overlay) + self.assertEqual(runner.state["backing"][overlay], + os.path.abspath(self.base)) + + def test_create_overlay_base_missing_fails(self): + runner = _fake_qemu_runner_factory({}) + with self.assertRaises(ChvFailure) as ctx: + create_overlay_disk(os.path.join(self.dir, "nope.qcow2"), + os.path.join(self.dir, "ov.qcow2"), + runner=runner) + self.assertEqual(ctx.exception.kind, + "disk_overlay_unsupported") + + def test_create_overlay_unknown_base_suffix_fails(self): + weird = os.path.join(self.dir, "base.img") + with open(weird, "wb") as fh: + fh.write(b"R" * 10) + runner = _fake_qemu_runner_factory({}) + with self.assertRaises(ChvFailure) as ctx: + create_overlay_disk(weird, + os.path.join(self.dir, "ov.qcow2"), + runner=runner) + self.assertEqual(ctx.exception.kind, + "disk_overlay_unsupported") + + def test_provision_overlay_and_data_disk(self): + runner = _fake_qemu_runner_factory({}) + disk = provision_task_disk(self.state, "chv-a", self.base, + "overlay", runner=runner) + self.assertEqual(disk, + os.path.join(self.state, "chv-a", "disk.qcow2")) + self.assertTrue(os.path.exists(disk)) + data = provision_data_disk(self.state, "chv-a", runner=runner) + self.assertEqual(data, + os.path.join(self.state, "chv-a", "data.qcow2")) + self.assertTrue(os.path.exists(data)) + # 数据盘已存在即复用，不重建。 + data2 = provision_data_disk(self.state, "chv-a", runner=runner) + self.assertEqual(data2, data) + + def test_delete_task_disk_removes_both_but_not_data(self): + runner = _fake_qemu_runner_factory({}) + provision_task_disk(self.state, "chv-a", self.base, + "overlay", runner=runner) + provision_data_disk(self.state, "chv-a", runner=runner) + raw = os.path.join(self.state, "chv-a", "disk.raw") + with open(raw, "wb") as fh: + fh.write(b"R") + self.assertTrue(delete_task_disk(self.state, "chv-a")) + self.assertFalse(os.path.exists( + os.path.join(self.state, "chv-a", "disk.qcow2"))) + self.assertFalse(os.path.exists(raw)) + # 数据盘还在：删任务盘绝不删用户数据。 + self.assertTrue(os.path.exists( + os.path.join(self.state, "chv-a", "data.qcow2"))) + self.assertTrue(delete_data_disk(self.state, "chv-a")) + self.assertFalse(delete_task_disk(self.state, "chv-a")) + + def test_quota_overlay_increment_zero(self): + # state_dir 已用逼近配额：copy 增量（base 大小）超帽， + # overlay 增量 0 不超帽。 + os.makedirs(self.state, exist_ok=True) + with open(os.path.join(self.state, "used.bin"), "wb") as fh: + fh.write(b"U" * 1000) + disk_gb = 1000 / (1024 ** 3) # 1000 字节配额 + with self.assertRaises(ChvFailure) as ctx: + chv_disk.check_disk_quota(self.base, disk_gb, + state_dir=self.state) + self.assertEqual(ctx.exception.kind, "disk_quota_exceeded") + chv_disk.check_disk_quota(self.base, disk_gb, + state_dir=self.state, + increment_bytes=0) + + def test_disk_mode_overlay_accepted_in_config(self): + from cloud.chv.config import build_chv_config + cfg = build_chv_config({}, {"CHV_DISK_MODE": "overlay"}) + self.assertEqual(cfg.disk_mode, "overlay") + with self.assertRaises(ValueError): + build_chv_config({}, {"CHV_DISK_MODE": "bogus"}) + + +class VmConfigOverlayTest(ChvOpt1Case): + def test_qcow2_disks_are_qcow2_image_type(self): + ex = self.make_exec(env_extra={"CHV_DISK_MODE": "overlay"}) + cfg = ex.build_vm_config( + os.path.join(self.state_dir, "chv-a", "disk.qcow2"), + seed_disk=None, + data_disk=os.path.join(self.state_dir, "chv-a", + "data.qcow2")) + self.assertEqual(cfg["disks"][0]["image_type"], "Qcow2") + self.assertEqual(cfg["disks"][1]["image_type"], "Qcow2") + self.assertEqual(len(cfg["disks"]), 2) + + def test_raw_disks_stay_raw(self): + ex = self.make_exec() + cfg = ex.build_vm_config( + os.path.join(self.state_dir, "chv-a", "disk.raw")) + self.assertEqual(cfg["disks"][0]["image_type"], "Raw") + + +# -- tenant 归属 -------------------------------------------------------------- +class TenantMarkingTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.state = os.path.join(self.tmp.name, "state") + + def tearDown(self): + self.tmp.cleanup() + + def test_snapshot_entry_carries_tenant(self): + url = "file:///snap/a" + entries, _ = record_snapshot(self.state, "chv-a", url, 20.0, + tenant_id="tenant-a") + self.assertEqual(entries[0]["tenant"], "tenant-a") + entries2, _ = record_snapshot(self.state, "chv-b", url, 20.0) + self.assertNotIn("tenant", entries2[0]) + + def test_find_snapshot_entry(self): + url = "file:///snap/a" + record_snapshot(self.state, "chv-a", url, 20.0, + tenant_id="tenant-a") + vm, entry = find_snapshot_entry(self.state, url) + self.assertEqual(vm, "chv-a") + self.assertEqual(entry["tenant"], "tenant-a") + self.assertEqual(find_snapshot_entry(self.state, + "file:///snap/nope"), + (None, None)) + + def test_vm_tenant_roundtrip_and_corrupt(self): + path = write_vm_tenant(self.state, "chv-a", "tenant-a", + disk="/x/disk.qcow2") + self.assertTrue(os.path.exists(path)) + self.assertEqual(read_vm_tenant(self.state, "chv-a"), + "tenant-a") + self.assertIsNone(read_vm_tenant(self.state, "chv-nope")) + with open(os.path.join(self.state, "chv-a", + "tenant.json"), "w") as fh: + fh.write("{broken") + self.assertIsNone(read_vm_tenant(self.state, "chv-a")) + + def test_persistent_record_roundtrip(self): + write_persistent_vm(self.state, "tenant-a", { + "vm": "chv-persist-tenant-a", "tenant_id": "tenant-a", + "state": "paused", "origin": "prewarm", + "gitea_synced": True}) + rec = read_persistent_vm(self.state, "tenant-a") + self.assertEqual(rec["vm"], "chv-persist-tenant-a") + self.assertEqual(rec["state"], "paused") + # 租户不符的登记读不到（不把 A 的登记当成 B 的）。 + self.assertIsNone(read_persistent_vm(self.state, "tenant-b")) + self.assertTrue(clear_persistent_vm(self.state, "tenant-a")) + self.assertIsNone(read_persistent_vm(self.state, "tenant-a")) + self.assertFalse(clear_persistent_vm(self.state, "tenant-a")) + + +class CrossTenantRestoreTest(ChvOpt1Case): + def _record_snapshot(self, vm, url, tenant): + os.makedirs(os.path.join(self.state_dir, vm), exist_ok=True) + record_snapshot(self.state_dir, vm, url, 20.0, + tenant_id=tenant) + + def _snap_url(self, vm, name="snap1"): + return "file://" + os.path.join(self.state_dir, vm, name) + + def test_cross_tenant_restore_refused(self): + url = self._snap_url("chv-a") + self._record_snapshot("chv-a", url, "tenant-a") + ex = self.make_exec() + events = [] + ex.audit_callback = lambda e, t, d, k: events.append((e, k)) + task_b = {"id": "task-b", "project_id": "tenant-b"} + with self.assertRaises(ChvFailure) as ctx: + ex._validate_restore(task_b, url) + self.assertEqual(ctx.exception.kind, + "snapshot_tenant_mismatch") + self.assertTrue(any( + e == "vm_restore_tenant_denied" for e, _ in events)) + + def test_same_tenant_restore_passes_validation(self): + url = self._snap_url("chv-a") + self._record_snapshot("chv-a", url, "tenant-a") + ex = self.make_exec() + snapshot_url, _net = ex._validate_restore( + {"id": "task-a2", "project_id": "tenant-a"}, url) + self.assertEqual(snapshot_url, url) + + def test_legacy_unmarked_snapshot_warns_but_passes(self): + url = self._snap_url("chv-a", "old") + self._record_snapshot("chv-a", url, None) + ex = self.make_exec() + events = [] + ex.audit_callback = lambda e, t, d, k: events.append((e, k)) + snapshot_url, _net = ex._validate_restore( + {"id": "task-a2", "project_id": "tenant-a"}, url) + self.assertEqual(snapshot_url, url) + self.assertTrue(any( + e == "vm_restore_tenant_unmarked" for e, _ in events)) + + def test_snapshot_vm_records_tenant(self): + ex = self.make_exec() + vm = "chv-snap1" + snap_dir = os.path.join(self.state_dir, vm, "snap1") + os.makedirs(snap_dir, exist_ok=True) + with open(os.path.join(snap_dir, "config.json"), "w") as fh: + fh.write("{}") + url = "file://" + snap_dir + out = ex.snapshot_vm({"id": "snap1", "project_id": "tenant-a"}, + url) + self.assertEqual(out, url) + _vm, entry = find_snapshot_entry(self.state_dir, url) + self.assertEqual(entry["tenant"], "tenant-a") + + +# -- 持久 VM 与预热池 ---------------------------------------------------------- +class PersistentPoolTest(ChvOpt1Case): + def _patch_overlay_provision(self, ex): + """overlay 建盘改走 fake qemu-img（本机无 qemu-img）。""" + runner = _fake_qemu_runner_factory({}) + + def fake_provision(state_dir, vm_name, rootfs, disk_mode, + runner=None): + if disk_mode != "overlay": + return chv_disk.provision_task_disk( + state_dir, vm_name, rootfs, disk_mode) + path = os.path.join(state_dir, vm_name, "disk.qcow2") + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "wb") as fh: + fh.write(b"QFI\xfb fake") + return path + + def fake_data(state_dir, vm_name, size_gb=10.0, runner=None): + path = os.path.join(state_dir, vm_name, "data.qcow2") + os.makedirs(os.path.dirname(path), exist_ok=True) + if not os.path.exists(path): + with open(path, "wb") as fh: + fh.write(b"QFI\xfb fake-data") + return path + + p1 = mock.patch("cloud.chv.executor.provision_task_disk", + side_effect=fake_provision) + p2 = mock.patch("cloud.chv.executor.provision_data_disk", + side_effect=fake_data) + p1.start() + p2.start() + self.addCleanup(p1.stop) + self.addCleanup(p2.stop) + return ex + + def _socket(self, vm): + os.makedirs(self.state_dir, exist_ok=True) + path = os.path.join(self.state_dir, vm + ".sock") + with open(path, "w") as fh: + fh.write("") + return path + + def test_pool_disabled_when_size_zero(self): + ex = self.make_exec(env_extra={"CHV_WARM_POOL_SIZE": "0"}) + self.assertEqual( + ex.prewarm_persistent_vm("proj1")["pooled"], False) + self.assertEqual( + ex.acquire_for_task(self._task())["mode"], "cold") + # 退化为旧 dispatch：冷启→跑→关机（_vms 不留登记）。 + out = ex.dispatch_task_persistent(self._task()) + self.assertEqual(out["status"], "completed") + self.assertEqual(ex._vms, {}) + + def test_prewarm_idempotent_and_acquire_resume(self): + ex = self.make_exec( + env_extra={"CHV_WARM_POOL_SIZE": "1", + "CHV_DISK_MODE": "overlay"}) + self._patch_overlay_provision(ex) + first = ex.prewarm_persistent_vm("proj1") + self.assertTrue(first["pooled"]) + self.assertEqual(first["mode"], "prewarmed") + self.assertEqual(first["vm"], "chv-persist-proj1") + self.assertEqual(len(self.launches), 1) + # socket 造出来后第二次 prewarm 即 already-warm，不重建。 + self._socket("chv-persist-proj1") + second = ex.prewarm_persistent_vm("proj1") + self.assertEqual(second["mode"], "already-warm") + self.assertEqual(len(self.launches), 1) + # acquire：vm.info=Paused → resume + SSH 就绪。 + self.vm_states["chv-persist-proj1"] = "Paused" + got = ex.acquire_for_task(self._task()) + self.assertEqual(got["mode"], "resumed") + self.assertTrue(got["warm_hit"]) + self.assertIn("resume_seconds", got) + self.assertEqual(got["vm"], "chv-persist-proj1") + self.assertTrue(any("resume" in c for c in self.calls)) + # 暂停回池后 origin 变 task，第二次 acquire 不再是 + # warm_hit，但仍是 resumed。 + self.assertTrue(ex.suspend_after_task( + {"id": "chv-persist-proj1"}, tenant_id="proj1")) + rec = read_persistent_vm(self.state_dir, "proj1") + self.assertEqual(rec["state"], "paused") + self.assertEqual(rec["origin"], "task") + got2 = ex.acquire_for_task(self._task(task_id="second")) + self.assertEqual(got2["mode"], "resumed") + self.assertFalse(got2["warm_hit"]) + + def test_acquire_falls_back_cold(self): + ex = self.make_exec( + env_extra={"CHV_WARM_POOL_SIZE": "1", + "CHV_DISK_MODE": "overlay"}) + self._patch_overlay_provision(ex) + ex.prewarm_persistent_vm("proj1") + self._socket("chv-persist-proj1") + # vm.info=Running：不是 paused，不认领，回落冷启。 + self.vm_states["chv-persist-proj1"] = "Running" + self.assertEqual( + ex.acquire_for_task(self._task())["mode"], "cold") + # socket 消失：登记清掉，回落冷启。 + os.remove(os.path.join(self.state_dir, + "chv-persist-proj1.sock")) + self.vm_states["chv-persist-proj1"] = "Paused" + self.assertEqual( + ex.acquire_for_task(self._task())["mode"], "cold") + self.assertIsNone(read_persistent_vm(self.state_dir, "proj1")) + + def test_acquire_tenant_mismatch_falls_back_cold(self): + ex = self.make_exec( + env_extra={"CHV_WARM_POOL_SIZE": "1", + "CHV_DISK_MODE": "overlay"}) + self._patch_overlay_provision(ex) + ex.prewarm_persistent_vm("proj1") + self._socket("chv-persist-proj1") + self.vm_states["chv-persist-proj1"] = "Paused" + # 盘的归属被改成另一租户：绝不 resume，回落冷启。 + with open(os.path.join(self.state_dir, "chv-persist-proj1", + "tenant.json"), "w") as fh: + json.dump({"tenant_id": "tenant-b", + "vm": "chv-persist-proj1"}, fh) + self.assertEqual( + ex.acquire_for_task(self._task())["mode"], "cold") + self.assertIsNone(read_persistent_vm(self.state_dir, "proj1")) + + def test_dispatch_persistent_two_tasks(self): + ex = self.make_exec( + env_extra={"CHV_WARM_POOL_SIZE": "1", + "CHV_DISK_MODE": "overlay"}) + self._patch_overlay_provision(ex) + ex.prewarm_persistent_vm("proj1") + self._socket("chv-persist-proj1") + self.vm_states["chv-persist-proj1"] = "Paused" + out1 = ex.dispatch_task_persistent(self._task(task_id="t1")) + self.assertEqual(out1["status"], "completed") + self.assertEqual(out1["vm_mode"], "resumed") + self.assertEqual(out1["vm"], "chv-persist-proj1") + self.assertIn("resume_seconds", out1) + # 任务结束已暂停回池：第二个任务仍是 resumed，且是 + # 同一台 VM（同一 overlay/数据盘）。 + out2 = ex.dispatch_task_persistent( + self._task(task_id="t2")) + self.assertEqual(out2["status"], "completed") + self.assertEqual(out2["vm_mode"], "resumed") + self.assertEqual(out2["vm"], "chv-persist-proj1") + rec = read_persistent_vm(self.state_dir, "proj1") + self.assertEqual(rec["state"], "paused") + # 任务盘仍在（暂停不删盘、不关机语义）。 + self.assertTrue(os.path.exists(os.path.join( + self.state_dir, "chv-persist-proj1", "disk.qcow2"))) + self.assertTrue(os.path.exists(os.path.join( + self.state_dir, "chv-persist-proj1", "data.qcow2"))) + + def test_dispatch_persistent_cold_when_pool_empty(self): + ex = self.make_exec( + env_extra={"CHV_WARM_POOL_SIZE": "1", + "CHV_DISK_MODE": "overlay"}) + self._patch_overlay_provision(ex) + # 池空（未 prewarm）：冷启持久 VM，跑完暂停回池。 + out = ex.dispatch_task_persistent(self._task(task_id="t1")) + self.assertEqual(out["status"], "completed") + self.assertEqual(out["vm_mode"], "cold") + self.assertEqual(out["vm"], "chv-persist-proj1") + rec = read_persistent_vm(self.state_dir, "proj1") + self.assertEqual(rec["state"], "paused") + self.assertEqual(rec["vm"], "chv-persist-proj1") + + def test_stop_keeps_protected_overlay_and_gc_skips_it(self): + ex = self.make_exec( + env_extra={"CHV_WARM_POOL_SIZE": "1", + "CHV_DISK_MODE": "overlay"}) + self._patch_overlay_provision(ex) + ex.prewarm_persistent_vm("proj1") + self._socket("chv-persist-proj1") + overlay = os.path.join(self.state_dir, "chv-persist-proj1", + "disk.qcow2") + self.assertTrue(ex.stop_vm({"id": "chv-persist-proj1"})) + # 有持久登记认领：stop 不删 overlay。 + self.assertTrue(os.path.exists(overlay)) + # GC 同样跳过认领的盘。 + self.assertEqual(ex.gc_stale_disks(), {"deleted": [], + "failed": []}) + self.assertTrue(os.path.exists(overlay)) + # 显式回收才删 overlay + 数据盘 + 登记。 + out = ex.recycle_persistent_vm({"id": "chv-persist-proj1"}, + tenant_id="proj1") + self.assertEqual(out["state"], "recycled") + self.assertIn("task", out["deleted_disks"]) + self.assertIn("data", out["deleted_disks"]) + self.assertFalse(os.path.exists(overlay)) + self.assertFalse(os.path.exists(os.path.join( + self.state_dir, "chv-persist-proj1", "data.qcow2"))) + self.assertIsNone(read_persistent_vm(self.state_dir, "proj1")) + + def test_refill_warm_pool_noop_when_warm(self): + ex = self.make_exec( + env_extra={"CHV_WARM_POOL_SIZE": "1", + "CHV_DISK_MODE": "overlay"}) + self._patch_overlay_provision(ex) + ex.prewarm_persistent_vm("proj1") + self._socket("chv-persist-proj1") + out = ex.refill_warm_pool("proj1") + self.assertEqual(out["mode"], "already-warm") + self.assertEqual(len(self.launches), 1) + + +if __name__ == "__main__": + unittest.main()