diff --git a/cloud/chv/executor.py b/cloud/chv/executor.py index c9920a8..95fa576 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 @@ -81,6 +82,46 @@ from cloud.executor_base import Executor _LOGGER = logging.getLogger("cloud.cloud_hypervisor") +# -- 混合模式 B 轨（feat-chv-hybrid-b，设计 DESIGN-HYBRID.md §9.1 +# 第 5–9 项；已定值直接用，不许改回提案口径）------------------- +# 常驻资格档位（§5）：只有 Paid/Heavy 有常驻资格，Free 无论多 +# 高频都没有。 +RESIDENT_TIERS = ("Paid", "Heavy") +# 常驻租约状态（§4.1）：登记 state 字段取值。legacy 的 +# paused/running 是阶段 1 持久 VM 的旧状态，不是常驻态。 +RESIDENT_STATES = ("resident-active", "resident-paused", + "hibernated", "demoted") +# 常驻有效态（占槽、续期、认领、网关放行都只认这三个）： +# demoted 已收回租约，不占槽、不续期、不复活。 +RESIDENT_ACTIVE_STATES = ("resident-active", "resident-paused", + "hibernated") +# 认领时可 resume 的登记状态：legacy paused + 常驻态里 VM 已 +# 暂停的两种（resident-active 的 VM 在任务间隙同样被 +# suspend_after_task 暂停；hibernated 的 VM 已停机，须走 +# restore，不进 resume 认领；demoted 已收回租约，不认领）。 +RESIDENT_CLAIMABLE_STATES = ("paused", "resident-active", + "resident-paused") +# 租约阈值（已定值，§4.2）：闲置 30 分钟挂起，挂起后再闲置 +# 24 小时降级（累计 24.5 小时无活动）。生产默认值不变； +# 测试用环境变量注入短阈值（见 _resident_thresholds）。 +RESIDENT_IDLE_SUSPEND_SECONDS = 30 * 60 +RESIDENT_IDLE_DEMOTE_SECONDS = 24 * 3600 +# 每宿主常驻上限（已定值，§4.1）：宿主 VM 槽位的 50%。 +RESIDENT_MAX_FRACTION = 0.5 +# 排队超时（已定值，§3.4）：无可用宿主时等 300 秒再明确拒绝。 +QUEUE_TIMEOUT_SECONDS = 300 +QUEUE_POLL_SECONDS = 5 +# 升级提示点（已定值，§5）：30 天内池任务 ≥100 次，或连续 7 天 +# 每天 ≥3 次；一周内同一租户最多记一次。 +UPGRADE_HINT_WINDOW_DAYS = 30 +UPGRADE_HINT_MIN_TASKS = 100 +UPGRADE_HINT_STREAK_DAYS = 7 +UPGRADE_HINT_MIN_PER_DAY = 3 +UPGRADE_HINT_DEDUP_DAYS = 7 +# 网关 provider（§6）：目前唯一的长连接网关是 sub2api。 +GATEWAY_PROVIDERS = ("sub2api",) + + def _pid_gone(pid): """进程是否已不在（/proc 消失，或已成僵尸：父进程未收， 但内存/tap 等资源已释放，等同已死）。""" @@ -365,12 +406,20 @@ class CloudHypervisorExecutor(Executor): - worker 层现状：通用路径已能走 dispatch_task/goal_task； 断点续跑（goal-resume-breakpoint）已由补缺口轨接线： worker 通用分支对任何提供该方法的执行器调用它， - Local 仍走原路径；网关任务仍被 worker 侧 loud fail - 守卫拦住，不放宽网关规则。 + Local 仍走原路径；网关任务的 worker 侧守卫（P1-1 起） + 已放行 CHV，网关分流改由本执行器按 VM 类 enforcement + （feat-chv-hybrid-b，见 gateway_allowed_for_task）： + 常驻 VM 可接网关、池 VM 不接（ChvFailure + gateway_pool_denied），改之前网关不上 CHV 的前置 + 依赖即由此落地。 - 网关现状：worker 通用路径在 sub2api 网关下对非 Local 执行器直接 loud fail（与 CloudExecutor 同款），v1 不 改这条门，所以 CHV + sub2api 组合目前在 worker 层就 - 明确失败，不宣称生产可用。 + 明确失败，不宣称生产可用。——本段是 v1 旧口径，已被 + feat-chv-hybrid-b 取代：worker 守卫自 P1-1 起已放行 + CHV，网关分流现由本执行器的 gateway_allowed_for_task + 按常驻/池分流（常驻可接、池不接）；旧文案留此作 + 演变记录，执行以 gateway_allowed_for_task 为准。 配置（只走构造参数或环境变量，不写文件；参数优先）： - CHV_BINARY：cloud-hypervisor 二进制（默认同名 PATH 查找） @@ -552,6 +601,7 @@ class CloudHypervisorExecutor(Executor): self.sleeper = sleeper if sleeper is not None else time.sleep self.audit_callback = None self.audit_failure_count = 0 + self._init_hybrid_b_runtime() self._vms_lock = threading.Lock() if runner is not None: self.runner = runner @@ -566,6 +616,17 @@ class CloudHypervisorExecutor(Executor): self.launcher = _default_popen_launcher self._vms = {} + def _init_hybrid_b_runtime(self): + """B 轨运行期状态（_init_runtime 的一段，拆出只为 + 满足 50 行函数上限，语义不变）：排队等待计数 + （_wait_for_capacity 在等待期间 +1，退出时 -1； + queue_depth() 供 L1 扩缩读队列深度）与租约降级前 + 的同步钩子（见 _sync_resident_before_demote； + None 即默认口径）。""" + self._queue_waiters = 0 + self._queue_lock = threading.Lock() + self.resident_sync_hook = None + def _delivered_secrets(self, name: str) -> set: """P2-20 唯一读入口：该 VM 已 seed 投递的密钥名集合。""" with self._guest_env_lock: @@ -2227,10 +2288,16 @@ class CloudHypervisorExecutor(Executor): tenant_id, extra_env, sync_workspace) self.pause_vm(vm_task) import time as _time - write_persistent_vm(self.state_dir, tenant_id, { + # feat-chv-hybrid-b：合并写（tier/last_active_at 等 + # 不再被字面量覆盖丢掉）；已有常驻态登记时保持其 + # 状态（预热不降级常驻租约），否则记 legacy paused。 + _prior = read_persistent_vm(self.state_dir, tenant_id) or {} + _state = _prior.get("state") \ + if _prior.get("state") in RESIDENT_STATES else "paused" + self._write_resident_record(tenant_id, { "vm": vm, "tenant_id": tenant_id, - "state": "paused", "origin": "prewarm", - "gitea_synced": synced, "updated_at": _time.time()}) + "state": _state, "origin": "prewarm", + "gitea_synced": synced}, now=_time.time()) self._emit("warm_pool_prewarm", vm_task, "vm=%s gitea_synced=%s" % (vm, synced)) return {"pooled": True, "vm": vm, "mode": "prewarmed", @@ -2259,13 +2326,36 @@ class CloudHypervisorExecutor(Executor): (vm, record)；任一不符即清登记、返回 (None, None) （调用方回落冷启）。归属不符与 socket 消失是登记 本身已坏，清掉；状态未知（vm.info 取不到）不在此 - 判——那是认领时的第二道门，见 acquire_for_task。""" + 判——那是认领时的第二道门，见 acquire_for_task。 + feat-chv-hybrid-b：可认领态扩展为 legacy paused + + 常驻态里 VM 已暂停的两种（resident-active / + resident-paused，见 RESIDENT_CLAIMABLE_STATES）； + 常驻态登记新增 tier 校验——tier 不是 Paid/Heavy + 即不认领（登记本身是坏的/被降档的，留给清理，不 + 静默按旧语义 resume；legacy paused 登记无 tier + 字段，走旧语义，不因此拒认）。""" record = read_persistent_vm(self.state_dir, tenant) - if record is None or record.get("state") != "paused": + if record is None \ + or record.get("state") not in RESIDENT_CLAIMABLE_STATES: return None, None vm = record.get("vm") if not vm: return None, None + if record.get("state") in RESIDENT_ACTIVE_STATES \ + and not self.resident_tier_eligible( + tenant, record.get("tier")): + # 常驻态却无常驻资格：不认领、不 resume；登记留给 + # 租约计时器/清理处理（不静默清掉——删租约是降级 + # 流程的职责，不是认领的职责）。 + self._emit("resident_tier_denied", + {"id": "persist-%s" % tenant, + "project_id": tenant}, + "host_id=%s tenant=%s vm=%s tier=%s 常驻态登记" + "无常驻资格，不认领、回落冷启" + % (self._host_id(), tenant, vm, + record.get("tier")), + "tier_denied") + return None, None owner = read_vm_tenant(self.state_dir, vm) if owner is not None and owner != tenant: # 盘的归属与请求方不一致：这台不是你的 VM，绝不 @@ -2323,12 +2413,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, { - "vm": vm, "tenant_id": tenant, - "state": "running", - "origin": record.get("origin") or "task", - "gitea_synced": bool(record.get("gitea_synced")), - "updated_at": _time.time()}) + # feat-chv-hybrid-b：合并写——常驻态登记 resume 后仍 + # 保持常驻态（不写回 legacy running），只续期；legacy + # 登记走旧语义（state=running）。tier 等字段不再被 + # 字面量覆盖丢掉。 + if record.get("state") in RESIDENT_ACTIVE_STATES: + self._write_resident_record( + tenant, {"last_active_at": _time.time()}, now=_time.time()) + else: + self._write_resident_record(tenant, { + "vm": vm, "tenant_id": tenant, + "state": "running", + "origin": record.get("origin") or "task", + "gitea_synced": bool(record.get("gitea_synced"))}, + now=_time.time()) self._emit("warm_pool_acquire", task, "vm=%s mode=resumed resume_seconds=%.2f" " warm_hit=%s" @@ -2358,16 +2456,681 @@ class CloudHypervisorExecutor(Executor): if tenant: clear_persistent_vm(self.state_dir, tenant) return False + def suspend_after_task(self, task_or_name: dict | str, + tenant_id: str | None = None) -> bool: + """任务间隙暂停：pause + 写持久登记（state=paused， + origin=task）。pause 失败或 VM 已死即清登记并返回 + False（调用方不得把死 VM 记成可用预热机）。 + feat-chv-hybrid-b：常驻租户的任务结束只暂停 VM、 + 只续期（last_active_at=now），不改租约状态—— + resident-active → resident-paused 只由租约计时器 + 在闲置满阈值时转（见 sweep_resident_leases），任务 + 结束不是降级信号；登记写改走 _write_resident_record + 合并写，tier/last_active_at 等字段不再被字面量覆盖丢掉。""" + vm = self.vm_name(task_or_name) + tenant = tenant_id or self._tenant_id( + task_or_name if isinstance(task_or_name, dict) + else {"id": task_or_name}) + try: + self.pause_vm(task_or_name) + except Exception as exc: + self._emit("vm_suspend_failed", + task_or_name if isinstance(task_or_name, dict) + else {"id": task_or_name}, + "vm=%s 暂停失败，清持久登记：%s" % (vm, exc), + "vm_suspend_failed") + if tenant: + clear_persistent_vm(self.state_dir, tenant) + return False if tenant: import time as _time + now = _time.time() record = read_persistent_vm(self.state_dir, tenant) or {} - write_persistent_vm(self.state_dir, tenant, { - "vm": vm, "tenant_id": tenant, - "state": "paused", "origin": "task", - "gitea_synced": bool(record.get("gitea_synced")), - "updated_at": _time.time()}) + if record.get("state") in RESIDENT_STATES: + self._write_resident_record( + tenant, {"last_active_at": now}, now=now) + else: + self._write_resident_record(tenant, { + "vm": vm, "tenant_id": tenant, + "state": "paused", "origin": "task", + "gitea_synced": bool(record.get("gitea_synced"))}, + now=now) return True + # -- 混合模式 B 轨（feat-chv-hybrid-b，DESIGN-HYBRID.md §9.1 + # 第 5–9 项）------------------------------------------------- + # 计时器驱动方式（实施时定，§4.2 二选一）：worker 心跳轮询。 + # executor 暴露 sweep_resident_leases() 做一轮扫描，worker + # 心跳 tick 调一次（接线点见汇报；本轨不改 worker.py，由 + # 协调员在集成时接一行调用）。不复用 P8 cron 语义：P8 是 + # 任务级定时调度（每项目每 tick 上限、指数退避重试），租约 + # 扫描是宿主级状态机轮询、无任务可调度，套 P8 只会把简单 + # 的扫描复杂化；心跳轮询与 _reconcile_warm_pool 的调用形态 + # 同款（worker 启动/心跳时调执行器方法）。 + # 审计：状态转换记 vm_resident_suspended / vm_resident_demoted / + # vm_resident_promoted、升级提示记 resident_upgrade_hint， + # 全部带 host_id + tenant_id，绝不带密钥值。 + + def _hybrid_env_float(self, name, default): + """读 B 轨阈值类环境变量（测试用短阈值注入）： + 非法/缺失/<=0 即回生产默认值，绝不因配错而停表。""" + try: + value = float(self._env.get(name) or default) + except (TypeError, ValueError): + return float(default) + return value if value > 0 else float(default) + + def _host_id(self): + """审计里的 host_id：CHV_HOST_ID 环境变量优先， + 未设即本机 hostname。""" + try: + host_id = self._env.get("CHV_HOST_ID") + except Exception: + host_id = None + return host_id or socket.gethostname() + + def _resident_thresholds(self): + """租约阈值（秒）：(挂起阈值, 降级阈值)。生产默认值 + 30 分钟 / 24 小时；测试经 CHV_RESIDENT_IDLE_SUSPEND_SECONDS / + CHV_RESIDENT_IDLE_DEMOTE_SECONDS 注入短阈值（构造时 + env 或事后改 self._env 均可，调用时现读）。""" + return ( + self._hybrid_env_float( + "CHV_RESIDENT_IDLE_SUSPEND_SECONDS", + RESIDENT_IDLE_SUSPEND_SECONDS), + self._hybrid_env_float( + "CHV_RESIDENT_IDLE_DEMOTE_SECONDS", + RESIDENT_IDLE_DEMOTE_SECONDS)) + + def resident_tier_eligible(self, tenant_id: str | None, + tier: str | None = None) -> bool: + """常驻资格（权益门，fail-closed，设计 §5）： + 只有 Paid/Heavy 有常驻资格；Free 无论多高频都没有； + 无 tier 行 = 无常驻资格（拒绝常驻，走池）。 + 不对称说明（有意为之，设计 §5）：P11 的既定语义是 + “无 tier 行的已有用户不受 tier 门约束（只受原有 USD + 配额约束）”——那是用量门。常驻是权益门，语义相反： + 无 tier 行 = 不授予新权益。两者不冲突，因为一个是 + “不限制已有用量”，一个是“不授予新权益”；实施时不许 + 把 P11 的豁免复用到常驻门上。 + tier 参数优先，否则读登记的 tier 字段；两者皆无即 + 不合格。""" + if tier is None: + record = read_persistent_vm(self.state_dir, tenant_id) \ + if tenant_id else None + tier = record.get("tier") if record else None + return tier in RESIDENT_TIERS + + def _write_resident_record(self, tenant, updates, now=None): + """合并写持久登记：保留旧字段，只更新给定字段 + + updated_at；新登记补 tenant_id/vm 默认值。阶段 1 的 + 三处字面量写登记（prewarm/acquire/suspend）至此都 + 经本方法，tier/state/last_active_at/lease_expires_at + 不再被字面量覆盖丢掉。""" + record = read_persistent_vm(self.state_dir, tenant) or {} + record.update(updates) + record.setdefault("tenant_id", tenant) + record.setdefault("vm", self.persistent_vm_name(tenant)) + record["updated_at"] = time.time() if now is None else now + write_persistent_vm(self.state_dir, tenant, record) + return record + + def touch_resident_activity(self, tenant_id: str | None, + now: float | None = None) -> bool: + """租约续期（设计 §4.1）：每次任务/消息/网关活动即续期 + （last_active_at=now）。仅当登记存在且为常驻有效态 + （resident-active/resident-paused/hibernated）时写； + 无登记不凭空造租约，demoted 不静默复活（复活走 + promote_to_resident）。lease_expires_at 是闲置规则 + 的派生值（last_active_at + 挂起阈值 + 降级阈值，即 + 再无活动时的降级点），不是日历到期日，续期即重算。 + 返回是否续期。""" + if not tenant_id: + return False + record = read_persistent_vm(self.state_dir, tenant_id) + if record is None \ + or record.get("state") not in RESIDENT_ACTIVE_STATES: + return False + now = time.time() if now is None else now + suspend, demote = self._resident_thresholds() + self._write_resident_record(tenant_id, { + "last_active_at": now, + "lease_expires_at": now + suspend + demote}, now=now) + return True + + def _resident_records(self): + """扫描 state_dir/persistent/*.json，返回常驻有效态 + 登记 [(tenant, record)]（demoted 已收回租约、不占槽， + 不在内；损坏/非 dict 的登记跳过）。""" + out = [] + pdir = os.path.join(self.state_dir, "persistent") + try: + entries = os.listdir(pdir) + except OSError: + return out + for entry in entries: + if not entry.endswith(".json"): + continue + try: + with open(os.path.join(pdir, entry), "r") as fh: + record = json.load(fh) + except (OSError, ValueError): + continue + if isinstance(record, dict) \ + and record.get("state") in RESIDENT_ACTIVE_STATES: + tenant = record.get("tenant_id") \ + or entry[:-len(".json")] + out.append((tenant, record)) + return out + + def _resident_count(self): + """当前常驻有效租约数（占槽数）。""" + return len(self._resident_records()) + + def resident_slot_cap(self) -> int: + """每宿主常驻上限（已定值，§4.1）：VM 槽位的 50% + （CHV_MAX_VMS × 50%，向下取整）。""" + return max(0, int(self.max_vms * RESIDENT_MAX_FRACTION)) + + def promote_to_resident(self, tenant_id: str, + tier: str | None = None) -> dict: + """新建/恢复常驻租约（§4.1/§4.4 执行层语义）： + - tier 校验 fail-closed：tier 须显式为 Paid/Heavy； + 缺失/不符即 ChvFailure tier_denied（平台层路由时 + 已拦一次，这里是执行层 preflight 第二道）。 + - 常驻上限：已常驻数达上限时不抢占已有常驻，返回 + {promoted: False, queued: True}（新常驻申请排队）。 + - 成功：登记 tier/state=resident-active/last_active_at/ + lease_expires_at，审计 vm_resident_promoted。 + 已是常驻有效态时幂等续期（不重复占槽）。 + 返回 {promoted, queued, vm, tier, state, reason?}。""" + if not tenant_id: + raise ValueError("tenant_id 不能为空") + if tier is None: + record = read_persistent_vm(self.state_dir, tenant_id) + tier = record.get("tier") if record else None + if tier not in RESIDENT_TIERS: + raise ChvFailure( + "tier_denied", + "常驻资格须为 Paid/Heavy 档位：租户 %s tier=%s，" + "无常驻资格（Free 无论多高频都没有；无 tier 行" + "即无资格，走池）" % (tenant_id, tier)) + now = time.time() + record = read_persistent_vm(self.state_dir, tenant_id) + if record is not None \ + and record.get("state") in RESIDENT_ACTIVE_STATES: + self.touch_resident_activity(tenant_id, now=now) + return {"promoted": True, "queued": False, + "vm": record.get("vm"), "tier": tier, + "state": record.get("state")} + if self._resident_count() >= self.resident_slot_cap(): + return {"promoted": False, "queued": True, + "reason": "resident_capacity_exceeded", + "tier": tier} + suspend, demote = self._resident_thresholds() + record = self._write_resident_record(tenant_id, { + "tier": tier, "state": "resident-active", + "last_active_at": now, + "lease_expires_at": now + suspend + demote}, now=now) + self._emit("vm_resident_promoted", + {"id": "persist-%s" % tenant_id, + "project_id": tenant_id}, + "host_id=%s tenant=%s vm=%s tier=%s" + % (self._host_id(), tenant_id, + record.get("vm"), tier)) + return {"promoted": True, "queued": False, + "vm": record.get("vm"), "tier": tier, + "state": "resident-active"} + + def _suspend_resident_vm(self, tenant, record): + """租约挂起的 VM 动作（sweep 的一步）：VM 已是 Paused + （任务间隙 suspend_after_task 已暂停）即直接转态， + 不调 pause；Running 即 pause_vm；状态未知且 socket + 已不在时视为 VM 已不在、无须 pause，直接转态；其余 + 未知时试调 pause_vm，失败即本租户中止（返回 False， + 租约状态不动）。""" + vm = record.get("vm") or self.persistent_vm_name(tenant) + state = self._vm_info_state({"id": vm}) + if state == "Paused": + return True + if state is None and not os.path.exists( + self.api_socket({"id": vm})): + return True + try: + self.pause_vm({"id": vm}) + return True + except Exception as exc: + self._emit("vm_suspend_failed", + {"id": vm, "project_id": tenant}, + "vm=%s 租约挂起 pause 失败：%s" % (vm, exc), + "vm_suspend_failed") + return False + + def sweep_resident_leases(self, now: float | None = None) -> dict: + """租约计时器一轮扫描（设计 §4.2，worker 心跳轮询驱动）： + - resident-active 且闲置 >= 挂起阈值 → pause_vm（已 + Paused 的直接转态）→ state=resident-paused，审计 + vm_resident_suspended（带 host_id + tenant_id）。 + - resident-paused 且闲置 >= 挂起阈值 + 降级阈值（即 + 累计 24.5 小时无活动）→ demote_resident，审计 + vm_resident_demoted。 + 单个租户失败（pause 失败/降级中止）不中断整轮，记入 + aborted。返回 {suspended, demoted, aborted}（租户 id + 列表）。""" + now = time.time() if now is None else now + suspend_s, demote_s = self._resident_thresholds() + out = {"suspended": [], "demoted": [], "aborted": []} + for tenant, record in self._resident_records(): + try: + last = record.get("last_active_at") \ + or record.get("updated_at") or now + idle = now - float(last) + state = record.get("state") + if state == "resident-active" and idle >= suspend_s: + if self._suspend_resident_vm(tenant, record): + self._write_resident_record( + tenant, {"state": "resident-paused"}, + now=now) + self._emit( + "vm_resident_suspended", + {"id": "persist-%s" % tenant, + "project_id": tenant}, + "host_id=%s tenant=%s vm=%s idle_seconds=%.0f" + % (self._host_id(), tenant, + record.get("vm"), idle)) + out["suspended"].append(tenant) + else: + out["aborted"].append(tenant) + elif state == "resident-paused" \ + and idle >= suspend_s + demote_s: + result = self.demote_resident(tenant, now=now) + if result.get("demoted"): + out["demoted"].append(tenant) + else: + out["aborted"].append(tenant) + except Exception as exc: + _LOGGER.error( + "chv_resident_sweep_failed tenant=%s error=%s", + tenant, exc) + out["aborted"].append(tenant) + return out + + def _sync_resident_before_demote(self, tenant_id, vm): + """降级前的同步步（设计 §4.3 第 1 步）： + 权威同步点在任务结束时（§3.3：任务完成即 Gitea push + + 记忆/技能落盘，之后 VM 才暂停），降级时盘上已是同步后 + 状态；因此默认本步直接通过。若部署配了 + resident_sync_hook（例如先 resume 再跑一次 chv-sync + push 的回调），则以钩子返回值为准：False 即同步失败， + 降级中止。demote_resident 的 sync_fn 参数优先于钩子。""" + hook = getattr(self, "resident_sync_hook", None) + if hook is None: + return True + return bool(hook(tenant_id, vm)) + + # 降级流程细则（设计 §4.3，注释放方法外、不计入 50 行上限）： + # 同步 → 快照 → 清 seed → 卸盘 → 销毁 → 登记 demoted。 + # - 同步失败即降级中止、VM 保留、记审计 + # （vm_resident_demote_aborted），绝不丢改动后降级； + # 快照失败同样中止。 + # - 清 seed：stop_vm 删 seed 目录 + 宿主私钥 + 已投递密钥 + # 记录；seed.iso 随 keep_disk=True 保留——它是快照 + # config.json 引用的 restore 输入（hibernate 的 V-7/D2 + # 同款），删了快照即不可恢复；“清 seed”指清 live seed， + # 不指毁快照输入。 + # - 卸盘/销毁：stop_vm(keep_disk=True) 关 VM；任务盘与 + # seed.iso 留作快照输入，数据盘 data.raw 是用户数据、 + # 永不随降级删（只走显式回收）。 + # - 登记：state=demoted，保留 last_active_at 与快照 URL + # （供下次唤醒时从 Gitea 重建或快照恢复），审计 + # vm_resident_demoted（带 host_id + tenant_id）。 + # - VM 已不在时跳过快照（无物可拍），只做登记。 + def demote_resident(self, tenant_id: str | None, + snapshot_url: str | None = None, + sync_fn: Callable[..., bool] | None = None, + now: float | None = None) -> dict: + """降级流程（设计 §4.3）：同步 → 快照 → 清 seed → + 卸盘 → 销毁 → 登记 demoted。同步/快照失败即中止、VM + 保留；仅 resident-paused/hibernated 可降级，其余返回 + {demoted: False, reason: not_demotable}。 + 返回 {demoted, vm, snapshot, reason?}。""" + now = time.time() if now is None else now + record = read_persistent_vm(self.state_dir, tenant_id) + if record is None or record.get("state") not in ( + "resident-paused", "hibernated"): + return {"demoted": False, "reason": "not_demotable", + "state": record.get("state") if record else None} + vm = record.get("vm") or self.persistent_vm_name(tenant_id) + task = {"id": vm, "project_id": tenant_id} + if not self._demote_sync_ok(task, tenant_id, vm, sync_fn): + return {"demoted": False, "reason": "sync_failed", + "vm": vm} + snapshot, failed = self._demote_snapshot( + task, vm, snapshot_url) + if failed: + return {"demoted": False, "reason": "snapshot_failed", + "vm": vm} + try: + self.stop_vm(task, keep_disk=True) + except Exception: + pass + updates = {"state": "demoted"} + if snapshot: + updates["snapshot_url"] = snapshot + self._write_resident_record(tenant_id, updates, now=now) + self._emit("vm_resident_demoted", task, + "host_id=%s tenant=%s vm=%s snapshot=%s" + % (self._host_id(), tenant_id, vm, snapshot)) + return {"demoted": True, "vm": vm, "snapshot": snapshot} + + def _demote_sync_ok(self, task, tenant_id, vm, sync_fn): + """降级同步步（demote_resident 的一步）：失败记 + vm_resident_demote_aborted 审计并返回 False（VM 保留， + 绝不丢改动后降级）；成功返回 True。""" + sync_callable = sync_fn if sync_fn is not None \ + else self._sync_resident_before_demote + try: + synced = bool(sync_callable(tenant_id, vm)) + except Exception: + synced = False + if not synced: + self._emit("vm_resident_demote_aborted", task, + "host_id=%s tenant=%s vm=%s 同步失败，降级中止、" + "VM 保留" % (self._host_id(), tenant_id, vm), + "resident_demote_sync_failed") + return synced + + def _demote_snapshot(self, task, vm, snapshot_url): + """降级快照步（demote_resident 的一步）：VM 活着即 + snapshot_vm（失败记中止审计）；VM 已不在时跳过快照 + （无物可拍）。返回 (snapshot_url_or_None, failed)。""" + if not (os.path.exists(self.api_socket({"id": vm})) + or self._vm_info_state({"id": vm}) is not None): + return None, False + if snapshot_url is None: + snapshot_url = "file://%s" % os.path.join( + self.state_dir, vm, "snapshot-demote") + try: + self.snapshot_vm(task, snapshot_url) + except Exception as exc: + self._emit("vm_resident_demote_aborted", task, + "host_id=%s tenant=%s vm=%s 快照失败，降级" + "中止、VM 保留：%s" + % (self._host_id(), task.get("project_id"), + vm, exc), + "resident_demote_snapshot_failed") + return None, True + return snapshot_url, False + + # -- 排队（设计 §3.4）-------------------------------------------- + def queue_depth(self) -> int: + """当前排队等待的任务数（L1 扩缩输入；等待期间计数， + 退出时减回）。""" + with self._queue_lock: + return self._queue_waiters + + def _queue_attempts(self): + """排队轮次（次数制，与 wait_api_ready 同款口径：fake + sleeper 不推进墙钟时墙钟循环会空转，次数循环在单测里 + 是确定性的；真实 sleeper 下总时长约等于超时）。 + 超时经 CHV_QUEUE_TIMEOUT_SECONDS 注入（默认 300 秒）， + 轮询间隔经 CHV_QUEUE_POLL_SECONDS（默认 5 秒）。""" + timeout = self._hybrid_env_float( + "CHV_QUEUE_TIMEOUT_SECONDS", QUEUE_TIMEOUT_SECONDS) + poll = self._hybrid_env_float( + "CHV_QUEUE_POLL_SECONDS", QUEUE_POLL_SECONDS) + return max(1, int(timeout / poll)), poll + + def _wait_for_capacity(self, task, exclude=None): + """排队等宿主容量（设计 §3.4）：容量有空即直接过； + 满时按到达顺序等（本方法是调用方线程内顺序等待，无 + 优先队列、不插队）、不抢占（只等同一容量门放行，不踢 + 别的 VM）、不升/降档位（只等槽位，不改路由）。 + 超时（默认 300 秒）后 ChvFailure host_capacity_exceeded + 明确拒绝，不无限积压。等待期间 queue_depth() +1。""" + try: + self._check_capacity(exclude=exclude) + return + except ChvFailure: + pass + attempts, poll = self._queue_attempts() + with self._queue_lock: + self._queue_waiters += 1 + try: + for _ in range(attempts): + self.sleeper(poll) + try: + self._check_capacity(exclude=exclude) + return + except ChvFailure: + continue + raise ChvFailure( + "host_capacity_exceeded", + "排队超时：%d 秒内无可用宿主（容量仍满），明确拒绝，" + "不无限积压" % int(attempts * poll)) + finally: + with self._queue_lock: + self._queue_waiters -= 1 + + def _wait_for_resident_slot(self, tenant_id): + """常驻租户的任务排队时仍等常驻槽位（设计 §3.4）： + 只等常驻计数 < 上限，不挪去池里跑（挪了等于改隔离 + 与计费口径），不抢占已有常驻。超时同样 + host_capacity_exceeded。""" + if self._resident_count() < self.resident_slot_cap(): + return + attempts, poll = self._queue_attempts() + with self._queue_lock: + self._queue_waiters += 1 + try: + for _ in range(attempts): + self.sleeper(poll) + if self._resident_count() < self.resident_slot_cap(): + return + raise ChvFailure( + "host_capacity_exceeded", + "排队超时：常驻槽位仍满（%d/%d），明确拒绝" + % (self._resident_count(), + self.resident_slot_cap())) + finally: + with self._queue_lock: + self._queue_waiters -= 1 + + # -- 升级提示点（设计 §5）---------------------------------------- + def _pool_task_file(self, tenant_id): + """池任务时间戳文件：state_dir/pool-tasks/.json + （执行层任务登记，不是平台层账本；只存时间戳与上次 + 提示时间，不存任务内容、不存密钥）。""" + safe = re.sub(r"[^A-Za-z0-9_.-]", "-", + str(tenant_id or "default")) + return os.path.join(self.state_dir, "pool-tasks", + safe + ".json") + + def _load_pool_tasks(self, tenant_id): + """读池任务登记：{tasks: [ts...], last_hint_at: ts|None}； + 无文件/损坏即空登记（不猜）。""" + try: + with open(self._pool_task_file(tenant_id), "r") as fh: + data = json.load(fh) + except (OSError, ValueError): + return {"tasks": [], "last_hint_at": None} + if not isinstance(data, dict): + return {"tasks": [], "last_hint_at": None} + tasks = data.get("tasks") or [] + if not isinstance(tasks, list): + tasks = [] + return {"tasks": [t for t in tasks + if isinstance(t, (int, float))], + "last_hint_at": data.get("last_hint_at")} + + def _save_pool_tasks(self, tenant_id, data): + """写池任务登记（atomic 写，与持久登记同款）。""" + path = self._pool_task_file(tenant_id) + os.makedirs(os.path.dirname(path), exist_ok=True) + atomic_write_text(path, json.dumps(data)) + + def _has_7day_streak(self, tasks, now): + """连续 7 天每天 ≥3 次（UTC 日桶；含今天往前 7 天）。""" + today = int(now // 86400) + days = {} + for ts in tasks: + day = int(ts // 86400) + days[day] = days.get(day, 0) + 1 + for offset in range(UPGRADE_HINT_STREAK_DAYS): + if days.get(today - offset, 0) < UPGRADE_HINT_MIN_PER_DAY: + return False + return True + + def _upgrade_hint_metric(self, tasks, now): + """本次提示的频率度量（进审计 detail，供平台层消费）。""" + if len(tasks) >= UPGRADE_HINT_MIN_TASKS: + return "30d_tasks=%d" % len(tasks) + return "7d_streak_%d_per_day" % UPGRADE_HINT_MIN_PER_DAY + + def _upgrade_hint_due(self, tasks, last_hint_at, now): + """是否该记升级提示：先去重（一周内同一租户最多一次）， + 再判高频定义（30 天 ≥100 次，或连续 7 天每天 ≥3 次）。""" + if last_hint_at is not None \ + and now - float(last_hint_at) \ + < UPGRADE_HINT_DEDUP_DAYS * 86400: + return False + if len(tasks) >= UPGRADE_HINT_MIN_TASKS: + return True + return self._has_7day_streak(tasks, now) + + def record_pool_task(self, tenant_id: str | None, + tier: str | None = None, + now: float | None = None) -> dict: + """记一次池任务并在达高频定义时发升级提示点（设计 §5）： + 只对 Free 租户记（Paid/Heavy 已有常驻资格，无须提示； + tier 未知时按 Free 记——提示只是事件，平台层消费时 + 再核档位）；频率只用于产生提示点，绝不用于改路由 + （本方法不返回路由决策、不改任何登记的状态字段）。 + 频率度量从本文件里的任务时间戳算，不另建账本。 + 返回 {recorded, hinted, metric}。""" + now = time.time() if now is None else now + if tier is None: + record = read_persistent_vm(self.state_dir, tenant_id) \ + if tenant_id else None + tier = record.get("tier") if record else None + if tier is not None and tier != "Free": + return {"recorded": False, "hinted": False, + "reason": "not_free_tier"} + data = self._load_pool_tasks(tenant_id) + cutoff = now - UPGRADE_HINT_WINDOW_DAYS * 86400 + tasks = [t for t in data["tasks"] if t >= cutoff] + tasks.append(now) + hinted = False + metric = None + if self._upgrade_hint_due(tasks, data.get("last_hint_at"), + now): + metric = self._upgrade_hint_metric(tasks, now) + data["last_hint_at"] = now + self._emit( + "resident_upgrade_hint", + {"id": "pool-%s" % tenant_id, + "project_id": tenant_id}, + "host_id=%s tenant=%s tier=Free metric=%s " + "window_days=%d" + % (self._host_id(), tenant_id, metric, + UPGRADE_HINT_WINDOW_DAYS)) + hinted = True + data["tasks"] = tasks + self._save_pool_tasks(tenant_id, data) + return {"recorded": True, "hinted": hinted, "metric": metric} + + # -- 网关分流（设计 §6）------------------------------------------ + def gateway_allowed_for_task(self, task: dict, + model_provider: str | None = None) -> bool: + """网关分流（设计 §6；worker 侧守卫自 P1-1 起已放行 CHV， + 分流 enforcement 落在本执行器——改之前网关不上 CHV 的 + 前置依赖即由此落地）： + - 非网关任务（model_provider 不是网关 provider）→ + True（不拦；本方法不作常驻判定用）。 + - 网关任务：常驻 VM 可接（登记为常驻有效态且 tier + 校验通过）、池 VM 不接。常驻有效态含 hibernated + （可 restore + resume 续跑）；demoted、无登记、 + legacy 态一律按池算。""" + provider = model_provider + if provider is None and isinstance(task, dict): + provider = task.get("_gateway_provider") + if provider not in GATEWAY_PROVIDERS: + return True + tenant = self._tenant_id(task) if isinstance(task, dict) \ + else None + if not tenant: + return False + record = read_persistent_vm(self.state_dir, tenant) + if record is None \ + or record.get("state") not in RESIDENT_ACTIVE_STATES: + return False + return self.resident_tier_eligible(tenant, + record.get("tier")) + + def _check_gateway_vm_class(self, task, model_provider=None): + """网关任务的 VM 类门：池 VM 以 gateway_pool_denied + 明确拒绝（ChvFailure，worker 现有 except 路径照常 + 写回 failed）；常驻 VM 放行。""" + if not self.gateway_allowed_for_task(task, model_provider): + tenant = self._tenant_id(task) \ + if isinstance(task, dict) else None + raise ChvFailure( + "gateway_pool_denied", + "网关任务只接常驻 VM：租户 %s 当前走池（无常驻" + "租约或档位无常驻资格），拒绝上 CHV；池 VM 用完" + "即销毁，长连接在上面活不过一个任务周期" + % tenant) + + def _hybrid_pool_preamble(self, task, model_provider=None): + """B 轨池路径前置（_run_with_vm 的开头，拆出只为满足 + 50 行函数上限）：任务到达即活动（续期租约）、网关 VM + 类门（常驻可接、池不接）、排队等容量（超时 + host_capacity_exceeded）、池任务记一次（Free 高频 + 提示点原料）。续期与记数失败不挡任务；网关门与排队 + 拒绝直接抛。""" + try: + _tenant = self._tenant_id(task) + if _tenant: + self.touch_resident_activity(_tenant) + except Exception: + pass + self._check_gateway_vm_class(task, model_provider) + self._wait_for_capacity(task, exclude=self.vm_name(task)) + try: + _tenant = self._tenant_id(task) + if _tenant: + self.record_pool_task( + _tenant, + task.get("tier") if isinstance(task, dict) else None) + except Exception: + pass + + def _hybrid_persistent_preamble(self, task, model_provider=None): + """B 轨持久路径前置（_run_with_persistent_vm 的开头， + 拆出只为满足 50 行函数上限）：任务到达即续期租约 + （仅常驻登记写，失败不挡任务）；网关任务先过 VM 类门 + （常驻可接、池不接）。""" + try: + _tenant = self._tenant_id(task) + if _tenant: + self.touch_resident_activity(_tenant) + except Exception: + pass + self._check_gateway_vm_class(task, model_provider) + + def _wait_for_persistent_slot(self, task, tenant, vm): + """持久路径冷启需新占槽位时先排队（_run_with_persistent_vm + 的一段，拆出只为满足 50 行函数上限）：常驻租户等常驻 + 槽位（不挪去池里跑），其余等通用宿主容量；超时 + host_capacity_exceeded 明确拒绝。""" + _record = read_persistent_vm(self.state_dir, tenant) \ + if tenant else None + if _record is not None and _record.get("state") \ + in RESIDENT_ACTIVE_STATES: + self._wait_for_resident_slot(tenant) + else: + self._wait_for_capacity(task, exclude=vm) + def _run_guest_on_vm(self, task, vm, inner_args, goal, extra_env=None, **kwargs): """在指定 VM 上跑客内命令并做中途死亡分类（ @@ -2412,11 +3175,14 @@ class CloudHypervisorExecutor(Executor): _run_guest_on_vm（登记进程已死即 interrupted）。""" tenant = self._tenant_id(task) vm_task = self._persistent_task(tenant) + self._hybrid_persistent_preamble( + task, kwargs.get("model_provider")) acquired = self.acquire_for_task(task, extra_env=extra_env) vm = acquired.get("vm") or self.vm_name(vm_task) mode = acquired["mode"] resume_seconds = acquired.get("resume_seconds") if mode == "cold": + self._wait_for_persistent_slot(vm_task, tenant, vm) self.wake_vm(vm_task, extra_env=extra_env) try: result = self._run_guest_on_vm( @@ -3372,35 +4138,48 @@ class CloudHypervisorExecutor(Executor): parse_goal_result），这里不再二次改写 goal 结论。 feat-chv-secfix（H2）：extra_env 透传给 wake_vm（密钥 子集在 provisioned 模式下随 seed 投递），再透传给 - _run_guest（远程命令 source 客内密钥文件）。""" + _run_guest（远程命令 source 客内密钥文件）。 + feat-chv-hybrid-b：wake 之前先走 _hybrid_pool_preamble + （续期 + 网关门 + 排队 + 池任务记数）。""" + self._hybrid_pool_preamble(task, kwargs.get("model_provider")) self.wake_vm(task, extra_env=extra_env) try: - try: - if goal: - return self._run_guest_goal( - task, inner_args, extra_env=extra_env, **kwargs) - result = self._run_guest(task, inner_args, - extra_env=extra_env, **kwargs) - except Exception as exc: - if self._vm_proc_dead(task): - return self._vm_death_result( - task, goal, str(exc)[:500]) - raise - if not goal and result.get("status") == "failed" \ - and self._vm_proc_dead(task): - result = dict(result) - result["status"] = "interrupted" - result["vm_dead"] = True - result["interrupted_reason"] = ( - "VM 中途死亡，任务未完成、断点可续跑：" - "客内命令已 failed 且登记的 VMM 进程已死") - return result + return self._run_guest_with_death_classification( + task, inner_args, goal, extra_env, kwargs) finally: try: self.shutdown_vm(task) except Exception: pass + def _run_guest_with_death_classification(self, task, inner_args, + goal, extra_env, kwargs): + """客内执行 + 中途死亡分类（_run_with_vm 的一段，拆出 + 只为满足 50 行函数上限）：客内命令抛传输异常、或非 + goal 结果已 failed 时，若登记的 VMM 进程已死，判 + interrupted（断点可续跑）；VM 还活着时的异常/失败 + 原样处理，不误判。""" + try: + if goal: + return self._run_guest_goal( + task, inner_args, extra_env=extra_env, **kwargs) + result = self._run_guest(task, inner_args, + extra_env=extra_env, **kwargs) + except Exception as exc: + if self._vm_proc_dead(task): + return self._vm_death_result( + task, goal, str(exc)[:500]) + raise + if not goal and result.get("status") == "failed" \ + and self._vm_proc_dead(task): + result = dict(result) + result["status"] = "interrupted" + result["vm_dead"] = True + result["interrupted_reason"] = ( + "VM 中途死亡，任务未完成、断点可续跑：" + "客内命令已 failed 且登记的 VMM 进程已死") + return result + def execute(self, task: dict) -> dict: """只读执行：客内 host.py status，不直接跑模型回合 （与 Local/Docker 的 execute 同款语义）。 @@ -3527,8 +4306,10 @@ class CloudHypervisorExecutor(Executor): worker 层已接线（补缺口轨）：非网关的 CHV 断点任务 经 worker 的通用断点分支调到这里，model_provider 按名透传给客内 host.py（与 Local/Docker 同款口径， - 首轮 thread 的 provider 必须一致）；网关任务在 worker - 侧仍被 loud fail 守卫拦住，不到这里。模型 key + 首轮 thread 的 provider 必须一致）；网关任务的分流 + 由 _run_with_vm 开头的 _check_gateway_vm_class 按 VM + 类 enforcement（常驻 VM 可接、池 VM 以 + gateway_pool_denied 拒绝）。模型 key 本身绝不经本方法传值，只按环境变量名注入。""" if not isinstance(task, dict) or task.get("id") is None: raise ValueError("任务缺少 id") diff --git a/cloud/tests/test_chv_hybrid_b.py b/cloud/tests/test_chv_hybrid_b.py new file mode 100644 index 0000000..5f92d6f --- /dev/null +++ b/cloud/tests/test_chv_hybrid_b.py @@ -0,0 +1,710 @@ +"""CHV 混合模式 B 轨（feat-chv-hybrid-b）测试。 + +只用标准库与 mock runner/launcher/api_client，绝不启动真实 +VM（与 test_chv_opt_stage1.py 同款边界）。覆盖（设计 +DESIGN-HYBRID.md §9.1 第 5–9 项）： +- 登记字段扩展 + tier 校验（含无 tier 行拒绝常驻） +- 租约挂起/降级计时（环境变量注入短阈值测状态机转换， + 含降级同步失败中止） +- 排队超时拒绝（host_capacity_exceeded，不插队、不抢占、 + 不升降档） +- 升级提示点去重（一周内同一租户最多一次） +- 网关分流（常驻可接、池不接） +""" + +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, + write_persistent_vm, + write_vm_tenant, + ) +except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.chv import ChvFailure + from cloud.chv import CloudHypervisorExecutor + from cloud.chv.disk import ( + read_persistent_vm, + write_persistent_vm, + write_vm_tenant, + ) + + +class FakeResult: + def __init__(self, returncode=0, stdout="", stderr=""): + self.returncode = returncode + self.stdout = stdout + self.stderr = stderr + + +class FakeProc: + def __init__(self, alive=True): + self._alive = alive + + def poll(self): + return None if self._alive else 1 + + def terminate(self): + self._alive = False + + def wait(self, timeout=None): + return 0 + + def kill(self): + self._alive = False + + +def _write(path, content=b"x"): + with open(path, "wb") as fh: + fh.write(content) + os.chmod(path, 0o755) + return path + + +class HybridBCase(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.sleeps = [] + self.events = [] + + def tearDown(self): + self.tmp.cleanup() + + def make_exec(self, runner=None, api_client=None, launcher=None, + env_extra=None, sleeper=None, **overrides): + env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888", + "CHV_HOST_KEY_MODE": "keyscan", + "CHV_SSH_FORWARD": "1", + "CHV_HOST_ID": "host-test-1"} + if env_extra: + env.update(env_extra) + if runner is None: + runner = self._default_runner + if api_client is None: + api_client = self._default_api + if launcher is None: + launcher = self._default_launcher + kwargs = dict( + binary=self.binary, ch_remote=self.ch_remote, + kernel=self.kernel, rootfs=self.rootfs, + ssh_key=self.ssh_key, kvm_path=self.kvm, + cpuinfo_path=self.cpuinfo, state_dir=self.state_dir, + runner=runner, launcher=launcher, api_client=api_client, + sleeper=sleeper if sleeper is not None + else (lambda s: self.sleeps.append(s)), + env=env) + kwargs.update(overrides) + ex = CloudHypervisorExecutor(**kwargs) + ex.audit_callback = lambda event, task, detail, kind=None: \ + self.events.append((event, detail, kind)) + 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, "", "") + if cmd and os.path.basename(str(cmd[0])) == "ch-remote": + if "snapshot" in cmd: + url = cmd[cmd.index("snapshot") + 1] + if url.startswith("file://"): + snap_dir = url[len("file://"):] + os.makedirs(snap_dir, exist_ok=True) + with open(os.path.join(snap_dir, "state.json"), + "w") as fh: + fh.write("{}") + return FakeResult(0, "", "") + return FakeResult(0, "", "") + return FakeResult(0, "", "") + + def _default_api(self, socket_path, method, path, payload=None): + self.api_calls.append((method, path, payload)) + if path == "/vm.info": + state = self.vm_states.get( + os.path.basename(socket_path)[:-len(".sock")]) + if state: + return (200, {"state": state}) + return (200, {}) + return (200, {}) + + def _default_launcher(self, cmd): + self.launches.append(list(cmd)) + return FakeProc(alive=True) + + def _task(self, task_id="abc123", project="proj1"): + return {"id": task_id, "prompt": "做一个加法函数", + "project_id": project} + + def _event_names(self): + return [e[0] for e in self.events] + + def _write_resident(self, ex, tenant, state, tier, last_active, + vm=None): + vm = vm or ex.persistent_vm_name(tenant) + write_persistent_vm(ex.state_dir, tenant, { + "vm": vm, "tenant_id": tenant, "tier": tier, + "state": state, "last_active_at": last_active, + "lease_expires_at": last_active + 1800 + 86400, + "origin": "task", "gitea_synced": True, + "updated_at": last_active}) + return vm + + def _live_vm(self, ex, vm): + """给一台 VM 造“活着”的外形：socket 文件 + vm.info。""" + sock = os.path.join(ex.state_dir, vm + ".sock") + os.makedirs(ex.state_dir, exist_ok=True) + with open(sock, "w") as fh: + fh.write("") + return sock + + +# -- 登记字段扩展 + tier 校验（第 5 项）------------------------------- +class ResidentRegistryTest(HybridBCase): + def test_promote_writes_extended_fields(self): + ex = self.make_exec() + out = ex.promote_to_resident("t1", "Paid") + self.assertTrue(out["promoted"]) + self.assertFalse(out["queued"]) + record = read_persistent_vm(ex.state_dir, "t1") + self.assertEqual(record["tier"], "Paid") + self.assertEqual(record["state"], "resident-active") + self.assertIn("last_active_at", record) + self.assertIn("lease_expires_at", record) + self.assertGreater(record["lease_expires_at"], + record["last_active_at"]) + self.assertIn("vm_resident_promoted", self._event_names()) + detail = [d for e, d, _k in self.events + if e == "vm_resident_promoted"][0] + self.assertIn("host_id=host-test-1", detail) + self.assertIn("tenant=t1", detail) + + def test_promote_free_tier_denied(self): + ex = self.make_exec() + with self.assertRaises(ChvFailure) as ctx: + ex.promote_to_resident("tfree", "Free") + self.assertEqual(ctx.exception.kind, "tier_denied") + self.assertIsNone(read_persistent_vm(ex.state_dir, "tfree")) + + def test_promote_no_tier_row_denied(self): + # 无 tier 行 = 无常驻资格（fail-closed，走池）：登记里 + # 没有 tier 字段、也不传 tier 参数时同样拒绝。 + ex = self.make_exec() + with self.assertRaises(ChvFailure) as ctx: + ex.promote_to_resident("tnope") + self.assertEqual(ctx.exception.kind, "tier_denied") + + def test_tier_eligible_matrix(self): + ex = self.make_exec() + self.assertTrue(ex.resident_tier_eligible("x", "Paid")) + self.assertTrue(ex.resident_tier_eligible("x", "Heavy")) + self.assertFalse(ex.resident_tier_eligible("x", "Free")) + self.assertFalse(ex.resident_tier_eligible("x", None)) + self.assertFalse(ex.resident_tier_eligible("nobody")) + + def test_promote_over_cap_queued_not_preempt(self): + # max_vms 默认 4 → 常驻上限 2；第 3 个排队，不抢占前两个。 + ex = self.make_exec() + self.assertTrue(ex.promote_to_resident("t1", "Paid")["promoted"]) + self.assertTrue(ex.promote_to_resident("t2", "Heavy")["promoted"]) + out = ex.promote_to_resident("t3", "Paid") + self.assertFalse(out["promoted"]) + self.assertTrue(out["queued"]) + self.assertEqual(out["reason"], "resident_capacity_exceeded") + self.assertEqual(ex._resident_count(), 2) + self.assertEqual( + read_persistent_vm(ex.state_dir, "t1")["state"], + "resident-active") + self.assertEqual( + read_persistent_vm(ex.state_dir, "t2")["state"], + "resident-active") + self.assertIsNone(read_persistent_vm(ex.state_dir, "t3")) + + def test_promote_idempotent_no_double_slot(self): + ex = self.make_exec() + ex.promote_to_resident("t1", "Paid") + out = ex.promote_to_resident("t1", "Paid") + self.assertTrue(out["promoted"]) + self.assertEqual(ex._resident_count(), 1) + + def test_touch_renews_only_resident(self): + ex = self.make_exec(env_extra={ + "CHV_RESIDENT_IDLE_SUSPEND_SECONDS": "60", + "CHV_RESIDENT_IDLE_DEMOTE_SECONDS": "120"}) + now = 1800000000.0 + # legacy paused 登记：touch 不写、不造租约字段。 + write_persistent_vm(ex.state_dir, "tleg", { + "vm": ex.persistent_vm_name("tleg"), "tenant_id": "tleg", + "state": "paused", "origin": "task", + "gitea_synced": True, "updated_at": now - 100}) + self.assertFalse(ex.touch_resident_activity("tleg", now=now)) + self.assertNotIn("last_active_at", + read_persistent_vm(ex.state_dir, "tleg")) + # 常驻登记：touch 续期并重算 lease_expires_at。 + vm = self._write_resident(ex, "t2", "resident-active", + "Paid", now - 100) + self.assertTrue(ex.touch_resident_activity("t2", now=now)) + record = read_persistent_vm(ex.state_dir, "t2") + self.assertEqual(record["last_active_at"], now) + self.assertEqual(record["lease_expires_at"], now + 60 + 120) + self.assertEqual(record["state"], "resident-active") + self.assertEqual(record["tier"], "Paid") + + def test_suspend_after_task_keeps_resident_state_and_tier(self): + ex = self.make_exec() + ex.promote_to_resident("t1", "Paid") + vm = ex.persistent_vm_name("t1") + self.assertTrue(ex.suspend_after_task({"id": vm}, + tenant_id="t1")) + record = read_persistent_vm(ex.state_dir, "t1") + self.assertEqual(record["state"], "resident-active") + self.assertEqual(record["tier"], "Paid") + self.assertIn("last_active_at", record) + + def test_acquire_resident_paused_paid_tier(self): + ex = self.make_exec(env_extra={"CHV_WARM_POOL_SIZE": "1"}) + now = 1800000000.0 + vm = self._write_resident(ex, "t1", "resident-paused", + "Paid", now - 10) + write_vm_tenant(ex.state_dir, vm, "t1") + self._live_vm(ex, vm) + self.vm_states[vm] = "Paused" + out = ex.acquire_for_task(self._task(project="t1")) + self.assertEqual(out["mode"], "resumed") + self.assertEqual(out["vm"], vm) + record = read_persistent_vm(ex.state_dir, "t1") + # resume 后仍是常驻态（不写回 legacy running），tier 保留。 + self.assertEqual(record["state"], "resident-paused") + self.assertEqual(record["tier"], "Paid") + + def test_acquire_resident_paused_free_tier_denied(self): + ex = self.make_exec(env_extra={"CHV_WARM_POOL_SIZE": "1"}) + now = 1800000000.0 + vm = self._write_resident(ex, "t1", "resident-paused", + "Free", now - 10) + write_vm_tenant(ex.state_dir, vm, "t1") + self._live_vm(ex, vm) + self.vm_states[vm] = "Paused" + out = ex.acquire_for_task(self._task(project="t1")) + self.assertEqual(out["mode"], "cold") + self.assertIn("resident_tier_denied", self._event_names()) + record = read_persistent_vm(ex.state_dir, "t1") + self.assertEqual(record["state"], "resident-paused") + self.assertEqual(record["tier"], "Free") + + def test_acquire_legacy_paused_without_tier_still_works(self): + # legacy paused 登记无 tier 字段：走旧语义，仍可认领。 + ex = self.make_exec(env_extra={"CHV_WARM_POOL_SIZE": "1"}) + vm = ex.persistent_vm_name("tleg") + write_persistent_vm(ex.state_dir, "tleg", { + "vm": vm, "tenant_id": "tleg", "state": "paused", + "origin": "prewarm", "gitea_synced": True, + "updated_at": 1800000000.0}) + write_vm_tenant(ex.state_dir, vm, "tleg") + self._live_vm(ex, vm) + self.vm_states[vm] = "Paused" + out = ex.acquire_for_task(self._task(project="tleg")) + self.assertEqual(out["mode"], "resumed") + + +# -- 租约挂起/降级计时（第 6 项）-------------------------------------- +class LeaseTimerTest(HybridBCase): + def _lease_exec(self): + return self.make_exec(env_extra={ + "CHV_RESIDENT_IDLE_SUSPEND_SECONDS": "60", + "CHV_RESIDENT_IDLE_DEMOTE_SECONDS": "120"}) + + def test_sweep_suspends_idle_active_already_paused(self): + ex = self._lease_exec() + now = 1800000000.0 + vm = self._write_resident(ex, "t1", "resident-active", + "Paid", now - 61) + self._live_vm(ex, vm) + self.vm_states[vm] = "Paused" + out = ex.sweep_resident_leases(now=now) + self.assertEqual(out["suspended"], ["t1"]) + self.assertEqual(out["demoted"], []) + record = read_persistent_vm(ex.state_dir, "t1") + self.assertEqual(record["state"], "resident-paused") + self.assertEqual(record["tier"], "Paid") + self.assertIn("vm_resident_suspended", self._event_names()) + detail = [d for e, d, _k in self.events + if e == "vm_resident_suspended"][0] + self.assertIn("host_id=host-test-1", detail) + self.assertIn("tenant=t1", detail) + # VM 本已 Paused：sweep 不应再发 pause。 + self.assertFalse(any(c[-1] == "pause" for c in self.calls + if c and "ch-remote" in c[0])) + + def test_sweep_suspends_running_vm_calls_pause(self): + ex = self._lease_exec() + now = 1800000000.0 + vm = self._write_resident(ex, "t1", "resident-active", + "Paid", now - 61) + self._live_vm(ex, vm) + self.vm_states[vm] = "Running" + out = ex.sweep_resident_leases(now=now) + self.assertEqual(out["suspended"], ["t1"]) + self.assertTrue(any(c[-1] == "pause" for c in self.calls + if c and "ch-remote" in c[0])) + self.assertEqual( + read_persistent_vm(ex.state_dir, "t1")["state"], + "resident-paused") + + def test_sweep_no_suspend_before_threshold(self): + ex = self._lease_exec() + now = 1800000000.0 + self._write_resident(ex, "t1", "resident-active", + "Paid", now - 30) + out = ex.sweep_resident_leases(now=now) + self.assertEqual(out, {"suspended": [], "demoted": [], + "aborted": []}) + self.assertEqual( + read_persistent_vm(ex.state_dir, "t1")["state"], + "resident-active") + + def test_sweep_demotes_after_24h(self): + ex = self._lease_exec() + now = 1800000000.0 + vm = self._write_resident(ex, "t1", "resident-paused", + "Paid", now - 61 - 121) + self._live_vm(ex, vm) + self.vm_states[vm] = "Paused" + # 用户数据盘与任务盘：降级后须保留（快照输入 + 用户数据）。 + vm_dir = os.path.join(ex.state_dir, vm) + os.makedirs(vm_dir, exist_ok=True) + with open(os.path.join(vm_dir, "data.raw"), "wb") as fh: + fh.write(b"D" * 10) + with open(os.path.join(vm_dir, "disk.raw"), "wb") as fh: + fh.write(b"K" * 10) + out = ex.sweep_resident_leases(now=now) + self.assertEqual(out["demoted"], ["t1"]) + record = read_persistent_vm(ex.state_dir, "t1") + self.assertEqual(record["state"], "demoted") + self.assertEqual(record["last_active_at"], now - 61 - 121) + self.assertIn("snapshot_url", record) + self.assertIn("vm_resident_demoted", self._event_names()) + detail = [d for e, d, _k in self.events + if e == "vm_resident_demoted"][0] + self.assertIn("host_id=host-test-1", detail) + self.assertIn("tenant=t1", detail) + self.assertTrue(os.path.exists( + os.path.join(vm_dir, "data.raw"))) + self.assertTrue(os.path.exists( + os.path.join(vm_dir, "disk.raw"))) + # 槽位已释放：可再 promote 别的租户。 + self.assertEqual(ex._resident_count(), 0) + + def test_demote_sync_failure_aborts(self): + ex = self._lease_exec() + now = 1800000000.0 + vm = self._write_resident(ex, "t1", "resident-paused", + "Paid", now - 61 - 121) + self._live_vm(ex, vm) + self.vm_states[vm] = "Paused" + out = ex.demote_resident( + "t1", sync_fn=lambda tenant, v: False, now=now) + self.assertFalse(out["demoted"]) + self.assertEqual(out["reason"], "sync_failed") + record = read_persistent_vm(ex.state_dir, "t1") + self.assertEqual(record["state"], "resident-paused") + self.assertNotIn("snapshot_url", record) + # VM 保留：socket 还在，未销毁。 + self.assertTrue(os.path.exists( + os.path.join(ex.state_dir, vm + ".sock"))) + self.assertIn("vm_resident_demote_aborted", + self._event_names()) + + def test_demote_snapshot_failure_aborts(self): + ex = self._lease_exec() + now = 1800000000.0 + vm = self._write_resident(ex, "t1", "resident-paused", + "Paid", now - 61 - 121) + self._live_vm(ex, vm) + self.vm_states[vm] = "Paused" + with mock.patch.object( + ex, "snapshot_vm", + side_effect=ChvFailure("snapshot_failed", "boom")): + out = ex.demote_resident("t1", now=now) + self.assertFalse(out["demoted"]) + self.assertEqual(out["reason"], "snapshot_failed") + self.assertEqual( + read_persistent_vm(ex.state_dir, "t1")["state"], + "resident-paused") + + def test_demote_not_demotable_states(self): + ex = self._lease_exec() + now = 1800000000.0 + self._write_resident(ex, "t1", "resident-active", + "Paid", now - 100000) + out = ex.demote_resident("t1", now=now) + self.assertFalse(out["demoted"]) + self.assertEqual(out["reason"], "not_demotable") + out = ex.demote_resident("ghost", now=now) + self.assertFalse(out["demoted"]) + self.assertEqual(out["reason"], "not_demotable") + + def test_sweep_demote_aborted_on_sync_hook_failure(self): + ex = self._lease_exec() + ex.resident_sync_hook = lambda tenant, vm: False + now = 1800000000.0 + vm = self._write_resident(ex, "t1", "resident-paused", + "Paid", now - 61 - 121) + self._live_vm(ex, vm) + self.vm_states[vm] = "Paused" + out = ex.sweep_resident_leases(now=now) + self.assertEqual(out["demoted"], []) + self.assertEqual(out["aborted"], ["t1"]) + self.assertEqual( + read_persistent_vm(ex.state_dir, "t1")["state"], + "resident-paused") + + +# -- 排队超时拒绝（第 7 项）-------------------------------------------- +class QueueTest(HybridBCase): + def _queue_exec(self): + return self.make_exec(env_extra={ + "CHV_MAX_VMS": "1", + "CHV_QUEUE_TIMEOUT_SECONDS": "10", + "CHV_QUEUE_POLL_SECONDS": "2"}) + + def test_wait_for_capacity_timeout_rejects(self): + ex = self._queue_exec() + occupant = FakeProc(alive=True) + ex._vms["chv-other"] = occupant + with self.assertRaises(ChvFailure) as ctx: + ex._wait_for_capacity(self._task(), exclude="chv-abc123") + self.assertEqual(ctx.exception.kind, "host_capacity_exceeded") + # 轮询 10/2 = 5 次 sleeper 调用后才拒绝。 + self.assertEqual(len(self.sleeps), 5) + # 不抢占：占位 VM 仍活着；队列计数已归零。 + self.assertIsNone(occupant.poll()) + self.assertEqual(ex.queue_depth(), 0) + + def test_wait_for_capacity_free_passes_immediately(self): + ex = self._queue_exec() + ex._wait_for_capacity(self._task(), exclude="chv-abc123") + self.assertEqual(self.sleeps, []) + + def test_run_with_vm_queues_before_wake(self): + ex = self.make_exec(env_extra={ + "CHV_MAX_VMS": "2", + "CHV_QUEUE_TIMEOUT_SECONDS": "10", + "CHV_QUEUE_POLL_SECONDS": "2"}) + ex._vms["chv-other1"] = FakeProc(alive=True) + ex._vms["chv-other2"] = FakeProc(alive=True) + # 常驻租户排队：档位与租约原样保留，不升降档。 + # （max_vms=2 → 常驻上限 1，promote 恰好占满 1 槽。） + ex.promote_to_resident("tq", "Paid") + wake_calls = [] + ex.wake_vm = lambda *a, **k: wake_calls.append(a) + with self.assertRaises(ChvFailure) as ctx: + ex._run_with_vm(self._task(project="tq"), + ["dispatch", "--name", "tq", + "--task", "hi"]) + self.assertEqual(ctx.exception.kind, "host_capacity_exceeded") + self.assertEqual(wake_calls, []) + record = read_persistent_vm(ex.state_dir, "tq") + self.assertEqual(record["tier"], "Paid") + self.assertEqual(record["state"], "resident-active") + + def test_wait_for_resident_slot_timeout(self): + ex = self.make_exec(env_extra={ + "CHV_MAX_VMS": "4", + "CHV_QUEUE_TIMEOUT_SECONDS": "10", + "CHV_QUEUE_POLL_SECONDS": "2"}) + ex.promote_to_resident("t1", "Paid") + ex.promote_to_resident("t2", "Heavy") + with self.assertRaises(ChvFailure) as ctx: + ex._wait_for_resident_slot("t3") + self.assertEqual(ctx.exception.kind, "host_capacity_exceeded") + # 不抢占、不挪池：前两个常驻原样。 + self.assertEqual(ex._resident_count(), 2) + self.assertEqual( + read_persistent_vm(ex.state_dir, "t1")["state"], + "resident-active") + + def test_wait_for_resident_slot_free_passes(self): + ex = self.make_exec(env_extra={"CHV_MAX_VMS": "4"}) + ex.promote_to_resident("t1", "Paid") + ex._wait_for_resident_slot("t2") + self.assertEqual(self.sleeps, []) + + +# -- 升级提示点去重（第 8 项）------------------------------------------ +class UpgradeHintTest(HybridBCase): + def test_hint_at_100_tasks_in_30d_and_dedup(self): + ex = self.make_exec() + now = 1800000000.0 + hinted_at = None + for i in range(100): + out = ex.record_pool_task("tfree", tier="Free", + now=now - 1000 + i * 10) + if out["hinted"]: + hinted_at = i + self.assertEqual(hinted_at, 99) + self.assertEqual( + self._event_names().count("resident_upgrade_hint"), 1) + detail = [d for e, d, _k in self.events + if e == "resident_upgrade_hint"][0] + self.assertIn("host_id=host-test-1", detail) + self.assertIn("tenant=tfree", detail) + self.assertIn("tier=Free", detail) + self.assertIn("30d_tasks=100", detail) + # 第 101 次：一周内同一租户不再记第二次。 + out = ex.record_pool_task("tfree", tier="Free", now=now + 10) + self.assertFalse(out["hinted"]) + self.assertEqual( + self._event_names().count("resident_upgrade_hint"), 1) + # 一周后：可再记一次。 + out = ex.record_pool_task("tfree", tier="Free", + now=now + 8 * 86400) + self.assertTrue(out["hinted"]) + self.assertEqual( + self._event_names().count("resident_upgrade_hint"), 2) + + def test_hint_on_7day_streak(self): + ex = self.make_exec() + now = 1800000000.0 + today = int(now // 86400) + hinted = False + for day_offset in range(6, -1, -1): + day_start = (today - day_offset) * 86400 + 3600 + for j in range(3): + out = ex.record_pool_task( + "tstreak", tier="Free", + now=day_start + j * 60) + hinted = hinted or out["hinted"] + self.assertTrue(hinted) + self.assertEqual( + self._event_names().count("resident_upgrade_hint"), 1) + detail = [d for e, d, _k in self.events + if e == "resident_upgrade_hint"][0] + self.assertIn("7d_streak_3_per_day", detail) + + def test_no_hint_below_streak(self): + ex = self.make_exec() + now = 1800000000.0 + today = int(now // 86400) + for day_offset in range(5, -1, -1): + day_start = (today - day_offset) * 86400 + 3600 + for j in range(3): + out = ex.record_pool_task( + "tshort", tier="Free", now=day_start + j * 60) + self.assertFalse(out["hinted"]) + self.assertEqual(self._event_names(), []) + + def test_paid_tier_not_recorded(self): + ex = self.make_exec() + out = ex.record_pool_task("tpaid", tier="Paid", + now=1800000000.0) + self.assertFalse(out["recorded"]) + self.assertFalse(out["hinted"]) + self.assertFalse(os.path.exists( + ex._pool_task_file("tpaid"))) + + +# -- 网关分流（第 9 项）------------------------------------------------- +class GatewaySplitTest(HybridBCase): + def test_gateway_allowed_resident_paid(self): + ex = self.make_exec() + ex.promote_to_resident("t1", "Paid") + task = self._task(project="t1") + self.assertTrue(ex.gateway_allowed_for_task( + task, model_provider="sub2api")) + self.assertTrue(ex.gateway_allowed_for_task( + dict(task, _gateway_provider="sub2api"))) + + def test_gateway_allowed_hibernated_resident(self): + ex = self.make_exec() + now = 1800000000.0 + self._write_resident(ex, "t1", "hibernated", "Paid", now - 10) + self.assertTrue(ex.gateway_allowed_for_task( + self._task(project="t1"), model_provider="sub2api")) + + def test_gateway_denied_pool_tenants(self): + ex = self.make_exec() + # 无登记：池。 + self.assertFalse(ex.gateway_allowed_for_task( + self._task(project="tpool"), model_provider="sub2api")) + # 常驻态但 tier 是 Free：无常驻资格，按池算。 + now = 1800000000.0 + self._write_resident(ex, "tfree", "resident-active", + "Free", now - 10) + self.assertFalse(ex.gateway_allowed_for_task( + self._task(project="tfree"), model_provider="sub2api")) + # demoted：租约已收回，按池算。 + self._write_resident(ex, "tdem", "demoted", "Paid", now - 10) + self.assertFalse(ex.gateway_allowed_for_task( + self._task(project="tdem"), model_provider="sub2api")) + # legacy paused 登记：不是常驻态，按池算。 + vm = ex.persistent_vm_name("tleg") + write_persistent_vm(ex.state_dir, "tleg", { + "vm": vm, "tenant_id": "tleg", "state": "paused", + "origin": "task", "gitea_synced": True, + "updated_at": now}) + self.assertFalse(ex.gateway_allowed_for_task( + self._task(project="tleg"), model_provider="sub2api")) + + def test_gateway_check_non_gateway_not_blocked(self): + ex = self.make_exec() + # 非网关任务不由本门拦（model_provider 不是网关）。 + self.assertTrue(ex.gateway_allowed_for_task( + self._task(project="tpool"), model_provider=None)) + self.assertTrue(ex.gateway_allowed_for_task( + self._task(project="tpool"), model_provider="xcmapi")) + + def test_run_with_vm_gateway_pool_denied_before_wake(self): + ex = self.make_exec() + wake_calls = [] + ex.wake_vm = lambda *a, **k: wake_calls.append(a) + with self.assertRaises(ChvFailure) as ctx: + ex._run_with_vm(self._task(project="tpool"), + ["dispatch", "--name", "tpool", + "--task", "hi"], + model_provider="sub2api") + self.assertEqual(ctx.exception.kind, "gateway_pool_denied") + self.assertEqual(wake_calls, []) + + def test_run_with_vm_gateway_resident_passes_gate(self): + ex = self.make_exec() + ex.promote_to_resident("t1", "Paid") + # 常驻租户的网关任务过门不抛错（全程 dispatch 由既有 + # 测试覆盖，这里只断言门本身放行）。 + ex._check_gateway_vm_class( + self._task(project="t1"), model_provider="sub2api") + + +if __name__ == "__main__": + unittest.main()