Когда логи кричат, а ты не слышишь: как я приводил логи в порядок в ETL‑коннекторе / Хабр

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

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

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

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

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

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

Fetcher — получает записи из БД.
Transformer — преобразует записи в нужный формат.
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 — действие не выполнено.

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

Ситуация	Уровень	Комментарий
«Запись не найдена» (нормальный кейс)	WARNING	Не ошибка сервиса, но стоит отметить.
«API вернул 429, повторяем»	WARNING	Деградация, но есть повтор.
«Пользователь ввёл неверные данные»	INFO	Клиентская ошибка, не сбой сервиса.
«Таймаут при запросе к API, есть retry»	WARNING	Временная проблема, работа продолжается.
«Батч обработан, часть записей пропущена»	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: каждый логирует по‑своему → карта логов

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

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

Компонент	Событие	Уровень	Обязательные поля	Зачем нужен	Контекст
Fetcher	fetch_started	DEBUG	`request_id`	Понять, когда началась выборка	Начало работы фетчера
Fetcher	fetch_finished	INFO	`request_id`, `total`	Понять, сколько записей пришло	Окончание выборки
Transformer	transform_finished	INFO	`request_id`, `total`, `errors`	Оценить результат трансформации	Окончание трансформации
Transformer	transform_failed	ERROR	`request_id`, `record_id`, `error_type`	Найти конкретную запись с ошибкой	Ошибка валидации/преобразования
Sender	send_finished	INFO	`request_id`, `total`, `errors`	Оценить результат отправки	Окончание отправки
Sender	send_failed	ERROR	`request_id`, `record_id`, `error_type`	Найти запись, которую не удалось отправить	Ошибка отправки в API

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

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

Живёт в репозитории проекта (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 минут.

Показатель	Было	Стало
Строк лога на запуск	~15 000	~20 (статистика + ошибки)
Время поиска ошибки	~30 минут	~2 минуты
Количество ложных алертов	~10 в день	~0

Структурированные логи — это не цель, а следствие. Цель — чтобы логи были полезными. А для этого нужны правила (уровни), сквозной контекст (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. Если у вас уже есть логи, начните с карты. Выпишите все события в таблицу — станет видно, где дубли, где путаница с уровнями и где не хватает контекста. А если хотите ещё быстрее — начните с одного компонента, например с сендера. Не пытайтесь переделать всё сразу.

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

Когда логи кричат, а ты не слышишь: как я приводил логи в порядок в ETL‑коннекторе