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.

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.

infra/ansible/sim-cluster.yml

@@ -29,12 +29,16 @@
         path: "{{ data_dir }}"
         state: directory
         mode: "0777"
+        owner: "10001"
+        group: "10001"
 
     - name: Create warehouse directory (T3 host path inside the k3d node)
       ansible.builtin.file:
         path: "{{ warehouse_dir }}"
         state: directory
         mode: "0777"
+        owner: "10001"
+        group: "10001"
 
     - name: List existing clusters
       ansible.builtin.command: k3d cluster list -o json

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
   }
 }
@@ -298,8 +340,12 @@ resource "kubernetes_config_map" "grafana_provisioning" {
   data = {
     "datasource.yml" = <<-EOT
       apiVersion: 1
+      deleteDatasources:
+        - name: Prometheus
+          orgId: 1
       datasources:
         - name: Prometheus
+          uid: swarm-prom
           type: prometheus
           access: proxy
           url: http://prometheus:9090
@@ -323,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
+    }
   }
 }
 

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

prototype/src/live.ts

@@ -20,24 +20,24 @@ export interface LivePose {
 // latest battery reading from telemetry. arg_max keeps it a single scan.
 const LIVE_SQL = `
 WITH pose AS (
-  SELECT drone_id,
+  SELECT drone,
          arg_max(pos_x, ts_ns) AS x,
          arg_max(pos_y, ts_ns) AS y,
          arg_max(yaw, ts_ns)   AS yaw,
          max(ts_ns)            AS ts_ns
   FROM state
   WHERE direction = 'sent'
-  GROUP BY drone_id
+  GROUP BY drone
 ),
 batt AS (
-  SELECT drone_id, arg_max(level_pct, ts_ns) AS battery
+  SELECT drone, arg_max(level_pct, ts_ns) AS battery
   FROM telemetry
   WHERE sensor = 'battery'
-  GROUP BY drone_id
+  GROUP BY drone
 )
-SELECT p.drone_id, p.x, p.y, p.yaw, p.ts_ns, b.battery
+SELECT p.drone, p.x, p.y, p.yaw, p.ts_ns, b.battery
-FROM pose p LEFT JOIN batt b USING (drone_id)
+FROM pose p LEFT JOIN batt b USING (drone)
-ORDER BY p.drone_id`;
+ORDER BY p.drone`;
 
 interface QueryResponse {
   columns?: string[];

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

simulator/explorer/server.py

@@ -1,33 +1,28 @@
-"""Data-plane explorer: a web view over the Hive-partitioned Parquet lake.
+"""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.
-"""
 
 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:
+def _refresh_views() -> None:
-    """Fresh connection with the three datasets pre-registered as views."""
+    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(
+            con.execute(f"CREATE OR REPLACE VIEW {ds} AS SELECT * FROM read_parquet({reader})")
-                f"CREATE VIEW {ds} AS SELECT * FROM read_parquet("
-                f"'{pattern}', hive_partitioning=true, union_by_name=true)"
-            )
         except duckdb.Error:
-            pass  # dataset not written yet; view simply won't exist
+            pass
-    return con
+    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()
+    started = time.monotonic()
-    try:
+    with _lock:
-        cur = con.sql(sql)
+        con = connect()
-        columns = cur.columns
+        try:
-        rows = cur.fetchmany(ROW_LIMIT)
+            cur = con.sql(sql)
-        return {
+            columns = cur.columns
-            "columns": columns,
+            rows = cur.fetchmany(ROW_LIMIT)
-            "rows": [[repr(v) if isinstance(v, bytes) else v for v in row] for row in rows],
+            _query_count += 1
-            "truncated": len(rows) == ROW_LIMIT,
+            _last_query_s = time.monotonic() - started
-        }
+            return {
-    except duckdb.Error as exc:
+                "columns": columns,
-        return {"error": str(exc)}
+                "rows": [[repr(v) if isinstance(v, bytes) else v for v in row] for row in rows],
-    finally:
+                "truncated": len(rows) == ROW_LIMIT,
-        con.close()
+            }
+        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()

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
+/metrics. Scans only the most recent flight partitions by default so CPU
-text, served with the standard library HTTP server.
+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(
+    for ds, name in (
-        "swarm_rows_total", "Telemetry rows written per drone and sensor", "gauge",
+        ("telemetry", "swarm_rows_total"),
-        [f'swarm_rows_total{{drone="{d}",sensor="{s}"}} {n}'
+        ("detections", "swarm_detections_total"),
-         for d, s, n in _q(con, f"SELECT drone, sensor, count(*) FROM read_parquet({glob('telemetry')}) GROUP BY 1,2")],
+        ("state", "swarm_state_frames_total"),
-    )
+    ):
-    metric(
+        reader = parquet_reader(DATA_DIR, ds)
-        "swarm_detections_total", "Detection events per drone and class", "gauge",
+        if ds == "telemetry":
-        [f'swarm_detections_total{{drone="{d}",cls="{c}"}} {n}'
+            metric(
-         for d, c, n in _q(con, f"SELECT drone, cls, count(*) FROM read_parquet({glob('detections')}) GROUP BY 1,2")],
+                name, f"Rows in dataset={ds} (recent {flight_window()} flights)", "gauge",
-    )
+                [f'{name}{{drone="{d}",sensor="{s}"}} {n}'
-    metric(
+                 for d, s, n in _q(con, f"SELECT drone, sensor, count(*) FROM read_parquet({reader}) GROUP BY 1,2")],
-        "swarm_state_frames_total", "Pose broadcast frames per drone and direction", "gauge",
+            )
-        [f'swarm_state_frames_total{{drone="{d}",direction="{dr}"}} {n}'
+        elif ds == "detections":
-         for d, dr, n in _q(con, f"SELECT drone, direction, count(*) FROM read_parquet({glob('state')}) GROUP BY 1,2")],
+            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')})
+            FROM read_parquet({telem})
-            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"))
+    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()

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": [] }
+    }
+  ]
+}

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

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"]

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

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__":