114 lines
3.7 KiB
Python
114 lines
3.7 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import time
|
|
import uuid
|
|
from contextvars import ContextVar
|
|
from datetime import UTC, datetime
|
|
|
|
from fastapi import Request
|
|
from prometheus_client import Counter, Gauge, Histogram
|
|
|
|
correlation_id_context: ContextVar[str] = ContextVar("correlation_id", default="")
|
|
|
|
HTTP_REQUESTS = Counter(
|
|
"mobilityops_http_requests_total",
|
|
"Completed MobilityOps HTTP requests.",
|
|
("method", "route", "status"),
|
|
)
|
|
HTTP_DURATION = Histogram(
|
|
"mobilityops_http_request_duration_seconds",
|
|
"MobilityOps HTTP request duration.",
|
|
("method", "route"),
|
|
buckets=(0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10),
|
|
)
|
|
HTTP_IN_PROGRESS = Gauge(
|
|
"mobilityops_http_requests_in_progress",
|
|
"MobilityOps HTTP requests currently executing.",
|
|
)
|
|
OUTBOX_EVENTS = Gauge(
|
|
"mobilityops_outbox_events",
|
|
"Persisted outbox events by state and scenario type.",
|
|
("scenario", "status"),
|
|
)
|
|
DATABASE_READY = Gauge(
|
|
"mobilityops_database_ready",
|
|
"Whether the canonical PostgreSQL database answered the most recent readiness probe.",
|
|
)
|
|
KNOWLEDGE_PROVIDER_REQUESTS = Counter(
|
|
"mobilityops_knowledge_provider_requests_total",
|
|
"RAGcore adapter requests by stage and outcome.",
|
|
("stage", "outcome"),
|
|
)
|
|
KNOWLEDGE_RETRIEVAL_SCORE = Histogram(
|
|
"mobilityops_knowledge_retrieval_score",
|
|
"Observed RAGcore fused/rerank retrieval scores.",
|
|
buckets=(0.005, 0.01, 0.015, 0.016, 0.0162, 0.0164, 0.02, 0.05, 0.1, 0.5, 1.0),
|
|
)
|
|
|
|
|
|
class JsonFormatter(logging.Formatter):
|
|
def format(self, record: logging.LogRecord) -> str:
|
|
payload: dict[str, object] = {
|
|
"timestamp": datetime.now(UTC).isoformat(),
|
|
"level": record.levelname.lower(),
|
|
"logger": record.name,
|
|
"message": record.getMessage(),
|
|
}
|
|
correlation_id = correlation_id_context.get()
|
|
if correlation_id:
|
|
payload["correlation_id"] = correlation_id
|
|
for key in ("method", "path", "status_code", "duration_ms", "client_ip"):
|
|
value = getattr(record, key, None)
|
|
if value is not None:
|
|
payload[key] = value
|
|
if record.exc_info:
|
|
payload["exception"] = self.formatException(record.exc_info)
|
|
return json.dumps(payload, separators=(",", ":"), default=str)
|
|
|
|
|
|
def configure_logging(level: str) -> None:
|
|
handler = logging.StreamHandler()
|
|
handler.setFormatter(JsonFormatter())
|
|
root = logging.getLogger()
|
|
root.handlers = [handler]
|
|
root.setLevel(level.upper())
|
|
|
|
|
|
def correlation_id_for(request: Request) -> str:
|
|
candidate = request.headers.get("X-Correlation-Id", "").strip()
|
|
try:
|
|
return str(uuid.UUID(candidate)) if candidate else str(uuid.uuid4())
|
|
except ValueError:
|
|
return str(uuid.uuid4())
|
|
|
|
|
|
UNMATCHED_ROUTE_LABEL = "<unmatched>"
|
|
|
|
|
|
def route_label(request: Request) -> str:
|
|
"""Return the route *template* for metrics labels.
|
|
|
|
Unmatched paths (404 probes, scanners) must not become their own label value:
|
|
every distinct URL would otherwise create a new Prometheus time series and the
|
|
metric cardinality would grow without bound.
|
|
"""
|
|
route = request.scope.get("route")
|
|
path = getattr(route, "path", None)
|
|
return str(path) if path else UNMATCHED_ROUTE_LABEL
|
|
|
|
|
|
def request_started() -> float:
|
|
HTTP_IN_PROGRESS.inc()
|
|
return time.perf_counter()
|
|
|
|
|
|
def request_finished(request: Request, status_code: int, started_at: float) -> float:
|
|
duration = time.perf_counter() - started_at
|
|
route = route_label(request)
|
|
HTTP_REQUESTS.labels(request.method, route, str(status_code)).inc()
|
|
HTTP_DURATION.labels(request.method, route).observe(duration)
|
|
HTTP_IN_PROGRESS.dec()
|
|
return duration
|