feat(chv): 混合模式 B 轨——租约 + 排队 + 升级提示 + 网关门
- 登记扩展 tier/state/last_active_at/lease_expires_at；认领三重核对 原样保留，新增 tier 校验（无 tier 行 = 无常驻资格，fail-closed； 与 P11 用量门不对称已写进注释） - 租约计时器：闲置 30 分钟挂起、再 24 小时降级；阈值经环境变量注入， 生产默认值不变；worker 心跳轮询驱动 sweep_resident_leases； 降级同步失败即中止、VM 保留 - 排队：300 秒超时后 host_capacity_exceeded 明确拒绝；不插队、 不抢占常驻槽位、不升降档；常驻任务仍等常驻槽位；队列深度供 L1 读 - 升级提示点 resident_upgrade_hint：30 天 ≥100 次或连续 7 天每天 ≥3 次， 一周内同一租户最多一次；频率只用于提示，绝不改路由 - 网关分流：常驻 VM 可接网关、池 VM 以 gateway_pool_denied 拒绝 （worker 侧守卫自 P1-1 起已放行 CHV，分流 enforcement 落执行器）
This commit is contained in:
host committed
1 parent
3cdfd0f79e
commit
667a427349
2 files changed
+1533
-42
No files matched your search
+823
-42
@@ -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/<tenant>.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")
|
||||
|
||||
@@ -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()
|
||||
Reference in new issue
Block a user