Три часа ночи, алерт, сервис отдаёт 500. Открываете логи и видите:
2026-05-12 03:14:22 ERROR Something went wrong
2026-05-12 03:14:22 ERROR Failed to process request
2026-05-12 03:14:23 ERROR Unexpected error occurred
2026-05-12 03:14:23 INFO Request completedКакой запрос сломался? Какой пользователь? Какой endpoint? Что за ошибка? В логе этого нет.
Вы начинаете прыгать по timestamp, пытаетесь руками сопоставить строки друг с другом, а если сервис обрабатывает 100 запросов в секунду, логи от разных запросов перемешаны в кашу, и отделить один от другого невозможно.
Проблема вообще не в количестве логов и не в уровне детализации. Проблема в формате. Текстовые логи написаны для человека, который сидит перед терминалом и читает их глазами. Но в три часа ночи при 100 RPS глазами ничего не найдёшь. Нужны логи, написанные для машины: фильтруемые, группируемые, с полями, по которым можно строить запросы.
Это и есть structured logging.
Текст vs JSON
Обычный лог:
2026-05-12 03:14:22 ERROR Failed to process payment for user 12345: connection timeoutВся информация закодирована в строке: user ID, тип ошибки, операция. Чтобы отфильтровать по user_id, нужен regex. Чтобы найти все таймауты, нужен другой regex. Чтобы посчитать количество ошибок по типам — третий. Каждый regex сломается, когда формат сообщения чуть изменится.
Structured лог — тот же самый набор данных, но каждое поле отдельно:
{
"timestamp": "2026-05-12T03:14:22.456Z",
"level": "error",
"message": "Failed to process payment",
"user_id": 12345,
"payment_id": "pay_abc123",
"error": "connection timeout",
"service": "payment-api",
"request_id": "req_xyz789",
"duration_ms": 5023
}Теперь jq 'select(.user_id == 12345)' даёт все события этого пользователя. В Kibana или Loki запрос service="payment-api" AND level="error" AND duration_ms > 3000 покажет все медленные ошибки за час. Без regex‑ов, без угадывания формата.
Python: structlog
В Python для structured logging есть structlog. Работает поверх стандартного logging или standalone, настраивается за 10 минут.
pip install structlogimport structlog
structlog.configure(
processors=[
structlog.contextvars.merge_contextvars,
structlog.processors.add_log_level,
structlog.processors.TimeStamper(fmt="iso"),
structlog.processors.format_exc_info,
structlog.processors.JSONRenderer(),
],
logger_factory=structlog.PrintLoggerFactory(),
)
log = structlog.get_logger()Тут стоит объяснить, что такое processors. В structlog каждое лог‑событие проходит через цепочку процессоров: один добавляет timestamp, другой — уровень, третий — контекстные переменные, последний сериализует в JSON. Можно добавлять свои: например, процессор, который маскирует email‑адреса или убирает поля с персональными данными.
Использование:
log.info("payment_started", user_id=123, amount=99.99, currency="USD")
log.error("payment_failed", user_id=123, error="timeout", retry_count=2)Вывод:
{"event": "payment_started", "user_id": 123, "amount": 99.99, "currency": "USD", "level": "info", "timestamp": "2026-05-12T03:14:22Z"}
{"event": "payment_failed", "user_id": 123, "error": "timeout", "retry_count": 2, "level": "error", "timestamp": "2026-05-12T03:14:27Z"}Тут еще обратите внимание: вы не конструируете строку руками (f"Payment failed for user {user_id}: {error}"). Вы передаёте поля как именованные аргументы. Structlog сам решает, как их отформатировать. В продакшене — JSON. В разработке можно переключить на красивый текстовый вывод с цветами (заменив JSONRenderer на ConsoleRenderer). Код логирования не меняется.
Request ID: связываем логи одного запроса
Самое ценное в structured logging — возможность привязать все логи одного запроса к одному идентификатору. Без этого при 100 RPS строки от разных запросов перемешиваются, и понять, какой лог к какому запросу относится, невозможно.
В FastAPI это делается через middleware и contextvars:
import uuid
from starlette.middleware.base import BaseHTTPMiddleware
class RequestIDMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
structlog.contextvars.clear_contextvars()
structlog.contextvars.bind_contextvars(
request_id=request_id,
method=request.method,
path=request.url.path,
)
response = await call_next(request)
response.headers["X-Request-ID"] = request_id
return response
app = FastAPI()
app.add_middleware(RequestIDMiddleware)bind_contextvars привязывает поля к текущему контексту выполнения (asyncio task). После этого каждый вызов log.info(...), log.error(...) в любом месте обработки запроса автоматически содержит request_id, method и path. Не нужно передавать их через все функции, не нужно протаскивать logger как аргумент.
@app.get("/api/orders/{order_id}")
async def get_order(order_id: int):
log.info("fetching_order", order_id=order_id)
order = await db.fetch_order(order_id)
if not order:
log.warning("order_not_found", order_id=order_id)
raise HTTPException(404)
log.info("order_fetched", order_id=order_id, status=order.status)
return order
Все три лог‑записи будут содержать один request_id. В Kibana фильтруете по нему и видите полный путь запроса: что искали, нашли ли, сколько заняло.
Если у вас Django вместо FastAPI, принцип тот же: middleware, который привязывает request_id к thread-local или contextvars. Для Flask — before_request hook.
Go: slog из стандартной библиотеки
В Go с версии 1.21 structured logging есть из коробки, без зависимостей. Пакет log/slog стал частью стандартной библиотеки.
import (
"log/slog"
"os"
)
func main() {
logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
Level: slog.LevelInfo,
}))
slog.SetDefault(logger)
slog.Info("server started", "port", 8080, "env", "production")
slog.Error("payment failed",
"user_id", 123,
"error", "connection timeout",
"duration_ms", 5023,
)
}До slog в Go‑сообществе были zerolog, zap, logrus — каждый со своим API, своими уровнями, своими форматами. slog не заменяет их полностью (у zerolog, например, zero‑allocation logging, что может быть критично при 100K RPS), но для большинства сервисов slog достаточен и не требует зависимости.
Request ID в Go привязывается через context:
func RequestIDMiddleware(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
requestID := r.Header.Get("X-Request-ID")
if requestID == "" {
requestID = uuid.New().String()
}
logger := slog.Default().With("request_id", requestID,
"method", r.Method, "path", r.URL.Path)
ctx := context.WithValue(r.Context(), "logger", logger)
w.Header().Set("X-Request-ID", requestID)
next.ServeHTTP(w, r.WithContext(ctx))
})
}Метод With создаёт дочерний логгер с привязанными полями. Все записи через этот логгер будут содержать request_id, method и path.
Что логировать, а что не логировать
Это вопрос, на который нет универсального ответа, но есть здравый смысл.
Логируйте начало и конец обработки запроса (с duration_ms — это бесценно для диагностики latency). Логируйте ошибки с контекстом: не просто «ошибка», а что за операция, какие входные данные, какой ответ от downstream. Логируйте обращения к внешним сервисам: какой сервис, сколько ждали, какой статус ответа. Логируйте бизнес‑события: заказ создан, платёж прошёл, пользователь заблокирован — то, что потом пригодится для расследования инцидентов и аналитики.
Не логируйте пароли, токены, номера карт, персональные данные. Это нарушение GDPR/ФЗ-152, за которое штрафуют. Не логируйте тела запросов и ответов целиком: объём огромный, конфиденциальные данные внутри, и в 99% случаев вам нужны два‑три поля, а не всё тело. Не логируйте на уровне DEBUG в продакшене: потоп данных забьёт диск и Kibana, а полезного в нём мало.
# Плохо
log.info("login", username=username, password=password)
log.info("request", body=request.body)
# Хорошо
log.info("login_attempt", username=username)
log.info("order_received", order_id=data["id"], amount=data["amount"])Как не утонуть при высокой нагрузке
При 10 000 RPS и 5 строках лога на запрос вы генерируете 50 000 строк в секунду. За сутки это больше 4 миллиардов строк. Даже в JSON это терабайты.
Решение в семплирование. Логируете 10% INFO‑запросов, но 100% ошибок. Объём падает на порядок, а диагностика работает: ошибки видны все, нормальный трафик — выборочно.
import random
import structlog
class SamplingProcessor:
def __init__(self, sample_rate=0.1):
self.sample_rate = sample_rate
def __call__(self, logger, method_name, event_dict):
if event_dict.get("level") in ("error", "warning", "critical"):
return event_dict
if random.random() > self.sample_rate:
raise structlog.DropEvent
return event_dictДобавляете SamplingProcessor(0.1) в цепочку процессоров structlog. 10% INFO, 100% WARNING и выше. Для большинства задач этого достаточно: если баг воспроизводится на 10% запросов, вы его увидите. Если на 0.01%, вам нужны не логи, а трейсинг.
Structured logging — возможность в три часа ночи за минуту найти конкретный запрос, понять что с ним произошло и кто виноват. JSON вместо текста, request_id для корреляции, slog или structlog вместо print и fmt.Println. Полчаса на настройку, и следующий алерт вы расследуете в разы быстрее.
Если у вас есть свои подходы к логированию или нашли ошибку, пишите в комментариях. Спасибо, что дочитали.
Если после настройки structured logging хочется пойти дальше — связать логи, метрики и трейсы в одну картину и быстрее разбирать инциденты в продакшене, у OTUS есть два бесплатных открытых урока по теме:
— 19 мая, 20:00 — «Введение в OpenTelemetry и основы наблюдаемости»
Разберёмся, как строится observability и зачем команде единый контекст для диагностики сервисов.
— 1 июня, 20:00 — «Мониторинг распределенных систем»
Поговорим о том, как отслеживать состояние сложных систем, находить источник сбоев и не тонуть в техническом шуме.
Открытые уроки проходят в рамках онлайн‑курсов OTUS: можно познакомиться с преподавателями‑практиками, посмотреть на формат обучения и задать вопросы по теме.
А чтобы не пропускать новые материалы про разработку, архитектуру, DevOps и эксплуатацию систем — подписывайтесь на блог.