Pipeline Logstash: input, filter, output

Filebeat собрал строки логов с серверов «Гонца» и отправил их дальше. Но строка 92.100.152.106 - - [30/Jun/2026:03:00:45 +0300] "GET /api/orders/104000 HTTP/1.1" 200 1801 для Elasticsearch просто текст. Чтобы построить график ошибок по минутам, нужны отдельные поля: время, код ответа, путь, адрес клиента. Разбирать поток событий на поля, дополнять и отправлять дальше умеет Logstash. Начнём с того, как устроен его конфиг.

Конвейер на заводе

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

Пайплайн Logstash устроен так же, из трёх стадий:

• input, вход: откуда приходят события. Из Filebeat, из Kafka, из файла, из stdin;
• filter, фильтры: что с событием сделать по дороге. Разобрать строку на поля, поправить время, удалить лишнее, отбросить шум;
• output, выход: куда событие отправить. В Elasticsearch, в файл, в Kafka, на экран.

Событие

По ленте едут события. Событие похоже на документ JSON: набор полей, которые могут быть вложенными. Когда на вход приходит строка текста, Logstash превращает её в событие с несколькими полями:

• message: сама строка;
• @timestamp: время события. Пока его никто не поправил, это время, когда событие попало в Logstash;
• @version: служебная версия формата, всегда "1";
• event.original: исходная строка, нетронутой копией;
• host.hostname: на какой машине работает Logstash.

Имена вроде event.original и host.hostname взяты из ECS, общей схемы полей Elastic. В ней у каждого понятия своё имя: адрес клиента всегда source.address, код ответа http.response.status_code. Это позволяет искать по логам разных сервисов одним запросом.