Конфигурация observability в Python: management-порт, логи и метрики

← назад к разделу

Когда сервис впервые запускается в проде, приходит первый вопрос: «Где смотреть метрики? Куда идут логи? Что значит DOWN на health-check?». Ответы зависят от того, как настроена observability при старте. Эта статья — про три составляющих такой настройки в FastAPI: отдельный management-порт, логирование через structlog и Prometheus-метрики.

Почему метрики и health-check нужно выносить на отдельный порт

Самый простой подход — добавить /metrics и /health прямо на основной сервер. Это работает, но создаёт неудобства:

• Scraping-трафик от Prometheus и проверки Kubernetes смешиваются с бизнес-запросами в одном event loop.
• Нельзя закрыть /metrics сетевой политикой только для внутреннего трафика, не трогая основной порт.
• Swagger UI (/docs) и схема API (/openapi.json) оказываются доступны на том же адресе, что и метрики, — их нужно отдельно отключать.

Решение — два ASGI-приложения: бизнес-трафик на основном порту (8080), management (/metrics, /health/*, /info) — на отдельном (8081):

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

def build_management_app(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())

Теперь Kubernetes-проверки ходят на 8081, Prometheus scraper тоже — а Ingress публикует только 8080.

Все параметры в одном классе Settings

Когда настройки разбросаны по отдельным переменным в разных файлах, легко пропустить, что LOG_LEVEL называется по-другому на разных серверах. Удобнее собрать всё в одном месте через 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 как переменная окружения — так же, как в сервисах на других языках. Все параметры читаются из окружения автоматически, с префиксом APP_.

Логи: читаемые локально, JSON в проде

В разработке приятно видеть строки вроде 10:42:03 [info] order_confirmed order_id=ORD-9912. В проде тот же формат — головная боль для Loki и Datadog: нужно писать regex для парсинга.

structlog решает это переключением рендерера в зависимости от окружения:

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

def add_service_info(settings):
    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 = [
        structlog.contextvars.merge_contextvars,
        structlog.stdlib.add_log_level,
        structlog.stdlib.add_logger_name,
        structlog.processors.TimeStamper(fmt="iso"),
        add_service_info(settings),
    ]

    renderer = structlog.processors.JSONRenderer() if is_prod else 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,
    )

Локально вывод выглядит так:

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

В проде — JSON, который агрегаторы логов парсят без дополнительных настроек:

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

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

Важно: configure_logging вызывается один раз при старте, не в каждом запросе. Повторный вызов structlog.configure перезаписывает настройки.

Histogram buckets для измерения латентности

По умолчанию Prometheus-гистограмма использует стандартные границы бакетов: 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0 секунд. Если у сервиса есть SLO «p99 < 500ms», стандартные бакеты дадут неточный histogram_quantile — граница 0.5 есть, но нет промежуточных значений рядом с ней.

Бакеты задаются при регистрации метрики, изменить их потом нельзя:

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

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"],
)

Стандартные labels service, env, version лучше вынести в фабричную функцию, чтобы не повторять в каждой метрике:

# 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._service = settings.service_name
        self._env     = settings.app_env
        self._version = settings.version

    def order_created(self, payment_method: str) -> None:
        _orders_created.labels(
            service=self._service,
            env=self._env,
            version=self._version,
            payment_method=payment_method,
        ).inc()

Label path: шаблон маршрута, не реальный URL

Типичная ошибка при первом подключении метрик — использовать request.url.path как значение label path. Для эндпоинта /orders/{order_id} каждый заказ создаёт отдельную time series: /orders/ORD-9912, /orders/ORD-9913, /orders/ORD-9914… Через несколько дней их становятся десятки тысяч.

В FastAPI шаблон маршрута доступен через request.scope["route"]:

# 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 настраивается один раз в 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 ParentBasedTraceIdRatio
from opentelemetry.sdk.resources import Resource
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=ParentBasedTraceIdRatio(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 и схему API с бизнес-порта — структура API не раскрывается внешнему трафику.

Частые ошибки

Один порт для бизнес-трафика и management. Scraping от Prometheus и проверки K8s идут на основной сервер. Нужно запускать два uvicorn.Server.

make_asgi_app() на бизнес-роутере. /metrics оказывается на бизнес-порту и доступен через Ingress. Prometheus scraper должен ходить на management-порт напрямую, минуя Ingress.

Стандартные бакеты гистограммы при жёстком SLO. DEFAULT_BUCKETS не покрывают нужный диапазон точно. Бакеты задаются один раз при регистрации — потом не изменить.

Raw URL в label path. /orders/ORD-9912, /orders/ORD-9913 — отдельные time series. Используйте request.scope["route"].path.

Значения сущностей как labels метрики. user_id=U-441 или order_id=ORD-9912 как label — взрыв кардинальности. Такие значения идут в traces (OTel span attributes), не в метрики.

structlog.configure() в обработчике запроса. Настройка перезаписывается на каждый запрос. Только один вызов при старте.

sampling_ratio=1.0 в проде на нагруженном сервисе. 100% трассировки дают большую нагрузку на экспортёр и хранилище. На dev/staging — ок, в проде — 0.01–0.1.

Коротко

• Два ASGI-приложения: бизнес-трафик на 8080, management (/metrics, /health/*, /info) на 8081. Так Prometheus scraper и K8s probes не мешают основному event loop.
• pydantic-settings собирает все observability-параметры в одном классе; переменные окружения с префиксом APP_.
• structlog переключает рендерер по APP_ENV: ConsoleRenderer для разработки, JSONRenderer для прода — без regex в агрегаторах логов.
• structlog.configure() вызывается один раз при старте. merge_contextvars ставится первым в цепочке процессоров.
• Бакеты гистограммы задаются при регистрации под конкретный SLO-порог. Изменить потом нельзя.
• Label path — шаблон маршрута (/orders/{order_id}), не реальный URL. Иначе каждый заказ создаёт отдельную time series.
• OTel настраивается один раз в lifespan. sampling_ratio берётся из Settings, не хардкодится.
• /docs и /openapi.json в проде — None: структура API недоступна внешнему трафику.

Что почитать дальше

• Context propagation — request_id в middleware, bind_contextvars, copy_context для offload в threads.
• Health checks — liveness, readiness с TTL-кешем, asyncpg ping.
• Logging — structlog processors, OTel-bridge, маскировка чувствительных данных.
• Metrics — RED-middleware, бизнес-счётчики, PrometheusInstrumentator.
• Tracing — OTel setup, FastAPIInstrumentor, ручные spans.