diff --git a/docs/07-observability.md b/docs/07-observability.md index 10f60fb..f48dc9e 100644 --- a/docs/07-observability.md +++ b/docs/07-observability.md @@ -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. diff --git a/infra/terraform/sim-env/main.tf b/infra/terraform/sim-env/main.tf index 5964d49..dbf20ce 100644 --- a/infra/terraform/sim-env/main.tf +++ b/infra/terraform/sim-env/main.tf @@ -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 + } } } diff --git a/prototype/src/App.tsx b/prototype/src/App.tsx index c6af2d0..9e0c2a6 100644 --- a/prototype/src/App.tsx +++ b/prototype/src/App.tsx @@ -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); diff --git a/simulator/docker-compose.yml b/simulator/docker-compose.yml index 38e6bb4..68b9504 100644 --- a/simulator/docker-compose.yml +++ b/simulator/docker-compose.yml @@ -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 diff --git a/simulator/explorer/server.py b/simulator/explorer/server.py index b649521..a9efd99 100644 --- a/simulator/explorer/server.py +++ b/simulator/explorer/server.py @@ -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() diff --git a/simulator/monitoring/exporter.py b/simulator/monitoring/exporter.py index b5d0113..2997cd8 100644 --- a/simulator/monitoring/exporter.py +++ b/simulator/monitoring/exporter.py @@ -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() diff --git a/simulator/monitoring/grafana/dashboards/swarm-platform.json b/simulator/monitoring/grafana/dashboards/swarm-platform.json new file mode 100644 index 0000000..91c118a --- /dev/null +++ b/simulator/monitoring/grafana/dashboards/swarm-platform.json @@ -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": [] } + } + ] +} diff --git a/simulator/monitoring/lake.py b/simulator/monitoring/lake.py new file mode 100644 index 0000000..1761f1c --- /dev/null +++ b/simulator/monitoring/lake.py @@ -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 diff --git a/simulator/monitoring/prometheus.yml b/simulator/monitoring/prometheus.yml index 885c7c0..a8c9431 100644 --- a/simulator/monitoring/prometheus.yml +++ b/simulator/monitoring/prometheus.yml @@ -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"] diff --git a/simulator/tests/test_lake.py b/simulator/tests/test_lake.py new file mode 100644 index 0000000..c9438a1 --- /dev/null +++ b/simulator/tests/test_lake.py @@ -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]) diff --git a/simulator/virtual_drone/main.py b/simulator/virtual_drone/main.py index 3f10389..a9dfbfc 100644 --- a/simulator/virtual_drone/main.py +++ b/simulator/virtual_drone/main.py @@ -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__":