Привет, Хабр!
Решение проблем при работе различных приложений очень часто превращается в увлекательный квест. Например, представьте себе неприятную ситуацию — ваше приложение на FastAPI падает в продуктиве. Вы открываете Kibana или CloudWatch и видите:
Далее вам необходимо ответить на несколько простых вопросов. Например, нужно узнать, сколько запросов сделал пользователь 12 345 за последнюю минуту или какое среднее время ответа у эндпоинта /api/order. Также было бы неплохо узнать, все ли ошибки связаны с таймаутами или есть другие причины.
Но ответить на эти вопросы будет непросто, потому что вся информация размазана по строкам, которые выглядят по‑разному, а ID пользователя то появляется, то исчезает.
И здесь хорошим решением является просто перестать логировать строки и начать логировать объекты по принципу один объект — одно событие. При этом все поля строго типизированы.
Давайте посмотрим, как это можно сделать в коде на Python.
Шаг 1. Простейший JSON‑логгер, который уже лучше, чем print
Начнём с того, что забудем про стандартный logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s').
Вместо этого мы будем сохранять каждое сообщение в JSON. Вот простой пример кода для создания таких записей.
import logging import json from datetime import datetime class JsonFormatter(logging.Formatter): def format(self, record): log_entry = { "timestamp": datetime.utcnow().isoformat(), "level": record.levelname, "logger": record.name, "message": record.getMessage(), "module": record.module, "function": record.funcName, "line": record.lineno } # Если есть исключение — добавляем стектрейс if record.exc_info: log_entry["exception"] = self.formatException(record.exc_info) # Если в record есть дополнительные атрибуты — добавляем их if hasattr(record, "extra"): log_entry.update(record.extra) return json.dumps(log_entry, ensure_ascii=False) # Настройка logger = logging.getLogger("my_app") handler = logging.StreamHandler() handler.setFormatter(JsonFormatter()) logger.addHandler(handler) logger.setLevel(logging.INFO)
Для того чтобы добавить в лог запись о входе пользователя в систему, нам потребуется выполнить следующий вызов:
logger.info(“User logged in”, extra={“user_id”: 12345, “ip”: “192.168.1.1”})
В итоге, на выходе мы получим запись в формате JSON:
Важно: параметр extra — это стандартный механизм logging. Но у него есть проблема: если вы забудете передать extra в конкретном месте, поля просто пропадут. Скажем честно, это не очень удобно.
Шаг 2. Контекстный логгер через LoggerAdapter
У нас уже есть простой рабочий пример, но он не слишком приспособлен к реальной жизни. Нам нужно, чтобы все логи внутри обработчика запроса автоматически содержали request_id, user_id, endpoint. Опять таки, не очень удобно передавать это в каждом вызове.
Реализуем данный механизм через logging.LoggerAdapter:
import logging from uuid import uuid4 class ContextLogger(logging.LoggerAdapter): def process(self, msg, kwargs): # Базовый контекст всегда добавляется context = self.extra.copy() if self.extra else {} # Если в kwargs передан дополнительный контекст — объединяем if "extra" in kwargs: context.update(kwargs["extra"]) kwargs["extra"] = context else: kwargs["extra"] = context return msg, kwargs
Теперь реализуем взаимодействие в эндпоинте FastAPI:
from fastapi import Request import logging logger = logging.getLogger("api") @app.post("/order") def create_order(request: Request, order_data: dict): # Создаём адаптер с контекстом запроса context = { "request_id": str(uuid4()), "user_id": request.headers.get("X-User-Id", "anonymous"), "endpoint": request.url.path, "method": request.method } context_logger = ContextLogger(logger, context) context_logger.info("Order creation started", extra={"order_id": order_data.get("id")}) try: # ... бизнес-логика ... context_logger.info("Order created successfully") except Exception as e: context_logger.error("Order creation failed", exc_info=True) raise
Теперь даже если вы просто пишете context_logger.info(“Something happened”) — в лог будет автоматически добавлен request_id, user_id, endpoint и method.
Теперь у нас есть готовые JSON для Elasticsearch/Kibana и мы можем применить фильтр по request_id — и увидеть весь путь запроса от входа до выхода. А еще можно посмотреть сколько заказов создал конкретный пользователь, выполнив аггрегацию по user_id.
Шаг 3. Как добавить trace_id для распределённой трассировки
В микросервисной архитектуре запрос проходит через 5 (а может и больше) сервисов. Логи разбросаны по разным инстансам. Давайте попробуем их связать. Для этого, мы используем заголовок X‑Trace‑ID, который передаётся от сервиса к сервису. Также, возможна ситуация, когда его нет, и в таком случае мы сгенерируем новый.
Для этого давайте немного модифицируем FastAPI.
from fastapi import Request from starlette.middleware.base import BaseHTTPMiddleware class LoggingMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): # Получаем trace_id из заголовков или генерируем trace_id = request.headers.get("X-Trace-ID") or str(uuid4()) # Кладём в state, чтобы достать в любом месте приложения request.state.trace_id = trace_id # Создаём контекстный логгер context = { "trace_id": trace_id, "request_id": str(uuid4()), "client_ip": request.client.host, "user_agent": request.headers.get("User-Agent") } # Сохраняем в request.state для доступа в эндпоинтах request.state.logger = ContextLogger(logger, context) # Логируем входящий запрос request.state.logger.info( "Incoming request", extra={"path": request.url.path, "query": str(request.query_params)} ) try: response = await call_next(request) request.state.logger.info( "Request completed", extra={"status_code": response.status_code} ) return response except Exception as e: request.state.logger.error("Unhandled exception", exc_info=True) raise
Теперь в каждом эндпоинте вы можете достать логгер через request.state.logger:
@app.get("/items/{item_id}") def get_item(request: Request, item_id: int): logger = request.state.logger logger.info("Fetching item", extra={"item_id": item_id}) # ...
Шаг 4. Пишем кастомный JSON‑форматтер с полным стектрейсом
Стандартный formatException() возвращает многострочную строку, но в JSON это выглядит ужасно. Давайте парсить стектрейс в массив, чтобы каждая строка была отдельным элементом.
import traceback class StructuredJsonFormatter(logging.Formatter): def format(self, record): log_entry = { "timestamp": datetime.utcnow().isoformat() + "Z", "level": record.levelname, "logger": record.name, "message": record.getMessage(), "module": f"{record.module}:{record.lineno}", "function": record.funcName } # Добавляем все атрибуты из extra if hasattr(record, "extra"): log_entry.update(record.extra) # Форматируем исключение как структуру if record.exc_info: exc_type, exc_value, exc_tb = record.exc_info # Берём последние 5 фреймов, чтобы не захламлять stack_summary = traceback.extract_tb(exc_tb) log_entry["exception"] = { "type": exc_type.__name__, "message": str(exc_value), "stack": [ { "file": frame.filename, "line": frame.lineno, "function": frame.name, "code": frame.line } for frame in stack_summary[-5:] # только последние 5 вызовов ] } return json.dumps(log_entry, ensure_ascii=False, default=str)
Обратите внимание на default=str — это спасёт от ошибок сериализации, если в extra попадёт, например, datetime или Decimal.
Вот пример вывода при ошибке:
{ "timestamp": "2026-08-06T14:23:45.123456Z", "level": "ERROR", "logger": "api.payment", "message": "Payment processing failed", "module": "payment.py:89", "function": "process_payment", "trace_id": "abc-123-def", "user_id": 12345, "exception": { "type": "ConnectionError", "message": "Timeout connecting to bank API", "stack": [ {"file": "/app/services/bank.py", "line": 45, "function": "_send_request", "code": "response = session.post(url)"}, {"file": "/app/services/payment.py", "line": 78, "function": "process_payment", "code": "bank_response = bank.send(data)"}, {"file": "/app/handlers/order.py", "line": 34, "function": "create_order", "code": "payment.process(order)"} ] } }
Теперь в Kibana вы можете сделать много всего интересного. Например можно выполнить визуализацию топ-5 типов исключений, или отфильтровать ошибки по конкретному файлу или функции. Также не лишней будет возможность посмотреть стектрейс не как одну строку, а как читаемый список.
Шаг 5. Логирование производительности: автоматический замер времени
Теперь давайте добавим в наш код автоматический подсчёт длительности запроса. Это очень важная информация для SRE и разработчиков.
import time class LoggingMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): start_time = time.perf_counter() # ... генерация context и logger ... try: response = await call_next(request) duration = time.perf_counter() - start_time request.state.logger.info( "Request finished", extra={ "status_code": response.status_code, "duration_ms": round(duration * 1000, 2), "size_bytes": len(response.body) if hasattr(response, "body") else 0 } ) # Добавляем заголовок для клиента, чтобы он знал, сколько шёл запрос response.headers["X-Response-Time"] = str(round(duration * 1000, 2)) return response except Exception as e: duration = time.perf_counter() - start_time request.state.logger.error( "Request failed", extra={"duration_ms": round(duration * 1000, 2)}, exc_info=True ) raise
Теперь у вас есть:
duration_ms — время выполнения каждого запроса. Можно строить графики p95, p99.
size_bytes — размер ответа. Отличный показатель для оптимизации сериализации.
Шаг 6. Как настроить всё это в приложении (готовый boilerplate)
Теперь давайте соберем всё в единый модуль logging_config.py:
import logging import logging.config import json from datetime import datetime import traceback from uuid import uuid4 class JsonFormatter(logging.Formatter): def format(self, record): log_entry = { "timestamp": datetime.utcnow().isoformat() + "Z", "level": record.levelname, "logger": record.name, "message": record.getMessage(), "location": f"{record.module}:{record.lineno}", "function": record.funcName, "pid": record.process, "thread": record.threadName } if hasattr(record, "extra"): log_entry.update(record.extra) if record.exc_info: log_entry["exception"] = { "type": record.exc_info[0].__name__, "message": str(record.exc_info[1]), "trace": traceback.format_tb(record.exc_info[2]) } return json.dumps(log_entry, ensure_ascii=False, default=str) def setup_logging(level=logging.INFO): config = { "version": 1, "formatters": { "json": {"()": JsonFormatter} }, "handlers": { "console": { "class": "logging.StreamHandler", "formatter": "json", "stream": "ext://sys.stdout" } }, "root": { "level": level, "handlers": ["console"] }, "loggers": { "uvicorn": {"level": logging.WARNING, "handlers": ["console"]}, "uvicorn.access": {"level": logging.WARNING, "handlers": ["console"]} } } logging.config.dictConfig(config) # Создаём корневой логгер return logging.getLogger("app")
В main.py:
from fastapi import FastAPI, Request from logging_config import setup_logging from middleware import LoggingMiddleware logger = setup_logging() app = FastAPI() app.add_middleware(LoggingMiddleware) @app.get("/health") def health(request: Request): request.state.logger.info("Health check called") return {"status": "ok"}
Антипаттерн, который убивает весь смысл
А теперь давайте немного поговорим о том, как не надо делать.
Не делайте так:
logger.info(f"User {user_id} bought {item_id}")
Дело в том, что парсер не сможет выделить user_id как отдельное поле. Вы не сможете построить фильтр «все логи для user_id=123». Если в user_id попадёт специальный символ — сломается вся строка.
Вместо подобных конструкций всегда используйте extra:
logger.info("User bought item", extra={"user_id": user_id, "item_id": item_id})
Что мы получили в итоге
Каждый лог — JSON. Любой парсер (Filebeat, Fluentd) может читать его без дополнительных настроек. При этом, все ключевые поля всегда присутствуют. trace_id, user_id, duration_ms. Это означает, что вы никогда не потеряете контекст.
Стек ошибок структурирован, то есть вы можете строить дашборды по типам исключений. Производительность нашего приложения также под контролем, так как вы знаете, какие эндпоинты тормозят, и можете это исправить до того, как пожалуются пользователи.
Внедрите эту схему в своём проекте, и через неделю вы не сможете смотреть на старые текстовые логи без содрогания. А когда ваш DevOps скажет спасибо за структурированные логи.
Если хотите глубже разобраться, как работать с логами в продакшене и быстрее находить причины сбоев, обратите внимание на два открытых урока OTUS:
17 августа, 20:00. «Системы логирования: ELK, EFK или Graylog?». Записаться
8 сентября, 20:00. «AI против бага: как разобрать инцидент в Python-проекте от логов до исправления». Записаться
Первый поможет выстроить подход к сбору и анализу логов, второй — применить их для расследования реальных проблем в Python-проектах.
Полный список бесплатных уроков августа смотрите в дайджесте.
