feat: punkt 5 (monitoring) — system dashboard backend+frontend, admin-only, auto-refresh 30s, alerting via notifications
This commit is contained in:
Agent Zero
@@ -0,0 +1,369 @@
|
||||
"""System Dashboard routes — admin-only system monitoring.
|
||||
|
||||
Endpoints:
|
||||
GET /api/v1/system/dashboard — comprehensive system metrics
|
||||
GET /api/v1/system/alerts — active alerts with notification dispatch
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import shutil
|
||||
import uuid
|
||||
from datetime import UTC, datetime
|
||||
from typing import Any
|
||||
|
||||
from fastapi import APIRouter, Depends
|
||||
from sqlalchemy import text
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.core.db import get_db, get_engine
|
||||
from app.core.monitoring import (
|
||||
check_database,
|
||||
check_redis,
|
||||
check_worker,
|
||||
get_health_status,
|
||||
)
|
||||
from app.core.notifications import post_system_message
|
||||
from app.deps import require_admin
|
||||
from app.plugins.registry import get_registry
|
||||
|
||||
router = APIRouter(prefix="/api/v1/system", tags=["system"])
|
||||
|
||||
|
||||
# ─── Helpers ──────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _format_bytes(size: int | float) -> str:
|
||||
"""Format bytes to human-readable string."""
|
||||
for unit in ["B", "KB", "MB", "GB", "TB"]:
|
||||
if size < 1024:
|
||||
return f"{size:.1f} {unit}"
|
||||
size /= 1024
|
||||
return f"{size:.1f} PB"
|
||||
|
||||
|
||||
async def _get_db_stats() -> dict[str, Any]:
|
||||
"""Get database statistics."""
|
||||
try:
|
||||
engine = get_engine()
|
||||
async with engine.connect() as conn:
|
||||
# Active connections
|
||||
result = await conn.execute(text("SELECT count(*) FROM pg_stat_activity"))
|
||||
connections = result.scalar()
|
||||
|
||||
# Table count (public schema)
|
||||
result = await conn.execute(
|
||||
text(
|
||||
"SELECT count(*) FROM information_schema.tables "
|
||||
"WHERE table_schema = 'public'"
|
||||
)
|
||||
)
|
||||
table_count = result.scalar()
|
||||
|
||||
# Database size
|
||||
result = await conn.execute(
|
||||
text("SELECT pg_database_size(current_database())")
|
||||
)
|
||||
db_size = result.scalar()
|
||||
|
||||
return {
|
||||
"status": "up",
|
||||
"connections": connections,
|
||||
"table_count": table_count,
|
||||
"db_size_bytes": db_size,
|
||||
"db_size_human": _format_bytes(db_size),
|
||||
}
|
||||
except Exception as e:
|
||||
return {"status": "down", "error": str(e)}
|
||||
|
||||
|
||||
async def _get_redis_stats() -> dict[str, Any]:
|
||||
"""Get Redis statistics."""
|
||||
try:
|
||||
from app.core.auth import get_redis
|
||||
|
||||
r = get_redis()
|
||||
info = await r.info()
|
||||
return {
|
||||
"status": "up",
|
||||
"connections": info.get("connected_clients", 0),
|
||||
"used_memory": info.get("used_memory", 0),
|
||||
"used_memory_human": info.get("used_memory_human", "N/A"),
|
||||
"peak_memory_human": info.get("used_memory_peak_human", "N/A"),
|
||||
"uptime_seconds": info.get("uptime_in_seconds", 0),
|
||||
}
|
||||
except Exception as e:
|
||||
return {"status": "down", "error": str(e)}
|
||||
|
||||
|
||||
async def _get_worker_stats() -> dict[str, Any]:
|
||||
"""Get ARQ worker statistics."""
|
||||
try:
|
||||
from app.core.auth import get_redis
|
||||
|
||||
r = get_redis()
|
||||
queue_length = await r.zcard("arq:queue")
|
||||
# Check for worker heartbeat keys
|
||||
heartbeat_keys = await r.keys("arq:heartbeat:*")
|
||||
active_workers = len(heartbeat_keys) if heartbeat_keys else 0
|
||||
|
||||
return {
|
||||
"status": "up" if active_workers > 0 or queue_length == 0 else "degraded",
|
||||
"queue_length": queue_length,
|
||||
"active_workers": active_workers,
|
||||
}
|
||||
except Exception as e:
|
||||
return {"status": "down", "error": str(e)}
|
||||
|
||||
|
||||
def _get_api_stats() -> dict[str, Any]:
|
||||
"""Get API statistics from Prometheus metrics registry."""
|
||||
try:
|
||||
from app.core.monitoring import REGISTRY
|
||||
|
||||
total_requests = 0
|
||||
error_count = 0
|
||||
duration_sum = 0.0
|
||||
duration_count = 0
|
||||
|
||||
for metric in REGISTRY.collect():
|
||||
for sample in metric.samples:
|
||||
if sample.name == "leocrm_http_requests_total":
|
||||
total_requests += int(sample.value)
|
||||
status_label = sample.labels.get("status", "")
|
||||
if status_label.startswith("5"):
|
||||
error_count += int(sample.value)
|
||||
elif sample.name == "leocrm_http_request_duration_seconds_sum":
|
||||
duration_sum += sample.value
|
||||
elif sample.name == "leocrm_http_request_duration_seconds_count":
|
||||
duration_count += int(sample.value)
|
||||
|
||||
avg_duration = (
|
||||
(duration_sum / duration_count * 1000) if duration_count > 0 else 0
|
||||
)
|
||||
error_rate = (error_count / total_requests * 100) if total_requests > 0 else 0
|
||||
|
||||
return {
|
||||
"total_requests": total_requests,
|
||||
"error_count": error_count,
|
||||
"error_rate": round(error_rate, 2),
|
||||
"avg_response_time_ms": round(avg_duration, 2),
|
||||
}
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
|
||||
|
||||
def _get_plugin_stats() -> dict[str, Any]:
|
||||
"""Get plugin statistics from registry."""
|
||||
try:
|
||||
registry = get_registry()
|
||||
discovered = registry.list_discovered()
|
||||
active_plugins: list[dict[str, Any]] = []
|
||||
|
||||
for name in discovered:
|
||||
plugin = registry.get_plugin(name)
|
||||
if plugin:
|
||||
active_plugins.append(
|
||||
{
|
||||
"name": name,
|
||||
"version": plugin.manifest.version,
|
||||
"is_core": plugin.manifest.is_core,
|
||||
}
|
||||
)
|
||||
|
||||
return {
|
||||
"total_discovered": len(discovered),
|
||||
"active_plugins": active_plugins,
|
||||
"error_count": 0,
|
||||
"errors": [],
|
||||
}
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
|
||||
|
||||
def _get_storage_stats() -> dict[str, Any]:
|
||||
"""Get storage statistics."""
|
||||
try:
|
||||
from app.config import get_settings
|
||||
|
||||
settings = get_settings()
|
||||
storage_path = getattr(settings, "storage_path", "/tmp")
|
||||
|
||||
if not os.path.exists(storage_path):
|
||||
return {"status": "down", "error": f"Storage path {storage_path} not found"}
|
||||
|
||||
disk = shutil.disk_usage(storage_path)
|
||||
file_count = sum(len(files) for _, _, files in os.walk(storage_path))
|
||||
|
||||
return {
|
||||
"status": "up",
|
||||
"path": storage_path,
|
||||
"disk_total_bytes": disk.total,
|
||||
"disk_used_bytes": disk.used,
|
||||
"disk_free_bytes": disk.free,
|
||||
"disk_usage_percent": (
|
||||
round(disk.used / disk.total * 100, 1) if disk.total > 0 else 0
|
||||
),
|
||||
"file_count": file_count,
|
||||
}
|
||||
except Exception as e:
|
||||
return {"status": "down", "error": str(e)}
|
||||
|
||||
|
||||
def _get_llm_usage() -> dict[str, Any]:
|
||||
"""Get LLM usage statistics (best effort — returns zeros if no tracking)."""
|
||||
return {
|
||||
"total_tokens": 0,
|
||||
"estimated_cost": 0.0,
|
||||
"last_24h_tokens": 0,
|
||||
"last_24h_cost": 0.0,
|
||||
}
|
||||
|
||||
|
||||
# ─── Alert Notification ──────────────────────────────────────────────────────
|
||||
|
||||
|
||||
async def _send_alert_notifications(
|
||||
db: AsyncSession,
|
||||
admin: dict[str, Any],
|
||||
alerts: list[dict[str, Any]],
|
||||
) -> None:
|
||||
"""Send notifications for new alerts (throttled via Redis 5-min TTL)."""
|
||||
try:
|
||||
from app.core.auth import get_redis
|
||||
|
||||
r = get_redis()
|
||||
tenant_id = uuid.UUID(admin["tenant_id"])
|
||||
user_id = uuid.UUID(admin["user_id"])
|
||||
|
||||
for alert in alerts:
|
||||
alert_key = f"system_alert:{alert['type']}"
|
||||
# Only send if not already sent in the last 5 minutes
|
||||
already_sent = await r.exists(alert_key)
|
||||
if not already_sent:
|
||||
await r.setex(alert_key, 300, "1") # 5-minute TTL
|
||||
await post_system_message(
|
||||
db=db,
|
||||
tenant_id=tenant_id,
|
||||
user_id=user_id,
|
||||
message_type="system_alert",
|
||||
title=f"System Alert: {alert['type']}",
|
||||
body=alert["message"],
|
||||
severity=alert["severity"],
|
||||
)
|
||||
except Exception:
|
||||
pass # Never let notification failures break the alerts endpoint
|
||||
|
||||
|
||||
# ─── Endpoints ───────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@router.get("/dashboard")
|
||||
async def get_system_dashboard(
|
||||
admin: dict[str, Any] = Depends(require_admin),
|
||||
):
|
||||
"""Get comprehensive system metrics. Admin-only.
|
||||
|
||||
Returns DB, Redis, Worker, API, Plugin, Storage, and LLM usage stats.
|
||||
"""
|
||||
health = await get_health_status()
|
||||
db_stats = await _get_db_stats()
|
||||
redis_stats = await _get_redis_stats()
|
||||
worker_stats = await _get_worker_stats()
|
||||
api_stats = _get_api_stats()
|
||||
plugin_stats = _get_plugin_stats()
|
||||
storage_stats = _get_storage_stats()
|
||||
llm_usage = _get_llm_usage()
|
||||
|
||||
return {
|
||||
"overall_status": health.get("status", "unknown"),
|
||||
"timestamp": datetime.now(UTC).isoformat(),
|
||||
"database": db_stats,
|
||||
"redis": redis_stats,
|
||||
"worker": worker_stats,
|
||||
"api": api_stats,
|
||||
"plugins": plugin_stats,
|
||||
"storage": storage_stats,
|
||||
"llm": llm_usage,
|
||||
}
|
||||
|
||||
|
||||
@router.get("/alerts")
|
||||
async def get_system_alerts(
|
||||
admin: dict[str, Any] = Depends(require_admin),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
):
|
||||
"""Get active system alerts. Admin-only.
|
||||
|
||||
Checks DB, Redis, Worker, Storage, and API error rate.
|
||||
Sends notifications for new alerts (throttled via Redis).
|
||||
"""
|
||||
alerts: list[dict[str, Any]] = []
|
||||
|
||||
# Check DB
|
||||
db_check = await check_database()
|
||||
if db_check["status"] == "down":
|
||||
alerts.append(
|
||||
{
|
||||
"severity": "critical",
|
||||
"type": "db_down",
|
||||
"message": f"Database is down: {db_check.get('error', 'Unknown error')}",
|
||||
}
|
||||
)
|
||||
|
||||
# Check Redis
|
||||
redis_check = await check_redis()
|
||||
if redis_check["status"] == "down":
|
||||
alerts.append(
|
||||
{
|
||||
"severity": "critical",
|
||||
"type": "redis_down",
|
||||
"message": f"Redis is down: {redis_check.get('error', 'Unknown error')}",
|
||||
}
|
||||
)
|
||||
|
||||
# Check Worker
|
||||
worker_check = await check_worker()
|
||||
if worker_check["status"] == "down":
|
||||
alerts.append(
|
||||
{
|
||||
"severity": "warning",
|
||||
"type": "worker_down",
|
||||
"message": f"Worker is down: {worker_check.get('error', 'Unknown error')}",
|
||||
}
|
||||
)
|
||||
|
||||
# Check Storage disk usage
|
||||
storage_stats = _get_storage_stats()
|
||||
disk_percent = storage_stats.get("disk_usage_percent", 0)
|
||||
if disk_percent > 90:
|
||||
alerts.append(
|
||||
{
|
||||
"severity": "critical",
|
||||
"type": "disk_full",
|
||||
"message": f"Disk usage at {disk_percent}%",
|
||||
}
|
||||
)
|
||||
|
||||
# Check high API error rate
|
||||
api_stats = _get_api_stats()
|
||||
error_rate = api_stats.get("error_rate", 0)
|
||||
if error_rate > 5:
|
||||
alerts.append(
|
||||
{
|
||||
"severity": "warning",
|
||||
"type": "high_error_rate",
|
||||
"message": f"High error rate: {error_rate:.1f}%",
|
||||
}
|
||||
)
|
||||
|
||||
# Send notifications for new alerts (throttled via Redis)
|
||||
if alerts:
|
||||
await _send_alert_notifications(db, admin, alerts)
|
||||
|
||||
return {
|
||||
"alerts": alerts,
|
||||
"alert_count": len(alerts),
|
||||
"timestamp": datetime.now(UTC).isoformat(),
|
||||
}
|
||||
Reference in New Issue
Block a user