| """Observability (P1): Prometheus metrics, structured JSON logs, request IDs, Sentry. |
| |
| - ``/metrics`` exposes Prometheus counters/histograms (request rate, latency, and |
| domain signals — payment decisions, auth outcomes, step-ups). |
| - ``metrics_middleware`` times every request, records it, and stamps an X-Request-ID. |
| - ``setup_logging`` emits one JSON object per log line (ingestable by Loki/ELK/Datadog). |
| - ``init_sentry`` wires error tracking when ``SENTRY_DSN`` is set. |
| All degrade gracefully if a dependency is missing. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import json |
| import logging |
| import os |
| import time |
| import uuid |
| from typing import Callable |
|
|
| try: |
| from prometheus_client import Counter, Histogram, CONTENT_TYPE_LATEST, generate_latest |
| _HTTP = Counter("amanpay_http_requests_total", "HTTP requests", |
| ["method", "path", "status"]) |
| _LAT = Histogram("amanpay_http_request_seconds", "HTTP request latency", |
| ["method", "path"]) |
| PAYMENTS = Counter("amanpay_payments_total", "Payment outcomes", ["decision"]) |
| AUTHN = Counter("amanpay_auth_total", "Auth outcomes", ["outcome"]) |
| KV_FAIL = Counter("amanpay_kv_failures_total", "KV-store operation failures", ["op"]) |
| READY = Counter("amanpay_readiness_checks_total", "Readiness checks", ["dep", "ok"]) |
| _PROM = True |
| except Exception: |
| _PROM = False |
|
|
|
|
| def _route(request) -> str: |
| """Templated path (avoids high-cardinality metric labels).""" |
| r = request.scope.get("route") |
| return getattr(r, "path", request.url.path) if r else request.url.path |
|
|
|
|
| async def metrics_middleware(request, call_next: Callable): |
| rid = request.headers.get("x-request-id") or uuid.uuid4().hex[:16] |
| start = time.time() |
| try: |
| response = await call_next(request) |
| status = response.status_code |
| except Exception: |
| status = 500 |
| raise |
| finally: |
| if _PROM: |
| path = _route(request) |
| _HTTP.labels(request.method, path, str(status)).inc() |
| _LAT.labels(request.method, path).observe(time.time() - start) |
| response.headers["X-Request-ID"] = rid |
| return response |
|
|
|
|
| def metrics_response(): |
| from fastapi import Response |
| if not _PROM: |
| return Response("prometheus_client not installed", status_code=501) |
| return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST) |
|
|
|
|
| def record_payment(decision: str) -> None: |
| if _PROM: |
| PAYMENTS.labels(decision or "unknown").inc() |
|
|
|
|
| def record_auth(outcome: str) -> None: |
| if _PROM: |
| AUTHN.labels(outcome).inc() |
|
|
|
|
| def record_kv_failure(op: str) -> None: |
| if _PROM: |
| KV_FAIL.labels(op).inc() |
|
|
|
|
| def record_readiness(dep: str, ok: bool) -> None: |
| if _PROM: |
| READY.labels(dep, "true" if ok else "false").inc() |
|
|
|
|
| class _JsonFormatter(logging.Formatter): |
| def format(self, record: logging.LogRecord) -> str: |
| obj = {"ts": round(record.created, 3), "level": record.levelname, |
| "logger": record.name, "msg": record.getMessage()} |
| if record.exc_info: |
| obj["exc"] = self.formatException(record.exc_info) |
| return json.dumps(obj) |
|
|
|
|
| def setup_logging() -> None: |
| """JSON logs when AMANPAY_JSON_LOGS=1 (default on in containers).""" |
| if os.getenv("AMANPAY_JSON_LOGS", "1").lower() in ("0", "false", "no"): |
| return |
| handler = logging.StreamHandler() |
| handler.setFormatter(_JsonFormatter()) |
| root = logging.getLogger() |
| root.handlers[:] = [handler] |
| root.setLevel(logging.INFO) |
|
|
|
|
| def init_sentry() -> None: |
| dsn = os.getenv("SENTRY_DSN") |
| if not dsn: |
| return |
| try: |
| import sentry_sdk |
| sentry_sdk.init(dsn=dsn, traces_sample_rate=float(os.getenv("SENTRY_TRACES", "0.1"))) |
| logging.getLogger("amanpay").info("Sentry error tracking enabled") |
| except Exception as exc: |
| logging.getLogger("amanpay").info("Sentry unavailable (%s)", exc) |
|
|
