Конфигурация observability в Python — management port, structlog profiles, Prometheus buckets

Настройка observability в FastAPI: отдельный management-порт для /metrics и /health, structlog с JSON в проде и text локально, SLO buckets для гистограммы латентности.

Опубликовано 19 июня 2026 ~6 мин чтения Автор: Вадим Викулин

Опирается на правила: R-OBS-CFG-1 … R-OBS-CFG-4 и R-OBS-CFG-X1 … R-OBS-CFG-X3 из Observability Style Guide → раздел 5. Конфигурация.

Важно знать

Отдельный management-порт (APP_MANAGEMENT_PORT=8081) — FastAPI sub-app или отдельный ASGI-процесс, изолированный от business-трафика.

APP_ENV управляет форматом логов: production → JSON-renderer, иное → человекочитаемый текст.

Explicit маршруты на management-сервере: только /metrics, /health/live, /health/ready, /info. /docs и /openapi.json в проде — без auth запрещены.

SLO buckets для гистограммы латентности задаются один раз при регистрации Histogram; route pattern вместо raw URL в label — иначе каждый /orders/123 создаёт отдельную time series.

Стандартные labels service/env/version через prometheus.labels(...) при регистрации, не в каждом .labels(...).

contextvars нативно проходят через await — context-loss при async возможен только при run_in_executor, там нужен copy_context().

/docs / /redoc / /openapi.json в проде без auth — запрещены: могут раскрыть структуру API и внутренние схемы.

Конфигурация observability в Python-сервисе собирается из трёх частей: инициализация structlog (app/platform/log/), Prometheus-метрики и management-приложение (app/platform/metrics/), Pydantic-настройки (app/config.py). Эта статья — про все три.

Отдельный management-порт

R-OBS-CFG-1 — два ASGI-приложения на разных портах: бизнес-трафик на основном, management (/metrics, /health/*, /info) — на отдельном.

# app/platform/metrics/management.py
from fastapi import FastAPI
from prometheus_client import make_asgi_app

def build_management_app(settings: Settings, ready_check) -> FastAPI:
    mgmt = FastAPI(
        title="management",
        docs_url=None,
        redoc_url=None,
        openapi_url=None,
    )

    mgmt.mount("/metrics", make_asgi_app())

    @mgmt.get("/health/live")
    async def liveness():
        return {"status": "UP"}

    @mgmt.get("/health/ready")
    async def readiness():
        ok = await ready_check()
        if not ok:
            from fastapi.responses import JSONResponse
            return JSONResponse({"status": "DOWN"}, status_code=503)
        return {"status": "UP"}

    @mgmt.get("/info")
    async def info():
        return {
            "service":   settings.service_name,
            "version":   settings.version,
            "env":       settings.app_env,
        }

    return mgmt

# app/main.py
import uvicorn, asyncio
from app.config import Settings
from app.platform.metrics.management import build_management_app
from app.app import build_app

async def main():
    settings = Settings()
    app = build_app(settings)
    mgmt = build_management_app(settings, ready_check=app.state.ready_check)

    config_biz  = uvicorn.Config(app,  host="0.0.0.0", port=settings.port,            log_config=None)
    config_mgmt = uvicorn.Config(mgmt, host="0.0.0.0", port=settings.management_port, log_config=None)

    await asyncio.gather(
        uvicorn.Server(config_biz).serve(),
        uvicorn.Server(config_mgmt).serve(),
    )

if __name__ == "__main__":
    asyncio.run(main())

Что это даёт:

Network policy в K8s разрешает Prometheus scraper подключаться на 8081, Ingress публикует только 8080.
Scraping-трафик и K8s probes не занимают event loop бизнес-сервера.
docs_url=None, openapi_url=None на management-приложении — /openapi.json не экспонируется, debug-интерфейс недоступен.

Pydantic Settings — конфиг одним классом

R-OBS-CFG-2 — все observability-параметры из переменных окружения через pydantic-settings:

# app/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict
from pydantic import Field

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_prefix="APP_", case_sensitive=False)

    app_env:         str = Field("development", description="production | staging | development")
    service_name:    str = "order-service"
    version:         str = Field("unknown", alias="BUILD_VERSION")
    port:            int = 8080
    management_port: int = 8081

    log_level:       str = "INFO"
    otel_endpoint:   str = Field("", alias="OTEL_EXPORTER_OTLP_ENDPOINT")
    sampling_ratio:  float = Field(0.1, description="0.01–1.0; 1.0 только на dev")

BUILD_VERSION инжектируется из CI как env var — та же переменная, что в Java и Go.

Конфиг логирования по APP_ENV

R-OBS-CFG-4 — structlog настраивается один раз при старте, не вызывается basicConfig повторно:

# app/platform/log/setup.py
import logging, sys
import structlog
from structlog.types import EventDict, WrappedLogger

def add_service_info(settings) -> structlog.types.Processor:
    def processor(logger: WrappedLogger, method: str, event_dict: EventDict) -> EventDict:
        event_dict["service"] = settings.service_name
        event_dict["env"]     = settings.app_env
        event_dict["version"] = settings.version
        return event_dict
    return processor

def configure_logging(settings) -> None:
    is_prod = settings.app_env == "production"

    shared_processors: list[structlog.types.Processor] = [
        structlog.contextvars.merge_contextvars,
        structlog.stdlib.add_log_level,
        structlog.stdlib.add_logger_name,
        structlog.processors.TimeStamper(fmt="iso"),
        add_service_info(settings),
    ]

    if is_prod:
        renderer = structlog.processors.JSONRenderer()
    else:
        renderer = structlog.dev.ConsoleRenderer(colors=True)

    structlog.configure(
        processors=[*shared_processors, renderer],
        wrapper_class=structlog.make_filtering_bound_logger(
            logging.getLevelName(settings.log_level)
        ),
        context_class=dict,
        logger_factory=structlog.PrintLoggerFactory(sys.stdout),
        cache_logger_on_first_use=True,
    )

В dev ConsoleRenderer показывает читаемый вывод:

10:42:03 [info     ] order_confirmed    order_id=ORD-9912 customer_id=C-441

В проде JSON-вывод парсится Loki/Datadog без regex:

{"event": "order_confirmed", "level": "info", "service": "order-service", "env": "production",
 "order_id": "ORD-9912", "customer_id": "C-441", "trace_id": "4b3e...", "timestamp": "2026-06-19T10:42:03Z"}

structlog.contextvars.merge_contextvars первым в цепочке автоматически добавляет trace_id, span_id, request_id, user_id из bound contextvars — устанавливаются в middleware, не в handler'ах.

Histogram buckets и стандартные labels

R-OBS-CFG-3 — SLO buckets задаются при регистрации, не по умолчанию:

# app/platform/metrics/http.py
from prometheus_client import Histogram, Counter, REGISTRY

HTTP_REQUEST_DURATION = Histogram(
    "http_request_duration_seconds",
    "HTTP request latency",
    ["method", "path", "status_class"],
    buckets=[0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0],
)

HTTP_REQUESTS_TOTAL = Counter(
    "http_requests_total",
    "Total HTTP requests",
    ["method", "path", "status_class"],
)

Для платёжных эндпоинтов с SLO «p99 < 500ms» bucket 0.5 даёт точный histogram_quantile(0.99, ...) без интерполяции. DEFAULT_BUCKETS — только для сервисов без жёстких SLO по латентности.

R-OBS-MTR-2 — стандартные labels через константу:

# app/platform/metrics/common.py
from prometheus_client import Counter

def make_business_counter(name: str, description: str, extra_labels: list[str]) -> Counter:
    return Counter(
        name,
        description,
        ["service", "env", "version", *extra_labels],
    )

# app/order/metrics.py
from app.platform.metrics.common import make_business_counter
from app.config import Settings

_orders_created = make_business_counter(
    "orders_created_total",
    "Orders successfully created",
    ["payment_method"],
)

class OrderMetrics:
    def __init__(self, settings: Settings) -> None:
        self._created = _orders_created.labels(
            service=settings.service_name,
            env=settings.app_env,
            version=settings.version,
        )

    def order_created(self, payment_method: str) -> None:
        self._created.labels(payment_method=payment_method).inc()

Route pattern вместо raw URL в label `path`

R-OBS-MTR-7 — в FastAPI route pattern извлекается из request.scope["route"], не из request.url.path:

# app/platform/metrics/middleware.py
import time
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from app.platform.metrics.http import HTTP_REQUEST_DURATION, HTTP_REQUESTS_TOTAL

class MetricsMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start = time.perf_counter()
        response = await call_next(request)
        duration = time.perf_counter() - start

        route = request.scope.get("route")
        path = route.path if route else "unknown"
        status_class = _status_class(response.status_code)

        HTTP_REQUESTS_TOTAL.labels(
            method=request.method, path=path, status_class=status_class
        ).inc()
        HTTP_REQUEST_DURATION.labels(
            method=request.method, path=path, status_class=status_class
        ).observe(duration)

        return response


def _status_class(code: int) -> str:
    if code < 400:
        return "success"
    if code < 500:
        return "client_error"
    return "server_error"

route.path возвращает /orders/{order_id} — фиксированное число time series независимо от числа заказов в системе.

OTel автоинструментация — конфиг при старте

R-OBS-TRC-1, R-OBS-TRC-5 — OTel настраивается один раз в lifespan, sampling ratio из Settings:

# app/platform/tracing/setup.py
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.trace.sampling import TraceIdRatioBased
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor

def configure_tracing(settings, app) -> None:
    if not settings.otel_endpoint:
        return

    provider = TracerProvider(
        sampler=TraceIdRatioBased(settings.sampling_ratio),
        resource=Resource.create({
            "service.name":    settings.service_name,
            "service.version": settings.version,
            "deployment.environment": settings.app_env,
        }),
    )
    provider.add_span_processor(
        BatchSpanProcessor(OTLPSpanExporter(endpoint=settings.otel_endpoint))
    )
    trace.set_tracer_provider(provider)

    FastAPIInstrumentor.instrument_app(app)
    SQLAlchemyInstrumentor().instrument()
    HTTPXClientInstrumentor().instrument()

# app/app.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.config import Settings
from app.platform.log.setup import configure_logging
from app.platform.tracing.setup import configure_tracing

def build_app(settings: Settings) -> FastAPI:
    configure_logging(settings)

    @asynccontextmanager
    async def lifespan(app: FastAPI):
        configure_tracing(settings, app)
        yield

    return FastAPI(
        title=settings.service_name,
        lifespan=lifespan,
        docs_url="/docs" if settings.app_env != "production" else None,
        openapi_url="/openapi.json" if settings.app_env != "production" else None,
    )

docs_url=None в проде убирает Swagger UI и /openapi.json с бизнес-порта — без дополнительных переменных.

Что запрещено

Антипаттерн	Правило	Что взамен
`/docs` и `/openapi.json` в проде без auth	`R-OBS-CFG-X1`	`docs_url=None, openapi_url=None` по `APP_ENV`
Один порт для бизнес и management	`R-OBS-CFG-X2`	два `uvicorn.Server`; `:8080` и `:8081`
`make_asgi_app()` на бизнес-роутере	`R-OBS-CFG-X2`	отдельное FastAPI-приложение для `/metrics`
`DEFAULT_BUCKETS` для эндпоинтов с жёстким SLO	`R-OBS-CFG-3`	кастомные buckets с границей на SLO-пороге
`print()` / `logging.basicConfig()` вместо structlog	`R-OBS-LOG-X2`	`structlog.get_logger(__name__).info(...)`
raw URL в label `path` (`/orders/ORD-9912`)	`R-OBS-MTR-X1`	`request.scope["route"].path` → `/orders/{order_id}`
`user_id` / `order_id` как label-значение метрики	`R-OBS-MTR-X1`	traces (OTel span attributes), не метрики
`sampling_ratio=1.0` в проде на нагруженном сервисе	`R-OBS-TRC-X1`	`0.01`–`0.1`; 100% только на dev/staging
`structlog.configure(...)` внутри handler'а или при каждом запросе	`R-OBS-CFG-4`	один раз в `configure_logging(settings)` при старте

Куда дальше

Context propagation — request_id middleware, bind_contextvars, copy_context для thread-offload.
Health checks — liveness, readiness с TTL-кешем, asyncpg ping.
Logging — structlog processors, OTel-bridge, маскировка PII.
Metrics — RED-middleware, бизнес-счётчики, PrometheusInstrumentator.
SLO и алерты — recording rules, multi-window burn-rate alerts, runbook.
Tracing — OTel setup, FastAPIInstrumentor, manual span через context manager.