T10: Monitoring, Performance, Documentation & Environment Config — 38 tests, ruff clean, docs OK
- Extended health endpoint: DB+Redis+Storage+Worker checks with degraded status - Prometheus metrics endpoint: admin-only, text/plain format - Metrics: http_requests_total, db_pool_connections, arq_jobs_total - Structured JSON logging (structlog): timestamp, level, method, path, status, duration_ms, tenant_id - Performance: page_size max 100 enforced (422), streaming CSV export (StreamingResponse) - Scripts: seed_perf_data.py, check_indexes.py - Docs: admin-guide.md (Deploy, Backup, Restore, Env-Vars, Troubleshooting), api-overview.md - README updated: prod setup, API section, env profiles, admin-guide link - .env.example: added SECRET_KEY, STORAGE_PATH, SMTP_* vars - 38 new tests, full regression 564/564 pass (0 failures) - Ruff: all checks passed
This commit is contained in:
leocrm-bot
@@ -0,0 +1,230 @@
|
||||
"""Monitoring: Prometheus metrics, structured JSON logging, health checks."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
import structlog
|
||||
from prometheus_client import (
|
||||
CollectorRegistry,
|
||||
Counter,
|
||||
Gauge,
|
||||
Histogram,
|
||||
generate_latest,
|
||||
)
|
||||
from sqlalchemy import text
|
||||
|
||||
from app.core.db import get_engine
|
||||
|
||||
# ─── Prometheus Metrics Registry ───
|
||||
|
||||
REGISTRY = CollectorRegistry()
|
||||
|
||||
http_requests_total = Counter(
|
||||
"leocrm_http_requests_total",
|
||||
"Total HTTP requests by method, path, and status",
|
||||
["method", "path", "status"],
|
||||
registry=REGISTRY,
|
||||
)
|
||||
|
||||
http_request_duration_seconds = Histogram(
|
||||
"leocrm_http_request_duration_seconds",
|
||||
"HTTP request duration in seconds",
|
||||
["method", "path"],
|
||||
registry=REGISTRY,
|
||||
buckets=(0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0),
|
||||
)
|
||||
|
||||
db_pool_connections = Gauge(
|
||||
"leocrm_db_pool_connections",
|
||||
"Database connection pool size",
|
||||
registry=REGISTRY,
|
||||
)
|
||||
|
||||
arq_jobs_total = Counter(
|
||||
"leocrm_arq_jobs_total",
|
||||
"Total ARQ background jobs by function and status",
|
||||
["function", "status"],
|
||||
registry=REGISTRY,
|
||||
)
|
||||
|
||||
|
||||
# ─── Structured Logging (structlog) ───
|
||||
|
||||
structlog.configure(
|
||||
processors=[
|
||||
structlog.contextvars.merge_contextvars,
|
||||
structlog.processors.add_log_level,
|
||||
structlog.processors.TimeStamper(fmt="iso"),
|
||||
structlog.processors.JSONRenderer(),
|
||||
],
|
||||
wrapper_class=structlog.make_filtering_bound_logger(20), # INFO level
|
||||
cache_logger_on_first_use=True,
|
||||
)
|
||||
|
||||
logger = structlog.get_logger("leocrm")
|
||||
|
||||
|
||||
def get_logger(name: str = "leocrm") -> structlog.stdlib.BoundLogger:
|
||||
"""Get a structured logger instance."""
|
||||
return structlog.get_logger(name)
|
||||
|
||||
|
||||
def record_request(
|
||||
method: str,
|
||||
path: str,
|
||||
status_code: int,
|
||||
duration_ms: float,
|
||||
tenant_id: str | None = None,
|
||||
) -> None:
|
||||
"""Record an API request in metrics and structured log."""
|
||||
http_requests_total.labels(method=method, path=path, status=str(status_code)).inc()
|
||||
http_request_duration_seconds.labels(method=method, path=path).observe(
|
||||
duration_ms / 1000.0
|
||||
)
|
||||
logger.info(
|
||||
"api_request",
|
||||
method=method,
|
||||
path=path,
|
||||
status=status_code,
|
||||
duration_ms=round(duration_ms, 2),
|
||||
tenant_id=tenant_id,
|
||||
)
|
||||
|
||||
|
||||
def record_error(
|
||||
event: str,
|
||||
method: str,
|
||||
path: str,
|
||||
status_code: int,
|
||||
error: str,
|
||||
traceback_str: str | None = None,
|
||||
tenant_id: str | None = None,
|
||||
) -> None:
|
||||
"""Log an error with stacktrace and request context."""
|
||||
logger.error(
|
||||
event=event,
|
||||
method=method,
|
||||
path=path,
|
||||
status=status_code,
|
||||
error=error,
|
||||
traceback=traceback_str,
|
||||
tenant_id=tenant_id,
|
||||
)
|
||||
|
||||
|
||||
def update_db_pool_gauge() -> None:
|
||||
"""Update the DB pool connections gauge from the current engine."""
|
||||
engine = get_engine()
|
||||
pool = engine.pool
|
||||
db_pool_connections.set(pool.size() + pool.checkedout())
|
||||
|
||||
|
||||
def record_arq_job(function: str, status: str) -> None:
|
||||
"""Record an ARQ background job completion."""
|
||||
arq_jobs_total.labels(function=function, status=status).inc()
|
||||
|
||||
|
||||
def generate_metrics() -> bytes:
|
||||
"""Generate Prometheus-formatted metrics output."""
|
||||
update_db_pool_gauge()
|
||||
return generate_latest(REGISTRY)
|
||||
|
||||
|
||||
# ─── Health Checks ───
|
||||
|
||||
|
||||
async def check_database() -> dict[str, Any]:
|
||||
"""Check database connectivity (async, no blocking I/O)."""
|
||||
try:
|
||||
engine = get_engine()
|
||||
async with engine.connect() as conn:
|
||||
result = await conn.execute(text("SELECT 1"))
|
||||
result.scalar()
|
||||
return {"status": "up", "latency_ms": 0}
|
||||
except Exception as e:
|
||||
return {"status": "down", "error": str(e)}
|
||||
|
||||
|
||||
async def check_redis() -> dict[str, Any]:
|
||||
"""Check Redis connectivity (async)."""
|
||||
try:
|
||||
import redis.asyncio as aioredis
|
||||
|
||||
from app.config import get_settings
|
||||
|
||||
settings = get_settings()
|
||||
r = aioredis.from_url(settings.redis_url, decode_responses=True)
|
||||
pong = await r.ping()
|
||||
await r.aclose()
|
||||
if pong:
|
||||
return {"status": "up", "latency_ms": 0}
|
||||
return {"status": "down", "error": "Redis returned False for PING"}
|
||||
except Exception as e:
|
||||
return {"status": "down", "error": str(e)}
|
||||
|
||||
|
||||
async def check_storage() -> dict[str, Any]:
|
||||
"""Check storage directory accessibility."""
|
||||
import os
|
||||
|
||||
from app.config import get_settings
|
||||
|
||||
settings = get_settings()
|
||||
storage_path = getattr(settings, "storage_path", "/tmp")
|
||||
try:
|
||||
if os.path.exists(storage_path) and os.access(storage_path, os.W_OK): # noqa: ASYNC240
|
||||
return {"status": "up", "path": storage_path}
|
||||
return {"status": "down", "error": f"Storage path {storage_path} not writable"}
|
||||
except Exception as e:
|
||||
return {"status": "down", "error": str(e)}
|
||||
|
||||
|
||||
async def check_worker() -> dict[str, Any]:
|
||||
"""Check if ARQ worker process is reachable (best-effort).
|
||||
In test/dev mode this checks if Redis is available for the worker queue.
|
||||
"""
|
||||
try:
|
||||
import redis.asyncio as aioredis
|
||||
|
||||
from app.config import get_settings
|
||||
|
||||
settings = get_settings()
|
||||
r = aioredis.from_url(settings.redis_url, decode_responses=True)
|
||||
# Check if arq queue key exists
|
||||
queue_length = await r.llen("arq:queue")
|
||||
await r.aclose()
|
||||
return {"status": "up", "queue_length": queue_length}
|
||||
except Exception as e:
|
||||
return {"status": "down", "error": str(e)}
|
||||
|
||||
|
||||
async def get_health_status() -> dict[str, Any]:
|
||||
"""Run all health checks and return aggregated status."""
|
||||
db_result = await check_database()
|
||||
redis_result = await check_redis()
|
||||
storage_result = await check_storage()
|
||||
worker_result = await check_worker()
|
||||
|
||||
all_up = all(
|
||||
c["status"] == "up"
|
||||
for c in [db_result, redis_result, storage_result, worker_result]
|
||||
)
|
||||
|
||||
if all_up:
|
||||
overall_status = "healthy"
|
||||
elif db_result["status"] == "down":
|
||||
overall_status = "degraded"
|
||||
else:
|
||||
overall_status = "degraded"
|
||||
|
||||
return {
|
||||
"status": overall_status,
|
||||
"version": "1.0.0",
|
||||
"checks": {
|
||||
"database": db_result,
|
||||
"redis": redis_result,
|
||||
"storage": storage_result,
|
||||
"worker": worker_result,
|
||||
},
|
||||
}
|
||||
Reference in New Issue
Block a user