Первые подводные камни логирования: практика на примере Python‑коннектора в Airflow — цифры до/после, спорные кейсы уровней, карта логов как контракт.

«Ты же сказал, что шаришь в логировании?» — примерно так выглядело начало разговора, когда лид открыл лог одного запуска и увидел 15 000 строк INFO. Так начинался мой путь к структурированным логам.

5000 записей → 15 000 строк лога → 30 минут на поиск ошибки. Так выглядел ETL‑коннектор до того, как я навёл порядок.

Вот что я сделал и что из этого вышло: реальные цифры, кейс, где новые логи помогли найти баг за 5 минут, разбор спорных кейсов уровней, инструменты (Vector → Loki), связь request_id с контекстом Airflow, карта логов как контракт и что я сознательно не стал делать.

Как всё начиналось

Общий алгоритм работы включал в себя множество этапов и разные особенности преобразований, но для простоты повествования допустим, что коннектор состоял из трёх шагов:

1. Fetcher — получает записи из БД.

2. Transformer — преобразует записи в нужный формат.

3. Sender — отправляет результат в API.

Логи сначала расставлял интуитивно: где кажется важным, там и logging.info. Либо если у пользователей появлялся запрос «хочу видеть, загрузился ли такой‑то объект с таким‑то идентификатором». Главное правило — не запускать всё локально для поиска ошибок, если есть логи. Пример:

logging.info(f"Всего записей: {len(rows)}")
for row in rows:
    logging.info(f"Получена запись: {row['id']}")

Когда записей было 100 — это работало. Когда стало 5000 — лог одного запуска вырос до ~15 000 строк (3 функции × 5000 записей). Время на поиск ошибки — до 30 минут: нужно было вручную сопоставлять строки из разных функций, искать тайминги и гадать, какая запись к какому запросу относится. И даже если понятно, что запись выгружена и виден её идентификатор, возникали вопросы к полям. Запись валидировалась через Pydantic — модель под каждый объект API.

Боль № 1: тысячи INFO на каждый цикл

Проблема: на каждую запись писали INFO. При 5000 записях это 15 000 строк, которые:

• быстро забивают хранилище и трафик;

• маскируют реальные ошибки;

• усложняют поиск проблемы.

Один и тот же идентификатор мог встречаться на этапе выгрузки из БД и на этапе трансформации. Что сломалось? Ошибка валидации Pydantic — я шёл в базу, оказывалось, что объект старый и не отвечает новым требованиям. Это нормальное поведение, а не сбой. Но API не поддерживает инкрементальную загрузку — это отдельная боль, которую планирую решить позже. Пока из‑за отсутствия дельты загружаются все объекты, и такие «ошибки» сыпались каждый запуск.

Первой идеей стала общая статистика: сколько отработано, сколько ошибок. Это было полезно для итогового отчёта по дагу, но проблему количества логов не решало. Следующий шаг — промежуточная статистика по этапам. Казалось бы, мы уже логируем ошибки и важные моменты — зачем убирать их и добавлять статистику, которая ничего не покажет?

Решение: собираем статистику в цикле, логируем один итоговый результат и отдельные ошибки. Логирование каждой записи переезжает на DEBUG:

def transform_records(rows):
    total = len(rows)
    processed = 0
    errors = 0

    for row in rows:
        try:
            # трансформация
            logging.debug(f"Получена запись: {row['id']}")
            processed += 1
        except Exception as e:
            errors += 1
            logging.error(
                "Ошибка трансформации записи",
                extra={
                    "record_id": row["id"],
                    "error_type": type(e).__name__,
                    "error_message": str(e),
                },
            )

    logging.info(
        "Трансформация завершена",
        extra={
            "total": total,
            "processed": processed,
            "errors": errors,
        },
    )

Цифры после: вместо 15 000 строк — 3–5 строк (по одной на этап) + отдельные ошибки в режиме отладки. Время поиска ошибки сократилось с ~30 минут до ~2 минут: достаточно посмотреть на строку с errors > 0 и перейти к конкретным ошибкам.

И именно логирование только ошибок решало проблему с большим логом. С другой стороны, вопрос бизнеса «а мой любимый отчёт с идентификатором 777 загрузился?» — это можно понять уже в самой API. Мне показалось такое логирование излишним: если объект загрузился, он есть в системе. Если нет — он есть в ошибках. Если нужны детали, есть режим дебага, где можно рассмотреть подробнее, если по какой‑то причине отчет таки не попал в api.

Боль № 2: три функции — три лога, а связь непонятна

Проблема: ошибка в Sender, но непонятно, откуда взялась запись. В разных функциях использовали разные идентификаторы, и без общего контекста связать их было нельзя.

Несмотря на уникальный идентификатор записи, который можно было выводить на разных этапах, это только вводило дополнительный шум. Нет конкретного этапа, и я вижу в логах, как запись родилась из БД, как впервые пошла на свидание (этап трансформации), как завела семью (загрузка в API — а так как в роли API выступал каталог данных, который как раз показывает происхождение, связи и прочее). И это было лишним. Лучше выводить необходимые поля с ошибками там, где это действительно нужно, а остальное либо в дебаг, либо вообще не показывать. При этом не всегда уникальный идентификатор проходит сквозь все этапы. Поэтому поле request_id решает эту сквозную проблему.

Решение: сквозной request_id, который протаскивается через все функции и (при необходимости) в HTTP‑заголовки.

Связь с Airflow

В Airflow я привязываю request_id к контексту запуска:

from airflow.utils.context import Context

def get_request_id(context: Context) -> str:
    # request_id = dag_run_id + task_id — это уже уникальный контекст
    return f"{context['dag_run'].run_id}:{context['task'].task_id}"

Теперь request_id не случайный UUID, а отражает реальный запуск в DAG. Это помогает сразу понять, в каком даге и задаче возникла проблема.

Если коннектор запускается вне Airflow (например, локально или из другого оркестратора), request_id генерируется как uuid4. Это позволяет сохранить единый подход: формат request_id отличается, но проброс и логирование работают одинаково.

Автоматический проброс через LoggerAdapter

Чтобы request_id гарантированно попадал во все логи (в том числе при исключениях), использую LoggerAdapter:

import logging

logger = logging.getLogger(__name__)

def run_with_context(context: dict, func, *args, **kwargs):
    adapter = logging.LoggerAdapter(logger, context)
    try:
        return func(adapter, *args, **kwargs)
    except Exception:
        # даже если код падает, request_id уже в контексте логера
        adapter.exception("Unhandled exception in pipeline")
        raise

Я прокидываю adapter через все функции пайплайна. Это чуть более многословно, чем contextvars, но зато request_id гарантированно попадает во все логи — даже если функция упала до явного логирования. Вызов выглядит так:

run_with_context({"request_id": request_id}, fetch_records)

Также request_id передаю в HTTP‑заголовках при вызове внешних API:

headers = {"X-Request-ID": request_id}

Если у вас микросервисы, request_id должен передаваться между сервисами через этот же заголовок — тогда цепочку можно отследить от начала до конца.

Про управляющую функцию

Помимо этого, для API был реализован базовый класс, который получал запись из самой API, логировал, потом сравнивал с объектом из БД и по необходимости обновлял. То же и с удалением. Так как выгружаются все объекты, возникала ситуация: каждый раз сыпались ошибки ERROR, что запись не удалось получить. Это логично — она уже была удалена. А потом — что нельзя удалить. В БД был маркер «удаленна запись или нет», но из‑за отсутствия дельты ранее удалённые записи тоже приходится прогонять.

Как понять, что там нужно логировать, а здесь нет? А если это писали разные разработчики? Ответ оказался простым: есть управляющая функция, и она должна быть источником логов. Внутренние функции API не должны спамить ERROR для нормальных кейсов — «запись не найдена» при удалённых объектах это не ошибка, это WARNING. А сам уровень логирования внутри API‑класса лучше вынести в DEBUG, оставив управляющей функции право решать, что важно, а что нет.

Боль № 3: уровни логирования — спорные кейсы

Без чётких правил каждый разработчик пишет «как чувствует». В каждом новом модуле или расширении исходного кода могут появиться свои особенности логирования из опыта коллег. А если в API будут стучаться ещё и разные коннекторы, где различные команды сами определяют, что логировать и как? Одна команда пишет понятный лог, вторая не очень понятный, а поддержка как‑то должна выживать в этой «битве логирования».

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

Правила

• DEBUG — низкоуровневая отладка (не для прода по умолчанию).

• INFO — бизнес‑события и статистика.

• WARNING — что‑то пошло не по плану, но работа продолжается.

• ERROR — действие не выполнено.

Спорные кейсы

Такой разбор помогает избежать споров и делает алерты более точными. Алерты у нас были настроены на level=INFO, поэтому каждая ошибка валидации старой записи поднимала шум. После того как я убрал эти ошибки из ERROR в WARNING и некоторые в DEBUG, ложные алерты практически исчезли.

Сравнение «до/после» в одном месте

# Было (строковый лог)
[2026-10-07T01:05:25] ERROR [req-abc123] Ошибка трансформации записи 789: ValueError — Invalid date format

# Стало (структурированный JSON)
{
  "timestamp": "2026-10-07T01:05:25Z",
  "level": "ERROR",
  "request_id": "req-abc123",
  "event": "transform_failed",
  "record_id": 789,
  "error_type": "ValueError",
  "error_message": "Invalid date format"
}

Разница: первую строку человек прочтёт, но программе придётся парсить регулярками. Вторую программа прочитает сразу — это уже данные.

Переход был постепенным: сначала добавил extra с полями к обычному logging, потом заменил logging на structlog для вывода JSON. Это позволило не переписывать всё сразу — каждый компонент мигрировал отдельно.

Боль № 4: каждый логирует по‑своему → карта логов

Разные формулировки и поля мешали автоматическому парсингу. Кроме того, я задал себе вопрос: а что я вообще логирую и где? Этих логов достаточно? (с таким‑то количеством спама в логах я ещё спрашиваю «достаточно ли», да уж…). Мне показалось, что хорошо бы для начала создать карту логов. Так я увижу ошибки в разных модулях, связи этих модулей, лишние и недостающие логи.

Карта логов — документ, который стал в итоге контрактом для команды. В чём разница: карта логов показывает, что сейчас логируется, отдельными таблицами — пробелы этого логирования (если есть), проблемы для каждой функции и затравка на контракт (как оно будет выглядеть). В этом документе появилось всё. Я могу пойти с ним к команде и сказать: «Смотрите, ребята, вот что есть». Теперь задача по уменьшению шума логов становится легкой. Делиться описаниями всех таблиц карты смысла не вижу — мой опыт вряд ли покроет ваши проблемы. В итоге получилось что‑то такое:

Каждую строку с логом стало необходимо обосновать или добавить в свой код. Зачем он нужен? Какой в нём смысл? Какой уровень. Поначалу выглядит излишне заморочено, но фактически решает сразу кучу проблем. Не «а вставлю сюда инфо, потом подумаю» и в итоге никто потом, конечно, не подумает.

Как поддерживаю карту:

• Живёт в репозитории проекта (docs/logging-map.md).

• Обновляет лид команды или автор изменений.

• Перед добавлением нового лога разработчик смотрит в карту. Если события нет — добавляет строку и согласовывает с лидом.

• Часть карты я автоматизировал: события и обязательные поля вынесены в константы, а простой тест проверяет, что в логах нет событий, которых нет в карте. Так карта — это не только документ, но и часть кода.

Карта — не «написали и забыли», а живой документ, который меняется вместе с проектом.

От карты к контракту: JSON‑логи и инструменты

Когда карта появилась, стало очевидно: раз поля фиксированы, логи можно выводить в JSON. Это даёт автоматический парсинг и фильтрацию.

Готовая карта фактически переросла в контракт логов. Если карта говорит, что сейчас есть и как хорошо бы делать, то контракт логов — делаем только так. Его удобно парсить ИИ‑агентами и соответственно корректировать код автоматически. Это упрощает логирование и делает его типовым. Новому разработчику проще понимать правила игры, а начинающему — не наступать на грабли.

Инструменты:

• Куда пишем: stdout (контейнерные логи).

• Как собираем: Vector — сборщик логов из контейнеров и нормализация формата.

• Где смотрим: Grafana Loki — поиск по request_id, дашборды по event.

• Фильтрация: пример запроса в Loki:

{app="etl-connector"} | json | event="transform_failed" | request_id="req-abc123"

• Стек‑трейсы: многострочные сообщения остаются в поле stack_trace как строка; Loki умеет их отображать.

Параллельно вынес часть статистики в метрики (Prometheus), чтобы не логировать то, что можно посчитать: например, общее количество обработанных записей, долю ошибок и время выполнения. Логи — для диагностики, метрики — для мониторинга.

Реальный кейс: как новые логи помогли найти баг

Через неделю после внедрения поймал баг: на одной из записей трансформация падала с ValueError из‑за некорректного формата даты. В старых логах это было бы «Трансформирую запись N» и дальше ошибка обработки — пришлось бы вручную сопоставлять тайминги и просматривать БД, приложение, API.

С новыми логами картина была чёткой:

• Строка Transform finished с errors=1.

• Отдельная строка transform_failed с request_id=req-abc123, record_id=789, error_type=ValueError.

Нашёл и исправил баг за 5 минут. Раньше на такой поиск уходило 30+ минут.

Важное про безопасность и PII

Логи могут содержать персональные или чувствительные данные. У меня простое правило: не логирую сырые данные записей, только идентификаторы (record_id) и типы ошибок. Но эта та тема, которую стоит еще дополнительно изучить.

Чего я НЕ стал делать (и почему)

• Не стал вводить tracing (OpenTelemetry) сразу. Для моего масштаба сквозной request_id + структурированные логи оказались достаточны. Tracing добавил бы сложность и накладные расходы.

• Не стал писать логи в БД. Это дорого и усложняет масштабирование.

• Не стал логировать в файлы внутри контейнера. Это усложняет сбор и ротацию. Пишу в stdout, а дальше Vector собирает и нормализует.

• Не стал использовать structlog сразу. Сначала навёл порядок вручную, а потом уже внедрил structlog для удобного вывода JSON.

• Не стал логировать каждую запись в цикле. Это главный источник шума.

Это не «плохие» инструменты, а просто не были нужны на том этапе. Выбирал минимально достаточное решение.

Что в итоге: цифры и выводы

Замерил время от момента, когда алерт срабатывал, до момента, когда находил причину. До внедрения — в среднем 30 минут, после — около 2 минут.

Структурированные логи — это не цель, а следствие. Цель — чтобы логи были полезными. А для этого нужны правила (уровни), сквозной контекст (request_id), контракт (карта логов) и дисциплина в применении.

Чек‑лист: как привести логи в порядок

• [ ] Выпишите все события в таблицу (компонент, событие, уровень, обязательные поля).

• [ ] Проверьте, нет ли дублей и избыточных логов в циклах.

• [ ] Убедитесь, что у каждого события есть request_id (или другой сквозной идентификатор).

• [ ] Пересмотрите уровни: DEBUG для отладки, INFO для бизнес‑событий, WARNING для некритичных проблем, ERROR для реальных сбоев.

• [ ] Проверьте, что в логах нет сырых персональных данных — только идентификаторы и типы ошибок.

• [ ] Добавьте структурированный вывод (JSON) или хотя бы фиксированный набор полей.

• [ ] Разместите карту логов в репозитории и договоритесь, как её обновлять.

• [ ] Настройте простой поиск/дашборд (Loki/Kibana) и проверьте, что по request_id легко найти всю цепочку.

Ссылки на инструменты и почему именно они

• structlog — выбрал за удобный JSON из коробки и поддержку contextvars для request_id.

• OpenTelemetry — не стал внедрять сразу: избыточно для моего масштаба.

• Grafana Loki — использую для поиска по request_id и дашбордов по event.

• Vector — нужен для сбора логов из контейнеров и нормализации формата.

• Prometheus — для метрик, чтобы не логировать то, что можно посчитать.

P. S. Если у вас уже есть логи, начните с карты. Выпишите все события в таблицу — станет видно, где дубли, где путаница с уровнями и где не хватает контекста. А если хотите ещё быстрее — начните с одного компонента, например с сендера. Не пытайтесь переделать всё сразу.

А как у вас устроены логи? Что пробовали, от чего отказались?