cgroups v2: управление ресурсами контейнеров в современном Linux

Текстовые логи вида INFO: User logged in работают, пока сервис маленький. Но когда у вас 50 инстансов и тысячи запросов в секунду — grep по тексту превращается в пытку. structlog решает проблему: каждая запись — JSON с типизированными полями, которые можно фильтровать и агрегировать.

Зачем structured logging

# Текстовые логи — grep и надежда
$ grep "payment failed" app.log | grep "user_id=42"
# Ничего, потому что формат был "Payment error for user 42"

# Структурированные логи — точный запрос
$ cat app.log | jq 'select(.event=="payment_failed" and .user_id==42)'
{"event": "payment_failed", "user_id": 42, "amount": 99.90, "error": "insufficient_funds"}

JSON-логи позволяют: фильтровать по любому полю в ELK/Loki, строить метрики по типу ошибок, трейсить запрос через request_id по всем сервисам.

structlog: архитектура процессоров

structlog — обёртка над стандартным logging с pipeline-архитектурой. Каждое событие проходит через цепочку процессоров:

import structlog

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.add_log_level,
        structlog.processors.CallsiteParameterAdder(
            [structlog.processors.CallsiteParameter.FILENAME,
             structlog.processors.CallsiteParameter.LINENO]
        ),
        structlog.processors.JSONRenderer(),
    ],
    wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
    logger_factory=structlog.PrintLoggerFactory(),
    cache_logger_on_first_use=True,
)

Процессоры — функции (logger, method_name, event_dict) -> event_dict. Можно писать свои: маскировать PII, добавлять trace_id, фильтровать по условию.

Настройка: dev vs prod

import sys
import structlog

def configure_logging(env: str = "prod"):
    shared_processors = [
        structlog.contextvars.merge_contextvars,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.add_log_level,
        structlog.processors.format_exc_info,
    ]

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

    structlog.configure(
        processors=shared_processors + [renderer],
        wrapper_class=structlog.make_filtering_bound_logger(
            logging.DEBUG if env == "dev" else logging.INFO
        ),
        logger_factory=structlog.PrintLoggerFactory(file=sys.stdout),
        cache_logger_on_first_use=True,
    )

ConsoleRenderer в dev — цветной текст с выделением ключей. JSONRenderer в prod — одна строка JSON на событие для сборщиков логов.

Контекст: bind() и contextvars

Главная сила structlog — привязка контекста. Один раз привязали request_id — он появляется во всех логах запроса:

import structlog
import uuid

log = structlog.get_logger()

# Привязка к экземпляру логгера
logger = log.bind(service="payment", version="2.1.0")
logger.info("service_started", port=8080)
# {"event": "service_started", "service": "payment", "version": "2.1.0", "port": 8080}

# contextvars — контекст на уровне корутины
async def handle_request(request):
    structlog.contextvars.clear_contextvars()
    structlog.contextvars.bind_contextvars(
        request_id=str(uuid.uuid4()),
        user_id=request.user.id,
    )
    # Все логи внутри этого запроса получат request_id автоматически
    log.info("request_started")
    result = await process(request)
    log.info("request_completed", status=200)

contextvars работает с asyncio — каждая корутина имеет свой контекст, логи не перемешиваются.

Интеграция с FastAPI

import time
import uuid
import structlog
from starlette.middleware.base import BaseHTTPMiddleware

log = structlog.get_logger()

class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        structlog.contextvars.clear_contextvars()
        request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))

        structlog.contextvars.bind_contextvars(
            request_id=request_id,
            method=request.method,
            path=request.url.path,
        )

        start = time.perf_counter()
        try:
            response = await call_next(request)
            elapsed_ms = (time.perf_counter() - start) * 1000
            log.info("request_completed", status=response.status_code, duration_ms=round(elapsed_ms, 2))
            response.headers["X-Request-ID"] = request_id
            return response
        except Exception as exc:
            log.error("request_failed", error=str(exc))
            raise

Каждый лог внутри обработчика автоматически содержит request_id, method, path — без явной передачи.

Связка с ELK/Loki

JSON-логи в stdout — стандарт для контейнеров. Docker/Kubernetes собирают stdout и отправляют в систему логирования:

• ELK: Logstash парсит JSON, индексирует в Elasticsearch. Kibana фильтрует по любому полю.
• Grafana Loki: легковесная альтернатива, индексирует только labels. Дешевле ELK в 10 раз.

# LogQL запрос в Loki
{service="payment"} | json | event="payment_failed" | amount > 100

Структурированные поля становятся фильтрами — не нужно писать regex по тексту.

Выводы

1. structlog + JSONRenderer — стандарт для продакшена. Текстовые логи — только для dev.
2. Pipeline процессоров — гибкая архитектура для обогащения и фильтрации.
3. contextvars — привязка request_id без передачи логгера через все функции.
4. Middleware в FastAPI — автоматический контекст на каждый запрос.
5. JSON в stdout → Docker → Loki/ELK — zero-config сбор и анализ.
6. Переход постепенный: structlog оборачивает стандартный logging.