Add pay business metrics

src/infrastructure/metrics.py

@@ -0,0 +1,155 @@
+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 = "pay"
+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.order import Order
+from src.infrastructure.database.models.payment import Payment
+
+
+async def _business_metrics() -> list[str]:
+    lines: list[str] = []
+    for model, metric_prefix in ((Order, "payment_orders"), (Payment, "payments")):
+        total = await _count(select(func.count()).select_from(model))
+        lines.append(_sample(f"{metric_prefix}_total", total, {"service": SERVICE_NAME}))
+        for status, count in await _rows(select(model.status, func.count()).group_by(model.status)):
+            lines.append(_sample(f"{metric_prefix}_by_status_total", int(count), {"service": SERVICE_NAME, "status": getattr(status, "value", str(status))}))
+        for window_name, delta in WINDOWS:
+            conditions = []
+            if delta is not None:
+                conditions.append(model.created_at >= func.now() - delta)
+            created = await _count(select(func.count()).select_from(model).where(*conditions))
+            lines.append(_sample(f"{metric_prefix}_created_total", created, {"service": SERVICE_NAME, "window": window_name}))
+
+    paid_total = await _count(select(func.count()).select_from(Payment).where(Payment.paid_at.is_not(None)))
+    failed_total = await _count(select(func.count()).select_from(Payment).where(Payment.status.in_(["failed", "cancelled", "expired"])))
+    lines.append(_sample("payments_succeeded_total", paid_total, {"service": SERVICE_NAME}))
+    lines.append(_sample("payments_failed_total", failed_total, {"service": SERVICE_NAME}))
+    return lines
+
+
+async def metrics_response() -> Response:
+    lines = [
+        "# TYPE payment_orders_total gauge",
+        "# TYPE payment_orders_by_status_total gauge",
+        "# TYPE payment_orders_created_total gauge",
+        "# TYPE payments_total gauge",
+        "# TYPE payments_by_status_total gauge",
+        "# TYPE payments_created_total gauge",
+        "# TYPE payments_succeeded_total gauge",
+        "# TYPE payments_failed_total gauge",
+    ]
+    try:
+        lines.extend(await _business_metrics())
+        lines.append(_sample("pay_metrics_scrape_success", 1, {"service": SERVICE_NAME}))
+    except Exception:
+        lines.append(_sample("pay_metrics_scrape_success", 0, {"service": SERVICE_NAME}))
+    lines.extend(_http_metrics("pay"))
+    return Response("\n".join(lines) + "\n", media_type="text/plain; version=0.0.4; charset=utf-8")

src/main.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 from contextlib import asynccontextmanager
 import secrets
 from typing import AsyncGenerator
-from fastapi import Depends, FastAPI
+from fastapi import Depends, FastAPI, Response
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.openapi.docs import get_redoc_html, get_swagger_ui_html
 from fastapi.responses import HTMLResponse
@@ -13,6 +13,7 @@ from src.infrastructure.config.settings import get_settings
 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.handler import application_exception_handler, unhandled_exception_handler
 from src.presentation.messaging import crypto_transfer_router,sbp_withdrawal_messaging_router
@@ -106,6 +107,8 @@ app: FastAPI = FastAPI(
 app.add_exception_handler(ApplicationException, application_exception_handler)
 app.add_exception_handler(Exception, unhandled_exception_handler)
 
+app.add_middleware(PrometheusMetricsMiddleware)
+
 app.include_router(order_router)
 app.include_router(orders_router)
 app.include_router(payment_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 {