From 8f98a0c9940f1805e44d81daad1bd4edbdc7fef3 Mon Sep 17 00:00:00 2001 From: Andriy Oblivantsev Date: Wed, 8 Jul 2026 13:27:53 +0100 Subject: [PATCH] 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. --- docs/07-observability.md | 2 + simulator/README.md | 15 ++ simulator/docker-compose.yml | 38 +++++ simulator/monitoring/exporter.py | 132 ++++++++++++++++++ .../monitoring/grafana/dashboards/swarm.json | 116 +++++++++++++++ .../provisioning/dashboards/provider.yml | 8 ++ .../provisioning/datasources/prometheus.yml | 9 ++ simulator/monitoring/prometheus.yml | 7 + 8 files changed, 327 insertions(+) create mode 100644 simulator/monitoring/exporter.py create mode 100644 simulator/monitoring/grafana/dashboards/swarm.json create mode 100644 simulator/monitoring/grafana/provisioning/dashboards/provider.yml create mode 100644 simulator/monitoring/grafana/provisioning/datasources/prometheus.yml create mode 100644 simulator/monitoring/prometheus.yml diff --git a/docs/07-observability.md b/docs/07-observability.md index c149fb0..10f60fb 100644 --- a/docs/07-observability.md +++ b/docs/07-observability.md @@ -53,4 +53,6 @@ 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. + 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/simulator/README.md b/simulator/README.md index 04c0448..d32bacc 100644 --- a/simulator/README.md +++ b/simulator/README.md @@ -50,6 +50,21 @@ print(con.sql(""" EOF ``` +## Monitoring: Grafana over the generated data + +The `monitoring` profile spins up a small metrics chain — a DuckDB-based exporter that scans the generated Parquet every few seconds, Prometheus, and a pre-provisioned Grafana dashboard: + +```bash +docker compose --profile monitoring up -d # exporter + prometheus + grafana +# generate some flights in parallel or beforehand: +FLIGHT_ID=$(date -u +%Y%m%dT%H%MZ)-sim docker compose up --scale drone=5 +``` + +- Grafana: `http://localhost:3000` (anonymous admin — demo only) → dashboard **Swarm Fleet — generated data** +- Prometheus: `http://localhost:9090` · exporter: `http://localhost:9105/metrics` + +Panels: telemetry rows by drone/sensor, detections by class, pose frames sent/received, battery per drone, RSSI and estimated distance per link, Parquet bytes/files on disk. The exporter is deliberately a demonstration of the observability doctrine from [07 — Observability](../docs/07-observability.md): fleet statistics are *derived from the data platform itself* — no agent on the drone, just SQL over the same Parquet everyone else reads. + ## Reproducibility Every run is deterministic per `(SEED, DRONE_ID)`: same route jitter, same sensor noise, same detection sequence. A bug report is a seed and a config, not a description. diff --git a/simulator/docker-compose.yml b/simulator/docker-compose.yml index d8b5225..b20724b 100644 --- a/simulator/docker-compose.yml +++ b/simulator/docker-compose.yml @@ -15,3 +15,41 @@ services: BROADCAST_ADDR: ${BROADCAST_ADDR:-255.255.255.255} volumes: - ./data:/data + + # --- Monitoring profile: docker compose --profile monitoring up --- + + exporter: + build: . + profiles: [monitoring] + command: ["python", "monitoring/exporter.py"] + environment: + DATA_DIR: /data + SCAN_INTERVAL_S: "5" + volumes: + - ./data:/data:ro + - ./monitoring:/app/monitoring:ro + ports: + - "9105:9105" + + prometheus: + image: prom/prometheus:v2.53.0 + profiles: [monitoring] + volumes: + - ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml:ro + ports: + - "9090:9090" + + grafana: + image: grafana/grafana:11.1.0 + profiles: [monitoring] + environment: + GF_AUTH_ANONYMOUS_ENABLED: "true" + GF_AUTH_ANONYMOUS_ORG_ROLE: Admin + GF_AUTH_DISABLE_LOGIN_FORM: "true" + volumes: + - ./monitoring/grafana/provisioning:/etc/grafana/provisioning:ro + - ./monitoring/grafana/dashboards:/var/lib/grafana/dashboards:ro + ports: + - "3000:3000" + depends_on: + - prometheus diff --git a/simulator/monitoring/exporter.py b/simulator/monitoring/exporter.py new file mode 100644 index 0000000..b5d0113 --- /dev/null +++ b/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() diff --git a/simulator/monitoring/grafana/dashboards/swarm.json b/simulator/monitoring/grafana/dashboards/swarm.json new file mode 100644 index 0000000..996b23b --- /dev/null +++ b/simulator/monitoring/grafana/dashboards/swarm.json @@ -0,0 +1,116 @@ +{ + "uid": "swarm-fleet", + "title": "Swarm Fleet — generated data", + "tags": ["swarm"], + "timezone": "utc", + "schemaVersion": 39, + "version": 1, + "refresh": "5s", + "time": { "from": "now-15m", "to": "now" }, + "panels": [ + { + "id": 1, "type": "stat", "title": "Telemetry rows", + "gridPos": { "h": 5, "w": 4, "x": 0, "y": 0 }, + "datasource": { "type": "prometheus", "uid": "swarm-prom" }, + "targets": [{ "expr": "sum(swarm_rows_total)", "instant": true, "refId": "A" }], + "options": { "reduceOptions": { "calcs": ["lastNotNull"] }, "colorMode": "value" }, + "fieldConfig": { "defaults": { "unit": "short", "color": { "mode": "fixed", "fixedColor": "blue" } }, "overrides": [] } + }, + { + "id": 2, "type": "stat", "title": "Detections", + "gridPos": { "h": 5, "w": 4, "x": 4, "y": 0 }, + "datasource": { "type": "prometheus", "uid": "swarm-prom" }, + "targets": [{ "expr": "sum(swarm_detections_total)", "instant": true, "refId": "A" }], + "options": { "reduceOptions": { "calcs": ["lastNotNull"] }, "colorMode": "value" }, + "fieldConfig": { "defaults": { "unit": "short", "color": { "mode": "fixed", "fixedColor": "green" } }, "overrides": [] } + }, + { + "id": 3, "type": "stat", "title": "Pose frames (sent+received)", + "gridPos": { "h": 5, "w": 4, "x": 8, "y": 0 }, + "datasource": { "type": "prometheus", "uid": "swarm-prom" }, + "targets": [{ "expr": "sum(swarm_state_frames_total)", "instant": true, "refId": "A" }], + "options": { "reduceOptions": { "calcs": ["lastNotNull"] }, "colorMode": "value" }, + "fieldConfig": { "defaults": { "unit": "short" }, "overrides": [] } + }, + { + "id": 4, "type": "stat", "title": "Parquet on disk", + "gridPos": { "h": 5, "w": 4, "x": 12, "y": 0 }, + "datasource": { "type": "prometheus", "uid": "swarm-prom" }, + "targets": [{ "expr": "sum(swarm_parquet_bytes)", "instant": true, "refId": "A" }], + "options": { "reduceOptions": { "calcs": ["lastNotNull"] }, "colorMode": "value" }, + "fieldConfig": { "defaults": { "unit": "bytes", "color": { "mode": "fixed", "fixedColor": "purple" } }, "overrides": [] } + }, + { + "id": 5, "type": "stat", "title": "Parquet files", + "gridPos": { "h": 5, "w": 4, "x": 16, "y": 0 }, + "datasource": { "type": "prometheus", "uid": "swarm-prom" }, + "targets": [{ "expr": "swarm_parquet_files", "instant": true, "refId": "A" }], + "options": { "reduceOptions": { "calcs": ["lastNotNull"] } }, + "fieldConfig": { "defaults": { "unit": "short" }, "overrides": [] } + }, + { + "id": 6, "type": "gauge", "title": "Fleet battery (min)", + "gridPos": { "h": 5, "w": 4, "x": 20, "y": 0 }, + "datasource": { "type": "prometheus", "uid": "swarm-prom" }, + "targets": [{ "expr": "min(swarm_battery_pct)", "instant": true, "refId": "A" }], + "fieldConfig": { + "defaults": { + "unit": "percent", "min": 0, "max": 100, + "thresholds": { "mode": "absolute", "steps": [ + { "color": "red", "value": null }, { "color": "yellow", "value": 30 }, { "color": "green", "value": 60 } + ] } + }, "overrides": [] + } + }, + { + "id": 7, "type": "timeseries", "title": "Telemetry rows by drone", + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 5 }, + "datasource": { "type": "prometheus", "uid": "swarm-prom" }, + "targets": [{ "expr": "sum by (drone) (swarm_rows_total)", "legendFormat": "{{drone}}", "refId": "A" }], + "fieldConfig": { "defaults": { "unit": "short", "custom": { "fillOpacity": 12 } }, "overrides": [] } + }, + { + "id": 8, "type": "timeseries", "title": "Rows by sensor (fleet)", + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 5 }, + "datasource": { "type": "prometheus", "uid": "swarm-prom" }, + "targets": [{ "expr": "sum by (sensor) (swarm_rows_total)", "legendFormat": "{{sensor}}", "refId": "A" }], + "fieldConfig": { "defaults": { "unit": "short", "custom": { "fillOpacity": 12 } }, "overrides": [] } + }, + { + "id": 9, "type": "timeseries", "title": "Battery per drone", + "gridPos": { "h": 8, "w": 8, "x": 0, "y": 13 }, + "datasource": { "type": "prometheus", "uid": "swarm-prom" }, + "targets": [{ "expr": "swarm_battery_pct", "legendFormat": "{{drone}}", "refId": "A" }], + "fieldConfig": { "defaults": { "unit": "percent", "min": 0, "max": 100 }, "overrides": [] } + }, + { + "id": 10, "type": "timeseries", "title": "RSSI per link", + "gridPos": { "h": 8, "w": 8, "x": 8, "y": 13 }, + "datasource": { "type": "prometheus", "uid": "swarm-prom" }, + "targets": [{ "expr": "swarm_rssi_dbm", "legendFormat": "{{drone}} ← {{peer}}", "refId": "A" }], + "fieldConfig": { "defaults": { "unit": "dBm" }, "overrides": [] } + }, + { + "id": 11, "type": "timeseries", "title": "Inter-drone distance", + "gridPos": { "h": 8, "w": 8, "x": 16, "y": 13 }, + "datasource": { "type": "prometheus", "uid": "swarm-prom" }, + "targets": [{ "expr": "swarm_peer_distance_m", "legendFormat": "{{drone}} ↔ {{peer}}", "refId": "A" }], + "fieldConfig": { "defaults": { "unit": "lengthm" }, "overrides": [] } + }, + { + "id": 12, "type": "bargauge", "title": "Detections by class", + "gridPos": { "h": 7, "w": 12, "x": 0, "y": 21 }, + "datasource": { "type": "prometheus", "uid": "swarm-prom" }, + "targets": [{ "expr": "sum by (cls) (swarm_detections_total)", "legendFormat": "{{cls}}", "instant": true, "refId": "A" }], + "options": { "displayMode": "gradient", "orientation": "horizontal", "reduceOptions": { "calcs": ["lastNotNull"] } }, + "fieldConfig": { "defaults": { "unit": "short", "color": { "mode": "continuous-GrYlRd" } }, "overrides": [] } + }, + { + "id": 13, "type": "timeseries", "title": "Pose frames by direction", + "gridPos": { "h": 7, "w": 12, "x": 12, "y": 21 }, + "datasource": { "type": "prometheus", "uid": "swarm-prom" }, + "targets": [{ "expr": "sum by (drone, direction) (swarm_state_frames_total)", "legendFormat": "{{drone}} {{direction}}", "refId": "A" }], + "fieldConfig": { "defaults": { "unit": "short", "custom": { "fillOpacity": 8 } }, "overrides": [] } + } + ] +} diff --git a/simulator/monitoring/grafana/provisioning/dashboards/provider.yml b/simulator/monitoring/grafana/provisioning/dashboards/provider.yml new file mode 100644 index 0000000..00c0b13 --- /dev/null +++ b/simulator/monitoring/grafana/provisioning/dashboards/provider.yml @@ -0,0 +1,8 @@ +apiVersion: 1 + +providers: + - name: swarm + folder: "" + type: file + options: + path: /var/lib/grafana/dashboards diff --git a/simulator/monitoring/grafana/provisioning/datasources/prometheus.yml b/simulator/monitoring/grafana/provisioning/datasources/prometheus.yml new file mode 100644 index 0000000..2cd546b --- /dev/null +++ b/simulator/monitoring/grafana/provisioning/datasources/prometheus.yml @@ -0,0 +1,9 @@ +apiVersion: 1 + +datasources: + - name: Prometheus + uid: swarm-prom + type: prometheus + access: proxy + url: http://prometheus:9090 + isDefault: true diff --git a/simulator/monitoring/prometheus.yml b/simulator/monitoring/prometheus.yml new file mode 100644 index 0000000..885c7c0 --- /dev/null +++ b/simulator/monitoring/prometheus.yml @@ -0,0 +1,7 @@ +global: + scrape_interval: 5s + +scrape_configs: + - job_name: swarm + static_configs: + - targets: ["exporter:9105"]