125 lines
5.0 KiB
Python
125 lines
5.0 KiB
Python
from datetime import datetime
|
|
|
|
from knaster.healer.engine import HealerConfig
|
|
from knaster.policy import Policy
|
|
from knaster.util import fill_template, parse_duration
|
|
|
|
|
|
def test_policy_modes_and_windows():
|
|
p = Policy(**{"default": "approve", "actions": {
|
|
"a": {"mode": "auto"}, "night": {"mode": "approve", "window": "22:00-04:00"},
|
|
"day": {"mode": "auto", "window": "08:00-17:00"}}})
|
|
assert p.decide("a") == "auto" and p.decide("unknown") == "approve"
|
|
assert p.in_window("night", datetime(2026, 1, 1, 23, 0))
|
|
assert p.in_window("night", datetime(2026, 1, 1, 3, 0))
|
|
assert not p.in_window("night", datetime(2026, 1, 1, 12, 0))
|
|
assert p.in_window("day", datetime(2026, 1, 1, 9, 0))
|
|
assert p.in_window("a")
|
|
|
|
|
|
def test_repo_configs_load():
|
|
from knaster.settings import REPO_ROOT
|
|
|
|
cfg = HealerConfig.from_file(REPO_ROOT / "config" / "healer-rules.yaml")
|
|
assert {r.id for r in cfg.rules} >= {"slot-endpoint-down", "gpu-xid"}
|
|
assert cfg.enabled_actions == [] # sicherer Default
|
|
pol = Policy.from_file(REPO_ROOT / "config" / "policy.yaml")
|
|
assert pol.decide("promote_candidate") == "approve" and pol.decide("demote_skill") == "auto"
|
|
|
|
|
|
def test_util():
|
|
assert parse_duration("5m") == 300 and parse_duration(10) == 10 and parse_duration("2h") == 7200
|
|
assert fill_template({"a": "{x}-{y}", "b": ["{x}"]}, {"x": "1", "y": "2"}) == {"a": "1-2", "b": ["1"]}
|
|
|
|
|
|
def alert(status="firing", **labels):
|
|
base = {"alertname": "SlotEndpointDown", "slot": "tier1-general", "docker_host": "gpu-main",
|
|
"container": "vllm", "pve_node": "pve1", "vmid": "201"}
|
|
base.update(labels)
|
|
return {"status": status, "labels": base, "fingerprint": "fp1"}
|
|
|
|
|
|
def test_healer_ladder_cooldown_and_resolve(make_system, clock):
|
|
sysm = make_system()
|
|
h = sysm.healer
|
|
dry = sysm.actions.dry
|
|
out = h.handle_alert(alert())
|
|
assert (out.status, out.action) == ("executed", "restart_container")
|
|
assert out.message.startswith("[dry-run]")
|
|
assert h.handle_alert(alert()).status == "cooldown"
|
|
clock.advance(301)
|
|
assert h.handle_alert(alert()).action == "restart_container"
|
|
clock.advance(301)
|
|
assert h.handle_alert(alert()).action == "restart_vm"
|
|
clock.advance(901)
|
|
out = h.handle_alert(alert())
|
|
assert out.action == "mark_slot"
|
|
slot = sysm.registry.slot("tier1-general")
|
|
assert slot.status == "degraded" and slot.set_by == "healer"
|
|
out = h.handle_alert(alert())
|
|
assert out.status == "exhausted"
|
|
assert h.handle_alert(alert()).message == "bereits gemeldet"
|
|
assert [c[0] for c in dry.calls].count("restart_container") == 2
|
|
|
|
out = h.handle_alert(alert(status="resolved"))
|
|
assert out.status == "resolved"
|
|
assert sysm.registry.slot("tier1-general").status == "ok"
|
|
|
|
|
|
def test_healer_does_not_clear_human_degrade(make_system):
|
|
sysm = make_system()
|
|
sysm.registry.set_status("tier1-general", "degraded", set_by="human")
|
|
sysm.healer.handle_alert(alert())
|
|
sysm.healer.handle_alert(alert(status="resolved"))
|
|
assert sysm.registry.slot("tier1-general").status == "degraded"
|
|
|
|
|
|
def test_healer_approval_step_creates_proposal(make_system, clock):
|
|
sysm = make_system()
|
|
a = alert(alertname="GpuXidError")
|
|
assert sysm.healer.handle_alert(a).action == "mark_slot"
|
|
assert sysm.healer.handle_alert(a).action == "restart_vm"
|
|
clock.advance(1201)
|
|
out = sysm.healer.handle_alert(a)
|
|
assert out.status == "proposed" and out.action == "reboot_host"
|
|
props = sysm.maintenance.proposals()
|
|
assert props[0].kind == "reboot_host" and props[0].source == "healer"
|
|
|
|
|
|
def test_healer_missing_label_and_unmatched(make_system):
|
|
sysm = make_system()
|
|
a = alert()
|
|
del a["labels"]["container"]
|
|
out = sysm.healer.handle_alert(a)
|
|
assert out.status == "skipped" and "container" in out.message
|
|
assert sysm.healer.handle_alert({"status": "firing", "labels": {"alertname": "Neu"}}).status == "unmatched"
|
|
|
|
|
|
def test_healer_throttle(make_system, clock):
|
|
sysm = make_system()
|
|
sysm.healer.config.max_actions_per_hour = 2
|
|
for i in range(2):
|
|
sysm.healer.handle_alert({**alert(), "fingerprint": f"fp{i}"})
|
|
out = sysm.healer.handle_alert({**alert(), "fingerprint": "fp9"})
|
|
assert out.status == "throttled"
|
|
|
|
|
|
def test_infra_actions_need_explicit_enable(make_system):
|
|
from knaster.healer.drivers import DryRunDriver
|
|
|
|
real = DryRunDriver()
|
|
sysm = make_system(infra=real)
|
|
sysm.actions.dry_run = False
|
|
sysm.healer.handle_alert(alert())
|
|
assert real.calls == [] # nicht in enabled_actions
|
|
sysm.actions.enabled_infra_actions = {"restart_container"}
|
|
sysm.healer.handle_alert({**alert(), "fingerprint": "other"})
|
|
assert real.calls[0][0] == "restart_container"
|
|
|
|
|
|
def test_alertmanager_payload(make_system):
|
|
sysm = make_system()
|
|
outs = sysm.healer.handle_alertmanager({"version": "4", "alerts": [alert(), alert(alertname="NodeDown",
|
|
instance="pve2")]})
|
|
assert [o.status for o in outs] == ["executed", "executed"]
|