From 971e0549383a378043b8281ee255ef6a3c7a7fce Mon Sep 17 00:00:00 2001 From: muse-agent Date: Mon, 2 Nov 2026 03:25:49 +0000 Subject: [PATCH] =?UTF-8?q?fix(chv):=20=E7=94=9F=E4=BA=A7=E7=BA=A7?= =?UTF-8?q?=E5=AE=A1=E6=A0=B8=20P1-01=EF=BD=9EP1-10=20=E4=BF=AE=E5=A4=8D?= =?UTF-8?q?=EF=BC=88=E5=A0=86=E5=8F=A0=E4=BA=8E=20#270=20=E4=B9=8B?= =?UTF-8?q?=E4=B8=8A=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cloud/chv/config.py | 35 +- cloud/chv/disk.py | 102 +++- cloud/chv/executor.py | 240 +++++++-- cloud/chv/experimental_pool.py | 19 +- cloud/chv/locking.py | 107 ++++ cloud/chv/orphans.py | 135 ++++- cloud/tests/test_chv_arch_maint_fe.py | 28 +- cloud/tests/test_chv_closeup.py | 28 +- cloud/tests/test_chv_cover100.py | 9 +- cloud/tests/test_chv_p1fix.py | 730 ++++++++++++++++++++++++++ docs/chv-architecture.md | 34 +- docs/chv-production-readiness.md | 198 +++++++ 12 files changed, 1545 insertions(+), 120 deletions(-) create mode 100644 cloud/chv/locking.py create mode 100644 cloud/tests/test_chv_p1fix.py diff --git a/cloud/chv/config.py b/cloud/chv/config.py index 693cfe1..852adc6 100644 --- a/cloud/chv/config.py +++ b/cloud/chv/config.py @@ -80,15 +80,39 @@ class ChvConfig: def _base_fragment(params: dict, env: dict) -> dict: """身份/资源基底片段：二进制、内核、根盘、CPU/内存、 state_dir、KVM/cpuinfo 路径、ch-remote、模型。无校验， - 取值口径与旧 build_chv_config 内联 base 逐字一致。""" + 取值口径与旧 build_chv_config 内联 base 逐字一致， + 只有一处新增（feat-chv-p1fix，P1-07）：能解析出的 + CPU/内存非正值直接 ValueError，不再一路进 VmConfig + 到 VMM 侧才炸；解析不出的格式仍留给 preflight 报 + config_invalid（构造时不改旧报错时机分工）。""" + # 显式 cpus=0 不再经 or 链静默回落成 2：0 是非法值， + # 应报出来（P1-07）；None 仍走 env/默认值。 + _cpus_raw = params.get("cpus") if params.get("cpus") is not None \ + else (env.get("CHV_CPU") or "2") + cpus = str(_cpus_raw) + # 解析不出的格式留给 preflight（旧时机分工不变）： + # 只对“解析成功且 <=0”的值在这里报，格式错误不报。 + try: + _cpus_parsed = int(cpus) + except (TypeError, ValueError): + _cpus_parsed = None + if _cpus_parsed is not None and _cpus_parsed <= 0: + raise ValueError("CHV_CPU 必须是大于 0 的整数") + memory = params.get("memory") or env.get("CHV_MEM") or "2G" + try: + _mem_parsed = _parse_memory_bytes(memory) + except ValueError: + _mem_parsed = None + if _mem_parsed is not None and _mem_parsed <= 0: + raise ValueError("CHV_MEM 必须是大于 0") return { "binary": params.get("binary") or env.get("CHV_BINARY") or "cloud-hypervisor", "kernel": params.get("kernel") or env.get("CHV_KERNEL"), "rootfs": params.get("rootfs") or params.get("image") or env.get("CHV_ROOTFS") or env.get("CHV_IMAGE"), - "cpus": str(params.get("cpus") or env.get("CHV_CPU") or "2"), - "memory": params.get("memory") or env.get("CHV_MEM") or "2G", + "cpus": cpus, + "memory": memory, "state_dir": params.get("state_dir") or env.get("CHV_STATE_DIR") or os.path.join(os.path.expanduser("~"), @@ -221,6 +245,11 @@ def _timeout_fragment(params: dict, env: dict) -> dict: task_timeout_raw = env.get("CHV_TASK_TIMEOUT") task_timeout = _to_float(task_timeout_raw) \ if task_timeout_raw else None + # feat-chv-p1fix（P1-07）：显式设的任务超时非正即非法。 + # None = 不限（旧语义，保留）；0/负数过去一路进 VmConfig + # 相关路径（runner timeout=0/负），现在构造时即 ValueError。 + if task_timeout is not None and task_timeout <= 0: + raise ValueError("CHV_TASK_TIMEOUT 必须是大于 0 的数字") try: api_timeout = _to_float( params.get("api_timeout") diff --git a/cloud/chv/disk.py b/cloud/chv/disk.py index 68addf6..a5323bc 100644 --- a/cloud/chv/disk.py +++ b/cloud/chv/disk.py @@ -165,10 +165,21 @@ def _save_snapshots(state_dir, vm_name, entries): def prune_snapshots(state_dir: str, vm_name: str, retention_days: float | None, now: float | None = None, - delete_files: bool = True) -> tuple[list, list]: + delete_files: bool = True, + events: list | None = None) -> tuple[list, list]: """按保留期剪枝：删过期条目；delete_files 时同时删 对应的 file:// 快照文件（只删登记里记的 path，不猜 - 文件名）。返回（保留条目，剪掉条目）。""" + 文件名）。返回（保留条目，剪掉条目）。 + + feat-chv-p1fix（P1-08）：删文件动作不再静默—— + events 非空列表时，每条剪枝追加一条事件： + ("deleted", entry, None) / ("delete_failed", entry, + error) / ("pruned_no_file", entry, None)（登记条目 + 被剪、但无对应文件可删）。调用方（executor.snapshot_vm） + 据此发 vm_snapshot_pruned / vm_snapshot_prune_failed + 审计；events 为 None 时行为与旧版一致（仍删、仍 + best-effort），不改旧调用点返回值。 + """ import time now = time.time() if now is None else now kept, pruned = [], [] @@ -181,8 +192,15 @@ def prune_snapshots(state_dir: str, vm_name: str, try: if os.path.exists(entry["path"]): os.remove(entry["path"]) - except OSError: - pass + if events is not None: + events.append(("deleted", entry, None)) + elif events is not None: + events.append(("pruned_no_file", entry, None)) + except OSError as exc: + if events is not None: + events.append(("delete_failed", entry, exc)) + elif events is not None: + events.append(("pruned_no_file", entry, None)) else: kept.append(entry) if pruned: @@ -192,13 +210,16 @@ def prune_snapshots(state_dir: str, vm_name: str, def check_snapshot_limits(state_dir: str, vm_name: str, max_count: int, max_total_gb: float, - retention_days: float | None) -> list: + retention_days: float | None, + events: list | None = None) -> list: """快照上限预检（snapshot_vm 调 ch-remote 之前调）： 先按保留期剪枝，再查数量（已有数 + 本次 > 上限即 snapshot_quota_exceeded）与总量（已有总量已达上限即 拒绝；本次大小事前未知，总量在 record 时二次收紧， - 见 record_snapshot）。返回剪枝后条目。""" - kept, _ = prune_snapshots(state_dir, vm_name, retention_days) + 见 record_snapshot）。返回剪枝后条目。events 透传给 + prune_snapshots（见 P1-08），供调用方审计剪枝动作。""" + kept, _ = prune_snapshots(state_dir, vm_name, retention_days, + events=events) if len(kept) + 1 > max_count: raise ChvFailure( "snapshot_quota_exceeded", @@ -221,11 +242,24 @@ 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, + events: list | None = None) -> tuple[list, list]: """快照成功后登记；file:// 目标记实际大小。登记后若 总量超上限，按最旧优先删登记外的旧快照文件（总量硬 帽，删动作由调用方审计——本函数返回被删的旧条目）。 - 返回（全部条目，被总量帽删掉的旧条目）。""" + 返回（全部条目，被总量帽删掉的旧条目）。 + + feat-chv-p1fix（P1-06）：单条快照本身大于总量帽时， + 在任何落盘/删旧动作之前直接拒绝 + （snapshot_quota_exceeded），不悄悄保留超大快照—— + 旧循环 `while len > 1` 在单条超帽时直接停、超大快照 + 被保留。大小未知（size 为 None，如远端目标或文件 + 缺失）时无法判定，不拒绝，沿用旧口径。 + feat-chv-p1fix（P1-08）：总量帽删文件动作记 events + （("deleted"/"delete_failed"/"pruned_no_file", entry, + error)），供调用方审计；删失败时登记条目仍移除 + （旧语义不变），但失败本身不再静默。 + """ import time path = None size = None @@ -235,23 +269,43 @@ def record_snapshot(state_dir: str, vm_name: str, snapshot_url: str, max_total_g size = os.path.getsize(path) except OSError: size = None + if max_total_gb is not None and size is not None \ + and size > max_total_gb * 1024 ** 3: + raise ChvFailure( + "snapshot_quota_exceeded", + "单条快照 %d 字节已超过总量上限 %.1f GB（VM %s），" + "拒绝记录" % (size, max_total_gb, vm_name)) entries = _load_snapshots(state_dir, vm_name) entries.append({"url": snapshot_url, "path": path, "size": size, "created_at": time.time()}) _save_snapshots(state_dir, vm_name, entries) - removed = [] - if max_total_gb is not None: - cap = max_total_gb * 1024 ** 3 - total = sum(e.get("size") or 0 for e in entries) - while len(entries) > 1 and total > cap: - oldest = entries.pop(0) - if oldest.get("path") and os.path.exists(oldest["path"]): - try: - os.remove(oldest["path"]) - except OSError: - pass - total -= oldest.get("size") or 0 - removed.append(oldest) - if removed: - _save_snapshots(state_dir, vm_name, entries) + removed = _tighten_snapshot_cap(entries, max_total_gb, events) + if removed: + _save_snapshots(state_dir, vm_name, entries) return entries, removed + + +def _tighten_snapshot_cap(entries, max_total_gb, events): + """总量帽收紧（record_snapshot 提取，守 50 行门禁）： + 总量超帽时按最旧优先删登记外的旧快照文件，返回被删 + 的旧条目；删动作记 events（见 record_snapshot）。""" + removed = [] + if max_total_gb is None: + return removed + cap = max_total_gb * 1024 ** 3 + total = sum(e.get("size") or 0 for e in entries) + while len(entries) > 1 and total > cap: + oldest = entries.pop(0) + if oldest.get("path") and os.path.exists(oldest["path"]): + try: + os.remove(oldest["path"]) + if events is not None: + events.append(("deleted", oldest, None)) + except OSError as exc: + if events is not None: + events.append(("delete_failed", oldest, exc)) + elif events is not None: + events.append(("pruned_no_file", oldest, None)) + total -= oldest.get("size") or 0 + removed.append(oldest) + return removed diff --git a/cloud/chv/executor.py b/cloud/chv/executor.py index 3bdddc4..f2fede5 100644 --- a/cloud/chv/executor.py +++ b/cloud/chv/executor.py @@ -55,6 +55,7 @@ from cloud.chv.hardening import ( validate_net_entries, validate_snapshot_url, ) +from cloud.chv.locking import StateLock from cloud.chv.ssh_transport import build_ssh_command as _build_ssh_cmd from cloud.chv.ssh_transport import known_hosts_path as _known_hosts_path from cloud.chv.ssh_transport import ( @@ -299,6 +300,7 @@ class CloudHypervisorExecutor(Executor): sleeper: "Callable[..., object]" audit_callback: object audit_failure_count: int + _state_lock: StateLock _vms: dict _guest_env_delivered: dict @@ -365,6 +367,11 @@ class CloudHypervisorExecutor(Executor): # 名 source 客内文件，未投递的密钥名一律拒绝（防止 # “wake 没投递、执行时却按名 source 一个旧文件”）。 self._guest_env_delivered = {} + # feat-chv-p1fix（P1-05）：容量/net/磁盘 check→act + # 与 purge/cleanup 的跨进程锁（flock，见 locking.py）。 + # 只在 wake/restore/cleanup 时加锁，不在构造时建锁 + # 文件（构造本身不应有 state_dir 写副作用）。 + self._state_lock = StateLock(self.state_dir) self.api_client = api_client if api_client is not None \ else _unix_socket_api self.sleeper = sleeper if sleeper is not None else time.sleep @@ -465,16 +472,23 @@ class CloudHypervisorExecutor(Executor): # P2-6：配置校验时机统一到 preflight。CPU 非整数、内存 # 字符串非法过去只在 build_vm_config 才炸，现在 preflight # 一次报出（kind=config_invalid）。 + # feat-chv-p1fix（P1-07）：构造时校验拦不住“属性事后 + # 改写/注入”的非法值，preflight 这里再验 >0——CPU + # 0/负、内存 0、任务超时 0/负一律干净 preflight 失败， + # 不进 VmConfig、不等 VMM 侧才炸。 try: - int(self.cpus) + if int(self.cpus) <= 0: + raise ValueError("CHV_CPU") except (TypeError, ValueError): missing.append(("config_invalid", - "CHV_CPU 必须是整数：%r" % (self.cpus,))) + "CHV_CPU 必须是大于 0 的整数：%r" + % (self.cpus,))) try: - _parse_memory_bytes(self.memory) + if _parse_memory_bytes(self.memory) <= 0: + raise ValueError("CHV_MEM") except ValueError: missing.append(("config_invalid", - "CHV_MEM 格式非法：%r" % (self.memory,))) + "CHV_MEM 格式非法或为 0：%r" % (self.memory,))) self._preflight_numeric_checks(missing) def _preflight_numeric_checks(self, missing: list) -> None: """preflight 数值面：端口/配额/超时/快照统一校验，纯移动。 @@ -513,6 +527,7 @@ class CloudHypervisorExecutor(Executor): except (TypeError, ValueError): missing.append(("config_invalid", "CHV_*_TIMEOUT/INTERVAL 必须是正数")) + self._preflight_task_timeout_check(missing) try: if int(getattr(self, "warm_pool_size", 0)) < 0: raise ValueError("warm_pool_size") @@ -520,6 +535,19 @@ class CloudHypervisorExecutor(Executor): missing.append(("config_invalid", "CHV_WARM_POOL_SIZE 必须是非负整数")) self._preflight_snapshot_checks(missing) + def _preflight_task_timeout_check(self, missing: list) -> None: + """preflight 任务超时面（feat-chv-p1fix，P1-07 提取： + 保持 _preflight_numeric_checks 在 50 行硬门禁内）。 + None = 不限（旧语义）；显式 0/负数即 preflight 失败。""" + try: + _task_timeout = getattr(self, "task_timeout", None) + if _task_timeout is not None \ + and float(_task_timeout) <= 0: + raise ValueError("task_timeout") + except (TypeError, ValueError): + missing.append(("config_invalid", + "CHV_TASK_TIMEOUT 必须是大于 0 的数字")) + def _preflight_snapshot_checks(self, missing: list) -> None: """preflight 快照/host-key 面：统一校验，纯移动。""" try: @@ -815,13 +843,15 @@ class CloudHypervisorExecutor(Executor): public_text: str) -> None: """写预置 known_hosts（to10 提取）。主机部分跟 SSH 目标走（客机 IP 时不再写死回环，否则钉死校验永远 - 对不上）；解析失败时回环仅作钉死写入，不放宽校验。""" + 对不上）。feat-chv-p1fix（P1-04 同款）：解析失败即 + 抛 ChvFailure，绝不回落回环写钉死——回环钉死会让 + SSH 校验钉在一个“解析失败后猜出来的目标”上； + wake 路径在 provision 之前已在 _wake_prechecks 里 + 严格解析过，此处失败只发生在直接调用时，抛错是 + 正确语义。""" from cloud.chv.hostkeys import known_hosts_line from cloud.chv.ssh_transport import write_known_hosts - try: - ssh_host = self.resolve_ssh_host() - except ChvFailure: - ssh_host = "127.0.0.1" + ssh_host = self.resolve_ssh_host() write_known_hosts( paths["known_hosts"], [known_hosts_line(self.ssh_port, public_text, @@ -921,12 +951,14 @@ class CloudHypervisorExecutor(Executor): 即 CHV + 网关密钥现在要求 provisioned 模式（已是 默认），keyscan 模式下的网关任务 fail-closed，不再 把密钥值拼进 argv 将就跑。""" - try: - ssh_host = self.resolve_ssh_host() - except ChvFailure: - # 纯命令构造（旧测试/干跑）不炸：回环缺省仅作 - # 构造兼容，wake/wait 路径仍严格解析。 - ssh_host = "127.0.0.1" + # feat-chv-p1fix（P1-04）：删除旧回环回落——构造失败 + # 即抛 ssh_target_undeclared，调用方无法再把“显式 + # 回环”与“解析失败后回落”混为一谈。显式回环仍可用： + # 设 CHV_SSH_HOST=127.0.0.1，或显式宣称 + # CHV_SSH_FORWARD=1；wake/wait 路径本来就严格解析， + # 本改动只收紧公开构造方法，不放宽任何严格路径， + # 也没有新增 dry-run 例外参数（干跑同样不许猜目标）。 + ssh_host = self.resolve_ssh_host() secrets, plain = split_secret_env(extra_env) guest_env_file = None if secrets: @@ -1065,6 +1097,13 @@ class CloudHypervisorExecutor(Executor): names.discard(exclude) return len(names) + def _state_lock_ctx(self) -> StateLock: + """当前 state_dir 的 StateLock（state_dir 若被事后改写， + 换用新路径的锁实例，避免锁在旧目录上）。""" + if self._state_lock.state_dir != self.state_dir: + self._state_lock = StateLock(self.state_dir) + return self._state_lock + def _check_capacity(self, exclude=None): """宿主容量门：持久化计数（见 _persistent_vm_count） 达到 max_vms 即明确拒绝（kind=host_capacity_exceeded， @@ -1442,7 +1481,18 @@ class CloudHypervisorExecutor(Executor): 密钥子集在 provisioned 模式下写进 seed 客内文件， keyscan 模式下有密钥子集即 fail-closed；起 VM 前的 preflight/容量/配额失败也发 vm_failed 审计（L3）； - net 隔离在起 VMM 前校验（H7），成功后登记 net.json。""" + net 隔离在起 VMM 前校验（H7），成功后登记 net.json。 + feat-chv-p1fix（P1-05）：全程持 state_dir flock—— + 容量/net/磁盘的 check→act 只有在“查完立即占位、 + 占位前不放行”时才成立；锁提前释放会在 socket 尚 + 未绑定前留出计数窗口，仍可超限。串行化代价只落在 + wake/restore/cleanup 之间，已运行 VM 的任务执行 + （_run_guest）不持此锁。""" + with self._state_lock_ctx(): + return self._wake_vm_locked(task, extra_env) + + def _wake_vm_locked(self, task: dict, extra_env: dict | None = None) -> dict: + """持 state 锁的 wake 本体（见 wake_vm）。""" if not isinstance(task, dict) or task.get("id") is None: raise ValueError("任务缺少 id") name = self.vm_name(task) @@ -1499,19 +1549,48 @@ class CloudHypervisorExecutor(Executor): feat-chv-secfix（L2/M3）：ch-remote 之前先过 URL 白名单（file:// 须在 state_dir/CHV_SNAPSHOT_DIR 内， 远端须显式放行）与数量/总量/保留期上限；成功后登 - 记并在总量超帽时按最旧优先剪枝（删动作审计）。""" + 记并在总量超帽时按最旧优先剪枝（删动作审计）。 + feat-chv-p1fix（P1-08）：保留期剪枝与总量帽剪枝的 + 删文件成功/失败均发审计；失败路径统一发 + vm_snapshot_failed 审计再抛（此前失败路径无任何 + 快照失败审计，只有 wake 前置失败才有 vm_failed）。 + 本体拆到 _snapshot_create 以守 50 行硬门禁。""" if not snapshot_url: raise ValueError("snapshot_url 不能为空") + task = task_or_name if isinstance(task_or_name, dict) \ + else {"id": task_or_name} + vm_name = self.vm_name(task_or_name) + try: + self._snapshot_create(task_or_name, task, vm_name, + snapshot_url) + except Exception as exc: + kind = getattr(exc, "kind", None) or type(exc).__name__ + self._emit("vm_snapshot_failed", task, + "vm=%s snapshot=%s kind=%s error=%s" + % (vm_name, snapshot_url, kind, exc), kind) + raise + self._emit("vm_snapshot", task, + "vm=%s snapshot=%s" + % (self.vm_name(task_or_name), snapshot_url)) + return snapshot_url + + def _snapshot_create(self, task_or_name, task, vm_name, + snapshot_url) -> None: + """快照创建段：白名单→上限/保留期剪枝→ch-remote→ + 落盘校验→登记/总量剪枝（见 snapshot_vm）。""" validate_snapshot_url( snapshot_url, self.state_dir, snapshot_dir=getattr(self, "snapshot_dir", None), - allow_remote=getattr(self, "allow_remote_snapshot", False)) - vm_name = self.vm_name(task_or_name) + allow_remote=getattr( + self, "allow_remote_snapshot", False)) + prune_events: list = [] check_snapshot_limits( self.state_dir, vm_name, getattr(self, "snapshot_max_count", 10), getattr(self, "snapshot_max_total_gb", 20.0), - getattr(self, "snapshot_retention_days", 30.0)) + getattr(self, "snapshot_retention_days", 30.0), + events=prune_events) + self._emit_prune_events(task, vm_name, prune_events, "保留期") self._ch_remote(task_or_name, "snapshot", "source_url=%s" % snapshot_url) # feat-chv-fix（P1-7）：只查 returncode 不够。file:// @@ -1526,19 +1605,39 @@ class CloudHypervisorExecutor(Executor): raise ChvFailure( "snapshot_not_persisted", "快照未落盘或为空：%s" % snapshot_url) - task = task_or_name if isinstance(task_or_name, dict) \ - else {"id": task_or_name} - _entries, _removed = record_snapshot( + record_events: list = [] + record_snapshot( self.state_dir, vm_name, snapshot_url, - getattr(self, "snapshot_max_total_gb", 20.0)) - 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)) - return snapshot_url + getattr(self, "snapshot_max_total_gb", 20.0), + events=record_events) + self._emit_prune_events(task, vm_name, record_events, + "总量超帽") + + def _emit_prune_events(self, task, vm_name, events, reason) -> None: + """剪枝事件转审计（feat-chv-p1fix，P1-08）：成功删 + 文件与登记剪枝发 vm_snapshot_pruned，删失败发 + vm_snapshot_prune_failed；无文件可删的登记剪枝只 + 发 pruned（删动作本身不存在，无失败可言）。""" + deleted = 0 + failed = 0 + for status, entry, error in events: + url = entry.get("url") if isinstance(entry, dict) else None + if status == "delete_failed": + failed += 1 + self._emit("vm_snapshot_prune_failed", task, + "vm=%s pruned snapshot=%s reason=%s " + "error=%s" % (vm_name, url, reason, error)) + else: + deleted += 1 + self._emit("vm_snapshot_pruned", task, + "vm=%s pruned snapshot=%s reason=%s " + "deleted=%s" % ( + vm_name, url, reason, + status == "deleted")) + if failed: + _LOGGER.warning("chv_snapshot_prune_failed vm=%s " + "reason=%s failed=%d", vm_name, reason, + failed) def _validate_restore(self, task, snapshot_url): """restore 前置校验：返回归一化后的 snapshot_url。 @@ -1588,7 +1687,14 @@ class CloudHypervisorExecutor(Executor): resume → 等 SSH 就绪。恢复后默认 paused，不 resume 就不算醒。任一步失败都关 VM 再抛；成功返回 {vm, api_socket, state: "running", ready: True, - mode: "restored"}。""" + mode: "restored"}。feat-chv-p1fix（P1-05）：与 wake + 同款，全程持 state 锁——restore 同样查容量、建 VM， + 不持锁会与并发 wake 互相超限。""" + with self._state_lock_ctx(): + return self._restore_vm_locked(task, snapshot_url) + + def _restore_vm_locked(self, task: dict, snapshot_url: str) -> dict: + """持 state 锁的 restore 本体（见 restore_vm）。""" snapshot_url = self._validate_restore(task, snapshot_url) name = self.vm_name(task) self._check_capacity(exclude=name) @@ -1781,15 +1887,22 @@ class CloudHypervisorExecutor(Executor): def _clean_one_vm_secrets(self, entry, vm_dir): """单 VM 残留清理：私钥一律删（live VM 私钥在 seed 盘 建好后即删，残留只可能是崩溃遗留）；非 live VM 的 - seed.iso/seed 目录同理删。返回 (删私钥, 删 seed)。""" + seed.iso/seed 目录同理删。返回 (删私钥, 删 seed, + 删失败)。 + + feat-chv-p1fix（P1-03）：删失败不再静默 pass—— + 第三元组就是失败标志，调用方据此发审计、写脏标记， + 删失败的 VM 绝不进 cleaned（调用方可区分成功与失败）。 + """ removed_priv = False + failed = False priv = os.path.join(vm_dir, "ssh_host_ed25519_key") if os.path.exists(priv): try: os.remove(priv) removed_priv = True except Exception: - pass + failed = True removed_seed = False if not self._is_live_vm(entry): for _suffix in ("seed.iso", "seed"): @@ -1802,13 +1915,18 @@ class CloudHypervisorExecutor(Executor): os.remove(_p) removed_seed = True except Exception: - pass - return removed_priv, removed_seed + failed = True + return removed_priv, removed_seed, failed def cleanup_stale_secrets(self) -> list: """崩溃残留扫描（H5）：私钥一律删、非 live VM 的 seed 同理删并审计；返回已清理 VM 名。worker 启动与 - cleanup_stale_vms 时调它，不依赖 socket 是否存在。""" + cleanup_stale_vms 时调它，不依赖 socket 是否存在。 + feat-chv-p1fix（P1-03）：删失败的 VM 不进 cleaned， + 另发 host_secret_residue_delete_failed 审计并写脏 + 标记（下次 provision_host_key 见标记即拒起，见 + vm_dirty_refused），直至人工清理——最敏感物残留 + 不再“重试前不可观测”。""" cleaned: list = [] try: entries = os.listdir(self.state_dir) @@ -1820,9 +1938,18 @@ class CloudHypervisorExecutor(Executor): vm_dir = os.path.join(self.state_dir, entry) if not os.path.isdir(vm_dir): continue - removed_priv, removed_seed = \ + removed_priv, removed_seed, failed = \ self._clean_one_vm_secrets(entry, vm_dir) - if removed_priv or removed_seed: + if failed: + self._emit("host_secret_residue_delete_failed", + {"id": entry}, + "vm=%s private_key_removed=%s " + "seed_removed=%s" % ( + entry, removed_priv, removed_seed)) + self._write_dirty_marker(entry, True) + # 删失败时即使部分删掉了，也不进 cleaned：残留仍在， + # “已清理”名单里绝不出现未清干净的 VM。 + if not failed and (removed_priv or removed_seed): cleaned.append(entry) self._emit("host_secret_residue_cleaned", {"id": entry}, @@ -1865,7 +1992,9 @@ class CloudHypervisorExecutor(Executor): def _stop_dead_socket_vms(self, entries): """socket 扫描：登记进程仍活的跳过，其余走 stop_vm。 - 返回已清理 VM 名；非 chv- 前缀 socket 一律跳过。""" + 返回已清理 VM 名；非 chv- 前缀 socket 一律跳过。 + feat-chv-p1fix（P1-05）：登记进程的读持 _vms_lock， + 不再无锁读 dict（与 purge/注册并发会读到中间态）。""" cleaned = [] for entry in entries: if not entry.endswith(".sock"): @@ -1873,7 +2002,8 @@ class CloudHypervisorExecutor(Executor): name = entry[:-len(".sock")] if not name.startswith("chv-"): continue - proc = self._vms.get(name) + with self._vms_lock: + proc = self._vms.get(name) if proc is not None: try: alive = proc.poll() is None @@ -1887,20 +2017,36 @@ class CloudHypervisorExecutor(Executor): def _purge_dead_registry(self): """登记进程已死但 socket 已被删的登记项顺手清掉 - （没有 socket 可删，不计入清理数）。""" - for name in list(self._vms.keys()): - proc = self._vms.get(name) + （没有 socket 可删，不计入清理数）。 + feat-chv-p1fix（P1-05）：先在锁内快照键列表，再 + 逐项在锁内读/删——旧“无锁 pop 与持锁遍历并发” + 会致 dictionary changed size during iteration， + 且 pop 本身无锁。""" + with self._vms_lock: + names = list(self._vms.keys()) + for name in names: + with self._vms_lock: + proc = self._vms.get(name) try: dead = proc is not None and proc.poll() is not None except Exception: dead = True if dead: - self._vms.pop(name, None) + with self._vms_lock: + self._vms.pop(name, None) def cleanup_stale_vms(self) -> list: """清理残留 VM：先跑残留盘 GC 与残留密钥扫描，再走 socket 扫描与登记清理。只清 socket 残留与已死进程 - 对应的 VM；state_dir 不存在返回 []，绝不抛异常。""" + 对应的 VM；state_dir 不存在返回 []，绝不抛异常。 + feat-chv-p1fix（P1-05）：全程持 state 锁，与 wake/ + restore 互斥——清理删盘/删 socket 时不许另一路 + wake 同时为同名 VM 供盘、建 VM。""" + with self._state_lock_ctx(): + return self._cleanup_stale_vms_locked() + + def _cleanup_stale_vms_locked(self) -> list: + """持 state 锁的 cleanup 本体（见 cleanup_stale_vms）。""" try: self.cleanup_stale_secrets() except Exception: diff --git a/cloud/chv/experimental_pool.py b/cloud/chv/experimental_pool.py index ffb5df3..0da8927 100644 --- a/cloud/chv/experimental_pool.py +++ b/cloud/chv/experimental_pool.py @@ -159,8 +159,14 @@ class WarmPool: 对每个条目尽力 shutdown_vm，条目清掉并落盘，返回 已清理的 VM 名列表。生产启动/关闭不走本方法，走 - cloud.chv.orphans.reconcile_orphans_for_executor()。""" + cloud.chv.orphans.reconcile_orphans_for_executor()。 + feat-chv-p1fix（P1-01 同款语义）：shutdown 失败的 + 条目不进 cleaned、不从内存条目里 pop（下次 reconcile + 可重试），并记 failed 审计；文件落盘仍只写 + warm/assigned 态（见 P2-16，实验路径的文件语义 + 本轨不改，内存态保留即重试依据）。""" cleaned = [] + failed = [] for name, entry in list(self._entries.items()): if entry["state"] not in ("orphan", "assigned"): continue @@ -168,7 +174,8 @@ class WarmPool: try: self.executor.shutdown_vm(vm) except Exception: - pass + failed.append(vm) + continue self._entries.pop(name, None) cleaned.append(vm) if cleaned: @@ -179,6 +186,14 @@ class WarmPool: "cleaned=%s" % ",".join(cleaned)) except Exception: pass + if failed: + try: + self.executor._emit( + "warm_pool_reconcile_failed", {"id": None}, + "failed=%s cleaned=%s" % ( + ",".join(failed), ",".join(cleaned))) + except Exception: + pass return cleaned # -- 池操作 ---------------------------------------------------------- diff --git a/cloud/chv/locking.py b/cloud/chv/locking.py new file mode 100644 index 0000000..58055ef --- /dev/null +++ b/cloud/chv/locking.py @@ -0,0 +1,107 @@ +"""CHV state_dir 跨进程锁（feat-chv-p1fix，P1-05）。 + +问题：容量 / net / 磁盘的 check→act 原来只靠进程内 +threading.Lock 护 `_vms` dict；多 worker 进程共用同一 +state_dir 时，两路 wake 可双双通过容量/磁盘/net 门并 +双双建 VM（check-then-act 竞态），`_purge_dead_registry` +的无锁 pop 与持锁遍历并发还可致字典迭代异常。 + +解法：同一把锁护住 check→act 全程——文件锁（flock） ++ 进程内可重入锁的组合： + +- 文件锁：state_dir/chv-state.lock 上的 `LOCK_EX`。 + 不同进程（多 worker）靠它互斥；同一进程内不同 + executor 实例（各有自己的 StateLock 对象）同样靠它 + 互斥，因为 flock 按打开的文件描述符区分，两个实例 + 各自 open → 第二个会阻塞，直到第一个释放。 +- 进程内可重入：同一 executor 实例内，wake 全程会 + 嵌套调用同样需要锁的方法（如 wake → 失败时 stop → + cleanup_stale_vms 等路径），因此 StateLock 自带 + threading.RLock + 深度计数：同一线程重入时不再重复 + flock，只在最外层加锁/释放，避免同一线程把自己锁死。 + 注意：同一线程经“不同 StateLock 实例”嵌套加锁不在 + 保证内——代码里不存在这条路径（wake/restore/cleanup/ + reconcile 各自只用本实例或本 registry 的锁，且 + reconcile 不在 wake 持锁期间被调用）。 + +持有范围（见 executor.py）：wake_vm 全程（prechecks → +prepare → seed → boot 含 SSH 就绪等待）、restore_vm 全程、 +cleanup_stale_vms 全程（含 gc/密钥扫描/socket 扫描/purge）、 +orphans.reconcile 全程。锁内串行化 wake 是有意的代价： +容量门只有“查完立即占位、占位前不放行”才成立；把锁 +提前释放到 vm.create 之后会在 socket 尚未绑定前留出 +计数窗口，仍可超限。串行 wake 不影响已运行 VM 的任务 +执行（_run_guest 不持此锁）。 + +本模块只依赖标准库，是叶子模块；fcntl 只在 POSIX 上 +可用——本执行器只面向 Linux 宿主（KVM 路径本身即 +Linux 专属），非 POSIX 平台 import 即失败，不静默回落 +纯线程锁（回落会把多 worker 保证悄悄降格）。 +""" + +import fcntl +import os +import threading + +LOCK_FILENAME = "chv-state.lock" + + +def lock_path_for_state_dir(state_dir: str) -> str: + """state_dir 对应的锁文件路径。""" + return os.path.join(state_dir, LOCK_FILENAME) + + +class StateLock: + """state_dir 跨进程 + 进程内可重入锁（上下文管理器）。 + + 用法：`with executor_state_lock:`。acquire 时先拿进程内 + RLock，再在深度为 0 时 open 锁文件并 flock LOCK_EX + （阻塞，直到其它进程/实例释放）；release 时深度归零才 + 解锁并关闭文件描述符。锁文件本身永不删除（删文件会 + 破坏 flock 的互斥语义：两个进程可能各持一个已删文件 + 的锁而互不感知）。 + """ + + def __init__(self, state_dir: str) -> None: + self.state_dir = state_dir + self.lock_path = lock_path_for_state_dir(state_dir) + self._rlock = threading.RLock() + self._depth = 0 + self._fd: int | None = None + + def acquire(self) -> None: + self._rlock.acquire() + try: + if self._depth == 0: + os.makedirs(self.state_dir, exist_ok=True) + fd = os.open(self.lock_path, + os.O_RDWR | os.O_CREAT, 0o600) + try: + fcntl.flock(fd, fcntl.LOCK_EX) + except Exception: + os.close(fd) + raise + self._fd = fd + self._depth += 1 + except Exception: + self._rlock.release() + raise + + def release(self) -> None: + try: + self._depth -= 1 + if self._depth == 0 and self._fd is not None: + try: + fcntl.flock(self._fd, fcntl.LOCK_UN) + finally: + os.close(self._fd) + self._fd = None + finally: + self._rlock.release() + + def __enter__(self) -> "StateLock": + self.acquire() + return self + + def __exit__(self, *exc_info: object) -> None: + self.release() diff --git a/cloud/chv/orphans.py b/cloud/chv/orphans.py index 7c7c22a..59ab09b 100644 --- a/cloud/chv/orphans.py +++ b/cloud/chv/orphans.py @@ -17,12 +17,19 @@ Registry 语义与旧 WarmPool._load_registry 完全一致 VM，绝不发给任务，只能清理。 - 损坏 fail-safe：文件不是合法 JSON / 顶层不是 dict 时， 整个文件 quarantine 成 .corrupt，清理从空开始； + 损坏本身发审计（feat-chv-p1fix，P1-02），绝不静默 + 返回空——损坏文件的孤儿 VM 此前只能靠 socket 扫描。 单条目结构非法（无 vm、或 state 不在 warm/assigned） - 时只丢该条，不连坐其它条目。 -- 清理后落盘为空 dict（原子 tmp + rename）；写失败只记 - 日志警告，不炸 worker 启动/关闭。shutdown_vm 抛错时该 - 条目仍计入已清理（旧 reconcile 语义：尽力清理、不因 - 单条失败卡住整批），_emit 失败同样 best-effort。 + 时只丢该条，不连坐其它条目；非法条目发审计并计数 + （feat-chv-p1fix，P1-02），落盘时不再写回。 +- 清理后落盘：全部成功时为空 dict；有失败项时只写回 + 失败项的原始条目（原子 tmp + rename），下次 reconcile + 可重试，失败项绝不被“已清理”掩盖（feat-chv-p1fix， + P1-01）。写失败只记日志警告，不炸 worker 启动/关闭。 + shutdown_vm 抛错的条目不进 cleaned、不触发“全清”， + 而是记 failed 列表并经 executor._emit 发审计； + _emit 失败同样 best-effort。整个 reconcile 在 + state_dir 的 flock 下跑（见 locking.py），与 wake 互斥。 """ import json @@ -91,52 +98,130 @@ def _orphan_entries(data: dict) -> list[dict]: """从原始 registry dict 里筛出可清理的上一世条目。 只留 state 为 warm/assigned 且有 vm 的条目；非法条目 - 静默丢弃（不连坐）。返回条目为 {"vm", "task_id", + 不连坐其它条目（计数与审计见 reconcile 本体，不再 + 静默丢弃）。返回条目为 {"vm", "task_id", "snapshot_url"} 的新 dict，不改原数据。""" entries: list[dict] = [] - for entry in data.values(): + for _key, entry in _orphan_entries_with_keys(data)[0]: + entries.append(entry) + return entries + + +def _orphan_entries_with_keys(data: dict) -> tuple[list, list]: + """筛条目并保留原始 registry 键：返回（可清理条目， + 非法条目键）。 + + 可清理条目为 (原始键, {"vm", "task_id", + "snapshot_url"})；非法条目只记键（键本身不含密钥， + 可进审计 detail），原始值不进审计。""" + entries: list = [] + invalid_keys: list = [] + for key, entry in data.items(): if not isinstance(entry, dict): + invalid_keys.append(key) continue if not entry.get("vm"): + invalid_keys.append(key) continue if entry.get("state") not in ("warm", "assigned"): + invalid_keys.append(key) continue - entries.append({ + entries.append((key, { "vm": entry["vm"], "task_id": entry.get("task_id"), "snapshot_url": entry.get("snapshot_url"), - }) - return entries + })) + return entries, invalid_keys + + +def _best_effort_emit(executor: object, event: str, detail: str) -> None: + """经 executor._emit 发审计（best-effort，兼容只接受 + (event, task, detail) 三参的假执行器；无 _emit 时跳过）。""" + emit = getattr(executor, "_emit", None) + if not callable(emit): + return + try: + emit(event, {"id": None}, detail) + except Exception: + pass def reconcile_orphans_for_executor(executor: object) -> list[str]: """生产路径唯一的残留清理入口：返回已清理的 VM 名列表。 只做清理：读 registry → 每个上一世条目尽力 - shutdown_vm → 落盘为空 → best-effort 审计事件。 + shutdown_vm → 落盘 → best-effort 审计事件。 绝不构造 WarmPool、绝不调 prewarm/acquire/release， - 绝不把任何条目发给任务。无 registry 路径时返回 []。""" + 绝不把任何条目发给任务。无 registry 路径时返回 []。 + + feat-chv-p1fix（P1-01/P1-02/P1-05）语义： + - shutdown 失败的条目不进 cleaned，记 failed 并审计， + 落盘时只写回失败项原始条目（可重试），绝不清库； + - registry 损坏时 quarantine 后仍发审计，不静默返回空； + - 非法条目发审计并计数，落盘时丢弃（已审计，不再 + 每次重启重复审计同一批非法条目）； + - 全程持 state_dir flock，与 wake/restore/cleanup 互斥。 + 返回值仍只含 cleaned（调用方兼容不变）；失败与 + 损坏/非法的可见性走 executor._emit（worker 接线后 + 即任务 timeline 审计），不靠返回值传递。 + """ path = registry_path_for_executor(executor) if not path: return [] - data, _corrupt = _read_registry(path) - if data is None: + if not os.path.exists(path): return [] - entries = _orphan_entries(data) + state_dir = getattr(executor, "state_dir", None) + if state_dir: + from cloud.chv.locking import StateLock + lock = StateLock(state_dir) + else: # pragma: no cover - state_dir 缺失时 path 已为 None + from contextlib import nullcontext + lock = nullcontext() # type: ignore[assignment] + with lock: + return _reconcile_locked(executor, path) + + +def _reconcile_locked(executor: object, path: str) -> list[str]: + """持锁的 reconcile 本体（见 reconcile_orphans_for_executor）。""" + data, corrupt = _read_registry(path) + if data is None: + if corrupt: + _best_effort_emit( + executor, "warm_pool_registry_corrupt", + "path=%s registry corrupt, quarantined" % path) + return [] + entries, invalid_keys = _orphan_entries_with_keys(data) + if invalid_keys: + _best_effort_emit( + executor, "warm_pool_registry_invalid_entries", + "invalid=%d keys=%s" + % (len(invalid_keys), ",".join(str(k) for k in invalid_keys))) cleaned: list[str] = [] - for entry in entries: + failed: list[str] = [] + failed_errors: list[str] = [] + failed_keys: list[str] = [] + for key, entry in entries: vm = entry["vm"] try: executor.shutdown_vm(vm) # type: ignore[attr-defined] - except Exception: - pass + except Exception as exc: + failed.append(vm) + failed_keys.append(key) + failed_errors.append("%s:%s" % (vm, type(exc).__name__)) + continue cleaned.append(vm) + if cleaned or failed or invalid_keys: + remaining = {key: data[key] for key in failed_keys + if key in data} + _write_registry(path, remaining) if cleaned: - _write_registry(path, {}) - try: - executor._emit( # type: ignore[attr-defined] - "warm_pool_reconcile", {"id": None}, - "cleaned=%s" % ",".join(cleaned)) - except Exception: - pass + _best_effort_emit( + executor, "warm_pool_reconcile", + "cleaned=%s" % ",".join(cleaned)) + if failed: + _best_effort_emit( + executor, "warm_pool_reconcile_failed", + "failed=%s cleaned=%s errors=%s" + % (",".join(failed), ",".join(cleaned), + ";".join(failed_errors))) return cleaned diff --git a/cloud/tests/test_chv_arch_maint_fe.py b/cloud/tests/test_chv_arch_maint_fe.py index 56a8a78..21b07a9 100644 --- a/cloud/tests/test_chv_arch_maint_fe.py +++ b/cloud/tests/test_chv_arch_maint_fe.py @@ -156,6 +156,8 @@ class BreakingCleanupTest(unittest.TestCase): class OrphansTest(unittest.TestCase): def test_valid_warm_and_assigned_cleaned_invalid_dropped(self): + # feat-chv-p1fix（P1-02）：非法条目不再静默丢弃， + # 先发 invalid 审计并计数，再发 cleaned 审计。 with tempfile.TemporaryDirectory() as d: path = _write_registry(d, { "a": {"state": "warm", "vm": "chv-a", @@ -171,19 +173,26 @@ class OrphansTest(unittest.TestCase): reconcile_orphans_for_executor(ex), ["chv-a", "chv-b"]) self.assertEqual(ex.shutdowns, ["chv-a", "chv-b"]) - self.assertEqual(ex.emits, ["warm_pool_reconcile"]) + self.assertEqual(ex.emits, [ + "warm_pool_registry_invalid_entries", + "warm_pool_reconcile"]) with open(path) as fh: self.assertEqual(json.load(fh), {}) - def test_invalid_only_returns_empty_and_leaves_file(self): + def test_invalid_only_returns_empty_and_drops_audited_entries(self): + # feat-chv-p1fix（P1-02）：仅非法条目时返回 [] 且 + # 不 shutdown；非法条目已审计计数，落盘时丢弃， + # 不再每次重启重复审计同一批非法条目。 with tempfile.TemporaryDirectory() as d: path = _write_registry(d, { "c": {"state": "warm", "task_id": "c"}}) ex = FakeProdExecutor(d) self.assertEqual(reconcile_orphans_for_executor(ex), []) self.assertEqual(ex.shutdowns, []) + self.assertEqual( + ex.emits, ["warm_pool_registry_invalid_entries"]) with open(path) as fh: - self.assertIn("task_id", fh.read()) + self.assertEqual(json.load(fh), {}) def test_corrupt_json_quarantined(self): with tempfile.TemporaryDirectory() as d: @@ -212,9 +221,12 @@ class OrphansTest(unittest.TestCase): self.assertEqual( reconcile_orphans_for_executor(NoDir()), []) - def test_shutdown_and_emit_failures_do_not_break_cleanup(self): + def test_shutdown_failure_not_cleaned_and_retained(self): + # feat-chv-p1fix（P1-01）：shutdown 失败的条目不进 + # cleaned、不触发清库，落盘保留原始条目可重试； + # _emit 失败仍 best-effort，不炸清理流程。 with tempfile.TemporaryDirectory() as d: - _write_registry(d, { + path = _write_registry(d, { "a": {"state": "warm", "vm": "chv-a"}}) ex = FakeProdExecutor(d) @@ -225,7 +237,11 @@ class OrphansTest(unittest.TestCase): ex._emit = lambda *a, **k: (_ for _ in ()).throw( RuntimeError("audit down")) self.assertEqual( - reconcile_orphans_for_executor(ex), ["chv-a"]) + reconcile_orphans_for_executor(ex), []) + with open(path) as fh: + self.assertEqual( + json.load(fh), + {"a": {"state": "warm", "vm": "chv-a"}}) def test_experimental_pool_registry_roundtrip_unchanged(self): # 实验池自身的 registry 语义（载入即 orphan、reconcile diff --git a/cloud/tests/test_chv_closeup.py b/cloud/tests/test_chv_closeup.py index edd2a85..9912b11 100644 --- a/cloud/tests/test_chv_closeup.py +++ b/cloud/tests/test_chv_closeup.py @@ -889,25 +889,31 @@ class DefensiveBranchTest(CloseupCase): ex.runner = keyscan_runner self.assertFalse(ex.capture_host_key(self._task())) - def test_build_ssh_command_fallback_and_secret_not_delivered(self): + def test_build_ssh_command_undeclared_raises_and_secret_not_delivered(self): + # feat-chv-p1fix（P1-04）：删除回环回落——目标未声明 + # 时构造即抛 ssh_target_undeclared，不再悄悄连回环。 ex = self.make_exec(env_extra={"CHV_SSH_FORWARD": "0"}) - cmd = ex.build_ssh_command(self._task(), ["true"]) - self.assertIn("127.0.0.1", " ".join(cmd)) + with self.assertRaises(ChvFailure) as ctx: + ex.build_ssh_command(self._task(), ["true"]) + self.assertEqual(ctx.exception.kind, "ssh_target_undeclared") ex = self.make_exec() with self.assertRaises(ChvFailure) as ctx: ex.build_ssh_command(self._task(), ["true"], extra_env={"SUB2API_API_KEY": "v"}) self.assertEqual(ctx.exception.kind, "secret_env_not_delivered") - def test_write_provisioned_known_hosts_fallback(self): + def test_write_provisioned_known_hosts_undeclared_raises(self): + # feat-chv-p1fix（P1-04 同款）：预置 known_hosts 解析 + # 失败即抛，不再回落回环写钉死。 ex = self.make_exec(env_extra={"CHV_SSH_FORWARD": "0", "CHV_HOST_KEY_MODE": "provisioned"}) vm_dir = os.path.join(self.state_dir, "chv-abc123") os.makedirs(vm_dir, exist_ok=True) paths = {"known_hosts": os.path.join(vm_dir, "known_hosts")} - ex._write_provisioned_known_hosts(paths, "ssh-ed25519 AAAA") - with open(paths["known_hosts"]) as fh: - self.assertIn("127.0.0.1", fh.read()) + with self.assertRaises(ChvFailure) as ctx: + ex._write_provisioned_known_hosts(paths, "ssh-ed25519 AAAA") + self.assertEqual(ctx.exception.kind, "ssh_target_undeclared") + self.assertFalse(os.path.exists(paths["known_hosts"])) def test_provision_host_key_dirty_refused(self): ex = self.make_exec() @@ -1083,22 +1089,26 @@ class DefensiveBranchTest(CloseupCase): self.assertFalse(ex._vm_is_running("chv-bad")) def test_clean_one_vm_secrets_remove_failures(self): + # feat-chv-p1fix（P1-03）：返回 3 元组，删失败以 + # failed=True 区分，不再静默 pass。 ex = self.make_exec() vm_dir = os.path.join(self.state_dir, "chv-x") os.makedirs(os.path.join(vm_dir, "ssh_host_ed25519_key"), exist_ok=True) - removed_priv, removed_seed = ex._clean_one_vm_secrets( + removed_priv, removed_seed, failed = ex._clean_one_vm_secrets( "chv-x", vm_dir) self.assertFalse(removed_priv) + self.assertTrue(failed) vm2 = os.path.join(self.state_dir, "chv-y") os.makedirs(os.path.join(vm2, "seed"), exist_ok=True) import shutil real_rmtree = shutil.rmtree with mock.patch("shutil.rmtree", side_effect=OSError("rmtree boom")): - removed_priv, removed_seed = ex._clean_one_vm_secrets( + removed_priv, removed_seed, failed = ex._clean_one_vm_secrets( "chv-y", vm2) self.assertFalse(removed_seed) + self.assertTrue(failed) self.assertTrue(real_rmtree is not None) def test_cleanup_stale_vms_subcall_failures_and_poll_branches(self): diff --git a/cloud/tests/test_chv_cover100.py b/cloud/tests/test_chv_cover100.py index db7595e..51f3156 100644 --- a/cloud/tests/test_chv_cover100.py +++ b/cloud/tests/test_chv_cover100.py @@ -559,14 +559,17 @@ class ExperimentalPoolCover100Test(unittest.TestCase): self.assertIn("warm-1", pool._entries) self.assertNotIn("old-1", pool._entries) - def test_reconcile_shutdown_failure_still_cleans(self): + def test_reconcile_shutdown_failure_not_cleaned(self): + # feat-chv-p1fix（P1-01 同款语义，实验池）：shutdown + # 失败的条目不进 cleaned、不从内存条目里 pop， + # 下次 reconcile 可重试。 stub = _PoolStub() stub.shutdown_raises = True pool = WarmPool(stub, registry_path=None) pool._entries["old-1"] = {"state": "orphan", "vm": "vm-old", "task": {"id": "old-1"}} - self.assertEqual(pool.reconcile_orphans(), ["vm-old"]) - self.assertEqual(pool._entries, {}) + self.assertEqual(pool.reconcile_orphans(), []) + self.assertIn("old-1", pool._entries) def test_reconcile_emit_failure_is_best_effort(self): stub = _PoolStub() diff --git a/cloud/tests/test_chv_p1fix.py b/cloud/tests/test_chv_p1fix.py new file mode 100644 index 0000000..f31ccad --- /dev/null +++ b/cloud/tests/test_chv_p1fix.py @@ -0,0 +1,730 @@ +"""CHV 生产级审核 P1 修复轨（feat-chv-p1fix，堆叠于 #270 之上）测试。 + +逐项对应审核报告 §5 的 P1-01～P1-10（P1-10 是文档项， +无代码测试；文档数字见 docs/chv-production-readiness.md +§15 与 docs/chv-architecture.md 头范围）： +- P1-01：生产 reconcile 失败项不进 cleaned、不清库、 + 落盘保留失败项原始条目、可重试；实验池同款语义。 +- P1-02：registry 损坏与非法条目发审计并计数，损坏不 + 静默返回空。 +- P1-03：残留私钥删失败发审计、写脏标记，调用方可区分 + 成功与失败（failed VM 不进 cleaned）。 +- P1-04：build_ssh_command 删除回环回落，构造失败即抛； + 显式回环（CHV_SSH_HOST / CHV_SSH_FORWARD=1）仍可用。 +- P1-05：state_dir flock（locking.StateLock）——线程内 + 互斥、跨进程互斥、wake 全程串行化、purge 快照遍历。 +- P1-06：单条快照超总量帽时拒绝记录，且不删旧快照、 + 不落盘新条目。 +- P1-07：CHV_CPU / CHV_MEM / CHV_TASK_TIMEOUT 的 >0 + 校验（config 构造层 + preflight 事后改写兜底）。 +- P1-08：快照保留期剪枝删失败发审计；snapshot_vm 失败 + 路径发 vm_snapshot_failed 审计。 +- P1-09：4 处三元 else 分支——str 形态 pause/resume/ + snapshot + 非 dict spill task。 + +全部用标准库与 mock runner/launcher/api_client，绝不启动 +真实 VM（本机无 /dev/kvm，真机验证仍是未验证项）。 +""" + +import json +import os +import subprocess +import sys +import tempfile +import threading +import time +import unittest +from pathlib import Path +from unittest import mock + +try: + from cloud.chv import ( + ChvFailure, + CloudHypervisorExecutor, + reconcile_orphans_for_executor, + ) + from cloud.chv.config import build_chv_config + from cloud.chv.disk import ( + prune_snapshots, + record_snapshot, + ) + from cloud.chv.locking import StateLock, lock_path_for_state_dir +except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.chv import ( + ChvFailure, + CloudHypervisorExecutor, + reconcile_orphans_for_executor, + ) + from cloud.chv.config import build_chv_config + from cloud.chv.disk import ( + prune_snapshots, + record_snapshot, + ) + from cloud.chv.locking import StateLock, lock_path_for_state_dir + + +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 P1FixCase(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, "disk.raw"), + 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.audit = [] + + 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_SSH_FORWARD": "1", + "CHV_HOST_KEY_MODE": "keyscan"} + 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 *a: self.audit.append(a) + return ex + + def _default_runner(self, cmd, timeout=None): + self.calls.append(list(cmd)) + return FakeResult(0, "", "") + + def _default_api(self, socket_path, method, path, payload=None): + return (200, {}) + + def _default_launcher(self, cmd): + return FakeProc(alive=True) + + def _task(self, task_id="abc123"): + return {"id": task_id, "prompt": "x", "project_id": "proj1"} + + def _audit_events(self): + return [a[0] for a in self.audit] + + +# -- P1-01 / P1-02：生产 reconcile -------------------------------------- + +class FakeProdExecutor: + def __init__(self, state_dir, fail_vms=()): + self.state_dir = state_dir + self.fail_vms = set(fail_vms) + self.shutdowns = [] + self.emits = [] + + def shutdown_vm(self, vm): + self.shutdowns.append(vm) + if vm in self.fail_vms: + raise RuntimeError("shutdown down") + + def _emit(self, event, task, detail): + self.emits.append(event) + + +def _write_registry(state_dir, data): + path = os.path.join(state_dir, "warm-pool.json") + os.makedirs(state_dir, exist_ok=True) + if isinstance(data, str): + with open(path, "w") as fh: + fh.write(data) + else: + with open(path, "w") as fh: + json.dump(data, fh) + return path + + +class ReconcileP1FixTest(unittest.TestCase): + def test_failure_not_cleaned_registry_retained_and_retryable(self): + with tempfile.TemporaryDirectory() as d: + path = _write_registry(d, { + "a": {"state": "warm", "vm": "chv-good"}, + "b": {"state": "warm", "vm": "chv-bad"}, + }) + ex = FakeProdExecutor(d, fail_vms={"chv-bad"}) + self.assertEqual(reconcile_orphans_for_executor(ex), + ["chv-good"]) + self.assertIn("warm_pool_reconcile_failed", ex.emits) + with open(path) as fh: + remaining = json.load(fh) + self.assertEqual( + remaining, + {"b": {"state": "warm", "vm": "chv-bad"}}) + # 修复后重试：失败项可被清理，库最终为空。 + ex.fail_vms = set() + self.assertEqual(reconcile_orphans_for_executor(ex), + ["chv-bad"]) + with open(path) as fh: + self.assertEqual(json.load(fh), {}) + + def test_corrupt_registry_audited_not_silent(self): + with tempfile.TemporaryDirectory() as d: + path = _write_registry(d, "{not json") + ex = FakeProdExecutor(d) + self.assertEqual(reconcile_orphans_for_executor(ex), []) + self.assertIn("warm_pool_registry_corrupt", ex.emits) + self.assertTrue(os.path.exists(path + ".corrupt")) + + def test_invalid_entries_audited_and_counted(self): + with tempfile.TemporaryDirectory() as d: + path = _write_registry(d, { + "a": {"state": "warm", "vm": "chv-a"}, + "c": {"state": "warm"}, + "e": {"state": "running", "vm": "chv-e"}, + }) + ex = FakeProdExecutor(d) + self.assertEqual(reconcile_orphans_for_executor(ex), + ["chv-a"]) + self.assertIn("warm_pool_registry_invalid_entries", + ex.emits) + with open(path) as fh: + self.assertEqual(json.load(fh), {}) + + +# -- P1-03：残留私钥删失败 ---------------------------------------------- + +class StaleSecretsP1FixTest(P1FixCase): + def test_delete_failure_not_cleaned_audited_and_dirty_marked(self): + from cloud.chv.hostkeys import dirty_marker_path + ex = self.make_exec() + vm_dir = os.path.join(self.state_dir, "chv-bad") + os.makedirs(vm_dir, exist_ok=True) + priv = os.path.join(vm_dir, "ssh_host_ed25519_key") + with open(priv, "w") as fh: + fh.write("PRIVATE") + real_remove = os.remove + + def remove(path): + if path == priv: + raise OSError("删不掉") + return real_remove(path) + + with mock.patch.object(os, "remove", side_effect=remove): + cleaned = ex.cleanup_stale_secrets() + self.assertNotIn("chv-bad", cleaned) + self.assertIn("host_secret_residue_delete_failed", + self._audit_events()) + self.assertTrue(os.path.exists( + dirty_marker_path(self.state_dir, "chv-bad"))) + self.assertTrue(os.path.exists(priv)) + + def test_delete_success_cleaned(self): + ex = self.make_exec() + vm_dir = os.path.join(self.state_dir, "chv-ok") + os.makedirs(vm_dir, exist_ok=True) + with open(os.path.join(vm_dir, "ssh_host_ed25519_key"), + "w") as fh: + fh.write("PRIVATE") + cleaned = ex.cleanup_stale_secrets() + self.assertIn("chv-ok", cleaned) + self.assertIn("host_secret_residue_cleaned", + self._audit_events()) + + +# -- P1-04：SSH 回落删除 ------------------------------------------------- + +class SshFallbackP1FixTest(P1FixCase): + def test_undeclared_build_raises_not_loopback(self): + ex = self.make_exec(env_extra={"CHV_SSH_FORWARD": "0"}) + with self.assertRaises(ChvFailure) as ctx: + ex.build_ssh_command(self._task(), ["true"]) + self.assertEqual(ctx.exception.kind, "ssh_target_undeclared") + + def test_explicit_loopback_still_allowed(self): + ex = self.make_exec(env_extra={"CHV_SSH_HOST": "127.0.0.1"}) + cmd = ex.build_ssh_command(self._task(), ["true"]) + self.assertIn("root@127.0.0.1", " ".join(cmd)) + ex2 = self.make_exec(env_extra={"CHV_SSH_FORWARD": "1"}) + cmd2 = ex2.build_ssh_command(self._task(), ["true"]) + self.assertIn("root@127.0.0.1", " ".join(cmd2)) + + +# -- P1-05：flock --------------------------------------------------------- + +class StateLockP1FixTest(unittest.TestCase): + def test_thread_mutual_exclusion(self): + with tempfile.TemporaryDirectory() as d: + state_dir = os.path.join(d, "state") + lock_a = StateLock(state_dir) + lock_b = StateLock(state_dir) + acquired = [] + lock_a.acquire() + try: + def _take(): + # RLock 须由加锁线程释放，故获取与释放同线程。 + lock_b.acquire() + acquired.append(True) + lock_b.release() + + t = threading.Thread(target=_take) + t.start() + time.sleep(0.3) + self.assertEqual(acquired, []) + self.assertTrue(t.is_alive()) + finally: + lock_a.release() + t.join(timeout=5) + self.assertEqual(acquired, [True]) + self.assertTrue(os.path.exists( + lock_path_for_state_dir(state_dir))) + + def test_cross_process_mutual_exclusion(self): + with tempfile.TemporaryDirectory() as d: + state_dir = os.path.join(d, "state") + os.makedirs(state_dir, exist_ok=True) + root = str(Path(__file__).resolve().parents[2]) + ready = os.path.join(state_dir, "child-locked") + script = ( + "import sys, time; sys.path.insert(0, %r); " + "from cloud.chv.locking import StateLock; " + "lock = StateLock(%r); lock.acquire(); " + "open(%r, 'w').write('locked'); " + "time.sleep(10)" % (root, state_dir, ready)) + proc = subprocess.Popen( + [sys.executable, "-c", script], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + try: + deadline = time.time() + 10 + while not os.path.exists(ready) and time.time() < deadline: + time.sleep(0.05) + self.assertTrue(os.path.exists(ready)) + self.assertIsNone(proc.poll()) + lock = StateLock(state_dir) + acquired = [] + + def _take(): + lock.acquire() + acquired.append(True) + lock.release() + + t = threading.Thread(target=_take) + t.start() + time.sleep(0.5) + # 子进程持锁期间，本进程的加锁必须阻塞。 + self.assertEqual(acquired, []) + proc.terminate() + proc.wait(timeout=5) + t.join(timeout=5) + self.assertEqual(acquired, [True]) + finally: + if proc.poll() is None: + proc.kill() + proc.wait(timeout=5) + + def test_purge_snapshot_traversal_under_concurrent_mutation(self): + # 旧无锁 pop + 持锁遍历并发可致字典迭代异常；新实现 + # 快照键列表后逐项持锁删，并发变更加锁路径下不得抛错。 + with tempfile.TemporaryDirectory(): + case = P1FixCase() + case.setUp() + try: + ex = case.make_exec() + ex.state_dir = case.state_dir + stop = threading.Event() + + def mutate(): + i = 0 + while not stop.is_set(): + name = "chv-m%d" % (i % 5) + with ex._vms_lock: + ex._vms[name] = FakeProc(alive=False) + i += 1 + + t = threading.Thread(target=mutate) + t.start() + try: + for _ in range(50): + ex._purge_dead_registry() + finally: + stop.set() + t.join(timeout=5) + finally: + case.tearDown() + + +class WakeSerializationP1FixTest(P1FixCase): + def test_concurrent_wakes_serialized_by_state_lock(self): + # 两路并发 wake：每路 api 往返睡 0.2 秒；串行则总耗时 + # 至少 0.35 秒，并行则约 0.2 秒。阈值留足余量防抖动。 + def sleeping_api(socket_path, method, path, payload=None): + if path == "/vmm.ping": + time.sleep(0.2) + return (200, {}) + + ex1 = self.make_exec(api_client=sleeping_api) + ex2 = self.make_exec(api_client=sleeping_api) + results = [] + start = time.time() + + def wake(ex, task_id): + try: + results.append(ex.wake_vm({"id": task_id})["state"]) + except Exception as exc: # noqa: BLE001 - 失败即测试失败 + results.append("error:%s" % exc) + + t1 = threading.Thread(target=wake, args=(ex1, "s1")) + t2 = threading.Thread(target=wake, args=(ex2, "s2")) + t1.start() + t2.start() + t1.join(timeout=30) + t2.join(timeout=30) + elapsed = time.time() - start + self.assertEqual(sorted(results), ["running", "running"]) + self.assertGreaterEqual(elapsed, 0.35) + self.assertTrue(os.path.exists( + lock_path_for_state_dir(self.state_dir))) + + +# -- P1-06：单条快照超帽 --------------------------------------------------- + +class SingleSnapshotCapP1FixTest(unittest.TestCase): + def test_single_oversize_snapshot_rejected_without_side_effects(self): + with tempfile.TemporaryDirectory() as tmp: + old = os.path.join(tmp, "vm1", "old.snap") + os.makedirs(os.path.join(tmp, "vm1"), exist_ok=True) + with open(old, "wb") as fh: + fh.write(b"O" * 100) + from cloud.chv.disk import _save_snapshots + _save_snapshots(tmp, "vm1", [{ + "url": "file://" + old, "path": old, "size": 100, + "created_at": 0}]) + new = os.path.join(tmp, "vm1", "new.snap") + with open(new, "wb") as fh: + fh.write(b"N" * 1000) + # 总量帽 500 字节：新单条 1000 字节本身已超帽。 + with self.assertRaises(ChvFailure) as ctx: + record_snapshot(tmp, "vm1", "file://" + new, + 500 / (1024 ** 3)) + self.assertEqual(ctx.exception.kind, + "snapshot_quota_exceeded") + # 拒绝无副作用：旧快照文件仍在、登记仍只有旧条目。 + self.assertTrue(os.path.exists(old)) + from cloud.chv.disk import _load_snapshots + entries = _load_snapshots(tmp, "vm1") + self.assertEqual(len(entries), 1) + self.assertEqual(entries[0]["path"], old) + + +# -- P1-07：>0 校验 --------------------------------------------------------- + +class PositiveConfigP1FixTest(P1FixCase): + def test_config_layer_rejects_non_positive(self): + base_env = {} + for env_extra, field in ( + ({"CHV_CPU": "0"}, "cpu"), + ({"CHV_CPU": "-2"}, "cpu"), + ({"CHV_MEM": "0"}, "mem"), + ({"CHV_MEM": "0G"}, "mem"), + ({"CHV_TASK_TIMEOUT": "0"}, "timeout"), + ({"CHV_TASK_TIMEOUT": "-5"}, "timeout")): + env = dict(base_env) + env.update(env_extra) + with self.assertRaises(ValueError, msg=field): + build_chv_config({}, env) + # 合法值与 None（不限）不受影响。 + cfg = build_chv_config({}, {"CHV_TASK_TIMEOUT": "30"}) + self.assertEqual(cfg.task_timeout, 30.0) + cfg = build_chv_config({}, {}) + self.assertIsNone(cfg.task_timeout) + + def test_preflight_rejects_tampered_attributes(self): + for attr, value in (("cpus", "0"), ("cpus", "-1"), + ("memory", "0"), ("task_timeout", 0), + ("task_timeout", -3)): + ex = self.make_exec() + setattr(ex, attr, value) + with self.assertRaises(ChvFailure) as ctx: + ex.preflight() + self.assertEqual(ctx.exception.kind, "config_invalid") + + +# -- P1-08：剪枝审计与快照失败审计 ----------------------------------------- + +class SnapshotAuditP1FixTest(P1FixCase): + def test_retention_prune_delete_failure_audited(self): + ex = self.make_exec() + vm_dir = os.path.join(self.state_dir, "chv-abc123") + os.makedirs(vm_dir, exist_ok=True) + old = os.path.join(self.state_dir, "old.snap") + with open(old, "wb") as fh: + fh.write(b"O" * 10) + with open(os.path.join(vm_dir, "snapshots.json"), "w") as fh: + json.dump([{"url": "file://" + old, "path": old, + "size": 10, "created_at": 0}], fh) + new = os.path.join(self.state_dir, "new.snap") + with open(new, "wb") as fh: + fh.write(b"N" * 10) + real_remove = os.remove + + def remove(path): + if path == old: + raise OSError("删不掉") + return real_remove(path) + + with mock.patch.object(os, "remove", side_effect=remove): + out = ex.snapshot_vm(self._task(), "file://" + new) + self.assertEqual(out, "file://" + new) + self.assertIn("vm_snapshot_prune_failed", + self._audit_events()) + self.assertTrue(os.path.exists(old)) + + def test_snapshot_failure_audited(self): + def failing_runner(cmd, timeout=None): + if cmd and cmd[0] == "ch-remote": + raise RuntimeError("ch-remote 挂了") + return FakeResult(0, "", "") + + ex = self.make_exec(runner=failing_runner) + # 目标须在 state_dir 内才过白名单（ch-remote 失败在后）。 + target = "file://" + os.path.join(self.state_dir, "x.snap") + with self.assertRaises(RuntimeError): + ex.snapshot_vm(self._task(), target) + self.assertIn("vm_snapshot_failed", self._audit_events()) + + +# -- P1-09：4 处三元 else 分支 ---------------------------------------------- + +class BranchCoverP1FixTest(P1FixCase): + def test_pause_resume_snapshot_str_forms(self): + ex = self.make_exec() + os.makedirs(self.state_dir, exist_ok=True) + self.assertTrue(ex.pause_vm("chv-abc")) + self.assertTrue(ex.resume_vm("chv-abc")) + snap = os.path.join(self.state_dir, "snap.raw") + with open(snap, "wb") as fh: + fh.write(b"S" * 10) + out = ex.snapshot_vm("chv-abc", "file://" + snap) + self.assertEqual(out, "file://" + snap) + self.assertIn("vm_pause", self._audit_events()) + self.assertIn("vm_resume", self._audit_events()) + self.assertIn("vm_snapshot", self._audit_events()) + + def test_spill_audit_non_dict_task(self): + ex = self.make_exec() + os.makedirs(self.state_dir, exist_ok=True) + ex._spill_audit("ev", "plain-task-id", "detail", None) + path = os.path.join(self.state_dir, "chv-audit-spill.jsonl") + with open(path) as fh: + row = json.loads(fh.readline()) + self.assertEqual(row["task_id"], "plain-task-id") + self.assertEqual(row["event"], "ev") + + +# -- 覆盖补齐：新分支的行覆盖（诚实口径，不凑数改代码） -------- + +class CoverageCompletionP1FixTest(P1FixCase): + def test_prune_events_no_file_and_no_delete(self): + from cloud.chv.disk import _save_snapshots + with tempfile.TemporaryDirectory() as tmp: + _save_snapshots(tmp, "vm1", [ + {"url": "file:///gone.snap", "path": "/gone.snap", + "size": 5, "created_at": 0}, + {"url": "https://example.invalid/x", "path": None, + "size": 5, "created_at": 0}]) + events = [] + kept, pruned = prune_snapshots( + tmp, "vm1", retention_days=30, now=10 ** 9, + events=events) + self.assertEqual(kept, []) + self.assertEqual(len(pruned), 2) + kinds = [e[0] for e in events] + self.assertEqual(kinds, ["pruned_no_file", "pruned_no_file"]) + events2 = [] + _save_snapshots(tmp, "vm1", [ + {"url": "file:///gone2.snap", "path": "/gone2.snap", + "size": 5, "created_at": 0}]) + kept2, _ = prune_snapshots( + tmp, "vm1", retention_days=30, now=10 ** 9, + delete_files=False, events=events2) + self.assertEqual(kept2, []) + self.assertEqual([e[0] for e in events2], + ["pruned_no_file"]) + + def test_record_without_cap_and_tighten_delete_failure_event(self): + with tempfile.TemporaryDirectory() as tmp: + entries, removed = record_snapshot( + tmp, "vm1", "https://example.invalid/new", None) + self.assertEqual(len(entries), 1) + self.assertEqual(removed, []) + old = os.path.join(tmp, "vm1", "old.snap") + os.makedirs(os.path.join(tmp, "vm1"), exist_ok=True) + with open(old, "wb") as fh: + fh.write(b"O" * 100) + from cloud.chv.disk import _save_snapshots + _save_snapshots(tmp, "vm1", [ + {"url": "file://" + old, "path": old, "size": 100, + "created_at": 0}]) + real_remove = os.remove + + def remove(path): + if path == old: + raise OSError("删不掉") + return real_remove(path) + + events = [] + with mock.patch.object(os, "remove", side_effect=remove): + entries2, removed2 = record_snapshot( + tmp, "vm1", "https://example.invalid/new2", + 0.0, events=events) + self.assertEqual(len(removed2), 1) + self.assertEqual([e[0] for e in events], ["delete_failed"]) + self.assertTrue(os.path.exists(old)) + + def test_tighten_no_file_entry_event(self): + from cloud.chv.disk import _save_snapshots + with tempfile.TemporaryDirectory() as tmp: + _save_snapshots(tmp, "vm1", [ + {"url": "https://example.invalid/old", "path": None, + "size": 100, "created_at": 0}]) + events = [] + entries, removed = record_snapshot( + tmp, "vm1", "https://example.invalid/new", + 0.0, events=events) + self.assertEqual(len(removed), 1) + self.assertEqual([e[0] for e in events], + ["pruned_no_file"]) + + def test_state_lock_ctx_refreshes_on_state_dir_change(self): + ex = self.make_exec() + old_lock = ex._state_lock + new_dir = os.path.join(self.dir, "state2") + ex.state_dir = new_dir + self.assertIsNot(ex._state_lock_ctx(), old_lock) + self.assertEqual(ex._state_lock.state_dir, new_dir) + + def test_gc_and_secrets_listdir_failure_returns_empty(self): + ex = self.make_exec() + ex.state_dir = os.path.join(self.dir, "missing-dir") + self.assertEqual(ex.gc_stale_disks(), + {"deleted": [], "failed": []}) + self.assertEqual(ex.cleanup_stale_secrets(), []) + + def test_cleanup_listdir_failure_returns_empty(self): + ex = self.make_exec() + os.makedirs(self.state_dir, exist_ok=True) + real_listdir = os.listdir + calls = {"n": 0} + + def listdir(path): + calls["n"] += 1 + if calls["n"] >= 3: + raise OSError("listdir 挂了") + return real_listdir(path) + + with mock.patch.object(os, "listdir", side_effect=listdir): + self.assertEqual(ex.cleanup_stale_vms(), []) + + def test_orphan_entries_compat_wrapper_and_no_emit_executor(self): + from cloud.chv.orphans import _orphan_entries + data = {"a": {"state": "warm", "vm": "chv-a", + "task_id": "a", "snapshot_url": None}, + "b": {"state": "running", "vm": "chv-b"}} + self.assertEqual(_orphan_entries(data), + [{"vm": "chv-a", "task_id": "a", + "snapshot_url": None}]) + + class NoEmit: + def __init__(self, state_dir): + self.state_dir = state_dir + self.shutdowns = [] + + def shutdown_vm(self, vm): + self.shutdowns.append(vm) + + with tempfile.TemporaryDirectory() as d: + _write_registry(d, { + "a": {"state": "warm", "vm": "chv-a"}}) + ex = NoEmit(d) + self.assertEqual(reconcile_orphans_for_executor(ex), + ["chv-a"]) + + def test_state_lock_flock_failure_closes_and_releases(self): + import fcntl + with tempfile.TemporaryDirectory() as d: + state_dir = os.path.join(d, "state") + lock = StateLock(state_dir) + with mock.patch.object(fcntl, "flock", + side_effect=OSError("flock 挂了")): + with self.assertRaises(OSError): + lock.acquire() + self.assertIsNone(lock._fd) + # 失败后锁仍可用（RLock 已释放、fd 已关闭）。 + lock.acquire() + lock.release() + + def test_pool_reconcile_failed_emit_failure_swallowed(self): + from cloud.chv.experimental_pool import WarmPool + stub = _PoolStubForP1Fix() + pool = WarmPool(stub, registry_path=None) + pool._entries["old-1"] = {"state": "orphan", "vm": "vm-old", + "task": {"id": "old-1"}} + self.assertEqual(pool.reconcile_orphans(), []) + self.assertIn("old-1", pool._entries) + + +class _PoolStubForP1Fix: + """实验池失败审计吞错测试用假执行器。""" + + def shutdown_vm(self, task_or_name): + raise RuntimeError("shutdown 失败") + + def _emit(self, event, task, detail="", kind=None): + raise RuntimeError("audit 失败") + + +if __name__ == "__main__": + unittest.main() diff --git a/docs/chv-architecture.md b/docs/chv-architecture.md index ed06307..297830d 100644 --- a/docs/chv-architecture.md +++ b/docs/chv-architecture.md @@ -1,6 +1,6 @@ # Cloud Hypervisor 执行器架构决策记录（ADR） -范围：cloud/chv/ 执行器链（#216 → #221 → #230 → #232 → #234 → #239 → #246 → #252 → #254 → 本轨 feat-chv-arch-maint-fe）。 +范围：cloud/chv/ 执行器链（#216 → #221 → #230 → #232 → #234 → #239 → #246 → #252 → #254 → feat-chv-arch-maint-fe → feat-chv-closeup → feat-chv-cover100 → 本轨 feat-chv-p1fix）。 本文件只记录已落代码的决策，不记录未验证为已验证；真机、压测、动态安全 仍见 docs/chv-production-readiness.md §11。 @@ -323,3 +323,35 @@ 缺 268 → 242）；未覆盖大头仍是 executor 深层生命周期 分支（需假 VMM 全链路）与 disk 配额分支，不靠删测试 或改分母凑数。 + +## ADR-17 P1 修复轨语义收紧（feat-chv-p1fix，生产级审核 §5） + +本轨只修审核报告 §5 的 P1 十项（P2 不动），决策均为 +“失败可见、猜测收紧、竞态加锁”，不放宽任何既有安全语义： + +- 失败不算成功：孤儿清理失败项不进 cleaned、不清库、 + 落盘保留失败项（P1-01）；残留私钥删失败不进 cleaned、 + 发审计并写脏标记（P1-03）；单条超帽快照拒绝记录而不 + 是悄悄保留（P1-06）；快照失败与剪枝删失败均发审计 + （P1-08）。“尽力清理、不因单条失败卡住整批”的旧语义 + 被修正为“尽力清理、失败项留痕可重试”——整批仍不卡， + 但失败不再被成功名单掩盖。 +- 损坏与非法不静默：registry 损坏与非法条目均发审计并 + 计数（P1-02）；非法条目落盘时丢弃（已审计，不再每次 + 重启重复审计）。 +- 不猜目标：SSH 命令构造删除回环回落，失败即抛 + （P1-04）；显式回环（CHV_SSH_HOST / CHV_SSH_FORWARD=1） + 仍可用，wake/wait 严格路径不受影响。 +- 竞态用文件锁：容量/net/磁盘 check→act 持 + `state_dir/chv-state.lock` 的 flock 全程（P1-05）， + 拒绝“只加进程内锁就声称多 worker 已解决”的写法； + 串行 wake 是有意的代价，已运行 VM 的任务执行不持锁。 +- 非正配置在入口拦：CPU/内存/任务超时非正在构造层与 + preflight 层双重校验（P1-07）；None 任务超时 = 不限 + 的旧语义保留。 +- 覆盖口径诚实：行覆盖与分支覆盖分开报（P1-09）； + `__init__.py` 计入 coverage.py 分母（4 行）；长函数 + 50 行硬门禁靠提取维持，新增 helper 均有命名职责 + （`_snapshot_create`、`_tighten_snapshot_cap`、 + `_preflight_task_timeout_check`、`_emit_prune_events`、 + `_state_lock_ctx`），不是为凑行数而拆。 diff --git a/docs/chv-production-readiness.md b/docs/chv-production-readiness.md index 97b4db1..7413b9b 100644 --- a/docs/chv-production-readiness.md +++ b/docs/chv-production-readiness.md @@ -580,6 +580,9 @@ ### 14.1 本轨量化结果（实测，非上一轨回执转述） +（本节数字为 closeup 轨快照，历史数字保留不改； +P1 修复轨后的最新数字见 §15，§14.1 不再是最新口径。） + - 后端：232 项 → 317 项（新增 `cloud/tests/test_chv_closeup.py` 85 项），全过。命令（仓库根目录，cwd 须为根目录—— 在 `cloud/` 下跑会撞 `cloud/queue.py` 遮蔽 stdlib queue， @@ -761,6 +764,10 @@ ### 14.7 本轨逐项评分（目标 10，无一项硬写 10） +（本节评分为 closeup 轨快照；P1 修复轨后的评分更新见 +§15.7，其中“80 行未覆”“317 项/36 项”扣分理由已过期， +不再作为当前扣分依据。） + - 代码质量 9.8 → 9.8，未达 10：executor/worker_adapter 覆盖到 100%，但 chv 仍有 80 行未覆（非 executor 模块），且默认口径 mypy 余 2 错在作用域外。差：代码内。 @@ -785,3 +792,194 @@ UNSTABLE、未合并、未部署、真机 0 项。差：外部为主。 - 动态安全 2.0 → 2.0，未达 10：本轨无任何动态验证， V-7 未跑。差：外部（KVM 宿主 + 审计结论）。 +## 15. P1 修复轨 feat-chv-p1fix（堆叠于 #270 之上） + +本节是工程生产级审核（`~/workspace/chv-prod-audit/REPORT.md`， +审核对象 #270，提交 `8c8d62f`）§5 的 P1-01～P1-10 修复轨。 +本轨只到 PR OPEN，不合并、不部署、不做合并试验（P0-01 +仍未做，见 §15.6）；P2-01～P2-21 一律不动。 + +### 15.1 本轨量化结果（实测） + +- 后端：362 项 → 389 项（新增 `cloud/tests/test_chv_p1fix.py` + 27 项），全过。命令（仓库根目录）： + `python3 -m unittest cloud.tests.test_chv_arch95 + cloud.tests.test_chv_arch95plus cloud.tests.test_chv_arch_maint_fe + cloud.tests.test_chv_fix cloud.tests.test_chv_gaps + cloud.tests.test_chv_optimize cloud.tests.test_chv_secfix + cloud.tests.test_chv_to10 cloud.tests.test_cloud_hypervisor + cloud.tests.test_chv_closeup cloud.tests.test_chv_cover100 + cloud.tests.test_chv_p1fix` +- 覆盖率（同命令 + `coverage run --source=cloud/chv`，再 + `--append test_cloud_hypervisor`，`.coverage` 写 /tmp）： + 行覆盖 100%（2097 行，缺 0）；分支覆盖 99% + （714 分支，28 处部分分支未全走——行 100% 不等于 + 分支 100%，PR 描述按此口径写，不写裸“100%”）。 + `cloud/chv/__init__.py` 计入分母：4 行，100%（仅 + docstring + 3 个 re-export，coverage.py 口径；审核报告 + 的 trace 口径把它排除在外，两口径差在此说明，不混用）。 + 审核报告 P1-09 点名的 4 处三元 else 分支（str 形态 + pause/resume/snapshot + 非 dict spill task）已由 + `test_chv_p1fix.py` 的 BranchCover 测试覆盖。 +- mypy：silent 口径 `cloud/chv/` 0 错（不变）； + 默认口径 `cloud/marker_tools.py` 仍 2 错（作用域外， + 本轨未改该文件）。 +- ruff：`--select F,E9,W,I` 口径 `cloud/chv/` + + 本轨新测试 0 错（不变；默认全规则基线不声称通过）。 +- 前端：Vitest 37 项不变（`chv.test.ts` 23 + + `chv.closeup.test.ts` 14），本轨未改前端代码；三轮 + 连跑 37/37/37，无抖动回归。 +- 全量：`python3 -m unittest discover -s cloud/tests` + （仓库根目录）1819 项全过（skipped=8），392 秒； + 上一轨引用口径 1792 项的差额来自本轨新增 27 项。 +- 长函数：AST 扫描 `cloud/chv/`，>50 行函数 0 个 + （本轨新增的 snapshot_vm / record_snapshot / + _preflight_numeric_checks 均经提取守住 50 行门禁， + 见 ADR-17）。 +- §13.1 的 main 头 `aebc6238` 与前端产物哈希 + `269820d1…f389` 仍是 to10 轨快照，本轨不更新它们： + 合并前必须按 §13.1 逐项重跑重核，绝不把旧快照写成 + 新数字（审核报告 P1-10 的要求原样保留）。 + +### 15.2 P1 逐项修法（证据：文件与测试名） + +- P1-01（orphans.py + experimental_pool.py）：失败项不进 + cleaned、不触发清库；生产 reconcile 落盘只写回失败项 + 原始条目并审计 `warm_pool_reconcile_failed`，下次可重试； + 实验池失败条目留在内存条目里（文件落盘语义属 P2-16， + 本轨不改）。同步改旧测试：`test_chv_arch_maint_fe.py` + 的 shutdown 失败用例与 `test_chv_cover100.py` 的 + `test_reconcile_shutdown_failure_still_cleans`（已改名 + `test_reconcile_shutdown_failure_not_cleaned`）——代码 + 与测试同改，不存在“只改一侧”。 + 新测试：`ReconcileP1FixTest.test_failure_not_cleaned_registry_retained_and_retryable`。 +- P1-02（orphans.py）：registry 损坏发 + `warm_pool_registry_corrupt` 审计（quarantine 后仍审计， + 不静默返回空）；非法条目发 + `warm_pool_registry_invalid_entries` 审计并计数，落盘 + 时丢弃已审计条目（不再每次重启重复审计）。 + 新测试：`test_corrupt_registry_audited_not_silent`、 + `test_invalid_entries_audited_and_counted`。 +- P1-03（executor.py `_clean_one_vm_secrets` / + `cleanup_stale_secrets`）：返回值由 2 元组改为 + (removed_priv, removed_seed, failed)，删失败发 + `host_secret_residue_delete_failed` 审计并写脏标记 + （下次 provision 见标记即 `vm_dirty_refused` 拒起）， + 删失败的 VM 不进 cleaned（即使部分删掉也不算已清理）。 + 旧测试 `test_clean_one_vm_secrets_remove_failures` 已 + 同步改 3 元组解包与 failed 断言。 +- P1-04（executor.py）：删除 `build_ssh_command` 的回环 + 回落（构造失败即抛 `ssh_target_undeclared`），不设 + dry-run 例外参数——干跑同样不许猜目标；同款回落 + `_write_provisioned_known_hosts` 一并删除（wake 路径 + 在 provision 前已在 `_wake_prechecks` 里严格解析， + 不受影响；wait 路径的 `capture_host_key` 本来就 + 解析失败返回 False，同样严格）。显式回环仍可用： + `CHV_SSH_HOST=127.0.0.1` 或 `CHV_SSH_FORWARD=1`。 + 旧测试 `test_build_ssh_command_fallback_and_secret_not_delivered` + 与 `test_write_provisioned_known_hosts_fallback` 已 + 同步改为抛错断言。 +- P1-05（新增 `cloud/chv/locking.py` + executor.py）： + 选文件锁方案（非单 worker 声明）：`StateLock` = + `flock`（`state_dir/chv-state.lock`，永不删除）+ + 进程内 RLock 可重入（同一实例嵌套不自锁；多实例/ + 多进程靠 flock 互斥）。持有范围：wake_vm 全程、 + restore_vm 全程、cleanup_stale_vms 全程、orphans + reconcile 全程；`_purge_dead_registry` 改为锁内快照 + 键列表、逐项锁内读/删，`_stop_dead_socket_vms` 的 + 登记读持锁。串行 wake 是有意的代价（容量门只有 + “查完立即占位”才成立，提前释放会在 socket 未绑定 + 前留出计数窗口）；已运行 VM 的任务执行不持此锁。 + 新测试：`StateLockP1FixTest`（线程互斥、跨进程互斥、 + purge 并发变更加锁）、`WakeSerializationP1FixTest` + （两路并发 wake 实测串行化）。 +- P1-06（disk.py `record_snapshot`）：单条快照本身大于 + 总量帽时，在任何落盘/删旧动作之前直接拒绝 + （`snapshot_quota_exceeded`），不悄悄保留超大快照； + 大小未知（size 为 None，如远端目标）时无法判定， + 沿用旧口径。新测试： + `SingleSnapshotCapP1FixTest`（拒绝无副作用：旧文件 + 与旧登记均在）。 +- P1-07（config.py + executor.py preflight）：构造层 + 拒绝能解析出的 CPU/内存非正值与显式任务超时非正值 + （`CHV_CPU`/`CHV_MEM`/`CHV_TASK_TIMEOUT`）；preflight + 再验属性事后改写（CPU 0/负、内存 0、任务超时 0/负 + 一律 `config_invalid` 干净失败，不进 VmConfig）。 + `None` 任务超时 = 不限（旧语义保留）；解析不出的 + 格式仍留给 preflight（构造时机分工不变）。 + 新测试：`PositiveConfigP1FixTest`。 +- P1-08（disk.py + executor.py `snapshot_vm`）：剪枝 + 事件外发（`prune_snapshots` / `check_snapshot_limits` / + `record_snapshot` 均接受 `events` 列表：deleted / + delete_failed / pruned_no_file）；`snapshot_vm` 据此发 + `vm_snapshot_pruned` / `vm_snapshot_prune_failed` + 审计，失败路径（白名单/上限/ch-remote/落盘校验/ + 登记拒绝）统一发 `vm_snapshot_failed` 审计再抛。 + 新测试：`SnapshotAuditP1FixTest`。 +- P1-09：见 §15.1（行 100% / 分支 99% 分开写； + `__init__.py` 计入分母说明；4 分支测试在 + `BranchCoverP1FixTest`）。 +- P1-10：本节即同步结果；§14.1/§14.7 已加“已被 §15 + 取代”指针（历史数字保留不改）；架构文档头范围与 + ADR-17 见 `docs/chv-architecture.md`；§13.1 快照数 + 字按要求保留、合并前重跑（见 §15.1 末）。 + +### 15.3 安全语义未放宽清单 + +egress 默认 fail-closed、key seed 投递（provisioned 才 +投递、keyscan 下密钥 fail-closed）、磁盘删除（删失败 +审计 + 脏标记）、默认 provisioned、tap/mac/ip 校验、 +StrictHostKeyChecking 钉死、网关 loud fail 守卫—— +本轨一律未改动，只收紧（回落删除、失败可见性、 +check→act 加锁）。前端无改动，不存在绕后端白名单 +的新路径。 + +### 15.4 新增与改动文件 + +- 新增：`cloud/chv/locking.py`（StateLock）、 + `cloud/tests/test_chv_p1fix.py`（27 项）。 +- 改动：`cloud/chv/orphans.py`、`cloud/chv/disk.py`、 + `cloud/chv/executor.py`、`cloud/chv/experimental_pool.py`、 + `cloud/chv/config.py`；旧测试同步改： + `cloud/tests/test_chv_arch_maint_fe.py`、 + `cloud/tests/test_chv_closeup.py`、 + `cloud/tests/test_chv_cover100.py`； + 文档：本节 + `docs/chv-architecture.md`（头范围 + + ADR-17）。 + +### 15.5 逐项完成情况 + +P1-01～P1-10 全部修完（每项均有新测试或同步改的旧 +测试，见 §15.2；无“修不动”的项）。未修完的部分： +无。但“修完”只等于代码层完成，不等于生产就绪， +见 §15.6。 + +### 15.6 仍未验证项（原样保留，不冒充） + +- 合并试验 P0-01 仍未做（本轨禁令）：隔离演练区以 + 当前远端 HEAD 为目标的三方合并 + 全量测试未跑， + 直接合入仍可能以链侧文件覆盖 main 新增功能。 +- CI 仍 UNSTABLE（账单秒级失败，本轨未改 CI 配置）。 +- KVM 宿主 provision、客机镜像真实构建与发布未做； + 真机验证 V-1～V-9 全部未跑。 +- 前端 live 点击验证未做；渲染测试未做。 +- 旧安全审计 No-Go 未复审、未推翻。 + +### 15.7 本轨逐项评分更新（目标 10，无一项硬写 10） + +- 代码质量 9.8 → 9.8，未达 10：P1 语义缺陷已修，但 + 分支覆盖 99% 未满、长函数门禁靠提取维持（边界脆弱）。 + 差：代码内（小）。 +- 测试 9.8 → 9.9，未达 10：后端 389 项、前端 37 项 + 全过，行覆盖 100% 恢复且口径诚实分开写；但真机 + 用例 0 项已跑、分支 28 处部分。差：外部（真机）为主。 +- 文档 9.9 → 9.9，未达 10：§15 与 ADR-17 已同步，但 + §13.1 仍是旧快照、真机证据表仍空。差：外部。 +- 架构 9.7 → 9.7，未达 10：新增 flock 锁是并发语义 + 收紧而非架构重构，ADR 取舍不变。差：外部。 +- 前端 9.6 → 9.6，未达 10：本轨未改前端。差：代码内 + （有意不做）+ 外部（live）。 +- 生产就绪 5.5 → 5.5，未达 10：P0-01 未做、CI 未绿、 + 未合并、未部署、真机 0 项。差：外部为主。 +- 动态安全 2.0 → 2.0，未达 10：本轨无任何动态验证。 + 差：外部。