Luca Sacchi Ricciardi
6c8c05b13b
- Buffer locale samples[] per aggiornamenti real-time - Poll /api/status ogni 10 secondi - Aggiunge nuovi campioni senza ricaricare history - Aggiunge AGENTS.md per istruzioni agenti OpenCode - Aggiunge team agenti in .opencode/agents/
464 lines
15 KiB
Python
464 lines
15 KiB
Python
"""Supabase pinger con API FastAPI, storage storico a buffer circolare e dashboard."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
import threading
|
|
import time
|
|
from contextlib import asynccontextmanager
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import psycopg2
|
|
import uvicorn
|
|
from dotenv import load_dotenv
|
|
from fastapi import FastAPI, HTTPException, Query, Request
|
|
from fastapi.responses import HTMLResponse
|
|
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s %(levelname)-8s %(message)s",
|
|
datefmt="%Y-%m-%d %H:%M:%S",
|
|
stream=sys.stdout,
|
|
)
|
|
log = logging.getLogger("supabase-pinger")
|
|
|
|
REQUIRED_VARS = [
|
|
"SUPABASE_DB_HOST",
|
|
"SUPABASE_DB_PORT",
|
|
"SUPABASE_DB_NAME",
|
|
"SUPABASE_DB_USER",
|
|
"SUPABASE_DB_PASSWORD",
|
|
]
|
|
|
|
|
|
@dataclass
|
|
class Settings:
|
|
db_host: str
|
|
db_port: int
|
|
db_name: str
|
|
db_user: str
|
|
db_password: str
|
|
ping_query: str
|
|
ping_interval_minutes: int
|
|
web_host: str
|
|
web_port: int
|
|
rrd_db_path: str
|
|
rrd_retention_hours: int
|
|
|
|
@property
|
|
def max_samples(self) -> int:
|
|
samples = int((self.rrd_retention_hours * 60) / self.ping_interval_minutes)
|
|
return max(1, samples)
|
|
|
|
|
|
def load_config() -> Settings:
|
|
"""Carica e valida la configurazione da ambiente/.env."""
|
|
load_dotenv()
|
|
|
|
missing = [name for name in REQUIRED_VARS if not os.environ.get(name)]
|
|
if missing:
|
|
raise RuntimeError(
|
|
f"Variabili obbligatorie mancanti: {', '.join(missing)}"
|
|
)
|
|
|
|
try:
|
|
db_port = int(os.environ["SUPABASE_DB_PORT"])
|
|
ping_interval = int(os.environ.get("PING_INTERVAL_MINUTES", "4320"))
|
|
web_port = int(os.environ.get("WEB_PORT", "8080"))
|
|
retention_hours = int(os.environ.get("RRD_RETENTION_HOURS", "48"))
|
|
if ping_interval <= 0 or web_port <= 0 or retention_hours <= 0:
|
|
raise ValueError
|
|
except ValueError as exc:
|
|
raise RuntimeError(
|
|
"SUPABASE_DB_PORT, PING_INTERVAL_MINUTES, WEB_PORT e "
|
|
"RRD_RETENTION_HOURS devono essere interi positivi."
|
|
) from exc
|
|
|
|
return Settings(
|
|
db_host=os.environ["SUPABASE_DB_HOST"],
|
|
db_port=db_port,
|
|
db_name=os.environ["SUPABASE_DB_NAME"],
|
|
db_user=os.environ["SUPABASE_DB_USER"],
|
|
db_password=os.environ["SUPABASE_DB_PASSWORD"],
|
|
ping_query=os.environ.get("PING_QUERY", "SELECT 1;"),
|
|
ping_interval_minutes=ping_interval,
|
|
web_host=os.environ.get("WEB_HOST", "0.0.0.0"),
|
|
web_port=web_port,
|
|
rrd_db_path=os.environ.get("RRD_DB_PATH", "data/connection_rrd.sqlite3"),
|
|
rrd_retention_hours=retention_hours,
|
|
)
|
|
|
|
|
|
class RRDStore:
|
|
"""Storage storico a buffer circolare, dimensionato a finestra 48h."""
|
|
|
|
def __init__(self, path: str, max_samples: int) -> None:
|
|
self.path = path
|
|
self.max_samples = max_samples
|
|
Path(path).parent.mkdir(parents=True, exist_ok=True)
|
|
self._init_db()
|
|
|
|
def _connect(self) -> sqlite3.Connection:
|
|
conn = sqlite3.connect(self.path)
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
def _init_db(self) -> None:
|
|
with self._connect() as conn:
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS samples (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
ts INTEGER NOT NULL,
|
|
success INTEGER NOT NULL,
|
|
latency_ms REAL,
|
|
error_message TEXT
|
|
);
|
|
"""
|
|
)
|
|
conn.commit()
|
|
|
|
def add_sample(
|
|
self, *, ts: int, success: bool, latency_ms: float | None, error_message: str | None
|
|
) -> None:
|
|
with self._connect() as conn:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO samples (ts, success, latency_ms, error_message)
|
|
VALUES (?, ?, ?, ?)
|
|
""",
|
|
(ts, 1 if success else 0, latency_ms, error_message),
|
|
)
|
|
conn.execute(
|
|
"""
|
|
DELETE FROM samples
|
|
WHERE id NOT IN (
|
|
SELECT id FROM samples ORDER BY id DESC LIMIT ?
|
|
)
|
|
""",
|
|
(self.max_samples,),
|
|
)
|
|
conn.commit()
|
|
|
|
def latest(self) -> dict[str, Any] | None:
|
|
with self._connect() as conn:
|
|
row = conn.execute(
|
|
"""
|
|
SELECT ts, success, latency_ms, error_message
|
|
FROM samples
|
|
ORDER BY id DESC
|
|
LIMIT 1
|
|
"""
|
|
).fetchone()
|
|
return self._row_to_dict(row) if row else None
|
|
|
|
def history(self, hours: int) -> list[dict[str, Any]]:
|
|
min_ts = int(time.time()) - (hours * 3600)
|
|
with self._connect() as conn:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT ts, success, latency_ms, error_message
|
|
FROM samples
|
|
WHERE ts >= ?
|
|
ORDER BY ts ASC
|
|
""",
|
|
(min_ts,),
|
|
).fetchall()
|
|
return [self._row_to_dict(row) for row in rows]
|
|
|
|
@staticmethod
|
|
def _row_to_dict(row: sqlite3.Row) -> dict[str, Any]:
|
|
return {
|
|
"timestamp": int(row["ts"]),
|
|
"success": bool(row["success"]),
|
|
"latency_ms": row["latency_ms"],
|
|
"error_message": row["error_message"],
|
|
}
|
|
|
|
|
|
def run_ping(settings: Settings) -> tuple[bool, float | None, str | None]:
|
|
"""Esegue il keep-alive e restituisce esito, latenza, eventuale errore."""
|
|
conn = None
|
|
started = time.perf_counter()
|
|
|
|
try:
|
|
conn = psycopg2.connect(
|
|
host=settings.db_host,
|
|
port=settings.db_port,
|
|
dbname=settings.db_name,
|
|
user=settings.db_user,
|
|
password=settings.db_password,
|
|
connect_timeout=10,
|
|
)
|
|
with conn.cursor() as cur:
|
|
cur.execute(settings.ping_query)
|
|
latency_ms = (time.perf_counter() - started) * 1000
|
|
return True, latency_ms, None
|
|
except psycopg2.Error as exc:
|
|
return False, None, str(exc)
|
|
finally:
|
|
if conn is not None:
|
|
try:
|
|
conn.close()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def collector_loop(settings: Settings, store: RRDStore, stop_event: threading.Event) -> None:
|
|
interval_seconds = settings.ping_interval_minutes * 60
|
|
log.info(
|
|
"Collector keep-alive avviato (interval=%s min, retention=%s ore, max_samples=%s)",
|
|
settings.ping_interval_minutes,
|
|
settings.rrd_retention_hours,
|
|
settings.max_samples,
|
|
)
|
|
|
|
while not stop_event.is_set():
|
|
success, latency_ms, error_message = run_ping(settings)
|
|
sample_ts = int(time.time())
|
|
store.add_sample(
|
|
ts=sample_ts,
|
|
success=success,
|
|
latency_ms=latency_ms,
|
|
error_message=error_message,
|
|
)
|
|
|
|
if success:
|
|
log.info("Keep-alive OK (latency_ms=%.2f)", latency_ms or 0.0)
|
|
else:
|
|
log.warning("Keep-alive failed: %s", error_message)
|
|
|
|
wait_seconds = 0
|
|
while wait_seconds < interval_seconds and not stop_event.is_set():
|
|
time.sleep(min(5, interval_seconds - wait_seconds))
|
|
wait_seconds += 5
|
|
|
|
log.info("Collector keep-alive arrestato")
|
|
|
|
|
|
def _build_dashboard_html() -> str:
|
|
return """
|
|
<!doctype html>
|
|
<html lang=\"it\">
|
|
<head>
|
|
<meta charset=\"utf-8\" />
|
|
<meta name=\"viewport\" content=\"width=device-width, initial-scale=1\" />
|
|
<title>Supabase Pinger Dashboard</title>
|
|
<style>
|
|
:root {
|
|
--bg: #101418;
|
|
--panel: #17202a;
|
|
--line: #3ecf8e;
|
|
--danger: #ff6b6b;
|
|
--text: #edf2f7;
|
|
--muted: #9fb3c8;
|
|
}
|
|
body { margin: 0; font-family: 'Segoe UI', sans-serif; background: radial-gradient(circle at 20% 20%, #1f2937, var(--bg)); color: var(--text); }
|
|
.wrap { max-width: 1100px; margin: 0 auto; padding: 24px; }
|
|
.panel { background: color-mix(in srgb, var(--panel), #000 15%); border: 1px solid #283544; border-radius: 14px; padding: 18px; }
|
|
.top { display: flex; justify-content: space-between; gap: 16px; flex-wrap: wrap; }
|
|
.badge { padding: 6px 10px; border-radius: 999px; font-weight: 700; font-size: 13px; }
|
|
.ok { background: rgba(62, 207, 142, .15); color: #7ff5bd; }
|
|
.ko { background: rgba(255, 107, 107, .15); color: #ff9f9f; }
|
|
.meta { color: var(--muted); font-size: 14px; }
|
|
canvas { margin-top: 18px; }
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<div class=\"wrap\">
|
|
<div class=\"panel\">
|
|
<div class=\"top\">
|
|
<h1>Supabase Pinger - SmokePing Style</h1>
|
|
<div>
|
|
<span id=\"statusBadge\" class=\"badge\">N/A</span>
|
|
</div>
|
|
</div>
|
|
<div class=\"meta\" id=\"meta\">Caricamento...</div>
|
|
<canvas id=\"latencyChart\"></canvas>
|
|
</div>
|
|
</div>
|
|
<script src=\"https://cdn.jsdelivr.net/npm/chart.js\"></script>
|
|
<script>
|
|
const badge = document.getElementById('statusBadge');
|
|
const meta = document.getElementById('meta');
|
|
const ctx = document.getElementById('latencyChart').getContext('2d');
|
|
|
|
// Buffer locale per campioni - mantiene solo i dati nel grafico
|
|
let samples = [];
|
|
const MAX_VISIBLE_SAMPLES = 100; // Limite visualizzazione per performance
|
|
|
|
const chart = new Chart(ctx, {
|
|
type: 'line',
|
|
data: { labels: [], datasets: [
|
|
{ label: 'Latenza ms', data: [], borderColor: '#3ecf8e', tension: 0.2 },
|
|
{ label: 'Errori', data: [], borderColor: '#ff6b6b', stepped: true, yAxisID: 'y1' }
|
|
] },
|
|
options: {
|
|
responsive: true,
|
|
scales: {
|
|
y: { title: { display: true, text: 'ms' } },
|
|
y1: { position: 'right', min: 0, max: 1, ticks: { stepSize: 1 } }
|
|
}
|
|
}
|
|
});
|
|
|
|
async function loadHistory() {
|
|
// Carica history iniziale una sola volta
|
|
const historyRes = await fetch('/api/history?hours=48');
|
|
const history = await historyRes.json();
|
|
samples = history.samples || [];
|
|
updateChart();
|
|
}
|
|
|
|
function updateChart() {
|
|
// Limita i campioni visibili per performance
|
|
const visibleSamples = samples.slice(-MAX_VISIBLE_SAMPLES);
|
|
|
|
chart.data.labels = visibleSamples.map(s => {
|
|
const d = new Date(s.timestamp * 1000);
|
|
return d.toLocaleTimeString([], { hour: '2-digit', minute: '2-digit' });
|
|
});
|
|
chart.data.datasets[0].data = visibleSamples.map(s => s.latency_ms);
|
|
chart.data.datasets[1].data = visibleSamples.map(s => s.success ? 0 : 1);
|
|
chart.update('none'); // update senza animazione per performance
|
|
}
|
|
|
|
async function pollStatus() {
|
|
try {
|
|
const statusRes = await fetch('/api/status');
|
|
const status = await statusRes.json();
|
|
|
|
// Aggiorna badge
|
|
if (status.success === true) {
|
|
badge.textContent = 'UP';
|
|
badge.className = 'badge ok';
|
|
} else if (status.success === false) {
|
|
badge.textContent = 'DOWN';
|
|
badge.className = 'badge ko';
|
|
} else {
|
|
badge.textContent = 'N/A';
|
|
badge.className = 'badge';
|
|
}
|
|
|
|
// Aggiorna meta con timestamp corrente
|
|
meta.textContent = 'Ultimo aggiornamento: ' + new Date().toLocaleTimeString() + ' | Campioni: ' + samples.length;
|
|
|
|
// Aggiunge nuovo sample se non gia presente
|
|
if (status.timestamp && !samples.find(s => s.timestamp === status.timestamp)) {
|
|
samples.push({
|
|
timestamp: status.timestamp,
|
|
success: status.success,
|
|
latency_ms: status.latency_ms,
|
|
error_message: status.error_message
|
|
});
|
|
|
|
// Rimuove campioni piu vecchi di 48h
|
|
const cutoff = Math.floor(Date.now() / 1000) - (48 * 3600);
|
|
samples = samples.filter(s => s.timestamp >= cutoff);
|
|
|
|
updateChart();
|
|
}
|
|
} catch (err) {
|
|
meta.textContent = 'Errore: ' + err.message;
|
|
}
|
|
}
|
|
|
|
// Inizializzazione
|
|
loadHistory().catch((err) => {
|
|
meta.textContent = 'Errore caricamento history: ' + err;
|
|
});
|
|
|
|
// Poll ogni 10 secondi per aggiornamenti real-time
|
|
setInterval(pollStatus, 10000);
|
|
</script>
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
|
|
def create_app(
|
|
config_override: Settings | None = None,
|
|
start_collector: bool = True,
|
|
) -> FastAPI:
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
settings = config_override or load_config()
|
|
store = RRDStore(settings.rrd_db_path, settings.max_samples)
|
|
|
|
app.state.settings = settings
|
|
app.state.store = store
|
|
app.state.collector_stop_event = threading.Event()
|
|
app.state.collector_thread = None
|
|
|
|
if start_collector:
|
|
thread = threading.Thread(
|
|
target=collector_loop,
|
|
args=(settings, store, app.state.collector_stop_event),
|
|
daemon=True,
|
|
name="collector-thread",
|
|
)
|
|
thread.start()
|
|
app.state.collector_thread = thread
|
|
|
|
yield
|
|
|
|
app.state.collector_stop_event.set()
|
|
if app.state.collector_thread is not None:
|
|
app.state.collector_thread.join(timeout=5)
|
|
|
|
app = FastAPI(
|
|
title="Supabase Pinger API",
|
|
version="1.1.0",
|
|
description=(
|
|
"API read-only per monitoraggio storico connessione Supabase "
|
|
"(retention RRD 48h) e dashboard SmokePing-style."
|
|
),
|
|
lifespan=lifespan,
|
|
)
|
|
|
|
@app.get("/", response_class=HTMLResponse, tags=["dashboard"])
|
|
def dashboard() -> str:
|
|
return _build_dashboard_html()
|
|
|
|
@app.get("/api/status", tags=["monitoring"])
|
|
def api_status(request: Request) -> dict[str, Any]:
|
|
latest = request.app.state.store.latest()
|
|
if latest is None:
|
|
return {
|
|
"success": None,
|
|
"timestamp": None,
|
|
"latency_ms": None,
|
|
"error_message": "Nessun campione disponibile",
|
|
}
|
|
return latest
|
|
|
|
@app.get("/api/history", tags=["monitoring"])
|
|
def api_history(
|
|
request: Request,
|
|
hours: int = Query(48, ge=1, le=48),
|
|
) -> dict[str, Any]:
|
|
if hours > request.app.state.settings.rrd_retention_hours:
|
|
raise HTTPException(status_code=400, detail="Finestra richiesta oltre retention")
|
|
samples = request.app.state.store.history(hours)
|
|
return {
|
|
"hours": hours,
|
|
"count": len(samples),
|
|
"samples": samples,
|
|
}
|
|
|
|
return app
|
|
|
|
|
|
app = create_app()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
cfg = load_config()
|
|
uvicorn.run("app:app", host=cfg.web_host, port=cfg.web_port, log_level="info")
|