From aa01e67a091d2a162a00a6421c676fd75e05d332 Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 1 Jul 2026 18:46:14 +0300 Subject: [PATCH] Add kyc business metrics --- src/infrastructure/metrics.py | 158 ++++++++++++++++++++++++++++++++++ src/main.py | 8 ++ 2 files changed, 166 insertions(+) create mode 100644 src/infrastructure/metrics.py diff --git a/src/infrastructure/metrics.py b/src/infrastructure/metrics.py new file mode 100644 index 0000000..3321508 --- /dev/null +++ b/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") diff --git a/src/main.py b/src/main.py index a5703af..aca1a4f 100644 --- a/src/main.py +++ b/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 {