Балансировка нагрузки фиксированной сессии — кластеризация брокера MQTT, часть 2

В последнем посте: Балансировка нагрузки — кластеризация брокера MQTT, часть 1, мы ввели балансировку нагрузки MQTT в целом: балансировка нагрузки может применяться либо на транспортном уровне, либо на прикладном уровне. Теперь пришло время погрузиться в балансировку нагрузки прикладного уровня, самую интересную часть: sticky-session.

Этот пост состоит из 2 частей, первая часть посвящена тому, что такое сеансы MQTT, и проблемам обработки сеансов в распределенном кластере брокера MQTT; вторая часть — запачкать руки, предоставив HAProxy 2.4 балансировщик нагрузки перед EMQX 4.3 кластер, чтобы в полной мере воспользоваться преимуществами балансировки нагрузки фиксированных сеансов.

MQTT-сессия

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

По этой причине брокер MQTT, который обслуживает клиента, должен поддерживать сеанс для клиента (запрос для каждого клиента, установив для флага «Clean-Session» значение false). Таким образом, темы, на которые в данный момент подписан подписчик, и сообщения (QoS1 и 2), доставленные в эти темы и т. д., сохраняются брокером, даже когда клиент отключен.

Когда клиент с сохраненным сеансом повторно подключается, ему не нужно повторно подписываться на темы, и брокер должен отправить клиенту все ожидающие сообщения.

Ранее мы писали статью о сеансы MQTTэто отличное чтение, если вас интересуют дополнительные технические подробности о сеансах MQTT.

Захват сеанса

Все становится немного сложнее, когда MQTT-брокеры образуют кластер. С точки зрения клиента, существует несколько брокеров, к которым можно подключиться, и трудно понять, к какому из них лучше всего подключиться. Нам нужен еще один критический компонент в сети: балансировщик нагрузки. Балансировщик нагрузки становится точкой доступа для всего кластера и направляет подключения от клиентов к одному из брокеров в кластере.

Если клиент подключается через балансировщик нагрузки к посреднику (например, узлу 1), а затем отключается и подключается позже, существует вероятность того, что новое соединение может быть направлено к другому посреднику в кластере (например, узлу 3). В этом случае node3 должен начать отправлять ожидающие сообщения клиенту, когда клиент был отключен.

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

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

Sticky Session на помощь

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

Чтобы балансировщик нагрузки распределял соединения «прилипающим» образом, брокеру необходимо знать идентификатор клиента (или иногда имя пользователя) в запросе на подключение — для этого балансировщик нагрузки должен проверять пакеты MQTT для поиска такой информации.

После получения идентификатора клиента (или имени пользователя) для кластера статического размера брокер может хэшировать идентификатор клиента (или имя пользователя) в идентификатор брокера. Или для большей гибкости балансировщик нагрузки может поддерживать таблицу сопоставления идентификатора клиента (или имени пользователя) с идентификатором целевого узла.

В следующем разделе мы продемонстрируем стратегию закрепления таблиц в HAProxy 2.4.

Прилепленная сессия с HAProxy 2.4

Чтобы свести к минимуму предварительные требования, в этом демонстрационном кластере мы запустим два узла EMQX и HAProxy 2.4 в док-контейнерах.

Создайте сеть Docker

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

docker network create test.net

Запустите два узла EMQX 4.3

Чтобы узлы могли соединяться друг с другом, имя контейнера и имя узла EMQX должны быть назначены в пространстве имен сети (test.net).

Начальный узел1

docker run -d \
    --name n1.test.net \
    --net test.net \
    -e EMQX_NODE_NAME=emqx@n1.test.net \
    -e EMQX_LISTENER __TCP__ EXTERNAL__PROXY_PROTOCOL=on \
    emqx/emqx:4.3.7

Начальный узел2

docker run -d \
    --name n2.test.net \
    --net test.net \
    -e EMQX_NODE_NAME=emqx@n2.test.net \
    -e EMQX_LISTENER __TCP__ EXTERNAL__PROXY_PROTOCOL=on \
    emqx/emqx:4.3.7

Обратите внимание на переменную окружения

EMQX_LISTENER __TCP__ EXTERNAL__PROXY_PROTOCOL. Это включает бинарный прокси-протокол для прослушивателей TCP, чтобы брокер мог получать такую ​​информацию, как реальный IP-адрес клиента, а не балансировщик нагрузки.

Присоединение узлов EMQX к кластеру

docker exec -it n2.test.net emqx_ctl cluster join emqx@n1.test.net

Если все пойдет как положено, должен быть напечатан такой журнал

[EMQX] emqx shutdown for join
Join the cluster successfully.
Cluster status: #{running_nodes => ['emqx@n1.test.net','emqx@n2.test.net'], stopped_nodes => []}

Запустите HAProxy 2.4

Создать файл /tmp/haproxy.configс содержанием ниже

global
    log stdout format raw daemon debug
    nbproc 1
    nbthread 2
    cpu-map auto:1/1-2 0-1
    # Enable the HAProxy Runtime API
    # e.g. echo "show table emqx_tcp_back" | sudo socat stdio tcp4-connect:172.100.239.4:9999
    stats socket :9999 level admin expose-fd listeners

defaults
    log global
    mode tcp
    option tcplog
    maxconn 1024000
    timeout connect 30000
    timeout client 600s
    timeout server 600s

frontend emqx_tcp
   mode tcp
   option tcplog
   bind *:1883
   default_backend emqx_tcp_back

backend emqx_tcp_back
    mode tcp

    # Create a stick table for session persistence
    stick-table type string len 32 size 100k expire 30m

    # Use ClientID / client_identifier as persistence key
    stick on req.payload(0,0),mqtt_field_value(connect,client_identifier)

    # send proxy-protocol v2 headers
    server emqx1 n1.test.net:1883 check-send-proxy send-proxy-v2
    server emqx2 n2.test.net:1883 check-send-proxy send-proxy-v2

Запустите haproxy в тестовой сети докеров:

docker run -d \
    --net test.net \
    --name proxy.test.net \
    -p 9999:9999 \
    -v /tmp/haproxy.cfg:/haproxy.cfg \
    haproxy:2.4 haproxy -f /haproxy.cfg

Проверьте это

Теперь мы используем популярный MQTT-клиент mosquitto (также в докере), чтобы протестировать его.

Заводим подписчика (с именем subscriber1), который подписывается на t/#тема

docker run --rm -it --net test.net eclipse-mosquitto \
    mosquitto_sub -h proxy.test.net -t 't/#' -I subscriber1

А затем опубликовать hello сообщение для t/xyzот другого клиента

docker run --rm -it --net test.net eclipse-mosquitto \
    mosquitto_pub -h proxy.test.net -t 't/xyz' -m 'hello'

Абонент должен распечатать hello сообщение, если все работает, как ожидалось.

Осмотрите Sticky Table в HAProxy

С помощью этой команды мы также можем проверить прикрепленную таблицу, созданную в HAProxy. Это требует socat команда, поэтому я запускаю ее с хоста докера.

show table emqx_tcp_back" | sudo socat stdio tcp4-connect:127.0.0.1:9999

Это должно напечатать текущие соединения, как показано ниже:

# table: emqx_external_tcp_listners, type: string, size:102400, used:1
0x7f930c033d90: key=subscriber1 use=0 exp=1793903 server_id=2 server_key=emqx2

В этом примере клиент subscriber1 привязан к серверу emqx2.

Первоначально опубликовано на