Add notify business metrics

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")

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 {

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(
+    observe_notify('email', msg_body.event, 'received')
-        to=msg_body.payload.email,
+    try:
-        subject='Экса — код подтверждения',
+        await sender.send(
-        body=html,
+            to=msg_body.payload.email,
-        plain=text,
+            subject='Экса — код подтверждения',
-        inline_png=(EXA_LOGO_CID, logo_png) if logo_png else None,
+            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')

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:
+    observe_notify('telegram', msg_body.event, 'received')
-        await sender.send(client=session, text=text)
+    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')