Настройка доставки логов в Elasticsearch через Filebeat и Logstash: конфигурации, grok-паттерны и буферизация

Короткий ответ: прямое подключение Filebeat к Elasticsearch оправдано, когда логи нужно собрать и сохранить, а разбор ограничивается готовыми процессорами ingest node. Logstash ставят в конвейер, когда требуется grok-разбор неструктурированных строк, обогащение, условная маршрутизация по разным индексам или отправка одного потока сразу в несколько приёмников.

Ниже разбираем оба варианта на конфигурациях Filebeat, Logstash и Elasticsearch версий 8.x. Примеры файлов и команд можно копировать как есть, меняя адреса хостов, имена индексов и пути к сертификатам.

Общая картина по выбору инструментов и стоимости хранения логов разобрана в материале о централизованном сборе логов в 2026 году, здесь же идём от конфигов к проверке результата.

Когда Filebeat напрямую в Elasticsearch, а когда через Logstash

Filebeat читает файлы, следит за ротацией и отправляет события дальше. Logstash принимает поток, прогоняет его через фильтры и раскладывает по выходам. Разница в том, где именно происходит обработка и сколько компонентов вам придётся эксплуатировать.

Ориентируйтесь на три вопроса: сколько типов логов нужно разобрать на поля, требуется ли условная логика с несколькими выходами, и какой запас по ресурсам есть на серверах.

Сравнительная таблица: прямой Filebeat и маршрутизация через Logstash

Практические сценарии. Логи Nginx на десяти серверах, сложного парсинга нет, в Kibana нужен только поиск по строкам: отправляйте Filebeat прямо в Elasticsearch. Неструктурированные логи Java-приложения, геообогащение по IP, разбор в поля и отправка копии событий в S3: без Logstash не обойтись.

Использование ingest pipelines Elasticsearch как альтернатива Logstash

Ingest node выполняет часть работы фильтров Logstash на стороне кластера. Доступны процессоры grok, dissect, date, geoip, user_agent, set, rename, convert, script и условный параметр if, который вычисляется скриптом Painless.

Пример pipeline для combined-формата Nginx:

PUT _ingest/pipeline/nginx-access
{
  "processors": [
    {
      "grok": {
        "field": "message",
        "patterns": [
          "%{IPORHOST:client_ip} %{USER:ident} %{USER:auth} \\[%{HTTPDATE:timestamp}\\] \"%{WORD:http_method} %{DATA:request} HTTP/%{NUMBER:http_version}\" %{NUMBER:http_response_code} %{NUMBER:bytes} \"%{DATA:referrer}\" \"%{DATA:user_agent}\""
        ],
        "ignore_failure": true
      }
    },
    {
      "date": {
        "field": "timestamp",
        "formats": ["dd/MMM/yyyy:HH:mm:ss Z"]
      }
    },
    {
      "convert": { "field": "bytes", "type": "integer" }
    }
  ]
}

Ограничения ingest pipelines: обработка ложится на узлы кластера и конкурирует с поиском, отладка grok через API медленнее, чем тест конфига Logstash, а сложные ветвления приходится писать на Painless. Если падает любой процессор без ignore_failure и on_failure, документ отклоняется с ошибкой 400, и событие не попадёт в индекс. Для простых случаев этого достаточно, для промышленного парсинга Logstash предпочтительнее.

Варианты приёмников помимо Elasticsearch, включая OpenSearch, разобраны в статье о сборе логов в OpenSearch.

Базовая конфигурация Filebeat для отправки в Elasticsearch или Logstash

Конфиг у Filebeat один: /etc/filebeat/filebeat.yml. Отличие между схемами доставки только в секции output. Пароли не храните в открытом виде, добавляйте их через keystore: filebeat keystore add ES_PASSWORD, затем подставляйте ${ES_PASSWORD} в конфиге.

Пример filebeat.yml для отправки в Elasticsearch

filebeat.inputs:
  - type: filestream
    id: nginx-access
    enabled: true
    paths:
      - /var/log/nginx/access.log
    fields:
      log_type: nginx_access
    fields_under_root: true

queue.disk:
  enabled: true
  max_size: 10GB
  path: /var/lib/filebeat/diskqueue

output.elasticsearch:
  hosts: ["https://es01.example.net:9200", "https://es02.example.net:9200"]
  username: "filebeat_writer"
  password: "${ES_PASSWORD}"
  index: "app-logs-%{[agent.version]}-%{+yyyy.MM.dd}"
  ssl:
    certificate_authorities: ["/etc/filebeat/certs/ca.crt"]
    verification_mode: full

setup.ilm.enabled: true
setup.ilm.rollover_alias: "app-logs"
setup.template.name: "app-logs"
setup.template.pattern: "app-logs-*"

logging.level: info
logging.to_files: true

Проверка перед запуском:

filebeat test config -c /etc/filebeat/filebeat.yml
filebeat test output -c /etc/filebeat/filebeat.yml

Первая команда валидирует YAML и логику конфига, вторая проверяет TCP-соединение, TLS-рукопожатие и аутентификацию с приёмником. Обе выполняйте до перезапуска службы.

Пример filebeat.yml для отправки в Logstash

output.logstash:
  hosts: ["logstash01.example.net:5044", "logstash02.example.net:5044"]
  loadbalance: true
  worker: 2
  compression_level: 3
  bulk_max_size: 1024
  backoff.init: 1s
  backoff.max: 60s
  ssl:
    enabled: true
    certificate_authorities: ["/etc/filebeat/certs/ca.crt"]
    verification_mode: full

При отправке в Logstash параметр index в Filebeat не задавайте: имя индекса и ILM определяются в pipeline Logstash. Параметр loadbalance распределяет события по всем перечисленным хостам, worker задаёт число соединений на каждый хост, worker: 2 удваивает пропускную способность при большом потоке.

Готовые конфигурации для сбора логов Nginx, системного журнала и Docker

Три типовых источника закрывают большинство задач: веб-сервер, системный журнал и контейнеры. Ниже входные секции Filebeat и соответствующие фильтры Logstash.

Сбор логов Nginx: access.log и error.log

filebeat.inputs:
  - type: filestream
    id: nginx-access
    enabled: true
    paths:
      - /var/log/nginx/access.log
    fields:
      log_type: nginx_access
    fields_under_root: true

  - type: filestream
    id: nginx-error
    enabled: true
    paths:
      - /var/log/nginx/error.log
    fields:
      log_type: nginx_error
    fields_under_root: true
    multiline.pattern: '^\d{4}/\d{2}/\d{2}'
    multiline.negate: true
    multiline.match: after

Блок multiline собирает многострочные трейсы error.log в одно событие: строка, начинающаяся с даты, открывает новое сообщение, остальные к ней присоединяются. Фильтры разбора для обоих типов показаны в разделе про grok.

Альтернатива ручному конфигу: команда filebeat modules enable nginx подключает готовый набор input, ingest pipeline и дашбордов, но при этом разбор уходит на сторону Elasticsearch. Выбирайте одно из двух, чтобы не получить двойной парсинг.

Проверьте, что индекс появился:

curl -X GET "localhost:9200/_cat/indices?v&h=index,docs.count,store.size"

Сбор системного журнала через journald или syslog

Вариант journald отдаёт структурированные поля сразу: unit, priority, hostname. Отдельный grok для таких событий почти не нужен.

filebeat.inputs:
  - type: journald
    id: journald-systemd
    include_matches:
      - "_SYSTEMD_UNIT=nginx.service"
      - "_SYSTEMD_UNIT=sshd.service"
    fields:
      log_type: journald
    fields_under_root: true

Если хост не использует systemd либо логи уже пишутся в текстовые файлы, берите filestream и разбирайте строку grok-паттерном SYSLOGLINE:

filebeat.inputs:
  - type: filestream
    id: syslog
    enabled: true
    paths:
      - /var/log/syslog
      - /var/log/auth.log
    fields:
      log_type: syslog
    fields_under_root: true

filter {
  if [log_type] == "syslog" {
    grok {
      match => { "message" => "%{SYSLOGLINE}" }
    }
    date {
      match => ["timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss"]
      target => "@timestamp"
    }
  }
}

Диагностика агента:

journalctl -u filebeat -n 50 --no-pager
filebeat test config -c /etc/filebeat/filebeat.yml

Сбор логов контейнеров Docker

filebeat.inputs:
  - type: container
    id: docker-containers
    paths:
      - /var/lib/docker/containers/*/*.log
    processors:
      - add_docker_metadata:
          host: "unix:///var/run/docker.sock"
    fields:
      log_type: docker
    fields_under_root: true

Input type: container читает json-file формат и разбирает его сам: текст сообщения попадает в поле message, поток stdout или stderr в поле stream. Процессор add_docker_metadata добавляет container.id, container.name и метки образа, по ним удобно фильтровать в Kibana. Без него события останутся без привязки к контейнеру.

Если приложение пишет в stdout JSON, включите декодирование в input:

    json.keys_under_root: true
    json.add_error_key: true
    json.message_key: message

Для динамических окружений используйте autodiscover с подсказками в метках контейнера:

filebeat.autodiscover:
  providers:
    - type: docker
      hints.enabled: true

Проверка источника и результата:

docker logs nginx --tail 20
curl -X GET "localhost:9200/_cat/indices?v&h=index,docs.count"

Настройка grok-паттернов для разбора неструктурированных логов

Синтаксис простой: %{PATTERN:field} берёт встроенный или свой паттерн и кладёт результат в поле. Готовые паттерны, которые закрывают большинство задач: COMBINEDAPACHELOG и COMMONAPACHELOG для веб-логов, NGINXACCESS для Nginx, SYSLOGLINE для syslog, HTTPDATE и TIMESTAMP_ISO8601 для дат.

Правила, которые экономят время: разбивайте длинную строку на два-три паттерна вместо одного гигантского, используйте точные шаблоны вместо жадного %{GREEDYDATA}, задавайте тип прямо в паттерне через %{NUMBER:bytes:int}, применяйте dissect там, где порядок полей фиксирован: он работает быстрее grok и не требует регулярок.

Пример grok-паттерна для логов Nginx

filter {
  if [log_type] == "nginx_access" {
    grok {
      match => {
        "message" => "%{IPORHOST:client_ip} %{USER:ident} %{USER:auth} \\[%{HTTPDATE:timestamp}\\] \"%{WORD:http_method} %{DATA:request} HTTP/%{NUMBER:http_version}\" %{NUMBER:http_response_code:int} %{NUMBER:bytes:int} \"%{DATA:referrer}\" \"%{DATA:user_agent}\""
      }
      tag_on_failure => ["_grokparsefailure", "nginx_access_unparsed"]
    }
    date {
      match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
      target => "@timestamp"
    }
    useragent {
      source => "user_agent"
      target => "ua"
    }
    mutate {
      remove_field => ["ident", "auth", "timestamp"]
    }
  }
}

Что получается на выходе: client_ip, http_method, request, http_response_code, bytes, referrer, user_agent, а фильтр useragent дополнительно раскладывает браузер и ОС в поля ua.name, ua.os, ua.device. Для error.log хватит отдельного паттерна: %{HTTPDATE:timestamp} %{LOGLEVEL:log_level} %{POSINT:pid}#%{NUMBER:tid} %{GREEDYDATA:message_text}, где текст ошибки уходит в message_text и не затирает исходное поле.

Отладка. В Kibana есть Grok Debugger, он принимает строку и паттерн и сразу показывает извлечённые поля. Проверяйте паттерн на реальных строках из файла, а не на придуманных: расхождения в кавычках и пробелах ломают разбор чаще всего.

Пример комплексного парсинга логов прикладных сервисов с grok-фильтрами и последующими алертами разобран в руководстве по анализу логов образовательных сервисов.

Обработка ошибок парсинга и тег _grokparsefailure

Когда ни один паттерн не совпал, Logstash добавляет в событие тег _grokparsefailure и пропускает его дальше. Событие не теряется, но остаётся неструктурированным, и по нему не работают поиск по полям и дашборды.

Рабочая схема: отправлять непрошедшие события в отдельный индекс и разбирать их отдельно.

output {
  if "nginx_access_unparsed" in [tags] {
    elasticsearch {
      hosts => ["https://es01.example.net:9200"]
      index => "logs-unparsed-%{+yyyy.MM.dd}"
      user => "logstash_writer"
      password => "${ES_PASSWORD}"
      ssl_enabled => true
      ssl_certificate_authorities => ["/etc/logstash/certs/ca.crt"]
    }
  } else {
    elasticsearch {
      hosts => ["https://es01.example.net:9200"]
      index => "nginx-access-%{+yyyy.MM.dd}"
      user => "logstash_writer"
      password => "${ES_PASSWORD}"
      ssl_enabled => true
      ssl_certificate_authorities => ["/etc/logstash/certs/ca.crt"]
    }
  }
}

Второй уровень страховки: dead letter queue. В logstash.yml включите запись событий, которые вызвали исключение при обработке:

dead_letter_queue.enable: true
path.dead_letter_queue: /var/lib/logstash/dead_letter_queue
dead_letter_queue.max_bytes: 1024mb

Каждая запись в DLQ хранит событие, тип плагина и причину сбоя. Разбирают её отдельным pipeline, чтобы не смешивать с основным потоком:

input {
  dead_letter_queue {
    path => "/var/lib/logstash/dead_letter_queue"
    pipeline_id => "main"
    commit_offsets => true
    tags => ["dlq"]
  }
}

Для точечной обработки ошибок внутри одного фильтра используйте on_failure: он позволяет добавить служебное поле и отправить событие в другой индекс, не включая DLQ для всего pipeline.

Буферизация и поведение Filebeat при недоступности приёмника

Filebeat читает файлы харвестерами, складывает события во внутренний spooler и отправляет их пачками. Каждое событие считается доставленным только после подтверждения от приёмника, после чего позиция фиксируется в registry. Это модель at-least-once: при аварии возможна повторная доставка события, но не потеря.

Очередь настраивается одной из двух схем. queue.mem держит события в памяти: быстро, но при перезапуске агента незафиксированные данные теряются, а размер ограничен flush.min_events и flush.timeout (по умолчанию 2048 событий и 1 секунда). queue.disk пишет события на диск и переживает падение процесса.

Настройка дисковой очереди Filebeat

queue.disk:
  enabled: true
  max_size: 10GB
  path: /var/lib/filebeat/diskqueue

Для продакшена ставьте дисковую очередь: объём 10 ГБ закрывает несколько часов недоступности приёмника на среднем потоке, память при этом не расходуется. Когда очередь заполнится, Filebeat перестаёт читать новые строки, старые события остаются на диске и уходят после восстановления связи. Байты в файлах не теряются: харвестеры ждут снятия backpressure, а позиция в registry не двигается вперёд.

Контроль размера очереди:

du -sh /var/lib/filebeat/diskqueue

Поведение при недоступности Logstash или Elasticsearch

При ошибке отправки Filebeat не роняет события: он повторяет попытку с увеличивающимся интервалом. Управляют этим два параметра: backoff.init (по умолчанию 1s) задаёт первую паузу, backoff.max (по умолчанию 60s) ограничивает максимальную. Дополнительно bulk_max_size определяет, сколько событий уходит в одном запросе, а worker задаёт число одновременных соединений на хост.

output.logstash:
  hosts: ["logstash01.example.net:5044"]
  backoff.init: 1s
  backoff.max: 60s
  bulk_max_size: 1024
  worker: 2

Как найти проблемы с подключением:

journalctl -u filebeat -n 100 --no-pager | grep -i -E "error|fail|refused"

Сценарий проверки надёжности, который стоит прогнать на стенде перед продакшеном.

1. Остановите Logstash: systemctl stop logstash.
2. Сгенерируйте трафик к Nginx или запишите несколько сотен строк в отслеживаемый файл.
3. Проверьте рост очереди: du -sh /var/lib/filebeat/diskqueue.
4. Запустите Logstash: systemctl start logstash, затем посмотрите логи агента на восстановление соединения.
5. Убедитесь, что все события дошли до Elasticsearch: сравните docs.count по индексу до и после.

Если события нужны и в другом хранилище или в Grafana Loki, схему маршрутизации с приоритетами разбирает материал про маршрутизацию логов и метрик в DevOps.

Проверка результата: команды и чек-лист

Проверять конвейер нужно на трёх уровнях: валидность конфигов, состояние очередей и наличие разобранных полей в индексах.

Команды для проверки конфигураций и состояния индексов

filebeat test config -c /etc/filebeat/filebeat.yml
filebeat test output -c /etc/filebeat/filebeat.yml
filebeat export config -c /etc/filebeat/filebeat.yml
/usr/share/logstash/bin/logstash --config.test_and_exit -f /etc/logstash/conf.d/
curl -X GET "localhost:9200/_cat/indices?v&h=index,docs.count,store.size"
curl -X GET "localhost:9200/nginx-access-*/_search?q=*&pretty&size=1"
curl -X GET "localhost:9600/_node/stats?pretty"

Последняя команда возвращает статистику JVM, очередей и счётчики событий Logstash и полезна при поиске узкого места. Для метрик самого Filebeat включите http.enabled: true и опрашивайте эндпоинт localhost:5066/stats.

В Kibana проверка занимает две минуты: создайте data view по шаблону индекса, откройте Discover, задайте фильтр по интересующему полю, например http_response_code: 500 и посмотрите, есть ли документы. Поля из grok появляются в списке доступных полей только после того, как Elasticsearch создаст по ним маппинг, то есть после первого разобранного события.

Чек-лист перед вводом в эксплуатацию

• Версии Filebeat, Logstash и Elasticsearch совместимы между собой.
• Аутентификация включена, пароли хранятся в keystore, а не в конфиге.
• TLS настроен на всех участках: агент, Logstash, Elasticsearch.
• Дисковая очередь Filebeat включена, размер согласован с запасом диска.
• Persistent queue Logstash включена при высоком потоке.
• grok-паттерны проверены на реальных строках, tag_on_failure настроен.
• Непрошедшие события уходят в отдельный индекс, DLQ включена.
• Доставка проверена при остановленном Logstash и остановленном Elasticsearch.
• ILM или ротация индексов настроены, старые данные не растут бесконечно.
• Мониторинг Filebeat и Logstash подключён, алерты по ошибкам доставки заданы.

Политики хранения, ротация индексов и переход данных на холодные уровни разобраны в руководстве по оптимизации хранения и ротации логов.

Безопасность и масштабирование конвейера

Логи содержат имена пользователей, IP-адреса и токены в URL. Передавать их по HTTP без шифрования нельзя, а доступ к индексам стоит ограничивать по ролям.

Настройка TLS и аутентификации

В Elasticsearch 8.x базовые функции безопасности работают из коробки, включая TLS и роли. Для агента создайте отдельного пользователя с ролью, которая разрешает только запись в нужные индексы, либо выпустите API key.

output.elasticsearch:
  hosts: ["https://es01.example.net:9200"]
  api_key: "${ES_API_KEY}"
  ssl:
    enabled: true
    certificate_authorities: ["/etc/filebeat/certs/ca.crt"]
    verification_mode: full

На стороне Logstash TLS включается и на входе, и на выходе: в input beats параметрами ssl_enabled, ssl_certificate, ssl_key и ssl_certificate_authorities, в output elasticsearch параметрами ssl_enabled и ssl_certificate_authorities. Отдельный сертификат для каждого узла упрощает ротацию: при компрометации ключа меняется один узел, а не весь конвейер.

Базовая оптимизация производительности Logstash

Три параметра определяют пропускную способность pipeline в logstash.yml:

Для надёжности включите persistent queue: queue.type: persisted и queue.max_bytes: 1gb. Она сохраняет события на диск между перезапусками и сглаживает пики, когда Elasticsearch отвечает медленно.

Когда одного инстанса перестаёт хватать, запускайте несколько Logstash с одинаковым pipeline и распределяйте нагрузку через балансировщик L4: Filebeat с loadbalance: true сам разложит события по хостам. Парк агентов удобно администрировать через Central Management, тогда конфиг меняется в Kibana, а не на сотне серверов по ssh.

Стартуйте с прямого подключения Filebeat к Elasticsearch, если логи простые, и добавляйте Logstash в тот момент, когда появляется второй приёмник, сложный grok или требование к маршрутизации. Разделение обязанностей дешевле, чем поддержка лишнего компонента с первой недели.