Настраиваем jmx exporter для kafka в prometheus

исходные данные - есть докер с запущенной кафкой.

1) задаем конфиг /etc/jmx_exporter/config.yaml

---
startDelaySeconds: 3
hostPort: KAFKA_HOST:KAFKA_PORT
username:
password:
ssl: false
whitelistObjectNames: ["kafka.server:type=BrokerTopicMetrics,*"]
rules:
  - pattern: ".*"

2) запуск экспортера, либо в компоуз, либо отдельно, например так:

$ cat /etc/systemd/system/jmx_exporter.service
[Unit]
Description=jmx exporter
Requires=docker.service
After=docker.service

[Service]
Restart=always
RestartSec=3
ExecStartPre=/bin/sh -c "/usr/bin/docker rm -f jmx_exporter 2> /dev/null || /bin/true"
ExecStart=/usr/bin/docker run --rm -a STDIN -a STDOUT -a STDERR -v "/etc/jmx_exporter/config.yaml:/opt/jmx_exporter/config.yml" --name jmx_exporter -p "8081:5556" sscaling/jmx-prometheus-exporter
ExecStop=/usr/bin/docker stop jmx_exporter

[Install]
WantedBy=multi-user.target

3) в самой кафке, например в компоузе, нужно указать jmx endpoint
порты:

    ports:
      - "0.0.0.0:${KAFKA_ADVERTISED_PORT}:${KAFKA_ADVERTISED_PORT}"
      - "0.0.0.0:${KAFKA_JMX_PORT}:${KAFKA_JMX_PORT}"

и опцию JMX

      KAFKA_JMX_PORT: ${KAFKA_JMX_PORT}
      KAFKA_JMX_OPTS: -Dcom.sun.management.jmxremote=true -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Djava.rmi.server.hostname=${KAFKA_ADVERTISED_HOST_NAME} -Djava.net.preferIPv4Stack=true -Dcom.sun.management.jmxremote.port=${KAFKA_JMX_PORT} -Dcom.sun.management.jmxremote.rmi.port=${KAFKA_JMX_PORT}

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

+ статья-пример, что можно взять для мониторинга из такого экспортёра: https://sematext.com/blog/kafka-metrics-to-monitor/#toc-network-request-rate-2

#kafka #prometheus

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

#kafka #zookeper

jinja как не ставить последнюю запятую в цикле

дано: переменная с количеством серверов
задача: составить строку вида

zookeeper.connect=zookeeper1:2181,zookeeper2:2181,zookeeper3:2181/kafka

в целом решается простым циклом for + 2 небольших нюанса:
1) вместо последней запятой стоит /kafka
2) необходимо убрать newline который добавляется на каждом проходе

Решение:

zookeeper.connect=
{%- for server in range(servers_count) -%}
zookeeper{{ server+1 }}:2181{% if not loop.last %},{% endif %}
{%- endfor %}/kafka

все довольно просто - запятая ставится конструкцией {% if not loop.last %},{% endif %}, а переносы строк убираются постановкой дефиса у открывающих и закрывающих тегов {% и %}.

з.ы. счет начинается с нуля, поэтому для красоты именования серверов инкрементирую переменную server.
з.з.ы еще подсказали вот такой способ:

zookeeper.connect={{ range(1, servers_count + 1) | map('regex_replace', '^(.*)$', 'zookeeper\\1:2181') | join(',') }}/kafka

#jinja #ansible #kafka #zookeeper

Debezium и некоторые мои исследования

Q: возможно ли переключить коннектор на другой сервер?
A: это делается через удаление коннектора методом delete и создания нового. Можно сделать обновление через PUT, но через delete надёжнее. Подробнее можно смотреть доку к кафка коннекторам. в доке дебезиума этого нихрена нет.

Q: можно ли объединить коннекторы в один топик? в том смысле что мы запускаем 2 дебезиума на разных серверах, например.
А: Как показали тесты, нет. записи в schemas_history начинают дублироваться, а параметры коннектора вообще в случае разнесения на разные группы не могут быть никак в одних топиках, иначе будет работать только один коннектор из двух. В итоге на каждый коннектор должен быть 5 топиков: 3 для хранения параметров коннектора и 2 для хранения схемы базы и истории изменений схемы.

Q: что происходит при ротации binlog, если коннектор был выключен
A: Если бинлог не удалился, но отротировался, то будет запрошен существующий бинлог и продолжит с последнего считанного места. После чего постепенно перейдет на чтение актуального бинлога.

Q: можно ли чтобы изменения с разных коннекторов, которые читаю разные таблицы писать в один топик?
А: да, это делается через трансформации, описывается регулярное выражение источник и из него можно формировать целевое название очереди. Этот же прицип можно использовать для переименования топика. Только у них есть баг что это не работает в некоторых версиях. помимо указанной версии я это еще в 1.1 встретил. в 1.3 всё ок.

Q: добавление новой таблицы "на лету"
A: Схема создается дебезиумом при первом инсерте в таблицу. и далее на лету подхватывается. Если таблица только создана и пустая, то это не поедет в кафку.

Q: как правильно настроить retention на топиках?
А: согласно документации требуется бесконечный или очень долгий только на history топиках Infinite (or very long) retention (no compaction!), partition count of 1. И на всех нужно выставить cleanup.policy=compact
В итоге мы делаем как-то так:
--partitions 1 --replication-factor 3 --config retention.ms=-1 --config min.insync.replicas=1 --config cleanup.policy=compact

Q: Можно ли полностью контролировать имена топиков в кафке?
А: да, вполне.
database.server.name - это просто логическое имя. оно будет использовано по-умолчанию как префикс для именования топика с данными (что не очень удобно), но к счастью все топики с данными через трансформацию можно переименовать.
statuses configs offsets - топики которые мы можем создать с любым именем и при старте дебезиума указать их в env.
схему можно задать в конфиге например так: "database.server.name": "debezium.database_name.logical_connector_name.schema"
историю схемы: "database.history.kafka.topic": "debezium.database_name.logical_connector_name.history"
а затем через трансформацию переименовать целевые топики с данными

    "transforms": "Reroute",
    "transforms.Reroute.type": "io.debezium.transforms.ByLogicalTableRouter",
    "transforms.Reroute.topic.regex": "debezium.logical_connector_name.logical_connector_name.schema.<REGEXP>",
    "transforms.Reroute.topic.replacement": "my_fucking_awesome_new_name_with_group_capture_$1",

#debezium #mysql #kafka

Debezium helm chart

Мы с командой Ситимобил написали helm chart для разворачивания debezium в k8s. К сожалению корпоративного репозитория у нас нет, поэтому выкладываю так (по согласованию с руководством, естественно).

Чарт умеет разворачивать дебезиум, на запуске создавать или обновлять коннектор, дополняет апстримовый контейнер прометеус-метриками, а также создаст все нужные очереди в kafka.

Обратите внимание, в чарте используется подход 1 чарт = 1 коннектор, т.к. для высоконагруженных баз объединение конфигов играет злую шутку и может привести к долгому даунтайму. Приятного использования!

https://github.com/bykvaadm/debezium-helm-chart

З.ы. Дашборд для метрик выложим попозже

#debezium
#helm
#kafka
#mysql
#kubernetes