Привет, Хабр!

Решение проблем при работе различных приложений очень часто превращается в увлекательный квест. Например, представьте себе неприятную ситуацию — ваше приложение на FastAPI падает в продуктиве. Вы открываете Kibana или CloudWatch и видите:

Далее вам необходимо ответить на несколько простых вопросов. Например, нужно узнать, сколько запросов сделал пользователь 12 345 за последнюю минуту или какое среднее время ответа у эндпоинта /api/order. Также было бы неплохо узнать, все ли ошибки связаны с таймаутами или есть другие причины.

Но ответить на эти вопросы будет непросто, потому что вся информация размазана по строкам, которые выглядят по‑разному, а ID пользователя то появляется, то исчезает.

И здесь хорошим решением является просто перестать логировать строки и начать логировать объекты по принципу один объект — одно событие. При этом все поля строго типизированы.

Давайте посмотрим, как это можно сделать в коде на Python.

Начнём с того, что забудем про стандартный 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 в конкретном месте, поля просто пропадут. Скажем честно, это не очень удобно.

У нас уже есть простой рабочий пример, но он не слишком приспособлен к реальной жизни. Нам нужно, чтобы все логи внутри обработчика запроса автоматически содержали 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.

В микросервисной архитектуре запрос проходит через 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})
    # ...

Стандартный 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 типов исключений, или отфильтровать ошибки по конкретному файлу или функции. Также не лишней будет возможность посмотреть стектрейс не как одну строку, а как читаемый список.

Теперь давайте добавим в наш код автоматический подсчёт длительности запроса. Это очень важная информация для 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 — размер ответа. Отличный показатель для оптимизации сериализации.

Теперь давайте соберем всё в единый модуль 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-проектах.

Полный список бесплатных уроков августа смотрите в дайджесте.