feat(simulator): Grafana monitoring over generated Parquet

Monitoring compose profile: dependency-free Prometheus exporter that
scans the Parquet output with DuckDB (rows, detections, pose frames,
battery, RSSI, link distance, bytes on disk), Prometheus scrape config
and a provisioned Grafana dashboard. Demonstrates the observability
doctrine: fleet statistics derived from the data platform itself.

simulator/monitoring/exporter.py

@@ -0,0 +1,132 @@
+"""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.
+"""
+
+from __future__ import annotations
+
+import os
+import threading
+import time
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+
+import duckdb
+
+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"))
+
+_lock = threading.Lock()
+_payload = "# swarm exporter starting\n"
+
+
+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
+
+
+def collect() -> str:
+    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:
+        lines.append(f"# HELP {name} {help_text}")
+        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")],
+    )
+    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')})
+            WHERE sensor='battery' GROUP BY drone""")],
+    )
+    metric(
+        "swarm_rssi_dbm", "Latest RSSI per drone-peer link", "gauge",
+        [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')})
+            WHERE sensor='rssi' GROUP BY drone, peer_id""")],
+    )
+
+    files = list(DATA_DIR.rglob("*.parquet"))
+    metric(
+        "swarm_parquet_bytes", "Bytes on disk per dataset", "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",
+           [f"swarm_parquet_files {len(files)}"])
+    con.close()
+    return "\n".join(lines) + "\n"
+
+
+def scanner() -> None:
+    global _payload
+    while True:
+        started = time.monotonic()
+        try:
+            payload = collect()
+        except Exception as exc:  # keep serving stale metrics over dying
+            payload = f"# collect error: {exc}\n"
+        with _lock:
+            _payload = payload
+        time.sleep(max(0.5, SCAN_INTERVAL_S - (time.monotonic() - started)))
+
+
+class Handler(BaseHTTPRequestHandler):
+    def do_GET(self) -> None:  # noqa: N802 — http.server API
+        if self.path != "/metrics":
+            self.send_response(404)
+            self.end_headers()
+            return
+        with _lock:
+            body = _payload.encode()
+        self.send_response(200)
+        self.send_header("Content-Type", "text/plain; version=0.0.4")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    def log_message(self, *_args: object) -> None:
+        pass  # scrapes every few seconds; keep the log quiet
+
+
+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")
+    ThreadingHTTPServer(("", PORT), Handler).serve_forever()