perf: bound lake scans and add platform Grafana dashboard
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.
This commit is contained in:
@@ -53,6 +53,13 @@ graph LR
|
||||
| Which link degraded first, and was it distance or interference? | RSSI telemetry joined with pose distance |
|
||||
| Is the new writer version flushing slower on ARM64? | CI simulation runs emit the same metrics; diff across fleet releases |
|
||||
|
||||
A working miniature of this doctrine ships in [`simulator/`](../simulator/): the `monitoring` Compose profile runs a DuckDB-based exporter over the generated Parquet plus Prometheus and a provisioned Grafana dashboard — fleet statistics derived from the data platform itself, with no agent on the (virtual) drones.
|
||||
A working miniature of this doctrine ships in [`simulator/`](../simulator/): the `monitoring` Compose profile runs a DuckDB-based exporter over the generated Parquet plus Prometheus and two provisioned Grafana dashboards:
|
||||
|
||||
| Dashboard | Focus |
|
||||
| --- | --- |
|
||||
| **Swarm Fleet — generated data** | Row counts, detections, pose frames, Parquet bytes, battery, RSSI |
|
||||
| **Swarm Platform — CPU & scan health** | Node CPU/memory (node-exporter), exporter/explorer scan durations, flight-window gauge |
|
||||
|
||||
The exporter and explorer **scan only the most recent flight partitions** (`METRIC_FLIGHT_WINDOW`, default 5) and cache tree/views — otherwise CPU climbs as every pod restart appends a new `flight=` tree to the shared lake. Drones in k3d set `KEEP_ALIVE=1` after sealing so they idle instead of exiting and spawning another flight.
|
||||
|
||||
Alerting on the ground follows standard practice (Prometheus alert rules for infrastructure, CI gates for regression in simulated staleness/throughput budgets). In flight there is nobody to page — the platform's job is to degrade in the documented order and record everything for the post-mortem.
|
||||
|
||||
@@ -63,6 +63,14 @@ resource "kubernetes_stateful_set" "drone" {
|
||||
name = "DATA_DIR"
|
||||
value = "/data"
|
||||
}
|
||||
env {
|
||||
name = "KEEP_ALIVE"
|
||||
value = "1"
|
||||
}
|
||||
env {
|
||||
name = "IMU_HZ"
|
||||
value = "20"
|
||||
}
|
||||
|
||||
volume_mount {
|
||||
name = "lake"
|
||||
@@ -118,9 +126,21 @@ resource "kubernetes_deployment" "exporter" {
|
||||
name = "EXPORTER_PORT"
|
||||
value = "9105"
|
||||
}
|
||||
env {
|
||||
name = "SCAN_INTERVAL_S"
|
||||
value = "30"
|
||||
}
|
||||
env {
|
||||
name = "METRIC_FLIGHT_WINDOW"
|
||||
value = "5"
|
||||
}
|
||||
port {
|
||||
container_port = 9105
|
||||
}
|
||||
resources {
|
||||
requests = { cpu = "50m", memory = "64Mi" }
|
||||
limits = { cpu = "500m", memory = "256Mi" }
|
||||
}
|
||||
volume_mount {
|
||||
name = "lake"
|
||||
mount_path = "/data"
|
||||
@@ -182,9 +202,25 @@ resource "kubernetes_deployment" "explorer" {
|
||||
name = "EXPLORER_PORT"
|
||||
value = "8088"
|
||||
}
|
||||
env {
|
||||
name = "VIEW_REFRESH_S"
|
||||
value = "30"
|
||||
}
|
||||
env {
|
||||
name = "TREE_CACHE_S"
|
||||
value = "15"
|
||||
}
|
||||
env {
|
||||
name = "METRIC_FLIGHT_WINDOW"
|
||||
value = "5"
|
||||
}
|
||||
port {
|
||||
container_port = 8088
|
||||
}
|
||||
resources {
|
||||
requests = { cpu = "50m", memory = "64Mi" }
|
||||
limits = { cpu = "750m", memory = "256Mi" }
|
||||
}
|
||||
volume_mount {
|
||||
name = "lake"
|
||||
mount_path = "/data"
|
||||
@@ -226,11 +262,17 @@ resource "kubernetes_config_map" "prometheus" {
|
||||
data = {
|
||||
"prometheus.yml" = <<-EOT
|
||||
global:
|
||||
scrape_interval: 5s
|
||||
scrape_interval: 15s
|
||||
scrape_configs:
|
||||
- job_name: swarm
|
||||
static_configs:
|
||||
- targets: ["exporter:9105"]
|
||||
- job_name: explorer
|
||||
static_configs:
|
||||
- targets: ["explorer:8088"]
|
||||
- job_name: node
|
||||
static_configs:
|
||||
- targets: ["node-exporter:9100"]
|
||||
EOT
|
||||
}
|
||||
}
|
||||
@@ -327,7 +369,91 @@ resource "kubernetes_config_map" "grafana_dashboard" {
|
||||
namespace = kubernetes_namespace.swarm.metadata[0].name
|
||||
}
|
||||
data = {
|
||||
"swarm.json" = file("${path.module}/../../../simulator/monitoring/grafana/dashboards/swarm.json")
|
||||
"swarm.json" = file("${path.module}/../../../simulator/monitoring/grafana/dashboards/swarm.json")
|
||||
"swarm-platform.json" = file("${path.module}/../../../simulator/monitoring/grafana/dashboards/swarm-platform.json")
|
||||
}
|
||||
}
|
||||
|
||||
# Host metrics for the k3d node (CPU / memory on the platform dashboard)
|
||||
resource "kubernetes_deployment" "node_exporter" {
|
||||
metadata {
|
||||
name = "node-exporter"
|
||||
namespace = kubernetes_namespace.swarm.metadata[0].name
|
||||
labels = local.labels
|
||||
}
|
||||
|
||||
spec {
|
||||
replicas = 1
|
||||
selector {
|
||||
match_labels = { app = "node-exporter" }
|
||||
}
|
||||
template {
|
||||
metadata {
|
||||
labels = merge(local.labels, { app = "node-exporter" })
|
||||
}
|
||||
spec {
|
||||
host_network = true
|
||||
host_pid = true
|
||||
container {
|
||||
name = "node-exporter"
|
||||
image = "prom/node-exporter:v1.8.2"
|
||||
args = [
|
||||
"--path.procfs=/host/proc",
|
||||
"--path.sysfs=/host/sys",
|
||||
"--path.rootfs=/host/root",
|
||||
"--collector.filesystem.mount-points-exclude=^/(sys|proc|dev|host|etc)($$|/)",
|
||||
]
|
||||
port {
|
||||
container_port = 9100
|
||||
}
|
||||
volume_mount {
|
||||
name = "proc"
|
||||
mount_path = "/host/proc"
|
||||
read_only = true
|
||||
}
|
||||
volume_mount {
|
||||
name = "sys"
|
||||
mount_path = "/host/sys"
|
||||
read_only = true
|
||||
}
|
||||
volume_mount {
|
||||
name = "root"
|
||||
mount_path = "/host/root"
|
||||
read_only = true
|
||||
}
|
||||
resources {
|
||||
requests = { cpu = "20m", memory = "32Mi" }
|
||||
limits = { cpu = "200m", memory = "64Mi" }
|
||||
}
|
||||
}
|
||||
volume {
|
||||
name = "proc"
|
||||
host_path { path = "/proc" }
|
||||
}
|
||||
volume {
|
||||
name = "sys"
|
||||
host_path { path = "/sys" }
|
||||
}
|
||||
volume {
|
||||
name = "root"
|
||||
host_path { path = "/" }
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
resource "kubernetes_service" "node_exporter" {
|
||||
metadata {
|
||||
name = "node-exporter"
|
||||
namespace = kubernetes_namespace.swarm.metadata[0].name
|
||||
}
|
||||
spec {
|
||||
selector = { app = "node-exporter" }
|
||||
port {
|
||||
port = 9100
|
||||
target_port = 9100
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -25,7 +25,7 @@ function displayAspect(): number {
|
||||
return Math.max(1, Math.min(MAX_ASPECT, q));
|
||||
}
|
||||
|
||||
const LIVE_POLL_MS = 1000;
|
||||
const LIVE_POLL_MS = 2000;
|
||||
|
||||
export default function App(): JSX.Element {
|
||||
const [droneCount, setDroneCount] = useState(8);
|
||||
|
||||
@@ -24,7 +24,8 @@ services:
|
||||
command: ["python", "monitoring/exporter.py"]
|
||||
environment:
|
||||
DATA_DIR: /data
|
||||
SCAN_INTERVAL_S: "5"
|
||||
SCAN_INTERVAL_S: "30"
|
||||
METRIC_FLIGHT_WINDOW: "5"
|
||||
volumes:
|
||||
- ${SWARM_T1_DIR:-../var/t1}:/data:ro
|
||||
- ./monitoring:/app/monitoring:ro
|
||||
@@ -37,6 +38,9 @@ services:
|
||||
command: ["python", "explorer/server.py"]
|
||||
environment:
|
||||
DATA_DIR: /data
|
||||
VIEW_REFRESH_S: "30"
|
||||
TREE_CACHE_S: "15"
|
||||
METRIC_FLIGHT_WINDOW: "5"
|
||||
volumes:
|
||||
- ${SWARM_T1_DIR:-../var/t1}:/data:ro
|
||||
- ./explorer:/app/explorer:ro
|
||||
|
||||
@@ -1,33 +1,28 @@
|
||||
"""Data-plane explorer: a web view over the Hive-partitioned Parquet lake.
|
||||
|
||||
Serves three things:
|
||||
/ single-page UI (partition tree + read-only SQL console)
|
||||
/api/tree partition hierarchy with file counts and bytes, live
|
||||
/api/query gated read-only DuckDB SQL, same statement rules as the
|
||||
peer query channel (SELECT/WITH only, single statement)
|
||||
|
||||
The point is doctrinal, not just convenient: the explorer reuses the exact
|
||||
read-only SQL contract that drones expose to each other, so "looking at the
|
||||
data plane" on the bench exercises the same path a peer would use in flight.
|
||||
"""
|
||||
"""Data-plane explorer: a web view over the Hive-partitioned Parquet lake."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
from pathlib import Path
|
||||
|
||||
import duckdb
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "monitoring"))
|
||||
from lake import flight_window, iter_parquet_files, parquet_reader # noqa: E402
|
||||
|
||||
DATA_DIR = Path(os.environ.get("DATA_DIR", "./data"))
|
||||
PORT = int(os.environ.get("EXPLORER_PORT", "8088"))
|
||||
ROW_LIMIT = int(os.environ.get("ROW_LIMIT", "500"))
|
||||
VIEW_REFRESH_S = float(os.environ.get("VIEW_REFRESH_S", "30"))
|
||||
TREE_CACHE_S = float(os.environ.get("TREE_CACHE_S", "15"))
|
||||
STATIC_DIR = Path(__file__).parent
|
||||
|
||||
# Same spirit as the forced-command gate on a real drone: one statement,
|
||||
# must be a read, no statement that could write, configure, or reach out.
|
||||
_ALLOWED_START = re.compile(r"^\s*(SELECT|WITH|DESCRIBE|SUMMARIZE|SHOW)\b", re.IGNORECASE)
|
||||
_FORBIDDEN = re.compile(
|
||||
r"\b(INSERT|UPDATE|DELETE|CREATE|DROP|ALTER|ATTACH|DETACH|COPY|EXPORT"
|
||||
@@ -35,9 +30,17 @@ _FORBIDDEN = re.compile(
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
_lock = threading.Lock()
|
||||
_con: duckdb.DuckDBPyConnection | None = None
|
||||
_views_at = 0.0
|
||||
_tree_cache: dict | None = None
|
||||
_tree_at = 0.0
|
||||
_last_tree_s = 0.0
|
||||
_last_query_s = 0.0
|
||||
_query_count = 0
|
||||
|
||||
|
||||
def gate(sql: str) -> str | None:
|
||||
"""Return a rejection reason, or None if the statement passes."""
|
||||
stripped = re.sub(r"--[^\n]*|/\*.*?\*/", " ", sql, flags=re.DOTALL).strip().rstrip(";")
|
||||
if not stripped:
|
||||
return "empty statement"
|
||||
@@ -50,27 +53,39 @@ def gate(sql: str) -> str | None:
|
||||
return None
|
||||
|
||||
|
||||
def connect() -> duckdb.DuckDBPyConnection:
|
||||
"""Fresh connection with the three datasets pre-registered as views."""
|
||||
def _refresh_views() -> None:
|
||||
global _con, _views_at
|
||||
con = duckdb.connect()
|
||||
for ds in ("telemetry", "detections", "state"):
|
||||
pattern = f"{DATA_DIR}/dataset={ds}/**/*.parquet"
|
||||
reader = parquet_reader(DATA_DIR, ds)
|
||||
try:
|
||||
con.execute(
|
||||
f"CREATE VIEW {ds} AS SELECT * FROM read_parquet("
|
||||
f"'{pattern}', hive_partitioning=true, union_by_name=true)"
|
||||
)
|
||||
con.execute(f"CREATE OR REPLACE VIEW {ds} AS SELECT * FROM read_parquet({reader})")
|
||||
except duckdb.Error:
|
||||
pass # dataset not written yet; view simply won't exist
|
||||
return con
|
||||
pass
|
||||
with _lock:
|
||||
if _con is not None:
|
||||
_con.close()
|
||||
_con = con
|
||||
_views_at = time.monotonic()
|
||||
|
||||
|
||||
def connect() -> duckdb.DuckDBPyConnection:
|
||||
if _con is None or time.monotonic() - _views_at > VIEW_REFRESH_S:
|
||||
_refresh_views()
|
||||
assert _con is not None
|
||||
return _con
|
||||
|
||||
|
||||
def tree() -> dict:
|
||||
"""Partition hierarchy: dataset -> flight -> drone -> leafs, with sizes."""
|
||||
global _tree_cache, _tree_at, _last_tree_s
|
||||
now = time.monotonic()
|
||||
if _tree_cache is not None and now - _tree_at < TREE_CACHE_S:
|
||||
return _tree_cache
|
||||
started = now
|
||||
root: dict = {}
|
||||
total_bytes = 0
|
||||
total_files = 0
|
||||
for f in sorted(DATA_DIR.rglob("*.parquet")):
|
||||
for f in sorted(iter_parquet_files(DATA_DIR)):
|
||||
rel = f.relative_to(DATA_DIR)
|
||||
size = f.stat().st_size
|
||||
total_bytes += size
|
||||
@@ -80,7 +95,6 @@ def tree() -> dict:
|
||||
node = node.setdefault("children", {}).setdefault(part, {})
|
||||
leaf = node.setdefault("children", {}).setdefault(rel.parts[-1], {})
|
||||
leaf["bytes"] = size
|
||||
# roll sizes up the tree
|
||||
node = root
|
||||
node["bytes"] = node.get("bytes", 0) + size
|
||||
node["files"] = node.get("files", 0) + 1
|
||||
@@ -88,27 +102,50 @@ def tree() -> dict:
|
||||
node = node["children"][part]
|
||||
node["bytes"] = node.get("bytes", 0) + size
|
||||
node["files"] = node.get("files", 0) + 1
|
||||
return {"tree": root, "total_bytes": total_bytes, "total_files": total_files}
|
||||
_last_tree_s = time.monotonic() - started
|
||||
_tree_cache = {"tree": root, "total_bytes": total_bytes, "total_files": total_files}
|
||||
_tree_at = now
|
||||
return _tree_cache
|
||||
|
||||
|
||||
def run_query(sql: str) -> dict:
|
||||
global _last_query_s, _query_count
|
||||
reason = gate(sql)
|
||||
if reason:
|
||||
return {"error": f"rejected by read-only gate: {reason}"}
|
||||
con = connect()
|
||||
try:
|
||||
cur = con.sql(sql)
|
||||
columns = cur.columns
|
||||
rows = cur.fetchmany(ROW_LIMIT)
|
||||
return {
|
||||
"columns": columns,
|
||||
"rows": [[repr(v) if isinstance(v, bytes) else v for v in row] for row in rows],
|
||||
"truncated": len(rows) == ROW_LIMIT,
|
||||
}
|
||||
except duckdb.Error as exc:
|
||||
return {"error": str(exc)}
|
||||
finally:
|
||||
con.close()
|
||||
started = time.monotonic()
|
||||
with _lock:
|
||||
con = connect()
|
||||
try:
|
||||
cur = con.sql(sql)
|
||||
columns = cur.columns
|
||||
rows = cur.fetchmany(ROW_LIMIT)
|
||||
_query_count += 1
|
||||
_last_query_s = time.monotonic() - started
|
||||
return {
|
||||
"columns": columns,
|
||||
"rows": [[repr(v) if isinstance(v, bytes) else v for v in row] for row in rows],
|
||||
"truncated": len(rows) == ROW_LIMIT,
|
||||
}
|
||||
except duckdb.Error as exc:
|
||||
return {"error": str(exc)}
|
||||
|
||||
|
||||
def metrics_text() -> str:
|
||||
return "\n".join([
|
||||
"# HELP swarm_explorer_query_duration_seconds Wall time of the last SQL query",
|
||||
"# TYPE swarm_explorer_query_duration_seconds gauge",
|
||||
f"swarm_explorer_query_duration_seconds {_last_query_s:.4f}",
|
||||
"# HELP swarm_explorer_tree_duration_seconds Wall time of the last partition tree build",
|
||||
"# TYPE swarm_explorer_tree_duration_seconds gauge",
|
||||
f"swarm_explorer_tree_duration_seconds {_last_tree_s:.4f}",
|
||||
"# HELP swarm_explorer_queries_total Read-only queries served",
|
||||
"# TYPE swarm_explorer_queries_total counter",
|
||||
f"swarm_explorer_queries_total {_query_count}",
|
||||
"# HELP swarm_metric_flight_window Flight partitions in DuckDB views",
|
||||
"# TYPE swarm_metric_flight_window gauge",
|
||||
f"swarm_metric_flight_window {flight_window()}",
|
||||
]) + "\n"
|
||||
|
||||
|
||||
class Handler(BaseHTTPRequestHandler):
|
||||
@@ -116,13 +153,11 @@ class Handler(BaseHTTPRequestHandler):
|
||||
self.send_response(code)
|
||||
self.send_header("Content-Type", ctype)
|
||||
self.send_header("Content-Length", str(len(body)))
|
||||
# Dev CORS: lets the prototype's live mode poll the API from another
|
||||
# origin. Everything behind this is read-only by construction.
|
||||
self.send_header("Access-Control-Allow-Origin", "*")
|
||||
self.end_headers()
|
||||
self.wfile.write(body)
|
||||
|
||||
def do_OPTIONS(self) -> None: # noqa: N802 — http.server API
|
||||
def do_OPTIONS(self) -> None: # noqa: N802
|
||||
self.send_response(204)
|
||||
self.send_header("Access-Control-Allow-Origin", "*")
|
||||
self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
|
||||
@@ -132,15 +167,17 @@ class Handler(BaseHTTPRequestHandler):
|
||||
def _json(self, payload: dict, code: int = 200) -> None:
|
||||
self._send(code, json.dumps(payload, default=str).encode(), "application/json")
|
||||
|
||||
def do_GET(self) -> None: # noqa: N802 — http.server API
|
||||
def do_GET(self) -> None: # noqa: N802
|
||||
if self.path in ("/", "/index.html"):
|
||||
self._send(200, (STATIC_DIR / "index.html").read_bytes(), "text/html; charset=utf-8")
|
||||
elif self.path == "/api/tree":
|
||||
self._json(tree())
|
||||
elif self.path == "/metrics":
|
||||
self._send(200, metrics_text().encode(), "text/plain; version=0.0.4")
|
||||
else:
|
||||
self._send(404, b"not found", "text/plain")
|
||||
|
||||
def do_POST(self) -> None: # noqa: N802 — http.server API
|
||||
def do_POST(self) -> None: # noqa: N802
|
||||
if self.path != "/api/query":
|
||||
self._send(404, b"not found", "text/plain")
|
||||
return
|
||||
@@ -156,6 +193,20 @@ class Handler(BaseHTTPRequestHandler):
|
||||
pass
|
||||
|
||||
|
||||
def _view_loop() -> None:
|
||||
while True:
|
||||
try:
|
||||
_refresh_views()
|
||||
except Exception:
|
||||
pass
|
||||
time.sleep(VIEW_REFRESH_S)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
print(f"data-plane explorer on :{PORT}, reading {DATA_DIR}")
|
||||
_refresh_views()
|
||||
threading.Thread(target=_view_loop, daemon=True).start()
|
||||
print(
|
||||
f"data-plane explorer on :{PORT}, views refresh every {VIEW_REFRESH_S}s, "
|
||||
f"last {flight_window()} flights, reading {DATA_DIR}"
|
||||
)
|
||||
ThreadingHTTPServer(("", PORT), Handler).serve_forever()
|
||||
|
||||
@@ -1,13 +1,14 @@
|
||||
"""Prometheus exporter over the simulator's Parquet output.
|
||||
|
||||
Periodically scans DATA_DIR with DuckDB and exposes fleet statistics as
|
||||
/metrics. Zero dependencies beyond duckdb: the exposition format is plain
|
||||
text, served with the standard library HTTP server.
|
||||
/metrics. Scans only the most recent flight partitions by default so CPU
|
||||
stays bounded as the lake grows across pod restarts.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
@@ -15,27 +16,29 @@ from pathlib import Path
|
||||
|
||||
import duckdb
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||||
from lake import flight_window, iter_parquet_files, parquet_reader # noqa: E402
|
||||
|
||||
DATA_DIR = Path(os.environ.get("DATA_DIR", "./data"))
|
||||
PORT = int(os.environ.get("EXPORTER_PORT", "9105"))
|
||||
SCAN_INTERVAL_S = float(os.environ.get("SCAN_INTERVAL_S", "5"))
|
||||
SCAN_INTERVAL_S = float(os.environ.get("SCAN_INTERVAL_S", "15"))
|
||||
|
||||
_lock = threading.Lock()
|
||||
_payload = "# swarm exporter starting\n"
|
||||
_last_scan_s = 0.0
|
||||
|
||||
|
||||
def _q(con: duckdb.DuckDBPyConnection, sql: str) -> list[tuple]:
|
||||
try:
|
||||
return con.sql(sql).fetchall()
|
||||
except duckdb.Error:
|
||||
return [] # partitions may not exist yet while the swarm warms up
|
||||
return []
|
||||
|
||||
|
||||
def collect() -> str:
|
||||
global _last_scan_s
|
||||
started = time.monotonic()
|
||||
con = duckdb.connect()
|
||||
# union_by_name: sensors have different schemas under one dataset glob
|
||||
glob = lambda ds: ( # noqa: E731
|
||||
f"'{DATA_DIR}/dataset={ds}/**/*.parquet', hive_partitioning=true, union_by_name=true"
|
||||
)
|
||||
lines: list[str] = []
|
||||
|
||||
def metric(name: str, help_text: str, mtype: str, rows: list[str]) -> None:
|
||||
@@ -43,27 +46,38 @@ def collect() -> str:
|
||||
lines.append(f"# TYPE {name} {mtype}")
|
||||
lines.extend(rows)
|
||||
|
||||
metric(
|
||||
"swarm_rows_total", "Telemetry rows written per drone and sensor", "gauge",
|
||||
[f'swarm_rows_total{{drone="{d}",sensor="{s}"}} {n}'
|
||||
for d, s, n in _q(con, f"SELECT drone, sensor, count(*) FROM read_parquet({glob('telemetry')}) GROUP BY 1,2")],
|
||||
)
|
||||
metric(
|
||||
"swarm_detections_total", "Detection events per drone and class", "gauge",
|
||||
[f'swarm_detections_total{{drone="{d}",cls="{c}"}} {n}'
|
||||
for d, c, n in _q(con, f"SELECT drone, cls, count(*) FROM read_parquet({glob('detections')}) GROUP BY 1,2")],
|
||||
)
|
||||
metric(
|
||||
"swarm_state_frames_total", "Pose broadcast frames per drone and direction", "gauge",
|
||||
[f'swarm_state_frames_total{{drone="{d}",direction="{dr}"}} {n}'
|
||||
for d, dr, n in _q(con, f"SELECT drone, direction, count(*) FROM read_parquet({glob('state')}) GROUP BY 1,2")],
|
||||
)
|
||||
for ds, name in (
|
||||
("telemetry", "swarm_rows_total"),
|
||||
("detections", "swarm_detections_total"),
|
||||
("state", "swarm_state_frames_total"),
|
||||
):
|
||||
reader = parquet_reader(DATA_DIR, ds)
|
||||
if ds == "telemetry":
|
||||
metric(
|
||||
name, f"Rows in dataset={ds} (recent {flight_window()} flights)", "gauge",
|
||||
[f'{name}{{drone="{d}",sensor="{s}"}} {n}'
|
||||
for d, s, n in _q(con, f"SELECT drone, sensor, count(*) FROM read_parquet({reader}) GROUP BY 1,2")],
|
||||
)
|
||||
elif ds == "detections":
|
||||
metric(
|
||||
name, "Detection events per drone and class", "gauge",
|
||||
[f'{name}{{drone="{d}",cls="{c}"}} {n}'
|
||||
for d, c, n in _q(con, f"SELECT drone, cls, count(*) FROM read_parquet({reader}) GROUP BY 1,2")],
|
||||
)
|
||||
else:
|
||||
metric(
|
||||
name, "Pose broadcast frames per drone and direction", "gauge",
|
||||
[f'{name}{{drone="{d}",direction="{dr}"}} {n}'
|
||||
for d, dr, n in _q(con, f"SELECT drone, direction, count(*) FROM read_parquet({reader}) GROUP BY 1,2")],
|
||||
)
|
||||
|
||||
telem = parquet_reader(DATA_DIR, "telemetry")
|
||||
metric(
|
||||
"swarm_battery_pct", "Latest battery level per drone", "gauge",
|
||||
[f'swarm_battery_pct{{drone="{d}"}} {v}'
|
||||
for d, v in _q(con, f"""
|
||||
SELECT drone, arg_max(level_pct, ts_ns)
|
||||
FROM read_parquet({glob('telemetry')})
|
||||
FROM read_parquet({telem})
|
||||
WHERE sensor='battery' GROUP BY drone""")],
|
||||
)
|
||||
metric(
|
||||
@@ -71,26 +85,24 @@ def collect() -> str:
|
||||
[f'swarm_rssi_dbm{{drone="{d}",peer="{p}"}} {v}'
|
||||
for d, p, v in _q(con, f"""
|
||||
SELECT drone, peer_id, arg_max(rssi_dbm, ts_ns)
|
||||
FROM read_parquet({glob('telemetry')})
|
||||
WHERE sensor='rssi' GROUP BY drone, peer_id""")],
|
||||
)
|
||||
metric(
|
||||
"swarm_peer_distance_m", "Latest inter-drone distance estimate", "gauge",
|
||||
[f'swarm_peer_distance_m{{drone="{d}",peer="{p}"}} {v}'
|
||||
for d, p, v in _q(con, f"""
|
||||
SELECT drone, peer_id, arg_max(distance_m, ts_ns)
|
||||
FROM read_parquet({glob('telemetry')})
|
||||
FROM read_parquet({telem})
|
||||
WHERE sensor='rssi' GROUP BY drone, peer_id""")],
|
||||
)
|
||||
|
||||
files = list(DATA_DIR.rglob("*.parquet"))
|
||||
files = iter_parquet_files(DATA_DIR)
|
||||
metric(
|
||||
"swarm_parquet_bytes", "Bytes on disk per dataset", "gauge",
|
||||
"swarm_parquet_bytes", "Bytes on disk per dataset (recent flights)", "gauge",
|
||||
[f'swarm_parquet_bytes{{dataset="{ds}"}} {sum(f.stat().st_size for f in files if f"dataset={ds}" in str(f))}'
|
||||
for ds in ("telemetry", "detections", "state")],
|
||||
)
|
||||
metric("swarm_parquet_files", "Parquet files on disk", "gauge",
|
||||
metric("swarm_parquet_files", "Parquet files scanned (recent flights)", "gauge",
|
||||
[f"swarm_parquet_files {len(files)}"])
|
||||
metric("swarm_metric_flight_window", "Flight partitions included per dataset", "gauge",
|
||||
[f"swarm_metric_flight_window {flight_window()}"])
|
||||
|
||||
_last_scan_s = time.monotonic() - started
|
||||
metric("swarm_exporter_scan_duration_seconds", "Wall time of the last metrics scan", "gauge",
|
||||
[f"swarm_exporter_scan_duration_seconds {_last_scan_s:.4f}"])
|
||||
con.close()
|
||||
return "\n".join(lines) + "\n"
|
||||
|
||||
@@ -101,11 +113,11 @@ def scanner() -> None:
|
||||
started = time.monotonic()
|
||||
try:
|
||||
payload = collect()
|
||||
except Exception as exc: # keep serving stale metrics over dying
|
||||
except Exception as exc:
|
||||
payload = f"# collect error: {exc}\n"
|
||||
with _lock:
|
||||
_payload = payload
|
||||
time.sleep(max(0.5, SCAN_INTERVAL_S - (time.monotonic() - started)))
|
||||
time.sleep(max(1.0, SCAN_INTERVAL_S - (time.monotonic() - started)))
|
||||
|
||||
|
||||
class Handler(BaseHTTPRequestHandler):
|
||||
@@ -123,10 +135,13 @@ class Handler(BaseHTTPRequestHandler):
|
||||
self.wfile.write(body)
|
||||
|
||||
def log_message(self, *_args: object) -> None:
|
||||
pass # scrapes every few seconds; keep the log quiet
|
||||
pass
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
threading.Thread(target=scanner, daemon=True).start()
|
||||
print(f"swarm exporter on :{PORT}/metrics, scanning {DATA_DIR} every {SCAN_INTERVAL_S}s")
|
||||
print(
|
||||
f"swarm exporter on :{PORT}/metrics, scanning last {flight_window()} flights "
|
||||
f"every {SCAN_INTERVAL_S}s under {DATA_DIR}"
|
||||
)
|
||||
ThreadingHTTPServer(("", PORT), Handler).serve_forever()
|
||||
|
||||
@@ -0,0 +1,90 @@
|
||||
{
|
||||
"uid": "swarm-platform",
|
||||
"title": "Swarm Platform — CPU & scan health",
|
||||
"tags": ["swarm", "platform"],
|
||||
"timezone": "utc",
|
||||
"schemaVersion": 39,
|
||||
"version": 1,
|
||||
"refresh": "10s",
|
||||
"time": { "from": "now-30m", "to": "now" },
|
||||
"panels": [
|
||||
{
|
||||
"id": 1, "type": "timeseries", "title": "Node CPU %",
|
||||
"gridPos": { "h": 8, "w": 12, "x": 0, "y": 0 },
|
||||
"datasource": { "type": "prometheus", "uid": "swarm-prom" },
|
||||
"targets": [{
|
||||
"expr": "100 - (avg by (instance) (rate(node_cpu_seconds_total{mode=\"idle\"}[1m])) * 100)",
|
||||
"legendFormat": "{{instance}}", "refId": "A"
|
||||
}],
|
||||
"fieldConfig": { "defaults": { "unit": "percent", "min": 0, "max": 100 }, "overrides": [] }
|
||||
},
|
||||
{
|
||||
"id": 2, "type": "timeseries", "title": "Node memory used %",
|
||||
"gridPos": { "h": 8, "w": 12, "x": 12, "y": 0 },
|
||||
"datasource": { "type": "prometheus", "uid": "swarm-prom" },
|
||||
"targets": [{
|
||||
"expr": "(1 - node_memory_MemAvailable_bytes / node_memory_MemTotal_bytes) * 100",
|
||||
"legendFormat": "used", "refId": "A"
|
||||
}],
|
||||
"fieldConfig": { "defaults": { "unit": "percent", "min": 0, "max": 100 }, "overrides": [] }
|
||||
},
|
||||
{
|
||||
"id": 3, "type": "timeseries", "title": "Exporter scan duration",
|
||||
"gridPos": { "h": 7, "w": 8, "x": 0, "y": 8 },
|
||||
"datasource": { "type": "prometheus", "uid": "swarm-prom" },
|
||||
"targets": [{
|
||||
"expr": "swarm_exporter_scan_duration_seconds",
|
||||
"legendFormat": "scan seconds", "refId": "A"
|
||||
}],
|
||||
"fieldConfig": { "defaults": { "unit": "s" }, "overrides": [] }
|
||||
},
|
||||
{
|
||||
"id": 4, "type": "timeseries", "title": "Explorer query duration",
|
||||
"gridPos": { "h": 7, "w": 8, "x": 8, "y": 8 },
|
||||
"datasource": { "type": "prometheus", "uid": "swarm-prom" },
|
||||
"targets": [{
|
||||
"expr": "swarm_explorer_query_duration_seconds",
|
||||
"legendFormat": "last query", "refId": "A"
|
||||
}],
|
||||
"fieldConfig": { "defaults": { "unit": "s" }, "overrides": [] }
|
||||
},
|
||||
{
|
||||
"id": 5, "type": "timeseries", "title": "Explorer tree build duration",
|
||||
"gridPos": { "h": 7, "w": 8, "x": 16, "y": 8 },
|
||||
"datasource": { "type": "prometheus", "uid": "swarm-prom" },
|
||||
"targets": [{
|
||||
"expr": "swarm_explorer_tree_duration_seconds",
|
||||
"legendFormat": "tree seconds", "refId": "A"
|
||||
}],
|
||||
"fieldConfig": { "defaults": { "unit": "s" }, "overrides": [] }
|
||||
},
|
||||
{
|
||||
"id": 6, "type": "stat", "title": "Parquet files scanned",
|
||||
"gridPos": { "h": 5, "w": 6, "x": 0, "y": 15 },
|
||||
"datasource": { "type": "prometheus", "uid": "swarm-prom" },
|
||||
"targets": [{ "expr": "swarm_parquet_files", "instant": true, "refId": "A" }],
|
||||
"options": { "reduceOptions": { "calcs": ["lastNotNull"] } }
|
||||
},
|
||||
{
|
||||
"id": 7, "type": "stat", "title": "Flight window",
|
||||
"gridPos": { "h": 5, "w": 6, "x": 6, "y": 15 },
|
||||
"datasource": { "type": "prometheus", "uid": "swarm-prom" },
|
||||
"targets": [{ "expr": "swarm_metric_flight_window", "instant": true, "refId": "A" }],
|
||||
"options": { "reduceOptions": { "calcs": ["lastNotNull"] } }
|
||||
},
|
||||
{
|
||||
"id": 8, "type": "stat", "title": "Telemetry rows (window)",
|
||||
"gridPos": { "h": 5, "w": 6, "x": 12, "y": 15 },
|
||||
"datasource": { "type": "prometheus", "uid": "swarm-prom" },
|
||||
"targets": [{ "expr": "sum(swarm_rows_total)", "instant": true, "refId": "A" }],
|
||||
"options": { "reduceOptions": { "calcs": ["lastNotNull"] } }
|
||||
},
|
||||
{
|
||||
"id": 9, "type": "stat", "title": "Min fleet battery %",
|
||||
"gridPos": { "h": 5, "w": 6, "x": 18, "y": 15 },
|
||||
"datasource": { "type": "prometheus", "uid": "swarm-prom" },
|
||||
"targets": [{ "expr": "min(swarm_battery_pct)", "instant": true, "refId": "A" }],
|
||||
"fieldConfig": { "defaults": { "unit": "percent", "min": 0, "max": 100 }, "overrides": [] }
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -0,0 +1,43 @@
|
||||
"""Scan the Hive-partitioned lake without re-reading every historical flight."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def flight_window() -> int:
|
||||
return max(1, int(os.environ.get("METRIC_FLIGHT_WINDOW", "5")))
|
||||
|
||||
|
||||
def recent_flights(data_dir: Path, dataset: str, limit: int | None = None) -> list[Path]:
|
||||
"""Most recently modified flight= partitions for a dataset."""
|
||||
limit = limit or flight_window()
|
||||
base = data_dir / f"dataset={dataset}"
|
||||
if not base.is_dir():
|
||||
return []
|
||||
flights = [p for p in base.iterdir() if p.is_dir() and p.name.startswith("flight=")]
|
||||
flights.sort(key=lambda p: p.stat().st_mtime, reverse=True)
|
||||
return flights[:limit]
|
||||
|
||||
|
||||
def parquet_reader(data_dir: Path, dataset: str, *, limit: int | None = None) -> str:
|
||||
"""DuckDB read_parquet() source limited to recent flights."""
|
||||
flights = recent_flights(data_dir, dataset, limit)
|
||||
if not flights:
|
||||
path = data_dir / f"dataset={dataset}" / "**" / "*.parquet"
|
||||
return f"'{path}', hive_partitioning=true, union_by_name=true"
|
||||
if len(flights) == 1:
|
||||
return f"'{flights[0]}/**/*.parquet', hive_partitioning=true, union_by_name=true"
|
||||
inner = ", ".join(f"'{f}/**/*.parquet'" for f in flights)
|
||||
return f"[{inner}], hive_partitioning=true, union_by_name=true"
|
||||
|
||||
|
||||
def iter_parquet_files(data_dir: Path, *, flight_limit: int | None = None) -> list[Path]:
|
||||
"""Parquet paths under recent flights only — avoids full-lake rglob."""
|
||||
limit = flight_limit or flight_window()
|
||||
out: list[Path] = []
|
||||
for ds in ("telemetry", "detections", "state"):
|
||||
for flight in recent_flights(data_dir, ds, limit):
|
||||
out.extend(flight.rglob("*.parquet"))
|
||||
return out
|
||||
@@ -1,7 +1,10 @@
|
||||
global:
|
||||
scrape_interval: 5s
|
||||
scrape_interval: 15s
|
||||
|
||||
scrape_configs:
|
||||
- job_name: swarm
|
||||
static_configs:
|
||||
- targets: ["exporter:9105"]
|
||||
- job_name: explorer
|
||||
static_configs:
|
||||
- targets: ["explorer:8088"]
|
||||
|
||||
@@ -0,0 +1,40 @@
|
||||
"""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])
|
||||
@@ -2,6 +2,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import random
|
||||
import select
|
||||
import time
|
||||
@@ -108,6 +109,10 @@ def run(cfg: Config) -> None:
|
||||
writer.seal()
|
||||
print(f"[{cfg.drone_id}] done: {frames_sent} state frames sent, "
|
||||
f"{len(peers_seen)} peers seen {sorted(peers_seen)}; sealed to {root}")
|
||||
if os.environ.get("KEEP_ALIVE", "0") == "1":
|
||||
print(f"[{cfg.drone_id}] KEEP_ALIVE=1 — idle after seal (no pod restart churn)")
|
||||
while True:
|
||||
time.sleep(3600)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user