diff --git a/cloud/chv/executor.py b/cloud/chv/executor.py index c9920a8..ea533e4 100644 --- a/cloud/chv/executor.py +++ b/cloud/chv/executor.py @@ -31,6 +31,7 @@ import logging import os import re import shutil +import socket import subprocess import threading import time @@ -187,6 +188,29 @@ _SSH_TRANSIENT_RESET_MARKERS = ( ) +# CHV 混合模式已定值（DESIGN-HYBRID.md §8，2026-11-08 定； +# feat-chv-hybrid-a A 轨只用常量/注释承载时延目标，不在单元 +# 测试里硬测时延）。W=每宿主干净池预热数；池上限=低频并发 +# 峰值的 50%，超上限走冷启不扩池；常驻上限=宿主 VM 槽位 +# 的 50%；排队超时后明确拒绝 host_capacity_exceeded。 +HYBRID_WARM_POOL_PER_HOST = 2 +HYBRID_POOL_CAP_FRACTION = 0.5 +HYBRID_RESIDENT_CAP_FRACTION = 0.5 +HYBRID_QUEUE_TIMEOUT_SECONDS = 300 +HYBRID_QUEUE_POLL_INTERVAL_SECONDS = 5 +HYBRID_WARM_RESUME_TARGET_SECONDS = 2 +HYBRID_COLD_RESTORE_TARGET_SECONDS = 30 +HYBRID_SNAPSHOT_PER_TENANT = 1 +HYBRID_SNAPSHOT_RETENTION_DAYS = 30 +HYBRID_RESIDENT_TIERS = ("paid", "heavy") +CLEAN_POOL_DIR_NAME = "clean_pool" +# 干净池预热时拒绝的租户同步键（Gitea 同步变量）：池 VM 出生 +# 时不带租户，显式带这些键进预热即调用方错误，fail-closed。 +_CLEAN_POOL_REFUSED_SYNC_KEYS = ( + "GITEA_URL", "GITEA_USER", "GITEA_TOKEN", "SYNC_REPO", + "SYNC_BRANCH") + + def _is_transient_ssh_reset(result) -> bool: """单次 SSH 是否撞上连接建立阶段的瞬时 reset：returncode 255 + stderr 含 kex/reset 标记。stderr 可能是 bytes， @@ -337,6 +361,13 @@ class CloudHypervisorExecutor(Executor): 重启后载入的条目须经 vm.info=Paused + socket 活性 + tenant 归属三重核对才认领，否则只当孤儿清理， 绝不直接发给任务（见 orphans.py 与 acquire_for_task）。 + 混合模式（feat-chv-hybrid-a，DESIGN-HYBRID.md §9.1 第 1–4 + 项）：dispatch_task_hybrid 按 tier 分流（Paid/Heavy 常驻、 + 其余走池）；CHV_WARM_POOL_SIZE 的计数口径从“每租户 1 台” + 改为“每宿主 2 台干净池”（W=2）；干净池 VM 出生时不带租户， + 分配时才挂载租户数据盘（任务盘来源不变量见 + _verify_task_disk_source）；restore_vm 在混合模式启用时 + 严格校验快照 tenant_id（无 tenant 的登记即拒绝）。 - 长任务中途死亡：_run_with_vm 在客内命令抛传输异常、 或非 goal 结果已 failed 且 VM 登记进程已死时，改判 interrupted（断点可续跑）而不是笼统 failed；VM 还活 @@ -1842,9 +1873,15 @@ class CloudHypervisorExecutor(Executor): return out[0], out[1] def _effective_extra_env(self, task: dict, - extra_env: dict | None) -> dict | None: + extra_env: dict | None, + skip_gitea_sync: bool = False) -> dict | None: """wake 生效的 extra_env：把 Gitea 同步变量并进来。 + skip_gitea_sync=True（干净池出生，feat-chv-hybrid-a）： + 原样返回 extra_env，不并宿主配置的 Gitea 同步变量—— + 池 VM 出生时不带租户，同步变量只能在分配时随租户绑定 + 给，不在出生时给。 + 优先级（低→高）：执行器宿主配置 CHV_GITEA_*（构造时 已映射成 gitea_sync_env）< 任务 dict 的 gitea_sync < 显式 extra_env 同名键（见 gitea_sync.extract_...）。 @@ -1855,6 +1892,8 @@ class CloudHypervisorExecutor(Executor): 变量时原样返回（本次启动不同步）。 必须在 _wake_prechecks 之前调：任务/配置来源的 GITEA_TOKEN 也要受 keyscan 模式 fail-closed 检查。""" + if skip_gitea_sync: + return extra_env from cloud.chv.gitea_sync import ( extract_gitea_sync_env, validate_gitea_sync_env, @@ -1947,26 +1986,38 @@ class CloudHypervisorExecutor(Executor): self._mark_guest_env_delivered(name, secrets) return seed_disk - def _boot_vm(self, task, name, task_disk, seed_disk, - net_entries) -> None: - """VMM 启动→vm.create→vm.boot→SSH 就绪→密钥已投递时 - 等 guest-env 对 SSH 用户可读（V-9，见 - _wait_guest_env_usable），纯移动。""" + def _create_vm(self, task, name, task_disk, seed_disk, + attach_data_disk=True) -> None: + """PUT vm.create（_boot_vm 的建盘段）：VmConfig 挂任务盘 + + 可选 seed 盘 + 可选数据盘；attach_data_disk=False 时 + 不挂数据盘（干净池出生）。失败即 ChvFailure。""" socket_path = self.api_socket(task) - self._launch_api_only(task) - self.wait_api_ready(task) - self._check_api_socket_permissions(socket_path) - status, body = self._api_call( + status, _body = self._api_call( socket_path, "PUT", "/vm.create", self.build_vm_config( task_disk, seed_disk=seed_disk, - data_disk=self._provision_data_disk_path(task))) + data_disk=self._provision_data_disk_path(task) + if attach_data_disk else None)) if status not in (200, 204): raise ChvFailure( "vm_create_failed", "vm.create 失败：status=%s" % status) self._emit("vm_create", task, "vm=%s disk=%s" % (name, task_disk)) + + def _boot_vm(self, task, name, task_disk, seed_disk, + net_entries, attach_data_disk=True) -> None: + """VMM 启动→vm.create→vm.boot→SSH 就绪→密钥已投递时 + 等 guest-env 对 SSH 用户可读（V-9，见 + _wait_guest_env_usable），纯移动。attach_data_disk=False + （干净池出生，feat-chv-hybrid-a）：VmConfig 不挂数据盘， + 池 VM 出生时只有干净系统盘，数据盘在分配时才挂载。""" + socket_path = self.api_socket(task) + self._launch_api_only(task) + self.wait_api_ready(task) + self._check_api_socket_permissions(socket_path) + self._create_vm(task, name, task_disk, seed_disk, + attach_data_disk=attach_data_disk) if net_entries: # feat-chv-closeup（P2-19）：登记失败即拒起， # 见 _record_net_entries（审计 + 返回 False）。 @@ -1996,9 +2047,28 @@ class CloudHypervisorExecutor(Executor): # 就绪（runcmd chown 比 SSH 就绪晚约 30 秒）。 self._wait_guest_env_usable(task) - def _wake_prepare(self, task: dict, name: str) -> tuple: + def _wake_prepare_clean(self, name: str) -> str: + """干净池出生的任务盘供给（_wake_prepare 的干净分支）： + 只供给任务系统盘（不建数据盘、不写 tenant 标记）；上一 + 租户的 tenant.json 若残留，先清掉，干净 VM 绝不继承归属。""" + _stale_tenant = os.path.join(self.state_dir, name, + "tenant.json") + try: + if os.path.exists(_stale_tenant): + os.remove(_stale_tenant) + except OSError: + pass + return provision_task_disk( + self.state_dir, name, self.rootfs, self.disk_mode) + + def _wake_prepare(self, task: dict, name: str, + clean_pool: bool = False) -> tuple: """wake 准备：reuse 审计、旧 VM 清理、任务盘供给、 直连 egress 审计，纯移动。返回 (task_disk, socket_path)。 + clean_pool=True（干净池出生，feat-chv-hybrid-a）：只供 + 给任务系统盘，不建数据盘、不写 tenant 归属标记，并清掉 + 上一租户可能留下的 tenant.json——干净池 VM 出生时不带 + 任何租户。 feat-chv-v3（F1）：旧 VM 在时先删旧任务盘再供给新盘。 根因：H5 在 seed 建好后删宿主私钥，re-wake 必然重新 生成 key 对；旧盘若复用，cloud-init 以 instance-id @@ -2027,7 +2097,10 @@ class CloudHypervisorExecutor(Executor): "stale_disk_cleanup_failed", "旧任务盘删不掉，拒绝复用旧盘启动：VM %s：%s" % (name, exc)) - task_disk = self._provision_task_disk(task) + if clean_pool: + task_disk = self._wake_prepare_clean(name) + else: + task_disk = self._provision_task_disk(task) socket_path = self.api_socket(task) if self.allow_direct_egress: self._emit("egress_direct_allowed", task, @@ -2035,7 +2108,28 @@ class CloudHypervisorExecutor(Executor): return task_disk, socket_path - def wake_vm(self, task: dict, extra_env: dict | None = None) -> dict: + def _wake_boot(self, task, name, task_disk, secrets, guest_env, + seed_extras, net_entries, clean_pool=False) -> None: + """wake 的启动段（seed 盘供给 + _boot_vm）：任一步失败都 + 关 VM（stop_vm）再抛，绝不带着半醒的 VM 去跑任务。""" + try: + seed_disk = self._provision_seed_disk( + task, name, secrets, guest_env=guest_env or None, + seed_extras=seed_extras) + self._boot_vm(task, name, task_disk, seed_disk, net_entries, + attach_data_disk=not clean_pool) + except Exception: + try: + self.stop_vm(task) + except Exception: + pass + if isinstance(task, dict): + self._emit("vm_failed", task, + "vm=%s wake 失败" % name) + raise + + def wake_vm(self, task: dict, extra_env: dict | None = None, + clean_pool: bool = False) -> dict: """主唤醒路径（API 模式，dispatch/run/goal 的前置步骤）： preflight → 容量门 → 磁盘配额门 → 每任务磁盘供给 → launcher 只起 VMM（--api-socket）→ 等 vmm.ping 通 → @@ -2052,12 +2146,17 @@ class CloudHypervisorExecutor(Executor): 密钥子集在 provisioned 模式下写进 seed 客内文件， keyscan 模式下有密钥子集即 fail-closed；起 VM 前的 preflight/容量/配额失败也发 vm_failed 审计（L3）； - net 隔离在起 VMM 前校验（H7），成功后登记 net.json。""" + net 隔离在起 VMM 前校验（H7），成功后登记 net.json。 + clean_pool=True（干净池出生）：不并 Gitea 同步变量、 + 不建数据盘、不写 tenant 标记、VmConfig 不挂数据盘； + host-key 预置仍按 provisioned 模式走（SSH 钉死所需， + 与租户 seed 不同——租户密钥/Gitea 一律不带）。""" if not isinstance(task, dict) or task.get("id") is None: raise ValueError("任务缺少 id") name = self.vm_name(task) try: - extra_env = self._effective_extra_env(task, extra_env) + extra_env = self._effective_extra_env( + task, extra_env, skip_gitea_sync=clean_pool) seed_extras = self._seed_extras() except ChvFailure as exc: self._emit("vm_failed", task, @@ -2066,21 +2165,10 @@ class CloudHypervisorExecutor(Executor): raise secrets, net_entries = self._wake_prechecks(task, name, extra_env) guest_env, _plain = split_guest_env(extra_env) - task_disk, socket_path = self._wake_prepare(task, name) - try: - seed_disk = self._provision_seed_disk( - task, name, secrets, guest_env=guest_env or None, - seed_extras=seed_extras) - self._boot_vm(task, name, task_disk, seed_disk, net_entries) - except Exception: - try: - self.stop_vm(task) - except Exception: - pass - if isinstance(task, dict): - self._emit("vm_failed", task, - "vm=%s wake 失败" % name) - raise + task_disk, socket_path = self._wake_prepare( + task, name, clean_pool=clean_pool) + self._wake_boot(task, name, task_disk, secrets, guest_env, + seed_extras, net_entries, clean_pool=clean_pool) self._emit("vm_ready", task, "vm=%s" % name) return {"vm": name, "api_socket": socket_path, "state": "running", "ready": True, "mode": "cold"} @@ -2323,12 +2411,20 @@ class CloudHypervisorExecutor(Executor): clear_persistent_vm(self.state_dir, tenant) return {"mode": "cold", "vm": None} resume_seconds = _time.time() - t0 - write_persistent_vm(self.state_dir, tenant, { + _kept = {} + # feat-chv-hybrid-a：常驻标记（tier/resident）随登记流转， + # acquire/suspend 重写登记时不得丢——丢了常驻租户下次 + # 就被当成池租户，租约与容量口径即破。 + if record.get("tier"): + _kept["tier"] = record.get("tier") + if record.get("resident") is True: + _kept["resident"] = True + write_persistent_vm(self.state_dir, tenant, dict({ "vm": vm, "tenant_id": tenant, "state": "running", "origin": record.get("origin") or "task", "gitea_synced": bool(record.get("gitea_synced")), - "updated_at": _time.time()}) + "updated_at": _time.time()}, **_kept)) self._emit("warm_pool_acquire", task, "vm=%s mode=resumed resume_seconds=%.2f" " warm_hit=%s" @@ -2361,11 +2457,20 @@ class CloudHypervisorExecutor(Executor): if tenant: import time as _time record = read_persistent_vm(self.state_dir, tenant) or {} - write_persistent_vm(self.state_dir, tenant, { + _kept = {} + # feat-chv-hybrid-a：同 _acquire_resume_recorded， + # 常驻标记随登记流转，暂停回池时不得丢。 + if record.get("tier"): + _kept["tier"] = record.get("tier") + if record.get("resident") is True: + _kept["resident"] = True + _record = { "vm": vm, "tenant_id": tenant, "state": "paused", "origin": "task", "gitea_synced": bool(record.get("gitea_synced")), - "updated_at": _time.time()}) + "updated_at": _time.time()} + _record.update(_kept) + write_persistent_vm(self.state_dir, tenant, _record) return True def _run_guest_on_vm(self, task, vm, inner_args, goal, @@ -2399,18 +2504,45 @@ class CloudHypervisorExecutor(Executor): "客内命令已 failed 且登记的 VMM 进程已死") return result + def _ensure_resident_capacity(self, tenant) -> None: + """常驻容量门（_run_with_persistent_vm 的常驻腿前置）： + 无 tenant 即拒绝（无归属不能占常驻槽位）；已有常驻登记 + 的租户直接放行；否则常驻数达上限时排队等槽位，超时仍满 + 即明确拒绝 host_capacity_exceeded，不抢占已有常驻。""" + if not tenant: + raise ChvFailure( + "resident_tenant_missing", + "常驻路由须带 tenant_id（无归属的任务不能占常驻" + "槽位），请走池路由") + _existing = read_persistent_vm(self.state_dir, tenant) + if _existing is not None \ + and _existing.get("resident") is True: + return + if self._resident_count() >= self.resident_cap(): + if not self._wait_for_resident_capacity(): + raise ChvFailure( + "host_capacity_exceeded", + "常驻槽位已满：已有 %d 台，上限 %d，排队 %d 秒" + "后仍无槽位，明确拒绝" + % (self._resident_count(), self.resident_cap(), + HYBRID_QUEUE_TIMEOUT_SECONDS)) + def _run_with_persistent_vm(self, task, inner_args, goal=False, - extra_env=None, **kwargs): + extra_env=None, hybrid_resident=False, + hybrid_tier=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）。""" + 名仍按真实任务 id 派生。中途死亡分类见 + _run_guest_on_vm（登记进程已死即 interrupted）。 + hybrid_resident=True（feat-chv-hybrid-a）：先过常驻 + 容量门，再把 tier/resident 写进持久登记；结果附 route/allocation。""" tenant = self._tenant_id(task) + if hybrid_resident: + self._ensure_resident_capacity(tenant) 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) @@ -2418,6 +2550,8 @@ class CloudHypervisorExecutor(Executor): resume_seconds = acquired.get("resume_seconds") if mode == "cold": self.wake_vm(vm_task, extra_env=extra_env) + if hybrid_resident: + self._mark_resident_record(tenant, hybrid_tier, vm) try: result = self._run_guest_on_vm( task, vm, inner_args, goal, @@ -2425,6 +2559,9 @@ class CloudHypervisorExecutor(Executor): if isinstance(result, dict): result = dict(result) result["vm_mode"] = mode + if hybrid_resident: + result["route"] = "resident" + result["allocation"] = "resident-" + mode if resume_seconds is not None: result["resume_seconds"] = resume_seconds return result @@ -2468,6 +2605,758 @@ class CloudHypervisorExecutor(Executor): model_provider=model_provider, sandbox=("read-only" if task.get("plan") else None)) + # -- 混合模式（feat-chv-hybrid-a，DESIGN-HYBRID.md §9.1 第 1–4 项） + # 路由：常驻租户（tier Paid/Heavy 且有 tenant）走常驻 VM， + # 其余走共享干净池（命中预热 / 冷启 / 排队）。池 VM 出生时 + # 不带租户（无数据盘、无 seed、无 Gitea 同步等待），只 boot + # 到 SSH 就绪 + pause；租户绑定延后到 acquire 时（数据盘 + # 挂载 + seed 注入 + 首次同步等待）。用过的池 VM 一律销毁， + # 绝不回池。CHV_WARM_POOL_SIZE 的计数口径从“每租户 1 台” + # 改为“每宿主 W 台干净池”（W=HYBRID_WARM_POOL_PER_HOST=2）； + # 阶段 1 的每租户持久 VM 路径（prewarm_persistent_vm / + # acquire_for_task）原样保留，只作常驻腿的底座。 + # 审计事件名按设计原文，detail 一律带 host_id，绝不带密钥值。 + + def _host_id(self) -> str: + """本宿主 id（审计用）：显式 CHV_HOST_ID > 主机名。 + 审计 detail 里只放这个 id，不放密钥、不放快照内容。""" + host_id = "" + try: + host_id = str(self._env.get("CHV_HOST_ID") or "").strip() + except Exception: + host_id = "" + if host_id: + return host_id + try: + return socket.gethostname() + except Exception: + return "unknown" + + # -- 口径与容量 ------------------------------------------------------ + def clean_pool_target(self) -> int: + """干净池目标台数（本宿主）：CHV_WARM_POOL_SIZE 的新口径 + ——每宿主 W 台干净池（W 定值见 warm_pool_per_host_target， + 部署上把 CHV_WARM_POOL_SIZE 设成 W 即 2）；为 0 即池关闭， + 混合路由整体退化为阶段 1 旧语义。旧口径“每租户 1 台” + 已废止：该值不再按租户计数。""" + try: + return max(0, int(getattr(self, "warm_pool_size", 0))) + except (TypeError, ValueError): + return 0 + + def warm_pool_per_host_target(self) -> int: + """每宿主干净池预热数 W（已定值 2 台/宿主）。""" + return HYBRID_WARM_POOL_PER_HOST + + def warm_pool_cap(self, low_freq_peak: float | None = None) \ + -> int | None: + """池上限 = 低频并发峰值的 50%（已定口径）：超上限的低频 + 任务走冷启，不扩池。峰值未给（参数与 CHV_LOW_FREQ_PEAK_ + CONCURRENCY 环境变量皆无）即 None（不封顶，由目标数定）。 + 20 并发 → 10 台，50 并发 → 25 台。""" + if low_freq_peak is None: + raw = self._env.get("CHV_LOW_FREQ_PEAK_CONCURRENCY") + if raw is None: + return None + try: + low_freq_peak = float(raw) + except (TypeError, ValueError): + return None + try: + peak = float(low_freq_peak) + except (TypeError, ValueError): + return None + if peak <= 0: + return 0 + return max(0, int(peak * HYBRID_POOL_CAP_FRACTION)) + + def clean_pool_effective_target( + self, low_freq_peak: float | None = None) -> int: + """干净池生效目标 = min（配置目标，池上限）：上限只会 + 压小池子，绝不会把它扩过配置目标。""" + target = self.clean_pool_target() + cap = self.warm_pool_cap(low_freq_peak) + if cap is None: + return target + return min(target, cap) + + def resident_cap(self) -> int: + """每宿主常驻上限 = VM 槽位的 50%（已定值）：48 槽位 → + 24 台；超上限的新常驻申请排队，不抢占已有常驻。""" + try: + max_vms = int(getattr(self, "max_vms", 4)) + except (TypeError, ValueError): + max_vms = 4 + return max(0, int(max_vms * HYBRID_RESIDENT_CAP_FRACTION)) + + def _resident_count(self) -> int: + """当前常驻租户数：persistent 登记里 resident=True 的 + 条目数（损坏/缺失的登记跳过，不误计）。""" + count = 0 + try: + entries = os.listdir( + os.path.join(self.state_dir, "persistent")) + except Exception: + return 0 + for entry in entries: + if not entry.endswith(".json"): + continue + try: + with open(os.path.join( + self.state_dir, "persistent", entry), + "r") as fh: + data = json.load(fh) + except Exception: + continue + if isinstance(data, dict) \ + and data.get("resident") is True: + count += 1 + return count + + def _mark_resident_record(self, tenant, tier, vm) -> None: + """把 tier/resident 写进该租户的持久登记（常驻腿的租约 + 标记，租约计时本身是另一轨，不在这里）：state/origin/ + gitea_synced 沿用旧登记，不丢。""" + record = read_persistent_vm(self.state_dir, tenant) or {} + write_persistent_vm(self.state_dir, tenant, { + "vm": vm, "tenant_id": tenant, + "state": record.get("state") or "running", + "origin": record.get("origin") or "task", + "gitea_synced": bool(record.get("gitea_synced")), + "tier": tier, "resident": True, + "updated_at": time.time()}) + + def _hybrid_route(self, task: dict) -> str: + """混合路由判定（只看档位 + 租户归属，不看频率）： + tier Paid/Heavy 且带 tenant_id → resident；其余（Free、 + 无 tier、无 tenant）一律 pool。无 tier 行 = 无常驻资格， + fail-closed 走池（与 P11 用量门的豁免语义相反，常驻是 + 权益门，不许把 P11 的豁免复用到常驻门上）。""" + tier = "" + if isinstance(task, dict): + tier = str(task.get("tier") or "").strip().lower() + if tier in HYBRID_RESIDENT_TIERS and self._tenant_id(task): + return "resident" + return "pool" + + def _wait_for_capacity(self, timeout_seconds=None) -> bool: + """排队等可用宿主（§3.4）：轮询 _persistent_vm_count， + 有空槽即 True；超时（已定 300 秒，CHV_QUEUE_TIMEOUT + 可覆盖，测试用 0 即不等待）仍满即 False，调用方明确 + 拒绝 host_capacity_exceeded，绝不静默降级。轮询按 + 次数封顶（timeout/interval），fake sleeper 不推进墙钟 + 时也不会空转。""" + if timeout_seconds is None: + try: + timeout_seconds = float(self._env.get( + "CHV_QUEUE_TIMEOUT", HYBRID_QUEUE_TIMEOUT_SECONDS)) + except (TypeError, ValueError): + timeout_seconds = float(HYBRID_QUEUE_TIMEOUT_SECONDS) + timeout_seconds = max(0.0, float(timeout_seconds)) + interval = HYBRID_QUEUE_POLL_INTERVAL_SECONDS + attempts = max(1, int(timeout_seconds / interval)) \ + if timeout_seconds > 0 else 1 + for _ in range(attempts): + if self._persistent_vm_count() < self.max_vms: + return True + self.sleeper(interval) + return self._persistent_vm_count() < self.max_vms + + def _wait_for_resident_capacity(self, timeout_seconds=None) -> bool: + """排队等常驻槽位（§4.1）：轮询 _resident_count，有常驻 + 槽位即 True；超时仍满即 False，调用方明确拒绝 + host_capacity_exceeded，不抢占已有常驻。与 + _wait_for_capacity 的区别：等的是常驻上限（槽位 50%）， + 不是通用 VM 容量——通用槽位空闲时常驻仍可能已满。""" + if timeout_seconds is None: + try: + timeout_seconds = float(self._env.get( + "CHV_QUEUE_TIMEOUT", HYBRID_QUEUE_TIMEOUT_SECONDS)) + except (TypeError, ValueError): + timeout_seconds = float(HYBRID_QUEUE_TIMEOUT_SECONDS) + timeout_seconds = max(0.0, float(timeout_seconds)) + interval = HYBRID_QUEUE_POLL_INTERVAL_SECONDS + attempts = max(1, int(timeout_seconds / interval)) \ + if timeout_seconds > 0 else 1 + for _ in range(attempts): + if self._resident_count() < self.resident_cap(): + return True + self.sleeper(interval) + return self._resident_count() < self.resident_cap() + + # -- 干净池登记 ------------------------------------------------------ + def _clean_pool_vm_name(self, slot: int) -> str: + """干净池第 slot 台 VM 名：chv-pool-（与租户无关， + 租户绑定只写登记 + tenant.json，不进 VM 名）。""" + return "chv-pool-%d" % int(slot) + + def _clean_pool_record_path(self, slot: int) -> str: + return os.path.join(self.state_dir, CLEAN_POOL_DIR_NAME, + "slot-%d.json" % int(slot)) + + def _read_clean_pool_record(self, slot: int): + """读干净池登记：无文件/损坏/VM 名不符即 None（不猜）。""" + try: + with open(self._clean_pool_record_path(slot), "r") as fh: + data = json.load(fh) + except (OSError, ValueError): + return None + if not isinstance(data, dict) \ + or data.get("vm") != self._clean_pool_vm_name(slot): + return None + return data + + def _write_clean_pool_record(self, slot: int, record: dict) -> str: + path = self._clean_pool_record_path(slot) + os.makedirs(os.path.dirname(path), exist_ok=True) + atomic_write_text(path, json.dumps(record)) + return path + + def _clear_clean_pool_record(self, slot: int) -> bool: + path = self._clean_pool_record_path(slot) + if os.path.exists(path): + os.remove(path) + return True + return False + + def _list_clean_pool_records(self): + """全部干净池登记，按 slot 排序：[(slot, record)]。""" + out = [] + try: + entries = os.listdir( + os.path.join(self.state_dir, CLEAN_POOL_DIR_NAME)) + except Exception: + return out + for entry in sorted(entries): + if not entry.startswith("slot-") \ + or not entry.endswith(".json"): + continue + try: + slot = int(entry[len("slot-"):-len(".json")]) + except ValueError: + continue + record = self._read_clean_pool_record(slot) + if record is not None: + out.append((slot, record)) + return out + + def _clean_pool_slot_for_vm(self, vm): + """该 VM 占的干净池 slot，无登记即 None。""" + for slot, record in self._list_clean_pool_records(): + if record.get("vm") == vm: + return slot + return None + + def _clean_pool_warm_vm(self, target): + """已暖的干净池 VM 名（paused 登记 + socket 活着），无即 None。""" + for i in range(target): + record = self._read_clean_pool_record(i) + if record is not None \ + and record.get("state") == "paused" \ + and os.path.exists(self.api_socket( + {"id": record.get("vm") or ""})): + return record.get("vm") + return None + + def _pick_clean_pool_slot(self, target): + """挑一个可建的干净池 slot：无登记 / 残留登记的 slot； + 正跑任务的 slot 绝不重建；全部无槽即 None。""" + for i in range(target): + record = self._read_clean_pool_record(i) + if record is None: + return i + if record.get("state") == "running": + continue + if record.get("state") == "paused" and os.path.exists( + self.api_socket({"id": record.get("vm") or ""})): + continue + return i + return None + + def _check_clean_prewarm_seed(self, extra_env) -> None: + """干净池预热的 seed 门禁：显式带密钥或 Gitea 同步变量 + 即 ChvFailure clean_pool_seed_refused（出生时带租户态 + 是调用方错误，同步只能在分配时随租户绑定给）。""" + secrets, _plain = split_secret_env(extra_env) + if secrets: + raise ChvFailure( + "clean_pool_seed_refused", + "干净池预热不带租户 seed：密钥 %s 只能在分配时" + "随租户绑定给，不能在出生时给" + % ",".join(sorted(secrets))) + if extra_env and any( + key in extra_env + for key in _CLEAN_POOL_REFUSED_SYNC_KEYS): + raise ChvFailure( + "clean_pool_seed_refused", + "干净池预热不带 Gitea 同步变量：同步只能在分配" + "时随租户绑定给，不能在出生时给") + + def _prewarm_clean_build(self, slot, vm, extra_env): + """干净池预热的建造段：冷启干净 VM → pause → 写登记 → + 发 warm_pool_prewarm 审计。返回 prewarmed 结果。""" + vm_task = {"id": "pool-%d" % slot} + self.wake_vm(vm_task, extra_env=extra_env, clean_pool=True) + self.pause_vm(vm_task) + self._write_clean_pool_record(slot, { + "vm": vm, "slot": slot, "tenant_id": None, + "state": "paused", "origin": "clean-prewarm", + "gitea_synced": False, "data_disk": None, + "updated_at": time.time()}) + self._emit("warm_pool_prewarm", vm_task, + "vm=%s clean=true host_id=%s gitea_synced=false" + % (vm, self._host_id())) + return {"pooled": True, "vm": vm, "mode": "prewarmed", + "gitea_synced": False, "clean": True, + "tenant_id": None} + + def prewarm_clean_pool(self, slot: int | None = None, + extra_env: dict | None = None) -> dict: + """干净池预热（从 prewarm_persistent_vm 拆出，§9.1 第 2 项）： + 池 VM 出生时不带租户——无数据盘、无 seed、无 Gitea 同步 + 等待，只 boot 到 SSH 就绪 + pause；租户绑定延后到 acquire + 时。幂等：已有 paused 登记且 socket 活着即 already-warm。 + 显式带密钥或 Gitea 同步变量进预热即 ChvFailure + clean_pool_seed_refused（出生时带租户态是调用方错误）。 + host-key 预置仍按 provisioned 模式走（SSH 钉死所需， + 与租户 seed 不同）。返回 {pooled, vm, mode, + gitea_synced: False, clean: True, tenant_id: None}。""" + target = self.clean_pool_target() + if target <= 0: + return {"pooled": False, "vm": None, + "reason": "warm_pool_disabled"} + if slot is None: + warm = self._clean_pool_warm_vm(target) + if warm: + return {"pooled": True, "vm": warm, + "mode": "already-warm", + "gitea_synced": False, "clean": True, + "tenant_id": None} + slot = self._pick_clean_pool_slot(target) + if slot is None: + # 全部 slot 不是已暖就是正跑任务：无槽可建。 + return {"pooled": False, "vm": None, + "reason": "clean_pool_all_busy"} + if int(slot) < 0 or int(slot) >= target: + return {"pooled": False, "vm": None, + "reason": "clean_pool_slot_out_of_range"} + slot = int(slot) + vm = self._clean_pool_vm_name(slot) + record = self._read_clean_pool_record(slot) + 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": False, "clean": True, + "tenant_id": None} + if record is not None and record.get("state") == "running": + # 正跑任务的 slot 绝不重建（重建等于杀掉租户任务）。 + return {"pooled": False, "vm": vm, + "reason": "clean_pool_slot_busy"} + self._check_clean_prewarm_seed(extra_env) + return self._prewarm_clean_build(slot, vm, extra_env) + + def refill_clean_pool(self, extra_env: dict | None = None) -> dict: + """干净池补建（后台调用方自行起线程跑，本方法本身同步）： + 逐 slot 预热，already-warm 的 slot 不重建；单 slot 失败 + 记 reason，不挡其它 slot。warm_pool_size==0 时空跑。""" + target = self.clean_pool_target() + results = [] + for i in range(target): + try: + results.append(self.prewarm_clean_pool( + slot=i, extra_env=extra_env)) + except Exception as exc: + results.append({"pooled": False, + "vm": self._clean_pool_vm_name(i), + "reason": "refill_failed", + "error": str(exc)[:200]}) + return {"target": target, "slots": results} + + # -- 租户数据盘：挂载 / 卸载（§9.1 第 3 项，ADR-5 正面实现）------ + def _tenant_data_disk_name(self, tenant) -> str: + """该租户的数据盘目录名：tenant-data-<安全字符租户>。 + 数据盘住在 VM 外（state_dir/tenant-data-<租户>/data.raw）， + 与任何一台池 VM 的自带盘都不是同一个文件。""" + safe = re.sub(r"[^A-Za-z0-9_.-]", "-", + str(tenant or "default")) + return "tenant-data-" + safe + + def _tenant_data_disk_path(self, tenant) -> str: + """该租户的数据盘路径（唯一合法的任务数据盘来源）。""" + return data_disk_path( + self.state_dir, self._tenant_data_disk_name(tenant)) + + def _provision_tenant_data_disk(self, tenant) -> str: + """租户数据盘供给：已存在即复用（旧盘里可能已有用户数据， + 不 resize、不重建）；不存在即新建空稀疏盘并写 tenant 归属 + 标记。首次使用的租户到这里才有数据盘（§3.1 回落：尚无 + 数据盘时分配时新建 + 首次同步等待）。""" + name = self._tenant_data_disk_name(tenant) + path = provision_data_disk(self.state_dir, name) + write_vm_tenant(self.state_dir, name, tenant, disk=path) + return path + + def _verify_task_disk_source(self, vm, tenant, data_disk) -> None: + """任务盘来源不变量（ADR-5 的正面实现，测试钉死）： + 任务数据盘必须是该租户自己的数据盘，绝不能是干净快照盘 + （池 VM 自带盘）、基础镜像或其它租户的盘。任一不符即 + ChvFailure task_disk_source_invalid（fail-closed，不补 + 默认值、不静默换盘、不把干净盘当任务盘用）。""" + if not tenant: + raise ChvFailure( + "task_disk_source_invalid", + "任务盘归属校验须带 tenant_id：无归属的任务不能" + "绑定数据盘") + if not data_disk: + raise ChvFailure( + "task_disk_source_invalid", + "任务数据盘缺失：租户 %s 没有可挂载的数据盘" + % tenant) + expected = os.path.realpath( + self._tenant_data_disk_path(tenant)) + actual = os.path.realpath(data_disk) + if actual != expected: + raise ChvFailure( + "task_disk_source_invalid", + "任务数据盘 %s 不是租户 %s 的数据盘（应为 %s），" + "拒绝挂载" % (data_disk, tenant, expected)) + if self.rootfs and actual == os.path.realpath(self.rootfs): + raise ChvFailure( + "task_disk_source_invalid", + "任务数据盘不能是基础镜像：租户 %s" % tenant) + pool_data = os.path.realpath( + data_disk_path(self.state_dir, vm)) + if actual == pool_data: + raise ChvFailure( + "task_disk_source_invalid", + "任务数据盘不能是干净池 VM 自带的盘：vm=%s，" + "租户 %s" % (vm, tenant)) + owner = read_vm_tenant( + self.state_dir, self._tenant_data_disk_name(tenant)) + if owner is not None and owner != tenant: + raise ChvFailure( + "task_disk_source_invalid", + "租户数据盘的归属标记是 %s，与请求方 %s 不一致，" + "拒绝挂载" % (owner, tenant)) + + def attach_tenant_data_disk(self, task: dict, vm: str, + data_disk: str | None = None) -> dict: + """分配时挂载租户数据盘（§3.1 首选方案）：供给/复用该租户 + 的数据盘 → 过任务盘来源不变量 → CHV 热挂载（PUT + /vm.add-disk，CHV 54 的热插盘端点）→ 客内挂载脚本 + （chv-data-mount，按 LABEL=CHVDATA 挂到 + /home/chv/workspace）→ 写 VM 归属标记。任一步失败即 + ChvFailure（fail-closed，调用方回落冷启或拒绝，不把 + 未挂盘的 VM 当已绑定发给任务）。返回 {vm, tenant_id, + data_disk}。""" + tenant = self._tenant_id(task) + if not tenant: + raise ChvFailure( + "task_disk_source_invalid", + "挂载数据盘须带 tenant_id：无归属的任务不能绑盘") + disk = data_disk or self._provision_tenant_data_disk(tenant) + self._verify_task_disk_source(vm, tenant, disk) + status, _body = self._api_call( + self.api_socket({"id": vm}), "PUT", "/vm.add-disk", + {"path": disk}) + if status not in (200, 204): + raise ChvFailure( + "data_disk_attach_failed", + "数据盘热挂载失败：vm=%s status=%s" + % (vm, status)) + result = self._call_runner(self.build_ssh_command( + {"id": vm}, ["/usr/local/bin/chv-data-mount"])) + if getattr(result, "returncode", 1) != 0: + raise ChvFailure( + "data_disk_mount_failed", + "客内挂载数据盘失败：vm=%s rc=%s" + % (vm, getattr(result, "returncode", None))) + write_vm_tenant(self.state_dir, vm, tenant, disk=disk) + return {"vm": vm, "tenant_id": tenant, "data_disk": disk} + + def detach_tenant_data_disk(self, task: dict, vm: str) -> bool: + """客内卸载租户数据盘（umount /home/chv/workspace，落盘 + 回外部存储）：成功 True；失败 False（调用方不得继续销毁 + VM，见 release_pool_vm——同步/卸载失败的 VM 不回收， + 等人工/重试，绝不丢改动后静默回收）。""" + try: + result = self._call_runner(self.build_ssh_command( + {"id": vm}, ["umount", "/home/chv/workspace"])) + except Exception: + return False + return getattr(result, "returncode", 1) == 0 + + def release_pool_vm(self, task: dict, vm: str, + sync_workspace: str | None = None) -> dict: + """池 VM 回收（§3.3）：同步回外部存储（sync_workspace 给 + 出时客内 chv-sync push，失败即不回收）→ 卸载租户数据盘 + （失败即不回收）→ 销毁 VM（stop_vm，用过的池 VM 一律 + 不回池）→ 清归属标记与干净池登记。返回 {released, vm, + tenant_id, reason?}。""" + tenant = self._tenant_id(task) + if sync_workspace: + try: + result = self._call_runner(self.build_ssh_command( + {"id": vm}, ["chv-sync", "push", sync_workspace])) + pushed = getattr(result, "returncode", 1) == 0 + except Exception: + pushed = False + if not pushed: + self._emit("warm_pool_release_failed", task, + "vm=%s host_id=%s tenant=%s " + "reason=sync_failed" + % (vm, self._host_id(), tenant), + "warm_pool_release_failed") + return {"released": False, "vm": vm, + "tenant_id": tenant, "reason": "sync_failed"} + if not self.detach_tenant_data_disk(task, vm): + self._emit("warm_pool_release_failed", task, + "vm=%s host_id=%s tenant=%s " + "reason=unmount_failed" + % (vm, self._host_id(), tenant), + "warm_pool_release_failed") + return {"released": False, "vm": vm, + "tenant_id": tenant, "reason": "unmount_failed"} + try: + self.stop_vm({"id": vm}) + except Exception: + pass + _tenant_path = os.path.join(self.state_dir, vm, "tenant.json") + try: + if os.path.exists(_tenant_path): + os.remove(_tenant_path) + except OSError: + pass + slot = self._clean_pool_slot_for_vm(vm) + if slot is not None: + self._clear_clean_pool_record(slot) + self._emit("warm_pool_release", task, + "vm=%s host_id=%s tenant=%s" + % (vm, self._host_id(), tenant)) + return {"released": True, "vm": vm, "tenant_id": tenant} + + # -- 干净池分配 ------------------------------------------------------ + def _clean_pool_candidate(self): + """找一台可分配的干净池 VM：登记 paused + socket 存在 + + vm.info 确认 Paused。socket 消失的登记当场清掉；状态非 + Paused 的跳过（不猜，留给清理）。返回 (slot, vm, record) + 或 None。""" + for slot, record in self._list_clean_pool_records(): + if record.get("state") != "paused": + continue + vm = record.get("vm") + if not vm: + continue + if not os.path.exists(self.api_socket({"id": vm})): + self._clear_clean_pool_record(slot) + continue + if self._vm_info_state({"id": vm}) != "Paused": + continue + return slot, vm, record + return None + + def _clean_pool_binding(self, task, vm, tenant): + """干净池分配的绑盘段：供给/复用租户数据盘 → 过任务盘 + 来源不变量。失败即发 warm_pool_acquire_failed 审计并 + 返回 None（调用方回落冷启，不报错）。""" + try: + data_disk = self._provision_tenant_data_disk(tenant) + self._verify_task_disk_source(vm, tenant, data_disk) + except ChvFailure as exc: + self._emit("warm_pool_acquire_failed", task, + "vm=%s host_id=%s tenant=%s kind=%s" + % (vm, self._host_id(), tenant, exc.kind), + "warm_pool_acquire_failed") + return None + return data_disk + + def _resume_clean_pool(self, task, vm, slot, tenant, data_disk): + """干净池分配的唤醒段：resume + 等 SSH 就绪 + 挂载租户 + 数据盘，计时返回 resume_seconds。失败即发 + warm_pool_acquire_failed 审计、销毁该 VM、清登记并 + 返回 None（调用方回落冷启）。""" + t0 = time.time() + try: + self.resume_vm({"id": vm}) + self.wait_guest_ready({"id": vm}) + self.attach_tenant_data_disk(task, vm, + data_disk=data_disk) + except Exception as exc: + self._emit("warm_pool_acquire_failed", task, + "vm=%s host_id=%s tenant=%s error=%s" + % (vm, self._host_id(), tenant, + str(exc)[:200]), + "warm_pool_acquire_failed") + try: + self.stop_vm({"id": vm}) + except Exception: + pass + self._clear_clean_pool_record(slot) + return None + return time.time() - t0 + + def _check_clean_pool_sync(self, vm, slot, sync_workspace) -> None: + """干净池分配的同步检查点（fail-closed 在此，不在预热时）： + sync_workspace 给出时等首次同步，等不到即解绑销毁该 VM + （已绑过租户态的 VM 绝不回干净池）再抛 + warm_pool_sync_incomplete，任务失败。""" + if not sync_workspace: + return + if not self._wait_sync_workspace({"id": vm}, sync_workspace): + try: + self.stop_vm({"id": vm}) + except Exception: + pass + self._clear_clean_pool_record(slot) + raise ChvFailure( + "warm_pool_sync_incomplete", + "池 VM %s 分配时在 %s 秒内未完成 Gitea 首次同步" + "（%s/.git 未出现），已解绑销毁，不派任务" + % (vm, self.ready_timeout, sync_workspace)) + + def acquire_clean_pool(self, task: dict, + extra_env: dict | None = None, + sync_workspace: str | None = None) -> dict: + """取一台干净池 VM（任务到达时的认领，§3.2 (a)）： + 命中 → resume + 等 SSH 就绪 + 挂载租户数据盘 + 首次同步 + 等待（sync_workspace 给出时），返回 {mode: resumed, vm, + data_disk, resume_seconds, warm_hit: True}； + 池空 / 无租户 → 发 warm_pool_miss 审计，返回 {mode: + cold, vm: None}，调用方回落冷启，不报错。 + resume_seconds 只计 resume+SSH 就绪+挂盘，不含租户态 + 同步时间（热恢复 ≤2 秒口径，§2.1）。 + warm_pool_sync_incomplete 的 fail-closed 语义保留，但 + 检查点从“预热时”移到“分配时”：同步等不到即解绑销毁 + 该 VM（已绑过租户态的 VM 绝不回干净池）再抛 + warm_pool_sync_incomplete，任务失败，不派给未同步的 VM。""" + if self.clean_pool_target() <= 0: + return {"mode": "cold", "vm": None} + tenant = self._tenant_id(task) + candidate = self._clean_pool_candidate() + if candidate is None or not tenant: + self._emit("warm_pool_miss", task, + "host_id=%s tenant=%s reason=%s" + % (self._host_id(), tenant, + "pool_empty" if candidate is None + else "tenant_missing")) + return {"mode": "cold", "vm": None} + slot, vm, record = candidate + data_disk = self._clean_pool_binding(task, vm, tenant) + if data_disk is None: + return {"mode": "cold", "vm": None} + resume_seconds = self._resume_clean_pool( + task, vm, slot, tenant, data_disk) + if resume_seconds is None: + return {"mode": "cold", "vm": None} + self._write_clean_pool_record(slot, { + "vm": vm, "slot": slot, "tenant_id": tenant, + "state": "running", "origin": "clean-prewarm", + "gitea_synced": False, "data_disk": data_disk, + "updated_at": time.time()}) + self._check_clean_pool_sync(vm, slot, sync_workspace) + self._emit("warm_pool_acquire", task, + "vm=%s host_id=%s tenant=%s mode=resumed " + "resume_seconds=%.2f warm_hit=true" + % (vm, self._host_id(), tenant, resume_seconds)) + return {"mode": "resumed", "vm": vm, + "data_disk": data_disk, + "resume_seconds": resume_seconds, + "warm_hit": True, "clean": True} + + # -- 池任务生命周期与混合入口 ---------------------------------------- + def _run_with_pool_vm(self, task, inner_args, goal=False, + extra_env=None, **kwargs): + """池任务生命周期（§3.2/§3.3）：命中预热 → 跑客内命令 → + finally 回收销毁（release_pool_vm，用过的池 VM 一律不 + 回池）；池空 → 冷启（_run_with_vm，跑完关机销毁）； + 无可用宿主 → 排队等槽位，300 秒超时后明确拒绝 + host_capacity_exceeded。warm_pool_sync_incomplete 从 + acquire 抛出时 fail-closed，不回落冷启。""" + acquired = self.acquire_clean_pool( + task, extra_env=extra_env, + sync_workspace=task.get("sync_workspace")) + if acquired["mode"] == "resumed": + vm = acquired["vm"] + 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["route"] = "pool" + result["allocation"] = "pool-hit" + result["vm_mode"] = "resumed" + result["warm_hit"] = True + result["resume_seconds"] = acquired.get( + "resume_seconds") + result["data_disk"] = acquired.get("data_disk") + return result + finally: + try: + self.release_pool_vm( + task, vm, + sync_workspace=task.get("sync_workspace")) + except Exception: + pass + if self._persistent_vm_count() >= self.max_vms: + if not self._wait_for_capacity(): + raise ChvFailure( + "host_capacity_exceeded", + "无可用宿主：已计数 %d 台，上限 %d，排队 %d 秒" + "后仍无槽位，明确拒绝" + % (self._persistent_vm_count(), self.max_vms, + HYBRID_QUEUE_TIMEOUT_SECONDS)) + result = self._run_with_vm( + task, inner_args, goal=goal, extra_env=extra_env, **kwargs) + if isinstance(result, dict): + result = dict(result) + result["route"] = "pool" + result["allocation"] = "pool-cold" + result["vm_mode"] = "cold" + return result + + def dispatch_task_hybrid(self, task: dict, + extra_env: dict | None = None, + model_provider: str | None = None) -> dict: + """混合路由任务入口（§9.1 第 1 项）：按 tier + 租约状态 + 分流——常驻租户走常驻 VM（resume/冷启常驻，任务间隙 + 暂停回常驻登记，不删盘），池租户走三路分配（命中预热 / + 冷启 / 排队）。warm_pool_size==0 时退化为 + dispatch_task_persistent（阶段 1 旧语义），与旧行为 + 一致。结果 dict 附 route（resident/pool）与 allocation + （resident-resumed/resident-cold/pool-hit/pool-cold）。""" + 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_persistent( + 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] + route = self._hybrid_route(task) + if route == "resident": + 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), + hybrid_resident=True, + hybrid_tier=str(task.get("tier") or "").lower()) + return self._run_with_pool_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 @@ -2568,7 +3457,74 @@ class CloudHypervisorExecutor(Executor): % (self.vm_name(task_or_name), snapshot_url, tenant)) return snapshot_url - def _check_snapshot_tenant(self, task, snapshot_url) -> None: + def _snapshot_strict_tenant(self) -> bool: + """快照 tenant 严格校验是否生效（feat-chv-hybrid-a， + DESIGN-HYBRID.md §2.2）：登记必填，无 tenant_id 的快照 + 视为非法登记，恢复时拒绝、不补默认值。生效条件：混合 + 模式已启用（干净池已开，即 CHV_WARM_POOL_SIZE>=1）或 + 显式 CHV_SNAPSHOT_STRICT_TENANT=1；显式设 0 即关闭。 + 未启用时沿用阶段 1 的 legacy 警告放行——旧 restore + 测试（test_chv_closeup RestoreSnapshotTest、 + test_chv_fix RestoreNoFallbackTest）依赖该语义，全量 + 严格默认须先更新那些旧测试，本轨不擅自改旧测试文件， + 故严格校验以混合模式开关为界，未决项见 A 轨汇报。""" + raw = self._env.get("CHV_SNAPSHOT_STRICT_TENANT") + if raw == "1": + return True + if raw == "0": + return False + return self.clean_pool_target() > 0 + + def _deny_restore(self, task, reason: str) -> None: + """严格模式的恢复拒绝（fail-closed）：审计 vm_restore_denied + 只带 host_id + 请求租户 + 原因，不带快照内容（不带 URL、 + 不带归属 tenant 值）；错误码一律 snapshot_tenant_mismatch。""" + requester = self._tenant_id(task) + self._emit("vm_restore_denied", task, + "host_id=%s tenant=%s reason=%s" + % (self._host_id(), requester, reason), + "snapshot_tenant_mismatch") + raise ChvFailure( + "snapshot_tenant_mismatch", + "快照 tenant 校验失败（%s），拒绝恢复：请求方 " + "tenant=%s" % (reason, requester)) + + def _warn_restore_unmarked(self, task, snapshot_url, + what: str) -> None: + """未登记 / 登记无 tenant 的恢复警告（legacy 放行路径）： + 发 vm_restore_tenant_unmarked 审计警告，不拒绝。what 为 + "unregistered"（未在登记中找到）或 "unmarked"（登记无 + tenant 字段）。""" + if what == "unregistered": + detail = ("vm=%s snapshot=%s 未在登记中找到，归属无法" + "核对（未登记快照，放行并警告）" + % (self.vm_name(task), snapshot_url)) + else: + detail = ("vm=%s snapshot=%s 登记无 tenant 字段" + "（legacy 旧快照，放行并警告）" + % (self.vm_name(task), snapshot_url)) + self._emit("vm_restore_tenant_unmarked", task, detail, + "snapshot_tenant_unmarked") + + def _deny_cross_tenant_restore(self, task, snapshot_url, marked, + requester) -> None: + """跨租户恢复拒绝（legacy 路径的旧审计名）：发 + vm_restore_tenant_denied 审计并抛 + snapshot_tenant_mismatch。""" + 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 _check_snapshot_tenant(self, task, snapshot_url, + strict: bool | None = None) -> None: """restore 的归属校验（feat-chv-opt-landing，阶段 1 必须补的代码项）：按 URL 在全部 VM 的快照登记里找 条目；条目带 tenant 且与请求方 tenant 不一致时， @@ -2580,39 +3536,37 @@ class CloudHypervisorExecutor(Executor): 未登记：同样放行 + 警告（未登记快照无从核对归属， 登记才是归属的唯一账本——生产须要求快照必登记， 见报告残留）。请求方无 tenant（None）而条目有 - tenant 时同样拒绝（无法证明归属即 fail-closed）。""" + tenant 时同样拒绝（无法证明归属即 fail-closed）。 + strict=True（feat-chv-hybrid-a，§2.2）：登记必填—— + 未登记 / 登记无 tenant / 归属不符一律拒绝 + （snapshot_tenant_mismatch），审计 vm_restore_denied + （带 host_id + 请求租户，不带快照内容），不补默认值。 + strict=None 时按 _snapshot_strict_tenant 自动判定。""" + if strict is None: + strict = self._snapshot_strict_tenant() _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") + if strict: + self._deny_restore(task, "unregistered") + self._warn_restore_unmarked(task, snapshot_url, + "unregistered") 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") + if strict: + self._deny_restore(task, "unmarked") + self._warn_restore_unmarked(task, snapshot_url, "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)) + if strict: + self._deny_restore(task, "mismatch") + self._deny_cross_tenant_restore(task, snapshot_url, + marked, requester) - def _validate_restore(self, task, snapshot_url): + def _validate_restore(self, task, snapshot_url, + strict_tenant: bool | None = None): """restore 前置校验：返回（归一化后的 snapshot_url， 归一化后的 net 条目）。 @@ -2639,7 +3593,8 @@ 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_tenant(task, snapshot_url, + strict=strict_tenant) self._check_snapshot_disk(snapshot_url) self.preflight() self.resolve_ssh_host() diff --git a/cloud/tests/test_chv_hybrid_a.py b/cloud/tests/test_chv_hybrid_a.py new file mode 100644 index 0000000..9cd4499 --- /dev/null +++ b/cloud/tests/test_chv_hybrid_a.py @@ -0,0 +1,688 @@ +"""CHV 混合模式 A 轨（feat-chv-hybrid-a）测试。 + +只用标准库与 mock runner/launcher/api_client，绝不启动真实 +VM（与 test_chv_opt_stage1.py 同款边界）。覆盖 DESIGN-HYBRID.md +§9.1 第 1–4 项： +- 混合路由分流（常驻 / 池命中 / 池空冷启 / 排队拒绝） +- 干净池预热不带租户（无数据盘、无 seed、无同步等待） +- 数据盘挂载归属 + “任务盘来源”不变量 +- snapshot_tenant_mismatch（含无 tenant_id 登记拒绝） +- 已定值常量（每宿主预热 2 台、池上限 50%、常驻上限 50%、 + 排队超时 300 秒、热恢复 ≤2 秒、冷恢复 ≤30 秒、快照 1 个/30 天） + +不改其它测试文件；阶段 1 的旧语义（prewarm_persistent_vm / +acquire_for_task / dispatch_task_persistent / _validate_restore +的 legacy 警告放行）原样保留，只测本轨新增行为。 +""" + +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.disk import ( + read_persistent_vm, + read_vm_tenant, + record_snapshot, + write_persistent_vm, + ) + from cloud.chv.executor import ( + HYBRID_COLD_RESTORE_TARGET_SECONDS, + HYBRID_QUEUE_TIMEOUT_SECONDS, + HYBRID_SNAPSHOT_PER_TENANT, + HYBRID_SNAPSHOT_RETENTION_DAYS, + HYBRID_WARM_POOL_PER_HOST, + HYBRID_WARM_RESUME_TARGET_SECONDS, + ) +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 ( + read_persistent_vm, + read_vm_tenant, + record_snapshot, + write_persistent_vm, + ) + from cloud.chv.executor import ( + HYBRID_COLD_RESTORE_TARGET_SECONDS, + HYBRID_QUEUE_TIMEOUT_SECONDS, + HYBRID_SNAPSHOT_PER_TENANT, + HYBRID_SNAPSHOT_RETENTION_DAYS, + HYBRID_WARM_POOL_PER_HOST, + HYBRID_WARM_RESUME_TARGET_SECONDS, + ) + + +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 + + +class HybridACase(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 = {} + self.events = [] + + 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_WARM_POOL_SIZE": "2"} + 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) + ex = CloudHypervisorExecutor(**kwargs) + ex.audit_callback = lambda e, t, d, k: self.events.append( + (e, d, k)) + return ex + + 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", tier=None): + task = {"id": task_id, "prompt": "做一个加法函数", + "project_id": project} + if tier is not None: + task["tier"] = tier + return task + + 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 _event_names(self): + return [e for e, _d, _k in self.events] + + def _event_details(self, name): + return [d for e, d, _k in self.events if e == name] + + +# -- 已定值常量 ------------------------------------------------------------- +class SettledValuesTest(HybridACase): + def test_hybrid_settled_values(self): + ex = self.make_exec() + self.assertEqual(HYBRID_WARM_POOL_PER_HOST, 2) + self.assertEqual(ex.warm_pool_per_host_target(), 2) + self.assertEqual(ex.clean_pool_target(), 2) + # 池上限 = 低频并发峰值的 50%：20 → 10，50 → 25。 + self.assertEqual(ex.warm_pool_cap(20), 10) + self.assertEqual(ex.warm_pool_cap(50), 25) + self.assertIsNone(ex.warm_pool_cap(None)) + # 生效目标取 min（配置目标，上限）：目标 2、峰值 20 → 2。 + self.assertEqual(ex.clean_pool_effective_target(20), 2) + ex_big = self.make_exec(env_extra={ + "CHV_WARM_POOL_SIZE": "30", + "CHV_LOW_FREQ_PEAK_CONCURRENCY": "20"}) + self.assertEqual(ex_big.clean_pool_effective_target(), 10) + # 常驻上限 = 槽位 50%：4 槽 → 2 台，48 槽 → 24 台。 + self.assertEqual(ex.resident_cap(), 2) + ex48 = self.make_exec(env_extra={"CHV_MAX_VMS": "48"}) + self.assertEqual(ex48.resident_cap(), 24) + # 排队超时 300 秒；热恢复 ≤2 秒、冷快照恢复 ≤30 秒（目标值， + # 只作常量/注释，不在单元测试里硬测时延）；快照 1 个/30 天。 + self.assertEqual(HYBRID_QUEUE_TIMEOUT_SECONDS, 300) + self.assertEqual(HYBRID_WARM_RESUME_TARGET_SECONDS, 2) + self.assertEqual(HYBRID_COLD_RESTORE_TARGET_SECONDS, 30) + self.assertEqual(HYBRID_SNAPSHOT_PER_TENANT, 1) + self.assertEqual(HYBRID_SNAPSHOT_RETENTION_DAYS, 30) + + +# -- 混合路由分流 ----------------------------------------------------------- +class HybridRouteTest(HybridACase): + def test_route_by_tier(self): + ex = self.make_exec() + self.assertEqual( + ex._hybrid_route(self._task(tier="paid")), "resident") + self.assertEqual( + ex._hybrid_route(self._task(tier="heavy")), "resident") + self.assertEqual( + ex._hybrid_route(self._task(tier="PAID")), "resident") + self.assertEqual( + ex._hybrid_route(self._task(tier="free")), "pool") + self.assertEqual(ex._hybrid_route(self._task()), "pool") + # 常驻是权益门：Paid 但无 tenant（无归属）即 fail-closed 走池。 + task = {"id": "x1", "prompt": "p", "tier": "paid"} + self.assertEqual(ex._hybrid_route(task), "pool") + + def test_dispatch_hybrid_validates_task(self): + ex = self.make_exec() + with self.assertRaises(ValueError): + ex.dispatch_task_hybrid({"prompt": "p"}) + with self.assertRaises(ValueError): + ex.dispatch_task_hybrid({"id": "x"}) + + def test_dispatch_hybrid_pool_disabled_falls_back(self): + ex = self.make_exec(env_extra={"CHV_WARM_POOL_SIZE": "0"}) + out = ex.dispatch_task_hybrid(self._task()) + self.assertEqual(out["status"], "completed") + # 退化为阶段 1 旧语义：结果不带 route/allocation。 + self.assertNotIn("route", out) + self.assertEqual(ex._vms, {}) + + def test_dispatch_pool_cold_when_pool_empty(self): + ex = self.make_exec() + out = ex.dispatch_task_hybrid(self._task(task_id="t1")) + self.assertEqual(out["status"], "completed") + self.assertEqual(out["route"], "pool") + self.assertEqual(out["allocation"], "pool-cold") + self.assertEqual(out["vm_mode"], "cold") + self.assertIn("warm_pool_miss", self._event_names()) + self.assertEqual(ex._vms, {}) + + def test_dispatch_pool_queue_timeout_refuses(self): + ex = self.make_exec(env_extra={ + "CHV_QUEUE_TIMEOUT": "0", "CHV_MAX_VMS": "1"}) + # 宿主已满（1 台活 VM 占满唯一槽位），池空 → 排队 0 秒即拒绝。 + ex._vms["chv-busy"] = FakeProc(alive=True) + with self.assertRaises(ChvFailure) as ctx: + ex.dispatch_task_hybrid(self._task(task_id="t1")) + self.assertEqual(ctx.exception.kind, "host_capacity_exceeded") + self.assertEqual(self.launches, []) + self.assertIn("warm_pool_miss", self._event_names()) + + def test_dispatch_resident_cold_when_no_resident_vm(self): + ex = self.make_exec() + out = ex.dispatch_task_hybrid( + self._task(task_id="t1", project="paid1", tier="paid")) + self.assertEqual(out["status"], "completed") + self.assertEqual(out["route"], "resident") + self.assertEqual(out["allocation"], "resident-cold") + self.assertEqual(out["vm"], "chv-persist-paid1") + rec = read_persistent_vm(self.state_dir, "paid1") + self.assertEqual(rec["state"], "paused") + self.assertEqual(rec["tier"], "paid") + self.assertTrue(rec["resident"]) + self.assertEqual(ex._resident_count(), 1) + + def test_dispatch_resident_resumed_from_prewarm(self): + ex = self.make_exec() + ex.prewarm_persistent_vm("paid1") + self._socket("chv-persist-paid1") + self.vm_states["chv-persist-paid1"] = "Paused" + out = ex.dispatch_task_hybrid( + self._task(task_id="t1", project="paid1", tier="paid")) + self.assertEqual(out["status"], "completed") + self.assertEqual(out["route"], "resident") + self.assertEqual(out["allocation"], "resident-resumed") + self.assertEqual(out["vm"], "chv-persist-paid1") + self.assertIn("resume_seconds", out) + rec = read_persistent_vm(self.state_dir, "paid1") + self.assertEqual(rec["state"], "paused") + self.assertEqual(rec["tier"], "paid") + self.assertTrue(rec["resident"]) + + def test_resident_cap_blocks_new_resident(self): + ex = self.make_exec(env_extra={"CHV_QUEUE_TIMEOUT": "0"}) + # 4 槽位 → 常驻上限 2 台：已有 2 个常驻租户，第 3 个排队即拒。 + for tenant in ("r1", "r2"): + write_persistent_vm(self.state_dir, tenant, { + "vm": "chv-persist-%s" % tenant, + "tenant_id": tenant, "state": "paused", + "origin": "task", "gitea_synced": False, + "tier": "paid", "resident": True}) + self.assertEqual(ex._resident_count(), 2) + with self.assertRaises(ChvFailure) as ctx: + ex.dispatch_task_hybrid( + self._task(task_id="t1", project="r3", tier="paid")) + self.assertEqual(ctx.exception.kind, "host_capacity_exceeded") + self.assertEqual(self.launches, []) + # 已有常驻的租户不受上限挡（自己的槽位）。 + ex2 = self.make_exec(env_extra={"CHV_QUEUE_TIMEOUT": "0"}) + write_persistent_vm(self.state_dir, "r1", { + "vm": "chv-persist-r1", "tenant_id": "r1", + "state": "paused", "origin": "task", + "gitea_synced": False, "tier": "paid", "resident": True}) + write_persistent_vm(self.state_dir, "r2", { + "vm": "chv-persist-r2", "tenant_id": "r2", + "state": "paused", "origin": "task", + "gitea_synced": False, "tier": "paid", "resident": True}) + self._socket("chv-persist-r1") + self.vm_states["chv-persist-r1"] = "Paused" + out = ex2.dispatch_task_hybrid( + self._task(task_id="t1", project="r1", tier="paid")) + self.assertEqual(out["route"], "resident") + self.assertEqual(out["allocation"], "resident-resumed") + + +# -- 干净池预热 ------------------------------------------------------------- +class CleanPoolPrewarmTest(HybridACase): + def test_prewarm_clean_pool_has_no_tenant_state(self): + ex = self.make_exec() + out = ex.prewarm_clean_pool() + self.assertTrue(out["pooled"]) + self.assertEqual(out["mode"], "prewarmed") + self.assertEqual(out["vm"], "chv-pool-0") + self.assertFalse(out["gitea_synced"]) + self.assertTrue(out["clean"]) + self.assertIsNone(out["tenant_id"]) + self.assertEqual(len(self.launches), 1) + vm_dir = os.path.join(self.state_dir, "chv-pool-0") + # 出生时不带租户：无数据盘、无 tenant 标记、无 seed。 + self.assertFalse(os.path.exists( + os.path.join(vm_dir, "data.raw"))) + self.assertIsNone(read_vm_tenant(self.state_dir, "chv-pool-0")) + self.assertFalse(os.path.exists( + os.path.join(vm_dir, "seed.iso"))) + self.assertFalse(os.path.exists( + os.path.join(vm_dir, "seed"))) + # 无 Gitea 同步等待：预热全程没有任何 ssh test -d 调用。 + self.assertFalse(any( + "test" in call for call in self.calls + if call and call[0] == "ssh")) + # 登记同样不带租户。 + record = ex._read_clean_pool_record(0) + self.assertEqual(record["state"], "paused") + self.assertIsNone(record["tenant_id"]) + self.assertFalse(record["gitea_synced"]) + # socket 造出来后第二次即 already-warm，不重建。 + self._socket("chv-pool-0") + second = ex.prewarm_clean_pool() + self.assertEqual(second["mode"], "already-warm") + self.assertEqual(len(self.launches), 1) + + def test_prewarm_clean_pool_slots_and_range(self): + ex = self.make_exec() + first = ex.prewarm_clean_pool(slot=1) + self.assertEqual(first["vm"], "chv-pool-1") + out = ex.prewarm_clean_pool(slot=5) + self.assertFalse(out["pooled"]) + self.assertEqual(out["reason"], "clean_pool_slot_out_of_range") + self.assertEqual(len(self.launches), 1) + + def test_prewarm_clean_pool_disabled_when_size_zero(self): + ex = self.make_exec(env_extra={"CHV_WARM_POOL_SIZE": "0"}) + self.assertEqual( + ex.prewarm_clean_pool()["reason"], "warm_pool_disabled") + self.assertEqual( + ex.acquire_clean_pool(self._task())["mode"], "cold") + self.assertEqual(ex.refill_clean_pool(), {"target": 0, + "slots": []}) + + def test_prewarm_clean_pool_refuses_tenant_seed(self): + ex = self.make_exec() + with self.assertRaises(ChvFailure) as ctx: + ex.prewarm_clean_pool( + extra_env={"GITEA_TOKEN": "secret-value"}) + self.assertEqual(ctx.exception.kind, "clean_pool_seed_refused") + with self.assertRaises(ChvFailure) as ctx: + ex.prewarm_clean_pool( + extra_env={"GITEA_URL": "https://gitea.tose.top"}) + self.assertEqual(ctx.exception.kind, "clean_pool_seed_refused") + self.assertEqual(self.launches, []) + + def test_prewarm_clean_pool_never_rebuilds_running_slot(self): + ex = self.make_exec() + ex._write_clean_pool_record(0, { + "vm": "chv-pool-0", "slot": 0, "tenant_id": "tenant-a", + "state": "running", "origin": "clean-prewarm", + "gitea_synced": False, "data_disk": "x", + "updated_at": 0}) + out = ex.prewarm_clean_pool(slot=0) + self.assertFalse(out["pooled"]) + self.assertEqual(out["reason"], "clean_pool_slot_busy") + self.assertEqual(self.launches, []) + # 自动选槽跳过正跑的 slot 0，建 slot 1。 + out2 = ex.prewarm_clean_pool() + self.assertEqual(out2["vm"], "chv-pool-1") + self.assertEqual(len(self.launches), 1) + + def test_refill_clean_pool(self): + ex = self.make_exec() + ex.prewarm_clean_pool(slot=0) + self._socket("chv-pool-0") + out = ex.refill_clean_pool() + self.assertEqual(out["target"], 2) + self.assertEqual(out["slots"][0]["mode"], "already-warm") + self.assertEqual(out["slots"][1]["mode"], "prewarmed") + self.assertEqual(len(self.launches), 2) + + +# -- 数据盘挂载归属 + 任务盘来源不变量 --------------------------------------- +class DataDiskAttachTest(HybridACase): + def test_verify_task_disk_source(self): + ex = self.make_exec() + tenant = "tenant-a" + disk = ex._provision_tenant_data_disk(tenant) + self.assertEqual(disk, ex._tenant_data_disk_path(tenant)) + self.assertTrue(os.path.exists(disk)) + # 合法的租户数据盘通过。 + ex._verify_task_disk_source("chv-pool-0", tenant, disk) + # 干净池 VM 自带的盘（干净快照盘）绝不能当任务盘。 + pool_data = os.path.join(self.state_dir, "chv-pool-0", + "data.raw") + with self.assertRaises(ChvFailure) as ctx: + ex._verify_task_disk_source("chv-pool-0", tenant, pool_data) + self.assertEqual(ctx.exception.kind, "task_disk_source_invalid") + # 基础镜像不能当任务盘。 + with self.assertRaises(ChvFailure) as ctx: + ex._verify_task_disk_source("chv-pool-0", tenant, + self.rootfs) + self.assertEqual(ctx.exception.kind, "task_disk_source_invalid") + # 其它租户的盘不能当本租户的任务盘。 + other = ex._provision_tenant_data_disk("tenant-b") + with self.assertRaises(ChvFailure) as ctx: + ex._verify_task_disk_source("chv-pool-0", tenant, other) + self.assertEqual(ctx.exception.kind, "task_disk_source_invalid") + # 无盘 / 无租户同样拒绝。 + with self.assertRaises(ChvFailure) as ctx: + ex._verify_task_disk_source("chv-pool-0", tenant, None) + self.assertEqual(ctx.exception.kind, "task_disk_source_invalid") + with self.assertRaises(ChvFailure) as ctx: + ex._verify_task_disk_source("chv-pool-0", None, disk) + self.assertEqual(ctx.exception.kind, "task_disk_source_invalid") + + def test_attach_refuses_wrong_disk_before_api_call(self): + ex = self.make_exec() + with self.assertRaises(ChvFailure) as ctx: + ex.attach_tenant_data_disk( + self._task(project="tenant-a"), "chv-pool-0", + data_disk=self.rootfs) + self.assertEqual(ctx.exception.kind, "task_disk_source_invalid") + # 不变量在 API 调用之前拦住：一次 add-disk 都没发。 + self.assertFalse(any(p == "/vm.add-disk" + for _m, p, _pl in self.api_calls)) + + def test_attach_tenant_data_disk(self): + ex = self.make_exec() + out = ex.attach_tenant_data_disk( + self._task(project="tenant-a"), "chv-pool-0") + self.assertEqual(out["tenant_id"], "tenant-a") + self.assertEqual(out["data_disk"], + ex._tenant_data_disk_path("tenant-a")) + self.assertTrue(any(p == "/vm.add-disk" + for _m, p, _pl in self.api_calls)) + _method, _path, payload = next( + (m, p, pl) for m, p, pl in self.api_calls + if p == "/vm.add-disk") + self.assertEqual(payload["path"], out["data_disk"]) + # 客内挂载脚本被调过（chv-data-mount）。 + self.assertTrue(any( + "/usr/local/bin/chv-data-mount" in call + for call in self.calls if call and call[0] == "ssh")) + self.assertEqual( + read_vm_tenant(self.state_dir, "chv-pool-0"), "tenant-a") + + def test_dispatch_pool_hit_binds_tenant_disk_and_destroys_vm(self): + ex = self.make_exec() + ex.prewarm_clean_pool(slot=0) + self._socket("chv-pool-0") + self.vm_states["chv-pool-0"] = "Paused" + out = ex.dispatch_task_hybrid( + self._task(task_id="t1", project="tenant-a")) + self.assertEqual(out["status"], "completed") + self.assertEqual(out["route"], "pool") + self.assertEqual(out["allocation"], "pool-hit") + self.assertEqual(out["vm"], "chv-pool-0") + self.assertTrue(out["warm_hit"]) + self.assertEqual(out["data_disk"], + ex._tenant_data_disk_path("tenant-a")) + self.assertIn("resume_seconds", out) + self.assertIn("warm_pool_acquire", self._event_names()) + self.assertIn("warm_pool_release", self._event_names()) + # 用过的池 VM 已销毁、不回池：登记清掉、VMM 登记空、归属清掉。 + self.assertIsNone(ex._read_clean_pool_record(0)) + self.assertEqual(ex._vms, {}) + self.assertIsNone(read_vm_tenant(self.state_dir, "chv-pool-0")) + # 租户数据盘还在（用户数据不随 VM 销毁而删）。 + self.assertTrue(os.path.exists( + ex._tenant_data_disk_path("tenant-a"))) + # 池 VM 自带的干净盘从未被当成任务盘建出来。 + self.assertFalse(os.path.exists(os.path.join( + self.state_dir, "chv-pool-0", "data.raw"))) + + def test_acquire_sync_incomplete_fails_closed_at_allocate(self): + # 分配时同步等不到：fail-closed，VM 解绑销毁，不派任务。 + def runner(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 "test" in cmd: + return FakeResult(1, "", "") + if cmd[-1] == "true": + return FakeResult(0, "", "") + if "dispatch" in cmd: + return FakeResult(0, json.dumps( + {"status": "completed", "text": "ok"}), "") + return FakeResult(0, "", "") + return FakeResult(0, "", "") + + ex = self.make_exec(runner=runner) + ex.prewarm_clean_pool(slot=0) + self._socket("chv-pool-0") + self.vm_states["chv-pool-0"] = "Paused" + task = self._task(task_id="t1", project="tenant-a") + task["sync_workspace"] = "/home/chv/workspace/repo" + with self.assertRaises(ChvFailure) as ctx: + ex.dispatch_task_hybrid(task) + self.assertEqual(ctx.exception.kind, + "warm_pool_sync_incomplete") + # 已绑过租户态的 VM 绝不回干净池：已销毁、登记清掉。 + self.assertIsNone(ex._read_clean_pool_record(0)) + self.assertEqual(ex._vms, {}) + self.assertFalse(any("dispatch" in call + for call in self.calls + if call and call[0] == "ssh")) + + def test_release_sync_failure_keeps_vm(self): + # 回收时 chv-sync push 失败：VM 不回收、不复用。 + def runner(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 "chv-sync" in cmd: + return FakeResult(1, "", "push 被拒绝") + if cmd[-1] == "true": + return FakeResult(0, "", "") + return FakeResult(0, "", "") + return FakeResult(0, "", "") + + ex = self.make_exec(runner=runner) + ex.prewarm_clean_pool(slot=0) + self._socket("chv-pool-0") + self.vm_states["chv-pool-0"] = "Paused" + acquired = ex.acquire_clean_pool(self._task(project="tenant-a")) + self.assertEqual(acquired["mode"], "resumed") + out = ex.release_pool_vm( + self._task(project="tenant-a"), "chv-pool-0", + sync_workspace="/home/chv/workspace/repo") + self.assertFalse(out["released"]) + self.assertEqual(out["reason"], "sync_failed") + self.assertIn("warm_pool_release_failed", self._event_names()) + # VM 还在（未销毁），登记仍是 running 绑定态。 + self.assertIn("chv-pool-0", ex._vms) + record = ex._read_clean_pool_record(0) + self.assertEqual(record["state"], "running") + self.assertEqual(record["tenant_id"], "tenant-a") + + +# -- 快照 tenant 校验 -------------------------------------------------------- +class SnapshotTenantStrictTest(HybridACase): + def _snap_url(self, vm, name="snap1"): + snap_dir = os.path.join(self.state_dir, vm, name) + os.makedirs(snap_dir, exist_ok=True) + with open(os.path.join(snap_dir, "config.json"), "w") as fh: + fh.write("{}") + with open(os.path.join(snap_dir, "state.json"), "w") as fh: + fh.write("{}") + return "file://" + snap_dir + + def _record(self, vm, url, tenant): + record_snapshot(self.state_dir, vm, url, 20.0, + tenant_id=tenant) + + def test_unmarked_snapshot_denied_in_hybrid_mode(self): + url = self._snap_url("chv-a") + self._record("chv-a", url, None) + ex = self.make_exec() # CHV_WARM_POOL_SIZE=2 → 严格校验生效 + with self.assertRaises(ChvFailure) as ctx: + ex.restore_vm({"id": "task-a", "project_id": "tenant-a"}, + url) + self.assertEqual(ctx.exception.kind, + "snapshot_tenant_mismatch") + denied = [d for _e, d, _k in self.events + if _e == "vm_restore_denied"] + self.assertTrue(denied) + # 审计带 host_id + 请求租户，不带快照内容（URL 不得出 + # 现在 detail 里）。 + self.assertIn("host_id=", denied[0]) + self.assertIn("tenant=tenant-a", denied[0]) + self.assertNotIn(url, denied[0]) + self.assertNotIn("snap1", denied[0]) + + def test_unregistered_snapshot_denied_in_hybrid_mode(self): + url = self._snap_url("chv-a") + ex = self.make_exec() + with self.assertRaises(ChvFailure) as ctx: + ex.restore_vm({"id": "task-a", "project_id": "tenant-a"}, + url) + self.assertEqual(ctx.exception.kind, + "snapshot_tenant_mismatch") + self.assertIn("vm_restore_denied", self._event_names()) + + def test_mismatched_snapshot_denied_in_hybrid_mode(self): + url = self._snap_url("chv-a") + self._record("chv-a", url, "tenant-a") + ex = self.make_exec() + with self.assertRaises(ChvFailure) as ctx: + ex.restore_vm({"id": "task-b", "project_id": "tenant-b"}, + url) + self.assertEqual(ctx.exception.kind, + "snapshot_tenant_mismatch") + self.assertIn("vm_restore_denied", self._event_names()) + + def test_same_tenant_snapshot_restores_in_hybrid_mode(self): + url = self._snap_url("chv-a") + self._record("chv-a", url, "tenant-a") + ex = self.make_exec() + out = ex.restore_vm( + {"id": "task-a", "project_id": "tenant-a"}, url) + self.assertEqual(out["mode"], "restored") + self.assertEqual(out["state"], "running") + + def test_explicit_strict_flag_denies_unmarked(self): + url = self._snap_url("chv-a") + self._record("chv-a", url, None) + ex = self.make_exec(env_extra={"CHV_WARM_POOL_SIZE": "0"}) + # 显式 strict_tenant=True 时即使池未启用也拒绝。 + with self.assertRaises(ChvFailure) as ctx: + ex._validate_restore( + {"id": "task-a", "project_id": "tenant-a"}, url, + strict_tenant=True) + self.assertEqual(ctx.exception.kind, + "snapshot_tenant_mismatch") + self.assertIn("vm_restore_denied", self._event_names()) + + def test_legacy_warn_pass_preserved_when_not_strict(self): + # 混合模式未启用时，_validate_restore 沿用阶段 1 的 + # legacy 警告放行（旧测试依赖该语义，本轨不改）。 + url = self._snap_url("chv-a") + self._record("chv-a", url, None) + ex = self.make_exec(env_extra={"CHV_WARM_POOL_SIZE": "0"}) + snapshot_url, _net = ex._validate_restore( + {"id": "task-a", "project_id": "tenant-a"}, url) + self.assertEqual(snapshot_url, url) + self.assertIn("vm_restore_tenant_unmarked", + self._event_names()) + self.assertNotIn("vm_restore_denied", self._event_names()) + + +if __name__ == "__main__": + unittest.main()