Передо мной встала задача сбора логов с парка серверов на ОС Windows и ОС Linux. Для того чтобы решить её я воспользовался стэком OpenSearch. Во время настройки OpenSearch мне не хватало в открытых источниках наглядных примеров, а информация на официальных сайтах ElasticSearch и OpenSearch мне показалась обрывочной, слабо привязанной к реальным ситуациям. Поэтому я решил поделиться своим опытом и описать основные моменты установки и некоторые сценарии настройки и применения стэка OpenSearch, которые я применил в своей практике.
Содержание
1. Введение
1.1. Коротко о том, что такое OpenSearch
1.2. Коротко о форках Elasticsearch
1.3. Что и зачем будем настраивать
1.4. Настраиваемая схема
2. Установка стэка OpenSearch
2.1. Подготовка Linux машины Node OpenSearch
2.2. Установка OpenSearch (аналог ElasticSearch)
2.3. Настройки производительности
2.4. Первый запуск OpenSearch
2.5. Смена пароля учетной записи OpenSearch
2.6. Переводим OpenSearch в режим кластера
2.7. Установка OpenSearch-Dashboards (аналог Kibana)
2.8. Установка NGINX и настройка переадресации порта OpenSearch-Dashboards
2.9. Настройка файрволла на запрет доступа по порту 5601
2.10. Установка Logstash-oss-with-OpenSearch-output-plugin
3. Установка Beat’ов
3.1. Установка Filebeat на Linux
3.2. Установка Filebeat на Windows
3.3. Установка Winlogbeat на Windows
4. Настройка на примере нескольких кейсов
4.1. Сценарий сбора данных в формате json
4.1.1. Исходные данные
4.1.2. Процесс сбора данных через Filebeat
4.1.3. Процесс обработки данных в Logstash
4.2. Сценарий сбора однострочных данных в текстовом формате
4.2.1. Исходные данные
4.2.2. Процесс сбора данных через Filebeat
4.2.3. Процесс обработки данных в Logstash
4.3. Сценарий сбора Логов IIS
4.3.1. Процесс сбора данных через Filebeat
4.3.2. Процесс обработки данных в Logstash
4.4. Сценарий сбора системных Логов Windows
4.4.1. Процесс сбора данных через Winlogbeat
4.4.2. Процесс обработки данных в Logstash
4.5. Сценарий сбора системных Логов Linux
4.5.1. Процесс сбора данных через Filebeat
4.5.2. Процесс обработки данных в Logstash
5. Итоговые файлы конфигураций
6. Эксплуатация
6.1. Создание шаблона индексов с помощью OpenSearch-Dashboards
6.2. Отображение данных в OpenSearch-Dashboards
6.3. Поиск в OpenSearch-Dashboards
6.4. Работа с OpenSearch через консоль (без OpenSearch-Dashboards)
7. Актуальные проблемы
7.1. Имена Topic’ов Kafka не должны содержать большие буквы
7.2. Filebeat некорректно создает Topic’и Kafka
7.3. Требуется переход на следующую строку после json сообщения
7.4. Необходимо обращать внимание на кодировку log-файлов
8. Послесловие
1.1. Коротко о том, что такое OpenSearch
1.2. Коротко о форках Elasticsearch
1.3. Что и зачем будем настраивать
1.4. Настраиваемая схема
2. Установка стэка OpenSearch
2.1. Подготовка Linux машины Node OpenSearch
2.2. Установка OpenSearch (аналог ElasticSearch)
2.3. Настройки производительности
2.4. Первый запуск OpenSearch
2.5. Смена пароля учетной записи OpenSearch
2.6. Переводим OpenSearch в режим кластера
2.7. Установка OpenSearch-Dashboards (аналог Kibana)
2.8. Установка NGINX и настройка переадресации порта OpenSearch-Dashboards
2.9. Настройка файрволла на запрет доступа по порту 5601
2.10. Установка Logstash-oss-with-OpenSearch-output-plugin
3. Установка Beat’ов
3.1. Установка Filebeat на Linux
3.2. Установка Filebeat на Windows
3.3. Установка Winlogbeat на Windows
4. Настройка на примере нескольких кейсов
4.1. Сценарий сбора данных в формате json
4.1.1. Исходные данные
4.1.2. Процесс сбора данных через Filebeat
4.1.3. Процесс обработки данных в Logstash
4.2. Сценарий сбора однострочных данных в текстовом формате
4.2.1. Исходные данные
4.2.2. Процесс сбора данных через Filebeat
4.2.3. Процесс обработки данных в Logstash
4.3. Сценарий сбора Логов IIS
4.3.1. Процесс сбора данных через Filebeat
4.3.2. Процесс обработки данных в Logstash
4.4. Сценарий сбора системных Логов Windows
4.4.1. Процесс сбора данных через Winlogbeat
4.4.2. Процесс обработки данных в Logstash
4.5. Сценарий сбора системных Логов Linux
4.5.1. Процесс сбора данных через Filebeat
4.5.2. Процесс обработки данных в Logstash
5. Итоговые файлы конфигураций
6. Эксплуатация
6.1. Создание шаблона индексов с помощью OpenSearch-Dashboards
6.2. Отображение данных в OpenSearch-Dashboards
6.3. Поиск в OpenSearch-Dashboards
6.4. Работа с OpenSearch через консоль (без OpenSearch-Dashboards)
7. Актуальные проблемы
7.1. Имена Topic’ов Kafka не должны содержать большие буквы
7.2. Filebeat некорректно создает Topic’и Kafka
7.3. Требуется переход на следующую строку после json сообщения
7.4. Необходимо обращать внимание на кодировку log-файлов
8. Послесловие
1. Введение
1.1. Коротко о том, что такое OpenSearch
OpenSearch – система полнотекстового поиска. Является форком Elasticsearch. Разработана компанией A9.com, дочерней компанией Amazon.com.
Стэк OpenSearch (OpenSearch + Logstash-oss-with-OpenSearch-output-plugin + OpenSearch–Dashboards) является бесплатным аналогом с отрытым исходным кодом стэку ELK (Elasticsearch + Logstash + Kibana). В совокупности с Beat’ами (Filebeat, Winlogbeat, и т.д.) образуют полный цикл управления логами: сбор, систематизация, поиск.
Можно найти и множество других применений этому стэку, но в этой статье речь пойдет главным образом именно о логах.
1.2. Коротко о форках Elasticsearch
Elasticsearch развивается под этим названием с 2010 года. С 2018 года, с версии 6.3 лицензия на ELK меняется, появляется платная и бесплатная версия. Позже Amazon создает форк Elasticsearch и называет его «Open Distro for Elasticsearch». А в 2021 году Amazon создает новый форк и называет его «OpenSearch». Немного позже Российская компания Arenadata становится официальным партнёром OpenSearch в России и выпускает отечественный форк под названием «Arenadata LogSearch (ADLS)» под платной лицензией.
1.3. Что и зачем будем настраивать
Стэк OpenSearch мы будем разворачивать из архивов, это особенно актуально в закрытых средах. Кроме того, мы не будем использовать Kubernetes и Docker, то есть будем производить установку в классической среде.
В качестве ОС под OpenSearch будем использовать Linux Ubuntu 20.04.
Логи будут транспортироваться с серверов, на которых они возникли на сервер OpenSearch в виде сообщений. Для того, чтобы гарантировать доставку сообщений, в качестве транспорта мы будем использовать сервис Apache Kafka.
Apache Kafka – это распределённый программный брокер сообщений, который гарантирует доставку сообщений.
Мы введем OpenSearch в кластер, но будет описан ввод только основной ноды.
Кластер OpenSearch позволяет решать сразу несколько задач. Во-первых, традиционно, кластер решает задачу масштабирования, распределения нагрузки и отказоустойчивости. Во-вторых, данные накопленные в OpenSearch, можно распределять по их значимости на «Горячие» («Hot»), «Теплые» («Warm») и «Холодные» («Cold»). А ноды кластера можно настроить так чтобы они принимали только «Горячие» данные, или «Теплые» данные, или только «Холодные». И если на ноду под «Холодные» данные выделять менее производительные мощности, то это позволит сэкономить ресурсы.
1.4. Настраиваемая схема
На серверы, с которых нужно собирать логи, мы установим Beat’ы. На серверы под управлением ОС Windows мы установим Filebeat и Winlogbeat. На серверы под управлением Linux мы установим только Filebeat. Beat’ы будут отправлять сообщения с логами в Kafk’у. Logstash будет брать эти сообщения из Kafka, обрабатывать их и отправлять в OpenSearch. Пользователь будет осуществлять просмотр и поиск по логам через OpenSearch-Dashboards. Схема взаимодействия представлена на Рис.1.
Рис.1
Договоримся, что машина Node OpenSearch будет иметь:
1. Имя «server-elk01»;
2. Ip адрес «10.0.0.70».
Машина с Kafka будет иметь:
1. Ip адрес «10.0.0.60».
2. Установка стэка OpenSearch
2.1. Подготовка Linux машины Node OpenSearch
Перед началом установки основных пакетов установим Java и Unzip, а также создадим пользователей для OpenSearch и Logstash.
1. Установка Java:
apt install openjdk-11-jdk2. Установка unzip:
apt install unzip3. Создание пользователя opensearch:
groupadd opensearch
useradd opensearch -g opensearch -M -s /bin/bash
passwd opensearch4. Создание пользователя logstash:
groupadd logstash
useradd logstash -g logstash -M -s /bin/bash
passwd logstash2.2. Установка OpenSearch (аналог ElasticSearch)
OpenSearch, как можно было заметить, является ядром всего стэка. OpenSearch содержит базу данных, в которой будут храниться логи. Кроме того, OpenSearch имеет API для обработки запросов как на ввод данных, так и на вывод. Так же OpenSearch индексирует поступившие в неё данные и осуществляет поиск по этим данным.
Приступаем к установке.
1. Переходим на официальный сайт OpenSearch (https://opensearch.org/downloads.html) и скачиваем архив tar.gz нужной версии. В этой статье я буду использовать OpenSearch версии 1.2.4 (https://artifacts.opensearch.org/releases/bundle/opensearch/1.2.4/opensearch-1.2.4-linux-x64.tar.gz). После скачивания перенесите архив на сервер «server-elk01» в удобный для вас каталог и перейдите в него в консоли.
2. Даем права на выполнение для архива:
chmod +x opensearch-1.2.4-linux-x64.tar.gz3. Распаковываем архив:
tar -xf opensearch-1.2.4-linux-x64.tar.gz4. Будем устанавливать OpenSearch в каталог «/opt/opensearch», поэтому создаем рабочий каталог для OpenSearch:
mkdir /opt/opensearch5. Переносим распакованные данные в рабочий каталог:
mv ./opensearch-1.2.4/* /opt/opensearch6. Удаляем каталог, оставшийся от распаковки:
rmdir ./opensearch-1.2.47. Делаем пользователя opensearch владельцем рабочего каталога OpenSearch:
chown -R opensearch:opensearch /opt/opensearch8. Запускает установочный скрипт от имени пользователя opensearch:
sudo -u opensearch /opt/opensearch/opensearch-tar-install.shДожидаемся сообщения «Node 'server-elk01' initialized» и нажимаем «Ctrl+C».
9. Создаем файл демона для работы OpenSearch:
nano /lib/systemd/system/opensearch.service
/lib/systemd/system/opensearch.service
[Unit]
Description=Opensearch
Documentation=https://opensearch.org/docs/latest
Wants=network-online.target
After=network-online.target
[Service]
Type=simple
RuntimeDirectory=opensearch
PrivateTmp=true
Restart=on-failure
RestartSec=60s
WorkingDirectory=/opt/opensearch
User=opensearch
Group=opensearch
ExecStart=/opt/opensearch/bin/opensearch
StandardOutput=journal
StandardError=inherit
# Specifies the maximum file descriptor number that can be opened by this process
LimitNOFILE=65535
# Specifies the maximum number of processes
LimitNPROC=4096
# Specifies the maximum size of virtual memory
LimitAS=infinity
# Specifies the maximum file size
LimitFSIZE=infinity
# Not use SWAP
LimitMEMLOCK=infinity
# Disable timeout logic and wait until process is stopped
TimeoutStopSec=0
# Allow a slow startup before the systemd notifier module kicks in to extend the timeout
TimeoutStartSec=75
[Install]
WantedBy=multi-user.targetЗа основу файла демона я взял файл оригинального демона Elasticsearch и переработал его.
10. Делаем пользователя root владельцем файла демона:
chown -R root:root /lib/systemd/system/opensearch.service11. Перечитаем конфигурацию systemd, чтобы система обнаружила вновь созданный демон:
systemctl daemon-reload12. Создаем каталог для логов opensearch:
mkdir /var/log/opensearch13. Делаем пользователя opensearch владельцем каталога логов:
chown -R opensearch /var/log/opensearch2.3. Настройки производительности
Для того чтобы получить высокую производительность недостаточно иметь хорошее железо, нужно еще и правильно настроить систему. Попробуем оптимизировать несколько параметров.
1. Настройки Java:
Производятся в файле «/opt/opensearch/config/jvm.options». Необходимо настроить два параметра: «Xmx» и «Xms».
Оба параметра рекомендуется установить в значение равное 50% от имеющейся физической памяти узла. Чем больше доступно памяти, тем лучше, но есть ограничения. Значения параметров «Xmx» и «Xms» не должны превышать значение параметра JVM «compressed object pointers» (по умолчанию равен 32 Gb). И, значения параметров «Xmx» и «Xms» не должны превышать значение параметра JVM «zero-based compressed oops» (по умолчанию равен 26 Gb).
У меня на машине «server-elk01» установлено 8 Gb оперативной памяти поэтому настройки получились такие:
nano /opt/opensearch/config/jvm.options
/opt/opensearch/config/jvm.options
...
-Xms4g
-Xmx4g
...2. Настройка виртуальной памяти:
nano /etc/sysctl.conf
/etc/sysctl.conf
...
# Добавить в конце файла
vm.max_map_count=262144Для того, чтобы изменения вступили в силу без перезагрузки хоста можно выполнить следующую команду в консоли:
sysctl -w vm.max_map_count=2621442.4. Первый запуск OpenSearch
На этом этапе уже все готово к тому, чтобы запустить OpenSearch. Глубокой настройкой займемся позже, а сейчас просто запустим OpenSearch.
1. Запустим настроенный нами демон OpenSearch:
systemctl start opensearch2. Проверим статус запуска демона OpenSearch:
systemctl status opensearch3. Настроим автозапуск демона OpenSearch:
systemctl enable opensearch4. Проверим работоспособность демона OpenSearch:
curl -X GET https://localhost:9200 -u 'admin:admin' --insecureЕсли вы устанавливаете OpenSearch в ознакомительных целях, то можете считать установку законченной и приступать к использованию OpenSearch, правда только в консольном режиме, Web-интерфейс (OpenSearch-Dashboards) мы еще не устанавливали.
2.5. Смена пароля учетной записи OpenSearch
По умолчанию в OpenSearch предустановлена админская учетная запись «admin» с паролем «admin», поэтому для повышения уровня безопасности сменим пароль этой учетной записи. Для этого выполним следующие шаги.
1. Остановим демон OpenSearch:
systemctl stop opensearch2. Дадим права на выполнение скрипта получения хэша пароля:
chmod +x /opt/opensearch/plugins/opensearch-security/tools/hash.sh3. Запускаем скрипт и вводим новый пароль «yN-3L(GMmAAw»:
/opt/opensearch/plugins/opensearch-security/tools/hash.shЗапоминаем полученный хэш, например:
$2y$12$OCvWNlMu8VbOarfdXdcjPOnHarqktJIcTYjwoykXdaJJfjcCTmfXOК слову, я указал реальный хэш к описанному паролю, в тестах можно их использовать.
4. Заменим хэш пароля пользователя admin:
nano /opt/opensearch/plugins/opensearch-security/securityconfig/internal_users.yml
/opt/opensearch/plugins/opensearch-security/securityconfig/internal_users.yml
...
admin:
hash: "$2y$12$OCvWNlMu8VbOarfdXdcjPOnHarqktJIcTYjwoykXdaJJfjcCTmfXO"
...5. Дадим права на выполнение скрипта для создания новых сертификатов:
chmod +x /opt/opensearch/plugins/opensearch-security/tools/securityadmin.sh6. Перейдём в каталог со скриптом:
cd /opt/opensearch/plugins/opensearch-security/tools7. Запустим демон OpenSearch:
systemctl start opensearch8. Запускаем скрипт для создания новых сертификатов:
./securityadmin.sh -cd ../securityconfig/ -icl -nhnv \
-cacert ../../../config/root-ca.pem \
-cert ../../../config/kirk.pem \
-key ../../../config/kirk-key.pem9. Проверяем работоспособность OpenSearch с новым паролем:
curl -X GET https://localhost:9200 -u 'admin:yN-3L(GMmAAw' --insecureАналогичным образом можно сменить пароль и у других учетных записей, которые можно обнаружить в «/opt/opensearch/plugins/opensearch-security/securityconfig/internal_users.yml», если вам это необходимо.
2.6. Переводим OpenSearch в режим кластера
Как я писал выше в реализуемой схеме мы введем OpenSearch в кластер, но введем только одну ноду. Потому как задачи перед кластером можно поставить разные то и настройки, и схемы могут сильно отличаться. Если вам потребуется только одна машина с OpenSearch, то можно и не вводить её в кластер, тогда этот шаг можно пропустить. А мы приступим.
1. Остановим демон OpenSearch:
systemctl stop opensearch2. Добавляем параметры в настройки OpenSearch:
nano /opt/opensearch/config/opensearch.yml
/opt/opensearch/config/opensearch.yml
# ------------------------------------ Node ------------------------------------
# Имя ноды:
node.name: os-node01-server-elk01
# Роли узла:
node.roles: [ master, data ]
#
# ---------------------------------- Network -----------------------------------
# Адрес узла - принимать на любых адресах:
network.host: 0.0.0.0
# Порт:
http.port: 9200
#
# ---------------------------------- Cluster -----------------------------------
# Имя кластера:
cluster.name: os_cluster
# Список узлов в голосовании по выбору master узла:
cluster.initial_master_nodes: ["os-node01"]
#
# --------------------------------- Discovery ----------------------------------
# Список master узлов кластера:
discovery.seed_hosts: ["10.0.0.70"]
#
# ----------------------------------- Paths ------------------------------------
# Директория с данными:
path.data: /opt/opensearch/data
# Директория с логами:
path.logs: /var/log/opensearch
#
######## Start OpenSearch Security Demo Configuration ########
# WARNING: revise all the lines below before you go into production
plugins.security.ssl.transport.pemcert_filepath: esnode.pem
plugins.security.ssl.transport.pemkey_filepath: esnode-key.pem
plugins.security.ssl.transport.pemtrustedcas_filepath: root-ca.pem
plugins.security.ssl.transport.enforce_hostname_verification: false
plugins.security.ssl.http.enabled: true
plugins.security.ssl.http.pemcert_filepath: esnode.pem
plugins.security.ssl.http.pemkey_filepath: esnode-key.pem
plugins.security.ssl.http.pemtrustedcas_filepath: root-ca.pem
plugins.security.allow_unsafe_democertificates: true
plugins.security.allow_default_init_securityindex: true
plugins.security.authcz.admin_dn:
- CN=kirk,OU=client,O=client,L=test, C=de
plugins.security.authcz.type: internal_opensearch
plugins.security.enable_snapshot_restore_privileges: true
plugins.security.check_snapshot_restore_write_privileges: true
plugins.security.restapi.roles_enabled: ["all_access", "security_rest_api_access"]
plugins.security.system_indices.enabled: true
plugins.security.system_indices.indices: [".opendistro-alerting-config", ".opendistro-alerting-alert*", ".opendistro-anomaly-results*", ".opendistro-anomaly-detector*", ".opendistro-anomaly-checkpoints", ".opendistro-anomaly-detection-state", ".opendistro-reports-*", ".opendistro-notifications-*", ".opendistro-notebooks", ".opendistro-observability", ".opendistro-asynchronous-search-response*", ".opendistro-metadata-store"]
node.max_local_storage_node: 3
######## End OpenSearch Security Demo Configuration ########Обращаем внимание на роль ноды (master, data). Роль «master» — означает, что нода управляет кластером. Роль «data» — означает, что нода содержит базу данных.
3. Запускаем демон OpenSearch:
systemctl start opensearch4. Проверим работоспособность OpenSearch:
curl -X GET https://localhost:9200 -u 'admin:yN-3L(GMmAAw' --insecure5. Проверим состояние кластера:
curl -X GET https://localhost:9200/_cluster/health?pretty -u 'admin:yN-3L(GMmAAw' --insecure6. Выясним кто мастер в кластере:
curl -X GET https://localhost:9200/_cat/master?pretty -u 'admin:yN-3L(GMmAAw' --insecure2.7. Установка OpenSearch-Dashboards (аналог Kibana)
OpenSearch-Dashboards является Web-интерфейсом для OpenSearch. Для тех кто собирается работать с OpenSearch из консоли этот шаг можно и пропустить. Так же вместо OpenSearch-Dashboards в качестве Web-интерфейса можно установить и другие утилиты, например, «Grafana».
Приступим к установке OpenSearch-Dashboards.
1. Переходим на официальный сайт OpenSearch (https://opensearch.org/downloads.html) и скачиваем архив tar.gz нужной версии. В этой статье я буду использовать OpenSearch-Dashboards версии 1.2.0 (https://artifacts.opensearch.org/releases/bundle/opensearch-dashboards/1.2.0/opensearch-dashboards-1.2.0-linux-x64.tar.gz). После скачивания перенесите архив на сервер «server-elk01» в удобный для вас каталог и перейдите в него в консоли.
2. Даем права на выполнение для архива:
chmod +x opensearch-dashboards-1.2.0-linux-x64.tar.gz3. Распаковываем архив:
tar -xf opensearch-dashboards-1.2.0-linux-x64.tar.gz4. Будем устанавливать OpenSearch-Dashboards в каталог «/opt/opensearch-dashboards», поэтому создаем рабочий каталог для OpenSearch-Dashboards:
mkdir /opt/opensearch-dashboards5. Переносим распакованные данные в рабочий каталог:
mv ./opensearch-dashboards-1.2.0-linux-x64/* ./opensearch-dashboards-1.2.0-linux-x64/.* /opt/opensearch-dashboards/6. Удаляем каталог, оставшийся от распаковки:
rmdir ./opensearch-dashboards-1.2.0-linux-x647. Делаем пользователя opensearch владельцем рабочего каталога OpenSearch-Dashboards:
chown -R opensearch:opensearch /opt/opensearch-dashboardsПо умолчанию OpenSearch-Dashboards доступен по порту 5601. Можно изменить порт в настройках OpenSearch-Dashboards, однако тогда придется запускать сервис под учетной записью root. Поэтому оставим порт по умолчанию и позже настроим переадресацию через NGINX.
Для информации, для смены порта по умолчанию необходимо внести следующие изменения:
/opt/opensearch-dashboards/config/opensearch_dashboards.yml
nano /opt/opensearch-dashboards/config/opensearch_dashboards.yml...
# Добавить в конец файла
server.port: 80
server.host: 10.0.0.708. Создаем файл демона для работы OpenSearch-Dashboards:
nano /lib/systemd/system/opensearch_dashboards.service
/lib/systemd/system/opensearch_dashboards.service
[Unit]
Description=Opensearch_dashboards
Documentation=https://opensearch.org/docs/latest
Wants=network-online.target
After=network-online.target
[Service]
Type=simple
RuntimeDirectory=opensearch_dashboards
PrivateTmp=true
WorkingDirectory=/opt/opensearch-dashboards
User=opensearch
Group=opensearch
ExecStart=/opt/opensearch-dashboards/bin/opensearch-dashboards
StandardOutput=journal
StandardError=inherit
# Specifies the maximum file descriptor number that can be opened by this process
LimitNOFILE=65535
# Specifies the maximum number of processes
LimitNPROC=4096
# Specifies the maximum size of virtual memory
LimitAS=infinity
# Specifies the maximum file size
LimitFSIZE=infinity
# Disable timeout logic and wait until process is stopped
TimeoutStopSec=0
# Allow a slow startup before the systemd notifier module kicks in to extend the timeout
TimeoutStartSec=75
[Install]
WantedBy=multi-user.target9. Делаем пользователя root владельцем файла демона:
chown -R root:root /lib/systemd/system/opensearch_dashboards.service10. Перечитаем конфигурацию systemd, чтобы система обнаружила вновь созданный демон:
systemctl daemon-reload11. Запустим настроенный нами демон OpenSearch-Dashboards:
systemctl start opensearch_dashboards12. Проверим статус запуска демона OpenSearch-Dashboards:
systemctl status opensearch_dashboards13. Настроим автозапуск демона OpenSearch-Dashboards:
systemctl enable opensearch_dashboards2.8. Установка NGINX и настройка переадресации порта OpenSearch-Dashboards
OpenSearch-Dashboards работает на порту 5601, но это неудобно, так как портом для Web-ресурсов по умолчанию является порт 80. Мы установим NGINX и настроим его так, чтобы он слушал 80 порт и перенаправлял с него все запросы на порт 5601. То есть NGINX будет работать в режиме прокси. Приступаем.
1. Установим Nginx на хост:
apt install nginx2. Настроим автозапуск Nginx:
systemctl enable nginx3. Не будем удалять, а сделаем резервную копию настроек Nginx по умолчанию:
cp /etc/nginx/sites-enabled/default /etc/nginx/sites-enabled/default.sav4. Настроим переадресацию OpenSearch-Dashboards с порта 5601 на 80:
nano /etc/nginx/sites-enabled/default
/etc/nginx/sites-enabled/default
map $http_upgrade $connection_upgrade { # WebSocket support
default upgrade;
'' '';
}
server {
listen 80;
server_name server-elk01 server-elk01.domain.ad 10.0.0.70;
location / {
proxy_pass http://127.0.0.1:5601; # full internal address
proxy_http_version 1.1;
proxy_set_header Host $server_name:$server_port;
proxy_set_header X-Forwarded-Host $http_host; # necessary for proper absolute redirects and TeamCity CSRF check
proxy_set_header X-Forwarded-Proto $scheme;
proxy_set_header X-Forwarded-For $remote_addr;
proxy_set_header Upgrade $http_upgrade; # WebSocket support
proxy_set_header Connection $connection_upgrade; # WebSocket support
}
}5. Активируем новые настройки Nginx без перезагрузки демона:
nginx -s reloadВ случае неудачи перезагрузим Nginx:
service nginx reload2.9. Настройка файрволла на запрет доступа по порту 5601
После того как мы настроили на NGINX переадресацию с 80 порта на порт 5601 сам порт 5601 продолжит принимать запросы. На этом шаге мы закроем порт 5601 файерволлом. Это, конечно, не обязательно, но, я считаю, что так будет больше порядка.
Само правило для файрволла не сложное, а вариантов как сделать так, чтобы после перезагрузки системы правило активировалось вновь можно придумать много. Я пойду по пути создания демона, который будет запускать bash-скрипт «/etc/rc.local». А сам скрипт будет содержать команду добавления правила в файрволл. Приступим.
1. Правило привяжем к сетевому интерфейсу, поэтому сначала необходимо узнать название сетевого интерфейса, для этого воспользуемся командой:
ip link showили командой:
ifconfigВ моей системе сетевой интерфейс имеет имя «ens160».
2. Создадим Bash-скрипт с правилом файрволла:
nano /etc/rc.local
/etc/rc.local
#!/bin/bash
iptables -A INPUT -i ens160 -p tcp --dport 5601 -j DROP
exit 0Соответственно имя интерфейса нужно изменить на имя своего интерфейса.
3. Дадим скрипту права на выполнение:
chmod +x /etc/rc.local4. Делаем пользователя root владельцем скрипта:
chown -R root:root /etc/rc.local5. Создадим файл демона для запуска скрипта:
nano /etc/systemd/system/rc-local.service
/etc/systemd/system/rc-local.service
[Unit]
Description=/etc/rc.local Compatibility
ConditionPathExists=/etc/rc.local
[Service]
Type=forking
ExecStart=/etc/rc.local start
TimeoutSec=0
StandardOutput=tty
RemainAfterExit=yes
SysVStartPriority=99
[Install]
WantedBy=multi-user.target6. Зададим права на файл демона:
chmod 644 /etc/systemd/system/rc-local.service7. Делаем пользователя root владельцем файла демона:
chown -R root:root /etc/systemd/system/rc-local.service8. Перечитаем конфигурацию systemd, чтобы система обнаружила вновь созданный демон:
systemctl daemon-reload9. Запустим настроенный нами демон rc-local:
systemctl start rc-local10. Проверим статус запуска демона rc-local:
systemctl status rc-local11. Настроим автозапуск демона rc-local:
systemctl enable rc-local2.10. Установка Logstash-oss-with-OpenSearch-output-plugin
Logstash в реализуемой схеме выполняет несколько задач. Во-первых, он как насос будет постоянно обращаться к Kafka и втягивать все новые сообщения, которые будут накапливаться в Kafka. Во-вторых, получаемые сообщения он будет преобразовывать правилами, которые мы в него позже заложим. И в-третьих, он будет раскладывать преобразованные сообщения по разным индексам в OpenSearch. Приступим к установке.
1. Переходим на официальный сайт OpenSearch (https://opensearch.org/downloads.html) и скачиваем архив tar.gz нужной версии. В этой статье я буду использовать Logstash-oss-with-OpenSearch-output-plugin версии 7.16.2 (https://artifacts.opensearch.org/logstash/logstash-oss-with-opensearch-output-plugin-7.16.2-linux-x64.tar.gz). После скачивания перенесите архив на сервер «server-elk01» в удобный для вас каталог и перейдите в него в консоли.
2. Даем права на выполнение для архива:
chmod +x logstash-oss-with-opensearch-output-plugin-7.16.2-linux-x64.tar.gz3. Распаковываем архив:
tar -xf logstash-oss-with-opensearch-output-plugin-7.16.2-linux-x64.tar.gz4. Будем устанавливать Logstash-oss-with-OpenSearch-output-plugin в каталог «/opt/logstash», поэтому создаем рабочий каталог для Logstash-oss-with-OpenSearch-output-plugin:
mkdir /opt/logstash5. Переносим распакованные данные в рабочий каталог:
mv ./logstash-7.16.2/* /opt/logstash/6. Удаляем каталог, оставшийся от распаковки:
rmdir ./logstash-7.16.27. Делаем пользователя logstash владельцем рабочего каталога Logstash-oss-with-OpenSearch-output-plugin:
chown -R logstash:logstash /opt/logstash8. Создаем файл демона для работы Logstash-oss-with-OpenSearch-output-plugin:
nano /etc/systemd/system/logstash.service
/etc/systemd/system/logstash.service
[Unit]
Description=logstash
[Service]
Type=simple
User=logstash
Group=logstash
ExecStart=/opt/logstash/bin/logstash "--path.settings" "/opt/logstash/config"
Restart=always
WorkingDirectory=/opt/logstash
Nice=19
LimitNOFILE=16384
# When stopping, how long to wait before giving up and sending SIGKILL?
# Keep in mind that SIGKILL on a process can cause data loss.
TimeoutStopSec=75
[Install]
WantedBy=multi-user.targetОтмечу, что в файле демона оригинального Logstash от Elastic отсутствует параметр «TimeoutStopSec=75». И у меня были случаи, когда при остановке демон Logstash зависал, и приходилось искать java-процесс и убивать его. Параметр «TimeoutStopSec=75» принудительно завершит выполнение демона, если через 75 секунд после команды «stop» процесс не завершится.
9. Зададим права на файл демона:
chmod 644 /etc/systemd/system/logstash.service10. Делаем пользователя root владельцем файла демона:
chown -R root:root /etc/systemd/system/logstash.service11. Перечитаем конфигурацию systemd, чтобы система обнаружила вновь созданный демон:
systemctl daemon-reload12. Зададим путь к базе данных и логам Logstash-oss-with-OpenSearch-output-plugin:
nano /opt/logstash/config/logstash.yml
/opt/logstash/config/logstash.yml
path.data: /opt/logstash/data
pipeline.ordered: auto
path.logs: /var/log/logstash13. Зададим путь для чтения файлов pipelines:
nano /opt/logstash/config/pipelines.yml
/opt/logstash/config/pipelines.yml
- pipeline.id: man
path.config: "/opt/logstash/config/conf.d/*.conf"14. Создадим каталог для файлов pipelines:
mkdir /opt/logstash/config/conf.d15. Создадим каталог для логов:
mkdir /var/log/logstash16. Делаем пользователя logstash владельцем каталога логов:
chown -R logstash /var/log/logstash17. На этом этапе следовало бы сконфигурировать файлы pipelines для обработки данных поступающих из kafka и уходящих в OpenSearch, но я опишу это позже в нескольких кейсах и предоставлю итоговые файлы конфигураций.
18. После окончания настройки еще раз делаем пользователя logstash владельцем всех файлов в рабочем каталоге Logstash-oss-with-OpenSearch-output-plugin:
chown -R logstash:logstash /opt/logstash19. Запустим настроенный нами демон Logstash-oss-with-OpenSearch-output-plugin:
systemctl start logstash20. Проверим статус запуска демона Logstash-oss-with-OpenSearch-output-plugin:
systemctl status logstash21. Настроим автозапуск демона Logstash-oss-with-OpenSearch-output-plugin:
systemctl enable logstash3. Установка Beat’ов
Beat’ы это своего рода агенты, которые устанавливаются на каждом сервер. Они следят за определенными объектами, в нашем случае за файлами с логами, и когда заметят, что в файле произошли изменения они (Beat’ы) сформируют сообщение, которое будет содержать эти изменения, и отправят их (сообщения), в нашем случае, в Kafka.
Сразу стоит отметить, что Elastic выпускает две версии beat’ов: (условно) «платные» (Filebeat, Winlogbeat и т.д.) и (условно) «бесплатные» (Filebeat-OSS-only, Winlogbeat-OSS-only и т.д.). На сколько я понял «платные» работают так же, как и «бесплатные», за исключением того, что в «платных» можно подключить платный функционал «X-Pack». То есть «платные» обладают расширенными возможностями, которые можно и не покупать, и при этом не обладают временным ограничением действия. Поэтому далее я буду использовать Filebeat и Winlogbeat, но без платного функционала «X-Pack».
3.1. Установка Filebeat на Linux
1. Переходим на официальный сайт Elastic в раздел Filebeat (https://www.elastic.co/downloads/beats/filebeat) и скачиваем архив deb нужной версии. В этой статье я буду использовать Filebeat версии 7.12.1 (https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-7.12.1-amd64.deb).
Я буду настраивать Filebeat на той же машине, где развернут OpenSearch, поэтому после скачивания перенесите архив на сервер «server-elk01» в удобный для вас каталог и перейдите в него в консоли.
2. Даем права на выполнение для архива:
chmod +x ./filebeat-7.12.1-amd64.deb3. Установим Filebeat:
dpkg -i ./filebeat-7.12.1-amd64.deb4. На этом этапе следовало бы сконфигурировать файл «/etc/filebeat/filebeat.yml», но я опишу это позже в нескольких кейсах и предоставлю итоговый файл конфигурации.
5. Подключаем модуль system в filebeat:
filebeat modules enable system6. Подключаем модуль kafka в filebeat:
filebeat modules enable kafka7. Проверим список подключенных модулей:
filebeat modules list8. Запустим демон Filebeat:
systemctl start filebeat9. Проверим статус запуска демона Filebeat:
systemctl status filebeat10. Настроим автозапуск демона Filebeat:
systemctl enable filebeat3.2. Установка Filebeat на Windows
1. Переходим на официальный сайт Elastic в раздел Filebeat (https://www.elastic.co/downloads/beats/filebeat) и скачиваем архив zip нужной версии. В этой статье я буду использовать Filebeat версии 7.12.1 (https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-7.12.1-windows-x86_64.zip).
2. Будем устанавливать Filebeat в каталог «C:\filebeat», поэтому после скачивания перенесите архив на сервер «server-windows01» и распакуйте его в каталог «C:\filebeat».
3. На этом этапе следовало бы сконфигурировать файл «C:\filebeat\filebeat.yml», но я опишу это позже в нескольких кейсах и предоставлю итоговый файл конфигурации.
4. Для создания службы нужно выполнить следующий скрипт с правами администратора:
powershell "C:\filebeat\install-service-filebeat.ps1"5. Для запуска Filebeat без службы можно использовать команды:
cd C:\filebeat\
filebeat -e -c filebeat.yml3.3. Установка Winlogbeat на Windows
1. Переходим на официальный сайт Elastic в раздел Winlogbeat (https://www.elastic.co/downloads/beats/winlogbeat) и скачиваем архив zip нужной версии. В этой статье я буду использовать Winlogbeat версии 7.12.1 (https://artifacts.elastic.co/downloads/beats/winlogbeat/winlogbeat-7.12.1-windows-x86_64.zip).
2. Будем устанавливать Winlogbeat в каталог «C:\winlogbeat», поэтому после скачивания перенесите архив на сервер «server-windows01» и распакуйте его в каталог «C:\winlogbeat».
3. На этом этапе следовало бы сконфигурировать файл «C:\winlogbeat\winlogbeat.yml», но я опишу это позже в нескольких кейсах и предоставлю итоговый файл конфигурации.
4. Для создания службы нужно выполнить следующий скрипт с правами администратора:
powershell "C:\winlogbeat\install-service-winlogbeat.ps1"5. Для запуска Winlogbeat без службы можно использовать команды:
cd C:\winlogbeat\
winlogbeat -e -c winlogbeat.yml4. Настройка на примере нескольких кейсов
Рассмотрим несколько ситуаций, в разрезе разных форматов собираемых данных и в разрезе разных Beat’ов, которые будут собирать данные. Посмотрим, как выглядят данные, которые нам предстоит собрать. Какой Beat их будет собирать. Как следует настроить Beat в каждой ситуации. И как следует настроить Logstash в каждой ситуации.
4.1. Сценарий сбора данных в формате json
4.1.1. Исходные данные
Файл (файлы) логов содержит сообщения в виде списка json объектов. Формат самого файла текстовый (не json), то есть квадратные скобки в начале и в конце файла не нужны.
Файл дописывается новыми сообщениями по мере необходимости.
Пример:
«C:\Temp\test_log_json2\2022-03-15.log»
{
"logId": "ID_1",
"uniqueSubId": "001",
"mess": "test_text"
}
{
"logId": "ID_2", "uniqueSubId": "002", "mess": "test_text2"
}
{
"logId": "ID_3",
"uniqueSubId": "003",
"mess": "test_text3"
}Зачем нужно добавлять пустые строки после каждого json-объекта я опишу далее в разделе «Актуальные проблемы».
4.1.2. Процесс сбора данных через Filebeat
Служба Filebeat будет периодически проверять файл на наличие изменений. При обнаружении изменений эти изменения будут отправлены в Topic Kafka. На выходе в начало сообщения добавим метку времени отправки данных с сервера. Во входящее сообщение добавим тэг «test-json», а на выходе будем отправлять в Topic Kafka только те сообщения, которые содержат этот тэг. Так мы промаркируем сообщения. Маркировка сообщений становится актуальна, когда входящих и исходящих потоков несколько. Я добавил маркировку сообщения в этом кейсе для полноты конфигурации.
Отдельного внимания заслуживает параметр «codec.format». Если его не указывать, то, по умолчанию, отправляемое сообщение будет в формате json и будет содержать целый набор полей, которые содержат в основном техническую (справочную) информацию (имя и версию ОС, имя и версию Beat’а и т.д.). Меня же интересует только основное сообщение, которое содержится в поле «message», остальные данные я буду считать объективно известными, за счет того, что все потоки данных разделены на отдельные Topic’и Kafka.
Полный список полей, отправляемых в сообщении с настройками по умолчанию: «event», «log», «message», «@timestamp», «@metadata», «ecs», «agent», «host», «service», «input», «fileset».
Пример конфигурации Filebeat:
«C:\filebeat\filebeat.yml»
filebeat.inputs:
- type: log
enabled: true
paths:
- C:\Temp\test_log_json2\*
multiline.pattern: '^{'
multiline.negate: true
multiline.match: after
processors:
- decode_json_fields:
fields: ["message"]
target: "json"
tags: ["test-json"]
filebeat.config.modules:
path: ${path.config}/modules.d/*.yml
reload.enabled: false
setup.template.settings:
index.number_of_shards: 1
setup.kibana:
processors:
- add_host_metadata:
when.not.contains.tags: forwarded
- add_cloud_metadata: ~
- add_docker_metadata: ~
- add_kubernetes_metadata: ~
output.kafka:
hosts: ["10.0.0.60:9092"]
topics:
- topic: "server-windows01-test-json"
when.contains:
tags: "test-json"
codec.format:
string: '%{[@timestamp]} %{[message]}'
partition.round_robin:
reachable_only: false
required_acks: 1
compression: gzip
max_message_bytes: 1000000
close_inactive: 10m4.1.3. Процесс обработки данных в Logstash
Демон Logstash при обнаружении новых данных в Topic'е Kafka начинает их прием и обработку. Данные внутри Logstash поступают сначала в модуль input, потом поступают в модуль filter и затем поступают в модуль output.
В модуле input примем данные из Topic'а Kafka и добавим в сообщение поле «type», которое будет содержать название Topic'а.
Пример конфигурации Logstash модуля input:
«/opt/logstash/config/conf.d/server-windows01-input.conf»
input {
kafka {
type => "server-windows01-test-json"
bootstrap_servers => "10.0.0.60:9092"
topics => "server-windows01-test-json"
}
}В модуле filter обрабатываются только те сообщения, в которых есть поле «type», значение которого равно имени Topic'а Kafka. Из начала сообщения вырежем метку времени и добавим её в новое поле «timestamp_filebeat». Декодируем основное сообщение из формата json, при этом все заголовки ключей станут одноименными полями, а значения ключей перейдут в значения полей. Добавим поле «timestamp_logstash», для фиксирования времени, в которое проводилась обработка сообщения.
Пример конфигурации Logstash модуля filter:
«/opt/logstash/config/conf.d/server-windows01-filter.conf»
filter {
if [type] == "server-windows01-test-json" {
grok {
match => { "message" => "%{TIMESTAMP_ISO8601:timestamp_filebeat} %{GREEDYDATA:message}" }
overwrite => [ "message" ]
}
json {
source => "message"
}
mutate {
add_field => { "timestamp_logstash" => "%{[@timestamp]}" }
}
}
}В модуле output обрабатываются только те сообщения, в которых есть поле «type», значение которого равно имени Topic'а Kafka. Каждое сообщение отправляется в OpenSearch в индекс, соответствующий текущей дате, имени Topic’а и префикса «Kafka».
Пример конфигурации Logstash модуля output:
«/opt/logstash/config/conf.d/server-windows01-output.conf»
output {
if [type] == "server-windows01-test-json" {
opensearch {
hosts => "https://localhost:9200"
user => "admin"
password => "yN-3L(GMmAAw"
index => "kafka-server-windows01-test-json-%{+YYYY.MM.dd}"
ssl_certificate_verification => false
}
}
}4.2. Сценарий сбора однострочных данных в текстовом формате
4.2.1. Исходные данные
Каждая строка файла будет извлекаться как новое сообщение и всё содержимое строки будет помещено в поле «message» как текст. В Filebeat такой сценарий работает по умолчанию. Нам остается только промаркировать сообщения и отправить их в нужный Topic Kafka.
Если содержание всех строк стандартизировано, то в Logstash на этапе Filter можно разложить сообщение на заранее известные поля. В этом примере мы этого делать не будем.
Пример:
«C:\Temp\log\2022-03-16.log»
[2022.03.16 13:01:09] Info | Just text
Next text
[2022.03.16 13:02:01] Info | New text4.2.2. Процесс сбора данных через Filebeat
Процесс сбора данных аналогичен ситуации сбора данных в формате json за исключением того, что нам не нужно парсить json, так же изменим тэг для маркировки входящих сообщений, и изменим Topic Kafka.
Пример конфигурации Filebeat:
«C:\filebeat\filebeat.yml»
filebeat.inputs:
- type: filestream
paths:
- C:\Temp\log\*.log
tags: ["simple-logs"]
filebeat.config.modules:
path: ${path.config}/modules.d/*.yml
reload.enabled: false
setup.template.settings:
index.number_of_shards: 1
setup.kibana:
processors:
- add_host_metadata:
when.not.contains.tags: forwarded
- add_cloud_metadata: ~
- add_docker_metadata: ~
- add_kubernetes_metadata: ~
output.kafka:
hosts: ["10.0.0.60:9092"]
topics:
- topic: "server-windows01-simple-logs"
when.contains:
tags: "simple-logs"
codec.format:
string: '%{[@timestamp]} %{[message]}'
partition.round_robin:
reachable_only: false
required_acks: 1
compression: gzip
max_message_bytes: 1000000
close_inactive: 10m4.2.3. Процесс обработки данных в Logstash
Процесс обработки данных в Logstash будет незначительно отличаться от сценария для ситуации с json-объектами, поэтому я приведу только примеры конфигураций.
Пример конфигурации Logstash модуля input:
«/opt/logstash/config/conf.d/server-windows01-input.conf»
input {
kafka {
type => "server-windows01-simple-logs"
bootstrap_servers => "10.0.0.60:9092"
topics => "server-windows01-simple-logs"
}
}Пример конфигурации Logstash модуля filter:
«/opt/logstash/config/conf.d/server-windows01-filter.conf»
filter {
if [type] == "server-windows01-simple-logs" {
grok {
match => { "message" => "%{TIMESTAMP_ISO8601:timestamp_filebeat} %{GREEDYDATA:message}" }
overwrite => [ "message" ]
}
mutate {
add_field => { "timestamp_logstash" => "%{[@timestamp]}" }
}
}
}Пример конфигурации Logstash модуля output:
«/opt/logstash/config/conf.d/server-windows01-output.conf»
output {
if [type] == "server-windows01-simple-logs" {
opensearch {
hosts => "https://localhost:9200"
user => "admin"
password => "yN-3L(GMmAAw"
index => "kafka-server-windows01-simple-logs-%{+YYYY.MM.dd}"
ssl_certificate_verification => false
}
}
}4.3. Сценарий сбора Логов IIS
Для сбора логов IIS в Filebeat предусмотрен отдельный модуль, который так и называется IIS. В IIS логи делятся на две категории: «ACCESS» и «ERROR». У меня пути к этим логам настроены по умолчанию.
4.3.1. Процесс сбора данных через Filebeat
Для того, чтобы промаркировать эти две категории («ACCESS» и «ERROR») и направить их в разные Topic’и Kafka я использовал параметр «@metadata.pipeline».
Так же стоит обратить внимание на параметры «var.paths», по какой-то причине в путях необходимо использовать обратные слэши.
Пример конфигурации Filebeat:
«C:\filebeat\filebeat.yml»
filebeat.inputs:
filebeat.modules:
- module: iis
access:
enabled: true
var.paths: ["C:/inetpub/logs/LogFiles/*/*.log"]
error:
enabled: true
var.paths: ["C:/Windows/System32/LogFiles/HTTPERR/*.log"]
filebeat.config.modules:
path: ${path.config}/modules.d/*.yml
reload.enabled: false
setup.template.settings:
index.number_of_shards: 1
setup.kibana:
processors:
- add_host_metadata:
when.not.contains.tags: forwarded
- add_cloud_metadata: ~
- add_docker_metadata: ~
- add_kubernetes_metadata: ~
output.kafka:
hosts: ["10.0.0.60:9092"]
topics:
- topic: "rgmtpaydox39-iis-access"
when.contains:
"@metadata.pipeline": "iis-access"
- topic: "rgmtpaydox39-iis-error"
when.contains:
"@metadata.pipeline": "iis-error"
codec.format:
string: '%{[@timestamp]} %{[message]}'
partition.round_robin:
reachable_only: false
required_acks: 1
compression: gzip
max_message_bytes: 1000000
close_inactive: 10m4.3.2. Процесс обработки данных в Logstash
Пример конфигурации Logstash модуля input:
«/opt/logstash/config/conf.d/server-windows01-input.conf»
input {
kafka {
type => "server-windows01-iis-access"
bootstrap_servers => "10.0.0.60:9092"
topics => "server-windows01-iis-access"
}
kafka {
type => "server-windows01-iis-error"
bootstrap_servers => "10.0.0.60:9092"
topics => "server-windows01-iis-error"
}
}Во время обработки «iis-access» из начала сообщения вырежем метку времени и добавим её в новое поле «timestamp_filebeat». Затем из полученного сообщения снова вырежем метку времени из начала сообщения и добавим её в новое поле «timestamp_windows». Затем преобразуем тип поля «timestamp_windows» из строкового в тип «дата». Это позволит в дальнейшем в OpenSearch-Dashboards создавать шаблоны индексов с фильтрацией по этому полю. Затем добавим еще одно поле «timestamp_logstash».
За счет наличия трех меток времени мы сможем точно отследить, когда событие возникло в windows, когда попало в обработку Filebeat и когда попало в обработку Logstash.
Обработка «iis-error» осуществляется аналогичным образом.
Пример конфигурации Logstash модуля filter:
«/opt/logstash/config/conf.d/server-windows01-filter.conf»
filter {
if [type] == "server-windows01-iis-access" {
grok {
match => { "message" => "%{TIMESTAMP_ISO8601:timestamp_filebeat} %{GREEDYDATA:message}" }
overwrite => [ "message" ]
}
grok {
match => [ "message", "%{TIMESTAMP_ISO8601:timestamp_windows} %{GREEDYDATA:message}" ]
}
date {
match => ["timestamp_windows", "ISO8601", "YYYY-MM-dd HH:mm:ss", "MMM dd, YYYY @ HH:mm:ss.ZZZ"]
target => "timestamp_windows"
locale => "en"
}
mutate {
add_field => { "timestamp_logstash" => "%{[@timestamp]}" }
}
}
if [type] == "server-windows01-iis-error" {
grok {
match => { "message" => "%{TIMESTAMP_ISO8601:timestamp_filebeat} %{GREEDYDATA:message}" }
overwrite => [ "message" ]
}
grok {
match => [ "message", "%{TIMESTAMP_ISO8601:timestamp_windows} %{GREEDYDATA:message}" ]
}
date {
match => ["timestamp_windows", "ISO8601", "YYYY-MM-dd HH:mm:ss", "MMM dd, YYYY @ HH:mm:ss.ZZZ"]
target => "timestamp_windows"
locale => "en"
}
mutate {
add_field => { "timestamp_logstash" => "%{[@timestamp]}" }
}
}
}Пример конфигурации Logstash модуля output:
«/opt/logstash/config/conf.d/server-windows01-output.conf»
output {
if [type] == "server-windows01-iis-access" {
opensearch {
hosts => "https://localhost:9200"
user => "admin"
password => "yN-3L(GMmAAw"
index => "kafka-server-windows01-iis-access-%{+YYYY.MM.dd}"
ssl_certificate_verification => false
}
}
if [type] == "server-windows01-iis-error" {
opensearch {
hosts => "https://localhost:9200"
user => "admin"
password => "yN-3L(GMmAAw"
index => "kafka-server-windows01-iis-error-%{+YYYY.MM.dd}"
ssl_certificate_verification => false
}
}
}4.4. Сценарий сбора системных Логов Windows
4.4.1. Процесс сбора данных через Winlogbeat
Системные логи Windows будем собирать с помощью Winlogbeat. Доступны 3 вида логов: «Application», «Security» и «System».
Winlogbeat имеет особенность в отличие от Filebeat. Подставляемый на выходе параметр «timestamp» будет содержать не время отправки с сервера, а время события (из лога), произошедшего в Windows. Временная метка отправки с сервера будет содержаться в поле «event» в параметре «created» («event.created»). Кроме того, я посчитал еще несколько полей информативными. В итоге на выходе отправляются следующие поля: «@timestamp», «message», «event», «log», «winlog». При этом «@timestamp» расположен в начале сообщения и отделен пробелом. Все остальные поля я разделил символом «|».
Полный список полей, отправляемых в сообщении с настройками по умолчанию: «event», «log», «message», «winlog», «@timestamp», «@metadata», «ecs», «agent», «host».
Так же как и в Filebeat вы можете убрать параметр «codec.format» и посмотреть полный список полей и их значений, которые попадут в Logstash в формате json.
К каждому виду логов можно добавить параметр «ignore_older», чтобы игнорировать слишком старые логи. Я добавил этот параметр к «Application» со значением «72h», чтобы игнорировать логи старше 72 часов.
Пример конфигурации Winlogbeat:
«C:\winlogbeat\winlogbeat.yml»
winlogbeat.event_logs:
- name: Application
ignore_older: 72h
tags: ["server-windows01-application"]
- name: Security
tags: ["server-windows01-security"]
- name: System
tags: ["server-windows01-system"]
output.kafka:
hosts: ["10.0.0.60:9092"]
topics:
- topic: "server-windows01-application"
when.contains:
tags: "server-windows01-application"
- topic: "server-windows01-security"
when.contains:
tags: "server-windows01-security"
- topic: "server-windows01-system"
when.contains:
tags: "server-windows01-system"
codec.format:
string: '%{[@timestamp]} %{[message]}|%{[event]}|%{[log]}|%{[winlog]}'
partition.round_robin:
reachable_only: false
required_acks: 1
compression: gzip
max_message_bytes: 1000000
logging.level: info
logging.to_files: true
logging.files:
path: C:\winlogbeat
name: winlogbeat.log
keepfiles: 74.4.2. Процесс обработки данных в Logstash
Пример конфигурации Logstash модуля input:
«/opt/logstash/config/conf.d/server-windows01-input.conf»
input {
kafka {
type => "server-windows01-application"
bootstrap_servers => "10.0.0.60:9092"
topics => "server-windows01-application"
}
kafka {
type => "server-windows01-security"
bootstrap_servers => "10.0.0.60:9092"
topics => "server-windows01-security"
}
kafka {
type => "server-windows01-system"
bootstrap_servers => "10.0.0.60:9092"
topics => "server-windows01-system"
}
}Во время обработки «application» из начала сообщения вырежем метку времени и добавим её в новое поле «timestamp_windows». Установим, что остальные поля разделены символом «|», в соответствии с конфигурацией Winlogbeat. Символ «|» необходимо экранировать символом «\». Названия полей обозначим с таким же именем и в таком же порядке, как и в конфигурации Winlogbeat. Зная, что поле «event» находится в формате json распарсим его как json и поместим в то же поле. С полями «log» и «winlog» поступим аналогично. Так как время обработки сообщения в Winlogbeat находится в поле «event» в параметре «created», то создадим поле «timestamp_winlogbeat» и поместим туда значение из «event.created». После этого создадим поле «timestamp_logstash» с меткой времени обработки сообщения в Logstash.
Сообщения «security» и «system» обрабатываются аналогичным образом.
Пример конфигурации Logstash модуля filter:
«/opt/logstash/config/conf.d/server-windows01-filter.conf»
filter {
if [type] == "server-windows01-application" {
grok {
match => { "message" => "%{TIMESTAMP_ISO8601:timestamp_windows} %{DATA:message}\|%{DATA:event}\|%{DATA:log}\|%{GREEDYDATA:winlog}" }
overwrite => [ "message" ]
}
json {
source => "event"
target => "event"
}
json {
source => "log"
target => "log"
}
json {
source => "winlog"
target => "winlog"
}
mutate {
add_field => { "timestamp_winlogbeat" => "%{[event][created]}" }
add_field => { "timestamp_logstash" => "%{[@timestamp]}" }
}
}
if [type] == "server-windows01-security" {
grok {
match => { "message" => "%{TIMESTAMP_ISO8601:timestamp_windows} %{DATA:message}\|%{DATA:event}\|%{DATA:log}\|%{GREEDYDATA:winlog}" }
overwrite => [ "message" ]
}
json {
source => "event"
target => "event"
}
json {
source => "log"
target => "log"
}
json {
source => "winlog"
target => "winlog"
}
mutate {
add_field => { "timestamp_winlogbeat" => "%{[event][created]}" }
add_field => { "timestamp_logstash" => "%{[@timestamp]}" }
}
}
if [type] == "server-windows01-system" {
grok {
match => { "message" => "%{TIMESTAMP_ISO8601:timestamp_windows} %{DATA:message}\|%{DATA:event}\|%{DATA:log}\|%{GREEDYDATA:winlog}" }
overwrite => [ "message" ]
}
json {
source => "event"
target => "event"
}
json {
source => "log"
target => "log"
}
json {
source => "winlog"
target => "winlog"
}
mutate {
add_field => { "timestamp_winlogbeat" => "%{[event][created]}" }
add_field => { "timestamp_logstash" => "%{[@timestamp]}" }
}
}
}Пример конфигурации Logstash модуля output:
«/opt/logstash/config/conf.d/server-windows01-output.conf»
output {
if [type] == "server-windows01-application" {
opensearch {
hosts => "https://localhost:9200"
user => "admin"
password => "yN-3L(GMmAAw"
index => "kafka-server-windows01-application-%{+YYYY.MM.dd}"
ssl_certificate_verification => false
}
}
if [type] == "server-windows01-security" {
opensearch {
hosts => "https://localhost:9200"
user => "admin"
password => "yN-3L(GMmAAw"
index => "kafka-server-windows01-security-%{+YYYY.MM.dd}"
ssl_certificate_verification => false
}
}
if [type] == "server-windows01-system" {
opensearch {
hosts => "https://localhost:9200"
user => "admin"
password => "yN-3L(GMmAAw"
index => "kafka-server-windows01-system-%{+YYYY.MM.dd}"
ssl_certificate_verification => false
}
}
}4.5. Сценарий сбора системных Логов Linux
4.5.1. Процесс сбора данных через Filebeat
Системные логи Linux будем собирать с помощью Filebeat. Конфигурирование Filebeat в Linux практически идентично конфигурированию Filebeat в Windows. За счет того, что при установке Filebeat на Linux мы включили модуль system, можно настроить только вывод данных, сменив его с elasticsearch на kafka, остальные настройки можно оставить по умолчанию и данные пойдут в формате json.
Поскольку системные логи Linux хранятся в одном файле «/var/log/syslog» (архивные не берем в расчет) и формат логов однострочный, то есть каждое сообщение хранится в отдельной строке, то в этой ситуации можно так же применить «Сценарий сбора однострочных данных в текстовом формате». Именно этот вариант я и представлю.
Пример конфигурации Filebeat:
«/etc/filebeat/filebeat.yml»
filebeat.inputs:
- type: log
enabled: true
paths:
- /var/log/syslog
filebeat.config.modules:
path: ${path.config}/modules.d/*.yml
reload.enabled: false
setup.template.settings:
index.number_of_shards: 1
setup.kibana:
output.kafka:
hosts: ["10.0.0.60:9092"]
topic: "server-elk01-syslog"
codec.format:
string: '%{[@timestamp]} %{[message]}'
partition.round_robin:
reachable_only: false
required_acks: 1
compression: gzip
max_message_bytes: 1000000
close_inactive: 10m
processors:
- add_host_metadata:
when.not.contains.tags: forwarded
- add_cloud_metadata: ~
- add_docker_metadata: ~
- add_kubernetes_metadata: ~
logging.level: info
logging.to_files: true
logging.files:
path: /var/log/filebeat
name: filebeat.log
keepfiles: 74.5.2. Процесс обработки данных в Logstash
Пример конфигурации Logstash модуля input:
«/opt/logstash/config/conf.d/server-elk01-input.conf»
input {
kafka {
type => "server-elk01-syslog"
bootstrap_servers => "10.0.0.60:9092"
topics => "server-elk01-syslog"
}
}Пример конфигурации Logstash модуля filter:
«/opt/logstash/config/conf.d/server-elk01-filter.conf»
filter {
if [type] == "server-elk01-syslog" {
grok {
match => { "message" => "%{TIMESTAMP_ISO8601:timestamp_filebeat} %{GREEDYDATA:message}" }
overwrite => [ "message" ]
}
mutate {
add_field => { "timestamp_logstash" => "%{[@timestamp]}" }
}
}
}Пример конфигурации Logstash модуля output:
«/opt/logstash/config/conf.d/server-elk01-output.conf»
output {
if [type] == "server-elk01-syslog" {
opensearch {
hosts => "https://localhost:9200"
user => "admin"
password => "yN-3L(GMmAAw"
index => "kafka-server-elk01-syslog-%{+YYYY.MM.dd}"
ssl_certificate_verification => false
}
}
}5. Итоговые файлы конфигураций
С учетом всех сценариев у меня получились следующие файлы конфигураций.
Пример конфигурации Filebeat на Linux:
«/etc/filebeat/filebeat.yml»
filebeat.inputs:
- type: log
enabled: true
paths:
- /var/log/syslog
filebeat.config.modules:
path: ${path.config}/modules.d/*.yml
reload.enabled: false
setup.template.settings:
index.number_of_shards: 1
setup.kibana:
output.kafka:
hosts: ["10.0.0.60:9092"]
topic: "server-elk01-syslog"
codec.format:
string: '%{[@timestamp]} %{[message]}'
partition.round_robin:
reachable_only: false
required_acks: 1
compression: gzip
max_message_bytes: 1000000
close_inactive: 10m
processors:
- add_host_metadata:
when.not.contains.tags: forwarded
- add_cloud_metadata: ~
- add_docker_metadata: ~
- add_kubernetes_metadata: ~
logging.level: info
logging.to_files: true
logging.files:
path: /var/log/filebeat
name: filebeat.log
keepfiles: 7Пример конфигурации Filebeat на Windows:
«C:\filebeat\filebeat.yml»
filebeat.inputs:
- type: filestream
paths:
- C:\Temp\log\*.log
tags: ["simple-logs"]
- type: log
enabled: true
paths:
- C:\Temp\test_log_json2\*
multiline.pattern: '^{'
multiline.negate: true
multiline.match: after
processors:
- decode_json_fields:
fields: ["message"]
target: "json"
tags: ["test-json"]
filebeat.modules:
- module: iis
access:
enabled: true
var.paths: ["C:/inetpub/logs/LogFiles/*/*.log"]
error:
enabled: true
var.paths: ["C:/Windows/System32/LogFiles/HTTPERR/*.log"]
filebeat.config.modules:
path: ${path.config}/modules.d/*.yml
reload.enabled: false
setup.template.settings:
index.number_of_shards: 1
setup.kibana:
processors:
- add_host_metadata:
when.not.contains.tags: forwarded
- add_cloud_metadata: ~
- add_docker_metadata: ~
- add_kubernetes_metadata: ~
output.kafka:
hosts: ["10.0.0.60:9092"]
topics:
- topic: "server-windows01-simple-logs"
when.contains:
tags: "simple-logs"
- topic: "server-windows01-test-json"
when.contains:
tags: "test-json"
- topic: "server-windows01-iis-access"
when.contains:
"@metadata.pipeline": "iis-access"
- topic: "server-windows01-iis-error"
when.contains:
"@metadata.pipeline": "iis-error"
codec.format:
string: '%{[@timestamp]} %{[message]}'
partition.round_robin:
reachable_only: false
required_acks: 1
compression: gzip
max_message_bytes: 1000000
close_inactive: 10mПример конфигурации Winlogbeat:
«C:\winlogbeat\winlogbeat.yml»
winlogbeat.event_logs:
- name: Application
ignore_older: 72h
tags: ["server-windows01-application"]
- name: Security
tags: ["server-windows01-security"]
- name: System
tags: ["server-windows01-system"]
output.kafka:
hosts: ["10.0.0.60:9092"]
topics:
- topic: "server-windows01-application"
when.contains:
tags: "server-windows01-application"
- topic: "server-windows01-security"
when.contains:
tags: "server-windows01-security"
- topic: "server-windows01-system"
when.contains:
tags: "server-windows01-system"
codec.format:
string: '%{[@timestamp]} %{[message]}|%{[event]}|%{[log]}|%{[winlog]}'
partition.round_robin:
reachable_only: false
required_acks: 1
compression: gzip
max_message_bytes: 1000000
logging.level: info
logging.to_files: true
logging.files:
path: C:\winlogbeat
name: winlogbeat.log
keepfiles: 7Пример конфигурации Logstash модуля input для сервера «server-elk01»:
«/opt/logstash/config/conf.d/server-elk01-input.conf»
input {
kafka {
type => "server-elk01-syslog"
bootstrap_servers => "10.0.0.60:9092"
topics => "server-elk01-syslog"
}
}Пример конфигурации Logstash модуля filter для сервера «server-elk01»:
«/opt/logstash/config/conf.d/server-elk01-filter.conf»
filter {
if [type] == "server-elk01-syslog" {
grok {
match => { "message" => "%{TIMESTAMP_ISO8601:timestamp_filebeat} %{GREEDYDATA:message}" }
overwrite => [ "message" ]
}
mutate {
add_field => { "timestamp_logstash" => "%{[@timestamp]}" }
}
}
}Пример конфигурации Logstash модуля output для сервера «server-elk01»:
«/opt/logstash/config/conf.d/server-elk01-output.conf»
output {
if [type] == "server-elk01-syslog" {
opensearch {
hosts => "https://localhost:9200"
user => "admin"
password => "yN-3L(GMmAAw"
index => "kafka-server-elk01-syslog-%{+YYYY.MM.dd}"
ssl_certificate_verification => false
}
}
}Пример конфигурации Logstash модуля input для сервера «server-windows01»:
«/opt/logstash/config/conf.d/server-windows01-input.conf»
input {
kafka {
type => "server-windows01-simple-logs"
bootstrap_servers => "10.0.0.60:9092"
topics => "server-windows01-simple-logs"
}
kafka {
type => "server-windows01-test-json"
bootstrap_servers => "10.0.0.60:9092"
topics => "server-windows01-test-json"
}
kafka {
type => "server-windows01-application"
bootstrap_servers => "10.0.0.60:9092"
topics => "server-windows01-application"
}
kafka {
type => "server-windows01-security"
bootstrap_servers => "10.0.0.60:9092"
topics => "server-windows01-security"
}
kafka {
type => "server-windows01-system"
bootstrap_servers => "10.0.0.60:9092"
topics => "server-windows01-system"
}
kafka {
type => "server-windows01-iis-access"
bootstrap_servers => "10.0.0.60:9092"
topics => "server-windows01-iis-access"
}
kafka {
type => "server-windows01-iis-error"
bootstrap_servers => "10.0.0.60:9092"
topics => "server-windows01-iis-error"
}
}Пример конфигурации Logstash модуля filter для сервера «server-windows01»:
«/opt/logstash/config/conf.d/server-windows01-filter.conf»
filter {
if [type] == "server-windows01-simple-logs" {
grok {
match => { "message" => "%{TIMESTAMP_ISO8601:timestamp_filebeat} %{GREEDYDATA:message}" }
overwrite => [ "message" ]
}
mutate {
add_field => { "timestamp_logstash" => "%{[@timestamp]}" }
}
}
if [type] == "server-windows01-test-json" {
grok {
match => { "message" => "%{TIMESTAMP_ISO8601:timestamp_filebeat} %{GREEDYDATA:message}" }
overwrite => [ "message" ]
}
json {
source => "message"
}
mutate {
add_field => { "timestamp_logstash" => "%{[@timestamp]}" }
}
}
if [type] == "server-windows01-application" {
grok {
match => { "message" => "%{TIMESTAMP_ISO8601:timestamp_windows} %{DATA:message}\|%{DATA:event}\|%{DATA:log}\|%{GREEDYDATA:winlog}" }
overwrite => [ "message" ]
}
json {
source => "event"
target => "event"
}
json {
source => "log"
target => "log"
}
json {
source => "winlog"
target => "winlog"
}
mutate {
add_field => { "timestamp_winlogbeat" => "%{[event][created]}" }
add_field => { "timestamp_logstash" => "%{[@timestamp]}" }
}
}
if [type] == "server-windows01-security" {
grok {
match => { "message" => "%{TIMESTAMP_ISO8601:timestamp_windows} %{DATA:message}\|%{DATA:event}\|%{DATA:log}\|%{GREEDYDATA:winlog}" }
overwrite => [ "message" ]
}
json {
source => "event"
target => "event"
}
json {
source => "log"
target => "log"
}
json {
source => "winlog"
target => "winlog"
}
mutate {
add_field => { "timestamp_winlogbeat" => "%{[event][created]}" }
add_field => { "timestamp_logstash" => "%{[@timestamp]}" }
}
}
if [type] == "server-windows01-system" {
grok {
match => { "message" => "%{TIMESTAMP_ISO8601:timestamp_windows} %{DATA:message}\|%{DATA:event}\|%{DATA:log}\|%{GREEDYDATA:winlog}" }
overwrite => [ "message" ]
}
json {
source => "event"
target => "event"
}
json {
source => "log"
target => "log"
}
json {
source => "winlog"
target => "winlog"
}
mutate {
add_field => { "timestamp_winlogbeat" => "%{[event][created]}" }
add_field => { "timestamp_logstash" => "%{[@timestamp]}" }
}
}
if [type] == "server-windows01-iis-access" {
grok {
match => { "message" => "%{TIMESTAMP_ISO8601:timestamp_filebeat} %{GREEDYDATA:message}" }
overwrite => [ "message" ]
}
grok {
match => [ "message", "%{TIMESTAMP_ISO8601:timestamp_windows} %{GREEDYDATA:message}" ]
}
date {
match => ["timestamp_windows", "ISO8601", "YYYY-MM-dd HH:mm:ss", "MMM dd, YYYY @ HH:mm:ss.ZZZ"]
target => "timestamp_windows"
locale => "en"
}
mutate {
add_field => { "timestamp_logstash" => "%{[@timestamp]}" }
}
}
if [type] == "server-windows01-iis-error" {
grok {
match => { "message" => "%{TIMESTAMP_ISO8601:timestamp_filebeat} %{GREEDYDATA:message}" }
overwrite => [ "message" ]
}
grok {
match => [ "message", "%{TIMESTAMP_ISO8601:timestamp_windows} %{GREEDYDATA:message}" ]
}
date {
match => ["timestamp_windows", "ISO8601", "YYYY-MM-dd HH:mm:ss", "MMM dd, YYYY @ HH:mm:ss.ZZZ"]
target => "timestamp_windows"
locale => "en"
}
mutate {
add_field => { "timestamp_logstash" => "%{[@timestamp]}" }
}
}
}Пример конфигурации Logstash модуля output для сервера «server-windows01»:
«/opt/logstash/config/conf.d/server-windows01-output.conf»
output {
if [type] == "server-windows01-simple-logs" {
opensearch {
hosts => "https://localhost:9200"
user => "admin"
password => "yN-3L(GMmAAw"
index => "kafka-server-windows01-simple-logs-%{+YYYY.MM.dd}"
ssl_certificate_verification => false
}
}
if [type] == "server-windows01-test-json" {
opensearch {
hosts => "https://localhost:9200"
user => "admin"
password => "yN-3L(GMmAAw"
index => "kafka-server-windows01-test-json-%{+YYYY.MM.dd}"
ssl_certificate_verification => false
}
}
if [type] == "server-windows01-application" {
opensearch {
hosts => "https://localhost:9200"
user => "admin"
password => "yN-3L(GMmAAw"
index => "kafka-server-windows01-application-%{+YYYY.MM.dd}"
ssl_certificate_verification => false
}
}
if [type] == "server-windows01-security" {
opensearch {
hosts => "https://localhost:9200"
user => "admin"
password => "yN-3L(GMmAAw"
index => "kafka-server-windows01-security-%{+YYYY.MM.dd}"
ssl_certificate_verification => false
}
}
if [type] == "server-windows01-system" {
opensearch {
hosts => "https://localhost:9200"
user => "admin"
password => "yN-3L(GMmAAw"
index => "kafka-server-windows01-system-%{+YYYY.MM.dd}"
ssl_certificate_verification => false
}
}
if [type] == "server-windows01-iis-access" {
opensearch {
hosts => "https://localhost:9200"
user => "admin"
password => "yN-3L(GMmAAw"
index => "kafka-server-windows01-iis-access-%{+YYYY.MM.dd}"
ssl_certificate_verification => false
}
}
if [type] == "server-windows01-iis-error" {
opensearch {
hosts => "https://localhost:9200"
user => "admin"
password => "yN-3L(GMmAAw"
index => "kafka-server-windows01-iis-error-%{+YYYY.MM.dd}"
ssl_certificate_verification => false
}
}
}6. Эксплуатация
Следующие шаги я разберу на примере сбора данных в формате json.
6.1. Создание шаблона индексов с помощью OpenSearch-Dashboards
Шаблон индексов позволяет просматривать сообщения одного или нескольких индексов сортируя их по дате. В качестве даты может выступать любое поле в формате «дата». Для того чтобы можно было осуществлять поиск с помощью OpenSearch-Dashboards необходимо создать шаблон индексов. Для этого перейдем в раздел «Stack Management», далее в раздел «Index Pattern» и нажмем кнопку «Create index pattern» (Рис.2).
Рис.2
Создадим шаблон индексов для данных из Topic'а Kafka без учета даты. Для этого введем в строке имени шаблона название индекса, но в конце, вместо даты введем символ «*» (Рис.3).
Рис.3
Параметром сортировки данных в шаблоне выберем «@timestamp», он совпадает с «timestamp_logstash» (Рис.4).
Рис.4
6.2. Отображение данных в OpenSearch-Dashboards
При переходе в раздел «Discover» в соответствующий шаблон индексов там отобразятся доступные сообщения. Можно раскрыть подробное содержание сообщения и увидеть, что все заголовки ключей json-сообщения стали одноименными полями, а значения ключей перешли в значения полей. Так же осталось поле «message», которое содержит изначальное json-сообщение (Рис.5). На этапе фильтрации сообщения в Logstash это поле («message») можно удалять, если в нем нет необходимости.
Если в json-сообщении содержится поле «message», то оригинальное поле «message», которое содержит изначальное json-сообщение, будет перезаписано на поле из json-сообщения.
Рис.5
6.3. Поиск в OpenSearch-Dashboards
Я рассмотрю совсем немного вариантов поиска, исключительно с целью обзора. Поиск будет только в формате «DQL» (формат по умолчанию).
Поиск осуществляется в разделе «Discover» в соответствующем шаблоне индексов.
1. Для поиска по всем доступным сообщениям во всех полях достаточно ввести нужную фразу в строке поиска (Рис.6).
Рис.6
2. Для поиска фраз по определенным полям в поисковой строке указывается имя нужного поля, затем, в качестве разделителя, идет символ двоеточия «:» и затем искомая фраза. Искомая фраза может содержать специальные символы «*», «?» и т.д. для формирования шаблона (Рис.7).
Рис.7
3. В поисковом запросе можно применять логические операции «or», «and», «not» (Рис.8).
Рис.8
6.4. Работа с OpenSearch через консоль (без OpenSearch-Dashboards)
Я приведу не большой список запросов для того, чтобы показать, как выглядит работа из консоли.
Запросить список индексов:
curl 'https://10.0.0.70:9200/_cat/indices?v&pretty' -u 'admin:yN-3L(GMmAAw' –insecureЗапросить структуру индекса:
curl 'https://10.0.0.70:9200/kafka-server-windows01-test-json-2022.03.30/_mapping?pretty' -u 'admin:yN-3L(GMmAAw' --insecureУдалить индекс:
curl -XDELETE 'https://10.0.0.70:9200/kafka-server-windows01-test-json-2022.03.30?pretty' -u 'admin:yN-3L(GMmAAw' –insecureВывести первые 10 сообщений из индекса (10 по умолчанию):
curl -XGET 'https://10.0.0.70:9200/kafka-server-windows01-test-json-2022.03.30/_search?pretty' -u 'admin:yN-3L(GMmAAw' --insecureВывести первые 20 сообщений из индекса:
curl -XGET 'https://10.0.0.70:9200/kafka-server-windows01-test-json-2022.03.30/_search?size=20&pretty' -u 'admin:yN-3L(GMmAAw' --insecureВывести первые 20 сообщений из шаблона индексов:
curl -XGET 'https://10.0.0.70:9200/kafka-server-windows01-test-json-*/_search?size=20&pretty' -u 'admin:yN-3L(GMmAAw' --insecureОсуществить поиск в индексе по запросу «*text*» через параметр «q»:
curl -XGET 'https://10.0.0.70:9200/kafka-server-windows01-test-json-2022.03.30/_search?q=*text*&pretty' -u 'admin:yN-3L(GMmAAw' --insecureДля поиска по составным запросам существуют несколько типов поиска:
term — точное совпадение искомой строки со строкой в индексе или термом;
match — все слова должны входить в строку, в любом порядке;
match_phrase — вся фраза должна входить в строку;
query_string — все слова входят в строку в любом порядке, можно искать по нескольким полям, используя регулярные выражения.
Осуществим поиск в индексе по полю «mess», искать будем точное совпадение фразы «test_text»:
curl -XGET 'https://10.0.0.70:9200/kafka-server-windows01-test-json-2022.03.30/_search?pretty ' -u 'admin:yN-3L(GMmAAw' --insecure -H 'Content-Type: application/json' -d '{"query" : {"term" : {"mess" : "test_text"}}}'7. Актуальные проблемы
В этом разделе я опишу проблемы, с которыми я встретился.
7.1. Имена Topic’ов Kafka не должны содержать большие буквы
Logstash не может читать Topic'и Kafka, если они (Topic’и) содержат большие буквы. Поэтому имена Topic'ов Kafka не должны содержать большие буквы.
7.2. Filebeat некорректно создает Topic’и Kafka
При старте как Filebeat, так и Logstash создают Topic в Kafka, если в конфигурации соответствующего сервиса Topic имеется, а в Kafka он еще отсутствует. Но если Topic создает Filebeat, то Logstash не может читать этот топик. Если же Topic создает Logstash, то все сервисы могут читать этот Topic. Поэтому при создании/изменении конфигурации обеих служб сначала необходимо запустить Logstash и дождаться пока Topic в Kafka будет создан. И только после этого можно запускать Filebeat.
7.3. Требуется переход на следующую строку после json сообщения
При считывании сообщений из логов в формате json, после закрывающей фигурной скобки «}», которая закрывает json, в том случае, когда это последнее сообщение в файле, необходимо добавить переход на следующую строку. Иначе эта фигурная скобка «}» не попадет в обработку json и последнее json-сообщение в файле обрабатывается не корректно (не читается как json).
7.4. Необходимо обращать внимание на кодировку log-файлов
По умолчанию корректно обрабатываются log-файлы только в формате «UTF-8». Поэтому необходимо либо создавать log-файл в формате «UTF-8», либо явно указываться кодировку в Logstash и/или в Filebeat.
Например, я столкнулся с тем, что PowerShell по умолчанию формирует файлы в формате «UTF-16». Пришлось сменить кодировку по умолчанию в скрипте на «UTF-8»:
$PSDefaultParameterValues['*:Encoding'] = 'utf8'8. Послесловие
Что мне хотелось бы добавить в эту статью, но по какой-то причине этого не произошло:
1. Настройка времени жизни индексов в OpenSearch;
2. Настройка Apache Kafka;
3. Настройка аутентификации Active Directory в OpenSearch-Dashboards по протоколу LDAP;
4. Создание визуализаций в OpenSearch-Dashboards;
5. Создание Dashboard’ов в OpenSearch-Dashboards.
На этом я заканчиваю. Всем спасибо за внимание к статье, надеюсь она окажется для кого-то полезной. И оставляйте ваши рецепты настройки OpenSearch в комментариях, думаю это так же будет интересно.
