Harden notify Prometheus metrics

Dockerfile

@@ -21,8 +21,9 @@ COPY --from=builder /app/src /app/src
 ENV PATH="/app/.venv/bin:$PATH" \
     PYTHONUNBUFFERED=1 \
     PYTHONDONTWRITEBYTECODE=1 \
-    PYTHONPATH=/app
+    PYTHONPATH=/app \
+    PROMETHEUS_MULTIPROC_DIR=/tmp/prometheus_multiproc
 
 EXPOSE 8000
 
-CMD ["sh", "-c", "granian --interface asgi ${APP_MODULE:-src.main:app} --host ${APP_HOST:-0.0.0.0} --port ${APP_PORT:-8000} --workers ${APP_WORKERS:-3} --loop uvloop"]
+CMD ["sh", "-c", "mkdir -p "$PROMETHEUS_MULTIPROC_DIR" && rm -f "$PROMETHEUS_MULTIPROC_DIR"/* && granian --interface asgi ${APP_MODULE:-src.main:app} --host ${APP_HOST:-0.0.0.0} --port ${APP_PORT:-8000} --workers ${APP_WORKERS:-1} --loop uvloop"]

pyproject.toml

@@ -17,4 +17,5 @@ dependencies = [
     "jinja2>=3.1.6",
     "aiosmtplib>=5.1.0",
     "aiohttp-socks>=0.11.0",
+    "prometheus-client==0.23.1",
 ]

src/infrastructure/metrics.py

@@ -1,46 +1,43 @@
 from __future__ import annotations
 
+import os
 import time
-from collections import defaultdict
+from typing import Any
-from typing import Iterable
 
 from fastapi import Request, Response
+from prometheus_client import CollectorRegistry, Counter, Histogram, CONTENT_TYPE_LATEST, generate_latest, multiprocess
 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_REQUESTS = Counter("notify_http_requests_total", "Total HTTP requests handled by notify service.", ("service", "method", "route", "status_code"))
-_http_latency_sum: dict[tuple[str, str], float] = defaultdict(float)
+HTTP_LATENCY = Histogram("notify_http_request_duration_seconds", "HTTP request duration in seconds for notify service.", ("service", "method", "route"), buckets=LATENCY_BUCKETS)
-_http_latency_count: dict[tuple[str, str], int] = defaultdict(int)
+NOTIFY_EVENTS = Counter("notify_events_total", "Notify events by channel, event and result.", ("service", "channel", "event", "result"))
-_http_latency_buckets: dict[tuple[str, str, float], int] = defaultdict(int)
+
-_notify_events: dict[tuple[str, str, str], int] = defaultdict(int)
+
+def _safe_label(value: Any) -> str:
+    text = str(value or "unknown").strip().lower().replace(" ", "_")
+    if not text or len(text) > 80:
+        return "unknown"
+    return text
 
 
 def observe_notify(channel: str, event: str, result: str) -> None:
-    _notify_events[(channel, event, result)] += 1
+    NOTIFY_EVENTS.labels(SERVICE_NAME, _safe_label(channel), _safe_label(event), _safe_label(result)).inc()
-
-
-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)
+    path = getattr(route, "path", None)
+    return str(path) if path else "__unmatched__"
 
 
 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:
@@ -50,36 +47,27 @@ class PrometheusMetricsMiddleware(BaseHTTPMiddleware):
         finally:
             elapsed = time.perf_counter() - started
             route = _route_template(request)
-            method = request.method
+            HTTP_REQUESTS.labels(SERVICE_NAME, request.method, route, status_code).inc()
-            _http_requests[(method, route, status_code)] += 1
+            HTTP_LATENCY.labels(SERVICE_NAME, request.method, route).observe(elapsed)
-            _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]:
+def _labels(labels: dict[str, str]) -> str:
-    yield "# TYPE notify_http_requests_total counter"
+    escaped = {key: value.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n") for key, value in labels.items()}
-    for (method, route, status_code), value in sorted(_http_requests.items()):
+    return ",".join(f'{key}="{value}"' for key, value in escaped.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):
+def _sample(name: str, value: int | float, labels: dict[str, str] | None = None) -> str:
-        for bucket in LATENCY_BUCKETS:
+    return f"{name}{{{_labels(labels)}}} {value}" if labels else f"{name} {value}"
-            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})
+def _client_metrics() -> bytes:
-        yield _sample("notify_http_request_duration_seconds_count", _http_latency_count[(method, route)], {"service": SERVICE_NAME, "method": method, "route": route})
+    if os.environ.get("PROMETHEUS_MULTIPROC_DIR"):
+        registry = CollectorRegistry()
+        multiprocess.MultiProcessCollector(registry)
+        return generate_latest(registry)
+    return generate_latest()
 
 
 async def metrics_response() -> Response:
-    lines = [
+    body = _client_metrics().decode("utf-8") + _sample("notify_metrics_scrape_success", 1, {"service": SERVICE_NAME}) + "\n"
-        "# TYPE notify_events_total counter",
+    return Response(body, media_type=CONTENT_TYPE_LATEST)
-        "# 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")

uv.lock

@@ -765,6 +765,7 @@ dependencies = [
     { name = "granian" },
     { name = "hvac" },
     { name = "jinja2" },
+    { name = "prometheus-client" },
     { name = "pydantic-settings" },
     { name = "python-ulid" },
     { name = "uvloop", marker = "sys_platform != 'win32'" },
@@ -782,6 +783,7 @@ requires-dist = [
     { name = "granian", specifier = ">=2.6.1" },
     { name = "hvac", specifier = ">=2.4.0" },
     { name = "jinja2", specifier = ">=3.1.6" },
+    { name = "prometheus-client", specifier = "==0.23.1" },
     { name = "pydantic-settings", specifier = ">=2.13.1" },
     { name = "python-ulid", specifier = ">=3.1.0" },
     { name = "uvloop", marker = "sys_platform != 'win32'", specifier = ">=0.22.1" },
@@ -796,6 +798,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/ac/8d/c1e93296e109a320e508e38118cf7d1fc2a4d1c2ec64de78565b3c445eb5/pamqp-3.3.0-py2.py3-none-any.whl", hash = "sha256:c901a684794157ae39b52cbf700db8c9aae7a470f13528b9d7b4e5f7202f8eb0", size = 33848, upload-time = "2024-01-12T20:37:21.359Z" },
 ]
 
+[[package]]
+name = "prometheus-client"
+version = "0.23.1"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/23/53/3edb5d68ecf6b38fcbcc1ad28391117d2a322d9a1a3eff04bfdb184d8c3b/prometheus_client-0.23.1.tar.gz", hash = "sha256:6ae8f9081eaaaf153a2e959d2e6c4f4fb57b12ef76c8c7980202f1e57b48b2ce", size = 80481, upload-time = "2025-09-18T20:47:25.043Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/b8/db/14bafcb4af2139e046d03fd00dea7873e48eafe18b7d2797e73d6681f210/prometheus_client-0.23.1-py3-none-any.whl", hash = "sha256:dd1913e6e76b59cfe44e7a4b83e01afc9873c1bdfd2ed8739f1e76aeca115f99", size = 61145, upload-time = "2025-09-18T20:47:23.875Z" },
+]
+
 [[package]]
 name = "propcache"
 version = "0.4.1"