eSlider
ea062af86a
Explorer and exporter were scanning the full Parquet lake every 1–5s (~2000 files, 200MB+), driving ~2.2 CPU cores. Limit metrics to the last N flights, cache DuckDB views and the partition tree, slow live polls to 2s, and keep drones alive after seal to stop restart churn. Add node-exporter, scan-duration metrics, and a Swarm Platform Grafana dashboard for node CPU/memory and scan health.
41 lines
1.5 KiB
Python
41 lines
1.5 KiB
Python
"""Lake scan helpers — bounded to recent flight partitions."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from monitoring.lake import flight_window, iter_parquet_files, parquet_reader, recent_flights
|
|
|
|
|
|
def test_parquet_reader_limits_to_recent_flights(tmp_path: Path) -> None:
|
|
for i, name in enumerate(("flight=aaa", "flight=bbb", "flight=ccc")):
|
|
hour = tmp_path / f"dataset=telemetry/{name}/drone=dr-01/sensor=imu/year=2026/month=07/day=08/hour=10"
|
|
hour.mkdir(parents=True)
|
|
f = hour / "data.parquet"
|
|
f.write_bytes(b"x" * (i + 1))
|
|
# Make later names newer
|
|
import os
|
|
import time
|
|
os.utime(f, (time.time() + i, time.time() + i))
|
|
|
|
reader = parquet_reader(tmp_path, "telemetry", limit=1)
|
|
assert "flight=ccc" in reader
|
|
assert "flight=bbb" not in reader
|
|
|
|
|
|
def test_iter_parquet_files_skips_old_flights(tmp_path: Path) -> None:
|
|
old = tmp_path / "dataset=state/flight=old/drone=dr-01/year=2026/month=07/day=08/hour=09"
|
|
old.mkdir(parents=True)
|
|
(old / "data.parquet").write_bytes(b"old")
|
|
new = tmp_path / "dataset=state/flight=new/drone=dr-01/year=2026/month=07/day=08/hour=10"
|
|
new.mkdir(parents=True)
|
|
new_file = new / "data.parquet"
|
|
new_file.write_bytes(b"new")
|
|
import os
|
|
import time
|
|
os.utime(new_file, (time.time() + 10, time.time() + 10))
|
|
|
|
files = iter_parquet_files(tmp_path, flight_limit=1)
|
|
assert len(files) == 1
|
|
assert "flight=new" in str(files[0])
|