ElasticSearch. Начало работы. Часть 3 из 3

Стек ELK (Elasticsearch, Logstash и Kibana)

Logstash — это конвейер обработки данных. Позволяет собирать данные из различных источников (например, из файлов логов), обрабатывать их (например, разбирать строку лога на отдельные поля) и отправлять в Elasticsearch.

Kibana — это веб-интерфейс для Elasticsearch. Позволяет подключаться к Elasticsearch, чтобы можно было искать, анализировать и, самое главное — визуализировать данные в виде графиков, таблиц и дашбордов.

Установка и настройка Kibana

Устанавливаем зависимости, если их еще нет

$ sudo apt-get install apt-transport-https gnupg

Скачиваем GPG-ключ, конвертируем его из ASCII в бинарный формат, сохраняем в директории /usr/share/keyrings

$ wget -qO - https://artifacts.elastic.co/GPG-KEY-elasticsearch | \
> sudo gpg --dearmor -o /usr/share/keyrings/elastic.gpg

Добавляем репозиторий Elastic, если его еще нет

$ echo "deb [signed-by=/usr/share/keyrings/elastic.gpg] https://artifacts.elastic.co/packages/8.x/apt stable main" | \
> sudo tee /etc/apt/sources.list.d/elastic-8.x.list

Обновляем список пакетов и устанавливаем Kibana

$ sudo apt update && sudo apt install kibana

Запускаем службу и добавляем в автозагрузку

# systemctl start kibana.service
# systemctl enable kibana.service

# systemctl status kibana.service 
● kibana.service - Kibana
     Loaded: loaded (/usr/lib/systemd/system/kibana.service; enabled; preset: enabled)
     Active: active (running) since Fri 2025-09-19 16:09:17 MSK; 7s ago
       Docs: https://www.elastic.co
   Main PID: 13003 (node)
      Tasks: 11 (limit: 18684)
     Memory: 347.3M (peak: 347.7M)
        CPU: 8.551s
     CGroup: /system.slice/kibana.service
             └─13003 /usr/share/kibana/bin/../node/glibc-217/bin/node /usr/share/kibana/bin/../src/cli/dist

... EVG-UBN systemd[1]: Started kibana.service - Kibana.
... EVG-UBN kibana[13003]: Kibana is currently running with legacy OpenSSL providers enabled! For details ...
... EVG-UBN kibana[13003]: {"log.level":"info","@timestamp":"2025-09-19T13:09:18.657Z","log.logger":"elast...
... EVG-UBN kibana[13003]: Native global console methods have been overridden in production environment.
... EVG-UBN kibana[13003]: [2025-09-19T16:09:19.967+03:00][INFO ][root] Kibana is starting
... EVG-UBN kibana[13003]: [2025-09-19T16:09:19.991+03:00][INFO ][node] Kibana process configured with rol...

Файл конфигурации

Рассмотрим основные директивы конфигурации в файле /etc/kibana/kibana.yml

# Порт для входящих соединений
server.port: 5601
# Значение по умолчанию localhost,
# (доступ только с этой машины)
server.host: "0.0.0.0"
# Узлы кластера для подключения
elasticsearch.hosts: ["http://localhost:9200"]
# Отправлять данные мониторинга
monitoring.kibana.collection.enabled: true

# Какие сообщения логировать
logging.root.level: warn
# Логи в json-файл + stdout
logging:
  appenders:
    file:
      type: file
      fileName: /var/log/kibana/kibana.json
      layout:
        type: json
  root:
    appenders:
      - default
      - file

# Отключаем набор инструментов от Elastic для
# создания корпоративных поисковых решений
enterpriseSearch.enabled: false

Директива server.port задает порт, на котором Kibana будет принимать входящие соединения, значение по умолчанию 5601. Обычно нет необходимости менять этот порт, если только он не занят другим приложением или нет специфических требований к сетевой конфигурации.

Директива server.host указывает ip-адрес сетевого интерфейса, который Kibana будет прослушивать, значение по умолчанию localhost. Значение по умолчанию означает, что доступ возможен только с той же машины. Для доступа по сети нужно изменить на ip-адрес сетевого интерфейса машины или на значение 0.0.0.0 (прослушивать все интерфейсы).

Директива server.publicBaseUrl используется, когда Kibana доступна через реверс-прокси или балансировщик нагрузки, и базовый URL, который видят пользователи, отличается от URL, по которому Kibana фактически слушает. Это помогает Kibana правильно генерировать ссылки внутри себя (например, для отчетов или расшаривания дашбордов). При использовании реверс-прокси (например, Nginx или Apache), который перенаправляет запросы на Kibana, нужно установить полный внешний URL, по которому пользователи будут обращаться к Kibana (например, https://domain.com/kibana).

Директива server.name — произвольное имя этого экземпляра Kibana, которое будет отображаться в логах и некоторых UI-элементах, значение по умолчанию — hostname машины. Нужно использовать осмысленное имя, которое поможет дентифицировать данный экземпляр Kibana, особенно если этих экземпляров несколько.

Директива elasticsearch.hosts указывает ip-адреса одного или нескольких узлов Elasticsearch, к которым Kibana должна подключаться, значение по умолчанию localhost:9200. Для высокой доступности и производительности, особенно в кластере, рекомендуется перечислить несколько узлов Elasticsearch в виде массива.

elasticsearch.hosts: ["http://192.168.100.11:9200", "http://192.168.100.12:9200"]

Директивы elasticsearch.username и elasticsearch.password — логин и пароль для аутентификации, если кластер Elasticsearch защищен с помощью X-Pack Security, значения по умолчанию — пусто.

$ sudo nano /etc/elasticsearch/elasticsearch.yml

xpack.security.enabled: true

$ sudo /usr/share/elasticsearch/bin/elasticsearch-setup-passwords interactive

Директива elasticsearch.serviceAccountToken — альтернативный (и более безопасный) способ аутентификации Kibana в Elasticsearch, который используется вместо логина-пароля, когда Security включен. Токен генерируется в Elasticsearch и предоставляет Kibana необходимые разрешения. Значение по умолчанию — пусто.

Директива logging отвечает за логи Kibana — уровень детализации, куда и в каком формате сохранять.

logging:
  appenders: # определяет, куда отправлять логи и в каком формате
    file: # определяем appender с именем file
      type: file        # тип appender — запись в файл
      fileName: /var/log/kibana/kibana.json # имя файла для записи логов
      layout:           # формат логов для этого appender
        type: json      # формат логов — JSON
  root: # корневой logger, который определяет, какие appender-ы будут использоваться
    appenders:
      - default         # ссылается на appender по умолчанию (обычно stdout)
      - file            # ссылается на определенный выше аппендер file

Директива logging.root.level задает минимальный уровень сообщений, которые будут записываться в лог. Все сообщения, чей уровень выше или равен установленному, будут логироваться. Возможные значения (от наименее подробного к наиболее подробному) — fatal, error, warn, info, debug, trace.

logging.root.level: warn

Ротацию логов можно организовать с помощью демона logrotate.service или использовать встроенные возможности Kibana.

logging:
  appenders:
    file_with_rotation:
      type: rolling-file  # запись в файл с ротацией
      fileName: /var/log/kibana/kibana.json
      layout:
        type: json
      policy:
        type: size-limit
        size: 0mb  # ротация при размере 50 Мб
      strategy:
        type: numeric
        max: 10  # хранить 10 файлов логов 
  root:
    appenders:
      - default
      - file_with_rotation

Logger — это источник сообщения. Разные части Kibana (например, плагины, сервер, обработчик HTTP-запросов) имеют свои собственные логгеры. У каждого логгера есть уникальное имя (например, elasticsearch.query, metrics.ops, http.server.response).

Appender — определяет, куда будут отправляться логи. Это может быть консоль, файл, или даже удаленный сервер. Типы аппендеров в Kibana — console (отправляет логи в stdout), file (записывает логи в файл), rolling-file (записывает в файл и выполняет ротацию).

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

logging:
  appenders:
    # Будет выводить важные сообщения в консоль.
    # Использует pattern-формат с подсветкой.
    console_output:
      type: console
      layout:
        type: pattern
        pattern: "[%date{%ISO8601}][%level][%logger] %message"
        highlight: true # включает цветную подсветку для консоли

    # Записывает общие сообщения Kibana (info, warn, error) в файл.
    # Использует ротацию по размеру файла, сохраняет 5 файлов.
    kibana_general_file:
      type: rolling-file
      fileName: /var/log/kibana/kibana-general.log
      policy:
        type: size-limit
        size: 100mb
      strategy:
        type: numeric
        max: 5
      layout:
        type: pattern
        pattern: "[%date{%ISO8601}][%level][%logger] %message"

    # Записывает события безопасности в отдельный json-файл.
    # Использует ротацию по времени, сохраняет 7 файлов.
    kibana_security_file:
      type: rolling-file
      fileName: /var/log/kibana/kibana-security.json
      policy:
        type: time-interval
        interval: 24h
        modulate: true
      strategy:
        type: numeric
        max: 7
      layout:
        type: json

    # Записывает все запросы к Elasticsearch в отдельный json-файл.
    # Может быть объемным, поэтому используется ротация по размеру.
    es_query_json_file:
      type: rolling-file
      fileName: /var/log/kibana/kibana-es-queries.json
      policy:
        type: size-limit
        size: 500mb
      strategy:
        type: numeric
        max: 3
      layout:
        type: json

    # Записывает внутренние метрики работы Kibana в json-файл.
    # Без ротации, поэтому нужно использовать logrotate.service.
    kibana_metrics_json_file:
      type: file
      fileName: /var/log/kibana/kibana-metrics.json
      layout:
        type: json
        
    # Перехватывает логи HTTP-запросов, удаляет конфиденциальную
    # информацию и затем передает их в http_requests_target_file.
    http_requests_rewriter:
      type: rewrite
      appenders: [http_requests_target_file] # куда перенаправлять после перезаписи
      policy:
        type: meta # политика перезаписи для метаданных лога
        mode: update # режим обновления существующих свойств
        properties:
          - path: "http.request.headers.authorization" # свойство, которое нужно изменить
            value: "[REDACTED]" # новое значение
          - path: "http.request.headers.cookie"
            value: "[REDACTED]"

    # Куда отправлять логи после обработки rewrite appender
    http_requests_target_file:
      type: rolling-file
      fileName: /var/log/kibana/kibana-http-requests.json
      policy:
        type: size-limit
        size: 200mb
      strategy:
        type: numeric
        max: 4
      layout:
        type: json

  # Это главный logger, который обрабатывает все сообщения,
  # для которых не настроен более специфичный logger.
  root:
    # Отправлять общие сообщения в консоль и в основной файл Kibana.
    appenders: [console_output, kibana_general_file]
    level: info

  loggers:
    # Логгер для запросов к Elasticsearch, подробный уровень (debug)
    - name: elasticsearch.query
      appenders: [es_query_json_file]
      level: debug

    # Логгер для HTTP-запросов и ответов к Kibana, подробный уровень.
    # Логи сначала проходят через http_requests_rewriter для цензуры.
    - name: http.server.response
      appenders: [http_requests_rewriter]
      level: debug

    # Логгер для метрик работы Kibana (производительность).
    # Логи отправляются в kibana_metrics_json_file.
    - name: metrics.ops
      appenders: [kibana_metrics_json_file]
      level: debug

    # Логгер для подсистемы безопасности Kibana, уровень info.
    # Логи отправляются в kibana_security_file.
    - name: plugins.security
      appenders: [kibana_security_file]
      level: info

    # Пример логгера, который только меняет уровень. Мы хотим, чтобы все, что
    # связано с сервером (кроме HTTP), логировалось только при уровне error.
    # Appenders будут унаследованы от root (console_output, kibana_general_file).
    - name: server
      level: error

Директива конфигурации logging.loggers нужна только в том случае, когда нужно установить другой уровень и/или другой приемник(и) сообщений, потому что по умолчанию level и appenders все логгеры получают от logging.root.

Обзор возможностей Kibana

Kibana обеспечивает удобный досуп к данным Elasticsearch, предоставляет возможности

• Исследование данных — просмотр сырых данных, их фильтрация и анализ
• Визуализация — создание графиков, диаграмм, карт на основе данных
• Панели мониторинга (дашборды) — объединение нескольких визуализаций в единое целое
• Управление стеком ELK — инструменты управления Elasticsearch, Kibana и Logstash

Иконка в левом верхнем углу открывает меню для доступа ко всем инструментам Kibana

• Analytics
  • Discover — поиск, просмотр и фильтрация сырых данных в Elasticsearch
  • Dashboards — создание и просмотр интерактивных панелей с визуализациями
  • Canvas — визуализация данных в виде презентаций или инфографики
  • Maps — работа с геопространственными данными и отображение их на карте
  • Machine Learning — применение алгоритмов машинного обучения для поиска аномалий и прогнозирования
  • Visualize Library — библиотека сохраненных визуализаций, которые можно использовать на дашбордах
• Observability
  • Overview — сводный обзор состояния наблюдаемых систем
  • Alerts — управление оповещениями, сгенерированными из данных
  • SLOs — мониторинг Service Level Objectives (целей уровня обслуживания)
  • Cases — управление инцидентами и расследованиями
  • Logs — централизованный просмотр и анализ логов из различных источников
  • Infrastructure — мониторинг метрик инфраструктуры (серверы, контейнеры)
  • Applications — мониторинг производительности и доступности приложений (APM)
  • Synthetics — мониторинг доступности и производительности с помощью синтетических тестов
  • User Experience — анализ взаимодействия пользователей с приложениями и сайтами
• Security
  • Dashboards — дашборды для мониторинга событий безопасности
  • Rules — управление правилами обнаружения угроз
  • Alerts — просмотр и управление оповещениями безопасности
  • Attack discovery — поиск и анализ потенциальных атак
  • Findings — управление обнаруженными уязвимостями или подозрительными действиями
  • Cases — управление расследованиями инцидентов безопасности
  • Timelines — хронологическое представление событий для расследования
  • Intelligence — анализ угроз и данных киберразведки
  • Explore — инструмент для интерактивного исследования событий безопасности
  • Manage — управление общими настройками безопасности
• Management
  • Dev Tools — консоль для прямого взаимодействия с Elasticsearch через запросы
  • Integrations — управление интеграциями для сбора данных
  • Fleet — централизованное управление агентами Elastic (например, Elastic Agent)
  • Osquery — интерфейс для запросов к данным операционной системы
  • Stack Monitoring — мониторинг здоровья и производительности стека ELK
  • Stack Management — общие настройки для Kibana, Elasticsearch, пользователей, ролей

Визуализация в Kibana

Первый шаг — настройка подключения к данным (Data Views / Наборы данных). Без этого Kibana не будет знать, с какими индексами Elasticsearch работать. Для этого переходим Меню → Management → Stack Management → Kibana → Data Views. Здесь указываем имя набора данных и шаблон имени индекса (например, nginx-log-* или app-data-*), чтобы Kibana мог найти индексы и работать с ним.

Допустим, мы настроили отправку в ElasticSearch логов трех служб — ElasticSearch, Logstash и Kibana (что мы на самом деле сделаем чуть позже)

• data stream logs-elastic_server-test из файла /var/log/elasticsearch/ecommerce_server.json
• data stream logs-elastic_search_slowlog-test из файла /var/log/elasticsearch/ecommerce_index_search_slowlog.json
• data stream logs-elastic_indexing_slowlog-test из файла /var/log/elasticsearch/ecommerce_index_indexing_slowlog.json
• data stream logs-logstash_common-test из файла /var/log/logstash/logstash-json.log
• data stream logs-logstash_slowlog-test из файла /var/log/logstash/logstash-slowlog-json.log
• data stream logs-kibana-test из файла /var/log/kibana/kibana.json

Мы можем создать три набора данных (Data Views) для просмотра логов ElasticSearch, Logstash и Kibana, используя шаблон имени индекса

• Набор данных «Логи Elastic», шаблон имени индекса logs-elastic*
• Набор данных «Логи Logstash», шаблон имени индекса logs-logstash*
• Набор данных «Логи Kibana», шаблон имени индекса logs-kibana*

Или создать шесть наборов данных (Data Views) для просмотра логов ElasticSearch (server, slowlog), Logstash (common, slowlog) и Kibana

• Набор данных «Логи Elastic (server)», шаблон имени индекса logs-elastic_server*
• Набор данных «Логи Elastic (search slowlog)», шаблон имени индекса logs-elastic_search_slowlog*
• Набор данных «Логи Elastic (indexing slowlog)», шаблон имени индекса logs-elastic_indexing_slowlog*
• Набор данных «Логи Logstash (common)», шаблон имени индекса logs-logstash_commom*
• Набор данных «Логи Logstash (slowlog)», шаблон имени индекса logs-logstash_slowlog*
• Набор данных «Логи Kibana», шаблон имени индекса logs-kibana-test

Второй шаг — исследование данных Меню → Analytics → Discover. После создания набора данных — это основное место для просмотра и анализа сырых данных. Здесь будут все документы, соответствующие выбранному набору данных. Можно применять фильтры (по конкретным полям, значениям), использовать строку поиска (Kibana Query Language) для сложных запросов, выбирать временной диапазон для отображения данных.

Кроме фильтрации данных, можно в левой колонке выбрать поля, которые будут показаны в таблице справа (нужно кликнуть плюсик). Потому как полей может быть много, а нам интересны только некоторые из них. После этого нужно сохранить полученный набор данных — кнопка Save вверху справа. Позже можно получить доступ к этому набору данных через Меню → Stack Management → Kibana → Saved Objects. Обратите внимание, что в Saved Objects будут доступны как наборы из Data Views, так и наборы из Discover.

Третий шаг — создание визуализаций (Visualize / Визуализация). Как только разобрались с данными в Discover, можно приступать к их графическому представлению. Этот инструмент позволяет создавать Гистограммы (Bar charts), Круговые диаграммы (Pie charts), Линейные графики (Line charts), Таблицы данных (Data tables), Метрики (Metrics).

Переходим Меню → Analytics → Visualize Library → Create visualization → Lens. Lens — современный и интуитивно понятный инструмент для создания визуализаций, позволяет перетаскивать поля и сразу видеть результат. Достаточно перетащить поле из левой колонки в центральную колонку для построения подходящей визуализации. В правой колонке можно задать дополнительные настройки.

После создания диаграммы, графика, таблицы — нужно сохранить визуализацию в библиотеке с помощью «Add to library». Можно сразу выбрать Dashboard, на котором она будет показываться или создать новый Dashboard — тогда четвертый шаг будет не нужен.

Четвертый шаг — создание дашборда (Dashboards / Панели мониторинга). Дашборд — это место, где мы собираем все визуализации в единый, интерактивный обзор. Просто добавляем ранее созданные визуализации с помощью «Add from library», располагаем их как удобно, и они все взаимодействуют с общими фильтрами и временными диапазонами.

Язык KQL (Kibana Query Language)

Основное место использования KQL-запросов — это строка поиска (search bar) вверху на страницах Discover, Dashboard и Visualize.

1. Свободный текстовый поиск

Самый простой способ — это просто ввести текст. Kibana будет искать это слово или фразу во всех полях документов.

Найти документы, содержащие слово error (кавычки можно опустить)

"error"

Чтобы найти точную фразу — нужно использовать двойные кавычки

"user authentication failed"

2. Поиск по конкретному полю

Это главная сила KQL. Есть возможность указать, в каком именно поле искать значение. Это делает поиск точным и быстрым.

Найти все логи с HTTP-статусом 200

response.status: 200

Найти все логи, полученные с хоста web-server-01

host.name: "web-server-01"

Кавычки здесь не обязательны, но полезны, если значение содержит пробелы или спецсимволы.

3. Логические операторы and, or, not

Найти все ошибки 404 с IP-адреса 192.168.1.100 (оператор and)

response.status: 404 and client.ip: "192.168.1.100"

Найти все логи с уровнем ERROR или WARN (оператор or)

log.level: ERROR or log.level: WARN

Найти все успешные запросы, которые были сделаны не методом GET (оператор not)

response.status: 200 and not http.request.method: GET

4. Группировка с помощью скобок

Найти все неудачные попытки входа пользователей john или jane

(user.name: "john" or user.name: "jane") and event.action: "login_failed"

5. Операторы сравнения

Найти все запросы, размер которых превысил 5000 байт

response.bytes > 5000

Найти все ошибки сервера (статус 500 и выше)

response.status >= 500

6. Wildcards (маски) и проверка существования

Найти все IP-адреса из подсети 192.168.1.x

client.ip: "192.168.1.*"

Найти все документы, где операционная система начинается с Win

host.os.name: "Win*"

Найти все события, в которых есть информация об ошибке (неважно, с каким значением)

error.message: *

Установка и настройка Logstash

Устанавливаем зависимости, если их еще нет

$ sudo apt install apt-transport-https gnupg

Скачиваем GPG-ключ, конвертируем его из ASCII в бинарный формат, сохраняем в директории /usr/share/keyrings

$ wget -qO - https://artifacts.elastic.co/GPG-KEY-elasticsearch | \
> sudo gpg --dearmor -o /usr/share/keyrings/elastic.gpg

Добавляем репозиторий Elastic, если его еще нет

$ echo "deb [signed-by=/usr/share/keyrings/elastic.gpg] https://artifacts.elastic.co/packages/8.x/apt stable main" | \
> sudo tee /etc/apt/sources.list.d/elastic-8.x.list

$ echo "deb [signed-by=/usr/share/keyrings/elastic.gpg] https://mirror.yandex.ru/mirrors/elastic/8/apt stable main" | \
> sudo tee /etc/apt/sources.list.d/elastic-8.x.list

Обновляем список пакетов и устанавливаем Logstash

$ sudo apt update && sudo apt install logstash

Запускаем службу и добавляем в автозагрузку

$ sudo systemctl start logstash.service
$ sudo systemctl enable logstash.service

$ sudo systemctl status logstash.service
● logstash.service - logstash
     Loaded: loaded (/usr/lib/systemd/system/logstash.service; enabled; preset: enabled)
     Active: active (running) since Fri 2025-11-21 12:24:57 MSK; 6min ago
   Main PID: 983 (java)
      Tasks: 39 (limit: 18682)
     Memory: 837.4M (peak: 838.0M)
        CPU: 1min 15.311s
     CGroup: /system.slice/logstash.service
             └─983 /usr/share/logstash/jdk/bin/java -Xms1g -Xmx1g -Djava.awt.headless=true -Dfile.encoding=UTF-8...
..........

Главный файл конфигурации

В первую очередь нужно обратить внимание на производительность конвейера, настройки очереди и мониторинг состояния службы Logstash.

$ sudo nano /etc/logstash/logstash.yml

# Производительность конвейера (Pipeline Settings),
# можно оставить по умолчанию (закомментированными)
# pipeline.workers: 2
# pipeline.batch.size: 125

# Настройки очереди (Queuing Settings)
queue.type: persisted
queue.max_bytes: 2gb

# Автоматическая перезагрузка конфигурации (Config Reloading)
config.reload.automatic: false
#config.reload.interval: 5s

# Включаем отправку метрик о работе самого Logstash в Elasticsearch.
# Будут доступны в Kibana в разделе Management => Stack Monitoring
xpack.monitoring.enabled: true
xpack.monitoring.elasticsearch.hosts: ["http://localhost:9200"]

# Путь к директории для записи логов службы logstash.service
path.logs: /var/log/logstash
# Формат логов, допустимые значения — plain и json (можно не
# указывать, значение по умолчанию — plain)
log.format: plain
# Уровень для записи в лог (по умолчанию info), допустимые
# значения — fatal, error, warn, info, debug, trace
log.level: warn

# Записывать логи каждого конвейера в отдельный файл
pipeline.separate_logs: true

# Пороговые значения времени для операций, при превышении
# которых соответствующее сообщение записывается в slowlog
# с определенным уровнем детализации (level)
slowlog.threshold.warn: 500ms  # (default: -1)
slowlog.threshold.info: 300ms  # (default: -1)
# slowlog.threshold.debug: -1
# slowlog.threshold.trace: -1

Опция pipeline.workers задает количество потоков (workers), которые будут параллельно выполнять этапы фильтрации и вывода (filter и output) в конвейере. Значение по умолчанию — количество ядер CPU на машине. Рекомендуется оставить значение по умолчанию (закомментированным) — Logstash сам определит кол-во ядер и будет использовать их все.

Опция pipeline.batch.size задает кол-во событий, которые Logstash должен накопить от источника (input), прежде чем отправить их единой пачкой (батчем) на обработку в filter и output. Значение по умолчанию — 125. Для начала рекомендуется оставить значение по умолчанию. Увеличение размера пачки повышает пропускную способность (throughput), но также увеличивает потребление памяти и задержку (latency) обработки каждого отдельного события. Для высоконагруженных систем это значение часто поднимают до 1000 или 2000.

Опция queue.type определяет, как Logstash хранит события, которые он получил, но еще не успел отправить в output. Значение по умолчанию — memory (хранить в оперативной памяти — быстро, но возможна потеря данных). Значение persisted предписывает хранить очередь на диске — это немного медленнее, но гарантирует, что события переживут перезапуск или сбой Logstash. После запуска сервис продолжит обработку с того места, где остановился. Рекомендуемое значение — persisted.

Опция queue.max_bytes — при использовании persisted задает максимальный размер очереди на диске (например, 1gb, 2048mb). Это защита от переполнения диска, если принимающий сервис (например, Elasticsearch) недоступен долгое время. Если Logstash обрабатывает 2 Гб логов в час — чтобы пережить 4-часовое отключение Elasticsearch, нужно установить 8gb.

Опция config.reload.automatic — если установлено в true, Logstash будет автоматически проверять файлы конфигурации на изменения и, если они есть, перезагружать конвейер без перезапуска всего сервиса. Значение по умолчанию — false. Для изучения и разработки рекомендуется установить true. В стабильной производственной среде часто оставляют false, чтобы избежать случайной загрузки некорректной конфигурации.

Опция xpack.monitoring.enabled включает или выключает отправку метрик о работе Logstash (количество событий, нагрузка на CPU, использование памяти и т.д.) в кластер Elasticsearch. Значение по умолчанию — false. Рекомендуется установить true — без этого непонятно, что происходит внутри Logstash, все ли в порядке.

$ sudo nano /etc/elasticsearch/elasticsearch.yml

# Включаем сбор и хранение данных о работе ElasticSearch
# и других компонентов ELK (Kibana, Logstash, Beats)
xpack.monitoring.collection.enabled: true

$ sudo systemclt restart elasticsearch.service

Опция xpack.monitoring.elasticsearch.hosts задает, куда отправлять данные — после этого в Kibana в разделе «Stack Monitoring» будет доступны данные мониторинга Logstash.

Опция slowlog.threshold.warn. Если какая-либо фаза обработки события в конвейере Logstash (например, вход, фильтрация или вывод) занимает более 500 миллисекунд, то в logstash-slowlog-plain.log (или json) будет записано сообщение уровня WARN. Это указывает на серьезную задержку, которая может влиять на производительность и требует внимания.

Опция slowlog.threshold.info. Если какая-либо фаза обработки события занимает более 300 миллисекунд, но менее 500 миллисекунд, будет записано сообщение уровня INFO. Это менее критично, чем WARN, но все же может быть полезно для мониторинга производительности и выявления потенциальных узких мест на ранних стадиях.

Конфигурация API мониторинга

У Logstash есть встроенный HTTP API (иногда его называют Monitoring API / Node API), который слушает на порту (по умолчанию 9600) и отдает служебную информацию о Logstash (состояние процесса и версию, загруженные пайплайны и их конфигурации, статистику по входящим и исходящим событиям, очередям, плагинам). Это Это API предназначено для мониторинга (Prometheus, CheckMK, Zabbix, Elasticsearch), health-check-ов (liveness/readiness в Kubernetes), отладки и диагностики (посмотреть, что происходит с пайплайнами).

$ sudo nano /etc/logstash/logstash.yml

# Мониторинг состояния службы Logstash (по умолчанию разрешено)
api.enabled: true

# Какие стетевые интерфейсы прослушивать. Если разрешены SSL и
# Basic Auth, то по умолчанию прослушиваются все итнерфейсы, в
# противном случае прослушивается только localhost.
api.http.host: 127.0.0.1

# Какой порт прослушивать, допускается указать интервал, например
# 9600-9700, по умолчанию 9600.
api.http.port: 9600

# Шифрование трафика, нужно создать файл сертфиката и ключа в
# формате .p12, который должен быть защищены паролем.
api.ssl.enabled: false
# api.ssl.keystore.path: /path/to/keystore.jks
# api.ssl.keystore.password: "y0uRp4$$w0rD"
# api.ssl.supported_protocols: [TLSv1.2,TLSv1.3]

# Защита доступа к API с помощью HTTP Basic Auth. Без SSL это
# лучше не делать в проде — пароль пойдёт в открытом виде.
api.auth.type: none
# api.auth.basic.username: "logstash-user"
# api.auth.basic.password: "Str0ngP@ss"

# Пароль должен содержать минимум 8 символов, хотя бы одну
# цифру, хотя бы одну заглавную букву, хотя бы одну строчную
# букву. Значение WARN — слабый пароль допустим, но в логах
# будет предупреждение. Значение ERROR — Logstash не стартует,
# если пароль слабый.
api.auth.basic.password_policy.mode: ERROR

# Logstash будет добавлять в ответу эту строку, позволяет
# отличать разные окружения — dev, test, stage, prod
# api.environment: "prod"

Примера запросв к API

$ curl http://127.0.0.1:9600/
$ curl http://127.0.0.1:9600/_node
$ curl http://127.0.0.1:9600/_node/stats
$ curl http://127.0.0.1:9600/_node/pipelines

Файл конфигурации Java-машины (JVM)

Управляет настройками Java-машины (JVM), на которой работает Logstash. Эти настройки влияют на производительность и стабильность.

$ sudo nano /etc/logstash/jvm.options

# Минимальный и максимальный размер кучи (Java Heap) 1 Гбайт
-Xms1g
-Xmx1g

Настройка -Xms устанавливает начальный размер оперативной памяти, -Xmx устанавливает максимальный размер. Рекомендуется всегда устанавливать одинаковое значение для этих настроек. Это предотвращает паузы в работе, связанные с изменением размера кучи.

Под стэк ELK не рекомендуется выделять больше 50% от общей оперативной памяти, вторая половина нужна ОС для файлового кэша. Logstash требуется в 2-4 раза меньше оперативной памяти, чем ElasticSearch. Если есть 8 Гб оперативной памяти, то 3 Гбайт следует выделить для ElasticSearch, а 1 Гбайт — для Logstash.

Файл конфигурации конвейеров

В файле конфигурации /etc/logstash/logstash.conf есть опция path.config. Если задать для нее значение /etc/logstash/conf.d/*.conf — то Logstash при старте возьмет все .conf файлы из этой директории, «склеит» их в один большой файл и запустит как один-единственный конвейер. Это устаревший подход, в настоящее время так делать не рекомендуется.

• Все источники, фильтры и выводы смешиваются — логика обработки разных логов будет перемешана
• Ошибка в одной части конфигурации может сломать весь конвейер
• Невозможно выделить разное количество ресурсов (например, workers) для разных задач

В настоящее время рекомендуется использовать отдельный файл конфигурации /etc/logstash/pipelines.yml. Это позволяет определять и запускать несколько независимых конвейеров одновременно. Каждый конвейер описывается отдельно, имеет уникальный идентификатор и свой файл конфигурации.

Представим, что нам нужно обрабатывать два типа логов

• Системные логи (syslog)
• Логи веб-сервера Apache

Редактируем файл конфигурации /etc/logstash/pipelines.yml

$ sudo nano /etc/logstash/pipelines.yml

# Список конвейеров, каждый начинается с дефиса
- pipeline.id: syslog-pipeline
  path.config: "/etc/logstash/conf.d/syslog.conf"
  pipeline.workers: 1
  queue.type: persisted
  
- pipeline.id: apache-pipeline
  path.config: "/etc/logstash/conf.d/apache.conf"
  pipeline.workers: 2
  queue.type: memory

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

$ sudo nano /etc/logstash/conf.d/syslog.conf

# Это конвейер для системных логов
input {
    tcp {
        port => 5140
        host => "0.0.0.0"
        codec => plain
        tags => ["syslog", "prod"]
    }
}

filter {
    # Тут фильтры для разбора лога syslog
    grok {
        match => { "message" => "%{SYSLOGLINE}" }
    }
}

output {
    elasticsearch {
        hosts => ["http://localhost:9200"]
        index => "syslog-%{+yyyy.MM.dd}"
    }
}

Создаем файл конфигурации для логов веб-сервера

$ sudo nano /etc/logstash/conf.d/apache.conf

# Это конвейер для логов Apache
input {
    file {
        path => "/var/log/apache2/access.log"
        # Используется только при первом чтении файла лога, может принимать значение
        # "beginning" или "end" — начинать чтение с начала файла или с конца. Далее
        # чтение начинается с позиции, сохраненной в файле, который задается опцией
        # sincedb_path, например sincedb_path => "/var/lib/logstash/sincedb_apache"
        start_position => "beginning"
    }
}

filter {
    # Тут фильтры для разбора лога Apache
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}" }
    }
}

output {
    elasticsearch {
        hosts => [ "http://localhost:9200" ]
        index => "apache-access-%{+yyyy.MM.dd}"
    }
}

По умолчанию сервис Logstash работает от имени пользователя logstash. Этот пользователь не имеет права читать файлы в директории /var/log, которые принадлежат пользователю root или группе adm. Поэтому нужно добавить пользователя logstash в системную группу adm.

$ sudo usermod -a -G adm logstash

Файл конфигурации ротации логов

Аппендеры (Appenders) определяют, куда будут отправляться логи — например, в консоль или в файл с ротацией.

• Консольные аппендеры — plain_console (вывод в консоль в виде plain text) и json_console (вывод в консоль в json-формате)
• Файловые аппендеры с ротацией — plain_rolling (запись в файл в виде plain text), json_rolling (запись в файл в json-формате), pipeline_routing_appender (для логирования отдельных конвейеров)

# Вывод в консоль в виде plain text
appender.console.type = Console
appender.console.name = plain_console
appender.console.layout.type = PatternLayout
appender.console.layout.pattern = [%d{ISO8601}][%-5p][%-25c]%notEmpty{[%X{pipeline.id}]}%notEmpty{[%X{plugin.id}]} %m%n

# Вывод в консоль в json-формате
appender.json_console.type = Console
appender.json_console.name = json_console
appender.json_console.layout.type = JSONLayout
appender.json_console.layout.compact = true
appender.json_console.layout.eventEol = true

# Запись в файл с ротацией в виде plain text
appender.rolling.type = RollingFile
appender.rolling.name = plain_rolling
appender.rolling.fileName = ${sys:ls.logs}/logstash-plain.log
appender.rolling.filePattern = ${sys:ls.logs}/logstash-plain-%d{yyyy-MM-dd}-%i.log.gz
appender.rolling.policies.type = Policies
# Два условия для ротации — по времени или по размеру
appender.rolling.policies.time.type = TimeBasedTriggeringPolicy
appender.rolling.policies.time.interval = 1  # ротировать каждый день
appender.rolling.policies.time.modulate = true  # всегда в полночь
appender.rolling.layout.type = PatternLayout
appender.rolling.layout.pattern = [%d{ISO8601}][%-5p][%-25c]%notEmpty{[%X{pipeline.id}]}%notEmpty{[%X{plugin.id}]} %m%n
appender.rolling.policies.size.type = SizeBasedTriggeringPolicy
appender.rolling.policies.size.size = 100MB  # ротировать, если размер файла достиг 100 Мбайт
# Условия для удаления старых файлов логов
appender.rolling.strategy.type = DefaultRolloverStrategy
appender.rolling.strategy.max = 30  # хранить не более 30 файлов
appender.rolling.strategy.action.type = Delete
appender.rolling.strategy.action.basepath = ${sys:ls.logs}
appender.rolling.strategy.action.condition.type = IfFileName
appender.rolling.strategy.action.condition.glob = logstash-plain-*
appender.rolling.strategy.action.condition.nested_condition.type = IfLastModified
appender.rolling.strategy.action.condition.nested_condition.age = 7D  # удалять логи старше 7 дней
appender.rolling.avoid_pipelined_filter.type = PipelineRoutingFilter

# Запись в файл с ротацией в json-формате
appender.json_rolling.type = RollingFile
appender.json_rolling.name = json_rolling
appender.json_rolling.fileName = ${sys:ls.logs}/logstash-json.log
appender.json_rolling.filePattern = ${sys:ls.logs}/logstash-json-%d{yyyy-MM-dd}-%i.log.gz
appender.json_rolling.policies.type = Policies
# Два условия для ротации — по времени или по размеру
appender.json_rolling.policies.time.type = TimeBasedTriggeringPolicy
appender.json_rolling.policies.time.interval = 1  # ротировать каждый день
appender.json_rolling.policies.time.modulate = true  # всегда в полночь
appender.json_rolling.layout.type = JSONLayout
appender.json_rolling.layout.compact = true
appender.json_rolling.layout.eventEol = true
appender.json_rolling.policies.size.type = SizeBasedTriggeringPolicy
appender.json_rolling.policies.size.size = 100MB  # ротировать, если размер файла достиг 100 Мбайт
# Условия для удаления старых файлов логов
appender.json_rolling.strategy.type = DefaultRolloverStrategy
appender.json_rolling.strategy.max = 30  # хранить не более 30 файлов
appender.json_rolling.strategy.action.type = Delete
appender.json_rolling.strategy.action.basepath = ${sys:ls.logs}
appender.json_rolling.strategy.action.condition.type = IfFileName
appender.json_rolling.strategy.action.condition.glob = logstash-json-*
appender.json_rolling.strategy.action.condition.nested_condition.type = IfLastModified
appender.json_rolling.strategy.action.condition.nested_condition.age = 7D  # удалять логи старше 7 дней
appender.json_rolling.avoid_pipelined_filter.type = PipelineRoutingFilter

# Запись в файл с ротацией для отдельных конвейеров
appender.routing.type = PipelineRouting
appender.routing.name = pipeline_routing_appender
appender.routing.pipeline.type = RollingFile
appender.routing.pipeline.name = appender-${ctx:pipeline.id}
appender.routing.pipeline.fileName = ${sys:ls.logs}/pipeline_${ctx:pipeline.id}.log
appender.routing.pipeline.filePattern = ${sys:ls.logs}/pipeline_${ctx:pipeline.id}.%i.log.gz
appender.routing.pipeline.layout.type = PatternLayout
appender.routing.pipeline.layout.pattern = [%d{ISO8601}][%-5p][%-25c] %m%n
# Одно условие для ротации — по размеру файла лога
appender.routing.pipeline.policy.type = SizeBasedTriggeringPolicy
appender.routing.pipeline.policy.size = 100MB  # ротировать, если размер файла достиг 100 Мбайт
# Условия для удаления старых файлов логов
appender.routing.pipeline.strategy.type = DefaultRolloverStrategy
appender.routing.pipeline.strategy.max = 30  # хранить не более 30 файлов
appender.routing.pipeline.strategy.action.type = Delete
appender.routing.pipeline.strategy.action.basepath = ${sys:ls.logs}
appender.routing.pipeline.strategy.action.condition.type = IfFileName
appender.routing.pipeline.strategy.action.condition.glob = pipeline_${ctx:pipeline.id}*.log.gz
appender.routing.pipeline.strategy.action.condition.nested_condition.type = IfLastModified
appender.routing.pipeline.strategy.action.condition.nested_condition.age = 7D  # удалять логи старше 7 дней

Логгеры (Loggers) определяют, какие сообщения (по уровню) будут отправляться и к каким аппендерам.

rootLogger.level = ${sys:ls.log.level}
rootLogger.appenderRef.console.ref = ${sys:ls.log.format}_console
rootLogger.appenderRef.rolling.ref = ${sys:ls.log.format}_rolling
rootLogger.appenderRef.routing.ref = pipeline_routing_appender

Переменная ${sys:ls.logs} содержит путь к директории логов, этот путь задается в главном файле конфигурациии с помощью опции path.logs. Переменная ${sys:ls.log.level} — это уровень лога, задается с помощью опции log.level. Переменная ${sys:ls.log.format} — это формат лога, задается с помощью опции log.format.

Логирование медленных операций

Если включено логирование медленных операций в файле /etc/logstash/logstash.yml

# Вывод в консоль в виде plain text
appender.console_slowlog.type = Console
appender.console_slowlog.name = plain_console_slowlog
appender.console_slowlog.layout.type = PatternLayout
appender.console_slowlog.layout.pattern = [%d{ISO8601}][%-5p][%-25c] %m%n

# Вывод в консоль в json-формате
appender.json_console_slowlog.type = Console
appender.json_console_slowlog.name = json_console_slowlog
appender.json_console_slowlog.layout.type = JSONLayout
appender.json_console_slowlog.layout.compact = true
appender.json_console_slowlog.layout.eventEol = true

# Запись в файл с ротацией в виде plain text
appender.rolling_slowlog.type = RollingFile
appender.rolling_slowlog.name = plain_rolling_slowlog
appender.rolling_slowlog.fileName = ${sys:ls.logs}/logstash-slowlog-plain.log
appender.rolling_slowlog.filePattern = ${sys:ls.logs}/logstash-slowlog-plain-%d{yyyy-MM-dd}-%i.log.gz
appender.rolling_slowlog.policies.type = Policies
# Два условия для ротации — по времени или по размеру
appender.rolling_slowlog.policies.time.type = TimeBasedTriggeringPolicy
appender.rolling_slowlog.policies.time.interval = 1  # ротировать каждый день
appender.rolling_slowlog.policies.time.modulate = true  # всегда в полночь
appender.rolling_slowlog.layout.type = PatternLayout
appender.rolling_slowlog.layout.pattern = [%d{ISO8601}][%-5p][%-25c] %m%n
appender.rolling_slowlog.policies.size.type = SizeBasedTriggeringPolicy
appender.rolling_slowlog.policies.size.size = 100MB  # ротировать, если размер файла достиг 100 Мбайт
# Условия для удаления старых файлов логов
appender.rolling_slowlog.strategy.type = DefaultRolloverStrategy
appender.rolling_slowlog.strategy.max = 30  # хранить не более 30 файлов

# Запись в файл с ротацией в json-формате
appender.json_rolling_slowlog.type = RollingFile
appender.json_rolling_slowlog.name = json_rolling_slowlog
appender.json_rolling_slowlog.fileName = ${sys:ls.logs}/logstash-slowlog-json.log
appender.json_rolling_slowlog.filePattern = ${sys:ls.logs}/logstash-slowlog-json-%d{yyyy-MM-dd}-%i.log.gz
appender.json_rolling_slowlog.policies.type = Policies
# Два условия для ротации — по времени или по размеру
appender.json_rolling_slowlog.policies.time.type = TimeBasedTriggeringPolicy
appender.json_rolling_slowlog.policies.time.interval = 1  # ротировать каждый день
appender.json_rolling_slowlog.policies.time.modulate = true  # всегда в полночь
appender.json_rolling_slowlog.layout.type = JSONLayout
appender.json_rolling_slowlog.layout.compact = true
appender.json_rolling_slowlog.layout.eventEol = true
appender.json_rolling_slowlog.policies.size.type = SizeBasedTriggeringPolicy
appender.json_rolling_slowlog.policies.size.size = 100MB  # ротировать, если размер файла достиг 100 Мбайт
# Условия для удаления старых файлов логов
appender.json_rolling_slowlog.strategy.type = DefaultRolloverStrategy
appender.json_rolling_slowlog.strategy.max = 30  # хранить не более 30 файлов

logger.slowlog.name = slowlog
logger.slowlog.level = trace
logger.slowlog.appenderRef.console_slowlog.ref = ${sys:ls.log.format}_console_slowlog
logger.slowlog.appenderRef.rolling_slowlog.ref = ${sys:ls.log.format}_rolling_slowlog
logger.slowlog.additivity = false

Логирование устаревших функций

Файл logstash-deprecation.log предназначен для записи предупреждений об устаревших функциях (deprecation warnings). Это сообщения Logstash генерирует, когда встречает уeстаревшие опции в файле logstash.yml или в файлах в директории conf.d. В моем случае файл содержал следующие записи

$ sudo cat /var/log/logstash/logstash-deprecation.log
[2025-11-25...][WARN][deprecation.logstash.runner] 'pipeline.buffer.type' setting is not explicitly defined. Before
moving to 9.x set it to 'heap' and tune heap size upward, or set it to 'direct' to maintain existing behavior.
[2025-11-25...][WARN][deprecation.logstash.monitoringextension.pipelineregisterhook] Internal collectors option for
Logstash monitoring is deprecated and targeted for removal in the next major version.
Please configure Elastic Agent to monitor Logstash. Documentation can be found at:
https://www.elastic.co/guide/en/logstash/current/monitoring-with-elastic-agent.html

Опция pipeline.buffer.type определяет, где Logstash будет выделять буферы памяти для своих плагинов. Эти буферы используются для временного хранения данных во время обработки.

• Значение heap. Буферы выделяются внутри основной «кучи» JVM (Java Heap), то есть той памяти, которая устанавливается опцией -Xmx в файле jvm.options.
• Значение direct. Буферы выделяются вне основной «кучи» JVM, в так называемой «прямой памяти» (direct memory) или «память вне кучи» (off-heap memory). Это память, которую JVM использует, но которая не входит в лимит -Xmx.

Поскольку опция pipeline.buffer.type явно не определена в logstash.yml, Logstash использует значение по умолчанию direct. Сообщение говорит, что в будущей версии (9.x) это поведение может измениться. Чтобы продолжить использование значения direct — нужно явно задать значение pipeline.buffer.type.

При использовании direct (текущее неявное поведение) — эти буферы выделяются сверх того, что задано в -Xmx. Это может быть быстрее, так как не мешает сборщику мусора, но требует, чтобы операционная система имела достаточно свободной памяти.

appender.deprecation_rolling.type = RollingFile
appender.deprecation_rolling.name = deprecation_plain_rolling
appender.deprecation_rolling.fileName = ${sys:ls.logs}/logstash-deprecation.log
appender.deprecation_rolling.filePattern = ${sys:ls.logs}/logstash-deprecation-%d{yyyy-MM-dd}-%i.log.gz
appender.deprecation_rolling.policies.type = Policies
appender.deprecation_rolling.policies.time.type = TimeBasedTriggeringPolicy
appender.deprecation_rolling.policies.time.interval = 1  # ротировать каждый день
appender.deprecation_rolling.policies.time.modulate = true  # всегда в полночь
appender.deprecation_rolling.layout.type = PatternLayout
appender.deprecation_rolling.layout.pattern = [%d{ISO8601}][%-5p][%-25c]%notEmpty{[%X{pipeline.id}]}%notEmpty{[%X{plugin.id}]} %m%n
appender.deprecation_rolling.policies.size.type = SizeBasedTriggeringPolicy
appender.deprecation_rolling.policies.size.size = 100MB  # ротировать, если размер файла достиг 100 Мбайт
appender.deprecation_rolling.strategy.type = DefaultRolloverStrategy
appender.deprecation_rolling.strategy.max = 30  # хранить не более 30 файлов

logger.deprecation.name = org.logstash.deprecation
logger.deprecation.level = WARN
logger.deprecation.appenderRef.deprecation_rolling.ref = deprecation_plain_rolling
logger.deprecation.additivity = false

logger.deprecation_root.name = deprecation
logger.deprecation_root.level = WARN
logger.deprecation_root.appenderRef.deprecation_rolling.ref = deprecation_plain_rolling
logger.deprecation_root.additivity = false

Конвейер обработки событий Logstash

В контексте Logstash «событие» (event) — это одна логическая запись данных, с которой работает конвейер. Упрощённо — «одна строка лога → одно событие» (но не всегда только строка). Можно думать, что событие — это структура типа «JSON-объект», у которой есть поля и значения.

{
    "@timestamp": "2025-12-01T10:00:00Z",
    "message": "Oct 11 22:14:15 myhost sshd[1234]: Failed password for root from 1.2.3.4 port 22 ssh2",
    "host_name": "myhost",
    "process_name": "sshd",
    "process_pid": "1234",
    "syslog_severity_name": "error",
    "tags": ["syslog", "prod"]
}

Плагин input (например, file, syslog, beats) читает данные — строку лога, сообщение от Beats, запись из Kafka и т.п. На основе этого создаёт событие — объект с полями (минимум @timestamp и message, плюс тех.поля типа host и т.д.).

Плагин filter (например, grok, mutate, date) берёт событие, изменяет/дополняет его поля и отдаёт дальше. Пример — grok из поля message вырезает дату, хост, процесс и записывает в новые поля события.

Плагин output (например, elasticsearch, file, stdout) просто берёт готовое событие и отправляет его куда нужно.

input {
  # Плагины ввода данных
}

filter {
  # Плагины обработки данных
}

output {
  # Плагины вывода данных
}

1. Плагины для секции input

Плагин stdin читает данные со стандартного ввода, полезен для быстрого тестирования конфигураций.

input {
    stdin {}
}

Плагин file читает данные из файлов, идеально подходит для чтения существующих файлов логов.

input {
    file {
        path => "/var/log/syslog"  # Путь к файлу или маска (например, "/var/log/*.log")
        start_position => "beginning" # Читать с начала файла при первом запуске
        sincedb_path => "/dev/null" # Отключает отслеживание прочитанных позиций (для тестирования)
    }
}

Плагин beats слушает входящие соединения от Filebeat и Metricbeat — это наиболее распространенный способ получения логов от агентов.

input {
    beats {
        port => 5044 # Порт, на котором Logstash будет слушать Beats
    }
}

Плагин syslog слушает указанный UDP/TCP порт, парсит syslog‑заголовок (timestamp, host, program, pid, pri и т.п.), кладёт эти данные в поля события (обычно host, program, pid, syslog_pri и т.д.).

input {
    syslog {
        port => 514
        host => "192.168.100.5"
        codec => "plain"
    }
}

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

input {
    generator {
        count => 10 # Количество генерируемых событий
        message => "Тестовое сообщение из Logstash генератора. Поле 1: %{sequence}, Поле 2: %{random_string}"
    }
}

Кроме перечисленных выше, существует множество других — tcp, udp, pipe, http, imap, kafka, redis и т.д.

2. Плагины для секции filter

Плагин grok парсит неструктурированные лог-строки в структурированные поля, используя регулярные выражения.

filter {
    grok {
        match => { "message" => "%{IP:client_ip} %{WORD:method} %{URIPATH:request_path} %{NUMBER:response_code}" }
    }
}

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

filter {
    mutate {
        add_field => { "env" => "dev" }
        remove_field => [ "original_message" ]
        rename => { "response_code" => "http_status" }
        convert => { "http_status" => "integer" }
    }
}

Плагин date парсит строки даты/времени из логов и использует их для установки поля @timestamp.

filter {
    date {
        match => [ "log_timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
        target => "@timestamp"
    }
}

Кроме перечисленных выше, существует множество других — geoip, http, json, json_encode, split, syslog_pri и т.д.

3. Плагины для секции output

Плагин elasticsearch отправляет обработанные события в ElasticSearch, это основной выходной плагин для ELK стека.

output {
    elasticsearch {
        hosts => ["http://localhost:9200"] # адрес ElasticSearch
        user => "elastic"  # если ElasticSearch защищен логином-паролем
        password => "qwerty"  # если ElasticSearch защищен логином-паролем
    }
}

Плагин stdout выводит обработанные события на стандартный вывод (консоль), полезен для отладки конфигураций.

output {
    stdout {
        codec => rubydebug  # форматирует вывод для удобства чтения
    }
}

Плагин file записывает обработанные события в указанный файл.

output {
    file {
        path => "/var/log/logstash/nginx-backend-errors.log"
    }
}

Кроме перечисленных выше, существует множество других — tcp, udp, pipe, http, redis, syslog, email и т.д.

Пример составления конвейера (pipeline)

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

$ sudo systemctl stop logstash.service

Создаем файл системного лога

$ mkdir /home/evgeniy/logstash
$ tail /var/log/syslog > /home/evgeniy/logstash/system-sample.log
$ cat /home/evgeniy/logstash/system-sample.log
2025-12-01T17:30:52.577667+03:00 EVG-UBN anacron[5041]: Normal exit (0 jobs run)
2025-12-01T17:30:52.579570+03:00 EVG-UBN systemd[1]: anacron.service: Deactivated successfully.
2025-12-01T17:35:01.094499+03:00 EVG-UBN CRON[5048]: (root) CMD (command -v debian-sa1 > /dev/null && debian-sa1 1 1)
2025-12-01T17:39:01.103761+03:00 EVG-UBN CRON[5124]: (root) CMD (  [ -x /usr/lib/php/sessionclean ] && if [ ! -d /run/...
2025-12-01T17:39:02.626571+03:00 EVG-UBN systemd[1]: Starting phpsessionclean.service - Clean php session files...
2025-12-01T17:39:02.710808+03:00 EVG-UBN systemd[1]: phpsessionclean.service: Deactivated successfully.
2025-12-01T17:39:02.711005+03:00 EVG-UBN systemd[1]: Finished phpsessionclean.service - Clean php session files.
2025-12-01T17:40:02.629163+03:00 EVG-UBN systemd[1]: Starting sysstat-collect.service - system activity accounting tool...
2025-12-01T17:40:02.638010+03:00 EVG-UBN systemd[1]: sysstat-collect.service: Deactivated successfully.
2025-12-01T17:40:02.638393+03:00 EVG-UBN systemd[1]: Finished sysstat-collect.service - system activity accounting tool.

Создаем файл конфигурации pipeline

$ nano /home/evgeniy/logstash/system-sample.conf

input {
    file {
        path => "/home/evgeniy/logstash/system-sample.log"
        start_position => "beginning"  # читать файл с начала, а не с конца
        sincedb_path => "/dev/null"  # забывать, что файл прочитан, начинать сначала
        ecs_compatibility => "disabled"  # старый формат события
    }
}

filter {
    # Пока никаких фильтров, пропускаем событие дальше
}

output {
    stdout {
        # Печатать каждое событие в консоль в виде «ruby‑хеша»
        # (по сути, это JSON‑подобная структура)
        codec => rubydebug
    }
}

Теперь запускаем Logstash с этим конфигом (без сервиса, вручную)

$ sudo /usr/share/logstash/bin/logstash --path.settings /etc/logstash -f /home/evgeniy/logstash/system-sample.conf

{
    "message" => "2025-12-01T17:30:52.577667+03:00 EVG-UBN anacron[5041]: Normal exit (0 jobs run)",
    "@timestamp" => 2025-12-03T14:59:43.015578382Z,
    "host" => "EVG-UBN",
    "path" => "/home/evgeniy/logstash/system-sample.log",
    "event" => {
        "original" => "2025-12-01T17:30:52.577667+03:00 EVG-UBN anacron[5041]: Normal exit (0 jobs run)"
    },
    "@version" => "1"
}
{
    "message" => "2025-12-01T17:30:52.579570+03:00 EVG-UBN systemd[1]: anacron.service: Deactivated successfully.",
    "@timestamp" => 2025-12-03T14:59:43.060390381Z,
    "host" => "EVG-UBN",
    "path" => "/home/evgeniy/logstash/system-sample.log",
    "event" => {
        "original" => "2025-12-01T17:30:52.579570+03:00 EVG-UBN systemd[1]: anacron.service: Deactivated successfully."
    },
    "@version" => "1"
}
..........

Для каждого события мы видим поля message (вся строка лога целиком), @timestamp (время поступления события в Logstash), host (имя машины), path (путь к файлу лога) и другие.

Редактируем файл конфигурации pipeline

$ nano /home/evgeniy/logstash/system-sample.conf

input {
    file {
        path => "/home/evgeniy/logstash/system-sample.log"
        start_position => "beginning"  # читать файл с начала, а не с конца
        sincedb_path => "/dev/null"  # забывать, что файл прочитан, начинать сначала
        ecs_compatibility => "disabled"  # старый формат события
    }
}

filter {
    grok {
        match => {
            "message" => "%{TIMESTAMP_ISO8601:timestamp_from_message} %{HOSTNAME} %{WORD:process_name}(?:\[%{POSINT:process_pid}\])?: %{GREEDYDATA:text_of_message}"
        }
    }

    date {
        # В поле @timestamp записываем время события из файла лога.
        # Это настоящее время, когда произошло событие, изначально
        # logstash.service записывает туда время, когда прочитана
        # строка из файла лога и создано событие.
        match  => [ "timestamp_from_message", "ISO8601" ]
        target => "@timestamp"
    }

    mutate {
        # Удаляем временное поле, оно нам больше не нужно.
        # Также удаляем поле event, оно дублирует message.
        remove_field => [ "timestamp_from_message", "event" ]
        # Изменяем имя поля host на новое имя host_name
        rename => { "host" => "host_name" }
    }
}

output {
    stdout {
        # Печатать каждое событие в консоль в виде «ruby‑хеша»
        # (по сути, это JSON‑подобная структура)
        codec => rubydebug
    }
}

Опять запускаем Logstash с этим конфигом

$ sudo /usr/share/logstash/bin/logstash --path.settings /etc/logstash -f /home/evgeniy/logstash/system-sample.conf

{
    "message" => "2025-12-01T17:30:52.577667+03:00 EVG-UBN anacron[5041]: Normal exit (0 jobs run)",
    "@timestamp" => 2025-12-01T14:30:52.577Z,
    "host_name" => "EVG-UBN",
    "path" => "/home/evgeniy/logstash/system-sample.log",
    "@version" => "1",
    "process_name" => "anacron",
    "process_pid" => "5041",
    "text_of_message" => "Normal exit (0 jobs run)"
}
{
    "message" => "2025-12-01T17:30:52.579570+03:00 EVG-UBN systemd[1]: anacron.service: Deactivated successfully.",
    "@timestamp" => 2025-12-01T14:30:52.579Z,
    "host_name" => "EVG-UBN",
    "path" => "/home/evgeniy/logstash/system-sample.log",
    "@version" => "1",
    "process_name" => "systemd",
    "process_pid" => "1",
    "text_of_message" => "anacron.service: Deactivated successfully."
}

Теперь у нас есть дополнительные поля, которые мы извлекли из поля message — process_name, process_pid, text_of_message. И мы записали в поле @timestamp дату и время события из лога вместо даты и времени, когда событие пришло в Logstash.

$ sudo systemctl start logstash.service

Именованные шаблоны для плагина grok

Посмотреть шаблоны можно в репозитриии Elastic на GitHub https://github.com/logstash-plugins/logstash-patterns-core/tree/main/patterns. Устаревшие шаблоны расположены в legacy, новые шаблоны расположены в ecs-v1. Рекомендуется использовать новые, соответствующие спецификации Elastic Common Schema (ECS).

$ ls -la /usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/logstash-patterns-core-4.3.4/patterns/ecs-v1
итого 136
drwxr-xr-x 2 root root  4096 ноя  3 14:00 .
drwxr-xr-x 4 root root  4096 ноя  3 14:00 ..
-rw-r--r-- 1 root root  4741 окт 22 03:36 aws
-rw-r--r-- 1 root root  5852 окт 22 03:36 bacula
-rw-r--r-- 1 root root   977 окт 22 03:36 bind
-rw-r--r-- 1 root root  4612 окт 22 03:36 bro
-rw-r--r-- 1 root root  1849 окт 22 03:36 exim
-rw-r--r-- 1 root root 15471 окт 22 03:36 firewalls
-rw-r--r-- 1 root root  5514 окт 22 03:36 grok-patterns
-rw-r--r-- 1 root root  4015 окт 22 03:36 haproxy
-rw-r--r-- 1 root root  1411 окт 22 03:36 httpd
-rw-r--r-- 1 root root  2223 окт 22 03:36 java
-rw-r--r-- 1 root root  1893 окт 22 03:36 junos
-rw-r--r-- 1 root root  1143 окт 22 03:36 linux-syslog
-rw-r--r-- 1 root root    74 окт 22 03:36 maven
-rw-r--r-- 1 root root   206 окт 22 03:36 mcollective
-rw-r--r-- 1 root root   842 окт 22 03:36 mongodb
-rw-r--r-- 1 root root 10242 окт 22 03:36 nagios
-rw-r--r-- 1 root root   198 окт 22 03:36 postgresql
-rw-r--r-- 1 root root  1137 окт 22 03:36 rails
-rw-r--r-- 1 root root   302 окт 22 03:36 redis
-rw-r--r-- 1 root root   214 окт 22 03:36 ruby
-rw-r--r-- 1 root root   637 окт 22 03:36 squid
-rw-r--r-- 1 root root  5366 окт 22 03:36 zeek

$ cat /usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/logstash-patterns-core-4.3.4/patterns/ecs-v1/grok-patterns

USERNAME [a-zA-Z0-9._-]+
USER %{USERNAME}
EMAILLOCALPART [a-zA-Z0-9!#$%&'*+\-/=?^_`{|}~]{1,64}(?:\.[a-zA-Z0-9!#$%&'*+\-/=?^_`{|}~]{1,62}){0,63}
EMAILADDRESS %{EMAILLOCALPART}@%{HOSTNAME}
INT (?:[+-]?(?:[0-9]+))
..........

$ cat /usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/logstash-patterns-core-4.3.4/patterns/ecs-v1/httpd

HTTPDUSER %{EMAILADDRESS}|%{USER}
HTTPDERROR_DATE %{DAY} %{MONTH} %{MONTHDAY} %{TIME} %{YEAR}
# Log formats
HTTPD_COMMONLOG %{IPORHOST:[source][address]} (?:-|%{HTTPDUSER:[apache][access][user][identity]}) (?:-|%{HTTPDUSER:[user][name]}) \[%{HTTPDATE:timestamp}\] "(?:%{WORD:[http][request][method]} %{NOTSPACE:[url][original]}(?: HTTP/%{NUMBER:[http][version]})?|%{DATA})" (?:-|%{INT:[http][response][status_code]:int}) (?:-|%{INT:[http][response][body][bytes]:int})
# :long - %{INT:[http][response][body][bytes]:int}
HTTPD_COMBINEDLOG %{HTTPD_COMMONLOG} "(?:-|%{DATA:[http][request][referrer]})" "(?:-|%{DATA:[user_agent][original]})"
..........

Отправка системных логов в ElasticSearch

Давайте создадим конвейер для обработки и отправки системных логов в ElasticSearch и посмотрим на эти логи в интерфейсе Kibana. Мы уже видели, как Logstash создает событие и какой формат у этого события.

{
    "message" => "2025-12-01T17:30:52.577667+03:00 EVG-UBN anacron[5041]: Normal exit (0 jobs run)",
    "@timestamp" => 2025-12-03T14:59:43.015578382Z,
    "host" => "EVG-UBN",
    "path" => "/home/evgeniy/logstash/system-sample.log",
    "event" => {
        "original" => "2025-12-01T17:30:52.577667+03:00 EVG-UBN anacron[5041]: Normal exit (0 jobs run)"
    },
    "@version" => "1"
}

Создаем ILM политику для системных логов

PUT http://localhost:9200/_ilm/policy/syslog-policy

{
    "policy": {
        "phases": {
            "hot": {
                "actions": {
                    "rollover": {
                        "max_age": "1d",
                        "max_primary_shard_size": "50gb"
                    }
                }
            },
            "delete": {
                "min_age": "30d",
                "actions": {
                    "delete": {}
                }
            }
        }
    }
}

Создаем шаблон индекса для системных логов

PUT http://localhost:9200/_index_template/syslog-template

{
    "index_patterns": ["syslog-0*"],
    "template": {
        "settings": {
            "index.lifecycle.name": "syslog-policy",
            "index.lifecycle.rollover_alias": "syslog-write",
            "number_of_shards": 1,
            "number_of_replicas": 0
        },
        "mappings": {
            "dynamic": "strict",
            "properties": {
                "@timestamp": { "type": "date" },
                "host": { "type": "keyword" },
                "path": { "type": "keyword" },
                "program": { "type": "keyword" },
                "proc_id": { "type": "integer" },
                "message": { "type": "text" },
                "original": { "type": "text", "index": false },
                "tags": { "type": "keyword" }
            }
        },
        "aliases": {
            "syslog-read": {
                "is_write_index": false
            },
            "syslog-write": {
                "is_write_index": true
            }
        }
    }
}

Создаем начальный индекс для системных логов

PUT http://localhost:9200/syslog-000001

Если все прошло успешно — ответ будет таким

{
    "acknowledged": true,
    "shards_acknowledged": true,
    "index": "syslog-000001"
}

День 1:
  syslog-000001 [ЗАПИСЬ] ← syslog-write
  syslog-000001 [ЧТЕНИЕ] ← syslog-read

День 2 (ролловер):
  syslog-000001 [ЧТЕНИЕ] ← syslog-read
  syslog-000002 [ЗАПИСЬ] ← syslog-write
  syslog-000002 [ЧТЕНИЕ] ← syslog-read

День 3 (ролловер):
  syslog-000001 [ЧТЕНИЕ] ← syslog-read
  syslog-000002 [ЧТЕНИЕ] ← syslog-read
  syslog-000003 [ЗАПИСЬ] ← syslog-write
  syslog-000003 [ЧТЕНИЕ] ← syslog-read

.......

День 30 (удаление):
  syslog-000001 удалён ILM политикой
  syslog-000002 [ЧТЕНИЕ] ← syslog-read
  syslog-000003 [ЧТЕНИЕ] ← syslog-read
  syslog-000004 [ЧТЕНИЕ] ← syslog-read
  ..........
  syslog-000030 [ЗАПИСЬ] ← syslog-write
  syslog-000030 [ЧТЕНИЕ] ← syslog-read

Редактируем файл конфигурации и создаем pipeline

$ sudo nano /etc/logstash/pipelines.yml

# Список конвейеров, каждый начинается с дефиса
- pipeline.id: syslog-pipeline
  path.config: "/etc/logstash/conf.d/syslog.conf"
  pipeline.workers: 1
  queue.type: persisted

$ sudo nano /etc/logstash/conf.d/syslog.conf

input {
    file {
        path => "/var/log/syslog"
        start_position => "beginning"
        sincedb_path => "/var/lib/logstash/sincedb_syslog"
        codec => "plain"
        ecs_compatibility => "disabled"
    }
}

filter {
    # Пример строки системного лога
    # 2025-12-14T15:35:00.538370+03:00 EVG-UBN systemd[1]: Started anacron.service - Run anacron jobs.
    grok {
        ecs_compatibility => "disabled"
        match => {
            "message" => "%{TIMESTAMP_ISO8601:datetime} %{HOSTNAME} %{DATA:program}(?:\[%{NUMBER:proc_id:int}\])?: %{GREEDYDATA:message}"
        }
        # Разрешаем grok перезаписать поле message
        overwrite => ["message"]
    }

    # Записываем время из лог-файла в @timestamp
    date {
        ecs_compatibility => "disabled"
        match => ["datetime", "ISO8601"]
        target => "@timestamp"
    }

    mutate {
        # Сохраняем оригинальную строку в поле original
        rename => {
            "[event][original]" => "original"
        }
        # Удаляем поля, которые не нужно записывать в индекс
        remove_field => ["@version", "event", "datetime"]
    }
}

output {
    elasticsearch {
        hosts => ["http://localhost:9200"]

        # Указываем алиас для записи, а не имя индекса напрямую.
        # Именно через этот алиас ILM управляет тем, в какой
        # физический индекс (syslog-000001, syslog-000002...)
        # будут попадать документы
        index => "syslog-write"
        
        # Отключаем управление ILM со стороны Logstash, потому
        # что мы создали политику и шаблон вручную. Если оставить
        # true (по умолчанию) — Logstash попытается создать свои
        # шаблон и политику, что перезапишет наши настройки.
        ilm_enabled => false
        
        # Запрещаем Logstash создавать свой шаблона для логов
        manage_template => false

        # Работаем с логами по старинке, не используем ECS
        ecs_compatibility => "disabled"
    }
}

$ sudo systemctl restart logstash.service

Тут важно отметить, что если не создавать политику и шаблон — Logstash сам создаст политику logstash-policy и шаблон logstash.

// Политика по умолчанию logstash-policy
{
    "policy": {
        "phases": {
            "hot": {
                "actions": {
                    "rollover": {
                        "max_age": "30d",
                        "max_size": "50gb"
                    }
                }
            },
            "delete": {
                "min_age": "never"
            }
        }
    }
}

// Шаблон индекса по умолчанию logstash
{
    "index_patterns": "logstash-*",
    "priority": 200,
    "template": {
        "settings": {
            "index.refresh_interval": "5s",
            "number_of_shards": 1
        },
        "mappings": {
            "dynamic_templates": [
                {
                    "message_field": {
                        "path_match": "message",           // только поле с именем message
                        "match_mapping_type": "string",    // если это является строкой
                        "mapping": {
                            "type": "text",                // → сохранить как text
                            "norms": false                 // не хранить norms (экономия места)
                        }
                    }
                },
                {
                    "string_fields": {
                        "match": "*",                      // любое поле
                        "match_mapping_type": "string",    // если это строка
                        "mapping": {
                            "type": "text",                // → сохранить как text (для поиска)
                            "fields": {
                                "keyword": {               // + копия как keyword (для агрегаций)
                                    "type": "keyword",
                                    "ignore_above": 256    // обрезать строки длиннее 256 символов
                                }
                            }
                        }
                    }
                }
            ],
            "properties": {
                "@timestamp": { "type": "date" },    // всегда дата
                "@version":   { "type": "keyword" }, // всегда keyword
                "geoip": {                           // geo-данные если есть
                    "ip":        { "type": "ip" },
                    "location":  { "type": "geo_point" },
                    "latitude":  { "type": "half_float" },
                    "longitude": { "type": "half_float" }
                }
            }
        }
    }
}

# Обычный индекс + ILM политика, всё создаем вручную
output {
    elasticsearch {
        hosts => ["http://localhost:9200"]
        index => "syslog-write"   # алиас для записи логов
        ilm_enabled => false      # Logstash не управляет ILM
        manage_template => false  # Logstash не трогает шаблоны
    }
}

# Обычный индекс + ILM политика, всё доверяем Logstash
output {
    elasticsearch {
        hosts => ["http://localhost:9200"]
        ilm_enabled => true
        # Logstash создаст политику «logstash-policy»
        # Logstash создаст шаблон «logstash»
        # Logstash создаст индекс «logstash-000001»
        # Logstash создаст алиас «logstash»
    }
}

# Обычный индекс + ILM политика, политику создаём сами, остальное доверяем Logstash
output {
    elasticsearch {
        hosts => ["http://localhost:9200"]
        ilm_enabled => true
        ilm_policy => "syslog-policy"       # политику создали вручную
        ilm_rollover_alias => "syslog"      # Logstash создаст алиас
        ilm_pattern => "000001"             # Logstash создаст syslog-000001
        manage_template => true             # Logstash создаст шаблон
        template_name => "syslog-template"  # имя для этого шаблона
        template_overwrite => false
        ecs_compatibility => "disabled"
    }
}

# Data Stream + ECS + системные (шаблон logs и политика logs@lifecycle), всё доверяем Logstash
output {
    elasticsearch {
        hosts => ["http://localhost:9200"]
        data_stream => "true"
        # Это не обычный индекс, а специальный индекс для логов (data stream)
        data_stream => "true"
        # Logstash проверит поля [data_stream][type], [data_stream][dataset],
        # [data_stream][namespace] и сам сформирует имя data stream, например
        # logs-generic-production
        data_stream_auto_routing => true
        # Если этих полей не будет — имя будет сформировано из значений ниже
        data_stream_type => "logs"          # это значение по умолчанию
        data_stream_dataset => "generic"    # это значение по умолчанию
        data_stream_namespace => "default"  # это значение по умолчанию
        ecs_compatibility => "v8"
    }
}

Для удобного просмотра нужно изменить настройку часового пояса — Меню → Management → Stack Management → Kibana → Advanced Settings → Timezone → Europe/Moscow. Elasticsearch хранит дату-время в UTC, но просматривать удобнее в своем часовом поясе, чтобы не гадать — событие было только что или три часа назад.

Можно дополнительно установить формат показа дата-время на более привычный для России — DD.MM.YYYY HH:mm:ss (может потребоваться Hard refresh страницы просмотра логов — Ctrl+Shift+R или Ctrl+F5).

• DD — день с ведущим нулём
• MM — месяц с ведущим нулём
• YYYY — год четырьмя цифрами
• HH — часы, 24-часовой формат
• mm — минуты
• ss — секунды

Первый шаг — настройка подключения к данным (Data Views / Наборы данных). Для этого переходим Меню → Management → Stack Management → Kibana → Data Views. Здесь указываем имя набора данных Syslog и шаблон syslog-*, чтобы Kibana мог найти все индексы и работать с ними как с единым целым.

Второй шаг — исследование данных Меню → Analytics → Discover. Здесь будут все документы, соответствующие выбранному набору данных. Можно применять фильтры (по конкретным полям, значениям), использовать строку поиска (Kibana Query Language) для сложных запросов, выбирать временной диапазон для отображения данных. Нужные поля из левой колонки перетаскиваем в таблицу справа для удобного просмотра. Когда все настроено — сохраняем этот набор данных. В дальнейшем доступ будет возможен через Меню → Stack Management → Kibana → Saved Objects.

Давайте отфильтруем документы с ошибками с использованием языка KQL и будем отправлять сообщение, когда ошибок много. Для этого в правом верхнем углу нажимаем ссылку Alerts и выбираем «Create search threshold rule».

message: "error" and program: "systemd"

Elastic каждые N минут запускает проверку и считает — сколько документов, созданных за последние 15 минут (time window), совпало с условием KQL-запроса. Если таких документов больше 10 — создается Alert со статусом Active. Внизу есть checkbox «Exclude matches from previous runs» — не учитывать повторно документы, которые уже совпали при прошлом запуске. Если окно больше интервала запуска (например окно 15 минут, запуск каждые 5 минут) — без checkbox одни и те же документы попадут в несколько запусков.

Alert when matches are found during the latest query run.

Select a data view
DATA VIEW Syslog

Define your query
message: "error" and program: "systemd"

Set the group, threshold, and time window
WHEN count()
OVER all documents
IS ABOVE  10
FOR THE LAST 15 minutes

Set the number of documents to send
3

Exclude matches from previous runs

Возможна группировка по какому-то полю документа — для этого нужно изменить условие OVER all documents на GROUPED OVER top 3 program. Здесь top 3 означает — взять 3 самых часто встречающихся значений поля program и для каждого из них посчитать отдельно.

• Rule schedule — как часто запускать проверку созданного нами правила (rule). Чем чаще — тем быстрее реакция, но больше нагрузка.
• Alert delay — защита от случайных срабатываний. Значение 2 означает, что alert будет создан, если правило сработало 2 раза подряд.
• Alert flapping detection — допустим, условие постоянно то выполняется, то не выполняется. Алерт прыгает между Active и Recovered. Если алерт сменил статус 4 раза за последние 20 проверок — считать его flapping и подавить лишние уведомления. Полезно когда метрика нестабильна и болтается около порога срабатывания.

На следующем этапе создания правила для алерта меня ожидал сюрприз — без лицензии отправить сообщение нельзя. То есть, алерт создается, его можно посмотреть в интерфейсе Kibana, но сообщить на почту или в телеграм — нельзя. Так что переходим к последнему этапу, где задаем имя для этого правила — «Systemd errors». Дополнительно можно добавить тег и написать рекомендации для получателя — что делать при получении сообщения.

Посмотреть созданные правила можно в Меню → Management → Stack Management → Alerts and Insights → Rules, посмотреть алерты можно в Меню → Management → Stack Management → Alerts and Insights → Alerts. Алерты можно фильтровать по статусу, правилу, группе и тегу. Статус Active означает, что правило сработало. Статус Recovered означает, алерт был активным, но условие больше не выполняется.

ECS (Elastic Common Schema)

ECS (Elastic Common Schema) — это единый «словарь» имён полей и типов данных для событий в Elastic. Цель — чтобы логи/метрики/трейсы из разных источников выглядели одинаково, а Kibana (Discover, Lens, Alerting) могла «понимать» их без кастомной логики — одинаковые фильтры, готовые визуализации, единые поля уровня, сервиса, хоста и т.д.

Ниже — на какие ECS-поля стоит обратить внимание в первую очередь для логов.

1. Базовые поля события (must-have)

• @timestamp — время события, это главный time-field для data view и графиков
• message — короткое человекочитаемое сообщение, даже при работе с JSON
• event.original — оригинальная «сырая» строка (помогает при отладке парсинга)

2. Идентификация источника — сервис/хост/окружение

• service.name — ключевое поле для фильтров и дашбордов (nginx, redis)
• service.version — полезно для анализа проблем после обновлений (если доступно)
• host.name — на каком хосте произошло событие (важно даже на стенде, в проде — критично)
• agent.type, agent.name, agent.version — помогает понять, какой beats/agent
• environment — в ECS как стандартного поля нет, обычно используют service.environment

3. Уровень и «тип» события (важно для фильтрации)

• log.level — нормализованный уровень (debug, info, warn, error, fatal)
• log.logger — имя логгера (для Java-стека часто есть loggerName)
• event.dataset и event.module — обычно согласуют с data_stream.dataset

4. Data stream поля (для data streams)

• data_stream.type — обычно logs
• data_stream.dataset — например, elasticsearch_server
• data_stream.namespace — например, dev, prod

5. Ошибки и исключения (для анализа проблем)

• error.message — читаемое описание ошибки
• error.type — тип (например, класс исключения)
• error.stack_trace — стектрейс (обычно text)

@timestamp                    — когда произошло событие
message                       — основной текст события
tags                          — массив меток для фильтрации
labels                        — пользовательские пары key:value

ecs.*                         — версия схемы
  └── ecs.version              (например "8.11")

event.*                       — метаданные события
  ├── event.original
  ├── event.module
  ├── event.dataset
  ├── event.kind               (event, alert, metric, signal)
  ├── event.category           (authentication, network, process, file...)
  ├── event.type               (start, end, creation, deletion, access...)
  ├── event.action             (login, logout, file-created...)
  ├── event.outcome            (success, failure, unknown)
  └── event.duration           (наносекунды)

service.*                     — служба-источник события
  ├── service.name
  └── service.version

host.*                        — хост-источник события
  ├── host.name
  ├── host.hostname
  ├── host.ip
  └── host.os.name

process.*                     — процесс-источник события
  ├── process.name
  ├── process.pid
  └── process.executable

data_stream.*                 — data stream поля
  ├── data_stream.type
  ├── data_stream.dataset
  └── data_stream.namespace

agent.*                       — какой beats/agent отправил
  ├── agent.type
  ├── agent.name
  └── agent.version

log.*                         — метаданные лога
  ├── log.file.path
  ├── log.level
  └── log.logger

source.* / destination.*      — сетевые соединения
  ├── source.ip
  ├── source.port
  ├── source.geo.country_name
  ├── destination.ip
  ├── destination.port
  └── destination.geo.country_name

network.*                     — сетевой уровень
  ├── network.protocol          (http, ssh, dns...)
  ├── network.direction         (inbound, outbound, internal)
  └── network.bytes

url.*                         — URL (отдельно от http)
  ├── url.original
  ├── url.full
  ├── url.domain
  ├── url.path
  └── url.query

http.*                        — поля для http-запросов
  ├── http.request.method
  ├── http.request.body.bytes
  ├── http.response.status_code
  └── http.response.body.bytes

user.*                        — данные пользователя
  ├── user.name
  ├── user.id
  └── user.email

file.*                        — операции с файлами
  ├── file.path
  ├── file.name
  └── file.size

error.*                       — информация об ошибке
  ├── error.message
  ├── error.type
  └── error.stack_trace

cloud.*                       — облачная инфраструктура
  ├── cloud.provider            (aws, gcp, azure)
  ├── cloud.region
  └── cloud.instance.id

container.*                   — контейнеры
  ├── container.id
  ├── container.name
  └── container.image.name

orchestrator.*                — оркестрация (Kubernetes...)
  ├── orchestrator.namespace
  ├── orchestrator.cluster.name
  └── orchestrator.resource.name

related.*                     — для корреляции событий
  ├── related.ip                (все IP из события в один массив)
  ├── related.user              (все имена пользователей)
  └── related.hash              (все хеши)

В маппинге индекса возможны секции properties и dynamic_templates

// Явное описание типов полей в properties
{
    "settings": {
        "number_of_shards": 5,
        "number_of_replicas": 2
    },
    "mappings": {
        // запрещаем использовать dynamic mapping
        "dynamic": "strict",
        // все поля и типы должны быть описаны ниже
        "properties": {
            "@timestamp": { "type": "date" },
            "message":    { "type": "match_only_text" }
        }
    }
}

// Динамическое описание в dynamic_templates
{
    "settings": {
        "number_of_shards": 5,
        "number_of_replicas": 2
    },
    "mappings": {
        // разрешаем использовать dynamic mapping
        "dynamic": true,
        // правила определения типа поля по имени
        "dynamic_templates": [
            {
                "ecs_ip": {
                    "path_match": ["*.ip"],
                    "mapping": { "type": "ip" }
                }
            }
        ]
    }
}

ECS базируется на dynamic templates — это правила, которые на основе имени поля определяют тип этого поля. После установки ElasticSearch нам доступен системный component template ecs@mappings, который можно использовать в своих шаблонах. Если все имена полей соответствуют схеме ECS — ElasticSearch при добавлении в индекс сам определит тип этого поля.

$ curl -X GET 'http://localhost:9200/_component_template/ecs@mappings?pretty'

{
    "component_templates": [
        {
            "name": "ecs@mappings",
            "component_template": {
                "version": 18,
                "_meta": {
                    "managed": true,
                    "description": "dynamic mappings based on ECS, installed by x-pack"
                },
                "deprecated": false,
                "template": {
                    "mappings": {
                        "dynamic_templates": [ // компонент использует dynamic templates
                            {
                                "ecs_timestamp": {
                                    "mapping": {
                                        "ignore_malformed": false,
                                        "type": "date"
                                    },
                                    "match": "@timestamp"
                                }
                            },
                            {
                                "ecs_message_match_only_text": {
                                    "path_match": [
                                        "message",
                                        "*.message"
                                    ],
                                    "mapping": {
                                        "type": "match_only_text"
                                    },
                                    "unmatch_mapping_type": "object"
                                }
                            },
                            {
                                "ecs_non_indexed_keyword": {
                                    "path_match": [
                                        "*event.original"
                                    ],
                                    "mapping": {
                                        "index": false,
                                        "type": "keyword",
                                        "doc_values": false
                                    }
                                }
                            },
                            {
                                "ecs_non_indexed_long": {
                                    "path_match": [
                                        "*.x509.public_key_exponent"
                                    ],
                                    "mapping": {
                                        "index": false,
                                        "type": "long",
                                        "doc_values": false
                                    }
                                }
                            },
                            {
                                "ecs_ip": {
                                    "path_match": [
                                        "ip",
                                        "*.ip",
                                        "*_ip"
                                    ],
                                    "mapping": {
                                        "type": "ip"
                                    },
                                    "match_mapping_type": "string"
                                }
                            },
                            {
                                "ecs_wildcard": {
                                    "path_match": [
                                        "*.io.text",
                                        "*.message_id",
                                        "*registry.data.strings",
                                        "*url.path"
                                    ],
                                    "mapping": {
                                        "type": "wildcard"
                                    },
                                    "unmatch_mapping_type": "object"
                                }
                            },
                            {
                                "ecs_path_match_wildcard_and_match_only_text": {
                                    "path_match": [
                                        "*.body.content",
                                        "*url.full",
                                        "*url.original"
                                    ],
                                    "mapping": {
                                        "fields": {
                                            "text": {
                                                "type": "match_only_text"
                                            }
                                        },
                                        "type": "wildcard"
                                    },
                                    "unmatch_mapping_type": "object"
                                }
                            },
                            {
                                "ecs_match_wildcard_and_match_only_text": {
                                    "mapping": {
                                        "fields": {
                                            "text": {
                                                "type": "match_only_text"
                                            }
                                        },
                                        "type": "wildcard"
                                    },
                                    "unmatch_mapping_type": "object",
                                    "match": [
                                        "*command_line",
                                        "*stack_trace"
                                    ]
                                }
                            },
                            {
                                "ecs_path_match_keyword_and_match_only_text": {
                                    "path_match": [
                                        "*.title",
                                        "*.executable",
                                        "*.name",
                                        "*.working_directory",
                                        "*.full_name",
                                        "*file.path",
                                        "*file.target_path",
                                        "*os.full",
                                        "*email.subject",
                                        "*vulnerability.description",
                                        "*user_agent.original"
                                    ],
                                    "mapping": {
                                        "fields": {
                                            "text": {
                                                "type": "match_only_text"
                                            }
                                        },
                                        "type": "keyword"
                                    },
                                    "unmatch_mapping_type": "object"
                                }
                            },
                            {
                                "ecs_date": {
                                    "path_match": [
                                        "*.timestamp",
                                        "*_timestamp",
                                        "*.not_after",
                                        "*.not_before",
                                        "*.accessed",
                                        "created",
                                        "*.created",
                                        "*.installed",
                                        "*.creation_date",
                                        "*.ctime",
                                        "*.mtime",
                                        "ingested",
                                        "*.ingested",
                                        "*.start",
                                        "*.end",
                                        "*.indicator.first_seen",
                                        "*.indicator.last_seen",
                                        "*.indicator.modified_at",
                                        "*threat.enrichments.matched.occurred"
                                    ],
                                    "mapping": {
                                        "type": "date"
                                    },
                                    "unmatch_mapping_type": "object"
                                }
                            },
                            {
                                "ecs_path_match_float": {
                                    "path_match": [
                                        "*.score.*",
                                        "*_score*"
                                    ],
                                    "mapping": {
                                        "type": "float"
                                    },
                                    "path_unmatch": "*.version",
                                    "unmatch_mapping_type": "object"
                                }
                            },
                            {
                                "ecs_usage_double_scaled_float": {
                                    "path_match": "*.usage",
                                    "mapping": {
                                        "scaling_factor": 1000,
                                        "type": "scaled_float"
                                    },
                                    "match_mapping_type": [
                                        "double",
                                        "long",
                                        "string"
                                    ]
                                }
                            },
                            {
                                "ecs_geo_point": {
                                    "path_match": [
                                        "*.geo.location"
                                    ],
                                    "mapping": {
                                        "type": "geo_point"
                                    }
                                }
                            },
                            {
                                "ecs_flattened": {
                                    "path_match": [
                                        "*structured_data",
                                        "*exports",
                                        "*imports"
                                    ],
                                    "mapping": {
                                        "type": "flattened"
                                    },
                                    "match_mapping_type": "object"
                                }
                            },
                            {
                                "all_strings_to_keywords": {
                                    "mapping": {
                                        "ignore_above": 1024,
                                        "type": "keyword"
                                    },
                                    "match_mapping_type": "string"
                                }
                            }
                        ]
                    }
                }
            }
        }
    ]
}

$ curl -X GET 'http://localhost:9200/_component_template/logs@mappings?pretty'

{
    "component_templates": [
        {
            "name": "logs@mappings",
            "component_template": {
                "version": 18,
                "_meta": {
                    "managed": true,
                    "description": "default mappings for the logs index template installed by x-pack"
                },
                "deprecated": false,
                "template": {
                    "mappings": {
                        "date_detection": false,
                        "properties": {
                            // эти четыре поля являются обязательными
                            "@timestamp": {
                                "type": "date"
                            },
                            "data_stream.namespace": {
                                "type": "constant_keyword"
                            },
                            "data_stream.dataset": {
                                "type": "constant_keyword"
                            },
                            "data_stream.type": {
                                "type": "constant_keyword",
                                "value": "logs"
                            }
                        }
                    }
                }
            }
        }
    ]
}

$ curl -X GET 'http://localhost:9200/_ilm/policy/30-days@lifecycle?pretty'

{
    "30-days@lifecycle": {
        "version": 1,
        "modified_date": "2026-03-02T13:33:46.825Z",
        "policy": {
            "_meta": {
                "managed": true,
                "description": "built-in ILM policy using the hot and warm phases with a retention of 30 days"
            },
            "phases": {
                "hot": {
                    "min_age": "0ms",
                    "actions": {
                        "rollover" : {
                            // условия rollover — 30 дней или 50 Гб
                            "max_age": "30d",
                            "max_primary_shard_size": "50gb"
                        }
                    }
                },
                "warm": {
                    "min_age": "2d",
                    "actions": {
                        "shrink": {
                            // shrink — уменьшение количества шардов
                            "number_of_shards": 1,
                            // запись в индекс после shrink запрещена
                            "allow_write_after_shrink": false
                        },
                        // forcemerge — принудительное слияние сегментов Lucene, уменьшает их количество,
                        // что ускоряет поиск и освобождает место, занятое удалёнными документами
                        "forcemerge" : {
                            "max_num_segments": 1
                        }
                    }
                },
                "delete": {
                    // удаление через 30 дней
                    "min_age": "30d",
                    "actions": {
                        "delete": {
                            "delete_searchable_snapshot": true
                        }
                    }
                }
            },
        },
        "in_use_by": { // где используется эта политика
            "indices": [ ], // индексы, использующие эту политику
            "data_streams": [ ], // data streams, использующие эту политику
            "composable_templates": [ ] // шаблоны индексов, ссылающиеся на эту политику
        }
    }
}

Пример шаблона, который использует системные компоненты

{
    "index_patterns": ["app-log-*"],
    "priority": 200,
    "composed_of": [
        "logs@mappings", // описывает четыре обязательных поля
        "ecs@mappings" // остальные поля определяются динамически
    ],
    "data_stream": {}, // это не обычный индекс, а data stream
    "template": {
        "settings": {
            "number_of_shards": 5,
            "number_of_replicas": 2,
            // используем системную политику 30-days@lifecycle
            "index.lifecycle.name": "30-days@lifecycle",
            // максимальное количество полей при динамическом добавлении полей
            "index.mapping.total_fields.limit": 100,
            // не добавлять динамически новое поле, если лимит уже достигнут
            "index.mapping.total_fields.ignore_dynamic_beyond_limit": true,
            "index.mapping.ignore_malformed": true
        },
        "mappings": {
            "dynamic": true, // использовать dynamic mapping
        }
    }
}

Для data streams существует более простая альтернатива ILM — Data Stream Lifecycle (DSL). Вместо создания отдельной политики с фазами достаточно указать срок хранения прямо в шаблоне. ILM оправдан только для сложных сценариев — когда нужны warm/cold фазы, shrink или forcemerge. Для большинства задач — DSL проще и современнее.

{
    "index_patterns": ["app-log-*"],
    "priority": 200,
    "composed_of": [
        "logs@mappings", // описывает четыре обязательных поля
        "ecs@mappings"   // остальные поля определяются динамически
    ],
    "data_stream": { // это не обычный индекс, а data stream
        "lifecycle": {
            "data_retention": "30d"
        }
    },
    "template": {
        "settings": {
            "number_of_shards": 5,
            "number_of_replicas": 2,
            // не используем системную политику 30-days@lifecycle
            "index.mapping.total_fields.limit": 100,
            "index.mapping.total_fields.ignore_dynamic_beyond_limit": true,
            "index.mapping.ignore_malformed": true
        }
    }
}

Установка и настройка Filebeat

Нужно скачать и установить пакет, запустить службу filebeat.service

$ curl -L -O https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-8.19.7-amd64.deb
$ sudo dpkg -i filebeat-8.19.7-amd64.deb

Редактируем файл конфигурации службы

$ sudo nano /etc/filebeat/filebeat.yml

# Загружать файлы конфигурации .yml из директории /etc/filebeat/modules.d/
filebeat.config.modules:
  path: ${path.config}/modules.d/*.yml
  reload.enabled: false
  #reload.period: 10s

# Настройки ниже разрешают управлять жизненным циклом логов, создавать шаблоны
# для логов и создавать дашборды в Kibana. Эти настройки имеют смысл, если
# используютчя модули, то есть логи отправляются напрямую в ElasticSearch.
setup:
  # Управление жизненным циклом индексов (ILM)
  ilm:
    enabled: true
    # Перезаписывать политику, если существует?
    overwrite: false
  # Настройки для шаблона индекса (mapping)
  template:
    enabled: true
    # Перезаписывать шаблоны, если существуют?
    overwrite: false
  # Настройки для установки дашбордов в Kibana
  dashboards:
    enabled: true

# Настройка нужна, чтобы Filebeat мог найти Kibana и создать дашборды
setup.kibana.host: "localhost:5601"

# Включаем отправку метрик о работе самого Filebeat в Elasticsearch.
# Будут доступны в Kibana в разделе Management => Stack Monitoring
monitoring:
  enabled: true
  # UUID можно получить с помощью запроса к API Elasticsearch
  cluster_uuid: "jL_MZQOHTriwCH2UFEs-Ow"
  elasticsearch:
    hosts: ["http://localhost:9200"]

# Узлы кластера ElasticSerach для отправки прочитаных логов
output.elasticsearch:
  hosts: ["localhost:9200"]
  # preset: balanced
  # protocol: "https"
  # api_key: "id:api_key"
  # username: "elastic"
  # password: "changeme"

# Хост и порт Logstash для отправки прочитанных файлов логов
# output.logstash:
  # hosts: ["localhost:5044"]
  # ssl.certificate_authorities: ["/etc/pki/root/ca.pem"]
  # ssl.certificate: "/etc/pki/client/cert.pem"
  # ssl.key: "/etc/pki/client/cert.key"

# Уровень логирования Filebeat, доступные значения — error,
# warning, info, debug. Значение по умолчанию — info.
logging.level: info

# Отключаем отправку логов в stderr, чтобы не дублировать
logging.to_stderr: false

# Отключаем отправку логов в syslog, чтобы не дублировать
logging.to_syslog: false

# Включаем логирование в файл, указываем путь, шаблон имени файла,
# права доступа, условия ротации, сколько старых файлов хранить
logging.to_files: true

logging.files:
  # Директория для файлов логов
  path: /var/log/filebeat
  # Базовое имя файла (будет создан /var/log/filebeat/filebeat.log)
  name: filebeat
  # Максимальный размер файла до ротации (10 MB)
  rotateeverybytes: 10485760
  # Сколько ротированных файлов хранить (filebeat.1, filebeat.2 ...)
  keepfiles: 7
  # Права доступа к файлу логов
  permissions: 0640
  # Перенаправлять stderr в файл логов — полезно для отладки
  # при критическом сбое Filebeat
  redirect_stderr: true

# Отключаем периодическую запись внутренних метрик в логи.
# У нас уже есть отправка метрик в ElasticSearch, так что
# дополнительно записывать метрики в лог нет необходимости.
logging.metrics.enabled: false
# Как часто писать метрики во внутренний лог (300 секунд)
logging.metrics.period: 300s
# Какие пространства имён метрик логировать. Достаточно
# stats — общая статистика по работе Filebeat.
logging.metrics.namespaces: [stats]

# Мониторинг состояния службы Filebeat (по умолчанию false)
http.enabled: false
http.host: "127.0.0.1"
http.port: 5066

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

У Filebeat есть два независимых канала вывода

• Собственная система логирования, настраивается через logging.to_files, logging.to_syslog, logging.to_stderr — они не исключают друг друга, можно включить сразу несколько.
• Стандартный поток ошибок (stderr) — низкоуровневый системный поток, в который Go Runtime пишет напрямую, минуя систему логирования Filebeat. Если Filebeat падает с panic или Go обнаруживает критическую ошибку выполнения — это уйдёт в stderr.

Опции logging.to_stderr и logging.files.redirect_stderr позволяют либо направить логи Filebeat в stderr (при записи в системный лог), либо перенаправить stderr в файл (когда запись идет в отдельный лог-файл). Смысл этого в том, чтобы все, связанное с Filebeat, было в одном месте — либо в системном логе, либо в отдельном файле лога.

$ sudo nano /etc/elasticsearch/elasticsearch.yml

# Включаем сбор и хранение данных о работе ElasticSearch
# и других компонентов ELK (Kibana, Logstash, Beats)
xpack.monitoring.collection.enabled: true

$ sudo systemclt restart elasticsearch.service

$ curl -s http://localhost:9200/
{
    "name" : "node-1",
    "cluster_name" : "ecommerce",
    "cluster_uuid" : "jL_MZQOHTriwCH2UFEs-Ow",
    "version" : {
        "number" : "8.19.5",
        "build_flavor" : "default",
        "build_type" : "deb",
        "build_hash" : "d6dd0417f05cd69706f4f103c69bbb8b7688db9c",
        "build_date" : "2025-10-03T16:35:50.165700789Z",
        "build_snapshot" : false,
        "lucene_version" : "9.12.2",
        "minimum_wire_compatibility_version" : "7.17.0",
        "minimum_index_compatibility_version" : "7.0.0"
    },
    "tagline" : "You Know, for Search"
}

Как правильно работать с модулями

Модули сильно упрощают процесс настройки, автоматически конфигурируя Kibana и Elasticsearch для работы с данными от Filebeat, что позволяет быстрее приступить к анализу логов. С другой стороны — модули умеют работать только с логами стандартного формата.

$ ls -la /etc/filebeat/modules.d/
итого 248
drwxr-xr-x 2 root root  4096 мар 11 12:12 .
drwxr-xr-x 3 root root  4096 мар 13 19:06 ..
-rw-r--r-- 1 root root   486 ноя  7 15:24 activemq.yml.disabled
-rw-r--r-- 1 root root   478 ноя  7 15:24 apache.yml.disabled
-rw-r--r-- 1 root root   282 ноя  7 15:24 auditd.yml.disabled
-rw-r--r-- 1 root root  2291 ноя  7 15:24 awsfargate.yml.disabled
-rw-r--r-- 1 root root 11645 ноя  7 15:24 aws.yml.disabled
..........

Верхний уровень каждого модуля имеет такую структуру

/usr/share/filebeat/module/nginx/
├── module.yml              ← описание всего модуля
├── access/                 ← набор для access логов
├── error/                  ← набор для error логов
└── ingress_controller/     ← набор для Kubernetes Nginx Ingress

Файл module.yml содержит метаданные модуля — название, описание, ссылки на документацию.

Каждая поддиректория (access, error, ingress_controller) имеет такую структуру

access/
├── manifest.yml            ← главный конфиг access логов
├── config/
│   └── nginx-access.yml    ← где искать access лог файлы
└── ingest/
    └── pipeline.yml        ← как создать ingest pipeline

Активация модуля для отправки логов Nginx

$ sudo filebeat modules enable nginx

Файл конфигурации для модуля Nginx имеет вид

$ sudo nano /etc/filebeat/modules.d/nginx.yml

# Module: nginx
# Docs: https://www.elastic.co/guide/en/beats/filebeat/8.19/filebeat-module-nginx.html

- module: nginx
  # Access logs
  access:
    enabled: false
    # Set custom paths for the log files. If left empty,
    # Filebeat will choose the paths depending on your OS.
    #var.paths:

  # Error logs
  error:
    enabled: false
    # Set custom paths for the log files. If left empty,
    # Filebeat will choose the paths depending on your OS.
    #var.paths:

  # Ingress-nginx controller logs. This is disabled by default. It could
  # be used in Kubernetes environments to parse ingress-nginx logs
  ingress_controller:
    enabled: false
    # Set custom paths for the log files. If left empty,
    # Filebeat will choose the paths depending on your OS.
    #var.paths:

После активации модуля нужно выполнить команду setup, которая подготовит ElasticSearch к получению логов от Filebeat.

1. Читает конфиги модулей из /etc/filebeat/modules.d/
2. Загружает Index Template → ElasticSearch (определяет маппинг полей для индексов)
3. Создаёт ILM политику → ElasticSearch (правила ротации индексов hot → warm → cold → delete)
4. Создаёт Ingest Pipeline → ElasticSearch (правила парсинга и обработки логов)
5. Загружает Dashboards → Kibana (готовые визуализации для модулей)

$ sudo filebeat setup
Index setup finished.
Loading dashboards (Kibana must be running and reachable)
Loaded dashboards
Loaded Ingest pipelines

action.auto_create_index: false

Работа команды setup управляется директивами в файле конфигурации Filebeat. Если все директивы имеют значение false — команда ничего не будет делать. Полезно один раз разрешить все настройки и выполнить команду setup, после чего все эти настройки запретить. При активации еще одного модуля — повторный запуск команды не нужен, все было сделано при первом запуске.

setup.ilm.enabled: false
setup.ilm.overwrite: false
setup.template.enabled: false
setup.template.overwrite: false
setup.dashboards.enabled: false

Команда setup создает одну ILM-политику filebeat и один шаблон filebeat-8.19.7 для всех модулей. Это значит, что логи от всех модулей будут записаны в один индекс filebeat-8.19.7. И это не обычный индекс, а data stream — то есть специальный индекс для хранения логов. При этом ILM-политика filebeat не имеет delete-фазы, то есть логи не будут удаляться — лучше это сразу изменить.

Чтобы посмотреть логи только от одного модуля — записи нужно фильтровать по полю event.module, которое может принимать значение nginx, apache, haproxy, redis и т.п. Индекс filebeat-8.19.7 может иметь достаточно много полей для записи логов от разных служб, можно из левой колонки выбрать только нужные. Отфильтрованные по event.module записи нужно сохранить для быстрого доступа в будущем Меню → Stack Management → Kibana → Saved Objects.

Команда setup создает много Ingest Pipelines filebeat-8.19.11-* — для всех модулей, которые есть в директории /etc/filebeat/modules.d.

Команда setup создает много дашбордов (не для всех модулей, но для многих) — посмотреть можно в разделе Меню → Analytics → Dashboards. Честно говоря, есть большие сомнения в необходимости создания всех этих дашбородов — гораздо удобнее создать только те дашборды, которые реально нужны для работы, вместо того, чтобы искать подходящие среди этого зоопарка.

Несложные преобразования данных

Filebeat умеет выполнять несложные преобразования данных до отправки в Logstash или ElasticSearch.

• Глобальные процессоры — определяются в корне конфигурации и применяются ко всем событиям
• Процессоры на уровне ввода (input) — применяются только к данным из конкретного источника

# отбросить, если выполнено условие
processors:
  - drop_event:
      when:
        regexp:
          message: "^DEBUG"

# извлечь поля из message в корень
processors:
  - decode_json_fields:
      fields: ["message"]
      target: ""
      overwrite_keys: true

# удалить указанные поля из события
processors:
  - drop_fields:
      fields: ["agent.ephemeral_id", "agent.hostname"]
      ignore_missing: true

# процессор на уровне ввода (input)
filebeat.inputs:
  - type: log
    paths: ["/var/log/app.log"]
    processors:
      - add_labels:
          labels:
            service: backend

Примеры отправки логов в ElasticSearch

Отправить логи приложения или службы в ElasticSearch для хранения и анализа можно с помощью Logstash или Filebeat.

• Filebeat — это очень легковесный агент, написанный на Go. Он потребляет минимум CPU и RAM. Его основная задача — эффективно и надежно собирать данные и отправлять их. Filebeat надежно отслеживает позицию чтения, умеет корректно обрабатывать ротацию логов (когда старый лог-файл архивируется, а новый создается), помнит, откуда продолжать чтение после перезапуска. У него есть встроенные механизмы обратного давления (backpressure), чтобы не перегружать приемник.
• Logstash — это JVM-приложение, которое требует значительно больше ресурсов (CPU и RAM) для работы, так как он предназначен для сложной обработки, парсинга и трансформации данных. Умеет извлекать структурированные поля из неструктурированных текстовых логов с помощью фильтра grok, объединять многострочные логи (например, стек-трейсы ошибок), обогащать логи дополнительной информацией (например, добавлять геолокацию по ip-адресу), удалять конфиденциальную информацию.

Устанавливать и запускать полноценный Logstash на каждом сервере, с которого нужно собирать логи, неэффективно и расточительно. Это будет сильно нагружать серверы приложений.

• Архитектура ELK предполагает, что легкие Beats (Filebeat, Metricbeat и т.д.) стоят на «краях» инфраструктуры (на серверах приложения, баз данных) и занимаются сбором и отправкой.
• Logstash же ставится на отдельных, мощных серверах как центральный «комбайн» для обработки всех собранных данных. Это позволяет масштабировать обработку независимо от сбора логов.

Рассмотрим несколько примеров отправки логов в ElasticSearch. У меня на одной машине установлены ElasticSearch, Logstash, Kibana — логи этих служб и будем отправлять. Мы можем поручить чтение логов Logstash или Filebeat. Filebeat может напрямую отправлять их в ElasticSearch или отправлять на предварительную обработку в Logstash (и уже Logstash отправит их дальше в ElasticSearch).

1. Отправка логов ELK-стэка в ElasticSearch

Активируем три модуля для работы с логами — Elasticsearch, Logstash, Kibana

$ sudo filebeat modules enable elasticsearch
$ sudo filebeat modules enable logstash
$ sudo filebeat modules enable kibana

Редактируем файлы конфигураций этих модулей

$ sudo nano /etc/filebeat/modules.d/elasticsearch.yml

- module: elasticsearch
  server:
    enabled: true  # включаем отправку логов
    var.paths:
      - /var/log/elasticsearch/ecommerce_server.json
  gc:
    enabled: false
  audit:
    enabled: false
  slowlog:
    enabled: true  # включаем отправку логов
    var.paths:
      - /var/log/elasticsearch/ecommerce_index_indexing_slowlog.json
      - /var/log/elasticsearch/ecommerce_index_search_slowlog.json
  deprecation:
    enabled: false

$ sudo nano /etc/filebeat/modules.d/logstash.yml

- module: logstash
  log:
    enabled: true
    var.paths:
      - /var/log/logstash/logstash-json.log
  slowlog:
    enabled: true
    var.paths:
      - /var/log/logstash/logstash-slowlog-json.log

$ sudo nano /etc/filebeat/modules.d/kibana.yml

- module: kibana
  log:
    enabled: true
    var.paths:
      - /var/log/kibana/kibana.json
  audit:
    enabled: false

Выполняем команду для подготовки ElasticSearch к приему логов от трех служб

$ sudo filebeat setup
Index setup finished.
Loading dashboards (Kibana must be running and reachable)
Loaded dashboards
Loaded Ingest pipelines

Запускаем службу и добавляем в автозагрузку

$ sudo systemctl start filebeat.service
$ sudo systemctl enable filebeat.service

Чтобы посмотреть логи только от одного модуля — записи нужно фильтровать по полю event.module, которое может принимать значение elasticsearch, logstash или kibana. Отфильтрованные записи можно сохранить — наборы данных будут доступны в разделе Меню → Stack Management → Kibana → Saved Objects.

2. Filebeat + Logstash. Отправка логов ELK стэка

Обычно Filebeat устанавливают на каждый сервер, где запущена какая-то служба (MySQL, Redis, Nginx, Apache). Filebeat отправляет данные на отдельный сервер Logstash — центральный «комбайн» для обработки всех собранных данных. Это основной сценарий сбора логов и отправки их в ElasticSearch. Рассмотренный выше способ отправки логов напрямую из Filebeat в Elasticsearch — используется достаточно редко, для простых случаев.

$ sudo systemctl stop logstash.service
$ sudo systemctl stop filebeat.service

В этот раз мы не будем использовать готовые модули Filebeat, как делали это для отправки логов из Filebeat в Elasticsearch, а используем вместо этого директиву filebeat.inputs. При этом мы отключаем отправку логов в ElasticSearch — можно использовать либо директиву output.elasticsearch, либо output.logstash, но не обе сразу.

$ sudo nano /etc/filebeat/filebeat.yml

filebeat.inputs:
# Этот input для чтения файлов логов Elastic, тип filestream — рекомендуемый
- type: filestream
  id: elastic-server
  enabled: true
  paths:
    - /var/log/elasticsearch/ecommerce_server.json
  fields:
    data_stream:
      type: logs
      dataset: elastic_server
      namespace: test
  fields_under_root: true

# Этот input для чтения файлов логов Elastic, тип filestream — рекомендуемый
- type: filestream
  id: elastic-search-slowlog
  enabled: true
  paths:
    - /var/log/elasticsearch/ecommerce_index_search_slowlog.json
  fields:
    data_stream:
      type: logs
      dataset: elastic_slowlog_search
      namespace: test
  fields_under_root: true

# Этот input для чтения файлов логов Elastic, тип filestream — рекомендуемый
- type: filestream
  id: elastic-indexing-slowlog
  enabled: true
  paths:
    - /var/log/elasticsearch/ecommerce_index_indexing_slowlog.json
  fields:
    data_stream:
      type: logs
      dataset: elastic_slowlog_indexing
      namespace: test
  fields_under_root: true

# Этот input для чтения файлов логов Logstash, тип filestream — рекомендуемый
- type: filestream
  id: logstash
  enabled: true
  paths:
    - /var/log/logstash/logstash-json.log
  fields:
    data_stream:
      type: logs
      dataset: logstash_common
      namespace: test
  fields_under_root: true

# Этот input для чтения файлов логов Logstash, тип filestream — рекомендуемый
- type: filestream
  id: logstash-slowlog
  enabled: true
  paths:
    - /var/log/logstash/logstash-slowlog-json.log
  fields:
    data_stream:
      type: logs
      dataset: logstash_slowlog
      namespace: test
  fields_under_root: true

# Этот input для чтения файлов логов Kibana, тип filestream — рекомендуемый
- type: filestream
  id: kibana
  enabled: true
  paths:
    - /var/log/kibana/kibana.json
  fields:
    data_stream:
      type: logs
      dataset: kibana
      namespace: test
  fields_under_root: true

# Не загружать файлы конфигурации .yml из директории /etc/filebeat/modules.d/
# filebeat.config.modules:
  # path: ${path.config}/modules.d/*.yml
  # reload.enabled: false

# Настройки ниже запрещают создавать дашборды в Kibana, шаблоны для логов
# в ElasticSearch и управлять жизненным циклом Index Lifecycle Management
setup:
  # Управление жизненным циклом индексов (ILM)
  ilm:
    enabled: false
    # Перезаписывать политику, если существует?
    overwrite: false
  # Настройки для шаблона индекса (mapping)
  template:
    enabled: false
    # Перезаписывать шаблоны, если существуют?
    overwrite: false
  # Настройки для установки дашбордов в Kibana
  dashboards:
    enabled: false

# Настройка нужна, чтобы Filebeat мог найти Kibana и создать дашборды
setup.kibana.host: "localhost:5601"

# Включаем отправку метрик о работе самого Filebeat в Elasticsearch.
# Будут доступны в Kibana в разделе Management => Stack Monitoring
monitoring:
  enabled: true
  cluster_uuid: "jL_MZQOHTriwCH2UFEs-Ow"
  elasticsearch:
    hosts: ["http://localhost:9200"]

# Отправка логов Elastic, Logstash и Kibana в Elastic
# output.elasticsearch:
  # hosts: ["localhost:9200"]

# Отправка логов Elastic, Logstash и Kibana в Logstash
output.logstash:
  hosts: ["localhost:5044"]

Теперь редактируем файлы конфигурации Logstash

$ sudo nano /etc/logstash/pipelines.yml

- pipeline.id: syslog
  path.config: "/etc/logstash/conf.d/syslog.conf"
  pipeline.workers: 1

- pipeline.id: elk-stack
  path.config: "/etc/logstash/conf.d/elk-stack.conf"
  pipeline.workers: 1

Файл конфигурации для обработки логов ELK-стэка — для начала в нем будет все предельно просто, нам нужно просто посмотреть, какие данные отправляет Filebeat, чтобы потом правильно их обработать.

$ sudo nano /etc/logstash/conf.d/elk-stack.conf

input {
    beats {
        port => 5044
    }
}

filter {
    # этот фильтр распарсит json-поле message на поля
    json {
        source => "message"
    }
    # после парсинга json-поле message нам не нужно
    mutate {
        remove_field => ["message"]
    }
}

output {
    # посмотрим, какие данные отправляет Filebeat
    stdout { codec => rubydebug }
}

Вручную запускаем Filebeat и смотрим вывод

$ sudo /usr/share/logstash/bin/logstash --path.settings /etc/logstash -f /etc/logstash/conf.d/elk-stack.conf

# Файл лога Elastic
{
    # Исходная строка файла лога
    "message" => "{\"@timestamp\":\"2026-02-09T13:05:15.434Z\", \"log.level\": \"INFO\", \"message\":\"Data stream lifecycle is issuing a request to force merge index [.fs-filebeat-8.19.7-2026.01.19-000002]\", \"ecs.version\": \"1.2.0\",\"service.name\":\"ES_ECS\",\"event.dataset\":\"elasticsearch.server\",\"process.thread.name\":\"elasticsearch[node-1][trigger_engine_scheduler][T#1]\",\"log.logger\":\"org.elasticsearch.datastreams.lifecycle.DataStreamLifecycleService\",\"elasticsearch.cluster.uuid\":\"jL_MZQOHTriwCH2UFEs-Ow\",\"elasticsearch.node.id\":\"3BLc3_3VQsulmOVe-BVZRA\",\"elasticsearch.node.name\":\"node-1\",\"elasticsearch.cluster.name\":\"ecommerce\"}",
    "host" => {
        "name" => "EVG-UBN"
    },
    "@version" => "1",
    # Время чтения из файла лога
    "@timestamp" => 2026-02-09T13:05:24.556Z,
    # Результат парсинга json-строки лога
    "json_data" => {
        "message"                    => "Data stream lifecycle is issuing a request to force merge index [.fs-filebeat-8.19.7-2026.01.19-000002]",
        "elasticsearch.cluster.uuid" => "jL_MZQOHTriwCH2UFEs-Ow",
        # Дата и время записи в лог
        "@timestamp"                 => "2026-02-09T13:05:15.434Z",
        "ecs.version"                => "1.2.0",
        "elasticsearch.node.name"    => "node-1",
        "log.level"                  => "INFO",
        "elasticsearch.node.id"      => "3BLc3_3VQsulmOVe-BVZRA",
        "event.dataset"              => "elasticsearch.server",
        "elasticsearch.cluster.name" => "ecommerce",
        "log.logger"                 => "org.elasticsearch.datastreams.lifecycle.DataStreamLifecycleService",
        "process.thread.name"        => "elasticsearch[node-1][trigger_engine_scheduler][T#1]",
        "service.name"               => "ES_ECS"
    },
    "ecs" => {
        "version" => "8.0.0"
    },
    # Агент чтения файла лога
    "agent" => {
        "ephemeral_id" => "cb3caf78-18c1-4118-b0a1-1483ab9c66f7",
        "version"      => "8.19.7",
        "name"         => "EVG-UBN",
        "id"           => "9fa7f789-e002-42dc-8537-d585a3f14178",
        "type"         => "filebeat"
    },
    "tags" => [
        [0] "beats_input_codec_plain_applied"
    ],
    "event" => {
        # Исходная строка файла лога
        "original" => "{\"@timestamp\":\"2026-02-09T13:05:15.434Z\", \"log.level\": \"INFO\", \"message\":\"Data stream lifecycle is issuing a request to force merge index [.fs-filebeat-8.19.7-2026.01.19-000002]\", \"ecs.version\": \"1.2.0\",\"service.name\":\"ES_ECS\",\"event.dataset\":\"elasticsearch.server\",\"process.thread.name\":\"elasticsearch[node-1][trigger_engine_scheduler][T#1]\",\"log.logger\":\"org.elasticsearch.datastreams.lifecycle.DataStreamLifecycleService\",\"elasticsearch.cluster.uuid\":\"jL_MZQOHTriwCH2UFEs-Ow\",\"elasticsearch.node.id\":\"3BLc3_3VQsulmOVe-BVZRA\",\"elasticsearch.node.name\":\"node-1\",\"elasticsearch.cluster.name\":\"ecommerce\"}"
    },
    "log" => {
        "offset" => 128573,
        "file"   => {
            "inode"     => "12320813",
            "device_id" => "2052",
            "path"      => "/var/log/elasticsearch/ecommerce_server.json"
        }
    },
    "input" => {
        "type" => "filestream"
    },
    # Поля, добавленные в Filebeat
    "data_stream" => {
        "namespace" => "test",
        "dataset"   => "elastic_server",
        "type"      => "logs"
    }
}

# Файл лога Logstash
{
    # Исходная строка файла лога
    "message" => "{\"level\":\"INFO\",\"loggerName\":\"logstash.javapipeline\",\"timeMillis\":1770641248880,\"thread\":\"Converge PipelineAction::Create<main>\",\"logEvent\":{\"message\":\"Pipeline `main` is configured with `pipeline.ecs_compatibility: v8` setting. All plugins in this pipeline will default to `ecs_compatibility => v8` unless explicitly configured otherwise.\"}}",
           "host" => {
        "name" => "EVG-UBN"
    },
    "@version" => "1",
    # Время чтения из файла лога
    "@timestamp" => 2026-02-09T12:59:34.559Z,
    # Результат парсинга json-строки лога
    "json_data" => {
        # Дата и время записи в лог
        "timeMillis" => 1770641248880,
        "level"      => "INFO",
        "thread" => "Converge PipelineAction::Create<main>",
        "logEvent" => {
            "message" => "Pipeline `main` is configured with `pipeline.ecs_compatibility: v8` setting. All plugins in this pipeline will default to `ecs_compatibility => v8` unless explicitly configured otherwise."
        },
        "loggerName" => "logstash.javapipeline"
    },
    "ecs" => {
        "version" => "8.0.0"
    },
    # Агент чтения файла лога
    "agent" => {
        "ephemeral_id" => "cb3caf78-18c1-4118-b0a1-1483ab9c66f7",
        "version"      => "8.19.7",
        "name"         => "EVG-UBN",
        "id"           => "9fa7f789-e002-42dc-8537-d585a3f14178",
        "type"         => "filebeat"
    },
    "tags" => [
        [0] "beats_input_codec_plain_applied"
    ],
    "event" => {
        "original" => "{\"level\":\"INFO\",\"loggerName\":\"logstash.javapipeline\",\"timeMillis\":1770641248880,\"thread\":\"Converge PipelineAction::Create<main>\",\"logEvent\":{\"message\":\"Pipeline `main` is configured with `pipeline.ecs_compatibility: v8` setting. All plugins in this pipeline will default to `ecs_compatibility => v8` unless explicitly configured otherwise.\"}}"
    },
    "log" => {
        "file" => {
            "inode"     => "12320782",
            "device_id" => "2052",
            "path"      => "/var/log/logstash/logstash-json.log"
        },
        "offset" => 33199
    },
    "input" => {
        "type" => "filestream"
    },
    # Поля, добавленные в Filebeat
    "data_stream" => {
        "namespace" => "test",
        "dataset"   => "logstash_common",
        "type"      => "logs"
    }
}

# Файл лога Kibana
{
    # Исходная строка файла лога
    "message" => "{\"service\":{\"node\":{\"roles\":[\"background_tasks\",\"ui\"]}},\"ecs\":{\"version\":\"8.11.0\"},\"@timestamp\":\"2026-02-09T16:00:40.442+03:00\",\"message\":\"[UnenrollInactiveAgentsTask] runTask ended: success\",\"log\":{\"level\":\"INFO\",\"logger\":\"plugins.fleet.fleet:unenroll-inactive-agents-task:1.0.1\"},\"process\":{\"pid\":2146,\"uptime\":12662.393442755},\"trace\":{\"id\":\"6665fcaab9affc1398b508d7a7fbffd5\"},\"transaction\":{\"id\":\"ecad843d0348788e\"}}",
    "host" => {
        "name" => "EVG-UBN"
    },
    "@version" => "1",
    # Время чтения из файла лога
    "@timestamp" => 2026-02-09T13:00:48.573Z,
    # Результат парсинга json-строки лога
    "json_data" => {
        "message" => "[UnenrollInactiveAgentsTask] runTask ended: success",
        "process" => {
            "uptime" => 12662.393442755,
            "pid"    => 2146
        },
        "trace" => {
            "id" => "6665fcaab9affc1398b508d7a7fbffd5"
        },
        # Дата и время записи в лог
        "@timestamp" => "2026-02-09T16:00:40.442+03:00",
        "ecs" => {
            "version" => "8.11.0"
        },
        "log" => {
            "level"  => "INFO",
            "logger" => "plugins.fleet.fleet:unenroll-inactive-agents-task:1.0.1"
        },
        "transaction" => {
            "id" => "ecad843d0348788e"
        },
        "service" => {
            "node" => {
                "roles" => [
                    [0] "background_tasks",
                    [1] "ui"
                ]
            }
        }
    },
    "ecs" => {
        "version" => "8.0.0"
    },
    # Агент чтения файла лога
    "agent" => {
        "ephemeral_id" => "cb3caf78-18c1-4118-b0a1-1483ab9c66f7",
        "version"      => "8.19.7",
        "name"         => "EVG-UBN",
        "id"           => "9fa7f789-e002-42dc-8537-d585a3f14178",
        "type"         => "filebeat"
    },
     "tags" => [
        [0] "beats_input_codec_plain_applied"
    ],
    "event" => {
        "original" => "{\"service\":{\"node\":{\"roles\":[\"background_tasks\",\"ui\"]}},\"ecs\":{\"version\":\"8.11.0\"},\"@timestamp\":\"2026-02-09T16:00:40.442+03:00\",\"message\":\"[UnenrollInactiveAgentsTask] runTask ended: success\",\"log\":{\"level\":\"INFO\",\"logger\":\"plugins.fleet.fleet:unenroll-inactive-agents-task:1.0.1\"},\"process\":{\"pid\":2146,\"uptime\":12662.393442755},\"trace\":{\"id\":\"6665fcaab9affc1398b508d7a7fbffd5\"},\"transaction\":{\"id\":\"ecad843d0348788e\"}}"
    },
    "log" => {
        "offset" => 10378437,
        "file" => {
            "inode"     => "12320780",
            "device_id" => "2052",
            "path"      => "/var/log/kibana/kibana.json"
        }
    },
    "input" => {
        "type" => "filestream"
    },
    # Поля, добавленные в Filebeat
    "data_stream" => {
        "namespace" => "test",
        "dataset"   => "kibana",
        "type"      => "logs"
    }
}

Теперь, когда мы посмотрели на те данные, которые нам отправляет Filebeat — можем изменить файл конфигурации, чтобы обрабатывать логи от ElasticSearch, Logstash и Kibana.

$ sudo nano /etc/logstash/conf.d/elk-stack.conf

input {
    beats {
        port => 5044
        ecs_compatibility => v8
    }
}

filter {
    # Парсим JSON из поля message в отдельное поле json_data, чтобы не было
    # случайных конфликтов с полями ECS
    json {
        source => "message"
        target => "json_data"
    }

    # В поле message записываем человекочитаемое сообщение из json_data
    if [json_data][message] { # для Elastic и Kibana
        mutate {
            copy => { "[json_data][message]" => "[message]" }
        }
    } else if [json_data][logEvent][message] { # для Logstash
        mutate {
            copy => { "[json_data][logEvent][message]" => "[message]" }
        }
    }

    # Заполняем поле service.name исходя из содержимого поля data_stream.dataset
    if ![service][name] and [data_stream][dataset] {
        if [data_stream][dataset] =~ /^elastic/ {
            mutate {
                add_field => { "[service][name]" => "elastic" }
            }
        } else if [data_stream][dataset] =~ /^logstash/ {
            mutate {
                add_field => { "[service][name]" => "logstash" }
            }
        } else if [data_stream][dataset] =~ /^kibana/ {
            mutate {
                add_field => { "[service][name]" => "kibana" }
            }
        }
    }

    # Заполняем поле event.module из поля data_stream.dataset (просто копируем)
    if ![event][module] and [data_stream][dataset] {
        mutate {
            copy => { "[data_stream][dataset]" => "[event][module]" }
        }
    }

    # Поле event.dataset обычно удобно делать равным data_stream.dataset
    if ![event][dataset] and [data_stream][dataset] {
        mutate {
            copy => { "[data_stream][dataset]" => "[event][dataset]" }
        }
    }

    # Если в JSON есть свой timestamp — используем его (это время, когда
    # произошло событие, а не то время, когда Filebeat прочитал из лога)
    if [json_data][@timestamp] {
        date {
            match => ["[json_data][@timestamp]", "ISO8601"]
            target => "@timestamp"
        }
    } else if [json_data][timeMillis] {
        date {
            match => ["[json_data][timeMillis]", "UNIX_MS"]
            target => "@timestamp"
        }
    } else if [json_data][timestamp] {
        date {
            match => ["[json_data][timestamp]", "ISO8601", "yyyy-MM-dd HH:mm:ss,SSS", "yyyy-MM-dd HH:mm:ss.SSS"]
            target => "@timestamp"
        }
    }

    # Заполняем поле log.level, ищем подходящие поля в json_data
    if ![log][level] {
        if [json_data][log.level] { # для Elastic
            mutate {
                copy => { "[json_data][log.level]" => "[log][level]" }
            }
        } else if [json_data][level] { # для Logstash
            mutate {
                copy => { "[json_data][level]" => "[log][level]" }
            }
        } else if [json_data][log][level] { # для Kibana
            mutate {
                copy => { "[json_data][log][level]" => "[log][level]" }
            }
        }
    }
    
    if [log][level] {
        mutate {
            uppercase => ["[log][level]"]  # INFO, WARN, ERROR, DEBUG
        }
    }

    # Заполняем поле log.logger, ищем подходящие поля в json_data
    if ![log][logger] {
        if [json_data][log.logger] { # для Elastic
            mutate {
                copy => { "[json_data][log.logger]" => "[log][logger]" }
            }
        } else if [json_data][loggerName] { # для Logstash
            mutate {
                copy => { "[json_data][loggerName]" => "[log][logger]" }
            }
        } else if [json_data][log][logger] { # для Kibana
            mutate {
                copy => { "[json_data][log][logger]" => "[log][logger]" }
            }
        }
    }
}

output {
    elasticsearch {
        hosts => ["http://localhost:9200"]
        # Это не обычный индекс, а специальный индекс для логов (data stream)
        data_stream => "true"
        # Logstash проверит поля [data_stream][type], [data_stream][dataset],
        # [data_stream][namespace] и сам сформирует имя data stream, например
        # logs-elastic_slowlog_indexing-test
        data_stream_auto_routing => true
        # Если этих полей не будет — имя будет сформировано из значений ниже
        data_stream_type => "logs"
        data_stream_dataset => "elastic_logstash_kibana"
        data_stream_namespace => "test"
    }
}

В ElasticSearch есть встроенный шаблон logs, который состоит из компонентов logs@mappings, logs@settings, ecs@mappings. Шаблон logs применяется для индексов logs-*-* (все наши подходят), использует встроенную ILM политику logs@lifecycle (см. logs@settings). Но политика logs@lifecycle не имеет delete-фазы, так что логи будут накапливаться бесконечно. Технически у нас есть право добавить delete-фазу, но встроенная политика может быть перезаписана при обновлении — так что лучше этого не делать.

Для создания шаблона индекса используем компоненты logs@mappings, ecs@mappings и встроенную ILM политику 30-days@lifecycle (не торопитесь выполнять, еще немного доработаем шаблон).

PUT _index_template/elastic_logstash_kibana_template

{
    "priority": 200, // приоритет выше, чем у шаблона logs
    "index_patterns": [
        "logs-elastic*-*",
        "logs-logstash*-*",
        "logs-kibana*-*"
    ],
    "composed_of": [
        "logs@mappings", // описывает четыре обязательных поля
        "ecs@mappings" // остальные поля определяются динамически
    ],
    "data_stream": {}, // это не обычный индекс, а data stream
    "template": {
        "settings": {
            "number_of_shards": 1,
            "number_of_replicas": 0,
            // используем системную политику 30-days@lifecycle
            "index.lifecycle.name": "30-days@lifecycle",
            "index.mapping.total_fields.limit": 1000,
            "index.mapping.total_fields.ignore_dynamic_beyond_limit": true,
            "index.mapping.ignore_malformed": true
        },
        "mappings": {
            "dynamic": true // использовать dynamic mapping
        }
    }
}

Мы уже говорили выше, что для data stream существует DSL — более простая альтернатива ILM. Так что вместо использования политики 30-days@lifecycle — можем просто указать срок хранения логов.

PUT _index_template/elastic_logstash_kibana_template

{
    "priority": 200, // приоритет выше, чем у шаблона logs
    "index_patterns": [
        "logs-elastic*-*",
        "logs-logstash*-*",
        "logs-kibana*-*"
    ],
    "composed_of": [
        "logs@mappings", // описывает четыре обязательных поля
        "ecs@mappings" // остальные поля определяются динамически
    ],
    "data_stream": { },  // это не обычный индекс, а data stream
    "template": {
        "lifecycle": {
            "data_retention": "30d"
        },
        "settings": {
            "number_of_shards": 1,
            "number_of_replicas": 0,
            // не используем системную политику 30-days@lifecycle
            "index.mapping.total_fields.limit": 1000,
            "index.mapping.total_fields.ignore_dynamic_beyond_limit": true,
            "index.mapping.ignore_malformed": true
        },
        "mappings": {
            "dynamic": true // использовать dynamic mapping
        }
    }
}

Запускаем службы в правильном порядке — сначала Logstash, чтобы он начал слушать порт, затем Filebeat

$ sudo systemctl start logstash.service
$ sudo systemctl start filebeat.service

В Kibana смотрим data streams в разделе Меню → Management → Stack Management → Index Management

Каждый data stream состоит их нескольких индексов, это можно увидеть, если кликнуть по ссылке в колонке Indices

В Kibana смотрим lifecycle политики в разделе Меню → Management → Stack Management → Index Lifecycle Policies

Изначально там только системные политики, в том числе политика 30-days@lifecycle, которую мы планировали использовать, но отказались в пользу DSL (Data Stream Lifecycle).

Подводя небольшой итог — логи можно обрабатывать и хранить «по старому» и «по новому». По старому — когда мы используем обычный индекс, сами создаем шаблон, ILM-политику, начальный индекс, не используем ECS, описываем в шаблоне все поля. По новому — когда мы используем data stream (специальный индекс для логов), ECS (Elastic Common Schema), системные компоненты logs@mappings и ecs@mappings + более простой DSL вместо ILM.

Как начать все с чистого листа

И вот когда все было готово, когда родная планета приняла сравнительно благоустроенный вид — все сломалось. Вроде бы все службы (elasticsearch.service, logstash.service, kibana.service, filebeat.service) работают, отправляют и принимают данные, но при этом в Kibana в разделе Management → Stack Management висит сообщение «No monitoring data found».

Причина нашлась в логах службы Kibana

cluster_block_exception: index [.kibana_task_manager_...] blocked by:
[TOO_MANY_REQUESTS/12/disk usage exceeded flood-stage watermark]

На диске, где Elasticsearch хранит свои данные (обычно /var/lib/elasticsearch), заканчивается свободное место. В Elasticsearch есть защитный механизм, называемый watermark (водяной знак).

• Low watermark (по умолчанию 85% заполненности диска) — Elasticsearch перестает выделять новые шарды (shards) на этот узел
• High watermark (по умолчанию 90%) — Elasticsearch пытается переместить шарды с этого узла на другие узлы кластера
• Flood-stage watermark (по умолчанию 95%) — Elasticsearch принудительно устанавливает для индексов на узле флаг read-only

У меня в индексах и логах нет никаких важных данных — так что можно выполнить полную очистку и начать с чистого листа

$ sudo systemctl stop filebeat.service
$ sudo systemctl stop logstash.service
$ sudo systemctl stop kibana.service
$ sudo systemctl stop elasticsearch.service

$ sudo rm -rf /var/lib/elasticsearch/*
$ sudo rm -rf /var/lib/logstash/*
$ sudo rm -rf /var/lib/filebeat/*

$ sudo rm -rf /var/log/elasticsearch/*
$ sudo rm -rf /var/log/logstash/*
$ sudo rm -rf /var/log/filebeat/*
$ sudo rm -rf /var/log/kibana/*

Запускаем службы в правильном порядке

$ sudo systemctl start elasticsearch.service
$ sudo systemctl start kibana.service
$ sudo systemctl start logstash.service
$ sudo systemctl start filebeat.service

Важный момент — при выполнении этой операции ElasticSearch получает новый cluser UUID. Так что нужно получить новый UUID и заменить в файлах конфигурации (мы использовали в файле конфигурации Filebeat).

Чтобы не выполнять все эти команды вручную — давайте напишем скрипт

#!/bin/bash

# Скрипт ПОЛНОСТЬЮ сбрасывает состояние тестового ELK-стека + Filebeat:
# - удаляет данные Elasticsearch
# - удаляет состояние Logstash
# - удаляет состояние Filebeat
# - очищает логи всех сервисов
# - запускает сервисы в правильном порядке
#
# ВНИМАНИЕ! Использовать ТОЛЬКО в тестовой/учебной среде. На боевом
# кластере это уничтожит все данные!

set -e  # При любой ошибке в команде — аварийный выход

echo '=== Stopping services ==='
systemctl stop filebeat.service || echo 'filebeat.service is not running'
systemctl stop logstash.service || echo 'logstash.service is not running'
systemctl stop kibana.service || echo 'kibana.service is not running'
systemctl stop elasticsearch.service || echo 'elasticsearch.service is not running'

echo '=== Cleaning Elasticsearch data and logs ==='
# Полная очистка данных Elasticsearch. После этого Elasticsearch будет как только
# что установленный, без индексов.
if [ -d /var/lib/elasticsearch ]; then
    rm -rf /var/lib/elasticsearch/*
    echo 'Removed /var/lib/elasticsearch/*'
else
    echo 'WARNING: /var/lib/elasticsearch does not exist (check your ES data path)'
fi
# Логи Elastic
if [ -d /var/log/elasticsearch ]; then
    rm -rf /var/log/elasticsearch/*
    echo 'Removed /var/log/elasticsearch/*'
else
    echo 'WARNING: /var/log/elasticsearch does not exist (check your ES log path)'
fi

echo '=== Cleaning Logstash state and logs ==='
# Потенциальное состояние Logstash (например, persistent queues, sincedb и т.п.)
if [ -d /var/lib/logstash ]; then
    rm -rf /var/lib/logstash/*
    echo 'Removed /var/lib/logstash/*'
else
    echo 'WARNING: /var/lib/logstash does not exist (check your Logstash data path)'
fi
# Логи Logstash
if [ -d /var/log/logstash ]; then
    rm -rf /var/log/logstash/*
    echo 'Removed /var/log/logstash/*'
else
    echo 'WARNING: /var/log/logstash does not exist (check your Logstash log path)'
fi

echo '=== Cleaning Filebeat state and logs ==='
# Состояние Filebeat, после этого он забудет все ранее прочитанные файлы логов
if [ -d /var/lib/filebeat ]; then
    rm -rf /var/lib/filebeat/*
    echo 'Removed /var/lib/filebeat/*'
else
    echo 'WARNING: /var/lib/filebeat does not exist (check your Filebeat data path)'
fi
# Логи Filebeat
if [ -d /var/log/filebeat ]; then
    rm -rf /var/log/filebeat/*
    echo 'Removed /var/log/filebeat/*'
else
    echo 'WARNING: /var/log/filebeat does not exist (check your Filebeat log path)'
fi

echo '=== Cleaning Kibana logs ==='
if [ -d /var/log/kibana ]; then
    rm -rf /var/log/kibana/*
    echo 'Removed /var/log/kibana/*'
else
    echo 'WARNING: /var/log/kibana does not exist (check your Kibana log path)'
fi

echo '=== Starting services in correct order ==='
# Сначала Elasticsearch — на него завязаны все остальные службы стэка
systemctl start elasticsearch.service
echo 'Started elasticsearch.service'

# Ждём, пока Elasticsearch реально поднимется (можно подправить timeout)
echo 'Waiting for Elasticsearch to start...'
sleep 20

# Затем Kibana (она использует Elasticsearch)
systemctl start kibana.service
echo 'Started kibana.service'

# Ждём, пока Kibana реально поднимется (можно подправить timeout)
echo 'Waiting for Kibana to start...'
sleep 60

# Затем Logstash (может использовать Elastic как output, читать шаблоны)
systemctl start logstash.service
echo 'Started logstash.service'

# Ждём, пока Logstash реально поднимется (можно подправить timeout)
echo 'Waiting for Logstash to start...'
sleep 10

# В конце Filebeat (ему нужно, чтобы Logstash/Elasticsearch уже слушали)
systemctl start filebeat.service
echo 'Started filebeat.service'

echo '=== Done. ELK stack and Filebeat have been reset to a clean state. ==='

Отправка сообщений в телеграм

Elastalert2 — отдельный Python-инструмент, который сам подключается к Elasticsearch, делает запросы по расписанию и отправляет уведомления.

$ sudo python3 -m venv /opt/elastalert2
$ sudo /opt/elastalert2/bin/pip install elastalert2

Создаем директории для файла конфигурации и правил

$ sudo mkdir /etc/elastalert2
$ sudo mkdir /etc/elastalert2/rules

/etc/elastalert2/
├── config.yaml
└── rules/
    ├── rule-one.yaml
    └── rule-two.yaml

Создаем файл конфигурации

$ sudo nano /etc/elastalert2/config.yaml

# Хост для подключения к Elasticsearch
es_host: localhost

# Порт для подключения к Elasticsearch
es_port: 9200

# Логин и пароль (если включена аутентификация)
# es_username: elastic
# es_password: qwerty

# Имя индекса для служебных данных
writeback_index: elastalert2_status

# Путь к директории с файлами правил
rules_folder: /etc/elastalert2/rules

# Как часто запускать проверку правил
run_every:
  minutes: 1

# Директива чуть-чуть сдвигает окно просмотра в прошлое, чтобы не
# читать совсем уж свежие данные, которые могут быть неполными
buffer_time:
  minutes: 1

# Таймаут подключения к Elasticsearch (секунды)
es_conn_timeout: 20

# Использовать SSL (если Elasticsearch за HTTPS)
use_ssl: false

# Уровень логирования, по умолчанию INFO
logging_level: WARNING

Для кластера ElasticSearch всесто директивы es_host используется директива es_hosts

es_hosts:
  - host: node1.example.com
    port: 9200
  - host: node2.example.com
    port: 9200
  - host: node3.example.com
    port: 9200

Настройки логирования рассмотрим отдельно

$ sudo nano /etc/elastalert2/config.yaml

# ... все прочие настройки ...

logging:
  version: 1
  incremental: false
  disable_existing_loggers: false

  formatters:
    logline:
      format: '%(asctime)s %(levelname)+8s %(name)+20s %(message)s'

  handlers:
    console:
      class: logging.StreamHandler
      formatter: logline
      level: INFO
      stream: ext://sys.stderr

  loggers:
    elastalert:
      level: INFO
      handlers: []
      propagate: true

  '':  # root logger
    level: INFO
    handlers:
      - console
    propagate: false

Что означают отдельные блоки

• formatters — формат строки лога, например 2026-04-19 14:23:57 DEBUG elastalert какое-то сообщение
• handlers — куда писать логи, значение ext://sys.stderr указывает на стандартный поток ошибок stderr
• loggers — настройки для модуля elastalert предписывают перехватывать логи уровня DEBUG и выше, но не обрабатывать самому, а передавать их дальше (propagate: true) — в корневой логгер
• '' (root) — корневой логгер (перехватывает всё, что не поймали другие логеры), мы назначаем ему console handler — то есть всё в итоге попадает в стандартный поток ошибок stderr

Формат одной строки лога

• %(asctime)s — время
• %(levelname)s — уровень (DEBUG, INFO, WARNING)
• %(name)s — имя модуля
• %(message)s — само сообщение

Создаём индекс для служебных данных

$ /opt/elastalert2/bin/elastalert-create-index --config /etc/elastalert2/config.yaml

Создаем службу elastalert2.service

$ sudo nano /etc/systemd/system/elastalert2.service

[Unit]
Description=ElastAlert2
After=network.target

[Service]
Type=simple
User=root
ExecStart=/opt/elastalert2/bin/elastalert --config /etc/elastalert2/config.yaml
Restart=on-failure

[Install]
WantedBy=multi-user.target

Сообщаем системе о новой службе

$ sudo systemctl daemon-reload

Запускаем и добавляем в автозагрузку

$ sudo systemctl enable elastalert2.service
$ sudo systemctl start elastalert2.service

Создаем правило для системного лога

$ sudo nano /etc/elastalert2/rules/syslog-errors.yaml

name: syslog-errors-and-warnings

# Индекс Elastic из которого читаем события и проверяем условие срабатывания
index: syslog-read

# Elastalert ведёт внутренний указатель — до какого момента уже просканировал.
# При следующем запуске читает только новые события начиная с этой отметки, не
# перечитывая всё окно timeframe заново. Значение true нужно только если логи
# приходят с большой задержкой и нужно убедиться что все события были учтены.
scan_entire_timeframe: false

# Срабатывает, когда количество событий превышает num_events за период timeframe
type: frequency
# Минимальное количество событий для срабатывания
num_events: 3
# Период времени (окно) в котором считаем события
timeframe:
  minutes: 10

# Минимальный интервал между повторными алертами по этому правилу, позволяет
# не отправлять сообщения каждую минуту, а только одно сообщение в 10 минут
realert:
  minutes: 10

# Фильтр событий — какие события считаем
filter:
  - query:
      query_string:
        # Сообщение содержит ERROR или WARN, и программа — systemd (program:logger
        # добавлено для тестирования, чтобы иметь возможность использовать logger)
        query: 'message:(ERROR OR WARN) AND program:(systemd OR logger)'

# Способ отправки алерта (debug — только в лог, telegram — в Телеграм и т.д.)
alert:
  - debug

# Использовать только текст из alert_text, без автодобавления полей события
alert_text_type: alert_text_only

# Шаблон текста алерта, {0} {1} {2} — плейсхолдеры из alert_text_args
alert_text: "Хост: {0}\nСообщений ERROR/WARN за 10 минут: {1}\nПоследнее сообщение: {2}"

# Поля события подставляемые в плейсхолдеры шаблона alert_text по порядку
alert_text_args:
  - host
  - num_hits
  - message

Поисковый запрос

{
    "filter": [
        {
            "query": {
                "query_string": {
                    "query": "message:(ERROR OR WARN) AND program:(systemd OR logger)"
                }
            }
        }
    ]
}

Отправим несколько сообщений в системный лог и проверим создание алерта

$ logger -p syslog.err "TEST ERROR message from logger"
$ logger -p syslog.warn "TEST WARN message from logger"
$ logger -p syslog.err "TEST ERROR message from logger"
$ logger -p syslog.warn "TEST WARN message from logger"
$ logger -p syslog.err "TEST ERROR message from logger"
$ logger -p syslog.warn "TEST WARN message from logger"

$ journalctl -u elastalert2 -n 50

апр 20 11:15:16 EVG-UBN elastalert[4010]: INFO:elastalert:Queried rule syslog-errors-and-warnings from 2026-04-20 11:00 MSK to 2026-04-20 11:15 MSK: 4 / 4 hits
апр 20 11:15:16 EVG-UBN elastalert[4010]: INFO:elastalert:Ran syslog-errors-and-warnings from 2026-04-20 11:00 MSK to 2026-04-20 11:15 MSK: 4 query hits (4 already seen), 0 matches, 0 alerts se>
апр 20 11:15:16 EVG-UBN elastalert[4010]: INFO:elastalert:syslog-errors-and-warnings range 883
апр 20 11:15:33 EVG-UBN elastalert[4010]: INFO:elastalert:Background configuration change check run at 2026-04-20 11:15 MSK
апр 20 11:15:33 EVG-UBN elastalert[4010]: INFO:elastalert:Background alerts thread 0 pending alerts sent at 2026-04-20 11:15 MSK
апр 20 11:15:33 EVG-UBN elastalert[4010]: INFO:elastalert:Disabled rules are: []
апр 20 11:15:33 EVG-UBN elastalert[4010]: INFO:elastalert:Sleeping for 59.999777 seconds
апр 20 11:16:18 EVG-UBN elastalert[4010]: INFO:elastalert:Queried rule syslog-errors-and-warnings from 2026-04-20 11:01 MSK to 2026-04-20 11:16 MSK: 10 / 10 hits
апр 20 11:16:18 EVG-UBN elastalert[4010]: INFO:elastalert:Alert for syslog-errors-and-warnings at 2026-04-20T08:15:31.485Z:
апр 20 11:16:18 EVG-UBN elastalert[4010]: INFO:elastalert:Хост: EVG-UBN
апр 20 11:16:18 EVG-UBN elastalert[4010]: Сообщений ERROR/WARN за 10 минут: 10
апр 20 11:16:18 EVG-UBN elastalert[4010]: Последнее сообщение: TEST ERROR message from logger
апр 20 11:16:18 EVG-UBN elastalert[4010]: INFO:elastalert:Ignoring match for silenced rule syslog-errors-and-warnings
апр 20 11:16:18 EVG-UBN elastalert[4010]: INFO:elastalert:Ran syslog-errors-and-warnings from 2026-04-20 11:01 MSK to 2026-04-20 11:16 MSK: 10 query hits (4 already seen), 2 matches, 1 alerts s>
апр 20 11:16:18 EVG-UBN elastalert[4010]: INFO:elastalert:syslog-errors-and-warnings range 900
апр 20 11:16:33 EVG-UBN elastalert[4010]: INFO:elastalert:Background configuration change check run at 2026-04-20 11:16 MSK
апр 20 11:16:33 EVG-UBN elastalert[4010]: INFO:elastalert:Background alerts thread 0 pending alerts sent at 2026-04-20 11:16 MSK
апр 20 11:16:33 EVG-UBN elastalert[4010]: INFO:elastalert:Disabled rules are: []
апр 20 11:16:33 EVG-UBN elastalert[4010]: INFO:elastalert:Sleeping for 59.999863 seconds

Тестирование прошло успешно, можно отправлять сообщения в телеграм

name: syslog-errors-and-warnings
index: syslog-read
scan_entire_timeframe: false

type: frequency
num_events: 3
timeframe:
  minutes: 10
realert:
  minutes: 10

filter:
  - query:
      query_string:
        query: 'message:(ERROR OR WARN) AND program:systemd'

alert:
  - telegram
telegram_bot_token: '.....'
telegram_room_id: '.....'

alert_text_type: alert_text_only
alert_text: "Хост: {0}\nСообщений ERROR/WARN за 10 минут: {1}\nПоследнее сообщение: {2}"
alert_text_args:
  - host
  - num_hits
  - message

Отправлять сообщения в телеграм можно через прокси

$ sudo nano /etc/systemd/system/elastalert2.service

[Unit]
Description=ElastAlert2
After=network.target

[Service]
Environment="http_proxy=http://123.123.123.123:3128"
Environment="HTTP_PROXY=http://123.123.123.123:3128"
Environment="https_proxy=http://123.123.123.123:3128"
Environment="HTTPS_PROXY=http://123.123.123.123:3128"
Environment="no_proxy=localhost,127.0.0.1"
Environment="NO_PROXY=localhost,127.0.0.1"
Type=simple
User=root
ExecStart=/opt/elastalert2/bin/elastalert --config /etc/elastalert2/config.yaml
Restart=on-failure

[Install]
WantedBy=multi-user.target

Сообщаем системе о новой службе

$ sudo systemctl daemon-reload
$ sudo systemctl restart elastalert2.service

Просмотр данных мониторинга

При настройке всех служб ELK-стэка есть возможность включить отправку данных мониторинга в ElasticSearch. А сам ElasticSearch имеет настройку, разрешающую прием этих данных.

# Файл конфигурации ElasticSearch
xpack.monitoring.collection.enabled: true

# Файл конфигурации Logstash
xpack.monitoring.enabled: true
xpack.monitoring.elasticsearch.hosts: ["http://localhost:9200"]

# Файл конфигурации Filebeat
monitoring.enabled: true
monitoring.elasticsearch:
  hosts: ["localhost:9200"]

# Файл конфигурации Kibana
monitoring.kibana.collection.enabled: true

Давайте проверим, что данные реально собираются

GET /_cat/indices/.monitoring-*?v

Данные мониторинга можно посмотреть в интерфейсе Kibana — Меню → Management → Stack Monitoring

Поиск: API • HTTP • HTTPS • Linux • База данных • Поиск • Сервер • Индекс • Elasticsearch • Kibana • Logstash

• Web-разработка
  • HTML и CSS
  • JavaScript и TypeScript
  • PHP и MySQL
  • CMS Битрикс
  • CMS WoprdPress
  • Yii2 и Laravel
  • Python и Flask
  • Web-аналитика
  • Разное
• ОС Linux
  • Команды
  • Сценарии
  • Сервисы
  • Разное
• 1С:Предприятие
  • Программирование
  • Язык запросов
  • Разное
• Разное
  • Локальная сеть
  • Windows

• 1С:Предприятие (31)
• API (32)
• Bash (44)
• CLI (131)
• CMS (139)
• CSS (50)
• Frontend (75)
• HTML (66)
• JavaScript (150)
• Laravel (72)
• Linux (187)
• MySQL (77)
• PHP (126)
• React.js (66)
• SSH (27)
• Ubuntu (73)
• Web-разработка (509)
• WordPress (73)
• Yii2 (69)
• БазаДанных (100)
• Битрикс (66)
• Блог (29)
• Верстка (43)
• ИнтернетМагаз… (84)
• КаталогТоваров (87)
• Класс (30)
• Клиент (35)
• Ключ (32)
• Команда (92)
• Компонент (60)
• Конфигурация (82)
• Корзина (32)
• ЛокальнаяСеть (33)
• Модуль (35)
• Навигация (31)
• Настройка (160)
• ПанельУправле… (29)
• Плагин (33)
• Пользователь (27)
• Практика (101)
• Сервер (96)
• Событие (28)
• Теория (106)
• Установка (82)
• Файл (53)
• Форма (58)
• Фреймворк (192)
• Функция (36)
• ШаблонСайта (68)

• Web-разработка
• 1С:Предприятие
• ОС Linux
• Регулярные выражения
• Разное