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.

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()