Использование кастомного конвейера загрузки для Filebeat Netflow

Содержание

Вопрос или проблема
Ответ или решение
Использование пользовательского канала обработки данных для Filebeat Netflow
Настройка канала обработки данных
Решение проблемы с "enrich coordination queue at capacity"
Заключение

Вопрос или проблема

Я использую Filebeat v8.8.0 и пытаюсь загружать данные Netflow через пользовательский конвейер загрузки для выполнения геолокации IP с использованием базы данных, отличной от Maxmind. В данный момент мои данные Netflow обогащаются с помощью базы данных Maxmind (я не указывал никаких настроек в Filebeat или ES, кроме загрузки баз данных Maxmind в соответствующие папки для ES).

Я уже создал конвейер, названный new_filebeat_pipeline. Однако я не уверен, где мне следует указать конвейер в моих конфигурационных файлах Filebeat. Я пробовал добавить pipeline: "new_filebeat_pipeline" в следующих 3 местах (отдельно), но мои данные всегда обогащаются данными Maxmind, а не моей пользовательской базой данных.

Настройка 1: В filebeat.yml

output.elasticsearch:
  ...
  pipeline: "new_filebeat_pipeline"

Настройка 2: В filebeat.yml

filebeat.inputs:
- type: netflow
   enabled: true
   pipeline: "new_filebeat_pipeline"

Настройка 3: В /etc/filebeat/modules.d/netflow.yml

- module: netflow
  log:
    enabled: true
    var: 
      ...
      input.pipeline: "new_filebeat_pipeline" # также пробовал 'pipeline: "new_filebeat_pipeline"'

Я всегда использовал netflow.yml для настройки своих данных Netflow, а не filebeat.inputs, т.е. Настройка 2 была добавлена только для этого теста и в противном случае не существует.

Я протестировал конвейер, используя существующий документ (уже содержащий обогащение Maxmind), и он смог сгенерировать поля, ожидаемые из новой базы данных, так что конвейер должен работать правильно. Но по какой-то причине мои данные Netflow все еще проходят через “стандартный” конвейер, а не через мой пользовательский конвейер.

Как мне следует настроить Filebeat для этого?

Дополнение: Я прочитал здесь, что “Этот модуль оборачивает входные данные netflow для обогащения записей потока геолокационной информацией о конечных IP-адресах с помощью конвейера загрузки Elasticsearch.” Так что я предположил, что если я использую модуль netflow, он всегда будет отправляться в стандартный конвейер загрузки. Поэтому я отключил его и вместо этого использовал filebeat.inputs, и теперь я вижу поля, добавленные с помощью пользовательского конвейера загрузки.

Хотя теперь я сталкиваюсь с другой проблемой, а именно, я получаю сообщение об ошибке Не удалось выполнить обогащение, очередь координации обогащения достигла емкости [1024/1024].

Ответ или решение

Использование пользовательского канала обработки данных для Filebeat Netflow

Ваша ситуация с использованием Filebeat версии 8.8.0 и желанием инжектировать данные Netflow через кастомный канал обработки данных для выполнения геолокации IP, используя базу данных, отличную от Maxmind, требует внимательного подхода к настройкам. Давайте рассмотрим, как правильно настроить Filebeat и решить вашу проблему.

Настройка канала обработки данных

Проверка конфигурации канала: Убедитесь, что ваш пользовательский канал обработки данных new_filebeat_pipeline действительно правильно создан и настроен. Можно сделать это через API Elasticsearch с помощью команды:
```
GET _ingest/pipeline/new_filebeat_pipeline
```
Убедитесь, что там указаны все необходимые процессоры для вашей базы данных геолокации.
Правильное указание канала в конфигурации Filebeat:
- Настройка в filebeat.yml:
  Ваша первоначальная попытка указания канала обработки в секции output.elasticsearch выглядит правильно:
```
output.elasticsearch:
...
pipeline: "new_filebeat_pipeline"
```
- Использование filebeat.inputs: Раздел в конфигурации filebeat.inputs тоже актуален, особенно если вы отключили модуль Netflow:
```
filebeat.inputs:
- type: netflow
enabled: true
pipeline: "new_filebeat_pipeline"
```
- Модуль Netflow: Если вы используете модуль Netflow, он по умолчанию использует стандартный канал обработки, который включает в себя Maxmind. Если модуль включен, ваши попытки указать пользовательский канал в файле /etc/filebeat/modules.d/netflow.yml будут проигнорированы. Чтобы избежать этого, убедитесь, что данный модуль отключен, если вы хотите использовать пользовательские настройки.

Решение проблемы с "enrich coordination queue at capacity"

Ошибка Could not perform enrichment, enrich coordination queue at capacity [1024/1024] указывает на то, что очередь для обогащения данных заполнена. Это может быть связано с высокой нагрузкой, а также недостаточными ресурсами для обработки входящих данных. Рассмотрим возможные решения:

Увеличение размера очереди:
Увеличьте размер очереди для обогащения, добавив/изменив следующие настройки в вашем Elasticsearch elasticsearch.yml:
```
ingest.enrich.lookup.size: 2048
```
Это увеличит вместимость очереди и позволит системе обрабатывать больше запросов одновременно.
Увеличение ресурсов Elasticsearch:
Проверьте, достаточно ли у вашего Elasticsearch ресурсов (CPU и памяти). При необходимости изучите возможность масштабирования кластера или добавления дополнительных нод.
Оптимизация пользовательского канала:
Пересмотрите собственный обработчик для уменьшения его времени выполнения. Оптимизация кодов или алгоритмов может значительно повысить производительность.
Проверка Rate Limiting: Если существует какое-либо ограничение по скорости на уровне сети или хранилища, это также может привести к перегрузке. Убедитесь, что ваша архитектура не имеет бутылочного горлышка, которое дизбалансирует поток данных.

Заключение

Для настройки Filebeat с пользовательским каналом обработки данных, обеспечивающего применение вашей базы данных геолокации, важно правильно настроить каналы в конфигурациях и учесть ресурсы вашего Elasticsearch. При выполнении вышеуказанных рекомендаций вы должны быть в состоянии эффективно обрабатывать и обогащать данные Netflow. Если у вас возникнут дополнительные вопросы или сложности, не стесняйтесь обращаться за помощью к сообществу Elastic или профессионалам в этой области.