From 910fc8d37f4eab8e7663e1ba421d557065b43598 Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 1 Jul 2026 18:47:26 +0300 Subject: [PATCH] Add notify business metrics --- src/infrastructure/metrics.py | 85 +++++++++++++++++++++++++ src/main.py | 10 ++- src/presentation/messaging/code.py | 21 ++++-- src/presentation/messaging/consumers.py | 11 +++- 4 files changed, 117 insertions(+), 10 deletions(-) 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..b23c0a4 --- /dev/null +++ b/src/infrastructure/metrics.py @@ -0,0 +1,85 @@ +from __future__ import annotations + +import time +from collections import defaultdict +from typing import Iterable + +from fastapi import Request, Response +from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint +from starlette.responses import Response as StarletteResponse + + +SERVICE_NAME = "notify" +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) +_notify_events: dict[tuple[str, str, str], int] = defaultdict(int) + + +def observe_notify(channel: str, event: str, result: str) -> None: + _notify_events[(channel, event, result)] += 1 + + +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 + + +def _http_metrics() -> Iterable[str]: + yield "# TYPE notify_http_requests_total counter" + for (method, route, status_code), value in sorted(_http_requests.items()): + yield _sample("notify_http_requests_total", value, {"service": SERVICE_NAME, "method": method, "route": route, "status_code": status_code}) + + yield "# TYPE notify_http_request_duration_seconds histogram" + for method, route in sorted(_http_latency_count): + for bucket in LATENCY_BUCKETS: + yield _sample("notify_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("notify_http_request_duration_seconds_bucket", _http_latency_count[(method, route)], {"service": SERVICE_NAME, "method": method, "route": route, "le": "+Inf"}) + yield _sample("notify_http_request_duration_seconds_sum", _http_latency_sum[(method, route)], {"service": SERVICE_NAME, "method": method, "route": route}) + yield _sample("notify_http_request_duration_seconds_count", _http_latency_count[(method, route)], {"service": SERVICE_NAME, "method": method, "route": route}) + + +async def metrics_response() -> Response: + lines = [ + "# TYPE notify_events_total counter", + "# TYPE notify_http_requests_total counter", + ] + for (channel, event, result), value in sorted(_notify_events.items()): + lines.append(_sample("notify_events_total", value, {"service": SERVICE_NAME, "channel": channel, "event": event, "result": result})) + lines.append(_sample("notify_metrics_scrape_success", 1, {"service": SERVICE_NAME})) + lines.extend(_http_metrics()) + 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 fa43d94..35550c8 100644 --- a/src/main.py +++ b/src/main.py @@ -2,13 +2,14 @@ from __future__ import annotations from contextlib import asynccontextmanager import secrets from typing import AsyncGenerator -from fastapi import Depends, FastAPI, status, APIRouter +from fastapi import Depends, FastAPI, Response, status, APIRouter from fastapi.openapi.docs import get_redoc_html, get_swagger_ui_html from fastapi.responses import HTMLResponse from fastapi.security import HTTPBasic, HTTPBasicCredentials from src.application.domain.exceptions import ApplicationException 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 from src.presentation.messaging.code import code_broker @@ -57,6 +58,8 @@ app: FastAPI = FastAPI( app.add_exception_handler(ApplicationException, application_exception_handler) app.add_exception_handler(Exception, unhandled_exception_handler) +app.add_middleware(PrometheusMetricsMiddleware) + v1_router = APIRouter(prefix='/v1') app.include_router(v1_router) app.include_router(code_broker) @@ -97,6 +100,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 { diff --git a/src/presentation/messaging/code.py b/src/presentation/messaging/code.py index aadf8ac..f9b170c 100644 --- a/src/presentation/messaging/code.py +++ b/src/presentation/messaging/code.py @@ -9,6 +9,7 @@ from datetime import datetime from typing import Literal, Optional from pydantic import BaseModel, EmailStr, Field from src.infrastructure.rabbit.broker import email_code_queue +from src.infrastructure.metrics import observe_notify code_broker = RabbitRouter(settings.RABBIT_URL) @@ -76,10 +77,16 @@ async def consume_email_code( year=datetime.now().year, ) - await sender.send( - to=msg_body.payload.email, - subject='Экса — код подтверждения', - body=html, - plain=text, - inline_png=(EXA_LOGO_CID, logo_png) if logo_png else None, - ) \ No newline at end of file + observe_notify('email', msg_body.event, 'received') + try: + await sender.send( + to=msg_body.payload.email, + subject='Экса — код подтверждения', + body=html, + plain=text, + inline_png=(EXA_LOGO_CID, logo_png) if logo_png else None, + ) + except Exception: + observe_notify('email', msg_body.event, 'error') + raise + observe_notify('email', msg_body.event, 'sent') \ No newline at end of file diff --git a/src/presentation/messaging/consumers.py b/src/presentation/messaging/consumers.py index 158c21c..17a6428 100644 --- a/src/presentation/messaging/consumers.py +++ b/src/presentation/messaging/consumers.py @@ -10,6 +10,7 @@ from datetime import datetime from typing import Literal, Optional from pydantic import BaseModel from src.infrastructure.rabbit.broker import telegram_notify_queue +from src.infrastructure.metrics import observe_notify notify_broker = RabbitRouter(settings.RABBIT_URL) @@ -64,5 +65,11 @@ async def consume_notify( trace_id=trace_id ) - async with ClientSession() as session: - await sender.send(client=session, text=text) \ No newline at end of file + observe_notify('telegram', msg_body.event, 'received') + try: + async with ClientSession() as session: + await sender.send(client=session, text=text) + except Exception: + observe_notify('telegram', msg_body.event, 'error') + raise + observe_notify('telegram', msg_body.event, 'sent') \ No newline at end of file