[Перевод] Отправка метрик Pi-Hole в Graylog с помощью Zabbix20.11.2024 18:16

Метрики с нескольких инстансов pi-hole на дашборде Graylog

В начале 2021 года я написал статью, посвященную сбору детальных метрик Pi-Hole с помощью telegraf и influxdb с последующим отображением на дашбордах Grafana. Прошло несколько лет и я перешел на Graylog (для логирования) и Zabbix (для контроля за ошибками и производительностью). В связи с этим я решил пересмотреть свою старую реализацию и перенести весь функционал с оригинального TIG-стека в Graylog, используя агент Zabbix для мониторинга и инициации загрузки данных.

Оригинальная архитектура

Моя первоначальная реализация извлекала данные из sqlite базы данных FTL pihole с помощью shell-скрипта и преобразовывала каждую строку в line protocol Influxdb, совместимый с плагином telegraph exec.

sqlite3 ->shell script->telegraf->influxdb

Shell-скрипт вызывался агентом telegraf и производил выходные данные, которые отправлялись в influxdb. Я был доволен результатом и для просмотра данных создал в Grafana дашборд. И эта схема работала некоторое время, пока в конце 2021 года я не столкнулся с проблемами с кардинальностью в Influxdb. Это было не ограничение Influxdb, а результат недочетов в моей собственной архитектуре. Когда я создавал базу данных в influxdb, я забыл установить период хранения и за несколько месяцев достиг лимита max-values-per-tag в 100000.

В конце 2020 года Influxdb анонсировал v2, призывая людей переходить на эту версию. Я хотел перейти на эту версию, узнав о ряде новых фич, которые она предлагала. Но как только я попытался настроить v2, я понял, что язык запросов полностью отличается от v1. Мне нужно было заново разбираться, как настроить базу данных influx и как делать к ней запросы для Grafana. Документации по запросам на тот момент было не так много, и я не хотел тратить много времени и сил на изучение нового языка запросов.

Прошло пару лет

Я был занят другими делами и некоторое время пренебрегал своим стеком Telegraf, Influxdb, Grafana. После миграции моего инстанса Graylog с 4.3 на 5.0 и тестирования Zabbix я решил пойти в другом направлении. Я использовал Zabbix для всех метрик производительности и ошибок, а Graylog — для всех логов и связанных с ними метрик. В Zabbix также были уведомления, которых мне не хватало в моей оригинальной системе.

Новая архитектура

Вместо того чтобы отправлять line protocol сообщения в Influxdb, я буду отправлять syslog-сообщения в Graylog. Эти syslog-сообщения, содержащие JSON требуемого формата, периодически отправляются с помощью агента Zabbix. Для отправки этих данных в Graylog я использую агент Zabbix, что дает мне возможность реализовать мониторинг процесса, которая отсутствовал в моем первоначальном проекте. Zabbix, используя UserParameter, будет периодически запускать скрипт извлечения данных, что похоже на то, как для его запуска я использовал telegraf. Это позволит мне отслеживать загрузку данных.

sqlite->shell script->syslog->graylog

Изменения в скрипте извлечения данных

Со времен, когда я работал над моей первоначальной реализацией, база данных FTL pihole претерпела ряд изменений. Одно из изменений — использование представления для таблицы queries в базе данных FTL.

$ /usr/bin/sqlite3 /etc/pihole/pihole-FTL.db
SQLite version 3.34.1 2021-01-20 14:10:07
Enter ".help" for usage hints.
sqlite> .schema queries
CREATE VIEW queries AS SELECT id, timestamp, type, status, CASE typeof(domain) WHEN 'integer' THEN (SELECT domain FROM domain_by_id d WHERE d.id = q.domain) ELSE domain END domain,CASE typeof(client) WHEN 'integer' THEN (SELECT ip FROM client_by_id c WHERE c.id = q.client) ELSE client END client,CASE typeof(forward) WHEN 'integer' THEN (SELECT forward FROM forward_by_id f WHERE f.id = q.forward) ELSE forward END forward,CASE typeof(additional_info) WHEN 'integer' THEN (SELECT content FROM addinfo_by_id a WHERE a.id = q.additional_info) ELSE additional_info END additional_info, reply_type, reply_time, dnssec FROM query_storage q
/* queries(id,timestamp,type,status,domain,client,forward,additional_info,reply_type,reply_time,dnssec) */;
sqlite>

Подробное описание каждого поля в операторе CREAT VIEW, который вы видите выше, можно найти в документации pihole. Ниже приведен мой обновленный скрипт для извлечения данных из sqlite и отправки их в Graylog через syslog.

#!/bin/bash
PATH=$PATH:/bin:/usr/bin:/usr/local/bin
SQLITE=/usr/bin/sqlite3
FTLDB=/etc/pihole/pihole-FTL.db
LATEST=/var/log/zabbix/latest
LAST=/var/log/zabbix/lastrec
MAXLOAD=50
SELECTLASTREC="SELECT id FROM queries ORDER by id desc limit 1;"

# получаем последнюю запись
if [[ ! -f "${LATEST}" ]]; then
  logger "Creating file ${LATEST}"
  touch ${LATEST}
fi
if [[ ! -w "${LATEST}" ]]; then
  logger "Can't write to file ${LATEST}"
  exit
fi
LATESTREC=$(${SQLITE} ${FTLDB} "${SELECTLASTREC}"|tee ${LATEST})
retval=$?
if [[ $retval -ne "0" ]]; then
  logger "${SQLITE} exited with a non zero value: ${retval} trying to get the last record"
  exit ${retval}
fi
logger "Latest record in sqlite is: $(<${LATEST}) value";
if [[ ! -f "${LAST}" ]]; then
  logger "Creating file ${LAST}"
  touch ${LAST}
  cp ${LATEST} ${LAST}
  LASTREC=LATESTREC
else
  LASTREC=$(<${LAST})
fi
if [[ ! -w "${LAST}" ]]; then
  logger "Can't write to file ${LAST}"
  exit
fi
logger "Last record inserted: ${LASTREC}";
DELTA=$((${LATESTREC}-${LASTREC}))
if [[ ${DELTA} -gt ${MAXLOAD} ]]; then
  logger "Caping load to only ${MAXLOAD} records"
  LATESTREC=$((${LASTREC}+${MAXLOAD}))
  logger "New LATEST record ${LATESTREC}"
else
  logger "loading all ${DELTA} records"
fi
# запрашиваем из базы данных все записи между lastrec и latest.
SELECTDELTA="select json_object('timestamp',strftime('%Y-%m-%d %H:%M:%f',datetime(q.timestamp,'unixepoch')),'type',q.type,'status',q.status,'domain',q.domain,'client',q.client,'client_name',n.name,'forward',q.forward,'id',q.id,'reply_type',q.reply_type,'reply_time',q.reply_time,'dnssec',q.dnssec) from queries as q inner join network_addresses as n on q.client = n.ip where q.id > ${LASTREC} and q.id <= ${LATESTREC};"

# подсчитываем строки, отправленные в syslog
LINES=$(${SQLITE} ${FTLDB} "${SELECTDELTA}" 2>>/tmp/blah |tee >(wc -l) >(logger -p local2.info) >/dev/null)

retval=$?
if [[ $retval -ne "0" ]]; then
  logger "${SQLITE} exited with a non zero value: ${retval}"
  exit ${retval}
fi
# обновляем lastrec значением latest
printf '%s\n' ${LATESTREC} > ${LAST}
printf '{\"loaded\":%s,\"delta\":%s}' ${LINES} ${DELTA}

Начиная с верхней части скрипта, одним из изменений, которые я сделал, было перемещение файлов состояния

LATEST=/var/log/zabbix/latest — последние данные из sqlite
LAST=/var/log/zabbix/lastrec — последняя запись, загруженная в Graylog

из каталога /tmp на постоянное хранение. Это гарантирует, что при перезагрузке машины она начнет загружаться с последней записи, загруженной в Graylog. Раньше я хранил их в /tmp, и перезагрузка системы приводила к пробелам в данных.

Двигаясь вниз по скрипту к строке, которая начинается с DELTA, я подсчитываю количество строк между последним загруженным в Graylog значением и последним значением в sqlite. Если это число меньше 50, все строки отправляются в Graylog. Если больше 50, то я ограничиваю количество записей, отправляемых в Graylog ровно до 50.

Следующее изменение, вероятно, самое большое — в строке SELECTDELTA. Это запрос для sqlite. Мое предыдущее решение форматировало выходные строки для line protocol Influxdb, новое же выводит JSON-объекты. Чтобы сделать его более читабельным, я скопировал и переформатировал оператор select ниже.

select json_object(
'timestamp',strftime('%Y-%m-%d %H:%M:%f',datetime(q.timestamp,'unixepoch')),
'type',q.type,
'status',q.status,
'domain',q.domain,
'client',q.client,
'client_name',n.name,
'forward',q.forward,
'id',q.id,
'reply_type',q.reply_type,
'reply_time',q.reply_time,
'dnssec',q.dnssec) from 
    queries as q 
    inner join network_addresses as n 
    on q.client = n.ip 
    where q.id > ${LASTREC} and q.id <= ${LATESTREC};

Первое серьезное изменение в строке запроса — использование функции json_object для вывода результирующего запроса в корректный JSON-объект для каждой строки выходных данных. Таймстемп нужно отправлять в Graylog в определенном формате. Подробнее о формате Graylog я расскажу чуть позже, но первое, что необходимо сделать, это преобразовать значение unixtime в date/time строку с помощью функции datetime. Параметр unixepoch необходим для правильной интерпретации числа, указанного в timestamp. Graylog также требует таймстемп с точностью до миллисекунд, что требует пересчета результирующей строки в формат секунд %f. Поскольку исходный таймстемп не имеет миллисекундной точности, нам нужно заполнить значения миллисекунд нулями.

usr/bin/sqlite3 /etc/pihole/pihole-FTL.db 'select datetime()';
2024-06-09 02:02:12
/usr/bin/sqlite3 /etc/pihole/pihole-FTL.db 'select strftime('\''%Y-%m-%d %H:%M:%f'\'',datetime())';
2024-06-09 02:02:16.000

Следующие несколько полей, за исключением client_name, взяты из стандартного представления запроса. Я хотел передать имя хоста в Graylog, чтобы в диаграммах Graylog можно было легко определить имя хоста клиента. Для этого я объединил представление запроса с таблицей network_addressess в on q.client = n.ip. Полученный запрос выдает следующие результаты

/usr/bin/sqlite3 /etc/pihole/pihole-FTL.db 'select json_object('\''timestamp'\'',strftime('\''%Y-%m-%d %H:%M:%f'\'',datetime(q.timestamp,'\''unixepoch'\'')),'\''type'\'',q.type,'\''status'\'',q.status,'\''domain'\'',q.domain,'\''client'\'',q.client,'\''client_name'\'',n.name,'\''forward'\'',q.forward,'\''id'\'',q.id,'\''reply_type'\'',q.reply_type,'\''reply_time'\'',q.reply_time,'\''dnssec'\'',q.dnssec) from queries as q inner join network_addresses as n on q.client = n.ip where q.id > 9054310 and q.id <= 9054313;'
{"timestamp":"2024-05-22 21:18:46.000","type":16,"status":2,"domain":"portal.azure.com","client":"192.168.1.90","client_name":"studio.home","forward":"192.168.1.1#53","id":9054311,"reply_type":3,"reply_time":0.1059,"dnssec":0}
{"timestamp":"2024-05-22 21:18:57.000","type":1,"status":3,"domain":"waa-pa.clients6.google.com","client":"192.168.1.90","client_name":"studio.home","forward":null,"id":9054312,"reply_type":4,"reply_time":0.0001,"dnssec":0}
{"timestamp":"2024-05-22 21:18:57.000","type":16,"status":2,"domain":"waa-pa.clients6.google.com","client":"192.168.1.90","client_name":"studio.home","forward":"192.168.1.1#53","id":9054313,"reply_type":1,"reply_time":0.0261,"dnssec":0}

Следующее важное изменение — переход с line protocol Influx на syslog в стандартном выводе. Раньше на стандартный вывод отправлялся line protocol, откуда агент Telegraf направлял его в Influxdb. Для Graylog же мне нужно отправлять данные через syslog. Вот как выглядит строка моего скрипта, в которой это происходит:

# подсчитываем строки, отправленные в syslog
LINES=$(${SQLITE} ${FTLDB} "${SELECTDELTA}" 2>>/tmp/blah \

|tee >(wc -l) >(logger -p local2.info) >/dev/null)

Первая часть просто присваивает переменной LINES вывод подоболочки (subshell), вызванной с помощью $(somecommand) (количество строк, отправленных в Graylog). Эта подоболочка выполняет запрос к базе данных FTL piholedb и отправляет строки в unix-утилиту tee. Утилита tee отправляет строки в unix-утилиту wc для подсчета количества строк, полученных в результате выполнения запроса к стандарту. Это значение присваивается LINES. Вторая часть tee отправляет строки на локальный сервер syslog. Сервер syslog на pihole настроен на пересылку всех сообщений syslog в Graylog.

$ grep graylog /etc/rsyslog.conf
*.* @graylog.home

Финальная часть скрипта проверяет последнюю команду, вернувшуюся с нулевым кодом завершения.

# Обновляем lastrec значением latest
printf '%s\n' ${LATESTREC} > ${LAST}
printf '{\"loaded\":%s,\"delta\":%s}' ${LINES} ${DELTA

Наконец, файл lastrec (LAST) обновляется последней записью, которая была отправлена в Graylog, и скрипт возвращает JSON-объект в zabbix для записи двух метрик процесса загрузки: количества загруженных записей и разницы (delta). Этот скрипт вызывается агентом zabbix, уже запущенным на моем хосте.

Конфигурация агента Zabbix

Для запуска скрипта используется агент Zabbix, запущенный на pi-hole. Скрипт находится в файле /usr/local/bin/piholestats. Чтобы настроить агент Zabbix на выполнение скрипта, нам нужно использовать параметр UserParameter.

$ grep ^UserParameter /etc/zabbix/zabbix_agent2.conf
UserParameter=pihole.records,/usr/local/bin/piholestats

После добавления этого параметра в файл агента Zabbix вы можете либо перезапустить агент

sudo service zabbix-agent2 restart

либо использовать параметр -R (runtime control)

zabbix_agentd -R userparameter_reload

Но будьте осторожны: если вы запустите этот скрипт не под пользователем zabbix, у вас могут возникнуть проблемы с правами доступа. Дело в том, что у пользователя, запускающего скрипт, будут созданы файлы состояния, используемые для отслеживания загрузки данных, если их еще нет. Если вы запустите скрипт под другим пользователем, а затем запустите его с агентом zabbix, у вас могут возникнуть проблемы с правами доступа.

/var/log/zabbix $ ls -lasrt la*
4 -rw-rw-r-- 1 zabbix zabbix 9 Jun 30 15:51 latest
4 -rw-rw-r-- 1 zabbix zabbix 9 Jun 30 15:51 lastrec

Я рекомендую для начала протестировать эту конфигурацию с сервера Zabbix с помощью команды zabbix_get.

Конфигурация сервера Zabbix

Вы можете протестировать скрипт UserParameter с сервера zabbix, выполнив команду zabbix_get

zabbix_get -s 192.168.1.6 -k pihole.records

Убедитесь, что он возвращает данные в формате JSON, прежде чем продолжить. Запустив этот скрипт, вы также должны увидеть в Graylog записи syslog из pi-hole.

Для сбора данных из агента я создал шаблон на своем сервере Zabbix. Этот шаблон содержит элемент, который вызывает скрипт из UserParameter, определенного выше, каждые 30 секунд, и два зависимых элемента для извлечения количества загруженных записей и разницы (delta) между последней загруженной записью и последней записью в базе данных FLT pi-hole.

Мастер-элемент, установленный в результате вызова скрипта UserParameter

Зависимый элемент, хранящий информацию о том, сколько записей было отправлено в Graylog

Шаг предварительной обработки с помощью JSONPath для извлечения количества отправленных записей

Такая конфигурация позволяет мне отслеживать процесс загрузки данных и легко изменять его частоту, если это необходимо. Теперь мне нужно обновить graylog, чтобы использовать данные, отправляемые syslog.

Конфигурация Graylog

Стыдно признаться, но я потратил огромное количество времени, пытаясь понять, как внести в Graylog данные из sqlite, только, чтобы в итоге узнать, что unixtime — это UTC. Я не знал, что Unix-время на самом деле определяется как UTC. Отсюда следует:

0 в Unix-времени — это ровно полночь по Гринвичу 1 января 1970 года, после этого Unix-время инкрементируется на 1 за каждую не дополнительную секунду. Например, 00:00:00 UTC 1 января 1971 года представлено в Unix-времени как 31536000. Время до Unix-эпохи представлено в виде отрицательных значений (в системах, которые их поддерживают), где значение декрементируется на 1 за каждую не дополнительную секунду до эпохи. Например, 00:00:00 UTC 1 января 1969 года представлено в Unix-времени как -31536000. Каждый день в Unix-времени состоит ровно из 86400 секунд.

Когда я только начинал переход от line protocol Influx DB к syslog, одно из изменений, с которым мне пришлось столкнуться, было связано с тем, что line protocol просто передает unix-таймстемп из sqlite в influxdb (через telegraf). Line protocol позволяет указывать таймстемп в сообщении в unix-формате, а pihole сохраняет свои запросы с unix-временем. В случае с syslog таймстемп будет записываться в момент получения сообщения. О сохранении исходного таймстемпа в sqlite я расскажу позже.

Я создал новый набор индексов, содержащий сообщения из pihole.

Набор индексов в Graylog для данных Pi-Hole.

Затем я создал поток для всех статистических данных pihole в graylog и назвал его piholestats.

Создаем поток с набором индексов pihole

Я направляю в этот поток все сообщения с syslog facility local2.

Определяем маршрут для всех сообщений с facility, определенным как local2

После того, как я направил сообщения с local2 в поток pihostats, мне нужно установить правила для обработки сообщений конвейером. Начину я с парсинга JSON-пейлоада в сообщении syslog. Я использую local2 в своей сети исключительно для логов pihole, поэтому я могу обрабатывать все сообщения, отправляемые в поток как JSON:

rule "JSON Parser" 
when
   true
then
    let json_string = regex("pi-hole[2]* zabbix: (.*)",to_string($message.message))["0"];
    set_fields(to_map(parse_json(to_string(json_string))));
end

В моей сети два инстанса pi-hole. Это позволяет мне тестировать, обновлять или останавливать один из них, не влияя на доступ в интернет. Поэтому в регулярном выражении выше есть pi-hole[2]* (pi-hole и pi-hole2).

Я добавил на первую стадию конвейера правило для разбора JSON-пейлоада сообщений syslog на отдельные поля:

Стадия 0 конвейера piholestats

Возвращаясь к таймстемпам, разобравшись, что unix-время неявно означает UTC, я прочитал несколько сообщений на Graylog. Это сообщение помогло мне понять, что обязательное поле timestamp — это то, с чем мне нужно работать.

По умолчанию Graylog записывает в поле timestamp таймстемп, указанный в логе, но в случае, если он не может автоматически распарсить его из сообщения syslog, он запишет туда время обработки сообщения Graylog, чтобы избежать потери этого сообщения («timestamp» — одно из трех обязательных полей для всех сообщений в Graylog, остальные два — «message» и «source»). Самый простой способ исправить это — использовать конвейеры обработки для извлечения таймстемпа из сообщения, чтобы мы могли заменить им поле timestamp. У @chris.black-gl мог заваляться где-нибудь пример правила для этого случая.

Итак, теперь я знаю, что нужно заменить — поле timestamp. Я полагаю, что заменяю его в приведенном выше правиле конвейера. Давайте взглянем на сообщение из вывода скрипта

{
  "timestamp": "2024-05-22 21:18:46.000",
  "type": 16,
  "status": 2,
  "domain": "portal.azure.com",
  "client": "192.168.1.90",
  "client_name": "studio.home",
  "forward": "192.168.1.1#53",
  "id": 9054311,
  "reply_type": 3,
  "reply_time": 0.1059,
  "dnssec": 0
}

В JSON этого сообщения есть поле «timestamp». Теперь проблема в самом значении, а точнее — в его точности. Это сообщение явно указывает нам на то, что это НЕ микросекунды (подтверждено Яном ниже в том же треде). Это сообщение, похоже, указывает на то, что поддерживаются миллисекунды. Это был, вероятно, самый важный тред, который помогл направить меня в нужное русло.

Параметр Timestamp должен быть таймстемпом эпохи unix, а не date string: https://archivedocs.graylog.org/en/latest/pages/gelf.html (обновленная ссылка)

timestamp number

Секунды с начала эпохи UNIX с необязательными десятичными знаками для миллисекунд ДОЛЖНЫ быть установлены клиентской библиотекой. Если они отсутствует, сервер установит текущий таймстемп (сейчас).

Наконец, исследуя все это, я наткнулся на эту справку от Elastic (изначально graylog был построен на elastic, но теперь поддерживает и opensearch).

Поскольку Graylog — это фронт для Elastic|Open, логично, что таймстемп должна поддерживать тип, который легко воспринимается Elastic.

Тип поля даты

В JSON нет типа данных для даты, поэтому даты в Elasticsearch могут быть либо:

* строками, содержащими форматированные даты, например,»2015–01–01» или »2015/01/01 12:10:30».

* числами, представляющими миллисекунды с начала эпохи.
* числами, представляющими секунды с начала эпохи (конфигурация).

Внутри программы даты преобразуются в UTC (если указан часовой пояс) и хранятся в long number, представляющем собой количество миллисекунд с начала эпохи.

Я уделил так много внимания этому шагу (определению формата ввода, необходимого для поля timestamp) потому, что он занял у меня очень много времени и был самой сложной частью работы. Документация Graylog изменилась (не знаю как давно), и я не смог найти спецификацию по приему сообщений или их особенностям. К счастью, сообщество и его модераторы восполняют эти пробелы. Завершить этот раздел хочу, подытожив, что sqlite хранит данные в UTC с точностью до секунд, а Graylog принимает данные в UTC и ожидает миллисекундной точности. Поэтому скрипт извлечения должен преобразовать данные из sqlite из секунд в миллисекунды.

Лукап-таблицы в Graylog

Поместив данные pihole в Graylog, я в принципе был доволен результатом, но также мне хотелось увидеть ряд дополнительных данных, которых нет в сообщении лога. Например, если вы посмотрите на приведенное выше сообщение в формате JSON, то оно имеет поле type со значением 16. Мне не хотелось выделять под эти значения память, поэтому я создал пару лукап-таблиц (lookup table — LUT).

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

Лукапы в стадии 2

Sqlite-запросы содержат несколько числовых значений, которые являются представлением для компактного хранения и простоты передачи. Эти значения преобразуются с помощью лукап-таблиц, чтобы их было легче читать как в сообщениях логов, так и на дашбордах. Первая лукап-таблица предназначена для статуса ответа на запрос. Я кодирую два значения в лукап-таблице status type, ориентируясь на статус отсюда.

# cat status-type.csv
status,status_name
0,Unknown|NOT_YET_KNOWN
1,Blocked|BLOCKED_GRAVITY
2,Allowed|ALLOWED_FORWARDED
3,Allowed|ALLOWED_CACHE
4,Blocked|BLOCKED_BLACKLIST_REGEX
5,Blocked|BLOCKED_BLACKLIST_EXACT
6,Blocked|BLOCKED_UPSTREAM_IP
7,Blocked|BLOCKED_UPSTREAM_DGW
8,Blocked|BLOCKED_UPSTREAM_NXDOMAIN
9,Blocked|BLOCKED_GRAVITY_DEEP_CNAME
10,Blocked|BLOCKED_BLACKLIST_REGEX_DEEP_CNAME
11,Blocked|BLOCKED_BLACKLIST_EXACT_DEEP_CNAME
12,Allowed|ALLOWED_RETRIED
13,Allowed|ALLOWED_RETRIED_IGNORING
14,Allowed|ALLOWED_ALREADY_FORWARDED
15,Blocked|BLOCKED_DB_BUSY
16,Blocked|BLOCKED_SPECIAL_DOMAIN
17,Allowed|ALLOWED_REPLIED_STALE_CACHE

Первое значение — это ключ, который отправляется в Graylog в качестве статуса ответа на запрос. Значение для этого ключа расшифровывается в Blocked или Allowed, за чем следует подробная причина. Файл представляет из себя csv и использует адаптер данных.

Адаптер данных статуса запроса pihole

Для этого адаптера также создается соответствующий кэш.

Кэш статуса запросов

Наконец, с помощью адаптера данных и кэша, созданного выше, можно построить лукап-таблицу.

Конфигурация лукап-таблицы.

Окончательная конфигурация таблицы выглядит следующим образом:

Эта таблица затем используется в правиле конвейера и задает два поля status_type (blocked или allowed) и status_detail — более подробное описание.

rule "DNS query status lookup"

when
    has_field("status")
    then
//        debug("get in status name rule");
//        debug($message.status);
//        let new_type = lookup_value("status-name", to_string($message.status));
        let type_array = split("\\|",to_string(lookup_value("status-name",to_string($message.status))));
//        debug(type_array);
//        let type_array = split("|",to_string(new_type));
//        debug(type_array[1]);
//        debug(type_array[0]);
       set_field("status_type", type_array[0]);
       set_field("status_detail",type_array[1]);
end

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

Я решил использовать в лукапе два значения, чтобы было проще сообщать о простом соотношении заблокированных и разрешенных статусов ответов. Я также хотел иметь возможность исследовать конкретные типы статуса ответа без необходимости обращаться к документации. Этот процесс настройки лукап-таблицы был мной повторен и для полей с типом запроса и типом ответа. CSV-файл для типов запросов взят отсюда.

# cat query-type.csv
type,query_type
1,A
2,AAAA
3,ANY
4,SRV
5,SOA
6,PTR
7,TXT
8,NAPTR
9,MX
10,DS
11,RRSIG
12,DNSKEY
13,NS
14,OTHER
15,SVCB
16,HTTPS

Поддерживаемые типы ответов приведены здесь.

# cat reply-name.csv
ID,reply_name
0,unknown
1,NODATA
2,NXDOMAIN
3,CNAME
4,IP
5,DOMAIN
6,RRNAME
7,SERVFAIL
8,REFUSED
9,NOTIMP
10,OTHER
11,DNSSEC
12,NONE
13,BLOB

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

Дашборды Graylog

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

Blocked vs Allowed — использует лукап-таблицу для преобразования кода в тип статуса
Request type — использует другую лукап-таблицу для преобразования кода типа запроса
DNS Server — поскольку у меня два сервера pi-hole, я могу видеть, какой сервер ответил на конкретный запрос
Query status detail — здесь используется та же лукап-таблица, что и в первом случае, с более подробной причиной.

Первый ряд виджетов дашборда данных pihole в graylog

Следующий ряд виджетов представляет из себя «топы» в pihole (графика опущена)

Топ клиентов
Топ заблокированных клиентом
Топ разрешенных доменов
Топ заблокированных доменов

Следующие три сложенных графика показывают ту же информацию, что и в первом ряду, в течение всего периода выполнения запроса.

Blocked vs Allowed — с течением времени
detailed status — с течением времени
by host — с течением времени

Последние две диаграммы показывают детализацию по IP-адресам для клиентов и серверов

Некоторые различия, которые вы можете видеть в общих подсчетах, связаны с ограничениями, установленными в виджете построения графиков. Например, количество IP на графике немного меньше, чем общее количество по серверу, потому что на графике по IP отображаются только топ 30 IP.

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

Выберите элемент легенды и добавьте его в запрос

Результат запроса с обновленными критериями фильтрации (обратите внимание на страницу № 3 под запросом)

Дополнительные диаграммы с примененным запросом

Страница 3 дашборда, конкретные сообщения, соответствующие критериям запроса

Дашборд Zabbix

Наконец, используя Zabbix для управления процессом загрузки данных, я могу контролировать их загрузку. Приведенные ниже данные показывают в Zabbix тот же период времени, что и все остальные графики Graylog.

Один день загрузки данных

Один час загрузки данных

На графиках можно заметить, что я ограничиваю количество сообщений, которые отправляются из pi-hole в graylog. Ограничение (50) было установлено произвольно. Я опасался перегрузить graylog логами dns-запросов. На графике выше видно, что один из pihole’ов часто отправляет больше чем 50 сообщений. Справа показано, сколько сообщений он должен отправить, а слева видно, что он отправил максимум за несколько запусков. Возможно, мне нужно пересмотреть этот момент, но пока, похоже, все работает как надо.

Наблюдения

Реализовав свое первоначальное решение по мониторингу двух инстансов pi-hole с помощью influxdb и Grafana, я начал понимать, что, хотя метрики были важны, мне не хватало анализа. Было сложно динамически изменять и поворачивать данные, чтобы ответить на конкретные вопросы. Я мог ответить на основные вопросы, но было сложнее диагностировать проблемы, не имея возможности увидеть их в контексте. Кроме того, у меня была не очень надежная архитектура. Я столкнулся с проблемами кардинальности. Но это не изъян Influx, а скорее плохое планирование с моей стороны (я не учесть скорость изменения данных).

Один из вопросов, на который мне нужно было ответить (в первую очередь самому себе), — «и что?». Здорово, что я могу собирать данные о различных системах с помощью Grafana, Telegraf и Influxdb, но что мне со всем этим делать? Можно ли из этого извлечь какую‑то пользу? У меня не были настроены оповещения, если бы я и решил, что на что‑то следует оперативно реагировать. Одной из причин, по которой я рассматривал influxdbv2, было оповещение. Когда я нашел Zabbix, я отказался от процесса обновления и начал переносить все на эту платформу.

DNS‑запросы имеют тенденцию смещаться в сторону первой записи в resolv.conf. Судя по количеству запросов, которые я вижу на обоих моих DNS‑серверах. Это не проблема, но это также означает, что вам, скорее всего, нужно планировать, что один сервер будет обрабатывать всю DNS‑нагрузку.

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

Лучшие практики по мониторингу инфраструктуры и отдельных её компонентов (приложения, баз данных, и тд) можно изучить под руководством экспертов на онлайн-курсе «Observability: мониторинг, логирование, трейсинг». На странице курса можно ознакомиться с программой и посмотреть записи открытых уроков.