466 lines
16 KiB
Python
466 lines
16 KiB
Python
from datetime import UTC, datetime, timedelta
|
||
|
||
from beaver_agent.hands.komodo import Komodo
|
||
from beaver_agent.hands.komodo_logs import (
|
||
KEEP,
|
||
SELF_MARK,
|
||
Seen,
|
||
StackScan,
|
||
Sweep,
|
||
brief,
|
||
digest,
|
||
read_stack,
|
||
)
|
||
from beaver_agent.jobs.logs import sweep as sweep_fleet
|
||
|
||
NOW = datetime(2026, 9, 20, 18, 0, tzinfo=UTC)
|
||
|
||
|
||
def log(*lines):
|
||
return {"stdout": "\n".join(lines), "stderr": ""}
|
||
|
||
|
||
def line(service, at, text):
|
||
return f"{service} | {at}Z {text}"
|
||
|
||
|
||
def a_stack(name, server="dell-home"):
|
||
return {
|
||
"id": name,
|
||
"name": name,
|
||
"info": {"state": "running", "server_name": server},
|
||
}
|
||
|
||
|
||
def fleet(answers, calls=None):
|
||
"""Komodo, отвечающий заготовками; calls копит запросы к нему."""
|
||
|
||
class Fake(Komodo):
|
||
async def read(self, request, params):
|
||
if calls is not None:
|
||
calls.append((request, params))
|
||
answer = answers.get(request, [])
|
||
return answer(params) if callable(answer) else answer
|
||
|
||
return Fake(url="u", key="k", secret="s")
|
||
|
||
|
||
def swept(scans, **kw):
|
||
kw.setdefault("in_scope", len(scans))
|
||
kw.setdefault("fleet", len(scans))
|
||
return Sweep(at=kw.pop("at", NOW), scans=tuple(scans), **kw)
|
||
|
||
|
||
KNOWN = ("beaver-agent", "gitea", "democalc", "mix", "healthbot")
|
||
|
||
|
||
def seen(tmp_path, *, fresh=False, known=KNOWN):
|
||
"""Состояние с уже снятой базовой линией: проход не первый."""
|
||
state = Seen.load(tmp_path / "логи.json")
|
||
if not fresh:
|
||
state.fresh = False
|
||
state.baselined = set(known)
|
||
return state
|
||
|
||
|
||
def test_a_line_gives_its_service_its_time_and_its_signal():
|
||
scan = read_stack(
|
||
"beaver-agent",
|
||
log(
|
||
line("beaver-gateway", "2026-09-20T17:40:01.123456789", "ERROR boom"),
|
||
line("beaver-gateway", "2026-09-20T17:41:02.000000000", "ERROR boom"),
|
||
line("beaver-obsidian", "2026-09-20T17:42:03.000000000", "Fully synced"),
|
||
),
|
||
)
|
||
(hit,) = scan.hits
|
||
assert hit.service == "beaver-gateway"
|
||
assert hit.signal == "ошибка"
|
||
assert hit.count == 2
|
||
assert hit.at == datetime(2026, 9, 20, 17, 41, 2, tzinfo=UTC)
|
||
assert scan.first == datetime(2026, 9, 20, 17, 40, 1, 123456, tzinfo=UTC)
|
||
assert scan.last == datetime(2026, 9, 20, 17, 42, 3, tzinfo=UTC)
|
||
|
||
|
||
def test_the_same_trouble_with_other_numbers_is_the_same_trouble():
|
||
scan = read_stack(
|
||
"gitea",
|
||
log(
|
||
line("gitea", "2026-09-20T17:40:01.0", "ERROR request 8871 failed"),
|
||
line("gitea", "2026-09-20T17:40:09.0", "ERROR request 9134 failed"),
|
||
),
|
||
)
|
||
(hit,) = scan.hits
|
||
assert hit.count == 2
|
||
|
||
|
||
def test_every_named_signal_is_caught():
|
||
texts = (
|
||
"CRITICAL disk is gone",
|
||
"panic: runtime error",
|
||
"container OOMKilled",
|
||
"Traceback (most recent call last):",
|
||
"service exited with code 137",
|
||
"beaver-gateway Restarting (1) 3 seconds ago",
|
||
"container is unhealthy",
|
||
"ERROR plain",
|
||
)
|
||
scan = read_stack(
|
||
"mix",
|
||
log(*(line("svc", f"2026-09-20T17:4{i}:00.0", t) for i, t in enumerate(texts))),
|
||
)
|
||
assert {h.signal for h in scan.hits} == {
|
||
"критично",
|
||
"паника",
|
||
"oom",
|
||
"traceback",
|
||
"падение",
|
||
"рестарт-цикл",
|
||
"healthcheck",
|
||
"ошибка",
|
||
}
|
||
assert "exited with code 0" not in {h.sample for h in scan.hits}
|
||
|
||
|
||
def test_a_clean_exit_and_our_own_inject_are_not_trouble():
|
||
scan = read_stack(
|
||
"beaver-agent",
|
||
log(
|
||
line("svc", "2026-09-20T17:40:00.0", "worker exited with code 0"),
|
||
line("beaver-gateway", "2026-09-20T17:41:00.0", f"{SELF_MARK}: ERROR boom"),
|
||
),
|
||
)
|
||
assert scan.hits == ()
|
||
|
||
|
||
def test_komodos_own_error_is_not_a_clean_stack():
|
||
scan = read_stack("gitea", {"error": "periphery unreachable"})
|
||
assert not scan.ok
|
||
assert scan.error == "periphery unreachable"
|
||
assert Sweep(at=NOW, scans=(scan,), in_scope=1, fleet=1).trouble == (
|
||
"gitea: periphery unreachable",
|
||
)
|
||
|
||
|
||
def test_the_first_pass_only_takes_the_baseline(tmp_path):
|
||
state = seen(tmp_path, fresh=True)
|
||
scan = read_stack(
|
||
"beaver-agent", log(line("svc", "2026-09-20T17:40:00.0", "ERROR старая беда"))
|
||
)
|
||
report = digest(state, swept([scan]))
|
||
assert report.cold and report.baseline == 1
|
||
assert report.new == ()
|
||
text = brief(report)
|
||
assert "базовую линию" in text
|
||
assert "прочитано 1 стеков из 1" in text
|
||
|
||
|
||
def test_only_the_new_line_wakes_the_master(tmp_path):
|
||
old = line("svc", "2026-09-20T17:40:00.0", "ERROR молотит третьи сутки")
|
||
state = seen(tmp_path, fresh=True)
|
||
first = digest(state, swept([read_stack("beaver-agent", log(old))]))
|
||
state.mark_told(first)
|
||
state.save()
|
||
|
||
fresh_line = line("svc", "2026-09-20T18:50:00.0", "CRITICAL диск кончился")
|
||
state = Seen.load(tmp_path / "логи.json")
|
||
report = digest(
|
||
state,
|
||
swept(
|
||
[read_stack("beaver-agent", log(old, old, fresh_line))],
|
||
at=NOW + timedelta(hours=3),
|
||
),
|
||
)
|
||
(new,) = report.new
|
||
assert new.signal == "критично"
|
||
text = brief(report)
|
||
assert "диск кончился" in text
|
||
assert "молотит третьи сутки" not in text
|
||
|
||
|
||
def test_a_repeat_carries_how_many_times(tmp_path):
|
||
noisy = [
|
||
line("svc", f"2026-09-20T17:4{i}:00.0", "ERROR раз за разом") for i in range(5)
|
||
]
|
||
report = digest(seen(tmp_path), swept([read_stack("beaver-agent", log(*noisy))]))
|
||
assert "×5" in brief(report)
|
||
|
||
|
||
def test_a_trouble_that_stopped_is_dated_by_the_event_not_by_the_look(tmp_path):
|
||
was = line("svc", "2026-09-20T12:03:00.0", "ERROR было и прошло")
|
||
state = seen(tmp_path)
|
||
state.mark_told(digest(state, swept([read_stack("beaver-agent", log(was))])))
|
||
|
||
later = NOW + timedelta(hours=6)
|
||
report = digest(state, swept([read_stack("beaver-agent", log())], at=later))
|
||
(quiet,) = report.quiet
|
||
assert quiet.last == datetime(2026, 9, 20, 12, 3, tzinfo=UTC)
|
||
text = brief(report)
|
||
assert "последнее в 20.09 12:03" in text
|
||
assert f"{later:%H:%M}" not in text.split("\n", 1)[1]
|
||
|
||
|
||
def test_a_stack_that_did_not_answer_is_not_a_stack_that_went_quiet(tmp_path):
|
||
was = line("svc", "2026-09-20T12:03:00.0", "ERROR было")
|
||
state = seen(tmp_path)
|
||
state.mark_told(digest(state, swept([read_stack("beaver-agent", log(was))])))
|
||
|
||
report = digest(
|
||
state,
|
||
swept(
|
||
[StackScan(stack="beaver-agent", error="timeout")],
|
||
at=NOW + timedelta(hours=3),
|
||
),
|
||
)
|
||
assert report.quiet == ()
|
||
assert report.trouble == ("beaver-agent: timeout",)
|
||
assert "beaver-agent: timeout" in brief(report)
|
||
|
||
|
||
def test_an_undelivered_inject_leaves_the_trouble_new(tmp_path):
|
||
scan = read_stack(
|
||
"beaver-agent", log(line("svc", "2026-09-20T17:40:00.0", "ERROR боль"))
|
||
)
|
||
state = seen(tmp_path)
|
||
first = digest(state, swept([scan]))
|
||
assert len(first.new) == 1
|
||
state.save() # инжект не доехал: mark_told не звали
|
||
|
||
state = Seen.load(state.path)
|
||
again = digest(
|
||
state, swept([read_stack("beaver-agent", log())], at=NOW + timedelta(hours=3))
|
||
)
|
||
assert [e.sample for e in again.new] == [e.sample for e in first.new]
|
||
assert again.quiet == ()
|
||
|
||
|
||
def test_the_state_survives_a_restart(tmp_path):
|
||
scan = read_stack(
|
||
"beaver-agent", log(line("svc", "2026-09-20T17:40:00.0", "ERROR боль"))
|
||
)
|
||
state = seen(tmp_path)
|
||
state.mark_told(digest(state, swept([scan])))
|
||
state.save()
|
||
|
||
state = Seen.load(state.path)
|
||
assert not state.fresh
|
||
report = digest(state, swept([scan], at=NOW + timedelta(hours=3)))
|
||
assert report.new == () and report.quiet == ()
|
||
|
||
|
||
def test_an_old_line_still_lying_in_the_tail_is_not_new_again(tmp_path):
|
||
# Тихий стек держит августовскую строку в хвосте месяцами: она не новая.
|
||
scan = read_stack(
|
||
"democalc", log(line("pg-1", "2026-08-22T15:49:51.0", "ERROR: relation"))
|
||
)
|
||
state = seen(tmp_path)
|
||
state.mark_told(digest(state, swept([scan])))
|
||
state.save()
|
||
|
||
state = Seen.load(state.path)
|
||
assert digest(state, swept([scan], at=NOW + timedelta(hours=3))).new == ()
|
||
|
||
|
||
def test_a_trouble_forgotten_by_time_comes_back_as_new(tmp_path):
|
||
scan = read_stack(
|
||
"beaver-agent", log(line("svc", "2026-09-20T17:40:00.0", "ERROR боль"))
|
||
)
|
||
state = seen(tmp_path)
|
||
state.mark_told(digest(state, swept([scan])))
|
||
late = NOW + KEEP + timedelta(days=1)
|
||
report = digest(state, swept([scan], at=late))
|
||
assert len(report.new) == 1
|
||
|
||
|
||
def test_a_silent_pass_is_not_worth_the_master(tmp_path):
|
||
state = seen(tmp_path)
|
||
assert not digest(state, swept([read_stack("beaver-agent", log())])).worth_telling
|
||
|
||
|
||
def test_the_scope_says_how_much_was_looked_at(tmp_path):
|
||
scans = [
|
||
read_stack(
|
||
"beaver-agent", log(line("svc", "2026-09-20T17:40:00.0", "ERROR боль"))
|
||
),
|
||
StackScan(stack="gitea", error="timeout"),
|
||
]
|
||
report = digest(seen(tmp_path), swept(scans, in_scope=2, fleet=53))
|
||
head = brief(report).splitlines()[1]
|
||
assert head.startswith(
|
||
"Область: dell-home, personal-main - прочитано 1 стеков из 2"
|
||
)
|
||
assert "во флоте 53" in head
|
||
assert "хвост 500 строк на стек: строки с 20.09 17:40 по 20.09 17:40" in head
|
||
assert "самый короткий хвост у beaver-agent - только с 20.09 17:40" in head
|
||
|
||
|
||
def test_a_tail_shorter_than_the_pause_between_passes_is_said_out_loud(tmp_path):
|
||
state = seen(tmp_path)
|
||
state.last_pass = NOW - timedelta(hours=3)
|
||
scan = read_stack(
|
||
"beaver-agent",
|
||
log(line("svc", "2026-09-20T17:59:00.0", "ERROR из последней минуты")),
|
||
)
|
||
report = digest(state, swept([scan]))
|
||
assert report.gaps == ("beaver-agent",)
|
||
assert "часть строк прошла мимо" in brief(report)
|
||
|
||
|
||
def test_a_silent_komodo_says_so_instead_of_reporting_clean(tmp_path):
|
||
down = Sweep(at=NOW, unreachable="ClientError: connection refused")
|
||
report = digest(seen(tmp_path), down)
|
||
assert report.worth_telling
|
||
text = brief(report)
|
||
assert "Komodo не ответил" in text
|
||
assert "Нового в этой области нет" not in text
|
||
|
||
|
||
def test_a_lasting_outage_is_told_once(tmp_path):
|
||
down = Sweep(at=NOW, unreachable="ClientError: connection refused")
|
||
state = seen(tmp_path)
|
||
state.mark_told(digest(state, down))
|
||
assert digest(state, down).trouble == ()
|
||
|
||
|
||
async def test_the_sweep_lists_every_page_and_skips_other_peoples_machines():
|
||
pages = [
|
||
[a_stack(f"s{i}") for i in range(500)],
|
||
[a_stack("gitea", "personal-main"), a_stack("veris-runner", "veris-main")],
|
||
]
|
||
calls = []
|
||
komodo = fleet(
|
||
{
|
||
"ListStacks": lambda p: pages[p["page"]],
|
||
"GetStacksSummary": {"total": 502},
|
||
"GetStackLog": log(),
|
||
},
|
||
calls,
|
||
)
|
||
out = await sweep_fleet(komodo, now=NOW)
|
||
asked = [p["stack"] for r, p in calls if r == "GetStackLog"]
|
||
assert "veris-runner" not in asked
|
||
assert "gitea" in asked and "s0" in asked
|
||
assert out.fleet == 502 and out.in_scope == 501
|
||
assert out.trouble == ()
|
||
assert all(p.get("timestamps") for r, p in calls if r == "GetStackLog")
|
||
|
||
|
||
async def test_the_sweep_shouts_when_the_list_and_the_summary_disagree():
|
||
komodo = fleet(
|
||
{
|
||
"ListStacks": [a_stack("beaver-agent")],
|
||
"GetStacksSummary": {"total": 53},
|
||
"GetStackLog": log(),
|
||
}
|
||
)
|
||
out = await sweep_fleet(komodo, now=NOW)
|
||
assert out.trouble == (
|
||
"листание дало 1 стеков, а GetStacksSummary - 53: список неполон",
|
||
)
|
||
|
||
|
||
async def test_a_stack_that_throws_does_not_end_the_sweep():
|
||
def logs(params):
|
||
if params["stack"] == "gitea":
|
||
msg = "connection reset"
|
||
raise RuntimeError(msg)
|
||
return log(line("svc", "2026-09-20T17:40:00.0", "ERROR боль"))
|
||
|
||
komodo = fleet(
|
||
{
|
||
"ListStacks": [a_stack("beaver-agent"), a_stack("gitea", "personal-main")],
|
||
"GetStacksSummary": {"total": 2},
|
||
"GetStackLog": logs,
|
||
}
|
||
)
|
||
out = await sweep_fleet(komodo, now=NOW)
|
||
assert len(out.read) == 1
|
||
assert out.trouble == ("gitea: RuntimeError: connection reset",)
|
||
|
||
|
||
async def test_a_dead_komodo_makes_a_sweep_that_admits_it():
|
||
class Dead(Komodo):
|
||
async def read(self, request, params):
|
||
msg = "connection refused"
|
||
raise OSError(msg)
|
||
|
||
out = await sweep_fleet(Dead(url="u", key="k", secret="s"), now=NOW)
|
||
assert out.read == ()
|
||
assert out.unreachable == "OSError: connection refused"
|
||
assert out.trouble == ("Komodo не ответил: OSError: connection refused",)
|
||
|
||
|
||
def test_a_line_without_a_stamp_is_not_a_find():
|
||
# Голову хвоста Komodo обрезает по месту: полстроки - не новая беда.
|
||
scan = read_stack(
|
||
"democalc",
|
||
log(
|
||
"49.260 UTC [8772] ERROR: relation does not exist",
|
||
line("postgres-1", "2026-09-20T17:40:00.0", "ERROR: настоящая"),
|
||
),
|
||
)
|
||
assert [h.sample for h in scan.hits] == ["ERROR: настоящая"]
|
||
assert scan.unparsed == 1 and scan.lines == 1
|
||
|
||
|
||
def test_many_unreadable_lines_are_said_out_loud(tmp_path):
|
||
scan = StackScan(stack="beaver-agent", lines=5, unparsed=40)
|
||
head = brief(digest(seen(tmp_path), swept([scan]))).splitlines()[1]
|
||
assert "строк без метки времени пропущено 40" in head
|
||
|
||
|
||
def test_an_empty_line_is_a_line_not_a_blind_spot():
|
||
scan = read_stack("dell-home-dawarich", log("app-1 | 2026-09-17T14:37:38.510Z"))
|
||
assert scan.unparsed == 0 and scan.lines == 1 and scan.hits == ()
|
||
|
||
|
||
def test_a_stack_read_for_the_first_time_does_not_dump_its_past(tmp_path):
|
||
state = seen(tmp_path)
|
||
scan = read_stack(
|
||
"lodge", log(line("svc", "2026-09-19T10:00:00.0", "ERROR старое, до нас"))
|
||
)
|
||
report = digest(state, swept([scan]))
|
||
assert report.new == ()
|
||
assert report.baselined == ("lodge",) and report.baseline == 1
|
||
assert "Впервые прочитаны" in brief(report)
|
||
|
||
later = read_stack(
|
||
"lodge",
|
||
log(
|
||
line("svc", "2026-09-19T10:00:00.0", "ERROR старое, до нас"),
|
||
line("svc", "2026-09-20T19:00:00.0", "ERROR уже при нас"),
|
||
),
|
||
)
|
||
after = digest(state, swept([later], at=NOW + timedelta(hours=3)))
|
||
assert [e.sample for e in after.new] == ["ERROR уже при нас"]
|
||
|
||
|
||
def test_a_first_pass_into_a_dead_komodo_does_not_burn_the_baseline(tmp_path):
|
||
state = seen(tmp_path, fresh=True)
|
||
down = Sweep(at=NOW, unreachable="OSError: connection refused")
|
||
report = digest(state, down)
|
||
assert not report.cold and state.fresh
|
||
assert report.trouble == ("Komodo не ответил: OSError: connection refused",)
|
||
|
||
scan = read_stack(
|
||
"beaver-agent", log(line("svc", "2026-09-20T17:40:00.0", "ERROR боль"))
|
||
)
|
||
later = digest(state, swept([scan], at=NOW + timedelta(hours=3)))
|
||
assert later.cold and later.new == ()
|
||
|
||
|
||
def test_a_secret_in_the_line_does_not_ride_into_the_chat():
|
||
scan = read_stack(
|
||
"gitea",
|
||
log(
|
||
line(
|
||
"svc",
|
||
"2026-09-20T17:40:00.0",
|
||
"ERROR auth failed token=ghp_ZZ123 for kJ8sdfLKJ23kjhsdf98234hkjsdf7823x",
|
||
)
|
||
),
|
||
)
|
||
(hit,) = scan.hits
|
||
assert "ghp_ZZ123" not in hit.sample
|
||
assert "kJ8sdfLKJ23kjhsdf98234hkjsdf7823x" not in hit.sample
|
||
assert hit.sample.startswith("ERROR auth failed")
|