Add kyc business metrics

src/infrastructure/metrics.py

@@ -0,0 +1,158 @@
+from __future__ import annotations
+
+import time
+from collections import defaultdict
+from datetime import timedelta
+from typing import Iterable
+
+from fastapi import Request, Response
+from sqlalchemy import func, select
+from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
+from starlette.responses import Response as StarletteResponse
+
+from src.infrastructure.database.context import async_session_maker
+
+
+SERVICE_NAME = "kyc"
+WINDOWS: tuple[tuple[str, timedelta | None], ...] = (
+    ("all", None),
+    ("30d", timedelta(days=30)),
+    ("7d", timedelta(days=7)),
+    ("24h", timedelta(days=1)),
+    ("1h", timedelta(hours=1)),
+)
+LATENCY_BUCKETS = (0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0)
+
+_http_requests: dict[tuple[str, str, str], int] = defaultdict(int)
+_http_latency_sum: dict[tuple[str, str], float] = defaultdict(float)
+_http_latency_count: dict[tuple[str, str], int] = defaultdict(int)
+_http_latency_buckets: dict[tuple[str, str, float], int] = defaultdict(int)
+
+
+def _labels(labels: dict[str, str]) -> str:
+    return ",".join(f'{key}="{value}"' for key, value in labels.items())
+
+
+def _sample(name: str, value: int | float, labels: dict[str, str] | None = None) -> str:
+    return f"{name}{{{_labels(labels)}}} {value}" if labels else f"{name} {value}"
+
+
+def _route_template(request: Request) -> str:
+    route = request.scope.get("route")
+    return str(getattr(route, "path", None) or request.url.path)
+
+
+class PrometheusMetricsMiddleware(BaseHTTPMiddleware):
+    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> StarletteResponse:
+        if request.url.path == "/metrics":
+            return await call_next(request)
+
+        started = time.perf_counter()
+        status_code = "500"
+        try:
+            response = await call_next(request)
+            status_code = str(response.status_code)
+            return response
+        finally:
+            elapsed = time.perf_counter() - started
+            route = _route_template(request)
+            method = request.method
+            _http_requests[(method, route, status_code)] += 1
+            _http_latency_sum[(method, route)] += elapsed
+            _http_latency_count[(method, route)] += 1
+            for bucket in LATENCY_BUCKETS:
+                if elapsed <= bucket:
+                    _http_latency_buckets[(method, route, bucket)] += 1
+
+
+async def _count(statement) -> int:
+    async with async_session_maker() as session:
+        result = await session.execute(statement)
+        return int(result.scalar_one() or 0)
+
+
+async def _rows(statement):
+    async with async_session_maker() as session:
+        result = await session.execute(statement)
+        return result.all()
+
+
+def _http_metrics(prefix: str) -> Iterable[str]:
+    yield f"# TYPE {prefix}_http_requests_total counter"
+    for (method, route, status_code), value in sorted(_http_requests.items()):
+        yield _sample(
+            f"{prefix}_http_requests_total",
+            value,
+            {"service": SERVICE_NAME, "method": method, "route": route, "status_code": status_code},
+        )
+
+    yield f"# TYPE {prefix}_http_request_duration_seconds histogram"
+    for method, route in sorted(_http_latency_count):
+        for bucket in LATENCY_BUCKETS:
+            yield _sample(
+                f"{prefix}_http_request_duration_seconds_bucket",
+                _http_latency_buckets.get((method, route, bucket), 0),
+                {"service": SERVICE_NAME, "method": method, "route": route, "le": str(bucket)},
+            )
+        yield _sample(
+            f"{prefix}_http_request_duration_seconds_bucket",
+            _http_latency_count[(method, route)],
+            {"service": SERVICE_NAME, "method": method, "route": route, "le": "+Inf"},
+        )
+        yield _sample(
+            f"{prefix}_http_request_duration_seconds_sum",
+            _http_latency_sum[(method, route)],
+            {"service": SERVICE_NAME, "method": method, "route": route},
+        )
+        yield _sample(
+            f"{prefix}_http_request_duration_seconds_count",
+            _http_latency_count[(method, route)],
+            {"service": SERVICE_NAME, "method": method, "route": route},
+        )
+
+from src.infrastructure.database.models.kyc import KycModel
+
+
+async def _business_metrics() -> list[str]:
+    lines: list[str] = []
+    total = await _count(select(func.count()).select_from(KycModel))
+    lines.append(_sample("kyc_requests_total", total, {"service": SERVICE_NAME}))
+    for status, count in await _rows(select(KycModel.status, func.count()).group_by(KycModel.status)):
+        status_value = str(status)
+        lines.append(_sample("kyc_requests_by_status_total", int(count), {"service": SERVICE_NAME, "status": status_value}))
+    approved = await _count(select(func.count()).select_from(KycModel).where(KycModel.done_state.is_(True)))
+    failed = await _count(select(func.count()).select_from(KycModel).where(KycModel.done_state.is_(False)))
+    lines.append(_sample("kyc_approvals_total", approved, {"service": SERVICE_NAME}))
+    lines.append(_sample("kyc_fails_total", failed, {"service": SERVICE_NAME}))
+    for window_name, delta in WINDOWS:
+        conditions = []
+        completed_conditions = []
+        if delta is not None:
+            conditions.append(KycModel.created_at >= func.now() - delta)
+            completed_conditions.append(KycModel.completed_at >= func.now() - delta)
+        created = await _count(select(func.count()).select_from(KycModel).where(*conditions))
+        approvals = await _count(select(func.count()).select_from(KycModel).where(KycModel.done_state.is_(True), *completed_conditions))
+        fails = await _count(select(func.count()).select_from(KycModel).where(KycModel.done_state.is_(False), *completed_conditions))
+        lines.append(_sample("kyc_requests_created_total", created, {"service": SERVICE_NAME, "window": window_name}))
+        lines.append(_sample("kyc_approvals_window_total", approvals, {"service": SERVICE_NAME, "window": window_name}))
+        lines.append(_sample("kyc_fails_window_total", fails, {"service": SERVICE_NAME, "window": window_name}))
+    return lines
+
+
+async def metrics_response() -> Response:
+    lines = [
+        "# TYPE kyc_requests_total gauge",
+        "# TYPE kyc_requests_by_status_total gauge",
+        "# TYPE kyc_requests_created_total gauge",
+        "# TYPE kyc_approvals_total gauge",
+        "# TYPE kyc_fails_total gauge",
+        "# TYPE kyc_approvals_window_total gauge",
+        "# TYPE kyc_fails_window_total gauge",
+    ]
+    try:
+        lines.extend(await _business_metrics())
+        lines.append(_sample("kyc_metrics_scrape_success", 1, {"service": SERVICE_NAME}))
+    except Exception:
+        lines.append(_sample("kyc_metrics_scrape_success", 0, {"service": SERVICE_NAME}))
+    lines.extend(_http_metrics("kyc"))
+    return Response("\n".join(lines) + "\n", media_type="text/plain; version=0.0.4; charset=utf-8")

src/main.py

@@ -19,6 +19,7 @@ from src.infrastructure.database.unit_of_work import UnitOfWork
 from src.infrastructure.vault import JwtKeyStore, start_jwt_keys_scheduler
 from src.infrastructure.utils import generate_instance_id
 from src.infrastructure.logger import logger
+from src.infrastructure.metrics import PrometheusMetricsMiddleware, metrics_response
 from src.infrastructure.config import settings
 from src.presentation.handlers import application_exception_handler,unhandled_exception_handler,validation_exception_handler
 from src.presentation.middleware import TraceIDMiddleware, SecurityHeadersMiddleware
@@ -103,6 +104,8 @@ app.add_exception_handler(ApplicationException, application_exception_handler)
 app.add_exception_handler(RequestValidationError, validation_exception_handler)
 app.add_exception_handler(Exception, unhandled_exception_handler)
 
+app.add_middleware(PrometheusMetricsMiddleware)
+
 app.include_router(kyc_router)
 
 
@@ -157,6 +160,11 @@ async def custom_redoc_html(_credentials: HTTPBasicCredentials = Depends(verify_
     )
 
 
+@app.get('/metrics', include_in_schema=False)
+async def metrics() -> Response:
+    return await metrics_response()
+
+
 @app.post('/ping')
 async def ping() -> dict[str, str]:
     return {