From fe14b8d7d7a75d17e40fa74f9813a60a0b1fc642 Mon Sep 17 00:00:00 2001 From: Aleksei Fedorov Date: Wed, 19 Aug 2026 19:17:17 +0000 Subject: [PATCH] tests --- tests/test_irq_anatomy_service.py | 207 +++++++++++++ tests/test_memory_service.py | 309 +++++++++++++++++++ tests/test_ml_http.py | 146 +++++++++ tests/test_ml_store_connect.py | 32 ++ tests/test_ml_thresholds.py | 35 +++ tests/test_process_inspect_service.py | 61 ++++ tests/test_runqueue_service.py | 198 ++++++++++++ tests/test_sched_debug_collector.py | 129 ++++++++ tests/test_syscall_snapshot_collector.py | 106 +++++++ tests/test_syscalls_service.py | 14 + tests/test_system_view_service.py | 20 ++ tests/test_threads_service.py | 233 +++++++++++++++ tests/test_waits_service.py | 366 +++++++++++++++++++++++ tests/test_wakeup_collector.py | 97 ++++++ tests/test_wakeups_service.py | 144 +++++++++ 15 files changed, 2097 insertions(+) create mode 100644 tests/test_irq_anatomy_service.py create mode 100644 tests/test_memory_service.py create mode 100644 tests/test_ml_http.py create mode 100644 tests/test_ml_store_connect.py create mode 100644 tests/test_ml_thresholds.py create mode 100644 tests/test_runqueue_service.py create mode 100644 tests/test_sched_debug_collector.py create mode 100644 tests/test_syscall_snapshot_collector.py create mode 100644 tests/test_threads_service.py create mode 100644 tests/test_waits_service.py create mode 100644 tests/test_wakeup_collector.py create mode 100644 tests/test_wakeups_service.py diff --git a/tests/test_irq_anatomy_service.py b/tests/test_irq_anatomy_service.py new file mode 100644 index 0000000..2d2fcbf --- /dev/null +++ b/tests/test_irq_anatomy_service.py @@ -0,0 +1,207 @@ +"""Tests for ``kernel_ai.services.irq_anatomy``.""" + +from kernel_ai.services import irq_anatomy as ia +from kernel_ai.services import syscall_anatomy as sa + + +def _stub_machine(monkeypatch, interrupts, sysfs, procfs=None, symbols=(), online=1, tasks=()): + """Stand in for the four files the module reads. + + ``sysfs`` and ``procfs`` are keyed by the path tail, so a test only has to + spell out the attributes it cares about. + """ + procfs = procfs or {} + monkeypatch.setattr(ia, "read_interrupts", lambda: interrupts) + monkeypatch.setattr(ia, "kernel_symbols", lambda: frozenset(symbols)) + monkeypatch.setattr(ia, "_cpus_online", lambda: online) + monkeypatch.setattr(ia, "_threaded_handler", lambda irq: dict(tasks).get(irq)) + + def fake_read(path): + for root, table in ((ia._SYS_IRQ, sysfs), (ia._PROC_IRQ, procfs)): + if path.startswith(root + "/"): + return table.get(path[len(root) + 1:], "") + return "" + + monkeypatch.setattr(ia, "_read", fake_read) + + +def test_a_network_line_is_followed_to_the_function_its_softirq_runs(monkeypatch): + _stub_machine( + monkeypatch, + interrupts={"56": ([43168173, 0], "xen-dyn-lateeoi -event eth0")}, + sysfs={ + "56/actions": "eth0", + "56/chip_name": "xen-dyn-lateeoi", + "56/name": "event", + "56/type": "edge", + }, + procfs={"56/smp_affinity_list": "0", "56/effective_affinity_list": "0"}, + symbols={"net_rx_action"}, + online=2, + ) + monkeypatch.setattr(ia, "read_softirqs", lambda: {"NET_RX": 16075193}) + + out = ia.describe("56") + assert out["kind"] == "line" + assert out["device"] == "eth0" + assert [step["stage"] for step in out["chain"]] == [ + "line", "handler", "raises", "runs", "thread", + ] + assert [step["symbol"] for step in out["chain"]][:4] == [ + "xen-dyn-lateeoi", "eth0", "NET_RX", "net_rx_action", + ] + assert out["softirq"]["vector"] == "NET_RX" + assert out["softirq"]["total"] == 16075193 + assert out["affinity"] == {"allowed": "0", "effective": "0"} + assert out["total"] == 43168173 + + +def test_the_softirq_vector_is_marked_as_reasoned_not_measured(monkeypatch): + """The kernel records no vector per line, and the card must not pretend.""" + _stub_machine( + monkeypatch, + interrupts={"7": ([12], "nvme0q1")}, + sysfs={"7/actions": "nvme0q1", "7/chip_name": "IR-PCI-MSI"}, + symbols={"blk_done_softirq"}, + ) + monkeypatch.setattr(ia, "read_softirqs", lambda: {"BLOCK": 4679716}) + + raises = next(s for s in ia.describe("7")["chain"] if s["stage"] == "raises") + assert raises["symbol"] == "BLOCK" + assert raises["inferred"] is True + assert raises["note"] == "by driver class" + + +def test_a_symbol_this_kernel_does_not_export_is_shown_as_unconfirmed(monkeypatch): + _stub_machine( + monkeypatch, + interrupts={"56": ([9], "eth0")}, + sysfs={"56/actions": "eth0", "56/chip_name": "IR-PCI-MSI"}, + symbols=set(), + ) + monkeypatch.setattr(ia, "read_softirqs", lambda: {}) + + runs = next(s for s in ia.describe("56")["chain"] if s["stage"] == "runs") + assert runs["confirmed"] is False + assert runs["note"] == "not in kallsyms" + + +def test_an_unrecognised_device_claims_no_vector_at_all(monkeypatch): + _stub_machine( + monkeypatch, + interrupts={"3": ([4], "some-odd-widget")}, + sysfs={"3/actions": "some-odd-widget", "3/chip_name": "IO-APIC"}, + symbols={"net_rx_action", "blk_done_softirq"}, + ) + monkeypatch.setattr(ia, "read_softirqs", lambda: {"NET_RX": 1}) + + out = ia.describe("3") + assert out["softirq"] is None + assert [step["stage"] for step in out["chain"]] == ["line", "handler", "thread"] + + +def test_a_threaded_handler_ends_the_chain_on_a_real_pid(monkeypatch): + _stub_machine( + monkeypatch, + interrupts={"9": ([0], "acpi")}, + sysfs={"9/actions": "acpi", "9/chip_name": "xen-pirq"}, + tasks=[("9", {"pid": 34, "comm": "irq/9-acpi"})], + ) + monkeypatch.setattr(ia, "read_softirqs", lambda: {}) + + last = ia.describe("9")["chain"][-1] + assert last["stage"] == "thread" + assert last["confirmed"] is True + assert "pid 34" in last["symbol"] + + +def test_a_lettered_row_is_a_counter_and_gets_no_invented_chain(monkeypatch): + _stub_machine( + monkeypatch, + interrupts={"MCP": ([8131], "Machine check polls")}, + sysfs={}, + ) + monkeypatch.setattr(ia, "read_softirqs", lambda: {}) + + out = ia.describe("MCP") + assert out["kind"] == "aggregate" + assert out["chain"] == [] + assert out["summary"] == "machine check polls" + assert "children" not in out + + +def test_the_hypervisor_counter_shows_the_channels_that_came_through_it(monkeypatch): + """On Xen, HYP is the door every event channel enters by, not a device.""" + _stub_machine( + monkeypatch, + interrupts={ + "HYP": ([737233233], "Hypervisor callback interrupts"), + "48": ([691592800], "xen-percpu -virq timer0"), + "56": ([43168173], "xen-dyn-lateeoi -event eth0"), + "31": ([50], "IO-APIC 31-edge not-a-channel"), + }, + sysfs={ + "48/chip_name": "xen-percpu", "48/actions": "timer0", + "56/chip_name": "xen-dyn-lateeoi", "56/actions": "eth0", + "31/chip_name": "IO-APIC", "31/actions": "not-a-channel", + }, + ) + monkeypatch.setattr(ia, "read_softirqs", lambda: {}) + + out = ia.describe("HYP") + assert [child["irq"] for child in out["children"]] == ["48", "56"] + assert out["children"][0]["device"] == "timer0" + assert out["children_total"] == 691592800 + 43168173 + + +def test_channels_that_never_fired_are_left_off_the_hypervisor_card(monkeypatch): + """A guest is wired with channels it never uses; a zero bar earns no row.""" + interrupts = {"HYP": ([100], "Hypervisor callback interrupts")} + sysfs = {} + for index in range(9): + irq = str(40 + index) + interrupts[irq] = ([10 - index], f"xen-dyn -event dev{index}") + sysfs[f"{irq}/chip_name"] = "xen-dyn" + sysfs[f"{irq}/actions"] = f"dev{index}" + interrupts["70"] = ([0], "xen-dyn -event spinlock0") + sysfs["70/chip_name"] = "xen-dyn" + sysfs["70/actions"] = "spinlock0" + + _stub_machine(monkeypatch, interrupts=interrupts, sysfs=sysfs) + monkeypatch.setattr(ia, "read_softirqs", lambda: {}) + + out = ia.describe("HYP") + assert "spinlock0" not in [child["device"] for child in out["children"]] + assert len(out["children"]) == ia._MAX_CHANNELS + # The nine that did fire, minus the six shown. + assert out["children_hidden"] == 3 + # The total still covers every channel, shown or not. + assert out["children_total"] == sum(range(2, 11)) + + +def test_a_line_the_machine_does_not_have_has_no_answer(monkeypatch): + _stub_machine(monkeypatch, interrupts={"56": ([1], "eth0")}, sysfs={}) + monkeypatch.setattr(ia, "read_softirqs", lambda: {}) + assert ia.describe("4096") is None + + +def test_a_path_cannot_be_smuggled_in_through_the_irq_name(monkeypatch): + _stub_machine(monkeypatch, interrupts={"56": ([1], "eth0")}, sysfs={}) + monkeypatch.setattr(ia, "read_softirqs", lambda: {}) + assert ia.describe("../../etc/passwd") is None + assert ia.describe("56/../55") is None + + +def test_the_kallsyms_filter_keeps_every_symbol_the_irq_card_asks_about(): + """The filter lives in syscall_anatomy, which cannot import this module. + + If a vector is added here without being added there, the collector would + stop publishing its symbol and the card would quietly report it missing. + """ + assert set(ia.VECTOR_SYMBOL.values()) <= sa._SOFTIRQ_SYMBOLS + + +def test_the_softirq_vectors_of_this_machine_are_all_described(): + """Every vector /proc/softirqs prints should have a function to name.""" + for vector in ia.read_softirqs(): + assert vector in ia.VECTOR_SYMBOL, vector diff --git a/tests/test_memory_service.py b/tests/test_memory_service.py new file mode 100644 index 0000000..4cee19d --- /dev/null +++ b/tests/test_memory_service.py @@ -0,0 +1,309 @@ +"""Tests for ``kernel_ai.services.memory``.""" + +import json +import os +import time + +from kernel_ai.services import memory as svc + +MAPS = """\ +aaaadc8b0000-aaaadca17000 r-xp 00000000 fd:00 922628 /usr/lib/systemd/systemd +aaaadca27000-aaaadca75000 r--p 00167000 fd:00 922628 /usr/lib/systemd/systemd +aaaadca75000-aaaadca76000 rw-p 001b5000 fd:00 922628 /usr/lib/systemd/systemd +aaaadca76000-aaaadca78000 rw-p 00000000 00:00 0 +aaaaee903000-aaaaeea89000 rw-p 00000000 00:00 0 [heap] +ffffa2ea0000-ffffa2ec2000 r-xp 00000000 fd:00 919057 /usr/lib/aarch64-linux-gnu/libgpg-error.so.0.32.1 +ffffa2ec2000-ffffa2ed1000 ---p 00022000 fd:00 919057 /usr/lib/aarch64-linux-gnu/libgpg-error.so.0.32.1 +ffffa2ed1000-ffffa2ed2000 r--p 00021000 fd:00 919057 /usr/lib/aarch64-linux-gnu/libgpg-error.so.0.32.1 +ffffa2ed2000-ffffa2ed3000 rw-p 00022000 fd:00 919057 /usr/lib/aarch64-linux-gnu/libgpg-error.so.0.32.1 +ffffc0000000-ffffc0002000 rw-p 00000000 00:00 0 [stack:4242] +ffffff000000-ffffff002000 rw-p 00000000 00:00 0 [stack] +ffffd0000000-ffffd0001000 r-xp 00000000 00:00 0 [vdso] +""" + +ROLLUP = """\ +aaaadc8b0000-ffffe4614000 ---p 00000000 00:00 0 [rollup] +Rss: 6584 kB +Pss: 3485 kB +Shared_Clean: 3564 kB +Shared_Dirty: 0 kB +Private_Clean: 408 kB +Private_Dirty: 2612 kB +Anonymous: 2612 kB +Swap: 0 kB +""" + +STATUS = """\ +Name: systemd +VmSize: 167648 kB +VmRSS: 8156 kB +VmData: 20416 kB +VmStk: 132 kB +VmExe: 1436 kB +VmLib: 12448 kB +VmSwap: 0 kB +Threads: 1 +""" + + +def test_adjacent_mappings_of_the_same_file_are_one_region(): + rows = svc.parse_maps( + "7ffff7f00000-7ffff7f20000 r-xp 00000000 fd:00 1 /lib/libc.so.6\n" + "7ffff7f20000-7ffff7f30000 r--p 00020000 fd:00 1 /lib/libc.so.6\n" + "7ffff7f30000-7ffff7f31000 rw-p 00030000 fd:00 1 /lib/libc.so.6\n" + ) + lib = [r for r in rows if r["kind"] == "library"] + assert len(lib) == 1 + assert lib[0]["size_kb"] == 0x31000 // 1024 + + +def test_the_kinds_are_told_apart(): + kinds = {r["kind"] for r in svc.parse_maps(MAPS)} + assert kinds == {"code", "file", "anonymous", "heap", "library", "stack", "vdso", "reserved"} + + +def test_a_guard_page_is_reserved_not_part_of_the_library(): + """The ---p hole between text and data is not library memory.""" + rows = svc.parse_maps(MAPS) + lib = [r for r in rows if r["kind"] == "library"] + reserved = [r for r in rows if r["kind"] == "reserved"] + assert len(lib) == 2 + assert reserved and all(r["path"] and "libgpg-error" in r["path"] for r in reserved) + + +def test_a_thread_stack_keeps_its_tid(): + stacks = [r for r in svc.parse_maps(MAPS) if r["kind"] == "stack"] + assert {r["tid"] for r in stacks} == {4242, None} + + +def test_a_process_that_is_gone_is_reported_as_gone(tmp_path, monkeypatch): + monkeypatch.setattr(svc, "PROC", str(tmp_path)) + assert svc.describe(9).get("error") == "no such process" + + +def test_the_map_is_grouped_and_the_totals_come_from_the_rollup(tmp_path, monkeypatch): + proc = tmp_path / "100" + proc.mkdir() + (proc / "comm").write_text("systemd\n") + (proc / "status").write_text(STATUS) + (proc / "maps").write_text(MAPS) + (proc / "smaps_rollup").write_text(ROLLUP) + monkeypatch.setattr(svc, "PROC", str(tmp_path)) + + out = svc.describe(100) + + assert out["comm"] == "systemd" + assert out["totals"]["rss_kb"] == 6584 + assert out["totals"]["pss_kb"] == 3485 + assert out["totals"]["private_kb"] == 3020 + assert out["executable"]["name"] == "systemd" + assert out["heap_kb"] == (0xaaaaeea89000 - 0xaaaaee903000) // 1024 + assert out["stack_count"] == 2 + assert [s["tid"] for s in out["stacks"]] == [100, 4242] + assert out["libraries"][0]["name"] == "libgpg-error.so.0.32.1" + assert "path" not in out["libraries"][0] + assert "path" not in (out["executable"] or {}) + assert "start" not in json.dumps(out) + assert out["sources"]["maps"]["available"] is True + assert out["sources"]["rollup"]["available"] is True + + +def test_without_the_map_the_totals_still_come_from_status(tmp_path, monkeypatch): + """pid 1 often refuses maps; status is still there and is not guessed from.""" + proc = tmp_path / "1" + proc.mkdir() + (proc / "comm").write_text("systemd\n") + (proc / "status").write_text(STATUS) + monkeypatch.setattr(svc, "PROC", str(tmp_path)) + monkeypatch.setattr(svc, "SNAPSHOT", str(tmp_path / "no-such-maps.json")) + + out = svc.describe(1) + + assert out["kinds"] == [] + assert out["libraries"] == [] + assert out["totals"]["rss_kb"] == 8156 + assert out["totals"]["virtual_kb"] == 167648 + assert out["totals"]["exe_kb"] == 1436 + assert out["sources"]["maps"]["available"] is False + assert out["sources"]["status"]["available"] is True + + +def test_a_shared_object_is_not_called_the_executable(): + text = ( + "7f00-7f10 r-xp 00000000 fd:00 1 /usr/lib/libc.so.6\n" + "4000-4010 r-xp 00000000 fd:00 2 /usr/bin/sleep\n" + ) + rows = svc.parse_maps(text) + assert [r["kind"] for r in rows] == ["library", "code"] + + +def test_a_collected_snapshot_has_no_addresses_or_directory_paths(tmp_path, monkeypatch): + """The world-readable file must not slide ASLR or name a home directory.""" + proc = tmp_path / "100" + proc.mkdir() + (proc / "comm").write_text("systemd\n") + (proc / "status").write_text(STATUS) + (proc / "maps").write_text(MAPS) + (proc / "smaps_rollup").write_text(ROLLUP) + monkeypatch.setattr(svc, "PROC", str(tmp_path)) + + out = svc.collect_all() + blob = json.dumps(out) + + assert "100" in out["processes"] + assert out["processes"]["100"]["libraries"][0]["name"] == "libgpg-error.so.0.32.1" + assert "aaaadc8b" not in blob + assert "ffffa2ea" not in blob + assert "/usr/lib" not in blob + assert "/home/" not in blob + assert "start" not in blob + assert "end" not in blob + + +def test_a_foreign_process_uses_the_collector_snapshot(tmp_path, monkeypatch): + """When the kernel refuses the map, a fresh snapshot still draws the card.""" + proc = tmp_path / "1" + proc.mkdir() + (proc / "comm").write_text("systemd\n") + (proc / "status").write_text(STATUS) + snap = tmp_path / "maps.json" + snap.write_text(json.dumps({ + "ts": time.time(), + "processes": { + "1": { + "comm": "systemd", + "totals": {"virtual_kb": 9000, "rss_kb": 100, "pss_kb": 80}, + "kinds": [{"kind": "library", "virtual_kb": 500, "count": 2}], + "libraries": [{"name": "libc.so.6", "virtual_kb": 400, "count": 2}], + "library_count": 1, + "stacks": [{"tid": 1, "virtual_kb": 132, "main": True}], + "stack_count": 1, + "executable": {"name": "systemd", "virtual_kb": 200}, + "heap_kb": 300, + } + }, + })) + monkeypatch.setattr(svc, "PROC", str(tmp_path)) + monkeypatch.setattr(svc, "SNAPSHOT", str(snap)) + + out = svc.describe(1) + + assert out["sources"]["maps"]["available"] is True + assert out["sources"]["maps"]["via"] == "collector" + assert out["libraries"][0]["name"] == "libc.so.6" + assert "path" not in out["libraries"][0] + assert out["totals"]["rss_kb"] == 8156 + assert out["totals"]["pss_kb"] == 80 + assert out["heap_kb"] == 300 + assert "start" not in json.dumps(out) + + +def test_a_stale_snapshot_is_not_used(tmp_path, monkeypatch): + proc = tmp_path / "1" + proc.mkdir() + (proc / "comm").write_text("systemd\n") + (proc / "status").write_text(STATUS) + snap = tmp_path / "maps.json" + snap.write_text(json.dumps({ + "ts": 0, + "processes": {"1": {"kinds": [{"kind": "library", "virtual_kb": 1, "count": 1}]}}, + })) + past = time.time() - 60 + os.utime(snap, (past, past)) + monkeypatch.setattr(svc, "PROC", str(tmp_path)) + monkeypatch.setattr(svc, "SNAPSHOT", str(snap)) + + out = svc.describe(1) + + assert out["kinds"] == [] + assert out["sources"]["maps"]["available"] is False + + +def test_a_thread_uses_the_process_snapshot(tmp_path, monkeypatch): + """The dossier pid is sometimes a tid; the map belongs to the thread-group.""" + proc = tmp_path / "4242" + proc.mkdir() + (proc / "comm").write_text("python\n") + (proc / "status").write_text(STATUS + "Tgid:\t100\n") + snap = tmp_path / "maps.json" + snap.write_text(json.dumps({ + "ts": time.time(), + "processes": { + "100": { + "comm": "python", + "totals": {"virtual_kb": 9000, "rss_kb": 100}, + "kinds": [{"kind": "library", "virtual_kb": 500, "count": 1}], + "libraries": [{"name": "libc.so.6", "virtual_kb": 400, "count": 1}], + "library_count": 1, + "stacks": [], + "stack_count": 0, + "executable": {"name": "python3.10", "virtual_kb": 200}, + "heap_kb": 300, + } + }, + "threads": {"4242": "100"}, + })) + monkeypatch.setattr(svc, "PROC", str(tmp_path)) + monkeypatch.setattr(svc, "SNAPSHOT", str(snap)) + + out = svc.describe(4242) + + assert out["sources"]["maps"]["via"] == "collector" + assert out["libraries"][0]["name"] == "libc.so.6" + assert out["executable"]["name"] == "python3.10" + + +def test_an_empty_maps_file_still_uses_the_snapshot(tmp_path, monkeypatch): + """An open that yields no areas is not a map — fall through to the collector.""" + proc = tmp_path / "1" + proc.mkdir() + (proc / "comm").write_text("systemd\n") + (proc / "status").write_text(STATUS) + (proc / "maps").write_text("") + snap = tmp_path / "maps.json" + snap.write_text(json.dumps({ + "ts": time.time(), + "processes": { + "1": { + "comm": "systemd", + "totals": {"virtual_kb": 9000, "rss_kb": 100}, + "kinds": [{"kind": "library", "virtual_kb": 500, "count": 1}], + "libraries": [{"name": "libc.so.6", "virtual_kb": 400, "count": 1}], + "library_count": 1, + "stacks": [], + "stack_count": 0, + "executable": {"name": "systemd", "virtual_kb": 200}, + "heap_kb": None, + } + }, + })) + monkeypatch.setattr(svc, "PROC", str(tmp_path)) + monkeypatch.setattr(svc, "SNAPSHOT", str(snap)) + + out = svc.describe(1) + + assert out["sources"]["maps"]["via"] == "collector" + assert out["kinds"][0]["kind"] == "library" + + +def test_collect_all_keeps_the_last_map_when_it_goes_empty(tmp_path, monkeypatch): + proc = tmp_path / "1" + proc.mkdir() + (proc / "comm").write_text("python\n") + (proc / "status").write_text(STATUS) + (proc / "maps").write_text("") + monkeypatch.setattr(svc, "PROC", str(tmp_path)) + + previous = {"1": { + "comm": "python", + "totals": {"virtual_kb": 9000, "rss_kb": 221184}, + "kinds": [{"kind": "library", "virtual_kb": 500, "count": 1}], + "libraries": [{"name": "libc.so.6", "virtual_kb": 400, "count": 1}], + "library_count": 1, + "stacks": [], + "stack_count": 0, + "executable": {"name": "python3.10", "virtual_kb": 200}, + "heap_kb": 300, + }} + out = svc.collect_all(previous=previous) + assert out["processes"]["1"]["executable"]["name"] == "python3.10" diff --git a/tests/test_ml_http.py b/tests/test_ml_http.py new file mode 100644 index 0000000..db573c3 --- /dev/null +++ b/tests/test_ml_http.py @@ -0,0 +1,146 @@ +"""Stage 9 HTTP attempt model, labels, join, gap, DNA mapping.""" + +from kernel_ai.ml.http_features import HTTP_FEATURE_ORDER, build_windows, features_of +from kernel_ai.ml.http_gap import report_gap +from kernel_ai.ml.http_join import join_success +from kernel_ai.ml.http_model import HttpAttemptModel +from kernel_ai.ml.http_parse import HttpEvent, parse_line +from kernel_ai.ml.http_polygon import benign_events, run_http_rce, run_http_wordlist, wordlist_events +from kernel_ai.ml.http_rules import classify_event, named_attempt_class +from kernel_ai.ml.http_tail import HttpLogTail +from kernel_ai.services.telemetry_orchestration import _ml_anomalies_to_mutations + + +def test_parse_nginx_and_ecs(): + nginx = parse_line( + '{"time_iso8601":"2026-08-15T12:00:00+00:00","remote_addr":"1.2.3.4",' + '"http_x_forwarded_for":"9.9.9.9","request_method":"GET",' + '"request_uri":"/.env?x=1","status":404}' + ) + assert nginx is not None + assert nginx.src_ip == "9.9.9.9" + assert nginx.path == "/.env" + assert nginx.status == 404 + ecs = parse_line( + '{"@timestamp":"2026-08-15T12:00:01Z","source.ip":"8.8.8.8",' + '"http.request.method":"POST","url.path":"/api",' + '"url.query":"q=1","http.response.status_code":200,' + '"http.request.body.content":"{\\"x\\":1}","event.dataset":"kernel_ai.http"}' + ) + assert ecs is not None + assert ecs.src_ip == "8.8.8.8" + assert ecs.dataset == "kernel_ai.http" + + +def test_rules_classify_patterns(): + ev = HttpEvent(1, "1.1.1.1", "GET", "/../../etc/passwd", "", 404, "", "", "nginx") + assert classify_event(ev) == "lfi" + ev = HttpEvent(1, "1.1.1.1", "GET", "/x", "id=1'+or+1=1", 200, "", "", "nginx") + assert classify_event(ev) == "sqli" + ev = HttpEvent(1, "1.1.1.1", "GET", "/vendor/phpunit/phpunit/src/Util/PHP/eval-stdin.php", "", 404, "", "", "nginx") + assert classify_event(ev) == "scanner" + + +def test_named_attempt_skips_quiet_model_hit(): + assert named_attempt_class("benign", "benign") is None + assert named_attempt_class("attempt", "anomaly") is None + assert named_attempt_class("attempt", "scanner") == "scanner" + + +def test_dashboard_poll_window_stays_benign(): + now = 1_700_000_000.0 + paths = [ + "/api/active-connections", + "/api/execution-context", + "/api/io-open-files", + "/api/io-pulse", + "/api/kernel-data", + "/api/syscalls-realtime", + ] + events = [ + HttpEvent(now + i * 0.7, "198.51.100.8", "GET", paths[i % 6], "", 200, "", "", "nginx") + for i in range(73) + ] + wins = build_windows(events, label=True) + assert wins and wins[0].label == "benign" + assert wins[0].why == "quiet" + assert named_attempt_class(wins[0].label, wins[0].cls) is None + + +def test_window_teacher_wordlist_vs_benign(): + attack = build_windows(wordlist_events(1_700_000_000.0), label=True) + quiet = build_windows(benign_events(1_700_000_000.0), label=True) + assert attack and attack[0].label == "attempt" + assert quiet and quiet[0].label == "benign" + assert set(attack[0].features) == set(HTTP_FEATURE_ORDER) + assert features_of([])["count"] == 0.0 + + +def test_http_model_separates_scanner_and_benign(): + result = run_http_wordlist() + assert result["teacher_attempt"] is True + assert result["model_attempt"][0] is True + assert result["model_benign"][0] is False + assert result["pass"] is True + + +def test_join_success_only_after_attempt_and_web_comm(): + result = run_http_rce() + assert result["pass"] is True + assert result["feature"] == "http_success:exec" + extra = join_success( + [{"ts": 10, "src_ip": "1.1.1.1", "cls": "scanner"}], + [{"ts": 12, "source": "stage4_sequence", "meta": {"comm": "cron"}}], + ) + assert extra == [] + + +def test_gap_report_hole_and_noise(): + attempts = [{"ts": 100.0, "src_ip": "1.1.1.1", "label": "attempt"}] + kernel = [ + {"ts": 500.0, "source": "stage2_isoforest", "feature": "ctxt_per_sec", "meta": {}}, + ] + gap = report_gap(attempts, kernel, slack_sec=30) + assert gap["attempt_without_kernel"] == 1 + assert gap["kernel_without_attempt"] == 1 + assert gap["attempt_with_kernel"] == 0 + + +def test_http_tail_starts_at_end(tmp_path): + log = tmp_path / "http.json" + line = ( + '{"time_iso8601":"2026-08-15T12:00:00+00:00","remote_addr":"1.2.3.4",' + '"request_method":"GET","request_uri":"/","status":200}\n' + ) + log.write_text(line) + follow = HttpLogTail(str(log)) + assert follow.read_new() == [] + replay = HttpLogTail(str(log), start="begin") + assert len(replay.read_new()) == 1 + log.write_text(line + line) + fresh = follow.read_new() + assert len(fresh) == 1 + + +def test_dna_mapper_keeps_http_classes_separate(): + rows = [ + {"feature": "http_attempt:scanner", "source": "stage9_http", "severity": "medium", + "message": "HTTP attempt scanner", "position": 0.22, "score": 0.8, + "meta": {"cls": "scanner", "src_ip": "203.0.113.9", "why": "scanner:14 x404"}}, + {"feature": "http_attempt:sqli", "source": "stage9_http", "severity": "high", + "message": "HTTP attempt sqli", "position": 0.32, "score": 0.9, + "meta": {"cls": "sqli", "src_ip": "203.0.113.10", "why": "pattern:sqli"}}, + {"feature": "http_attempt:scanner", "source": "stage9_http", "severity": "medium", + "message": "older scanner", "position": 0.22, "score": 0.7, "meta": {}}, + {"feature": "ctxt_per_sec", "source": "stage2_isoforest", "severity": "medium", + "message": "forest", "position": 0.18, "score": 0.2, "meta": {}}, + ] + muts = _ml_anomalies_to_mutations(rows) + types = [m["type"] for m in muts] + assert types.count("http_attempt:scanner") == 1 + assert "http_attempt:sqli" in types + assert "ctxt_per_sec" in types + scanner = next(m for m in muts if m["type"] == "http_attempt:scanner") + assert scanner["source"] == "ml" + assert scanner["ml_source"] == "stage9_http" + assert "113.9" in scanner["description"] or "why" in scanner diff --git a/tests/test_ml_store_connect.py b/tests/test_ml_store_connect.py new file mode 100644 index 0000000..0d73081 --- /dev/null +++ b/tests/test_ml_store_connect.py @@ -0,0 +1,32 @@ +"""Gunicorn + ProtectHome must not make libpq open ~/.postgresql/*.crt.""" + +from kernel_ai.ml.store import ( + _NO_CLIENT_CERT, + _NO_CLIENT_KEY, + connect, + connect_kwargs, +) + + +def test_connect_kwargs_do_not_use_home_client_cert(): + kwargs = connect_kwargs(autocommit=True) + assert kwargs["autocommit"] is True + assert kwargs["sslcert"] == _NO_CLIENT_CERT + assert kwargs["sslkey"] == _NO_CLIENT_KEY + assert "/home/" not in kwargs["sslcert"] + assert "/home/" not in kwargs["sslkey"] + + +def test_connect_passes_non_home_cert_paths(monkeypatch): + seen = {} + + def fake_connect(dsn, **kwargs): + seen["dsn"] = dsn + seen["kwargs"] = kwargs + return object() + + monkeypatch.setattr("kernel_ai.ml.store.psycopg.connect", fake_connect) + connect("postgresql://example/db") + assert seen["dsn"] == "postgresql://example/db" + assert seen["kwargs"]["sslcert"] == _NO_CLIENT_CERT + assert seen["kwargs"]["sslkey"] == _NO_CLIENT_KEY diff --git a/tests/test_ml_thresholds.py b/tests/test_ml_thresholds.py new file mode 100644 index 0000000..02e229f --- /dev/null +++ b/tests/test_ml_thresholds.py @@ -0,0 +1,35 @@ +"""Live-host calibration for Stage 1 z and Stage 2 IsolationForest emit.""" + +from types import SimpleNamespace + +from kernel_ai.ml.baseline import Score +from kernel_ai.ml.config import MLConfig +from kernel_ai.ml.worker import _build_anomalies + + +def _score(name: str, z: float) -> Score: + return Score(name=name, value=100.0 + z, mean=100.0, std=1.0, z=z, warm=False) + + +def test_calibrated_defaults(): + cfg = MLConfig() + assert cfg.z_warn == 6.0 + assert cfg.z_crit == 7.0 + assert cfg.if_cooldown_sec == 45.0 + assert cfg.if_min_score == 0.10 + + +def test_stage1_drops_poll_band_keeps_real_spikes(): + cfg = SimpleNamespace(z_warn=6.0, z_crit=7.0, alpha=0.03) + scores = { + "ctxt_per_sec": _score("ctxt_per_sec", 4.5), + "tcp_retrans_per_sec": _score("tcp_retrans_per_sec", 5.2), + "pgfault_per_sec": _score("pgfault_per_sec", 10.2), + "cpu_busy_pct": _score("cpu_busy_pct", 6.5), + } + out = _build_anomalies(scores, cfg) + features = {row["feature"] for row in out} + assert features == {"pgfault_per_sec", "cpu_busy_pct"} + by_feat = {row["feature"]: row for row in out} + assert by_feat["pgfault_per_sec"]["severity"] == "high" + assert by_feat["cpu_busy_pct"]["severity"] == "medium" diff --git a/tests/test_process_inspect_service.py b/tests/test_process_inspect_service.py index 4812f3a..70695f8 100644 --- a/tests/test_process_inspect_service.py +++ b/tests/test_process_inspect_service.py @@ -3,6 +3,15 @@ from kernel_ai.services import process_inspect as svc +def test_only_real_shared_memory_counts_as_an_ipc_channel(): + assert svc._is_ipc_shared_mapping("/dev/shm/pulse-shm-1") + assert svc._is_ipc_shared_mapping("/memfd:wayland-cursor") + assert svc._is_ipc_shared_mapping("/SYSV00000000") + assert not svc._is_ipc_shared_mapping("/dev/dri/card0") + assert not svc._is_ipc_shared_mapping("/usr/share/fonts/truetype/x") + assert not svc._is_ipc_shared_mapping("/var/cache/fontconfig/abc") + + def test_get_ipc_links_summary_empty_proc(monkeypatch): monkeypatch.setattr(svc.os, "listdir", lambda path: [] if path == "/proc" else []) monkeypatch.setattr(svc.psutil, "net_connections", lambda kind: []) @@ -14,6 +23,58 @@ def test_get_ipc_links_summary_empty_proc(monkeypatch): assert out["stats"]["shared_tcp_socket_inodes"] == 0 +def test_family_label_and_peer_split(): + assert svc._family_label("AddressFamily.AF_UNIX", "SOCK_STREAM") == "unix" + assert svc._family_label("AddressFamily.AF_INET", "SocketKind.SOCK_STREAM") == "tcp" + assert svc._family_label("AddressFamily.AF_INET6", "SOCK_DGRAM") == "udp6" + assert svc._split_ip_port("127.0.0.1:8000") == ("127.0.0.1", 8000) + assert svc._split_ip_port("/run/foo.sock") is None + assert svc._hex_ipv4("0100007F") == "127.0.0.1" + + +def test_annotate_connection_peers_pairs_loopback(monkeypatch): + class _Addr: + def __init__(self, ip, port): + self.ip = ip + self.port = port + + class _Conn: + def __init__(self, pid, lip, lport, rip, rport): + self.pid = pid + self.laddr = _Addr(lip, lport) + self.raddr = _Addr(rip, rport) + + class _Proc: + def __init__(self, pid, name): + self.info = {"pid": pid, "name": name} + + monkeypatch.setattr( + svc.psutil, + "process_iter", + lambda attrs: [_Proc(10, "gunicorn"), _Proc(20, "sshd")], + ) + monkeypatch.setattr( + svc.psutil, + "net_connections", + lambda kind: [ + _Conn(10, "127.0.0.1", 8000, "127.0.0.1", 44312), + _Conn(20, "127.0.0.1", 44312, "127.0.0.1", 8000), + ], + ) + rows = svc._annotate_connection_peers( + [ + { + "local_address": "127.0.0.1:8000", + "remote_address": "127.0.0.1:44312", + "peer_pid": None, + "peer_name": None, + } + ] + ) + assert rows[0]["peer_pid"] == 20 + assert rows[0]["peer_name"] == "sshd" + + def test_get_process_fds_info_basic(monkeypatch): class _FakeProc: def num_fds(self): diff --git a/tests/test_runqueue_service.py b/tests/test_runqueue_service.py new file mode 100644 index 0000000..7dccf0d --- /dev/null +++ b/tests/test_runqueue_service.py @@ -0,0 +1,198 @@ +"""Tests for ``kernel_ai.services.runqueue``.""" + +import json + +import pytest + +from kernel_ai.services import runqueue as svc + + +def _task(comm="worker", state="R", cpu=0, cgroup="/system.slice/app.service", + vlag=0.5, deadline=100.4, avg=100.0, prio=120, current=False, slice_ms=0.7): + row = { + "comm": comm, "state": state, "cpu": cpu, "cgroup": cgroup, + "vlag_ms": vlag, "deadline_v": deadline, "avg_vruntime": avg, + "eligible": vlag >= 0, "prio": prio, "slice_ms": slice_ms, + "switches": 12, "sum_exec_ms": 900.0, + } + if current: + row["current"] = True + return row + + +@pytest.fixture(autouse=True) +def _no_owner_lookup(monkeypatch, tmp_path): + """Keep the service off this host's real /proc for owner resolution.""" + monkeypatch.setattr(svc, "PROC", str(tmp_path / "proc")) + + +def _snapshot(tmp_path, monkeypatch, tasks, cpus=None, reader_tid=None, loadavg=None): + snap = tmp_path / "sched_debug.json" + payload = {"ts": 0, "tasks": tasks, "cpus": cpus or {"0": {"nr_running": len(tasks)}}} + if reader_tid is not None: + payload["reader_tid"] = reader_tid + snap.write_text(json.dumps(payload)) + monkeypatch.setattr(svc, "SCHED_DEBUG_SNAPSHOT", str(snap)) + + proc = tmp_path / "proc" + proc.mkdir(exist_ok=True) + (proc / "loadavg").write_text(loadavg or "0.52 0.31 0.14 2/431 90210\n") + + +def test_only_runnable_tasks_are_in_the_queue(tmp_path, monkeypatch): + _snapshot(tmp_path, monkeypatch, { + "10": _task(comm="gunicorn", state="R"), + "11": _task(comm="asleep", state="S"), + "12": _task(comm="waiting-on-disk", state="D"), + }) + + out = svc.describe() + + assert out["queued"] == 1 + assert [r["comm"] for r in out["cpus"][0]["queue"]] == ["gunicorn"] + # D is not queued for a CPU but every load average has counted it since 1993. + assert out["uninterruptible"] == 1 + + +def test_the_load_average_is_reported_with_the_kernels_live_count(tmp_path, monkeypatch): + _snapshot(tmp_path, monkeypatch, {"10": _task()}, loadavg="1.50 0.90 0.40 3/512 4242\n") + + load = svc.describe()["load"] + + assert (load["avg1"], load["avg5"], load["avg15"]) == (1.5, 0.9, 0.4) + assert load["running"] == 3 + assert load["total"] == 512 + + +def test_the_task_on_the_cpu_heads_its_queue_and_the_rest_go_by_deadline(tmp_path, monkeypatch): + _snapshot(tmp_path, monkeypatch, { + "10": _task(comm="late", deadline=100.9), + "11": _task(comm="running", deadline=100.8, current=True), + "12": _task(comm="soon", deadline=100.2), + }) + + queue = svc.describe()["cpus"][0]["queue"] + + assert [r["comm"] for r in queue] == ["running", "soon", "late"] + assert queue[1]["due_ms"] == 0.2 + + +def test_the_next_task_is_named_when_one_queue_holds_the_competition(tmp_path, monkeypatch): + _snapshot(tmp_path, monkeypatch, { + "10": _task(comm="running", current=True), + "11": _task(comm="soon", deadline=100.2), + "12": _task(comm="later", deadline=100.6), + }) + + nxt = svc.describe()["cpus"][0]["next"] + + assert nxt == {"tid": 11, "exact": True, "reason": None} + + +def test_an_ineligible_task_is_not_named_next(tmp_path, monkeypatch): + """Eligibility is the kernel's gate: a task in debt waits however near its + deadline is.""" + _snapshot(tmp_path, monkeypatch, { + "10": _task(comm="running", current=True), + "11": _task(comm="in-debt", deadline=100.1, vlag=-3.0), + "12": _task(comm="in-credit", deadline=100.5, vlag=2.0), + }) + + assert svc.describe()["cpus"][0]["next"]["tid"] == 12 + + +def test_across_cgroups_the_pick_is_offered_but_not_claimed_as_exact(tmp_path, monkeypatch): + """Each cgroup's cfs_rq keeps its own virtual clock. + + Comparing deadlines across two of them is not the comparison the kernel + makes, and the group deadlines that would decide it are not printed. + """ + _snapshot(tmp_path, monkeypatch, { + "10": _task(comm="running", current=True), + "11": _task(comm="ours", cgroup="/system.slice/app.service", deadline=100.2), + # Its own queue keeps its own virtual clock, and 8000 is as near to its + # own V as 100.2 is to ours. + "12": _task(comm="theirs", cgroup="/user.slice", deadline=8000.4, avg=8000.0), + }) + + nxt = svc.describe()["cpus"][0]["next"] + + assert nxt["exact"] is False + assert nxt["tid"] == 11 + assert "service queues first" in nxt["reason"] + + +def test_a_real_time_task_stops_the_question_being_answered(tmp_path, monkeypatch): + _snapshot(tmp_path, monkeypatch, { + "10": _task(comm="running", current=True), + "11": _task(comm="irq/24-nvme", prio=49), + }) + + nxt = svc.describe()["cpus"][0]["next"] + + assert nxt["exact"] is False + assert "real-time" in nxt["reason"] + + +def test_the_collector_is_marked_as_the_observer(tmp_path, monkeypatch): + """Reading the file is work, so the reader is usually the task on the CPU.""" + _snapshot(tmp_path, monkeypatch, { + "10": _task(comm="python3", current=True), + "11": _task(comm="gunicorn"), + }, reader_tid="10") + + out = svc.describe() + by_tid = {r["tid"]: r for r in out["cpus"][0]["queue"]} + + assert out["observer_tid"] == 10 + assert by_tid[10]["observer"] is True + assert by_tid[11]["observer"] is False + + +def test_queues_are_kept_apart_per_cpu(tmp_path, monkeypatch): + _snapshot(tmp_path, monkeypatch, { + "10": _task(comm="on-zero", cpu=0), + "11": _task(comm="on-one", cpu=1), + }, cpus={"0": {"nr_running": 1}, "1": {"nr_running": 1}}) + + cpus = svc.describe()["cpus"] + + assert [c["cpu"] for c in cpus] == [0, 1] + assert [len(c["queue"]) for c in cpus] == [1, 1] + + +def test_without_a_collector_the_queue_says_so_instead_of_being_empty(tmp_path, monkeypatch): + monkeypatch.setattr(svc, "SCHED_DEBUG_SNAPSHOT", str(tmp_path / "absent.json")) + proc = tmp_path / "proc" + proc.mkdir(exist_ok=True) + (proc / "loadavg").write_text("0.10 0.20 0.30 1/100 5\n") + + out = svc.describe() + + assert out["source"] == {"available": False, "reason": "no-collector"} + assert out["cpus"] == [] + assert out["scheduler"]["name"] is None + # The load average needs no privilege, so it is still answered. + assert out["load"]["avg1"] == 0.1 + + +def test_a_stale_snapshot_is_refused(tmp_path, monkeypatch): + _snapshot(tmp_path, monkeypatch, {"10": _task()}) + monkeypatch.setattr(svc, "SNAPSHOT_MAX_AGE_S", -1.0) + + out = svc.describe() + + assert out["source"]["available"] is False + assert out["source"]["reason"] == "stale" + + +def test_the_unit_is_read_out_of_the_cgroup_path(tmp_path, monkeypatch): + _snapshot(tmp_path, monkeypatch, { + "10": _task(cgroup="/system.slice/kernel-ai.service"), + "11": _task(cgroup="/"), + }) + + units = [r["unit"] for r in svc.describe()["cpus"][0]["queue"]] + + assert "kernel-ai.service" in units + assert None in units diff --git a/tests/test_sched_debug_collector.py b/tests/test_sched_debug_collector.py new file mode 100644 index 0000000..eebaa6e --- /dev/null +++ b/tests/test_sched_debug_collector.py @@ -0,0 +1,129 @@ +"""Tests for the root sched_debug collector. + +The collector is a standalone script run by systemd, not part of the package, +so it is loaded by path the same way the syscall collector loads the modules it +shares with the app. + +Every line in the fixture below is copied verbatim from +``/sys/kernel/debug/sched/debug`` on the production host (kernel 6.8, EEVDF), +including the two rows that break naive parsing: the task on the CPU, which the +kernel prefixes with ">R" instead of a space and a state letter, and rsyslog's +thread, whose name contains spaces. +""" + +import importlib.util +import os + +_PATH = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "deploy", "ebpf", "sched_debug_collector.py", +) +_spec = importlib.util.spec_from_file_location("sched_debug_collector", _PATH) +collector = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(collector) + + +DEBUG_TEXT = """cpu#0, 2299.918 MHz + .nr_running : 3 + .nr_switches : 1254336106 + +cfs_rq[0]:/system.slice/kernel-ai.service + .left_vruntime : 0.000001 + .avg_vruntime : 17852.068599 + .nr_running : 1 + +cfs_rq[0]:/user.slice + .avg_vruntime : 1049460.510581 + .nr_running : 1 + +runnable tasks: + S task PID tree-key switches prio wait-time sum-exec sum-sleep +------------------------------------------------------------------------------------------------------------- + Spool_workqueue_ 3 1428.137431 E 1428.824774 0.700000 0.025650 4 120 0.000000 0.025650 0.000000 0.000000 0 0 / + S rs:main Q:Reg 40319 11788.388078 E 11789.071062 0.700000 13523.318844 352958 120 0.000000 13523.318844 0.000000 0.000000 0 0 /system.slice/rsyslog.service + S gunicorn 160967 17852.182045 E 17852.768599 0.700000 4439.487427 37447 120 0.000000 4439.487427 0.000000 0.000000 0 0 /system.slice/kernel-ai.service + N gunicorn 160968 17853.000000 N 17853.700000 0.700000 4192.098178 33313 120 0.000000 4192.098178 0.000000 0.000000 0 0 /system.slice/kernel-ai.service +>R cat 164270 1049460.510581 E 1049460.842993 0.700000 2.771451 16 120 0.000000 2.771451 0.000000 0.000000 0 0 /user.slice + +""" + + +def _parsed(): + return collector.parse(DEBUG_TEXT) + + +def test_the_task_holding_the_cpu_is_not_dropped(): + """">R" replaces the state letter, so the row needs positional parsing.""" + tasks, cpus = _parsed() + + assert "164270" in tasks + assert tasks["164270"]["current"] is True + assert tasks["164270"]["comm"] == "cat" + assert cpus["0"]["current_tid"] == 164270 + + +def test_only_the_running_task_carries_the_flag(): + tasks, _ = _parsed() + + assert [tid for tid, row in tasks.items() if row.get("current")] == ["164270"] + + +def test_a_thread_whose_name_contains_spaces_is_kept(): + tasks, _ = _parsed() + + assert tasks["40319"]["comm"] == "rs:main Q:Reg" + assert tasks["40319"]["cgroup"] == "/system.slice/rsyslog.service" + + +def test_a_name_that_fills_the_field_does_not_swallow_the_state(): + tasks, _ = _parsed() + + assert tasks["3"]["comm"] == "pool_workqueue_" + assert tasks["3"]["state"] == "S" + + +def test_the_kernels_own_eligibility_verdict_is_carried_through(): + tasks, _ = _parsed() + + assert tasks["160967"]["eligible"] is True + assert tasks["160968"]["eligible"] is False + + +def test_lag_is_measured_against_the_fair_clock_of_the_same_cgroup(): + tasks, _ = _parsed() + + row = tasks["160967"] + assert row["avg_vruntime"] == 17852.069 + # V - vruntime: this thread is a hair ahead of fair, so it owes time. + assert row["vlag_ms"] == round(17852.068599 - 17852.182045, 3) + assert row["vlag_ms"] < 0 + + +def test_the_deadline_and_the_slice_come_from_the_row(): + tasks, _ = _parsed() + + row = tasks["160967"] + assert row["deadline_v"] == 17852.769 + assert row["slice_ms"] == 0.7 + assert row["prio"] == 120 + + +def test_the_runqueue_depth_is_the_one_printed_above_the_cfs_queues(): + """The cpu banner's nr_running counts every class, and comes first.""" + _, cpus = _parsed() + + assert cpus["0"]["nr_running"] == 3 + + +def test_the_table_header_is_not_read_as_a_task(): + tasks, _ = _parsed() + + assert "task" not in {row["comm"] for row in tasks.values()} + assert len(tasks) == 5 + + +def test_a_row_from_a_kernel_without_eevdf_is_refused(): + """Pre-6.6 rows have no eligibility column, so there is nothing to report.""" + old = " S systemd 1 53191.075522 314688 120 0.000000 /" + + assert collector.parse_row(old) is None diff --git a/tests/test_syscall_snapshot_collector.py b/tests/test_syscall_snapshot_collector.py new file mode 100644 index 0000000..0621f73 --- /dev/null +++ b/tests/test_syscall_snapshot_collector.py @@ -0,0 +1,106 @@ +"""Tests for the parsing the root syscall collector does. + +The collector is a standalone script run by systemd, not part of the package, +so it is loaded by path. Every fixture here is shaped like the real file it +stands for: ``/proc/locks`` with a blocked waiter, the ``tfd:`` lines of an +epoll set, and the octal open flags of a pipe's two ends. +""" + +import importlib.util +import os + +_PATH = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "deploy", "ebpf", "syscall_snapshot_collector.py", +) +_spec = importlib.util.spec_from_file_location("syscall_snapshot_collector", _PATH) +collector = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(collector) + + +LOCKS_TEXT = """1: POSIX ADVISORY WRITE 166 00:19:144 0 EOF +2: FLOCK ADVISORY WRITE 151877 ca:01:71958 0 EOF +2: -> FLOCK ADVISORY WRITE 151999 ca:01:71958 0 EOF +""" + +# /proc//fdinfo/ of a systemd epoll set, trimmed to three entries. +EPOLL_TEXT = """pos:\t0 +flags:\t02000002 +mnt_id:\t16 +ino:\t294 +tfd: 99 events: 18 data: 6546d479d470 pos:0 ino:826 sdev:8 +tfd: 24 events: 1 data: 6546d471bde0 pos:0 ino:fc9be38 sdev:8 +tfd: 133 events: 80000019 data: 6546d47b90b0 pos:0 ino:e47c349 sdev:8 +""" + + +def _fake_reads(monkeypatch, files): + monkeypatch.setattr(collector, "_read", lambda path: files.get(path, "")) + + +def test_a_blocked_waiter_is_read_as_one(monkeypatch): + """The "->" of a blocked task shifts every field after it by one.""" + _fake_reads(monkeypatch, {"/proc/locks": LOCKS_TEXT, + "/proc/166/comm": "multipathd", + "/proc/151877/comm": "snapd", + "/proc/151999/comm": "snap"}) + + rows = collector._locks() + + assert [(r["pid"], r["waiting"]) for r in rows] == [ + (166, False), (151877, False), (151999, True)] + assert rows[0]["kind"] == "POSIX" and rows[0]["mode"] == "WRITE" + # The holder and the task blocked behind it name the same file. + assert rows[1]["inode"] == rows[2]["inode"] == "ca:01:71958" + assert rows[2]["comm"] == "snap" + + +def test_the_descriptors_an_epoll_set_watches_are_listed(monkeypatch): + _fake_reads(monkeypatch, {"/proc/1/fdinfo/4": EPOLL_TEXT}) + links = {"/proc/1/fd/99": "socket:[826]", "/proc/1/fd/24": "anon_inode:[timerfd]"} + + def readlink(path): + if path not in links: + raise OSError(2, "No such file or directory") + return links[path] + + monkeypatch.setattr(collector.os, "readlink", readlink) + + watch = collector._epoll_watch("1", 4) + + assert watch["total"] == 3 + assert [w["fd"] for w in watch["watched"]] == [99, 24, 133] + # events are hex, and the third entry carries the edge-triggered bit. + assert [w["events"] for w in watch["watched"]] == [0x18, 0x1, 0x80000019] + assert watch["watched"][0]["target"] == "socket:[826]" + # A descriptor whose link cannot be read is kept, without a target. + assert watch["watched"][2]["target"] is None + + +def test_a_set_larger_than_the_cap_still_reports_its_size(monkeypatch): + lines = "\n".join( + f"tfd:{i:9d} events: 1 data: 0 pos:0 ino:1 sdev:8" for i in range(60)) + _fake_reads(monkeypatch, {"/proc/1/fdinfo/4": lines}) + monkeypatch.setattr(collector.os, "readlink", lambda path: "socket:[1]") + + watch = collector._epoll_watch("1", 4) + + assert watch["total"] == 60 + assert len(watch["watched"]) == collector.MAX_WATCHED + + +def test_a_thread_with_no_epoll_set_gets_nothing(monkeypatch): + _fake_reads(monkeypatch, {}) + + assert collector._epoll_watch("1", 4) is None + + +def test_the_two_ends_of_a_pipe_are_told_apart_by_open_flags(monkeypatch): + """The low two bits of the octal flags are the access mode.""" + _fake_reads(monkeypatch, { + "/proc/100/fdinfo/3": "pos:\t0\nflags:\t0100000\nmnt_id:\t14\n", + "/proc/200/fdinfo/1": "pos:\t0\nflags:\t0100001\nmnt_id:\t14\n", + }) + + assert collector._access_mode("100", "3") == "read" + assert collector._access_mode("200", "1") == "write" diff --git a/tests/test_syscalls_service.py b/tests/test_syscalls_service.py index 519b024..889d910 100644 --- a/tests/test_syscalls_service.py +++ b/tests/test_syscalls_service.py @@ -3,9 +3,23 @@ import json import time +import pytest + from kernel_ai.services import syscalls as svc +@pytest.fixture(autouse=True) +def _no_collector_on_this_host(monkeypatch): + """Keep the tests reading the machine they describe, not the one they run on. + + The service answers from the root collector's snapshot whenever a fresh one + exists. On a host where the collector happens to be running that snapshot + would win over the /proc each test sets up, so the path starts out pointing + at nothing and the tests about the snapshot point it back at their own. + """ + monkeypatch.setattr(svc, "_SYSCALLS_SNAPSHOT", "/nonexistent/kernel-ai/syscalls.json") + + def test_get_real_system_calls_non_linux_uses_fallback(monkeypatch): monkeypatch.setattr(svc.platform, "system", lambda: "Darwin") out = svc.get_real_system_calls( diff --git a/tests/test_system_view_service.py b/tests/test_system_view_service.py index 86a2142..d66e8da 100644 --- a/tests/test_system_view_service.py +++ b/tests/test_system_view_service.py @@ -10,10 +10,30 @@ def test_parse_cgroup_path_prefers_unified_entry(monkeypatch): def test_get_isolation_context_handles_empty_process_list(monkeypatch): + monkeypatch.setattr(svc, "_read_isolation_snapshot", lambda: None) monkeypatch.setattr(svc.psutil, "process_iter", lambda _fields: []) out = svc.get_isolation_context() assert "namespaces" in out assert out["processes_scanned"] == 0 + assert out["source"] == "self" + + +def test_get_isolation_context_prefers_fresh_collector_snapshot(monkeypatch): + snap = { + "ts": 1, + "source": "collector", + "namespaces": [{"id": "net", "worlds": [{"sample": ["nginx"]}]}], + "processes_scanned": 80, + "top_cgroups": [], + } + monkeypatch.setattr(svc, "_read_isolation_snapshot", lambda: snap) + out = svc.get_isolation_context() + assert out["source"] == "collector" + assert out["namespaces"][0]["worlds"][0]["sample"] == ["nginx"] + + +def test_rank_isolation_samples_puts_nginx_first(): + assert svc._rank_isolation_samples(["kthreadd", "nginx", "sshd"])[:2] == ["nginx", "sshd"] def test_read_namespace_inode_parses_inode(monkeypatch): diff --git a/tests/test_threads_service.py b/tests/test_threads_service.py new file mode 100644 index 0000000..b879fac --- /dev/null +++ b/tests/test_threads_service.py @@ -0,0 +1,233 @@ +"""Tests for ``kernel_ai.services.threads``.""" + +import json +import os + +from kernel_ai.services import threads as svc + + +def _thread(tmp_path, pid, tid, comm="worker", state="S", cpu=0, + ran_ns=0, waited_ns=0, turns=0, vol=0, forced=0): + """Write the four files the service reads for one thread.""" + task = tmp_path / str(pid) / "task" / str(tid) + task.mkdir(parents=True) + # /proc//stat: the CPU last used is field 39 of the line, which is the + # 37th field counting from the state. + tail = " ".join(["0"] * 35) + (task / "stat").write_text(f"{tid} ({comm}) {state} {tail} {cpu}\n") + (task / "schedstat").write_text(f"{ran_ns} {waited_ns} {turns}\n") + (task / "status").write_text( + f"Name:\t{comm}\nThreads:\t1\n" + f"voluntary_ctxt_switches:\t{vol}\n" + f"nonvoluntary_ctxt_switches:\t{forced}\n" + ) + (task / "comm").write_text(f"{comm}\n") + return task + + +def _snapshots(tmp_path, monkeypatch, parked=None, scheduled=None): + """Point the service at collector snapshots written just now.""" + syscalls = tmp_path / "syscalls.json" + tasks = tmp_path / "tasks.json" + sched = tmp_path / "sched_debug.json" + syscalls.write_text(json.dumps({"ts": 0, "syscalls": []})) + sched.write_text(json.dumps({"ts": 0, "tasks": scheduled or {}})) + monkeypatch.setattr(svc, "SYSCALLS_SNAPSHOT", str(syscalls)) + monkeypatch.setattr(svc, "SCHED_DEBUG_SNAPSHOT", str(sched)) + if parked is None: + monkeypatch.setattr(svc, "TASKS_SNAPSHOT", str(tmp_path / "absent.json")) + return + tasks.write_text(json.dumps({"ts": 0, "tasks": parked})) + monkeypatch.setattr(svc, "TASKS_SNAPSHOT", str(tasks)) + + +def test_a_process_that_is_gone_is_reported_as_gone(tmp_path, monkeypatch): + monkeypatch.setattr(svc, "PROC", str(tmp_path)) + assert svc.describe(4242).get("error") == "no such process" + + +def test_every_task_of_the_process_becomes_a_thread(tmp_path, monkeypatch): + _thread(tmp_path, 100, 100, comm="gunicorn") + _thread(tmp_path, 100, 101, comm="pool-1") + _thread(tmp_path, 100, 102, comm="jemalloc_bg") + monkeypatch.setattr(svc, "PROC", str(tmp_path)) + (tmp_path / "100" / "comm").write_text("gunicorn\n") + + out = svc.describe(100) + + assert out["thread_count"] == 3 + assert {t["name"] for t in out["threads"]} == {"gunicorn", "pool-1", "jemalloc_bg"} + assert [t["leader"] for t in out["threads"] if t["tid"] == 100] == [True] + + +def test_the_thread_on_a_cpu_is_listed_first(tmp_path, monkeypatch): + _thread(tmp_path, 100, 100, ran_ns=9_000_000_000) + _thread(tmp_path, 100, 101, state="R", ran_ns=1_000_000) + monkeypatch.setattr(svc, "PROC", str(tmp_path)) + + out = svc.describe(100) + + assert out["threads"][0]["tid"] == 101 + assert out["threads"][0]["state_label"] == "on cpu or queued" + assert out["totals"]["on_cpu"] == 1 + + +def test_time_queued_is_reported_beside_time_running(tmp_path, monkeypatch): + _thread(tmp_path, 100, 100, ran_ns=4_058_474_400, waited_ns=5_978_895_548, turns=31040) + monkeypatch.setattr(svc, "PROC", str(tmp_path)) + + thread = svc.describe(100)["threads"][0] + + assert thread["ran_ms"] == 4058.5 + assert thread["waited_ms"] == 5978.9 + assert thread["turns"] == 31040 + + +def test_the_dossier_keeps_seeing_the_leaders_own_switch_counters(tmp_path, monkeypatch): + _thread(tmp_path, 100, 100, vol=11, forced=3) + _thread(tmp_path, 100, 101, vol=70, forced=7) + monkeypatch.setattr(svc, "PROC", str(tmp_path)) + + out = svc.describe(100) + + assert out["voluntary_ctxt_switches"] == 11 + assert out["nonvoluntary_ctxt_switches"] == 3 + assert out["totals"]["voluntary"] == 81 + assert out["totals"]["forced"] == 10 + + +def test_the_call_a_thread_is_parked_in_comes_from_the_collector(tmp_path, monkeypatch): + _thread(tmp_path, 100, 100) + _thread(tmp_path, 100, 101) + monkeypatch.setattr(svc, "PROC", str(tmp_path)) + _snapshots(tmp_path, monkeypatch, parked={ + "100": {"pid": 100, "nr": 232, "name": "epoll_wait", "wchan": "ep_poll", + "fd": 12, "fd_target": "anon_inode:[eventpoll]"}, + }) + + out = svc.describe(100) + by_tid = {t["tid"]: t for t in out["threads"]} + + assert by_tid[100]["parked_in"] == "epoll_wait" + assert by_tid[100]["wchan"] == "ep_poll" + assert by_tid[100]["fd_target"] == "anon_inode:[eventpoll]" + # The other thread is running or in no call; nothing is invented for it. + assert "parked_in" not in by_tid[101] + assert out["sources"]["parked_in"]["available"] is True + + +def test_the_scheduler_verdict_comes_from_the_debugfs_snapshot(tmp_path, monkeypatch): + _thread(tmp_path, 100, 100, state="R") + monkeypatch.setattr(svc, "PROC", str(tmp_path)) + _snapshots(tmp_path, monkeypatch, scheduled={ + "100": {"eligible": True, "vlag_ms": 16.905, "slice_ms": 0.7, "prio": 120, + "current": True, "deadline_v": 4103.5, "avg_vruntime": 4103.1}, + }) + + out = svc.describe(100) + thread = out["threads"][0] + + assert thread["eligible"] is True + assert thread["vlag_ms"] == 16.905 + assert thread["nice"] == 0 + assert thread["on_cpu"] is True + # The deadline is worth having as the distance to the runqueue's clock: it + # is what EEVDF compares, and the raw virtual timestamp says nothing. + assert thread["due_ms"] == 0.4 + assert out["scheduler"] == {"name": "EEVDF", "slice_ms": 0.7} + + +def test_a_sleeping_thread_is_given_no_lag_and_no_deadline(tmp_path, monkeypatch): + """The kernel prints its task table per CPU, not per runqueue. + + A thread that went to sleep hours ago is still in that table, holding the + vruntime it was frozen at, so V minus that grows for as long as it sleeps. + Reporting it would put "owed 35,000,000 ms" against an idle kworker. + """ + _thread(tmp_path, 100, 100, state="S") + _thread(tmp_path, 100, 101, state="R") + monkeypatch.setattr(svc, "PROC", str(tmp_path)) + _snapshots(tmp_path, monkeypatch, scheduled={ + "100": {"eligible": True, "vlag_ms": 35210010.846, "prio": 120, + "slice_ms": 0.7, "deadline_v": 1.0, "avg_vruntime": 4103.1}, + "101": {"eligible": True, "vlag_ms": 0.9, "prio": 120, "slice_ms": 0.7, + "deadline_v": 4103.4, "avg_vruntime": 4103.1}, + }) + + by_tid = {t["tid"]: t for t in svc.describe(100)["threads"]} + + assert "vlag_ms" not in by_tid[100] + assert "eligible" not in by_tid[100] + assert "due_ms" not in by_tid[100] + assert "on_cpu" not in by_tid[100] + # The properties that hold whether or not it is queued still come through. + assert by_tid[100]["nice"] == 0 + assert by_tid[101]["vlag_ms"] == 0.9 + assert by_tid[101]["due_ms"] == 0.3 + + +def test_a_cfs_kernel_is_not_reported_as_eevdf(tmp_path, monkeypatch): + """A pre-6.6 kernel prints a task table with no eligibility and no deadline. + + The collector parses no rows out of it, and an empty table is the evidence: + the card is told the scheduler is unknown rather than shown an EEVDF column + it can never fill. + """ + _thread(tmp_path, 100, 100, state="R") + monkeypatch.setattr(svc, "PROC", str(tmp_path)) + _snapshots(tmp_path, monkeypatch, scheduled={}) + + out = svc.describe(100) + + assert out["sources"]["scheduler"]["available"] is True + assert out["scheduler"]["name"] is None + assert "eligible" not in out["threads"][0] + + +def test_without_a_collector_the_columns_are_absent_and_named_as_such(tmp_path, monkeypatch): + _thread(tmp_path, 100, 100) + monkeypatch.setattr(svc, "PROC", str(tmp_path)) + monkeypatch.setattr(svc, "SYSCALLS_SNAPSHOT", str(tmp_path / "nothing.json")) + monkeypatch.setattr(svc, "TASKS_SNAPSHOT", str(tmp_path / "nothing.json")) + monkeypatch.setattr(svc, "SCHED_DEBUG_SNAPSHOT", str(tmp_path / "nothing.json")) + + out = svc.describe(100) + + assert "parked_in" not in out["threads"][0] + assert "eligible" not in out["threads"][0] + assert out["sources"]["parked_in"]["reason"] == "no-collector" + assert out["scheduler"]["name"] is None + + +def test_a_collector_from_before_the_thread_index_says_so(tmp_path, monkeypatch): + _thread(tmp_path, 100, 100) + monkeypatch.setattr(svc, "PROC", str(tmp_path)) + # The syscall collector is publishing, but no per-thread index beside it. + _snapshots(tmp_path, monkeypatch) + + out = svc.describe(100) + + assert out["sources"]["parked_in"] == {"available": False, "reason": "no-thread-index"} + + +def test_a_stale_snapshot_is_refused(tmp_path, monkeypatch): + _thread(tmp_path, 100, 100) + monkeypatch.setattr(svc, "PROC", str(tmp_path)) + _snapshots(tmp_path, monkeypatch, parked={"100": {"name": "read"}}) + old = os.path.getmtime(tmp_path / "tasks.json") - 3600 + os.utime(tmp_path / "tasks.json", (old, old)) + + out = svc.describe(100) + + assert out["sources"]["parked_in"]["reason"] == "stale" + assert "parked_in" not in out["threads"][0] + + +def test_a_thread_that_exits_mid_read_is_skipped_not_faked(tmp_path, monkeypatch): + _thread(tmp_path, 100, 100) + (tmp_path / "100" / "task" / "101").mkdir() # a task dir with no files left + monkeypatch.setattr(svc, "PROC", str(tmp_path)) + + out = svc.describe(100) + + assert [t["tid"] for t in out["threads"]] == [100] diff --git a/tests/test_waits_service.py b/tests/test_waits_service.py new file mode 100644 index 0000000..839b947 --- /dev/null +++ b/tests/test_waits_service.py @@ -0,0 +1,366 @@ +"""Tests for ``kernel_ai.services.waits``.""" + +import json +import os + +import pytest + +from kernel_ai.services import waits as svc + + +def _thread(tmp_path, pid, tid, comm="worker", state="S"): + task = tmp_path / str(pid) / "task" / str(tid) + task.mkdir(parents=True, exist_ok=True) + tail = " ".join(["0"] * 35) + (task / "stat").write_text(f"{tid} ({comm}) {state} {tail} 0\n") + (task / "comm").write_text(f"{comm}\n") + (tmp_path / str(pid) / "comm").write_text(f"{comm}\n") + return task + + +def _snapshots(tmp_path, monkeypatch, parked=None, pipes=None, locks=None): + tasks = tmp_path / "tasks.json" + endpoints = tmp_path / "endpoints.json" + tasks.write_text(json.dumps({"ts": 0, "tasks": parked or {}})) + endpoints.write_text(json.dumps({"ts": 0, "pipes": pipes or {}, "locks": locks or []})) + monkeypatch.setattr(svc, "TASKS_SNAPSHOT", str(tasks)) + monkeypatch.setattr(svc, "ENDPOINTS_SNAPSHOT", str(endpoints)) + + +@pytest.fixture(autouse=True) +def _proc(tmp_path, monkeypatch): + monkeypatch.setattr(svc, "PROC", str(tmp_path)) + # A live wakeup snapshot on this host must not leak into these tests. + from kernel_ai.services import wakeups as wakeups_service + monkeypatch.setattr(wakeups_service, "SNAPSHOT", str(tmp_path / "no-wakeups.json")) + + +def test_a_thread_that_is_gone_is_reported_as_gone(tmp_path, monkeypatch): + _snapshots(tmp_path, monkeypatch) + assert svc.describe(4242, 4242).get("error") == "no such thread" + + +def test_threads_on_the_same_word_are_one_group(tmp_path, monkeypatch): + for tid in (100, 101, 102): + _thread(tmp_path, 100, tid, comm="gunicorn") + _thread(tmp_path, 100, 103, comm="gunicorn", state="R") + _snapshots(tmp_path, monkeypatch, parked={ + "100": {"pid": 100, "name": "futex", "word": "0x7f10", "op": 393, "val": 0}, + "101": {"pid": 100, "name": "futex", "word": "0x7f10", "op": 393, "val": 0}, + "102": {"pid": 100, "name": "futex", "word": "0x7f99", "op": 128, "val": 2}, + }) + + out = svc.describe(100, 101) + on = out["waiting_on"] + + assert on["kind"] == "futex" + assert [w["tid"] for w in on["waiters"]] == [100, 101] + assert [w["self"] for w in on["waiters"]] == [False, True] + # The thread on another word and the one that is not waiting at all could + # both be holding this lock; only the running one can be holding it and + # doing something with it right now. + assert on["candidates"]["total"] == 2 + assert {c["tid"] for c in on["candidates"]["sample"]} == {102, 103} + assert [c["tid"] for c in on["candidates"]["running"]] == [103] + + +def test_a_sleeping_thread_is_not_ruled_out_as_the_holder(tmp_path, monkeypatch): + """A lock held across a blocking call is exactly how deadlocks read.""" + _thread(tmp_path, 100, 100) + _thread(tmp_path, 100, 101, state="S") + _snapshots(tmp_path, monkeypatch, parked={ + "100": {"pid": 100, "name": "futex", "word": "0x7f10", "op": 128}, + "101": {"pid": 100, "name": "epoll_wait", "fd": 4}, + }) + + pool = svc.describe(100, 100)["waiting_on"]["candidates"] + + assert pool["total"] == 1 + assert pool["running"] == [] + assert [c["parked_in"] for c in pool["sample"]] == ["epoll_wait"] + + +def test_the_same_address_in_another_process_is_another_lock(tmp_path, monkeypatch): + """A private futex is keyed by address inside one address space.""" + _thread(tmp_path, 100, 100, comm="gunicorn") + _thread(tmp_path, 200, 200, comm="gunicorn") + _snapshots(tmp_path, monkeypatch, parked={ + "100": {"pid": 100, "name": "futex", "word": "0x7f10", "op": 393}, + "200": {"pid": 200, "name": "futex", "word": "0x7f10", "op": 393}, + }) + + on = svc.describe(100, 100)["waiting_on"] + + assert [w["tid"] for w in on["waiters"]] == [100] + assert on["scope"] == "this process only" + + +def test_the_operation_is_decoded_into_its_flags(tmp_path, monkeypatch): + _thread(tmp_path, 100, 100) + _snapshots(tmp_path, monkeypatch, parked={ + "100": {"pid": 100, "name": "futex", "word": "0x7f10", "op": 393, "val": 0}, + }) + + op = svc.describe(100, 100)["waiting_on"]["op"] + + # 393 = FUTEX_WAIT_BITSET | FUTEX_PRIVATE_FLAG | FUTEX_CLOCK_REALTIME + assert op["cmd"] == 9 + assert op["name"] == "FUTEX_WAIT_BITSET" + assert op["private"] is True + assert op["realtime"] is True + assert op["pi"] is False + + +def test_a_plain_contended_mutex_is_told_apart_from_a_condition_variable(tmp_path, monkeypatch): + _thread(tmp_path, 100, 100) + _snapshots(tmp_path, monkeypatch, parked={ + "100": {"pid": 100, "name": "futex", "word": "0x7f10", "op": 128, "val": 2}, + }) + + op = svc.describe(100, 100)["waiting_on"]["op"] + + assert (op["cmd"], op["name"], op["private"]) == (0, "FUTEX_WAIT", True) + assert op["realtime"] is False + + +def test_the_owner_of_an_ordinary_futex_is_never_claimed(tmp_path, monkeypatch): + """Userspace takes the lock without telling the kernel, so no one knows.""" + _thread(tmp_path, 100, 100) + _snapshots(tmp_path, monkeypatch, parked={ + "100": {"pid": 100, "name": "futex", "word": "0x7f10", "op": 128}, + }) + + owner = svc.describe(100, 100)["waiting_on"]["owner"] + + assert owner["known"] is False + assert "userspace" in owner["why"] + + +def test_a_priority_inheriting_futex_says_where_its_owner_lives(tmp_path, monkeypatch): + _thread(tmp_path, 100, 100) + _snapshots(tmp_path, monkeypatch, parked={ + "100": {"pid": 100, "name": "futex", "word": "0x7f10", "op": 134}, + }) + + on = svc.describe(100, 100)["waiting_on"] + + assert on["op"]["pi"] is True + assert on["owner"]["known"] is False + assert "word itself" in on["owner"]["why"] + + +def test_the_far_end_of_a_pipe_is_named(tmp_path, monkeypatch): + _thread(tmp_path, 100, 100, comm="rsyslogd") + _snapshots(tmp_path, monkeypatch, parked={ + "100": {"pid": 100, "name": "read", "fd": 5, "fd_target": "pipe:[4242]"}, + }, pipes={ + "4242": { + "readers": [{"pid": 100, "fd": 5, "comm": "rsyslogd"}], + "writers": [{"pid": 900, "fd": 1, "comm": "systemd-journal"}], + }, + }) + + on = svc.describe(100, 100)["waiting_on"] + + assert on["kind"] == "pipe" + assert on["direction"] == "reading" + assert on["other_end"] == [{"pid": 900, "fd": 1, "comm": "systemd-journal"}] + assert on["same_end"] == [] + + +def test_a_pipe_whose_far_end_is_closed_says_nothing_instead_of_inventing(tmp_path, monkeypatch): + _thread(tmp_path, 100, 100) + _snapshots(tmp_path, monkeypatch, parked={ + "100": {"pid": 100, "name": "read", "fd": 3, "fd_target": "pipe:[7]"}, + }, pipes={"7": {"readers": [{"pid": 100, "fd": 3, "comm": "worker"}], "writers": []}}) + + on = svc.describe(100, 100)["waiting_on"] + + assert on["other_end"] == [] + assert on["direction"] == "reading" + + +def test_inherited_descriptors_on_the_same_end_are_kept_apart(tmp_path, monkeypatch): + """A child holding the same read end is not the other side of the pipe.""" + _thread(tmp_path, 100, 100, comm="mlflow") + _snapshots(tmp_path, monkeypatch, parked={ + "100": {"pid": 100, "name": "read", "fd": 0, "fd_target": "pipe:[11]"}, + }, pipes={ + "11": { + "readers": [{"pid": 100, "fd": 0, "comm": "mlflow"}, + {"pid": 101, "fd": 0, "comm": "python"}], + "writers": [{"pid": 102, "fd": 1, "comm": "shell"}], + }, + }) + + on = svc.describe(100, 100)["waiting_on"] + + assert [r["pid"] for r in on["other_end"]] == [102] + assert [r["pid"] for r in on["same_end"]] == [101] + + +def test_an_epoll_set_names_what_it_watches(tmp_path, monkeypatch): + _thread(tmp_path, 100, 100, comm="systemd") + _snapshots(tmp_path, monkeypatch, parked={ + "100": {"pid": 100, "name": "epoll_wait", "fd": 4, "watching": {"total": 3, "watched": [ + {"fd": 9, "events": 0x1, "target": "socket:[555]"}, + {"fd": 8, "events": 0x19, "target": "anon_inode:[timerfd]"}, + {"fd": 7, "events": 0x1, "target": "pipe:[77]"}, + ]}}, + }) + monkeypatch.setattr(svc, "socket_labels", lambda: {"555": "unix /run/systemd/journal/stdout"}) + + on = svc.describe(100, 100)["waiting_on"] + + assert on["kind"] == "epoll" + assert on["total"] == 3 + assert on["kinds"] == {"socket": 1, "timer": 1, "pipe": 1} + by_fd = {row["fd"]: row for row in on["watched"]} + assert by_fd[9]["label"] == "unix /run/systemd/journal/stdout" + assert by_fd[9]["waiting_for"] == "data" + assert by_fd[8]["label"] == "timer" + assert by_fd[7]["label"] == "pipe:[77]" + + +def test_the_same_thing_watched_many_times_is_counted_once(tmp_path, monkeypatch): + _thread(tmp_path, 100, 100, comm="systemd") + watched = [{"fd": 80 + i, "events": 0x18, "target": "socket:[555]"} for i in range(4)] + _snapshots(tmp_path, monkeypatch, parked={ + "100": {"pid": 100, "name": "epoll_wait", "fd": 4, + "watching": {"total": 40, "watched": watched}}, + }) + monkeypatch.setattr(svc, "socket_labels", lambda: {"555": "unix /run/systemd/journal/stdout"}) + + on = svc.describe(100, 100)["waiting_on"] + + assert len(on["watched"]) == 1 + assert on["watched"][0]["count"] == 4 + assert on["watched"][0]["fd"] == 80 + # The set is larger than the sample the collector kept, and says so. + assert on["total"] == 40 and on["shown"] == 4 + + +def test_socket_addresses_come_out_of_proc_net(tmp_path, monkeypatch): + net = tmp_path / "net" + net.mkdir() + (net / "unix").write_text( + "Num RefCount Protocol Flags Type St Inode Path\n" + "ffff: 00000003 00000000 00000000 0001 03 4242 /run/dbus/system_bus_socket\n") + (net / "tcp").write_text( + " sl local_address rem_address st tx_queue rx_queue tr tm->when retrnsmt uid timeout inode\n" + " 0: 0100007F:1F90 00000000:0000 0A 00000000:00000000 00:00000000" + " 00000000 1000 0 900 1 0000000000000000 100 0 0 10 0\n" + " 1: 0100007F:1F90 0100007F:C350 01 00000000:00000000 00:00000000" + " 00000000 1000 0 901 1 0000000000000000 20 4 30 10 -1\n") + monkeypatch.setattr(svc, "PROC", str(tmp_path)) + monkeypatch.chdir(tmp_path) + real_read = svc._read + monkeypatch.setattr(svc, "_read", lambda p: real_read( + str(net / os.path.basename(p)) if p.startswith("/proc/net/") else p)) + + labels = svc.socket_labels() + + assert labels["4242"] == "unix /run/dbus/system_bus_socket" + assert labels["900"] == "tcp listening on 127.0.0.1:8080" + assert labels["901"] == "tcp 127.0.0.1:8080 to 127.0.0.1:50000" + + +def test_file_locks_of_the_process_are_carried_along(tmp_path, monkeypatch): + _thread(tmp_path, 100, 100) + _snapshots(tmp_path, monkeypatch, parked={}, locks=[ + {"id": "1", "kind": "FLOCK", "mode": "WRITE", "pid": 100, "inode": "ca:01:7", "waiting": False}, + {"id": "1", "kind": "FLOCK", "mode": "WRITE", "pid": 200, "inode": "ca:01:7", "waiting": True}, + ]) + + out = svc.describe(100, 100) + + assert [row["pid"] for row in out["locks"]] == [100] + assert out["waiting_on"] is None + + +def test_without_the_collector_the_answer_is_missing_not_guessed(tmp_path, monkeypatch): + _thread(tmp_path, 100, 100) + monkeypatch.setattr(svc, "TASKS_SNAPSHOT", str(tmp_path / "nothing.json")) + monkeypatch.setattr(svc, "ENDPOINTS_SNAPSHOT", str(tmp_path / "nothing.json")) + + out = svc.describe(100, 100) + + assert out["waiting_on"] is None + assert out["call"] is None + assert out["sources"]["parked_in"]["reason"] == "no-collector" + assert out["sources"]["endpoints"]["reason"] == "no-collector" + assert out["seen_waking"]["available"] is False + + +def _wakeups(tmp_path, monkeypatch, edges): + from kernel_ai.services import wakeups as wakeups_service + path = tmp_path / "wakeups.json" + path.write_text(json.dumps({ + "ts": 0, "window_s": 0.25, "events": 10, "lost": 0, + "distinct_edges": len(edges), "contexts": {"task": 10}, + "observer_tid": 999, "edges": edges, "wakers": [], "wakees": [], + })) + monkeypatch.setattr(wakeups_service, "SNAPSHOT", str(path)) + + +def test_a_waker_seen_in_the_window_is_named_but_not_called_the_owner(tmp_path, monkeypatch): + """The kernel still does not record the holder; the window only saw a release.""" + _thread(tmp_path, 100, 100) + _thread(tmp_path, 100, 101, comm="worker", state="R") + _snapshots(tmp_path, monkeypatch, parked={ + "100": {"pid": 100, "name": "futex", "word": "0x7f10", "op": 128}, + }) + _wakeups(tmp_path, monkeypatch, [{ + "waker_tid": 101, "waker_comm": "worker", "waker_pid": 100, "waker_kernel": False, + "tid": 100, "comm": "worker", "pid": 100, "kernel": False, + "count": 3, "contexts": {"task": 3}, "new": False, + }]) + + out = svc.describe(100, 100) + seen = out["seen_waking"] + owner = out["waiting_on"]["owner"] + + assert owner["known"] is False + assert seen["available"] is True + assert seen["wakers"][0]["tid"] == 101 + assert seen["wakers"][0]["count"] == 3 + assert seen["wakers"][0]["of"] == [100] + + +def test_a_waker_of_a_sibling_on_the_same_word_is_the_same_observation(tmp_path, monkeypatch): + """Whoever woke another waiter on this word released or signaled it.""" + _thread(tmp_path, 100, 100) + _thread(tmp_path, 100, 101) + _thread(tmp_path, 100, 102, comm="holder", state="R") + _snapshots(tmp_path, monkeypatch, parked={ + "100": {"pid": 100, "name": "futex", "word": "0x7f10", "op": 128}, + "101": {"pid": 100, "name": "futex", "word": "0x7f10", "op": 128}, + }) + _wakeups(tmp_path, monkeypatch, [{ + "waker_tid": 102, "waker_comm": "holder", "waker_pid": 100, "waker_kernel": False, + "tid": 101, "comm": "worker", "pid": 100, "kernel": False, + "count": 2, "contexts": {"task": 2}, "new": False, + }]) + + seen = svc.describe(100, 100)["seen_waking"] + + assert [w["tid"] for w in seen["wakers"]] == [102] + assert seen["wakers"][0]["of"] == [101] + + +def test_an_interrupt_waking_the_thread_is_not_a_lock_holder(tmp_path, monkeypatch): + _thread(tmp_path, 100, 100) + _snapshots(tmp_path, monkeypatch, parked={ + "100": {"pid": 100, "name": "futex", "word": "0x7f10", "op": 128}, + }) + _wakeups(tmp_path, monkeypatch, [{ + "waker_tid": 0, "waker_comm": "", "waker_pid": None, "waker_kernel": None, + "tid": 100, "comm": "worker", "pid": 100, "kernel": False, + "count": 4, "contexts": {"hardirq": 4}, "new": False, + }]) + + waker = svc.describe(100, 100)["seen_waking"]["wakers"][0] + + assert waker["idle"] is True + assert waker["comm"] == "idle cpu" + assert waker["contexts"] == {"hardirq": 4} diff --git a/tests/test_wakeup_collector.py b/tests/test_wakeup_collector.py new file mode 100644 index 0000000..5de154f --- /dev/null +++ b/tests/test_wakeup_collector.py @@ -0,0 +1,97 @@ +"""Tests for the root wakeup collector. + +The fixture is shaped exactly like ``trace`` in a tracing instance: a header of +comment lines, the latency flags in their fixed columns, and one awkward line +whose waker name contains a space, which is what breaks naive parsing. +""" + +import importlib.util +import os + +_PATH = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "deploy", "ebpf", "wakeup_collector.py", +) +_spec = importlib.util.spec_from_file_location("wakeup_collector", _PATH) +collector = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(collector) + + +TRACE = """# tracer: nop +# +# _-----=> irqs-off/BH-disabled +# / _----=> need-resched +# | / _---=> hardirq/softirq + bash-171012 [000] dN... 2741591.914644: sched_wakeup: comm=kauditd pid=22 prio=120 target_cpu=000 + kauditd-22 [000] d.... 2741591.914654: sched_wakeup: comm=systemd-journal pid=128826 prio=119 target_cpu=000 + kauditd-22 [000] d.... 2741591.915189: sched_wakeup: comm=systemd-journal pid=128826 prio=119 target_cpu=000 + rs:main Q:Reg-40318 [000] dNs.. 2741591.915046: sched_wakeup: comm=rcu_sched pid=16 prio=120 target_cpu=000 + -0 [000] dNh.. 2741591.915446: sched_wakeup: comm=auditd pid=75798 prio=116 target_cpu=000 + bash-171012 [000] d.... 2741591.916000: sched_wakeup_new: comm=sleep pid=171015 prio=120 target_cpu=000 +""" + + +def test_an_edge_counts_every_time_it_happens(): + edges, contexts, events, lost = collector.parse(TRACE) + + assert events == 6 and lost == 0 + edge = edges[(22, 128826)] + assert edge["count"] == 2 + assert edge["waker_comm"] == "kauditd" + assert edge["comm"] == "systemd-journal" + + +def test_a_waker_whose_name_has_a_space_is_still_read(): + edges, _, _, _ = collector.parse(TRACE) + + assert edges[(40318, 16)]["waker_comm"] == "rs:main Q:Reg" + + +def test_the_context_of_each_wakeup_comes_from_the_latency_flags(): + edges, contexts, _, _ = collector.parse(TRACE) + + # 'h' in the third column is a hard interrupt, 's' a softirq, '.' task code. + assert edges[(0, 75798)]["contexts"] == {"hardirq": 1} + assert edges[(40318, 16)]["contexts"] == {"softirq": 1} + assert edges[(171012, 22)]["contexts"] == {"task": 1} + assert contexts == {"task": 4, "softirq": 1, "hardirq": 1} + + +def test_a_task_waking_for_the_first_time_is_marked(): + edges, _, _, _ = collector.parse(TRACE) + + assert edges[(171012, 171015)]["new"] is True + assert edges[(171012, 22)]["new"] is False + + +def test_events_the_kernel_dropped_are_counted_not_hidden(): + text = TRACE + "CPU:0 [LOST 4231 EVENTS]\n" + + _, _, events, lost = collector.parse(text) + + assert lost == 4231 + assert events == 6 + + +def test_the_header_and_anything_unparsable_are_skipped(): + _, _, events, _ = collector.parse("# tracer: nop\nnot a trace line at all\n") + + assert events == 0 + + +def test_the_busiest_ends_are_rolled_up_over_all_edges(): + edges, _, _, _ = collector.parse(TRACE) + wakers = {} + for edge in edges.values(): + row = wakers.setdefault(edge["waker_tid"], + {"tid": edge["waker_tid"], "comm": edge["waker_comm"], + "count": 0, "partners": 0}) + row["count"] += edge["count"] + row["partners"] += 1 + + top = {row["tid"]: row for row in collector._top(wakers)} + + # kauditd woke one task twice; bash woke two different tasks once each. + assert (top[22]["count"], top[22]["partners"]) == (2, 1) + assert (top[171012]["count"], top[171012]["partners"]) == (2, 2) + assert (top[0]["count"], top[0]["partners"]) == (1, 1) diff --git a/tests/test_wakeups_service.py b/tests/test_wakeups_service.py new file mode 100644 index 0000000..21e3375 --- /dev/null +++ b/tests/test_wakeups_service.py @@ -0,0 +1,144 @@ +"""Tests for ``kernel_ai.services.wakeups``.""" + +import json + +import pytest + +from kernel_ai.services import wakeups as svc + + +def _snapshot(tmp_path, monkeypatch, payload): + path = tmp_path / "wakeups.json" + path.write_text(json.dumps(payload)) + monkeypatch.setattr(svc, "SNAPSHOT", str(path)) + return path + + +@pytest.fixture(autouse=True) +def _no_units(monkeypatch): + monkeypatch.setattr(svc, "_unit", lambda pid: None) + + +PAYLOAD = { + "ts": 0, + "window_s": 0.25, + "events": 156, + "lost": 0, + "distinct_edges": 31, + "contexts": {"task": 122, "softirq": 30, "hardirq": 4}, + "observer_tid": 999, + "edges": [ + {"waker_tid": 22, "waker_comm": "kauditd", "waker_pid": 22, "waker_kernel": True, + "tid": 128826, "comm": "systemd-journal", "pid": 128826, "kernel": False, + "count": 24, "contexts": {"task": 24}, "new": False}, + {"waker_tid": 0, "waker_comm": "", "waker_pid": None, "waker_kernel": None, + "tid": 75798, "comm": "auditd", "pid": 75798, "kernel": False, + "count": 9, "contexts": {"hardirq": 9}, "new": False}, + {"waker_tid": 999, "waker_comm": "python3", "waker_pid": 999, "waker_kernel": False, + "tid": 400, "comm": "systemd-journal", "pid": 400, "kernel": False, + "count": 3, "contexts": {"task": 3}, "new": False}, + ], + "wakers": [{"tid": 22, "comm": "kauditd", "count": 24, "partners": 1}], + "wakees": [{"tid": 128826, "comm": "systemd-journal", "count": 24, "partners": 1}], +} + + +def test_the_window_is_reported_as_a_rate_and_a_length(tmp_path, monkeypatch): + _snapshot(tmp_path, monkeypatch, PAYLOAD) + + out = svc.describe() + + assert out["available"] is True + assert out["window_s"] == 0.25 + assert out["events"] == 156 + assert out["rate_per_s"] == 624 + assert out["lost"] == 0 + + +def test_each_context_carries_what_it_means(tmp_path, monkeypatch): + _snapshot(tmp_path, monkeypatch, PAYLOAD) + + contexts = svc.describe()["contexts"] + + assert contexts["hardirq"]["count"] == 4 + assert "hardware" in contexts["hardirq"]["means"] + assert contexts["task"]["count"] == 122 + + +def test_an_edge_names_both_ends(tmp_path, monkeypatch): + _snapshot(tmp_path, monkeypatch, PAYLOAD) + + edge = svc.describe()["edges"][0] + + assert edge["waker"]["comm"] == "kauditd" + assert edge["waker"]["kernel"] is True + assert edge["woken"]["comm"] == "systemd-journal" + assert edge["count"] == 24 + assert edge["why"] == "the waker itself, in ordinary code" + + +def test_a_wakeup_from_an_idle_cpu_is_told_as_an_interrupt(tmp_path, monkeypatch): + """Tid 0 is not a task: it is the cpu with nothing to run.""" + _snapshot(tmp_path, monkeypatch, PAYLOAD) + + edge = svc.describe()["edges"][1] + + assert edge["waker"]["idle"] is True + assert edge["waker"]["comm"] == "idle cpu" + assert edge["why"] == "an interrupt arriving while the cpu had nothing to run" + + +def test_the_collector_is_marked_where_it_appears(tmp_path, monkeypatch): + """Reading files wakes what serves them, and the sampler reads files.""" + _snapshot(tmp_path, monkeypatch, PAYLOAD) + + edges = svc.describe()["edges"] + + assert edges[2]["waker"]["observer"] is True + assert edges[0]["waker"]["observer"] is False + + +def test_a_thread_can_ask_who_was_seen_waking_it(tmp_path, monkeypatch): + _snapshot(tmp_path, monkeypatch, PAYLOAD) + + out = svc.for_thread(128826) + + assert out["available"] is True + assert [row["waker"]["comm"] for row in out["wakers"]] == ["kauditd"] + assert out["wakers"][0]["count"] == 24 + + +def test_a_thread_nobody_was_seen_waking_gets_an_empty_answer(tmp_path, monkeypatch): + _snapshot(tmp_path, monkeypatch, PAYLOAD) + + assert svc.for_thread(4242)["wakers"] == [] + + +def test_without_the_collector_nothing_is_claimed(tmp_path, monkeypatch): + monkeypatch.setattr(svc, "SNAPSHOT", str(tmp_path / "missing.json")) + + out = svc.describe() + + assert out["available"] is False + assert out["edges"] == [] + assert out["source"]["reason"] == "no-collector" + + +def test_a_stale_window_is_refused(tmp_path, monkeypatch): + import os + import time + + path = _snapshot(tmp_path, monkeypatch, PAYLOAD) + old = time.time() - 120 + os.utime(path, (old, old)) + + out = svc.describe() + + assert out["available"] is False + assert out["source"]["reason"] == "stale" + + +def test_dropped_events_reach_the_payload(tmp_path, monkeypatch): + _snapshot(tmp_path, monkeypatch, dict(PAYLOAD, lost=4231)) + + assert svc.describe()["lost"] == 4231