eSlider
ea062af86a
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.
213 lines
7.3 KiB
Python
213 lines
7.3 KiB
Python
"""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
|
|
|
|
_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"
|
|
r"|INSTALL|LOAD|SET|RESET|PRAGMA|CALL|IMPORT|VACUUM|CHECKPOINT)\b",
|
|
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:
|
|
stripped = re.sub(r"--[^\n]*|/\*.*?\*/", " ", sql, flags=re.DOTALL).strip().rstrip(";")
|
|
if not stripped:
|
|
return "empty statement"
|
|
if ";" in stripped:
|
|
return "multiple statements are not allowed"
|
|
if not _ALLOWED_START.match(stripped):
|
|
return "only SELECT / WITH / DESCRIBE / SUMMARIZE / SHOW are allowed"
|
|
if _FORBIDDEN.search(stripped):
|
|
return "statement contains a forbidden keyword"
|
|
return None
|
|
|
|
|
|
def _refresh_views() -> None:
|
|
global _con, _views_at
|
|
con = duckdb.connect()
|
|
for ds in ("telemetry", "detections", "state"):
|
|
reader = parquet_reader(DATA_DIR, ds)
|
|
try:
|
|
con.execute(f"CREATE OR REPLACE VIEW {ds} AS SELECT * FROM read_parquet({reader})")
|
|
except duckdb.Error:
|
|
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:
|
|
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(iter_parquet_files(DATA_DIR)):
|
|
rel = f.relative_to(DATA_DIR)
|
|
size = f.stat().st_size
|
|
total_bytes += size
|
|
total_files += 1
|
|
node = root
|
|
for part in rel.parts[:-1]:
|
|
node = node.setdefault("children", {}).setdefault(part, {})
|
|
leaf = node.setdefault("children", {}).setdefault(rel.parts[-1], {})
|
|
leaf["bytes"] = size
|
|
node = root
|
|
node["bytes"] = node.get("bytes", 0) + size
|
|
node["files"] = node.get("files", 0) + 1
|
|
for part in rel.parts[:-1]:
|
|
node = node["children"][part]
|
|
node["bytes"] = node.get("bytes", 0) + size
|
|
node["files"] = node.get("files", 0) + 1
|
|
_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}"}
|
|
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):
|
|
def _send(self, code: int, body: bytes, ctype: str) -> None:
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", ctype)
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.send_header("Access-Control-Allow-Origin", "*")
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
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")
|
|
self.send_header("Access-Control-Allow-Headers", "Content-Type")
|
|
self.end_headers()
|
|
|
|
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
|
|
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
|
|
if self.path != "/api/query":
|
|
self._send(404, b"not found", "text/plain")
|
|
return
|
|
length = int(self.headers.get("Content-Length", "0"))
|
|
try:
|
|
sql = json.loads(self.rfile.read(length)).get("sql", "")
|
|
except (json.JSONDecodeError, AttributeError):
|
|
self._json({"error": "invalid request body"}, 400)
|
|
return
|
|
self._json(run_query(sql))
|
|
|
|
def log_message(self, *_args: object) -> None:
|
|
pass
|
|
|
|
|
|
def _view_loop() -> None:
|
|
while True:
|
|
try:
|
|
_refresh_views()
|
|
except Exception:
|
|
pass
|
|
time.sleep(VIEW_REFRESH_S)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
_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()
|