Надёжный сбор метрик • dernasherbrezon

После нескольких месяцев эксплуатации fluent-bit для сборка метрик, я понял, что эта конфигурация совсем не работает. Я честно следовал инструкциям и читал исходный код, но оказалось, что буферизация во fluent-bit полностью сломана. Нет, она работает согласно документации, но вот на практике абсолютно бесполезна.

Во-первых, согласно документации fluent-bit (версия 4.0) сохраняет данные на диск небольшими кусочками. Эти кусочки (chunks) имеют максимальный размер - 2мб. На практике у меня всегда был размер 4кб. При отключении сети, fluent-bit создавал сотни файлов размером 4кб на каждую метрику. У меня есть теория, согласно которой, fluent-bit создаёт один файл на каждый сэмпл. За один месяц он может создать 259200 файлов для одной метрики и с лёгкостью израсходовать все свободные inode. Браво.

Во-вторых, как только пропадает сеть, fluent-bit начинает перепосылать все эти кусочки данных. Звучит разумно, но дьявол в деталях. Он начинает их перепосылать параллельно и независимо друг от друга! Такое ощущение, что авторы сами не поняли зачем они написали этот код. Весь смысл любого мониторинга и отправки логов в том, чтобы сохранять порядок отправки! Какой смысл перепосылать все кусочки данных, если удалённый сервис недоступен? Ну и как только удалённый сервис становится доступен, то loki, например, говорит, что данные идут непоследовательно и сразу же отбрасывает их. Это невероятно плохо, потому что логи нельзя терять! Одна пропущенная ошибка в логах и непонятно, почему всё сломалось.

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

fluent-bit был самым лучшим решением, которое мне удалось найти. Но не настолько лучшим, как написанное своими руками!

Поиск дизайна

На самом деле самописное решение в данном случае может быть не настолько дикой идеей. Всё, что требовалось от fluent-bit - это прочитать journald и отправить по HTTP. С этим может справиться даже скрипт на bash. Для логов можно написать небольшой скрипт, который читает из journald и, если нет соединения с Loki, просто пытается переподключится. Persistence может быть сконфигурирован на стороне journald:

Скрипт или небольшая программа читает данные из journald и отправляет в Loki. Каждые Х записей, скрипт берёт текущую позицию в journald и сохраняет как чекпоинт в файл. В случае краха системы или процесса в худшем случае он переотправит Х записей. Стандартная at-least-once гарантия, которая вполне себе подходит для обработки логов.

К сожалению, этот дизайн не подходит для обработки метрик. Их нужно где-то локально хранить в случае потери соединения с сервером.

Я использую influxdb для хранения метрик, и авторы как раз для моего случая рекомендуют поднимать реплику, которая будет автоматически синхронизироваться с мастер-копией.

Но тут есть несколько подводных камней:

нужно поднимать реплику. Каждые год-два influx будет выпускать новую версию базы и непонятно будет ли эта реплика работать в новой версии.
для сбора логов всё равно придётся делать самописный скрипт и не хочется добавлять ещё один способ доставки данных офлайн
непонятно сколько ресурсов будет потреблять реплика influxdb

В результате я стал думать: “А как объединить передачу метрик и логов с помощью одного скрипта или минимального набора технологий? Как гарантировать порядок получения сообщений при отправке на центральный сервер?” Это навело меня на ещё одну идею:

Суть заключается в том, чтобы отправлять сообщения в локальный MQTT сервер (mosquitto). Он достаточно легковесный (в памяти занимает около 7мб) и позволяет создавать персистентные топики. Telegraf будет подключаться к этому серверу и забирать оттуда сообщения. Если соединения нет, то ничего страшного - данные будут накапливаться на устройстве и рано или поздно выкачаются сервером. Причём MQTT будет гарантировать порядок сообщений!

У этого подхода есть пара недостатков:

в mosquitto нет механизма очистки старых сообщений. Можно указать время, через которое ВСЕ сообщения для данного клиента удаляться. Я рассчитал, что 250Гб SSD мне должно хватить надолго и этот недостаток не так важен
вместо fluent-bit используется Telegraf. К сожалению, fluent-bit может работать только как сервер MQTT и не может подписываться на другой сервер MQTT. Потребление памяти не так критично, так как сервис будет запущен на центральном сервере
Telegraf пишет ошибку в логи каждый раз, когда пытается переподключиться к MQTT серверу. Достаточно досадная вещь, которую я рассчитывал каким-то образом поправить в настройках нотификаций.

Я радостно начал конфигурировать Telegraf и у меня успешно получилось настроить отправку метрик. Однако, в тот момент, когда надо было добавить получение логов, что-то пошло не так. Надо либо входящий json притягивать в формат метрик influx, либо писать логи в формате influx, а потом конвертировать в json для Loki. Но это ещё не всё: надо настроить роутинг в зависимости от тегов либо в influx либо в Loki и заполнять эти теги из названия топика. В итоге я не справился с птичим языком конфигов Telegraf и решил написать свой сервис на Java. Он максимально простой, легко отлаживаемый и понятный.

Реализация

На самом деле mqtt-collector можно было реализовать двумя скриптами. На bash для отправки логов:

journalctl --all --output=json --follow | jq -c "{ streams: [{     stream: {       hostname: ._HOSTNAME,       syslog_identifier: .SYSLOG_IDENTIFIER, priority: .PRIORITY,           },     values: [[       ((.__REALTIME_TIMESTAMP + \"000\") | tostring),       .MESSAGE     ]]   }] }" | mosquitto_pub -h localhost -p 1883 -t logs -l -u ${MQTT_USERNAME} -P ${MQTT_PASSWORD} -q 1

И на python для отправки метрик:

...

client = mqtt.Client(client_id="system_metrics_publisher")
client.on_connect = on_connect
client.on_disconnect = on_disconnect
client.username_pw_set(username=os.getenv('MQTT_USERNAME'), password=os.getenv('MQTT_PASSWORD'))
client.connect(BROKER, PORT, keepalive=60)
client.loop_start()

def get_system_metrics():
    ...
    metrics_set = (
        f"metrics.cpu,{tag_set} cpu_p={cpu_percent} {timestamp_ns}\n"
        f"node_memory,{tag_set} MemAvailable_bytes={mem_available},MemTotal_bytes={memory.total} {timestamp_ns}\n"
        f"node_network,{tag_set} receive_bytes_total={net_bytes_recv},transmit_bytes_total={net_bytes_sent} {timestamp_ns}\n"
    )
    
    return f"{metrics_set}"

while True:
    line_protocol = get_system_metrics()
    client.publish(TOPIC, line_protocol, qos=1)
    time.sleep(INTERVAL)

Однако на практике я выбрал С. Во-первых, у меня много кода написано на С и я знаю как написать многопоточное приложение, в отличие от Python. Во-вторых, мне действительно не хочется привносить ещё одну технологию, которую я не понимаю как поддерживать. Какие пакеты ей нужны? Как их обновлять? Какая версия Python нужна? В-третьих, приложение достаточно простое и код на С получился примерно похожим на Python.

data-collector я написал на Java с закрытыми глазами и не помню хоть что-то достойного упоминания.

Послесловие

После того, как я выбрал сбор метрик через MQTT, я нашёл множество программ похожих на mqtt-collector: mqtt-collector, linux2mqtt, lnxlink, RPi-Reporter-MQTT2HA-Daemon. У них у всех один недостаток: они написаны на Python или Ruby и не поддерживают отправку логов. Тем не менее мне хотелось бы их упомянуть, возможно, они будут кому-то полезны.