169 lines
6.6 KiB
Python
169 lines
6.6 KiB
Python
import importlib.util
|
|
import sqlite3
|
|
import time
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
from k1link.fleet.monitor import SCHEMA, MonitorReceiver, MonitorReplica
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def module(name):
|
|
spec = importlib.util.spec_from_file_location(
|
|
name, ROOT / "apps/node-agent/monitor" / (name + ".py")
|
|
)
|
|
result = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(result)
|
|
return result
|
|
|
|
|
|
def record(seq, at, boot, value=25):
|
|
return dict(seq=seq, at=at, boot_id=boot, uptime=seq, values={"cpu.usage": value}, events=[])
|
|
|
|
|
|
def envelope(source, rows, latest=None):
|
|
return dict(
|
|
schema=SCHEMA,
|
|
source_id=source,
|
|
storage="ready",
|
|
latest=latest or rows[-1],
|
|
definitions={
|
|
"cpu.usage": dict(label="CPU", group="CPU", resource="cpu", unit="%", reason=None)
|
|
},
|
|
batch=dict(schema=SCHEMA, source_id=source, samples=rows),
|
|
)
|
|
|
|
|
|
def test_replica_retries_restart_backfill_and_latest_are_separate(tmp_path):
|
|
source, boot = str(uuid4()), str(uuid4())
|
|
now = time.time()
|
|
rows = [record(i, now - 40 + i, boot) for i in range(1, 41)]
|
|
archive = MonitorReplica(tmp_path)
|
|
assert archive.ingest("board", envelope(source, rows[:32], rows[-1]))["after"] == 32
|
|
result = archive.query("board")
|
|
assert result["latest"]["seq"] == 40 and result["backlog"] == 8
|
|
assert archive.db.execute("SELECT count(*) FROM samples").fetchone()[0] == 32
|
|
archive.close()
|
|
archive = MonitorReplica(tmp_path)
|
|
# A lost reply cannot duplicate history; ACK comes from the durable commit.
|
|
assert archive.ingest("board", envelope(source, rows[:32], rows[-1]))["after"] == 32
|
|
assert archive.ingest("board", envelope(source, rows[32:]))["after"] == 40
|
|
assert archive.db.execute("SELECT count(*) FROM samples").fetchone()[0] == 40
|
|
assert archive.query("board")["backlog"] == 0
|
|
# Retention on the board can leave a real gap; no zero samples are invented.
|
|
assert archive.ingest("board", envelope(source, [record(51, now + 11, boot)]))["after"] == 51
|
|
archive.close()
|
|
|
|
|
|
def test_conflicting_or_nonfinite_batch_rolls_back_and_missing_values_are_null(tmp_path):
|
|
source, boot = str(uuid4()), str(uuid4())
|
|
now = time.time()
|
|
archive = MonitorReplica(tmp_path)
|
|
original = record(2, now - 20, boot)
|
|
archive.ingest("board", envelope(source, [original]))
|
|
bad = envelope(source, [record(1, now - 21, boot), {**original, "values": {"cpu.usage": 99}}])
|
|
with pytest.raises(ValueError):
|
|
archive.ingest("board", bad)
|
|
assert archive.db.execute("SELECT count(*) FROM samples").fetchone()[0] == 1
|
|
with pytest.raises(ValueError):
|
|
archive.ingest("board", envelope(source, [record(3, now, boot, float("nan"))]))
|
|
archive.ingest("board", envelope(source, [record(3, now, boot, None)]))
|
|
point = archive.query("board")["series"][-1]
|
|
assert point["max"] is None and point["count"] == 0
|
|
archive.close()
|
|
|
|
|
|
def test_receiver_failure_isolated_and_offline_history_reopens(tmp_path, monkeypatch):
|
|
source, boot = str(uuid4()), str(uuid4())
|
|
archive = MonitorReplica(tmp_path)
|
|
archive.ingest("board", envelope(source, [record(1, time.time(), boot)]))
|
|
archive.close()
|
|
receiver = MonitorReceiver(tmp_path)
|
|
try:
|
|
deadline = time.monotonic() + 2
|
|
while not receiver.query("board")["available"] and time.monotonic() < deadline:
|
|
time.sleep(0.01)
|
|
assert receiver.query("board")["available"]
|
|
monkeypatch.setattr(
|
|
receiver.archive,
|
|
"ingest",
|
|
lambda *_: (_ for _ in ()).throw(sqlite3.OperationalError("synthetic full disk")),
|
|
)
|
|
start = time.monotonic()
|
|
assert receiver.submit("board", envelope(source, [record(2, time.time(), boot)])) is None
|
|
assert time.monotonic() - start < 0.1
|
|
time.sleep(0.05)
|
|
assert receiver.acks.get("board") is None
|
|
finally:
|
|
receiver.close()
|
|
|
|
|
|
def test_linux_rates_distinguish_absent_initial_reset_and_gap(tmp_path):
|
|
now = [100]
|
|
|
|
def write(path, text):
|
|
target = tmp_path / path
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_text(text)
|
|
|
|
def counters(cpu, idle, rx):
|
|
write("proc/stat", f"cpu {cpu} 0 0 {idle} 0 0 0 0\ncpu0 {cpu} 0 0 {idle} 0 0 0 0\n")
|
|
write("sys/class/net/eth0/statistics/rx_bytes", str(rx))
|
|
|
|
write("sys/class/net/eth0/speed", "-1")
|
|
write("sys/bus/usb/devices/1-2/idVendor", "0000")
|
|
write("sys/bus/usb/devices/1-2/speed", "480")
|
|
write(
|
|
"sys/fs/cgroup/system.slice/mission-core-node.service/memory.events", "oom 3\noom_kill 2\n"
|
|
)
|
|
metrics = module("linux_metrics").LinuxMetrics(tmp_path, lambda: now[0])
|
|
counters(100, 100, 1000)
|
|
first = metrics.sample()
|
|
assert first["cpu.usage"] is None and first["net.eth0.rx_bytes"] is None
|
|
now[0] += 1
|
|
counters(150, 150, 1512)
|
|
second = metrics.sample()
|
|
assert second["cpu.usage"] == 50 and second["net.eth0.rx_bytes"] == 512
|
|
assert second["net.eth0.speed"] is None
|
|
assert second["usb.1-2.speed"] == 480 and second["usb.1-2.traffic"] is None
|
|
assert second["service.mission-core-node.service.oom"] == 2
|
|
now[0] += 1
|
|
counters(5, 5, 100)
|
|
assert metrics.sample()["net.eth0.rx_bytes"] is None
|
|
now[0] += 11
|
|
counters(25, 25, 900)
|
|
assert metrics.sample()["cpu.usage"] is None
|
|
|
|
|
|
def test_journal_retains_classification_without_freeform_data():
|
|
classify = module("journal_events").classify
|
|
value = classify(
|
|
{
|
|
"__REALTIME_TIMESTAMP": "123000000",
|
|
"_TRANSPORT": "kernel",
|
|
"MESSAGE": "Out of memory: Killed process 23 secret argument",
|
|
}
|
|
)
|
|
assert value == dict(code="system-oom", kind="Kernel", at=123, locations=[])
|
|
assert classify({"MESSAGE": "private Wi-Fi password"}) is None
|
|
|
|
|
|
def test_long_range_aggregates_keep_peaks_counts_and_exclude_incomplete_minute(tmp_path):
|
|
source, boot = str(uuid4()), str(uuid4())
|
|
now = int(time.time()//60)*60
|
|
archive = MonitorReplica(tmp_path)
|
|
rows = [record(1, now-120, boot, 10), record(2, now-110, boot, 90),
|
|
record(3, now-60, boot, 30), record(4, now+1, boot, 100)]
|
|
value = envelope(source, rows)
|
|
archive.ingest('board', value)
|
|
archive.ingest('board', value) # Duplicate delivery cannot inflate aggregates.
|
|
result = archive.query('board', window=21600, end=now+10)
|
|
assert result['end'] == now
|
|
assert sum(point['count'] for point in result['series']) == 3
|
|
assert max(point['max'] for point in result['series']) == 90
|
|
assert min(point['min'] for point in result['series']) == 10
|
|
archive.close()
|