"""CHV 上一世残留 VM 清理(生产路径唯一入口)。 本模块是叶子模块,只依赖标准库,绝不 import cloud.chv.experimental_pool(实验接口)与 cloud.sandbox。 生产代码(worker 启动/关闭)只调 reconcile_orphans_for_executor(),它直接读 registry 文件、 对每个上一世条目调 executor.shutdown_vm(),不构造任何 池对象、不调 prewarm/acquire/release。 Registry 语义与旧 WarmPool._load_registry 完全一致 (feat-chv-arch95plus 的 fail-safe 口径,本轨只搬位置、 不改语义): - 文件:state_dir/warm-pool.json,内容为 {<合成名>: {"state", "vm", "task_id", "snapshot_url"}}。 - 读到的 warm/assigned 条目一律当 orphan:旧 VMM 进程 已随旧 worker 消失,无法验证它还是当初那个干净快照 VM,绝不发给任务,只能清理。 - 损坏 fail-safe:文件不是合法 JSON / 顶层不是 dict 时, 整个文件 quarantine 成 .corrupt,清理从空开始; 单条目结构非法(无 vm、或 state 不在 warm/assigned) 时只丢该条,不连坐其它条目。 - 清理后落盘为空 dict(原子 tmp + rename);写失败只记 日志警告,不炸 worker 启动/关闭。shutdown_vm 抛错时该 条目仍计入已清理(旧 reconcile 语义:尽力清理、不因 单条失败卡住整批),_emit 失败同样 best-effort。 """ import json import logging import os _LOGGER = logging.getLogger("cloud.cloud_hypervisor") _REGISTRY_FILENAME = "warm-pool.json" def registry_path_for_executor(executor: object) -> str | None: """执行器的 registry 路径:state_dir 下的 warm-pool.json。 无 state_dir 时返回 None(调用方按“无残留”处理)。""" state_dir = getattr(executor, "state_dir", None) if not state_dir: return None return os.path.join(state_dir, _REGISTRY_FILENAME) def _quarantine_registry(path: str) -> None: """把损坏的 registry 文件挪到 .corrupt 旁(尽力而为)。""" try: os.replace(path, path + ".corrupt") except OSError: _LOGGER.warning("chv_pool_registry_quarantine_failed " "path=%s", path) def _read_registry(path: str) -> tuple[dict | None, bool]: """读 registry 原始 dict:(data, corrupt)。 无文件 → (None, False);损坏 → quarantine 后 (None, True); 顶层非 dict 同样算损坏。调用方按 corrupt 决定是否记 registry_corrupt(实验池需要区分“无文件”与“损坏”)。""" try: with open(path, "r") as fh: data = json.load(fh) except FileNotFoundError: return None, False except (OSError, ValueError): _quarantine_registry(path) return None, True if not isinstance(data, dict): _quarantine_registry(path) return None, True return data, False def _write_registry(path: str, data: dict) -> None: """原子落盘 registry(tmp + rename);写失败只记日志。""" try: directory = os.path.dirname(path) if directory: os.makedirs(directory, exist_ok=True) tmp_path = path + ".tmp" with open(tmp_path, "w") as fh: json.dump(data, fh) os.replace(tmp_path, path) except OSError: _LOGGER.warning("chv_pool_registry_save_failed path=%s", path) def _orphan_entries(data: dict) -> list[dict]: """从原始 registry dict 里筛出可清理的上一世条目。 只留 state 为 warm/assigned 且有 vm 的条目;非法条目 静默丢弃(不连坐)。返回条目为 {"vm", "task_id", "snapshot_url"} 的新 dict,不改原数据。""" entries: list[dict] = [] for entry in data.values(): if not isinstance(entry, dict): continue if not entry.get("vm"): continue if entry.get("state") not in ("warm", "assigned"): continue entries.append({ "vm": entry["vm"], "task_id": entry.get("task_id"), "snapshot_url": entry.get("snapshot_url"), }) return entries def reconcile_orphans_for_executor(executor: object) -> list[str]: """生产路径唯一的残留清理入口:返回已清理的 VM 名列表。 只做清理:读 registry → 每个上一世条目尽力 shutdown_vm → 落盘为空 → best-effort 审计事件。 绝不构造 WarmPool、绝不调 prewarm/acquire/release, 绝不把任何条目发给任务。无 registry 路径时返回 []。""" path = registry_path_for_executor(executor) if not path: return [] data, _corrupt = _read_registry(path) if data is None: return [] entries = _orphan_entries(data) cleaned: list[str] = [] for entry in entries: vm = entry["vm"] try: executor.shutdown_vm(vm) # type: ignore[attr-defined] except Exception: pass cleaned.append(vm) if cleaned: _write_registry(path, {}) try: executor._emit( # type: ignore[attr-defined] "warm_pool_reconcile", {"id": None}, "cleaned=%s" % ",".join(cleaned)) except Exception: pass return cleaned