007 add telemetry

2026-03-08 11:43:15 +03:00
parent 150675cc75
commit 8d47917ba2
15 changed files with 952 additions and 62 deletions
@@ -46,7 +46,7 @@ class AdapterConfig(BaseModel):

 class ObservabilityConfig(BaseModel):
    service_name: str = "g2s-aggregator"
-    otlp_endpoint: str = "http://localhost:4317"
+    otlp_endpoint: str = ""
    log_level: str = "INFO"


@@ -1,10 +1,55 @@
 """HTTP middleware registration."""

-from fastapi import FastAPI
+from uuid import uuid4
+from time import perf_counter
+
+from fastapi import FastAPI, Request
+from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.responses import Response
+from structlog.contextvars import bind_contextvars, clear_contextvars
+
+from app.observability.metrics import get_metrics
+
+
+class RequestCorrelationMiddleware(BaseHTTPMiddleware):
+    """Assign and propagate request correlation identifiers."""
+
+    def __init__(self, app: FastAPI, *, request_id_header: str) -> None:
+        super().__init__(app)
+        self._request_id_header = request_id_header
+
+    async def dispatch(self, request: Request, call_next) -> Response:
+        request_id = str(uuid4())
+        request.state.request_id = request_id
+        bind_contextvars(request_id=request_id)
+
+        started_at = perf_counter()
+        status_code = 500
+        is_error = True
+
+        try:
+            response = await call_next(request)
+            status_code = response.status_code
+            is_error = status_code >= 500
+            response.headers[self._request_id_header] = request_id
+            return response
+        finally:
+            duration_ms = (perf_counter() - started_at) * 1000
+            get_metrics().record_http_request(
+                method=request.method,
+                route=request.url.path,
+                status_code=status_code,
+                duration_ms=duration_ms,
+                is_error=is_error,
+            )
+            clear_contextvars()


 def install_middleware(app: FastAPI) -> None:
    """Register middleware components for the API."""

-    # Middleware stack is introduced in later tasks.
-    _ = app
+    settings = app.state.settings
+    app.add_middleware(
+        RequestCorrelationMiddleware,
+        request_id_header=settings.controller.request_id_header,
+    )
@@ -5,6 +5,7 @@ from fastapi import FastAPI
 from app.config import Settings, get_settings
 from app.controllers.middleware import install_middleware
 from app.controllers.v1.delivery import router as delivery_router
+from app.observability import setup_observability


 def create_app(settings: Settings | None = None) -> FastAPI:
@@ -13,6 +14,7 @@ def create_app(settings: Settings | None = None) -> FastAPI:
    application = FastAPI(title="G2S Aggregator", version="0.1.0")
    application.state.settings = resolved_settings

+    setup_observability(application, resolved_settings.observability)
    install_middleware(application)
    application.include_router(delivery_router, prefix=resolved_settings.controller.api_prefix)
    return application
@@ -0,0 +1,5 @@
+"""Observability setup and helpers."""
+
+from app.observability.setup import setup_observability
+
+__all__ = ["setup_observability"]
@@ -0,0 +1,48 @@
+"""Structured logging setup with request and trace correlation."""
+
+import logging
+from typing import Any
+
+import structlog
+from opentelemetry import trace
+from opentelemetry.trace import SpanContext
+
+
+def configure_structured_logging(log_level: str) -> None:
+    resolved_level = _resolve_log_level(log_level)
+    logging.basicConfig(level=resolved_level, format="%(message)s")
+
+    structlog.configure(
+        processors=[
+            structlog.contextvars.merge_contextvars,
+            structlog.stdlib.add_log_level,
+            add_trace_context,
+            structlog.processors.TimeStamper(fmt="iso", utc=True),
+            structlog.processors.JSONRenderer(),
+        ],
+        logger_factory=structlog.stdlib.LoggerFactory(),
+        wrapper_class=structlog.make_filtering_bound_logger(resolved_level),
+        cache_logger_on_first_use=True,
+    )
+
+
+def add_trace_context(
+    _: Any,
+    __: str,
+    event_dict: dict[str, Any],
+) -> dict[str, Any]:
+    span = trace.get_current_span()
+    span_context = span.get_span_context()
+    if _is_valid_span_context(span_context):
+        event_dict["trace_id"] = format(span_context.trace_id, "032x")
+        event_dict["span_id"] = format(span_context.span_id, "016x")
+    return event_dict
+
+
+def _resolve_log_level(log_level: str) -> int:
+    candidate = log_level.strip().upper()
+    return logging.getLevelNamesMapping().get(candidate, logging.INFO)
+
+
+def _is_valid_span_context(span_context: SpanContext) -> bool:
+    return bool(span_context.is_valid and span_context.trace_id and span_context.span_id)
@@ -0,0 +1,129 @@
+"""OpenTelemetry metrics setup and recording helpers."""
+
+from typing import Sequence
+
+from opentelemetry import metrics
+from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter
+from opentelemetry.sdk.metrics import MeterProvider
+from opentelemetry.sdk.metrics.export import (
+    MetricReader,
+    PeriodicExportingMetricReader,
+)
+from opentelemetry.sdk.resources import Resource
+
+_METER_PROVIDER: MeterProvider | None = None
+_METRICS: "ObservabilityMetrics" | None = None
+_METER_NAME = "g2s.observability"
+
+
+class ObservabilityMetrics:
+    def __init__(
+        self,
+        meter_name: str,
+        *,
+        meter_provider: MeterProvider | None = None,
+    ) -> None:
+        if meter_provider is None:
+            meter = metrics.get_meter(meter_name)
+        else:
+            meter = meter_provider.get_meter(meter_name)
+        self._request_count = meter.create_counter(
+            name="http.server.request.count",
+            unit="1",
+            description="Total number of HTTP requests.",
+        )
+        self._error_count = meter.create_counter(
+            name="http.server.error.count",
+            unit="1",
+            description="Total number of HTTP server errors.",
+        )
+        self._request_latency = meter.create_histogram(
+            name="http.server.request.latency",
+            unit="ms",
+            description="HTTP request latency.",
+        )
+        self._provider_availability = meter.create_histogram(
+            name="delivery.provider.availability",
+            unit="1",
+            description="Provider availability sample (1=available, 0=unavailable).",
+        )
+
+    def record_http_request(
+        self,
+        *,
+        method: str,
+        route: str,
+        status_code: int,
+        duration_ms: float,
+        is_error: bool,
+    ) -> None:
+        attributes = {
+            "http.method": method,
+            "http.route": route,
+            "http.status_code": status_code,
+        }
+        self._request_count.add(1, attributes)
+        self._request_latency.record(duration_ms, attributes)
+        if is_error:
+            self._error_count.add(1, attributes)
+
+    def record_provider_availability(
+        self,
+        *,
+        provider: str,
+        is_available: bool,
+    ) -> None:
+        self._provider_availability.record(
+            1.0 if is_available else 0.0,
+            {"provider": provider},
+        )
+
+
+def setup_metrics(
+    *,
+    service_name: str,
+    otlp_endpoint: str,
+    meter_provider: MeterProvider | None = None,
+    metric_readers: Sequence[MetricReader] | None = None,
+) -> MeterProvider:
+    global _METER_PROVIDER, _METRICS
+
+    if meter_provider is None and _METER_PROVIDER is not None:
+        return _METER_PROVIDER
+
+    if meter_provider is None:
+        readers = list(metric_readers or ())
+        if otlp_endpoint:
+            metric_exporter = OTLPMetricExporter(
+                endpoint=otlp_endpoint,
+                insecure=otlp_endpoint.startswith("http://"),
+            )
+            readers.append(PeriodicExportingMetricReader(metric_exporter))
+        meter_provider = MeterProvider(
+            resource=Resource.create({"service.name": service_name}),
+            metric_readers=readers,
+        )
+
+    _METER_PROVIDER = meter_provider
+    _METRICS = ObservabilityMetrics(
+        _METER_NAME,
+        meter_provider=meter_provider,
+    )
+    return meter_provider
+
+
+def get_metrics() -> ObservabilityMetrics:
+    global _METRICS
+
+    if _METRICS is None:
+        _METRICS = ObservabilityMetrics(_METER_NAME)
+    return _METRICS
+
+
+def reset_metrics_for_tests() -> None:
+    global _METER_PROVIDER, _METRICS
+
+    if _METER_PROVIDER is not None:
+        _METER_PROVIDER.shutdown()
+    _METER_PROVIDER = None
+    _METRICS = None
@@ -0,0 +1,26 @@
+"""High-level observability setup wiring."""
+
+from fastapi import FastAPI
+
+from app.config import ObservabilityConfig
+from app.observability.logging import configure_structured_logging
+from app.observability.metrics import setup_metrics
+from app.observability.tracing import (
+    instrument_fastapi_app,
+    instrument_httpx_client,
+    setup_tracing,
+)
+
+
+def setup_observability(app: FastAPI, config: ObservabilityConfig) -> None:
+    configure_structured_logging(config.log_level)
+    setup_tracing(
+        service_name=config.service_name,
+        otlp_endpoint=config.otlp_endpoint,
+    )
+    setup_metrics(
+        service_name=config.service_name,
+        otlp_endpoint=config.otlp_endpoint,
+    )
+    instrument_fastapi_app(app)
+    instrument_httpx_client()
@@ -0,0 +1,80 @@
+"""OpenTelemetry tracing setup and instrumentation helpers."""
+
+from typing import cast
+
+from fastapi import FastAPI
+from opentelemetry import trace
+from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
+from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
+from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
+from opentelemetry.sdk.resources import Resource
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import BatchSpanProcessor
+from opentelemetry.trace import Tracer, TracerProvider as TraceAPIProvider
+
+_TRACER_PROVIDER: TraceAPIProvider | None = None
+_HTTPX_INSTRUMENTED = False
+
+
+def setup_tracing(
+    *,
+    service_name: str,
+    otlp_endpoint: str,
+    tracer_provider: TraceAPIProvider | None = None,
+) -> TraceAPIProvider:
+    global _TRACER_PROVIDER
+
+    if tracer_provider is None and _TRACER_PROVIDER is not None:
+        return _TRACER_PROVIDER
+
+    if tracer_provider is None:
+        tracer_provider = TracerProvider(
+            resource=Resource.create({"service.name": service_name}),
+        )
+        if otlp_endpoint:
+            span_exporter = OTLPSpanExporter(
+                endpoint=otlp_endpoint,
+                insecure=otlp_endpoint.startswith("http://"),
+            )
+            tracer_provider.add_span_processor(BatchSpanProcessor(span_exporter))
+
+    _TRACER_PROVIDER = tracer_provider
+    return tracer_provider
+
+
+def instrument_fastapi_app(app: FastAPI) -> None:
+    tracer_provider = _TRACER_PROVIDER
+    if tracer_provider is None:
+        return
+    FastAPIInstrumentor.instrument_app(app, tracer_provider=tracer_provider)
+
+
+def instrument_httpx_client() -> None:
+    global _HTTPX_INSTRUMENTED
+
+    tracer_provider = _TRACER_PROVIDER
+    if tracer_provider is None:
+        return
+
+    if _HTTPX_INSTRUMENTED:
+        return
+    HTTPXClientInstrumentor().instrument(tracer_provider=tracer_provider)
+    _HTTPX_INSTRUMENTED = True
+
+
+def get_tracer(name: str) -> Tracer:
+    tracer_provider = _TRACER_PROVIDER
+    if tracer_provider is None:
+        return trace.get_tracer(name)
+    return cast(Tracer, tracer_provider.get_tracer(name))
+
+
+def reset_tracing_for_tests() -> None:
+    global _TRACER_PROVIDER, _HTTPX_INSTRUMENTED
+
+    if _HTTPX_INSTRUMENTED:
+        HTTPXClientInstrumentor().uninstrument()
+    if isinstance(_TRACER_PROVIDER, TracerProvider):
+        _TRACER_PROVIDER.shutdown()
+    _TRACER_PROVIDER = None
+    _HTTPX_INSTRUMENTED = False
@@ -13,6 +13,8 @@ from app.domain.price import (
    filter_and_sort_prices,
    normalize_delivery_request,
 )
+from app.observability.metrics import get_metrics
+from app.observability.tracing import get_tracer
 from app.schemas.request import DeliveryRequest
 from app.schemas.response import DeliveryPrice

@@ -44,33 +46,41 @@ class AggregatorService:
        self._filter_and_sort_prices = filter_and_sort_prices_fn

    async def get_all_prices(self, request: DeliveryRequest) -> list[DeliveryPrice]:
-        normalized_request = normalize_delivery_request(
-            request, weight_round_scale=self._weight_round_scale
-        )
-        provider_request = self._to_provider_request(normalized_request)
+        tracer = get_tracer(__name__)
+        with tracer.start_as_current_span("AggregatorService.get_all_prices") as span:
+            span.set_attribute("from_city", request.from_city)
+            span.set_attribute("to_city", request.to_city)
+            span.set_attribute("weight_kg", float(request.weight_kg))

-        provider_results = await asyncio.gather(
-            *(
-                self._get_provider_price(
-                    provider=provider,
-                    request=provider_request,
-                    cache_key=self._build_cache_key(
-                        provider_name=provider.name,
-                        request=normalized_request,
-                    ),
-                )
-                for provider in self._providers
-            ),
-            return_exceptions=True,
-        )
+            normalized_request = normalize_delivery_request(
+                request, weight_round_scale=self._weight_round_scale
+            )
+            provider_request = self._to_provider_request(normalized_request)

-        successful_results = [
-            price
-            for price in provider_results
-            if isinstance(price, DeliveryPrice)
-        ]
-        filtered_and_sorted = self._filter_and_sort_prices(successful_results)
-        return [self._coerce_delivery_price(price) for price in filtered_and_sorted]
+            provider_results = await asyncio.gather(
+                *(
+                    self._get_provider_price(
+                        provider=provider,
+                        request=provider_request,
+                        cache_key=self._build_cache_key(
+                            provider_name=provider.name,
+                            request=normalized_request,
+                        ),
+                    )
+                    for provider in self._providers
+                ),
+                return_exceptions=True,
+            )
+
+            successful_results = [
+                price
+                for price in provider_results
+                if isinstance(price, DeliveryPrice)
+            ]
+            filtered_and_sorted = self._filter_and_sort_prices(successful_results)
+            result = [self._coerce_delivery_price(price) for price in filtered_and_sorted]
+            span.set_attribute("tariffs_found", len(result))
+            return result

    async def _get_provider_price(
        self,
@@ -79,31 +89,59 @@ class AggregatorService:
        request: DeliveryRequest,
        cache_key: str,
    ) -> DeliveryPrice:
-        cached_price = await self._get_cached_price(cache_key)
-        if cached_price is not None:
-            return cached_price
+        tracer = get_tracer(__name__)
+        with tracer.start_as_current_span("DeliveryProvider.get_price") as span:
+            span.set_attribute("provider", provider.name)
+            span.set_attribute("from_city", request.from_city)
+            span.set_attribute("to_city", request.to_city)
+            span.set_attribute("weight_kg", float(request.weight_kg))

-        fresh_price = await provider.get_price(request)
-        await self._set_cached_price(
-            cache_key,
-            fresh_price,
-            ttl=getattr(provider, "cache_ttl_seconds", None),
-        )
-        return fresh_price
+            cached_price = await self._get_cached_price(cache_key)
+            span.set_attribute("cache_hit", cached_price is not None)
+            if cached_price is not None:
+                return cached_price
+
+            try:
+                fresh_price = await provider.get_price(request)
+            except Exception:
+                get_metrics().record_provider_availability(
+                    provider=provider.name,
+                    is_available=False,
+                )
+                raise
+
+            get_metrics().record_provider_availability(
+                provider=provider.name,
+                is_available=True,
+            )
+            await self._set_cached_price(
+                cache_key,
+                fresh_price,
+                ttl=getattr(provider, "cache_ttl_seconds", None),
+            )
+            return fresh_price

    async def _get_cached_price(self, cache_key: str) -> DeliveryPrice | None:
-        if self._cache is None:
-            return None
-        try:
-            payload = await self._cache.get(cache_key)
-        except Exception:
-            return None
-        if payload is None:
-            return None
-        try:
-            return self._coerce_delivery_price(payload)
-        except Exception:
-            return None
+        tracer = get_tracer(__name__)
+        with tracer.start_as_current_span("PriceCache.get") as span:
+            if self._cache is None:
+                span.set_attribute("cache_hit", False)
+                return None
+            try:
+                payload = await self._cache.get(cache_key)
+            except Exception:
+                span.set_attribute("cache_hit", False)
+                return None
+            if payload is None:
+                span.set_attribute("cache_hit", False)
+                return None
+            try:
+                price = self._coerce_delivery_price(payload)
+            except Exception:
+                span.set_attribute("cache_hit", False)
+                return None
+            span.set_attribute("cache_hit", True)
+            return price

    async def _set_cached_price(
        self,
@@ -112,12 +150,14 @@ class AggregatorService:
        *,
        ttl: int | None,
    ) -> None:
-        if self._cache is None:
-            return
-        try:
-            await self._cache.set(cache_key, payload, ttl=ttl)
-        except Exception:
-            return
+        tracer = get_tracer(__name__)
+        with tracer.start_as_current_span("PriceCache.set"):
+            if self._cache is None:
+                return
+            try:
+                await self._cache.set(cache_key, payload, ttl=ttl)
+            except Exception:
+                return

    @staticmethod
    def _to_provider_request(request: NormalizedDeliveryRequest) -> DeliveryRequest: