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
@@ -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