feat(chv-hybrid-c): autoscaler 池目标数 + 回收同步门 + 审计事件目录
- autoscaler: desired_workers += pool_target(W=2 台/宿主, pool_target_for_hosts);Autoscaler/CLI 支持 --pool-target / --online-hosts;L1 其余语义(30s 轮、scale up fast/down slow、 fail-closed、veto、LIFO)原样复用,L2 不自动加主机 - executor: recycle_vm/recycle_persistent_vm 前加 Gitea 同步门 (chv-sync push 失败即 state/status=sync_failed、VM 不回收、 记 vm_recycle_sync_failed 审计);未配同步/VM 已死走旧路径 - 新增 cloud/chv/hybrid_events.py:9 个设计事件名一字不差, 全部必填 host_id、密钥字段名拒绝入目录 - orphans facade 已存在且复用 registry/orphan 语义,不动 experimental_pool(wired_to_dispatch 恒 False) - 测试:cloud/tests/test_chv_hybrid_c.py 32 项;全量 4611 项 0 失败 0 错误 15 skipped
This commit is contained in:
1 parent
3cdfd0f79e
commit
39c51b7bf2
4 files changed
+953
-11
No files matched your search
+84
-6
@@ -62,6 +62,29 @@ from cloud.logging_config import get_logger, log_event, setup_logging
|
|||||||
from cloud.queue import ready_tasks
|
from cloud.queue import ready_tasks
|
||||||
from cloud.workers import DEFAULT_ALIVE_SECONDS as _WORKER_ALIVE_SECONDS
|
from cloud.workers import DEFAULT_ALIVE_SECONDS as _WORKER_ALIVE_SECONDS
|
||||||
|
|
||||||
|
# CHV 混合模式每宿主预热数 W(DESIGN-HYBRID.md §1.2 已定值:
|
||||||
|
# 2 台/宿主;池目标数 P = W × 在线宿主数)。
|
||||||
|
CHV_WARM_POOL_PER_HOST = 2
|
||||||
|
|
||||||
|
|
||||||
|
def pool_target_for_hosts(online_hosts,
|
||||||
|
per_host=CHV_WARM_POOL_PER_HOST):
|
||||||
|
"""池目标数 P = W × 在线宿主数(§1.2 口径)。
|
||||||
|
|
||||||
|
在线宿主数/每宿主预热数为负或非整数时按 0 计,
|
||||||
|
绝不算出负的池目标。"""
|
||||||
|
try:
|
||||||
|
hosts = int(online_hosts)
|
||||||
|
except (TypeError, ValueError):
|
||||||
|
return 0
|
||||||
|
try:
|
||||||
|
per = int(per_host)
|
||||||
|
except (TypeError, ValueError):
|
||||||
|
return 0
|
||||||
|
if hosts <= 0 or per <= 0:
|
||||||
|
return 0
|
||||||
|
return hosts * per
|
||||||
|
|
||||||
|
|
||||||
def queue_depth(db_path, executor=None):
|
def queue_depth(db_path, executor=None):
|
||||||
"""当前就绪任务数:依赖已满足且状态为 queued 的任务。
|
"""当前就绪任务数:依赖已满足且状态为 queued 的任务。
|
||||||
@@ -126,19 +149,35 @@ def queue_depth(db_path, executor=None):
|
|||||||
|
|
||||||
|
|
||||||
def desired_workers(depth, per_worker_capacity,
|
def desired_workers(depth, per_worker_capacity,
|
||||||
min_workers, max_workers):
|
min_workers, max_workers, pool_target=0):
|
||||||
"""按深度算目标 worker 数。
|
"""按深度算目标 worker 数。
|
||||||
|
|
||||||
depth=0 时也要保留 min_workers(默认 1):冷启动要快,
|
depth=0 时也要保留 min_workers(默认 1):冷启动要快,
|
||||||
且 scaler 缩到 0 后再拉起仍有进程启动延迟;想缩到 0
|
且 scaler 缩到 0 后再拉起仍有进程启动延迟;想缩到 0
|
||||||
可以把 min_workers 配成 0,scaler 仍在跑所以能再拉起。
|
可以把 min_workers 配成 0,scaler 仍在跑所以能再拉起。
|
||||||
|
|
||||||
|
CHV 混合模式(DESIGN-HYBRID.md §1.2,C 轨):期望值
|
||||||
|
作用域扩展——队列深度算出的 fleet 需要量之外,再加
|
||||||
|
池目标数 pool_target(= W × 在线宿主数,W=2,见
|
||||||
|
pool_target_for_hosts)。pool_target=0 时与旧公式
|
||||||
|
一字不差。池目标数计入 max_workers 上限:max_workers
|
||||||
|
须配得下 min_workers + pool_target,否则期望值会被
|
||||||
|
上限截断、池子补不满(配错时 state 文件里的
|
||||||
|
pool_target 与 desired 可见此截断,不静默多扩)。
|
||||||
"""
|
"""
|
||||||
if per_worker_capacity <= 0:
|
if per_worker_capacity <= 0:
|
||||||
per_worker_capacity = 1
|
per_worker_capacity = 1
|
||||||
|
try:
|
||||||
|
pool_target = int(pool_target) if pool_target is not None else 0
|
||||||
|
except (TypeError, ValueError):
|
||||||
|
pool_target = 0
|
||||||
|
if pool_target < 0:
|
||||||
|
pool_target = 0
|
||||||
if depth <= 0:
|
if depth <= 0:
|
||||||
want = min_workers
|
want = min_workers
|
||||||
else:
|
else:
|
||||||
want = math.ceil(depth / per_worker_capacity)
|
want = math.ceil(depth / per_worker_capacity)
|
||||||
|
want += pool_target
|
||||||
return max(min_workers, min(max_workers, want))
|
return max(min_workers, min(max_workers, want))
|
||||||
|
|
||||||
|
|
||||||
@@ -264,7 +303,31 @@ class Autoscaler:
|
|||||||
per_worker_capacity=2, interval=30, scale_down_after=3,
|
per_worker_capacity=2, interval=30, scale_down_after=3,
|
||||||
drain_timeout=60, state_file=None, log_dir=None,
|
drain_timeout=60, state_file=None, log_dir=None,
|
||||||
worker_cmd=None, executor=None,
|
worker_cmd=None, executor=None,
|
||||||
guard_alive_seconds=_WORKER_ALIVE_SECONDS):
|
guard_alive_seconds=_WORKER_ALIVE_SECONDS,
|
||||||
|
pool_target=None, online_hosts=None,
|
||||||
|
warm_pool_per_host=CHV_WARM_POOL_PER_HOST):
|
||||||
|
self.db_path = db_path
|
||||||
|
self.worker_args = list(worker_args)
|
||||||
|
self.min_workers = min_workers
|
||||||
|
self.max_workers = max_workers
|
||||||
|
self.per_worker_capacity = per_worker_capacity
|
||||||
|
self.interval = interval
|
||||||
|
self.scale_down_after = scale_down_after
|
||||||
|
self.drain_timeout = drain_timeout
|
||||||
|
self.guard_alive_seconds = guard_alive_seconds
|
||||||
|
# CHV 混合模式池目标数(§1.2):显式 pool_target 优先;
|
||||||
|
# 未给时按 online_hosts × warm_pool_per_host(W=2)算;
|
||||||
|
# 两者皆无即 0(旧行为)。L2 仍不自动加主机。
|
||||||
|
if pool_target is not None:
|
||||||
|
try:
|
||||||
|
self.pool_target = max(0, int(pool_target))
|
||||||
|
except (TypeError, ValueError):
|
||||||
|
self.pool_target = 0
|
||||||
|
elif online_hosts is not None:
|
||||||
|
self.pool_target = pool_target_for_hosts(
|
||||||
|
online_hosts, warm_pool_per_host)
|
||||||
|
else:
|
||||||
|
self.pool_target = 0
|
||||||
self.db_path = db_path
|
self.db_path = db_path
|
||||||
self.worker_args = list(worker_args)
|
self.worker_args = list(worker_args)
|
||||||
self.min_workers = min_workers
|
self.min_workers = min_workers
|
||||||
@@ -407,7 +470,8 @@ class Autoscaler:
|
|||||||
self._write_state()
|
self._write_state()
|
||||||
return self.last_depth, self.last_desired, len(self.workers)
|
return self.last_depth, self.last_desired, len(self.workers)
|
||||||
desired = desired_workers(depth, self.per_worker_capacity,
|
desired = desired_workers(depth, self.per_worker_capacity,
|
||||||
self.min_workers, self.max_workers)
|
self.min_workers, self.max_workers,
|
||||||
|
pool_target=self.pool_target)
|
||||||
self.last_depth = depth
|
self.last_depth = depth
|
||||||
self.last_desired = desired
|
self.last_desired = desired
|
||||||
self._reap()
|
self._reap()
|
||||||
@@ -477,8 +541,9 @@ class Autoscaler:
|
|||||||
if action != "none":
|
if action != "none":
|
||||||
log_event(self.logger, "autoscaler_reconcile",
|
log_event(self.logger, "autoscaler_reconcile",
|
||||||
message="autoscaler reconcile: depth=%s desired=%s "
|
message="autoscaler reconcile: depth=%s desired=%s "
|
||||||
"actual=%s action=%s"
|
"actual=%s action=%s pool_target=%s"
|
||||||
% (depth, desired, len(self.workers), action))
|
% (depth, desired, len(self.workers), action,
|
||||||
|
self.pool_target))
|
||||||
self._write_state()
|
self._write_state()
|
||||||
return depth, desired, len(self.workers)
|
return depth, desired, len(self.workers)
|
||||||
|
|
||||||
@@ -489,6 +554,7 @@ class Autoscaler:
|
|||||||
"timestamp": time.time(),
|
"timestamp": time.time(),
|
||||||
"depth": self.last_depth,
|
"depth": self.last_depth,
|
||||||
"desired": self.last_desired,
|
"desired": self.last_desired,
|
||||||
|
"pool_target": self.pool_target,
|
||||||
"actual": len(self.workers),
|
"actual": len(self.workers),
|
||||||
"workers": [{"id": wid, "pid": entry["proc"].pid,
|
"workers": [{"id": wid, "pid": entry["proc"].pid,
|
||||||
"started_at": entry["started_at"]}
|
"started_at": entry["started_at"]}
|
||||||
@@ -571,6 +637,16 @@ def main(argv=None):
|
|||||||
"已显式给 --executor 时以显式为准);"
|
"已显式给 --executor 时以显式为准);"
|
||||||
"不给时按全部任务算(可能轻微多扩,"
|
"不给时按全部任务算(可能轻微多扩,"
|
||||||
"受 max-workers 封顶)")
|
"受 max-workers 封顶)")
|
||||||
|
parser.add_argument("--pool-target", type=int, default=None,
|
||||||
|
help="CHV 混合模式池目标数(§1.2):期望值"
|
||||||
|
" = 队列深度算出的 fleet 需要量 + 该数;"
|
||||||
|
"不给时按 --online-hosts × "
|
||||||
|
"--warm-pool-per-host 算;两者皆无即 0")
|
||||||
|
parser.add_argument("--online-hosts", type=int, default=None,
|
||||||
|
help="在线 CHV 宿主数:与 --warm-pool-per-host"
|
||||||
|
" 相乘得池目标数")
|
||||||
|
parser.add_argument("--warm-pool-per-host", type=int, default=2,
|
||||||
|
help="每宿主预热数 W,已定值 2 台/宿主(§1.2)")
|
||||||
parser.add_argument("worker_args", nargs=argparse.REMAINDER,
|
parser.add_argument("worker_args", nargs=argparse.REMAINDER,
|
||||||
help="传给 cloud.worker 的参数(写在 -- 之后)")
|
help="传给 cloud.worker 的参数(写在 -- 之后)")
|
||||||
args = parser.parse_args(argv)
|
args = parser.parse_args(argv)
|
||||||
@@ -595,7 +671,9 @@ def main(argv=None):
|
|||||||
per_worker_capacity=args.per_worker_capacity,
|
per_worker_capacity=args.per_worker_capacity,
|
||||||
interval=args.interval, scale_down_after=args.scale_down_after,
|
interval=args.interval, scale_down_after=args.scale_down_after,
|
||||||
drain_timeout=args.drain_timeout, state_file=args.state_file,
|
drain_timeout=args.drain_timeout, state_file=args.state_file,
|
||||||
log_dir=args.log_dir, executor=args.executor)
|
log_dir=args.log_dir, executor=args.executor,
|
||||||
|
pool_target=args.pool_target, online_hosts=args.online_hosts,
|
||||||
|
warm_pool_per_host=args.warm_pool_per_host)
|
||||||
|
|
||||||
def _on_signal(signum, frame):
|
def _on_signal(signum, frame):
|
||||||
scaler.request_stop()
|
scaler.request_stop()
|
||||||
|
|||||||
+150
-5
@@ -2469,16 +2469,33 @@ class CloudHypervisorExecutor(Executor):
|
|||||||
sandbox=("read-only" if task.get("plan") else None))
|
sandbox=("read-only" if task.get("plan") else None))
|
||||||
|
|
||||||
def recycle_persistent_vm(self, task_or_name: dict | str,
|
def recycle_persistent_vm(self, task_or_name: dict | str,
|
||||||
tenant_id: str | None = None) -> dict:
|
tenant_id: str | None = None,
|
||||||
|
extra_env: dict | None = None,
|
||||||
|
sync: bool = True) -> dict:
|
||||||
"""显式回收持久 VM(唯一删用户盘的入口):先 shutdown
|
"""显式回收持久 VM(唯一删用户盘的入口):先 shutdown
|
||||||
(stop_vm,keep_disk=False),再删 overlay 与数据盘、
|
(stop_vm,keep_disk=False),再删 overlay 与数据盘、
|
||||||
清 tenant 标记与持久登记。返回 {vm, state: recycled,
|
清 tenant 标记与持久登记。返回 {vm, state: recycled,
|
||||||
deleted_disks}。与 recycle_vm 的区别:后者只关 VM
|
deleted_disks}。与 recycle_vm 的区别:后者只关 VM
|
||||||
不删持久盘(持久盘须显式回收才删,用户数据不静默丢)。"""
|
不删持久盘(持久盘须显式回收才删,用户数据不静默丢)。
|
||||||
|
|
||||||
|
C 轨新增(§3.3 同款回收同步门,见 recycle_vm):sync
|
||||||
|
默认开;配了 Gitea 同步且 push 失败时不 shutdown、
|
||||||
|
不删盘,返回 sync_failed(vm_retained=True,
|
||||||
|
deleted_disks=[]),等人工/重试。
|
||||||
|
"""
|
||||||
vm = self.vm_name(task_or_name)
|
vm = self.vm_name(task_or_name)
|
||||||
tenant = tenant_id or self._tenant_id(
|
tenant = tenant_id or self._tenant_id(
|
||||||
task_or_name if isinstance(task_or_name, dict)
|
task_or_name if isinstance(task_or_name, dict)
|
||||||
else {"id": task_or_name})
|
else {"id": task_or_name})
|
||||||
|
if sync:
|
||||||
|
task_dict = task_or_name if isinstance(
|
||||||
|
task_or_name, dict) else {"id": task_or_name}
|
||||||
|
ok, sync_detail = self._recycle_sync_gate(
|
||||||
|
task_dict, extra_env=extra_env)
|
||||||
|
if not ok:
|
||||||
|
return self._sync_failed_result(vm, sync_detail, [])
|
||||||
|
else:
|
||||||
|
sync_detail = "sync skipped by caller"
|
||||||
self.shutdown_vm(task_or_name)
|
self.shutdown_vm(task_or_name)
|
||||||
deleted = []
|
deleted = []
|
||||||
if delete_task_disk(self.state_dir, vm):
|
if delete_task_disk(self.state_dir, vm):
|
||||||
@@ -2496,7 +2513,10 @@ class CloudHypervisorExecutor(Executor):
|
|||||||
else {"id": task_or_name},
|
else {"id": task_or_name},
|
||||||
"vm=%s deleted_disks=%s" % (vm, deleted))
|
"vm=%s deleted_disks=%s" % (vm, deleted))
|
||||||
return {"vm": vm, "state": "recycled",
|
return {"vm": vm, "state": "recycled",
|
||||||
"deleted_disks": deleted}
|
"deleted_disks": deleted,
|
||||||
|
"sync_ok": True,
|
||||||
|
"sync_detail": sync_detail,
|
||||||
|
"vm_retained": False}
|
||||||
|
|
||||||
def snapshot_vm(self, task_or_name: dict | str, snapshot_url: str) -> str:
|
def snapshot_vm(self, task_or_name: dict | str, snapshot_url: str) -> str:
|
||||||
"""给(已暂停的)VM 做快照:ch-remote snapshot <裸 URL>。
|
"""给(已暂停的)VM 做快照:ch-remote snapshot <裸 URL>。
|
||||||
@@ -3152,7 +3172,108 @@ class CloudHypervisorExecutor(Executor):
|
|||||||
只改数据库状态,必须真关 VM。"""
|
只改数据库状态,必须真关 VM。"""
|
||||||
return self.shutdown_vm(task_or_name)
|
return self.shutdown_vm(task_or_name)
|
||||||
|
|
||||||
def recycle_vm(self, task: dict) -> dict:
|
def sync_guest_to_gitea(self, task: dict,
|
||||||
|
extra_env: dict | None = None) -> tuple[bool, str]:
|
||||||
|
"""回收前把客内已提交的工作区改动 push 回 Gitea
|
||||||
|
(DESIGN-HYBRID.md §3.3 回收流程第 1 步)。
|
||||||
|
|
||||||
|
复用客内 chv-sync 的 push 腿(chv-guest/chv-sync.sh):
|
||||||
|
经 SSH 跑 `chv-sync push`;非 fast-forward 即失败,
|
||||||
|
绝不 --force、不自动 rebase(与 chv-sync.sh 的
|
||||||
|
fail-closed 语义同款)。返回 (ok, detail):
|
||||||
|
- 未配 Gitea 同步(extract_gitea_sync_env 为空)→
|
||||||
|
(True, "no gitea sync configured"),无物可推,
|
||||||
|
不拦回收;
|
||||||
|
- VM 已不运行 → (True, "vm not running"):死掉的
|
||||||
|
客机没有可 push 的活工作树,拦回收也推不出去,
|
||||||
|
不拿已死的 VM 卡清理(其盘仍在 state_dir 里,
|
||||||
|
由显式回收/GC 按原口径处理);
|
||||||
|
- push rc==0 → (True, ...);rc!=0 或 SSH 本身失败
|
||||||
|
→ (False, 末行输出截断 200 字符)。chv-sync 的
|
||||||
|
remote URL 是干净 https、token 只走 credential
|
||||||
|
store 文件,所以输出里没有 token(与 syncd 的
|
||||||
|
_run_sync 同款截断口径)。
|
||||||
|
边界(诚实写明):push 只推已提交的提交,未提交的
|
||||||
|
工作树改动不参与同步——与客内 syncd 常驻循环同款
|
||||||
|
口径,本方法不自动 commit(自动 commit 是新语义,
|
||||||
|
设计未定);同步失败的判定只看 push 本身。
|
||||||
|
本方法不重试:push 是幂等的,失败后调用方
|
||||||
|
(recycle_vm / 人工)重新调回收即重试。
|
||||||
|
"""
|
||||||
|
# 方法内 import:与 wake_vm 同款,避免顶层循环导入。
|
||||||
|
from cloud.chv.gitea_sync import extract_gitea_sync_env
|
||||||
|
sync = extract_gitea_sync_env(
|
||||||
|
task, extra_env,
|
||||||
|
config_sync=getattr(self, "gitea_sync_env", None))
|
||||||
|
if not sync:
|
||||||
|
return True, "no gitea sync configured"
|
||||||
|
if not self._vm_is_running(task):
|
||||||
|
return True, "vm not running: nothing to sync"
|
||||||
|
try:
|
||||||
|
ssh_cmd = self.build_ssh_command(task, ["chv-sync", "push"])
|
||||||
|
result = self._call_runner(ssh_cmd,
|
||||||
|
timeout=self.task_timeout)
|
||||||
|
except Exception as exc:
|
||||||
|
return False, "ssh failed: %s" % str(exc)[:200]
|
||||||
|
if getattr(result, "returncode", 1) == 0:
|
||||||
|
return True, "chv-sync push ok"
|
||||||
|
tail = ((getattr(result, "stdout", "") or "")
|
||||||
|
+ (getattr(result, "stderr", "") or "")).strip()
|
||||||
|
tail = tail.splitlines()[-1] if tail else ""
|
||||||
|
return False, "chv-sync push failed: %s" % tail[:200]
|
||||||
|
|
||||||
|
def _emit_recycle_sync_failed(self, task, vm, reason):
|
||||||
|
"""回收同步失败的审计(§3.3“同步失败记审计”):
|
||||||
|
事件 vm_recycle_sync_failed(见 hybrid_events),
|
||||||
|
带 host_id + vm + tenant_id(有则带),绝不带密钥值。
|
||||||
|
审计本身 best-effort:格式化失败时退化成裸 vm 明细,
|
||||||
|
绝不因此掩盖 sync_failed 的返回。"""
|
||||||
|
task_dict = task if isinstance(task, dict) else {"id": task}
|
||||||
|
try:
|
||||||
|
from cloud.chv.hybrid_events import (
|
||||||
|
VM_RECYCLE_SYNC_FAILED, format_detail)
|
||||||
|
fields = {"vm": vm, "reason": reason}
|
||||||
|
tenant = self._tenant_id(task_dict)
|
||||||
|
if tenant:
|
||||||
|
fields["tenant_id"] = tenant
|
||||||
|
detail = format_detail(VM_RECYCLE_SYNC_FAILED, fields)
|
||||||
|
except Exception:
|
||||||
|
detail = "vm=%s reason=%s" % (vm, str(reason)[:200])
|
||||||
|
try:
|
||||||
|
self._emit(VM_RECYCLE_SYNC_FAILED, task_dict, detail,
|
||||||
|
kind="sync_failed")
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _sync_failed_result(vm, detail, deleted_disks=None):
|
||||||
|
"""同步失败的回收返回(§3.3):state/status 均为
|
||||||
|
sync_failed,VM 保留。recycle_vm 与
|
||||||
|
recycle_persistent_vm 共用,免得两处拼同一份 dict;
|
||||||
|
各自再补自己的字段(preserved/lost 或 deleted_disks)。"""
|
||||||
|
out = {"vm": vm, "state": "sync_failed",
|
||||||
|
"status": "sync_failed", "sync_ok": False,
|
||||||
|
"sync_error": detail, "vm_retained": True}
|
||||||
|
if deleted_disks is not None:
|
||||||
|
out["deleted_disks"] = deleted_disks
|
||||||
|
return out
|
||||||
|
|
||||||
|
def _recycle_sync_gate(self, task, extra_env=None):
|
||||||
|
"""回收同步门(§3.3 新增语义的共用判定):
|
||||||
|
同步失败 → (False, detail);可回收 → (True, detail)。
|
||||||
|
调用方在 False 时必须不 shutdown、不删盘、不发
|
||||||
|
recycled 审计,只记 sync_failed 审计并返回
|
||||||
|
state/status 均为 sync_failed 的结果。"""
|
||||||
|
ok, detail = self.sync_guest_to_gitea(task,
|
||||||
|
extra_env=extra_env)
|
||||||
|
if ok:
|
||||||
|
return True, detail
|
||||||
|
self._emit_recycle_sync_failed(task, self.vm_name(task),
|
||||||
|
detail)
|
||||||
|
return False, detail
|
||||||
|
|
||||||
|
def recycle_vm(self, task: dict, extra_env: dict | None = None,
|
||||||
|
sync: bool = True) -> dict:
|
||||||
"""回收 VM(任务结束后的销毁语义):
|
"""回收 VM(任务结束后的销毁语义):
|
||||||
先 shutdown,再返回保留/丢失清单。保留的只有宿主侧
|
先 shutdown,再返回保留/丢失清单。保留的只有宿主侧
|
||||||
的 task_checkpoints / task_attempts(数据库里的断点与
|
的 task_checkpoints / task_attempts(数据库里的断点与
|
||||||
@@ -3160,9 +3281,30 @@ class CloudHypervisorExecutor(Executor):
|
|||||||
丢失的是客机内的临时文件与未拷出的 worktree 临时物
|
丢失的是客机内的临时文件与未拷出的 worktree 临时物
|
||||||
——recycle 绝不声称“恢复临时文件”。需要拷出的工件
|
——recycle 绝不声称“恢复临时文件”。需要拷出的工件
|
||||||
必须在 shutdown 之前由调用方经 SSH 拷出,本方法不
|
必须在 shutdown 之前由调用方经 SSH 拷出,本方法不
|
||||||
假装替调用方拷了。"""
|
假装替调用方拷了。
|
||||||
|
|
||||||
|
C 轨新增(DESIGN-HYBRID.md §3.3):sync=True(默认)
|
||||||
|
时先过回收同步门——配了 Gitea 同步且 `chv-sync push`
|
||||||
|
失败时,任务标同步失败(返回 state/status 均为
|
||||||
|
sync_failed,sync_ok=False,vm_retained=True),VM
|
||||||
|
不回收、不复用,等人工/重试,绝不丢改动后静默回收;
|
||||||
|
审计记 vm_recycle_sync_failed。未配同步、VM 已死、
|
||||||
|
或 sync=False(调用方显式跳过)时走旧回收路径。
|
||||||
|
"""
|
||||||
if not isinstance(task, dict) or task.get("id") is None:
|
if not isinstance(task, dict) or task.get("id") is None:
|
||||||
raise ValueError("任务缺少 id")
|
raise ValueError("任务缺少 id")
|
||||||
|
if sync:
|
||||||
|
ok, sync_detail = self._recycle_sync_gate(
|
||||||
|
task, extra_env=extra_env)
|
||||||
|
if not ok:
|
||||||
|
out = self._sync_failed_result(self.vm_name(task),
|
||||||
|
sync_detail)
|
||||||
|
out["preserved"] = ["task_checkpoints",
|
||||||
|
"task_attempts"]
|
||||||
|
out["lost"] = []
|
||||||
|
return out
|
||||||
|
else:
|
||||||
|
sync_detail = "sync skipped by caller"
|
||||||
self.shutdown_vm(task)
|
self.shutdown_vm(task)
|
||||||
self._emit("vm_recycled", task,
|
self._emit("vm_recycled", task,
|
||||||
"vm=%s" % self.vm_name(task))
|
"vm=%s" % self.vm_name(task))
|
||||||
@@ -3171,6 +3313,9 @@ class CloudHypervisorExecutor(Executor):
|
|||||||
"state": "recycled",
|
"state": "recycled",
|
||||||
"preserved": ["task_checkpoints", "task_attempts"],
|
"preserved": ["task_checkpoints", "task_attempts"],
|
||||||
"lost": ["guest /tmp", "guest worktree temp files"],
|
"lost": ["guest /tmp", "guest worktree temp files"],
|
||||||
|
"sync_ok": True,
|
||||||
|
"sync_detail": sync_detail,
|
||||||
|
"vm_retained": False,
|
||||||
}
|
}
|
||||||
|
|
||||||
def _stop_dead_socket_vms(self, entries):
|
def _stop_dead_socket_vms(self, entries):
|
||||||
|
|||||||
@@ -0,0 +1,146 @@
|
|||||||
|
"""CHV 混合模式审计事件目录(C 轨,feat-chv-hybrid-c)。
|
||||||
|
|
||||||
|
唯一依据:~/workspace/chv-multi-host/DESIGN-HYBRID.md
|
||||||
|
§7 与 §9.1 第 13 项。事件名字符串与设计原文一字不差,
|
||||||
|
供 A/B 两轨的发射点 import(A/B 工作树里还没有本文件,
|
||||||
|
他们先按设计原文的事件名字符串直接发射,协调员集成时
|
||||||
|
统一收到本模块名下)。
|
||||||
|
|
||||||
|
规则(设计 §7):
|
||||||
|
- 全部事件带 host_id(宿主名,默认 socket.gethostname())。
|
||||||
|
- 绝不带密钥值:字段名命中密钥名(见 SECRET_FIELD_NAMES)
|
||||||
|
即拒绝,错误信息只含字段名、不含值;调用方仍须遵守
|
||||||
|
executor._emit 的既有契约(detail 只含 VM 名、路径、
|
||||||
|
URL、kind,不含任何 key 值)。
|
||||||
|
- 本模块只定名 + 必填字段 + 校验/格式化,不发事件、
|
||||||
|
不读库、不碰 VM;发射仍走各轨现有的 _emit / 日志路径。
|
||||||
|
|
||||||
|
另有一枚 C 轨事件 vm_recycle_sync_failed:不在 §9.1-13
|
||||||
|
的名单里,但 §3.3 要求“同步失败 → 任务标同步失败
|
||||||
|
(审计)”,回收同步失败的审计须有个名字,故在此单列
|
||||||
|
并注明来源,不混进设计名单。
|
||||||
|
"""
|
||||||
|
|
||||||
|
import socket
|
||||||
|
|
||||||
|
# -- 设计原文事件名(§9.1 第 13 项 / §7,一字不差) ------------
|
||||||
|
WARM_POOL_ACQUIRE = "warm_pool_acquire"
|
||||||
|
WARM_POOL_MISS = "warm_pool_miss"
|
||||||
|
WARM_POOL_RELEASE = "warm_pool_release"
|
||||||
|
VM_RESIDENT_SUSPENDED = "vm_resident_suspended"
|
||||||
|
VM_RESIDENT_DEMOTED = "vm_resident_demoted"
|
||||||
|
VM_RESIDENT_PROMOTED = "vm_resident_promoted"
|
||||||
|
RESIDENT_UPGRADE_HINT = "resident_upgrade_hint"
|
||||||
|
SNAPSHOT_TENANT_MISMATCH = "snapshot_tenant_mismatch"
|
||||||
|
VM_RESTORE_DENIED = "vm_restore_denied"
|
||||||
|
|
||||||
|
# 设计 §9.1-13 名单一共 9 个事件(acquire/miss/release
|
||||||
|
# 是 §7“warm_pool_acquire/miss/release”的三枚展开,
|
||||||
|
# miss 的全名 warm_pool_miss 见 §7“须记 warm_pool_miss
|
||||||
|
# 审计”)。
|
||||||
|
HYBRID_EVENTS = (
|
||||||
|
WARM_POOL_ACQUIRE,
|
||||||
|
WARM_POOL_MISS,
|
||||||
|
WARM_POOL_RELEASE,
|
||||||
|
VM_RESIDENT_SUSPENDED,
|
||||||
|
VM_RESIDENT_DEMOTED,
|
||||||
|
VM_RESIDENT_PROMOTED,
|
||||||
|
RESIDENT_UPGRADE_HINT,
|
||||||
|
SNAPSHOT_TENANT_MISMATCH,
|
||||||
|
VM_RESTORE_DENIED,
|
||||||
|
)
|
||||||
|
|
||||||
|
# -- C 轨回收同步审计(§3.3,非 §9.1-13 名单) ---------------
|
||||||
|
VM_RECYCLE_SYNC_FAILED = "vm_recycle_sync_failed"
|
||||||
|
|
||||||
|
ALL_KNOWN_EVENTS = HYBRID_EVENTS + (VM_RECYCLE_SYNC_FAILED,)
|
||||||
|
|
||||||
|
# -- 必填字段:全部事件都须带 host_id -------------------------
|
||||||
|
# 其余字段按事件语义:vm = VM 名,tenant_id = 租户,
|
||||||
|
# reason = 失败原因短文本(已截断,绝不含密钥值)。
|
||||||
|
REQUIRED_FIELDS = {
|
||||||
|
WARM_POOL_ACQUIRE: ("host_id", "vm"),
|
||||||
|
WARM_POOL_MISS: ("host_id",),
|
||||||
|
WARM_POOL_RELEASE: ("host_id", "vm"),
|
||||||
|
VM_RESIDENT_SUSPENDED: ("host_id", "tenant_id", "vm"),
|
||||||
|
VM_RESIDENT_DEMOTED: ("host_id", "tenant_id", "vm"),
|
||||||
|
VM_RESIDENT_PROMOTED: ("host_id", "tenant_id", "vm"),
|
||||||
|
RESIDENT_UPGRADE_HINT: ("host_id", "tenant_id"),
|
||||||
|
SNAPSHOT_TENANT_MISMATCH: ("host_id", "tenant_id", "vm"),
|
||||||
|
VM_RESTORE_DENIED: ("host_id", "tenant_id", "vm"),
|
||||||
|
VM_RECYCLE_SYNC_FAILED: ("host_id", "vm"),
|
||||||
|
}
|
||||||
|
|
||||||
|
# 密钥字段名:命中即拒绝。exact 匹配 + token/secret/
|
||||||
|
# password 子串匹配(host_key 等公开指纹名不在此列,
|
||||||
|
# 不可误拦)。
|
||||||
|
SECRET_FIELD_NAMES = frozenset({
|
||||||
|
"gitea_token", "token", "secret", "password", "api_key",
|
||||||
|
"apikey", "private_key", "access_key", "session_key",
|
||||||
|
"credential", "credentials", "seed", "auth",
|
||||||
|
})
|
||||||
|
_SECRET_SUBSTRINGS = ("token", "secret", "password")
|
||||||
|
|
||||||
|
|
||||||
|
def _is_secret_field(name: str) -> bool:
|
||||||
|
lowered = str(name).lower()
|
||||||
|
if lowered in SECRET_FIELD_NAMES:
|
||||||
|
return True
|
||||||
|
return any(part in lowered for part in _SECRET_SUBSTRINGS)
|
||||||
|
|
||||||
|
|
||||||
|
def default_host_id() -> str:
|
||||||
|
"""宿主名:审计事件的 host_id 缺省值。"""
|
||||||
|
return socket.gethostname()
|
||||||
|
|
||||||
|
|
||||||
|
def validate_event_fields(event: str, fields: dict) -> dict:
|
||||||
|
"""校验一个事件的字段,返回清洗后的副本(去 None 值)。
|
||||||
|
|
||||||
|
未知事件名、缺必填字段、host_id 为空、字段名命中密钥
|
||||||
|
名时一律 ValueError;错误信息只含事件名/字段名,
|
||||||
|
绝不含字段值。"""
|
||||||
|
if event not in ALL_KNOWN_EVENTS:
|
||||||
|
raise ValueError("未知审计事件:%s" % event)
|
||||||
|
cleaned = {k: v for k, v in dict(fields or {}).items()
|
||||||
|
if v is not None}
|
||||||
|
secret = sorted(k for k in cleaned if _is_secret_field(k))
|
||||||
|
if secret:
|
||||||
|
raise ValueError(
|
||||||
|
"审计事件 %s 不得带密钥字段:%s" % (event, ",".join(secret)))
|
||||||
|
missing = [k for k in REQUIRED_FIELDS[event]
|
||||||
|
if k not in cleaned
|
||||||
|
or (isinstance(cleaned[k], str)
|
||||||
|
and not cleaned[k].strip())]
|
||||||
|
if missing:
|
||||||
|
raise ValueError(
|
||||||
|
"审计事件 %s 缺必填字段:%s" % (event, ",".join(missing)))
|
||||||
|
return cleaned
|
||||||
|
|
||||||
|
|
||||||
|
def build_event(event: str, fields: dict | None = None,
|
||||||
|
host_id: str | None = None) -> dict:
|
||||||
|
"""组装一个事件的字段 dict:host_id 未给时取本机宿主名,
|
||||||
|
再过 validate_event_fields。返回的 dict 可直接用于
|
||||||
|
format_detail 或交审计回调。"""
|
||||||
|
merged = dict(fields or {})
|
||||||
|
if host_id is not None:
|
||||||
|
merged["host_id"] = host_id
|
||||||
|
elif not merged.get("host_id"):
|
||||||
|
merged["host_id"] = default_host_id()
|
||||||
|
return validate_event_fields(event, merged)
|
||||||
|
|
||||||
|
|
||||||
|
def format_detail(event: str, fields: dict | None = None,
|
||||||
|
host_id: str | None = None) -> str:
|
||||||
|
"""把事件字段格式化成 executor._emit 的 detail 字符串
|
||||||
|
("k=v k=v",键排序,值截断 200 字符、去换行)。
|
||||||
|
先过 build_event 校验,所以带 host_id、不带密钥字段。"""
|
||||||
|
cleaned = build_event(event, fields, host_id=host_id)
|
||||||
|
parts = []
|
||||||
|
for key in sorted(cleaned):
|
||||||
|
value = str(cleaned[key]).replace("\n", " ").replace("\r", " ")
|
||||||
|
if len(value) > 200:
|
||||||
|
value = value[:200]
|
||||||
|
parts.append("%s=%s" % (key, value))
|
||||||
|
return " ".join(parts)
|
||||||
@@ -0,0 +1,573 @@
|
|||||||
|
"""CHV 混合模式 C 轨测试(feat-chv-hybrid-c)。
|
||||||
|
|
||||||
|
只覆盖本轨的三个改动点(DESIGN-HYBRID.md §9.1 第 10–13 项):
|
||||||
|
- autoscaler desired_workers 作用域扩展(§1.2,W=2);
|
||||||
|
- 回收同步失败即任务标同步失败、VM 不回收(§3.3);
|
||||||
|
- 审计事件目录(§7,事件名与设计原文一致、带 host_id、
|
||||||
|
不带密钥值)与 orphans facade 核实(§1.3/ADR-5/ADR-12)。
|
||||||
|
|
||||||
|
只用标准库与 mock runner/launcher/api_client,绝不启动
|
||||||
|
真实 VM、不联网、不碰生产库。真机验证(WSL 3 台排布、
|
||||||
|
热恢复 ≤2 秒等)不在本文件声称已验证。
|
||||||
|
"""
|
||||||
|
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
import socket
|
||||||
|
import subprocess
|
||||||
|
import sys
|
||||||
|
import tempfile
|
||||||
|
import time
|
||||||
|
import unittest
|
||||||
|
from pathlib import Path
|
||||||
|
from unittest import mock
|
||||||
|
|
||||||
|
try:
|
||||||
|
from cloud.autoscaler import (
|
||||||
|
CHV_WARM_POOL_PER_HOST,
|
||||||
|
Autoscaler,
|
||||||
|
desired_workers,
|
||||||
|
pool_target_for_hosts,
|
||||||
|
)
|
||||||
|
from cloud.chv import hybrid_events
|
||||||
|
from cloud.chv.executor import CloudHypervisorExecutor
|
||||||
|
from cloud.chv.orphans import reconcile_orphans_for_executor
|
||||||
|
from cloud.server import init_db
|
||||||
|
except ImportError:
|
||||||
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||||||
|
from cloud.autoscaler import (
|
||||||
|
CHV_WARM_POOL_PER_HOST,
|
||||||
|
Autoscaler,
|
||||||
|
desired_workers,
|
||||||
|
pool_target_for_hosts,
|
||||||
|
)
|
||||||
|
from cloud.chv import hybrid_events
|
||||||
|
from cloud.chv.executor import CloudHypervisorExecutor
|
||||||
|
from cloud.chv.orphans import reconcile_orphans_for_executor
|
||||||
|
from cloud.server import init_db
|
||||||
|
|
||||||
|
|
||||||
|
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
|
||||||
|
self.terminated = False
|
||||||
|
self.killed = False
|
||||||
|
|
||||||
|
def poll(self):
|
||||||
|
return None if self._alive else 1
|
||||||
|
|
||||||
|
def terminate(self):
|
||||||
|
self.terminated = True
|
||||||
|
self._alive = False
|
||||||
|
|
||||||
|
def wait(self, timeout=None):
|
||||||
|
return 0
|
||||||
|
|
||||||
|
def kill(self):
|
||||||
|
self.killed = True
|
||||||
|
self._alive = False
|
||||||
|
|
||||||
|
|
||||||
|
def _write(path, content=b"x"):
|
||||||
|
with open(path, "wb") as fh:
|
||||||
|
fh.write(content)
|
||||||
|
os.chmod(path, 0o755)
|
||||||
|
return path
|
||||||
|
|
||||||
|
|
||||||
|
# -- 11. desired_workers + 池目标数 --------------------------------------
|
||||||
|
class DesiredWorkersPoolTestCase(unittest.TestCase):
|
||||||
|
def test_pool_target_zero_unchanged(self):
|
||||||
|
# pool_target=0(缺省)时与旧公式一字不差。
|
||||||
|
self.assertEqual(desired_workers(0, 2, 1, 2), 1)
|
||||||
|
self.assertEqual(desired_workers(0, 2, 0, 2), 0)
|
||||||
|
self.assertEqual(desired_workers(1, 2, 1, 2), 1)
|
||||||
|
self.assertEqual(desired_workers(3, 2, 1, 2), 2)
|
||||||
|
self.assertEqual(desired_workers(10, 2, 1, 2), 2)
|
||||||
|
self.assertEqual(desired_workers(5, 1, 1, 3), 3)
|
||||||
|
self.assertEqual(desired_workers(5, 1, 1, 3, pool_target=0), 3)
|
||||||
|
|
||||||
|
def test_pool_target_additive(self):
|
||||||
|
# 期望值 = fleet 需要量 + 池目标数(§1.2)。
|
||||||
|
self.assertEqual(desired_workers(0, 2, 1, 20, pool_target=12), 13)
|
||||||
|
self.assertEqual(desired_workers(4, 2, 1, 20, pool_target=12), 14)
|
||||||
|
self.assertEqual(desired_workers(3, 2, 1, 20, pool_target=2), 4)
|
||||||
|
# depth=0 且 min_workers=0 时池目标仍维持(预热不因
|
||||||
|
# 队列空而归零)。
|
||||||
|
self.assertEqual(desired_workers(0, 2, 0, 20, pool_target=12), 12)
|
||||||
|
|
||||||
|
def test_pool_target_capped_by_max_workers(self):
|
||||||
|
# 池目标数计入 max_workers 上限:配不下即截断,不静默多扩。
|
||||||
|
self.assertEqual(desired_workers(0, 2, 1, 2, pool_target=12), 2)
|
||||||
|
self.assertEqual(desired_workers(10, 2, 1, 3, pool_target=12), 3)
|
||||||
|
|
||||||
|
def test_pool_target_negative_or_none_is_zero(self):
|
||||||
|
self.assertEqual(desired_workers(0, 2, 1, 2, pool_target=-5), 1)
|
||||||
|
self.assertEqual(desired_workers(0, 2, 1, 2, pool_target=None), 1)
|
||||||
|
|
||||||
|
|
||||||
|
class PoolTargetForHostsTestCase(unittest.TestCase):
|
||||||
|
def test_w_is_two_per_host(self):
|
||||||
|
self.assertEqual(CHV_WARM_POOL_PER_HOST, 2)
|
||||||
|
|
||||||
|
def test_pool_target_is_w_times_online_hosts(self):
|
||||||
|
self.assertEqual(pool_target_for_hosts(6), 12)
|
||||||
|
self.assertEqual(pool_target_for_hosts(22), 44)
|
||||||
|
self.assertEqual(pool_target_for_hosts(1), 2)
|
||||||
|
self.assertEqual(pool_target_for_hosts(0), 0)
|
||||||
|
|
||||||
|
def test_invalid_hosts_is_zero(self):
|
||||||
|
self.assertEqual(pool_target_for_hosts(-1), 0)
|
||||||
|
self.assertEqual(pool_target_for_hosts(None), 0)
|
||||||
|
self.assertEqual(pool_target_for_hosts("not-a-number"), 0)
|
||||||
|
|
||||||
|
|
||||||
|
def _insert_task(db_path, task_id):
|
||||||
|
now = time.time()
|
||||||
|
import sqlite3
|
||||||
|
conn = sqlite3.connect(db_path)
|
||||||
|
try:
|
||||||
|
conn.execute(
|
||||||
|
"INSERT INTO tasks(id, project_id, prompt, status,"
|
||||||
|
" idempotency_key, depends_on, priority, created_by,"
|
||||||
|
" created_at, updated_at, last_heartbeat)"
|
||||||
|
" VALUES(?, 'p1', 'x', 'queued', NULL, '[]', 0,"
|
||||||
|
" 'tester', ?, ?, NULL)",
|
||||||
|
(task_id, now, now))
|
||||||
|
conn.commit()
|
||||||
|
finally:
|
||||||
|
conn.close()
|
||||||
|
|
||||||
|
|
||||||
|
class AutoscalerPoolReconcileTestCase(unittest.TestCase):
|
||||||
|
"""L1 其余语义不变:30 秒一轮(interval 缺省)、scale up
|
||||||
|
fast / scale down slow(scale_down_after 冷却)原样复用,
|
||||||
|
只验证期望值多了池目标数。"""
|
||||||
|
|
||||||
|
def setUp(self):
|
||||||
|
self.tmp = tempfile.TemporaryDirectory()
|
||||||
|
self.db_path = os.path.join(self.tmp.name, "control.db")
|
||||||
|
init_db(self.db_path)
|
||||||
|
self.state_file = os.path.join(self.tmp.name, "state.json")
|
||||||
|
self.fake_cmd = [sys.executable, "-c",
|
||||||
|
"import time; time.sleep(120)"]
|
||||||
|
|
||||||
|
def tearDown(self):
|
||||||
|
if hasattr(self, "scaler"):
|
||||||
|
for wid in list(self.scaler.workers):
|
||||||
|
try:
|
||||||
|
self.scaler._stop_worker(wid)
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
self.tmp.cleanup()
|
||||||
|
|
||||||
|
def _scaler(self, **kwargs):
|
||||||
|
kwargs.setdefault("min_workers", 1)
|
||||||
|
kwargs.setdefault("max_workers", 5)
|
||||||
|
kwargs.setdefault("per_worker_capacity", 2)
|
||||||
|
kwargs.setdefault("scale_down_after", 1)
|
||||||
|
kwargs.setdefault("drain_timeout", 5)
|
||||||
|
kwargs.setdefault("state_file", self.state_file)
|
||||||
|
self.scaler = Autoscaler(
|
||||||
|
self.db_path, [], worker_cmd=self.fake_cmd, **kwargs)
|
||||||
|
return self.scaler
|
||||||
|
|
||||||
|
def test_empty_queue_still_holds_pool(self):
|
||||||
|
scaler = self._scaler(pool_target=2)
|
||||||
|
depth, desired, actual = scaler.reconcile()
|
||||||
|
self.assertEqual((depth, desired, actual), (0, 3, 3))
|
||||||
|
|
||||||
|
def test_depth_plus_pool_and_scale_down_slow(self):
|
||||||
|
scaler = self._scaler(pool_target=2, scale_down_after=2)
|
||||||
|
for tid in ("A", "B", "C", "D"):
|
||||||
|
_insert_task(self.db_path, tid)
|
||||||
|
depth, desired, actual = scaler.reconcile()
|
||||||
|
self.assertEqual((depth, desired, actual), (4, 4, 4))
|
||||||
|
# 任务全完成:深度归零但池目标仍在,期望值回落到 3
|
||||||
|
# 而不是 1;缩容仍须等满 2 轮冷却(scale down slow)。
|
||||||
|
import sqlite3
|
||||||
|
conn = sqlite3.connect(self.db_path)
|
||||||
|
try:
|
||||||
|
conn.execute("UPDATE tasks SET status='completed'")
|
||||||
|
conn.commit()
|
||||||
|
finally:
|
||||||
|
conn.close()
|
||||||
|
scaler.reconcile()
|
||||||
|
self.assertEqual(len(scaler.workers), 4)
|
||||||
|
self.assertEqual(scaler.last_action, "scale_down_pending_1/2")
|
||||||
|
depth, desired, actual = scaler.reconcile()
|
||||||
|
self.assertEqual((depth, desired, actual), (0, 3, 3))
|
||||||
|
|
||||||
|
def test_online_hosts_derives_pool_target(self):
|
||||||
|
scaler = self._scaler(online_hosts=6, max_workers=20)
|
||||||
|
self.assertEqual(scaler.pool_target, 12)
|
||||||
|
depth, desired, actual = scaler.reconcile()
|
||||||
|
self.assertEqual((depth, desired, actual), (0, 13, 13))
|
||||||
|
|
||||||
|
def test_explicit_pool_target_wins_over_hosts(self):
|
||||||
|
scaler = self._scaler(pool_target=2, online_hosts=6)
|
||||||
|
self.assertEqual(scaler.pool_target, 2)
|
||||||
|
|
||||||
|
def test_state_file_records_pool_target(self):
|
||||||
|
scaler = self._scaler(pool_target=2)
|
||||||
|
scaler.reconcile()
|
||||||
|
with open(self.state_file) as fh:
|
||||||
|
state = json.load(fh)
|
||||||
|
self.assertEqual(state["pool_target"], 2)
|
||||||
|
self.assertEqual(state["desired"], 3)
|
||||||
|
|
||||||
|
def test_main_passes_pool_flags(self):
|
||||||
|
import cloud.autoscaler as autoscaler_mod
|
||||||
|
with mock.patch.object(
|
||||||
|
autoscaler_mod, "Autoscaler") as mock_cls, \
|
||||||
|
mock.patch("signal.signal"):
|
||||||
|
mock_cls.return_value.run_loop = mock.Mock()
|
||||||
|
autoscaler_mod.main([
|
||||||
|
"--db", self.db_path, "--pool-target", "12",
|
||||||
|
"--", "--interval", "10"])
|
||||||
|
kwargs = mock_cls.call_args.kwargs
|
||||||
|
self.assertEqual(kwargs["pool_target"], 12)
|
||||||
|
with mock.patch.object(
|
||||||
|
autoscaler_mod, "Autoscaler") as mock_cls, \
|
||||||
|
mock.patch("signal.signal"):
|
||||||
|
mock_cls.return_value.run_loop = mock.Mock()
|
||||||
|
autoscaler_mod.main([
|
||||||
|
"--db", self.db_path, "--online-hosts", "6",
|
||||||
|
"--", "--interval", "10"])
|
||||||
|
kwargs = mock_cls.call_args.kwargs
|
||||||
|
self.assertEqual(kwargs["online_hosts"], 6)
|
||||||
|
self.assertEqual(kwargs["warm_pool_per_host"], 2)
|
||||||
|
|
||||||
|
|
||||||
|
# -- 12. 回收同步 ---------------------------------------------------------
|
||||||
|
class RecycleSyncCase(unittest.TestCase):
|
||||||
|
"""mock runner/launcher/api_client,不启动真实 VM。
|
||||||
|
push_rc 控制客内 `chv-sync push` 的返回值。"""
|
||||||
|
|
||||||
|
TOKEN = "SECRET-TOKEN-XYZ-DO-NOT-LEAK"
|
||||||
|
|
||||||
|
def setUp(self):
|
||||||
|
self.tmp = tempfile.TemporaryDirectory()
|
||||||
|
self.dir = self.tmp.name
|
||||||
|
self.binary = _write(os.path.join(self.dir, "cloud-hypervisor"))
|
||||||
|
self.ch_remote = _write(os.path.join(self.dir, "ch-remote"))
|
||||||
|
self.kernel = _write(os.path.join(self.dir, "vmlinux"))
|
||||||
|
self.rootfs = _write(os.path.join(self.dir, "disk.raw"), b"R" * 100)
|
||||||
|
self.ssh_key = _write(os.path.join(self.dir, "id_chv"))
|
||||||
|
self.kvm = _write(os.path.join(self.dir, "kvm"))
|
||||||
|
self.cpuinfo = os.path.join(self.dir, "cpuinfo")
|
||||||
|
with open(self.cpuinfo, "w") as fh:
|
||||||
|
fh.write("flags : fpu vmx sse\n")
|
||||||
|
self.state_dir = os.path.join(self.dir, "state")
|
||||||
|
self.calls = []
|
||||||
|
self.audit = []
|
||||||
|
self.proc = FakeProc(alive=True)
|
||||||
|
self.push_rc = 0
|
||||||
|
self.push_stdout = "已 push"
|
||||||
|
self.push_stderr = ""
|
||||||
|
self.raise_on_push = False
|
||||||
|
|
||||||
|
def tearDown(self):
|
||||||
|
self.tmp.cleanup()
|
||||||
|
|
||||||
|
def _runner(self, cmd, timeout=None):
|
||||||
|
self.calls.append(list(cmd))
|
||||||
|
joined = " ".join(str(c) for c in cmd)
|
||||||
|
if "chv-sync" in joined:
|
||||||
|
if self.raise_on_push:
|
||||||
|
raise RuntimeError("ssh broken")
|
||||||
|
return FakeResult(self.push_rc, self.push_stdout,
|
||||||
|
self.push_stderr)
|
||||||
|
return FakeResult(0, "", "")
|
||||||
|
|
||||||
|
def _make_exec(self, sync_configured=True):
|
||||||
|
env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888",
|
||||||
|
"CHV_HOST_KEY_MODE": "keyscan",
|
||||||
|
"CHV_SSH_FORWARD": "1"}
|
||||||
|
if sync_configured:
|
||||||
|
env.update({
|
||||||
|
"CHV_GITEA_URL": "https://gitea.tose.top",
|
||||||
|
"CHV_GITEA_USER": "tester",
|
||||||
|
"CHV_GITEA_TOKEN": self.TOKEN,
|
||||||
|
"CHV_SYNC_REPO": "tester/repo",
|
||||||
|
})
|
||||||
|
ex = CloudHypervisorExecutor(
|
||||||
|
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=self._runner,
|
||||||
|
launcher=lambda cmd: self.proc,
|
||||||
|
api_client=lambda *a, **k: (200, {}),
|
||||||
|
sleeper=lambda s: None, env=env)
|
||||||
|
ex.audit_callback = lambda e, t, d, k=None: self.audit.append(
|
||||||
|
(e, t.get("id") if isinstance(t, dict) else t, d, k))
|
||||||
|
return ex
|
||||||
|
|
||||||
|
def _task(self):
|
||||||
|
return {"id": "abc123", "prompt": "x", "project_id": "proj1"}
|
||||||
|
|
||||||
|
def _mark_vm_running(self, ex, vm_name="chv-abc123"):
|
||||||
|
os.makedirs(self.state_dir, exist_ok=True)
|
||||||
|
with open(os.path.join(self.state_dir, vm_name + ".sock"),
|
||||||
|
"w") as fh:
|
||||||
|
fh.write("")
|
||||||
|
ex._vms[vm_name] = self.proc
|
||||||
|
|
||||||
|
def _audit_events(self):
|
||||||
|
return [e for e, _t, _d, _k in self.audit]
|
||||||
|
|
||||||
|
def _audit_details(self):
|
||||||
|
return [d for _e, _t, d, _k in self.audit]
|
||||||
|
|
||||||
|
def test_no_sync_configured_recycles_as_before(self):
|
||||||
|
# 未配 Gitea 同步:无物可推,走旧回收路径(回归)。
|
||||||
|
ex = self._make_exec(sync_configured=False)
|
||||||
|
self._mark_vm_running(ex)
|
||||||
|
out = ex.recycle_vm(self._task())
|
||||||
|
self.assertEqual(out["state"], "recycled")
|
||||||
|
self.assertTrue(out["sync_ok"])
|
||||||
|
self.assertFalse(out["vm_retained"])
|
||||||
|
self.assertTrue(self.proc.terminated)
|
||||||
|
self.assertIn("vm_recycled", self._audit_events())
|
||||||
|
self.assertFalse(any("chv-sync" in " ".join(c)
|
||||||
|
for c in self.calls))
|
||||||
|
|
||||||
|
def test_sync_push_ok_then_recycle(self):
|
||||||
|
ex = self._make_exec()
|
||||||
|
self._mark_vm_running(ex)
|
||||||
|
self.push_rc = 0
|
||||||
|
out = ex.recycle_vm(self._task())
|
||||||
|
self.assertEqual(out["state"], "recycled")
|
||||||
|
self.assertTrue(out["sync_ok"])
|
||||||
|
self.assertFalse(out["vm_retained"])
|
||||||
|
self.assertTrue(self.proc.terminated)
|
||||||
|
self.assertIn("vm_recycled", self._audit_events())
|
||||||
|
self.assertTrue(any("chv-sync" in " ".join(c)
|
||||||
|
for c in self.calls))
|
||||||
|
self.assertNotIn("vm_recycle_sync_failed",
|
||||||
|
self._audit_events())
|
||||||
|
|
||||||
|
def test_sync_push_failed_marks_sync_failed_and_retains_vm(self):
|
||||||
|
# §3.3 新增语义:同步失败即任务标同步失败、VM 不回收。
|
||||||
|
ex = self._make_exec()
|
||||||
|
self._mark_vm_running(ex)
|
||||||
|
self.push_rc = 1
|
||||||
|
self.push_stderr = "push 被拒绝(远端已有新提交)"
|
||||||
|
out = ex.recycle_vm(self._task())
|
||||||
|
self.assertEqual(out["state"], "sync_failed")
|
||||||
|
self.assertEqual(out["status"], "sync_failed")
|
||||||
|
self.assertFalse(out["sync_ok"])
|
||||||
|
self.assertTrue(out["vm_retained"])
|
||||||
|
self.assertIn("push", out["sync_error"])
|
||||||
|
# VM 保留:进程未停、socket 未删、未发 recycled/shutdown。
|
||||||
|
self.assertFalse(self.proc.terminated)
|
||||||
|
self.assertTrue(os.path.exists(os.path.join(
|
||||||
|
self.state_dir, "chv-abc123.sock")))
|
||||||
|
self.assertNotIn("vm_recycled", self._audit_events())
|
||||||
|
self.assertNotIn("vm_shutdown", self._audit_events())
|
||||||
|
self.assertNotIn("vm_destroy", self._audit_events())
|
||||||
|
# 记审计:事件名正确、带 host_id、不带密钥值。
|
||||||
|
self.assertIn("vm_recycle_sync_failed", self._audit_events())
|
||||||
|
details = " ".join(self._audit_details())
|
||||||
|
self.assertIn("host_id=", details)
|
||||||
|
self.assertNotIn(self.TOKEN, details)
|
||||||
|
self.assertNotIn(self.TOKEN, json.dumps(self.audit))
|
||||||
|
|
||||||
|
def test_sync_ssh_error_also_retains_vm(self):
|
||||||
|
ex = self._make_exec()
|
||||||
|
self._mark_vm_running(ex)
|
||||||
|
self.raise_on_push = True
|
||||||
|
out = ex.recycle_vm(self._task())
|
||||||
|
self.assertEqual(out["state"], "sync_failed")
|
||||||
|
self.assertTrue(out["vm_retained"])
|
||||||
|
self.assertFalse(self.proc.terminated)
|
||||||
|
self.assertIn("vm_recycle_sync_failed", self._audit_events())
|
||||||
|
|
||||||
|
def test_sync_failed_then_retry_after_push_ok(self):
|
||||||
|
# 人工/重试路径:push 修好后再次回收即成功,不卡死。
|
||||||
|
ex = self._make_exec()
|
||||||
|
self._mark_vm_running(ex)
|
||||||
|
self.push_rc = 1
|
||||||
|
out = ex.recycle_vm(self._task())
|
||||||
|
self.assertEqual(out["state"], "sync_failed")
|
||||||
|
self.push_rc = 0
|
||||||
|
out = ex.recycle_vm(self._task())
|
||||||
|
self.assertEqual(out["state"], "recycled")
|
||||||
|
self.assertTrue(self.proc.terminated)
|
||||||
|
|
||||||
|
def test_sync_skip_flag_recycles_despite_push_failure(self):
|
||||||
|
ex = self._make_exec()
|
||||||
|
self._mark_vm_running(ex)
|
||||||
|
self.push_rc = 1
|
||||||
|
out = ex.recycle_vm(self._task(), sync=False)
|
||||||
|
self.assertEqual(out["state"], "recycled")
|
||||||
|
self.assertTrue(self.proc.terminated)
|
||||||
|
|
||||||
|
def test_vm_not_running_does_not_block_cleanup(self):
|
||||||
|
# VM 已死:没有活工作树可推,不拿死 VM 卡清理。
|
||||||
|
ex = self._make_exec()
|
||||||
|
out = ex.recycle_vm(self._task())
|
||||||
|
self.assertEqual(out["state"], "recycled")
|
||||||
|
|
||||||
|
def test_recycle_persistent_vm_sync_failed_keeps_disks(self):
|
||||||
|
ex = self._make_exec()
|
||||||
|
self._mark_vm_running(ex, vm_name="chv-persist-proj1")
|
||||||
|
self.push_rc = 1
|
||||||
|
out = ex.recycle_persistent_vm({"id": "chv-persist-proj1"})
|
||||||
|
self.assertEqual(out["state"], "sync_failed")
|
||||||
|
self.assertEqual(out["status"], "sync_failed")
|
||||||
|
self.assertTrue(out["vm_retained"])
|
||||||
|
self.assertEqual(out["deleted_disks"], [])
|
||||||
|
self.assertFalse(self.proc.terminated)
|
||||||
|
self.assertNotIn("vm_persistent_recycled",
|
||||||
|
self._audit_events())
|
||||||
|
self.assertIn("vm_recycle_sync_failed", self._audit_events())
|
||||||
|
|
||||||
|
def test_recycle_vm_missing_id_still_raises(self):
|
||||||
|
ex = self._make_exec()
|
||||||
|
with self.assertRaises(ValueError):
|
||||||
|
ex.recycle_vm({})
|
||||||
|
|
||||||
|
|
||||||
|
# -- 13. 审计事件目录 ------------------------------------------------------
|
||||||
|
class HybridEventsTestCase(unittest.TestCase):
|
||||||
|
def test_design_event_names_verbatim(self):
|
||||||
|
# 事件名与 DESIGN-HYBRID.md §9.1-13 原文一字不差。
|
||||||
|
self.assertEqual(hybrid_events.HYBRID_EVENTS, (
|
||||||
|
"warm_pool_acquire",
|
||||||
|
"warm_pool_miss",
|
||||||
|
"warm_pool_release",
|
||||||
|
"vm_resident_suspended",
|
||||||
|
"vm_resident_demoted",
|
||||||
|
"vm_resident_promoted",
|
||||||
|
"resident_upgrade_hint",
|
||||||
|
"snapshot_tenant_mismatch",
|
||||||
|
"vm_restore_denied",
|
||||||
|
))
|
||||||
|
|
||||||
|
def test_every_event_requires_host_id(self):
|
||||||
|
for event in hybrid_events.HYBRID_EVENTS:
|
||||||
|
self.assertIn("host_id",
|
||||||
|
hybrid_events.REQUIRED_FIELDS[event],
|
||||||
|
event)
|
||||||
|
self.assertIn(
|
||||||
|
"host_id",
|
||||||
|
hybrid_events.REQUIRED_FIELDS[
|
||||||
|
hybrid_events.VM_RECYCLE_SYNC_FAILED])
|
||||||
|
|
||||||
|
def test_build_event_defaults_host_id(self):
|
||||||
|
fields = hybrid_events.build_event(
|
||||||
|
"warm_pool_miss", {"tenant_id": "t1"})
|
||||||
|
self.assertEqual(fields["host_id"], socket.gethostname())
|
||||||
|
self.assertEqual(fields["tenant_id"], "t1")
|
||||||
|
|
||||||
|
def test_build_event_rejects_missing_required(self):
|
||||||
|
with self.assertRaises(ValueError):
|
||||||
|
hybrid_events.build_event("warm_pool_acquire", {})
|
||||||
|
with self.assertRaises(ValueError):
|
||||||
|
hybrid_events.build_event(
|
||||||
|
"vm_resident_suspended",
|
||||||
|
{"host_id": "h1", "tenant_id": "t1"})
|
||||||
|
with self.assertRaises(ValueError):
|
||||||
|
hybrid_events.build_event("no_such_event",
|
||||||
|
{"host_id": "h1"})
|
||||||
|
|
||||||
|
def test_secret_field_rejected_without_leaking_value(self):
|
||||||
|
sentinel = "SECRET-TOKEN-XYZ-DO-NOT-LEAK"
|
||||||
|
with self.assertRaises(ValueError) as ctx:
|
||||||
|
hybrid_events.build_event(
|
||||||
|
"warm_pool_acquire",
|
||||||
|
{"vm": "chv-x", "GITEA_TOKEN": sentinel})
|
||||||
|
self.assertNotIn(sentinel, str(ctx.exception))
|
||||||
|
self.assertIn("GITEA_TOKEN", str(ctx.exception))
|
||||||
|
|
||||||
|
def test_format_detail_has_host_id_and_no_secret(self):
|
||||||
|
detail = hybrid_events.format_detail(
|
||||||
|
"vm_resident_promoted",
|
||||||
|
{"tenant_id": "t1", "vm": "chv-persist-t1"})
|
||||||
|
self.assertIn("host_id=%s" % socket.gethostname(), detail)
|
||||||
|
self.assertIn("vm=chv-persist-t1", detail)
|
||||||
|
self.assertIn("tenant_id=t1", detail)
|
||||||
|
|
||||||
|
|
||||||
|
# -- 10. orphans facade 核实 ----------------------------------------------
|
||||||
|
class OrphansFacadeTestCase(unittest.TestCase):
|
||||||
|
def test_facade_exists_and_cleans_only_orphans(self):
|
||||||
|
# facade 已存在:读 registry → 只清理 warm/assigned
|
||||||
|
# 条目(只清理、不分配),非法条目丢弃不连坐。
|
||||||
|
with tempfile.TemporaryDirectory() as tmp:
|
||||||
|
registry = {
|
||||||
|
"warm-1": {"state": "warm", "vm": "chv-a",
|
||||||
|
"task_id": "t1"},
|
||||||
|
"asg-1": {"state": "assigned", "vm": "chv-b",
|
||||||
|
"task_id": "t2"},
|
||||||
|
"bad-1": {"state": "warm"},
|
||||||
|
"dead-1": {"state": "recycled", "vm": "chv-c"},
|
||||||
|
}
|
||||||
|
with open(os.path.join(tmp, "warm-pool.json"),
|
||||||
|
"w") as fh:
|
||||||
|
json.dump(registry, fh)
|
||||||
|
|
||||||
|
class Exec:
|
||||||
|
state_dir = tmp
|
||||||
|
cleaned = []
|
||||||
|
|
||||||
|
def shutdown_vm(self, vm):
|
||||||
|
self.cleaned.append(vm)
|
||||||
|
return True
|
||||||
|
|
||||||
|
def _emit(self, event, task, detail, kind=None):
|
||||||
|
pass
|
||||||
|
|
||||||
|
ex = Exec()
|
||||||
|
cleaned = reconcile_orphans_for_executor(ex)
|
||||||
|
self.assertEqual(cleaned, ["chv-a", "chv-b"])
|
||||||
|
self.assertEqual(ex.cleaned, ["chv-a", "chv-b"])
|
||||||
|
with open(os.path.join(tmp, "warm-pool.json")) as fh:
|
||||||
|
self.assertEqual(json.load(fh), {})
|
||||||
|
|
||||||
|
def test_facade_quarantines_corrupt_registry(self):
|
||||||
|
with tempfile.TemporaryDirectory() as tmp:
|
||||||
|
path = os.path.join(tmp, "warm-pool.json")
|
||||||
|
with open(path, "w") as fh:
|
||||||
|
fh.write("{not json")
|
||||||
|
|
||||||
|
class Exec:
|
||||||
|
state_dir = tmp
|
||||||
|
|
||||||
|
def shutdown_vm(self, vm):
|
||||||
|
raise AssertionError("must not be called")
|
||||||
|
|
||||||
|
self.assertEqual(reconcile_orphans_for_executor(Exec()), [])
|
||||||
|
self.assertTrue(os.path.exists(path + ".corrupt"))
|
||||||
|
|
||||||
|
def test_warm_pool_still_not_wired_to_dispatch(self):
|
||||||
|
from cloud.chv.experimental_pool import WarmPool
|
||||||
|
self.assertIs(WarmPool.wired_to_dispatch, False)
|
||||||
|
|
||||||
|
def test_orphans_module_does_not_import_experimental_pool(self):
|
||||||
|
# 生产 facade 不 import 实验模块(ADR-12):全新解释器
|
||||||
|
# 里只 import orphans,experimental_pool 不得被加载。
|
||||||
|
code = (
|
||||||
|
"import cloud.chv.orphans, sys;"
|
||||||
|
" assert 'cloud.chv.experimental_pool' not in sys.modules;"
|
||||||
|
" print('orphans-clean')"
|
||||||
|
)
|
||||||
|
proc = subprocess.run(
|
||||||
|
[sys.executable, "-c", code],
|
||||||
|
capture_output=True, text=True, cwd=str(
|
||||||
|
Path(__file__).resolve().parents[2]))
|
||||||
|
self.assertEqual(proc.returncode, 0, proc.stderr)
|
||||||
|
self.assertIn("orphans-clean", proc.stdout)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
unittest.main()
|
||||||
Reference in new issue
Block a user