feat(chv): hybrid executor core — routing, clean pool, data disk, snapshot tenant check
Track A (§9.1 items 1-4, DESIGN-HYBRID.md): - dispatch_task_hybrid: route by tier + lease state (resident resume/cold vs pool hit/cold/queue, 300s queue timeout -> host_capacity_exceeded) - clean pool prewarm split from prewarm_persistent_vm: tenant-less birth (no data disk, no seed, no Gitea sync wait), binding deferred to acquire; warm_pool_sync_incomplete fail-closed at allocation - tenant data disk mount at acquire (attach_tenant_data_disk, PUT /vm.add-disk + chv-data-mount) and unmount+sync at release; _verify_task_disk_source invariant: task disk must be the tenant's data disk, never the clean snapshot disk - restore_vm validates tenant_id before restore: unregistered/unmarked snapshot denied in strict (hybrid) mode, mismatch -> snapshot_tenant_ mismatch fail-closed, audit vm_restore_denied (host_id + tenant only) - settled values: 2 warm per host, pool cap 50% of low-freq peak, resident cap 50% of VM slots, 1 snapshot per tenant per 30 days Tests: cloud/tests/test_chv_hybrid_a.py (27 tests, all passing) Full cloud suite: 4606 tests, 0 failures, 0 errors, 15 skipped (TMPDIR on /home; ~91 min, slowed by concurrent track-C suite)
This commit is contained in:
host committed
1 parent
3cdfd0f79e
commit
172accb2ec
2 files changed
+1707
-64
No files matched your search
+1019
-64
File diff suppressed because it is too large.
Load diff
@@ -0,0 +1,688 @@
|
||||
"""CHV 混合模式 A 轨（feat-chv-hybrid-a）测试。
|
||||
|
||||
只用标准库与 mock runner/launcher/api_client，绝不启动真实
|
||||
VM（与 test_chv_opt_stage1.py 同款边界）。覆盖 DESIGN-HYBRID.md
|
||||
§9.1 第 1–4 项：
|
||||
- 混合路由分流（常驻 / 池命中 / 池空冷启 / 排队拒绝）
|
||||
- 干净池预热不带租户（无数据盘、无 seed、无同步等待）
|
||||
- 数据盘挂载归属 + “任务盘来源”不变量
|
||||
- snapshot_tenant_mismatch（含无 tenant_id 登记拒绝）
|
||||
- 已定值常量（每宿主预热 2 台、池上限 50%、常驻上限 50%、
|
||||
排队超时 300 秒、热恢复 ≤2 秒、冷恢复 ≤30 秒、快照 1 个/30 天）
|
||||
|
||||
不改其它测试文件；阶段 1 的旧语义（prewarm_persistent_vm /
|
||||
acquire_for_task / dispatch_task_persistent / _validate_restore
|
||||
的 legacy 警告放行）原样保留，只测本轨新增行为。
|
||||
"""
|
||||
|
||||
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,
|
||||
read_vm_tenant,
|
||||
record_snapshot,
|
||||
write_persistent_vm,
|
||||
)
|
||||
from cloud.chv.executor import (
|
||||
HYBRID_COLD_RESTORE_TARGET_SECONDS,
|
||||
HYBRID_QUEUE_TIMEOUT_SECONDS,
|
||||
HYBRID_SNAPSHOT_PER_TENANT,
|
||||
HYBRID_SNAPSHOT_RETENTION_DAYS,
|
||||
HYBRID_WARM_POOL_PER_HOST,
|
||||
HYBRID_WARM_RESUME_TARGET_SECONDS,
|
||||
)
|
||||
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,
|
||||
read_vm_tenant,
|
||||
record_snapshot,
|
||||
write_persistent_vm,
|
||||
)
|
||||
from cloud.chv.executor import (
|
||||
HYBRID_COLD_RESTORE_TARGET_SECONDS,
|
||||
HYBRID_QUEUE_TIMEOUT_SECONDS,
|
||||
HYBRID_SNAPSHOT_PER_TENANT,
|
||||
HYBRID_SNAPSHOT_RETENTION_DAYS,
|
||||
HYBRID_WARM_POOL_PER_HOST,
|
||||
HYBRID_WARM_RESUME_TARGET_SECONDS,
|
||||
)
|
||||
|
||||
|
||||
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 HybridACase(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.events = []
|
||||
|
||||
def tearDown(self):
|
||||
self.tmp.cleanup()
|
||||
|
||||
def make_exec(self, runner=None, api_client=None, launcher=None,
|
||||
env_extra=None, **overrides):
|
||||
env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888",
|
||||
"CHV_HOST_KEY_MODE": "keyscan",
|
||||
"CHV_SSH_FORWARD": "1",
|
||||
"CHV_WARM_POOL_SIZE": "2"}
|
||||
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=lambda s: None, env=env)
|
||||
kwargs.update(overrides)
|
||||
ex = CloudHypervisorExecutor(**kwargs)
|
||||
ex.audit_callback = lambda e, t, d, k: self.events.append(
|
||||
(e, d, k))
|
||||
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, "", "")
|
||||
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", tier=None):
|
||||
task = {"id": task_id, "prompt": "做一个加法函数",
|
||||
"project_id": project}
|
||||
if tier is not None:
|
||||
task["tier"] = tier
|
||||
return task
|
||||
|
||||
def _socket(self, vm):
|
||||
os.makedirs(self.state_dir, exist_ok=True)
|
||||
path = os.path.join(self.state_dir, vm + ".sock")
|
||||
with open(path, "w") as fh:
|
||||
fh.write("")
|
||||
return path
|
||||
|
||||
def _event_names(self):
|
||||
return [e for e, _d, _k in self.events]
|
||||
|
||||
def _event_details(self, name):
|
||||
return [d for e, d, _k in self.events if e == name]
|
||||
|
||||
|
||||
# -- 已定值常量 -------------------------------------------------------------
|
||||
class SettledValuesTest(HybridACase):
|
||||
def test_hybrid_settled_values(self):
|
||||
ex = self.make_exec()
|
||||
self.assertEqual(HYBRID_WARM_POOL_PER_HOST, 2)
|
||||
self.assertEqual(ex.warm_pool_per_host_target(), 2)
|
||||
self.assertEqual(ex.clean_pool_target(), 2)
|
||||
# 池上限 = 低频并发峰值的 50%：20 → 10，50 → 25。
|
||||
self.assertEqual(ex.warm_pool_cap(20), 10)
|
||||
self.assertEqual(ex.warm_pool_cap(50), 25)
|
||||
self.assertIsNone(ex.warm_pool_cap(None))
|
||||
# 生效目标取 min（配置目标，上限）：目标 2、峰值 20 → 2。
|
||||
self.assertEqual(ex.clean_pool_effective_target(20), 2)
|
||||
ex_big = self.make_exec(env_extra={
|
||||
"CHV_WARM_POOL_SIZE": "30",
|
||||
"CHV_LOW_FREQ_PEAK_CONCURRENCY": "20"})
|
||||
self.assertEqual(ex_big.clean_pool_effective_target(), 10)
|
||||
# 常驻上限 = 槽位 50%：4 槽 → 2 台，48 槽 → 24 台。
|
||||
self.assertEqual(ex.resident_cap(), 2)
|
||||
ex48 = self.make_exec(env_extra={"CHV_MAX_VMS": "48"})
|
||||
self.assertEqual(ex48.resident_cap(), 24)
|
||||
# 排队超时 300 秒；热恢复 ≤2 秒、冷快照恢复 ≤30 秒（目标值，
|
||||
# 只作常量/注释，不在单元测试里硬测时延）；快照 1 个/30 天。
|
||||
self.assertEqual(HYBRID_QUEUE_TIMEOUT_SECONDS, 300)
|
||||
self.assertEqual(HYBRID_WARM_RESUME_TARGET_SECONDS, 2)
|
||||
self.assertEqual(HYBRID_COLD_RESTORE_TARGET_SECONDS, 30)
|
||||
self.assertEqual(HYBRID_SNAPSHOT_PER_TENANT, 1)
|
||||
self.assertEqual(HYBRID_SNAPSHOT_RETENTION_DAYS, 30)
|
||||
|
||||
|
||||
# -- 混合路由分流 -----------------------------------------------------------
|
||||
class HybridRouteTest(HybridACase):
|
||||
def test_route_by_tier(self):
|
||||
ex = self.make_exec()
|
||||
self.assertEqual(
|
||||
ex._hybrid_route(self._task(tier="paid")), "resident")
|
||||
self.assertEqual(
|
||||
ex._hybrid_route(self._task(tier="heavy")), "resident")
|
||||
self.assertEqual(
|
||||
ex._hybrid_route(self._task(tier="PAID")), "resident")
|
||||
self.assertEqual(
|
||||
ex._hybrid_route(self._task(tier="free")), "pool")
|
||||
self.assertEqual(ex._hybrid_route(self._task()), "pool")
|
||||
# 常驻是权益门：Paid 但无 tenant（无归属）即 fail-closed 走池。
|
||||
task = {"id": "x1", "prompt": "p", "tier": "paid"}
|
||||
self.assertEqual(ex._hybrid_route(task), "pool")
|
||||
|
||||
def test_dispatch_hybrid_validates_task(self):
|
||||
ex = self.make_exec()
|
||||
with self.assertRaises(ValueError):
|
||||
ex.dispatch_task_hybrid({"prompt": "p"})
|
||||
with self.assertRaises(ValueError):
|
||||
ex.dispatch_task_hybrid({"id": "x"})
|
||||
|
||||
def test_dispatch_hybrid_pool_disabled_falls_back(self):
|
||||
ex = self.make_exec(env_extra={"CHV_WARM_POOL_SIZE": "0"})
|
||||
out = ex.dispatch_task_hybrid(self._task())
|
||||
self.assertEqual(out["status"], "completed")
|
||||
# 退化为阶段 1 旧语义：结果不带 route/allocation。
|
||||
self.assertNotIn("route", out)
|
||||
self.assertEqual(ex._vms, {})
|
||||
|
||||
def test_dispatch_pool_cold_when_pool_empty(self):
|
||||
ex = self.make_exec()
|
||||
out = ex.dispatch_task_hybrid(self._task(task_id="t1"))
|
||||
self.assertEqual(out["status"], "completed")
|
||||
self.assertEqual(out["route"], "pool")
|
||||
self.assertEqual(out["allocation"], "pool-cold")
|
||||
self.assertEqual(out["vm_mode"], "cold")
|
||||
self.assertIn("warm_pool_miss", self._event_names())
|
||||
self.assertEqual(ex._vms, {})
|
||||
|
||||
def test_dispatch_pool_queue_timeout_refuses(self):
|
||||
ex = self.make_exec(env_extra={
|
||||
"CHV_QUEUE_TIMEOUT": "0", "CHV_MAX_VMS": "1"})
|
||||
# 宿主已满（1 台活 VM 占满唯一槽位），池空 → 排队 0 秒即拒绝。
|
||||
ex._vms["chv-busy"] = FakeProc(alive=True)
|
||||
with self.assertRaises(ChvFailure) as ctx:
|
||||
ex.dispatch_task_hybrid(self._task(task_id="t1"))
|
||||
self.assertEqual(ctx.exception.kind, "host_capacity_exceeded")
|
||||
self.assertEqual(self.launches, [])
|
||||
self.assertIn("warm_pool_miss", self._event_names())
|
||||
|
||||
def test_dispatch_resident_cold_when_no_resident_vm(self):
|
||||
ex = self.make_exec()
|
||||
out = ex.dispatch_task_hybrid(
|
||||
self._task(task_id="t1", project="paid1", tier="paid"))
|
||||
self.assertEqual(out["status"], "completed")
|
||||
self.assertEqual(out["route"], "resident")
|
||||
self.assertEqual(out["allocation"], "resident-cold")
|
||||
self.assertEqual(out["vm"], "chv-persist-paid1")
|
||||
rec = read_persistent_vm(self.state_dir, "paid1")
|
||||
self.assertEqual(rec["state"], "paused")
|
||||
self.assertEqual(rec["tier"], "paid")
|
||||
self.assertTrue(rec["resident"])
|
||||
self.assertEqual(ex._resident_count(), 1)
|
||||
|
||||
def test_dispatch_resident_resumed_from_prewarm(self):
|
||||
ex = self.make_exec()
|
||||
ex.prewarm_persistent_vm("paid1")
|
||||
self._socket("chv-persist-paid1")
|
||||
self.vm_states["chv-persist-paid1"] = "Paused"
|
||||
out = ex.dispatch_task_hybrid(
|
||||
self._task(task_id="t1", project="paid1", tier="paid"))
|
||||
self.assertEqual(out["status"], "completed")
|
||||
self.assertEqual(out["route"], "resident")
|
||||
self.assertEqual(out["allocation"], "resident-resumed")
|
||||
self.assertEqual(out["vm"], "chv-persist-paid1")
|
||||
self.assertIn("resume_seconds", out)
|
||||
rec = read_persistent_vm(self.state_dir, "paid1")
|
||||
self.assertEqual(rec["state"], "paused")
|
||||
self.assertEqual(rec["tier"], "paid")
|
||||
self.assertTrue(rec["resident"])
|
||||
|
||||
def test_resident_cap_blocks_new_resident(self):
|
||||
ex = self.make_exec(env_extra={"CHV_QUEUE_TIMEOUT": "0"})
|
||||
# 4 槽位 → 常驻上限 2 台：已有 2 个常驻租户，第 3 个排队即拒。
|
||||
for tenant in ("r1", "r2"):
|
||||
write_persistent_vm(self.state_dir, tenant, {
|
||||
"vm": "chv-persist-%s" % tenant,
|
||||
"tenant_id": tenant, "state": "paused",
|
||||
"origin": "task", "gitea_synced": False,
|
||||
"tier": "paid", "resident": True})
|
||||
self.assertEqual(ex._resident_count(), 2)
|
||||
with self.assertRaises(ChvFailure) as ctx:
|
||||
ex.dispatch_task_hybrid(
|
||||
self._task(task_id="t1", project="r3", tier="paid"))
|
||||
self.assertEqual(ctx.exception.kind, "host_capacity_exceeded")
|
||||
self.assertEqual(self.launches, [])
|
||||
# 已有常驻的租户不受上限挡（自己的槽位）。
|
||||
ex2 = self.make_exec(env_extra={"CHV_QUEUE_TIMEOUT": "0"})
|
||||
write_persistent_vm(self.state_dir, "r1", {
|
||||
"vm": "chv-persist-r1", "tenant_id": "r1",
|
||||
"state": "paused", "origin": "task",
|
||||
"gitea_synced": False, "tier": "paid", "resident": True})
|
||||
write_persistent_vm(self.state_dir, "r2", {
|
||||
"vm": "chv-persist-r2", "tenant_id": "r2",
|
||||
"state": "paused", "origin": "task",
|
||||
"gitea_synced": False, "tier": "paid", "resident": True})
|
||||
self._socket("chv-persist-r1")
|
||||
self.vm_states["chv-persist-r1"] = "Paused"
|
||||
out = ex2.dispatch_task_hybrid(
|
||||
self._task(task_id="t1", project="r1", tier="paid"))
|
||||
self.assertEqual(out["route"], "resident")
|
||||
self.assertEqual(out["allocation"], "resident-resumed")
|
||||
|
||||
|
||||
# -- 干净池预热 -------------------------------------------------------------
|
||||
class CleanPoolPrewarmTest(HybridACase):
|
||||
def test_prewarm_clean_pool_has_no_tenant_state(self):
|
||||
ex = self.make_exec()
|
||||
out = ex.prewarm_clean_pool()
|
||||
self.assertTrue(out["pooled"])
|
||||
self.assertEqual(out["mode"], "prewarmed")
|
||||
self.assertEqual(out["vm"], "chv-pool-0")
|
||||
self.assertFalse(out["gitea_synced"])
|
||||
self.assertTrue(out["clean"])
|
||||
self.assertIsNone(out["tenant_id"])
|
||||
self.assertEqual(len(self.launches), 1)
|
||||
vm_dir = os.path.join(self.state_dir, "chv-pool-0")
|
||||
# 出生时不带租户：无数据盘、无 tenant 标记、无 seed。
|
||||
self.assertFalse(os.path.exists(
|
||||
os.path.join(vm_dir, "data.raw")))
|
||||
self.assertIsNone(read_vm_tenant(self.state_dir, "chv-pool-0"))
|
||||
self.assertFalse(os.path.exists(
|
||||
os.path.join(vm_dir, "seed.iso")))
|
||||
self.assertFalse(os.path.exists(
|
||||
os.path.join(vm_dir, "seed")))
|
||||
# 无 Gitea 同步等待：预热全程没有任何 ssh test -d 调用。
|
||||
self.assertFalse(any(
|
||||
"test" in call for call in self.calls
|
||||
if call and call[0] == "ssh"))
|
||||
# 登记同样不带租户。
|
||||
record = ex._read_clean_pool_record(0)
|
||||
self.assertEqual(record["state"], "paused")
|
||||
self.assertIsNone(record["tenant_id"])
|
||||
self.assertFalse(record["gitea_synced"])
|
||||
# socket 造出来后第二次即 already-warm，不重建。
|
||||
self._socket("chv-pool-0")
|
||||
second = ex.prewarm_clean_pool()
|
||||
self.assertEqual(second["mode"], "already-warm")
|
||||
self.assertEqual(len(self.launches), 1)
|
||||
|
||||
def test_prewarm_clean_pool_slots_and_range(self):
|
||||
ex = self.make_exec()
|
||||
first = ex.prewarm_clean_pool(slot=1)
|
||||
self.assertEqual(first["vm"], "chv-pool-1")
|
||||
out = ex.prewarm_clean_pool(slot=5)
|
||||
self.assertFalse(out["pooled"])
|
||||
self.assertEqual(out["reason"], "clean_pool_slot_out_of_range")
|
||||
self.assertEqual(len(self.launches), 1)
|
||||
|
||||
def test_prewarm_clean_pool_disabled_when_size_zero(self):
|
||||
ex = self.make_exec(env_extra={"CHV_WARM_POOL_SIZE": "0"})
|
||||
self.assertEqual(
|
||||
ex.prewarm_clean_pool()["reason"], "warm_pool_disabled")
|
||||
self.assertEqual(
|
||||
ex.acquire_clean_pool(self._task())["mode"], "cold")
|
||||
self.assertEqual(ex.refill_clean_pool(), {"target": 0,
|
||||
"slots": []})
|
||||
|
||||
def test_prewarm_clean_pool_refuses_tenant_seed(self):
|
||||
ex = self.make_exec()
|
||||
with self.assertRaises(ChvFailure) as ctx:
|
||||
ex.prewarm_clean_pool(
|
||||
extra_env={"GITEA_TOKEN": "secret-value"})
|
||||
self.assertEqual(ctx.exception.kind, "clean_pool_seed_refused")
|
||||
with self.assertRaises(ChvFailure) as ctx:
|
||||
ex.prewarm_clean_pool(
|
||||
extra_env={"GITEA_URL": "https://gitea.tose.top"})
|
||||
self.assertEqual(ctx.exception.kind, "clean_pool_seed_refused")
|
||||
self.assertEqual(self.launches, [])
|
||||
|
||||
def test_prewarm_clean_pool_never_rebuilds_running_slot(self):
|
||||
ex = self.make_exec()
|
||||
ex._write_clean_pool_record(0, {
|
||||
"vm": "chv-pool-0", "slot": 0, "tenant_id": "tenant-a",
|
||||
"state": "running", "origin": "clean-prewarm",
|
||||
"gitea_synced": False, "data_disk": "x",
|
||||
"updated_at": 0})
|
||||
out = ex.prewarm_clean_pool(slot=0)
|
||||
self.assertFalse(out["pooled"])
|
||||
self.assertEqual(out["reason"], "clean_pool_slot_busy")
|
||||
self.assertEqual(self.launches, [])
|
||||
# 自动选槽跳过正跑的 slot 0，建 slot 1。
|
||||
out2 = ex.prewarm_clean_pool()
|
||||
self.assertEqual(out2["vm"], "chv-pool-1")
|
||||
self.assertEqual(len(self.launches), 1)
|
||||
|
||||
def test_refill_clean_pool(self):
|
||||
ex = self.make_exec()
|
||||
ex.prewarm_clean_pool(slot=0)
|
||||
self._socket("chv-pool-0")
|
||||
out = ex.refill_clean_pool()
|
||||
self.assertEqual(out["target"], 2)
|
||||
self.assertEqual(out["slots"][0]["mode"], "already-warm")
|
||||
self.assertEqual(out["slots"][1]["mode"], "prewarmed")
|
||||
self.assertEqual(len(self.launches), 2)
|
||||
|
||||
|
||||
# -- 数据盘挂载归属 + 任务盘来源不变量 ---------------------------------------
|
||||
class DataDiskAttachTest(HybridACase):
|
||||
def test_verify_task_disk_source(self):
|
||||
ex = self.make_exec()
|
||||
tenant = "tenant-a"
|
||||
disk = ex._provision_tenant_data_disk(tenant)
|
||||
self.assertEqual(disk, ex._tenant_data_disk_path(tenant))
|
||||
self.assertTrue(os.path.exists(disk))
|
||||
# 合法的租户数据盘通过。
|
||||
ex._verify_task_disk_source("chv-pool-0", tenant, disk)
|
||||
# 干净池 VM 自带的盘（干净快照盘）绝不能当任务盘。
|
||||
pool_data = os.path.join(self.state_dir, "chv-pool-0",
|
||||
"data.raw")
|
||||
with self.assertRaises(ChvFailure) as ctx:
|
||||
ex._verify_task_disk_source("chv-pool-0", tenant, pool_data)
|
||||
self.assertEqual(ctx.exception.kind, "task_disk_source_invalid")
|
||||
# 基础镜像不能当任务盘。
|
||||
with self.assertRaises(ChvFailure) as ctx:
|
||||
ex._verify_task_disk_source("chv-pool-0", tenant,
|
||||
self.rootfs)
|
||||
self.assertEqual(ctx.exception.kind, "task_disk_source_invalid")
|
||||
# 其它租户的盘不能当本租户的任务盘。
|
||||
other = ex._provision_tenant_data_disk("tenant-b")
|
||||
with self.assertRaises(ChvFailure) as ctx:
|
||||
ex._verify_task_disk_source("chv-pool-0", tenant, other)
|
||||
self.assertEqual(ctx.exception.kind, "task_disk_source_invalid")
|
||||
# 无盘 / 无租户同样拒绝。
|
||||
with self.assertRaises(ChvFailure) as ctx:
|
||||
ex._verify_task_disk_source("chv-pool-0", tenant, None)
|
||||
self.assertEqual(ctx.exception.kind, "task_disk_source_invalid")
|
||||
with self.assertRaises(ChvFailure) as ctx:
|
||||
ex._verify_task_disk_source("chv-pool-0", None, disk)
|
||||
self.assertEqual(ctx.exception.kind, "task_disk_source_invalid")
|
||||
|
||||
def test_attach_refuses_wrong_disk_before_api_call(self):
|
||||
ex = self.make_exec()
|
||||
with self.assertRaises(ChvFailure) as ctx:
|
||||
ex.attach_tenant_data_disk(
|
||||
self._task(project="tenant-a"), "chv-pool-0",
|
||||
data_disk=self.rootfs)
|
||||
self.assertEqual(ctx.exception.kind, "task_disk_source_invalid")
|
||||
# 不变量在 API 调用之前拦住：一次 add-disk 都没发。
|
||||
self.assertFalse(any(p == "/vm.add-disk"
|
||||
for _m, p, _pl in self.api_calls))
|
||||
|
||||
def test_attach_tenant_data_disk(self):
|
||||
ex = self.make_exec()
|
||||
out = ex.attach_tenant_data_disk(
|
||||
self._task(project="tenant-a"), "chv-pool-0")
|
||||
self.assertEqual(out["tenant_id"], "tenant-a")
|
||||
self.assertEqual(out["data_disk"],
|
||||
ex._tenant_data_disk_path("tenant-a"))
|
||||
self.assertTrue(any(p == "/vm.add-disk"
|
||||
for _m, p, _pl in self.api_calls))
|
||||
_method, _path, payload = next(
|
||||
(m, p, pl) for m, p, pl in self.api_calls
|
||||
if p == "/vm.add-disk")
|
||||
self.assertEqual(payload["path"], out["data_disk"])
|
||||
# 客内挂载脚本被调过（chv-data-mount）。
|
||||
self.assertTrue(any(
|
||||
"/usr/local/bin/chv-data-mount" in call
|
||||
for call in self.calls if call and call[0] == "ssh"))
|
||||
self.assertEqual(
|
||||
read_vm_tenant(self.state_dir, "chv-pool-0"), "tenant-a")
|
||||
|
||||
def test_dispatch_pool_hit_binds_tenant_disk_and_destroys_vm(self):
|
||||
ex = self.make_exec()
|
||||
ex.prewarm_clean_pool(slot=0)
|
||||
self._socket("chv-pool-0")
|
||||
self.vm_states["chv-pool-0"] = "Paused"
|
||||
out = ex.dispatch_task_hybrid(
|
||||
self._task(task_id="t1", project="tenant-a"))
|
||||
self.assertEqual(out["status"], "completed")
|
||||
self.assertEqual(out["route"], "pool")
|
||||
self.assertEqual(out["allocation"], "pool-hit")
|
||||
self.assertEqual(out["vm"], "chv-pool-0")
|
||||
self.assertTrue(out["warm_hit"])
|
||||
self.assertEqual(out["data_disk"],
|
||||
ex._tenant_data_disk_path("tenant-a"))
|
||||
self.assertIn("resume_seconds", out)
|
||||
self.assertIn("warm_pool_acquire", self._event_names())
|
||||
self.assertIn("warm_pool_release", self._event_names())
|
||||
# 用过的池 VM 已销毁、不回池：登记清掉、VMM 登记空、归属清掉。
|
||||
self.assertIsNone(ex._read_clean_pool_record(0))
|
||||
self.assertEqual(ex._vms, {})
|
||||
self.assertIsNone(read_vm_tenant(self.state_dir, "chv-pool-0"))
|
||||
# 租户数据盘还在（用户数据不随 VM 销毁而删）。
|
||||
self.assertTrue(os.path.exists(
|
||||
ex._tenant_data_disk_path("tenant-a")))
|
||||
# 池 VM 自带的干净盘从未被当成任务盘建出来。
|
||||
self.assertFalse(os.path.exists(os.path.join(
|
||||
self.state_dir, "chv-pool-0", "data.raw")))
|
||||
|
||||
def test_acquire_sync_incomplete_fails_closed_at_allocate(self):
|
||||
# 分配时同步等不到：fail-closed，VM 解绑销毁，不派任务。
|
||||
def runner(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 "test" in cmd:
|
||||
return FakeResult(1, "", "")
|
||||
if cmd[-1] == "true":
|
||||
return FakeResult(0, "", "")
|
||||
if "dispatch" in cmd:
|
||||
return FakeResult(0, json.dumps(
|
||||
{"status": "completed", "text": "ok"}), "")
|
||||
return FakeResult(0, "", "")
|
||||
return FakeResult(0, "", "")
|
||||
|
||||
ex = self.make_exec(runner=runner)
|
||||
ex.prewarm_clean_pool(slot=0)
|
||||
self._socket("chv-pool-0")
|
||||
self.vm_states["chv-pool-0"] = "Paused"
|
||||
task = self._task(task_id="t1", project="tenant-a")
|
||||
task["sync_workspace"] = "/home/chv/workspace/repo"
|
||||
with self.assertRaises(ChvFailure) as ctx:
|
||||
ex.dispatch_task_hybrid(task)
|
||||
self.assertEqual(ctx.exception.kind,
|
||||
"warm_pool_sync_incomplete")
|
||||
# 已绑过租户态的 VM 绝不回干净池：已销毁、登记清掉。
|
||||
self.assertIsNone(ex._read_clean_pool_record(0))
|
||||
self.assertEqual(ex._vms, {})
|
||||
self.assertFalse(any("dispatch" in call
|
||||
for call in self.calls
|
||||
if call and call[0] == "ssh"))
|
||||
|
||||
def test_release_sync_failure_keeps_vm(self):
|
||||
# 回收时 chv-sync push 失败：VM 不回收、不复用。
|
||||
def runner(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 "chv-sync" in cmd:
|
||||
return FakeResult(1, "", "push 被拒绝")
|
||||
if cmd[-1] == "true":
|
||||
return FakeResult(0, "", "")
|
||||
return FakeResult(0, "", "")
|
||||
return FakeResult(0, "", "")
|
||||
|
||||
ex = self.make_exec(runner=runner)
|
||||
ex.prewarm_clean_pool(slot=0)
|
||||
self._socket("chv-pool-0")
|
||||
self.vm_states["chv-pool-0"] = "Paused"
|
||||
acquired = ex.acquire_clean_pool(self._task(project="tenant-a"))
|
||||
self.assertEqual(acquired["mode"], "resumed")
|
||||
out = ex.release_pool_vm(
|
||||
self._task(project="tenant-a"), "chv-pool-0",
|
||||
sync_workspace="/home/chv/workspace/repo")
|
||||
self.assertFalse(out["released"])
|
||||
self.assertEqual(out["reason"], "sync_failed")
|
||||
self.assertIn("warm_pool_release_failed", self._event_names())
|
||||
# VM 还在（未销毁），登记仍是 running 绑定态。
|
||||
self.assertIn("chv-pool-0", ex._vms)
|
||||
record = ex._read_clean_pool_record(0)
|
||||
self.assertEqual(record["state"], "running")
|
||||
self.assertEqual(record["tenant_id"], "tenant-a")
|
||||
|
||||
|
||||
# -- 快照 tenant 校验 --------------------------------------------------------
|
||||
class SnapshotTenantStrictTest(HybridACase):
|
||||
def _snap_url(self, vm, name="snap1"):
|
||||
snap_dir = os.path.join(self.state_dir, vm, name)
|
||||
os.makedirs(snap_dir, exist_ok=True)
|
||||
with open(os.path.join(snap_dir, "config.json"), "w") as fh:
|
||||
fh.write("{}")
|
||||
with open(os.path.join(snap_dir, "state.json"), "w") as fh:
|
||||
fh.write("{}")
|
||||
return "file://" + snap_dir
|
||||
|
||||
def _record(self, vm, url, tenant):
|
||||
record_snapshot(self.state_dir, vm, url, 20.0,
|
||||
tenant_id=tenant)
|
||||
|
||||
def test_unmarked_snapshot_denied_in_hybrid_mode(self):
|
||||
url = self._snap_url("chv-a")
|
||||
self._record("chv-a", url, None)
|
||||
ex = self.make_exec() # CHV_WARM_POOL_SIZE=2 → 严格校验生效
|
||||
with self.assertRaises(ChvFailure) as ctx:
|
||||
ex.restore_vm({"id": "task-a", "project_id": "tenant-a"},
|
||||
url)
|
||||
self.assertEqual(ctx.exception.kind,
|
||||
"snapshot_tenant_mismatch")
|
||||
denied = [d for _e, d, _k in self.events
|
||||
if _e == "vm_restore_denied"]
|
||||
self.assertTrue(denied)
|
||||
# 审计带 host_id + 请求租户，不带快照内容（URL 不得出
|
||||
# 现在 detail 里）。
|
||||
self.assertIn("host_id=", denied[0])
|
||||
self.assertIn("tenant=tenant-a", denied[0])
|
||||
self.assertNotIn(url, denied[0])
|
||||
self.assertNotIn("snap1", denied[0])
|
||||
|
||||
def test_unregistered_snapshot_denied_in_hybrid_mode(self):
|
||||
url = self._snap_url("chv-a")
|
||||
ex = self.make_exec()
|
||||
with self.assertRaises(ChvFailure) as ctx:
|
||||
ex.restore_vm({"id": "task-a", "project_id": "tenant-a"},
|
||||
url)
|
||||
self.assertEqual(ctx.exception.kind,
|
||||
"snapshot_tenant_mismatch")
|
||||
self.assertIn("vm_restore_denied", self._event_names())
|
||||
|
||||
def test_mismatched_snapshot_denied_in_hybrid_mode(self):
|
||||
url = self._snap_url("chv-a")
|
||||
self._record("chv-a", url, "tenant-a")
|
||||
ex = self.make_exec()
|
||||
with self.assertRaises(ChvFailure) as ctx:
|
||||
ex.restore_vm({"id": "task-b", "project_id": "tenant-b"},
|
||||
url)
|
||||
self.assertEqual(ctx.exception.kind,
|
||||
"snapshot_tenant_mismatch")
|
||||
self.assertIn("vm_restore_denied", self._event_names())
|
||||
|
||||
def test_same_tenant_snapshot_restores_in_hybrid_mode(self):
|
||||
url = self._snap_url("chv-a")
|
||||
self._record("chv-a", url, "tenant-a")
|
||||
ex = self.make_exec()
|
||||
out = ex.restore_vm(
|
||||
{"id": "task-a", "project_id": "tenant-a"}, url)
|
||||
self.assertEqual(out["mode"], "restored")
|
||||
self.assertEqual(out["state"], "running")
|
||||
|
||||
def test_explicit_strict_flag_denies_unmarked(self):
|
||||
url = self._snap_url("chv-a")
|
||||
self._record("chv-a", url, None)
|
||||
ex = self.make_exec(env_extra={"CHV_WARM_POOL_SIZE": "0"})
|
||||
# 显式 strict_tenant=True 时即使池未启用也拒绝。
|
||||
with self.assertRaises(ChvFailure) as ctx:
|
||||
ex._validate_restore(
|
||||
{"id": "task-a", "project_id": "tenant-a"}, url,
|
||||
strict_tenant=True)
|
||||
self.assertEqual(ctx.exception.kind,
|
||||
"snapshot_tenant_mismatch")
|
||||
self.assertIn("vm_restore_denied", self._event_names())
|
||||
|
||||
def test_legacy_warn_pass_preserved_when_not_strict(self):
|
||||
# 混合模式未启用时，_validate_restore 沿用阶段 1 的
|
||||
# legacy 警告放行（旧测试依赖该语义，本轨不改）。
|
||||
url = self._snap_url("chv-a")
|
||||
self._record("chv-a", url, None)
|
||||
ex = self.make_exec(env_extra={"CHV_WARM_POOL_SIZE": "0"})
|
||||
snapshot_url, _net = ex._validate_restore(
|
||||
{"id": "task-a", "project_id": "tenant-a"}, url)
|
||||
self.assertEqual(snapshot_url, url)
|
||||
self.assertIn("vm_restore_tenant_unmarked",
|
||||
self._event_names())
|
||||
self.assertNotIn("vm_restore_denied", self._event_names())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user