Миграция с Yandex IoT Core на MQTT-решения с открытым исходным кодом

После закрытия Yandex IoT Core вы можете развернуть альтернативное MQTT-решение самостоятельно: на виртуальной машине Yandex Compute Cloud или в кластере Yandex Managed Service for Kubernetes.

Вместо Yandex IoT Core можно использовать следующие опенсорс-решения, которые можно запустить в Yandex Cloud:

• Eclipse Mosquitto — рекомендуемый вариант для большинства сценариев, где Yandex IoT Core использовался как MQTT-брокер.
• TBMQ Community Edition — вариант для высоконагруженных MQTT-сценариев, где нужны кластеризация и долговременная доставка сообщений.
• ThingsBoard Community Edition — вариант для сценариев, где, кроме MQTT, нужны управление устройствами, дашборды, механизм rule engine и пользовательский интерфейс.

Как выбрать целевое решениеКак выбрать целевое решение

В разделе приведено краткое сравнение трех решений. В текущем руководстве описаны подробные шаги развертывания для Eclipse Mosquitto. Чтобы использовать TBMQ и ThingsBoard, смотрите документацию соответствующего решения.

Выбирайте Eclipse Mosquitto, если:

• устройства используют MQTT и подключаются к Yandex IoT Core по TLS;
• основная логика работает вокруг публикации и подписки на топики;
• вы готовы заменить интеграции Yandex IoT Core с Yandex Cloud Functions, Yandex Serverless Containers или Yandex Data Streams собственным обработчиком сообщений;
• вам достаточно одного брокера или отказоустойчивости уровня active/passive.

Выбирайте TBMQ Community Edition, если:

• у вас высокая нагрузка на MQTT-брокер;
• нужны кластеризация, горизонтальное масштабирование и более надежная доставка сообщений;
• вы готовы поддерживать дополнительную инфраструктуру, например Apache Kafka® и базу данных.

Выбирайте ThingsBoard Community Edition, если:

• Yandex IoT Core был частью более широкой IoT-системы;
• вам нужны управление устройствами, дашборды, механизм rule engine, пользовательский интерфейс, хранение телеметрии и обработка событий;
• вы готовы адаптировать устройства и бэкенд под модель данных ThingsBoard.

Что изменится после миграцииЧто изменится после миграции

При переносе с Yandex IoT Core на пользовательское MQTT-решение изменятся несколько важных частей архитектуры.

ЭндпоинтЭндпоинт

В Yandex IoT Core устройства подключались к управляемому эндпоинту сервиса. После миграции вам понадобится создать собственный эндпоинт, например:

mqtt.example.com:8883

Эндпоинт может указывать:

• на публичный IP-адрес виртуальной машины с Eclipse Mosquitto;
• на сетевой балансировщик перед брокером;
• на Service типа LoadBalancer в Yandex Managed Service for Kubernetes.

Для MQTT используйте TCP-порт 8883 и TLS. L7-балансировщик для обычного MQTT не подходит, так как MQTT использует TCP, а не HTTP.

Если часть клиентов подключалась по MQTT поверх WebSocket (например, веб- или браузерные клиенты), настройте в Mosquitto отдельного слушателя с protocol websockets и TLS (обычно на порту 443) и откройте этот порт в группе безопасности.

АутентификацияАутентификация

В Yandex IoT Core устройства могли использовать:

• X.509-сертификаты;
• логин и пароль.

В Mosquitto можно использовать такие же подходы:

• password_file — для аутентификации по логину и паролю;
• require_certificate и use_identity_as_username — для аутентификации по клиентскому сертификату;
• acl_file — для ограничения доступа к топикам.

ТопикиТопики

В Yandex IoT Core устройства и реестры используют следующие топики:

$devices/<device_id>/events
$devices/<device_id>/state
$devices/<device_id>/commands
$devices/<device_id>/config
$registries/<registry_id>/events
$registries/<registry_id>/state
$registries/<registry_id>/commands
$registries/<registry_id>/config

Топики state и config — постоянные: брокер хранит в них последнее сообщение. Их аналог в обычном MQTT реализуется через retained-сообщения (подробнее в разделе ниже).

На время миграции сохраните эту структуру в Mosquitto без изменений ($devices/..., $registries/...), чтобы не трогать прошивку устройств. Все примеры ниже используют именно ее. Переход на топики без символа $ — отдельный необязательный шаг: во время него нужно одновременно обновить прошивку устройств, ACL и подписчиков.

Постоянные топики state и configПостоянные топики state и config

В Yandex IoT Core топики state и config являются постоянными: сервис хранит последнее сообщение в таком топике.

В обычном MQTT похожее поведение реализуется с помощью флага retain. После миграции:

• при публикации состояния устройства в state передавайте сообщение с retain = true;
• при публикации конфигурации устройства в config также используйте retain = true;
• проверьте, что целевой брокер хранит retained-сообщения на постоянном диске.

Пример публикации retained-сообщения через mosquitto_pub:

mosquitto_pub \
  -h mqtt.example.com \
  -p 8883 \
  --cafile rootCA.crt \
  -u <device_id> \
  -P '<password>' \
  -t 'devices/<device_id>/state' \
  -m '{"online":true}' \
  -q 1 \
  -r

Триггеры и экспорт сообщенийТриггеры и экспорт сообщений

Если вы использовали триггеры для вызова функции Cloud Functions или контейнера Serverless Containers, после миграции их нужно заменить собственным подписчиком MQTT.

Такой подписчик должен:

• подписываться на MQTT-топики;
• преобразовывать сообщение в нужный формат;
• вызывать функцию, контейнер, HTTP API или отправлять сообщение в поток данных Yandex Data Streams;
• логировать ошибки и повторять обработку при сбоях.

Пример такого обработчика приведен в разделе Замените триггеры.

Архитектура миграции на MosquittoАрхитектура миграции на Mosquitto

Рекомендуемая минимальная архитектура:

1. Устройства подключаются к Mosquitto по MQTTS на порт 8883.
2. Mosquitto проверяет логин/пароль или клиентский сертификат устройства.
3. Mosquitto применяет ACL и разрешает устройству публиковать и читать только свои топики.
4. Бэкенд-подписчик получает сообщения из топиков events и state.
5. Бэкенд-подписчик передает сообщения в ваши обработчики, базы данных, функции Cloud Functions, контейнеры Serverless Containers или потоки данных Yandex Data Streams.
6. Команды и конфигурация публикуются бэкенд-приложением в топики commands и config.

┌─────────────┐        MQTT/TLS          ┌──────────────┐
│ Устройства  │  ─────────────────────▶  │  Mosquitto   │
└─────────────┘                          └──────┬───────┘
                                                │ MQTT subscribe
                                                ▼
                                      ┌────────────────────┐
                                      │ Backend subscriber │
                                      └─────────┬──────────┘
                                                │
                 ┌──────────────────────────────┼──────────────────────────────┐
                 ▼                              ▼                              ▼
        Cloud Functions              Serverless Containers      Yandex Data Streams

Выгрузите данные из Yandex IoT CoreВыгрузите данные из Yandex IoT Core

Перед миграцией выгрузите список реестров, устройств, сертификатов и алиасов топиков.

Установите и настройте Yandex Cloud CLI, если он еще не установлен:

yc init

Проверьте активную конфигурацию CLI:

yc config get cloud-id
yc config get folder-id
yc config get compute-default-zone

Создайте каталог для экспорта:

FOLDER_ID="$(yc config get folder-id)"
mkdir -p iot-export

Выгрузите список реестров и брокеров. Передавайте --folder-id явно, чтобы экспорт не зависел от каталога, указанного в локальной конфигурации CLI:

yc iot registry list --folder-id "$FOLDER_ID" --format json > iot-export/registries.json
yc iot broker list --folder-id "$FOLDER_ID" --format json > iot-export/brokers.json

Выгрузите устройства, сертификаты реестров и алиасы топиков:

for registry_id in $(jq -r '.[].id' iot-export/registries.json); do
  yc iot device list \
    --folder-id "$FOLDER_ID" \
    --registry-id "$registry_id" \
    --format json > "iot-export/devices-$registry_id.json"

  yc iot registry certificate list \
    --folder-id "$FOLDER_ID" \
    --registry-id "$registry_id" \
    --format json > "iot-export/registry-certs-$registry_id.json"

  yc iot registry list-device-topic-aliases \
    --folder-id "$FOLDER_ID" \
    "$registry_id" \
    --format json > "iot-export/aliases-$registry_id.json"
done

Выгрузите сертификаты устройств:

for file in iot-export/devices-*.json; do
  registry_id="${file#iot-export/devices-}"
  registry_id="${registry_id%.json}"

  mkdir -p "iot-export/device-certs-$registry_id"

  jq -r '.[].id' "$file" | while read -r device_id; do
    yc iot device certificate list \
      --folder-id "$FOLDER_ID" \
      --device-id "$device_id" \
      --format json > "iot-export/device-certs-$registry_id/$device_id.json"
  done
done

Разверните Mosquitto на виртуальной машинеРазверните Mosquitto на виртуальной машине

Ниже приведен пример развертывания Mosquitto в Docker на виртуальной машине. Для продакшен-сценария настройте резервное копирование диска, мониторинг и процедуру восстановления.

Создайте виртуальную машинуСоздайте виртуальную машину

Создайте ВМ Yandex Compute Cloud с Linux-образом и публичным IP-адресом или с доступом через сетевой балансировщик.

Рекомендации:

• используйте отдельный сервисный аккаунт для администрирования инфраструктуры;
• откройте во входящих правилах группы безопасности только порт 8883/TCP;
• ограничьте доступ по SSH;
• храните TLS-ключи и пароли в Yandex Lockbox или в другом защищенном хранилище секретов;
• подключите Yandex Monitoring и Yandex Cloud Logging.

Перед созданием ВМ выберите SSH-ключ и административный CIDR, с которого разрешен доступ по SSH:

ls -1 ~/.ssh/*.pub

SSH_PUBLIC_KEY="$HOME/.ssh/<key>.pub"
SSH_PRIVATE_KEY="${SSH_PUBLIC_KEY%.pub}"

ssh-keygen -y -f "$SSH_PRIVATE_KEY" > /tmp/mqtt-broker-key.pub
diff -u /tmp/mqtt-broker-key.pub "$SSH_PUBLIC_KEY"
chmod 600 "$SSH_PRIVATE_KEY"

ADMIN_CIDR="<your-admin-ip-or-vpn-cidr>/32"

Если приватный ключ защищен ключевой фразой (passphrase), загрузите его в ssh-agent до запуска автоматизации или выберите отдельный ключ для миграции. Для CI и песочницы не полагайтесь только на автоопределение публичного IP через внешние HTTP-сервисы: исходный адрес SSH может отличаться от адреса, который видит такой сервис. В этом случае задайте ADMIN_CIDR явно, например, CIDR вашего VPN или бастионного хоста.

Создайте группу безопасности и настройте минимальные входящие правила:

yc vpc security-group create \
  --name mqtt-broker-sg \
  --network-name default

yc vpc security-group update-rules mqtt-broker-sg \
  --add-rule "direction=ingress,protocol=tcp,port=8883,v4-cidrs=0.0.0.0/0" \
  --add-rule "direction=ingress,protocol=tcp,port=22,v4-cidrs=${ADMIN_CIDR}"

В некоторых версиях Yandex Cloud CLI нет команды yc vpc security-group rule add. Используйте yc vpc security-group update-rules --add-rule ... или проверьте актуальный синтаксис через yc vpc security-group update-rules --help.

Пример создания ВМ с помощью CLI:

yc compute instance create \
  --name mqtt-broker \
  --zone ru-central1-a \
  --network-interface subnet-name=default-kz1-a,nat-ip-version=ipv4,security-group-names=mqtt-broker-sg \
  --create-boot-disk image-folder-id=standard-images,image-family=ubuntu-2204-lts,size=15,type=network-hdd \
  --memory 2 \
  --cores 2 \
  --core-fraction 20 \
  --metadata-from-file user-data=cloud-init.yaml \
  --metadata ssh-keys="ubuntu:$(cat "$SSH_PUBLIC_KEY")"

Установите Docker и Docker Compose на ВМ.

Подготовьте каталог MosquittoПодготовьте каталог Mosquitto

Выполните команды:

mkdir -p mosquitto/config/certs mosquitto/data mosquitto/log
cd mosquitto

sudo chown -R 1883:1883 data log

Создайте файл docker-compose.yaml:

services:
  mosquitto:
    image: eclipse-mosquitto:2
    container_name: mosquitto
    restart: unless-stopped
    ports:
      - "8883:8883"
    volumes:
      - ./config:/mosquitto/config:ro
      - ./data:/mosquitto/data
      - ./log:/mosquitto/log

Подготовьте TLS-сертификат сервераПодготовьте TLS-сертификат сервера

Для продакшен-окружения используйте сертификат для домена, по которому устройства будут подключаться к брокеру, например:

mqtt.example.com

Если устройства умеют проверять публичные корневые сертификаты, можно использовать публичный TLS-сертификат.

Если устройства используют собственный корневой сертификат, положите сертификат сервера и ключ в каталог:

mosquitto/config/certs/server.crt
mosquitto/config/certs/server.key

Если вы используете аутентификацию устройств по клиентским сертификатам, также положите корневой сертификат, которым подписаны клиентские сертификаты:

mosquitto/config/certs/device-ca.crt

Вариант 1. Настройте логин и парольВариант 1. Настройте логин и пароль

Создайте файл config/mosquitto.conf:

persistence true
persistence_location /mosquitto/data/
autosave_interval 60

log_dest stdout

listener 8883 0.0.0.0
protocol mqtt

certfile /mosquitto/config/certs/server.crt
keyfile /mosquitto/config/certs/server.key

allow_anonymous false
password_file /mosquitto/config/passwords
acl_file /mosquitto/config/acl

Создайте файл паролей. Для первого пользователя используйте флаг -c и не создавайте файл заранее: mosquitto_passwd -c должен сам создать новый password_file.

docker run --rm -it \
  -v "$PWD/config:/mosquitto/config" \
  eclipse-mosquitto:2 \
  mosquitto_passwd -c /mosquitto/config/passwords <device_id>

Для неинтерактивной подготовки используйте -b, но не сохраняйте открытые пароли в shell history:

docker run --rm \
  -v "$PWD/config:/mosquitto/config" \
  eclipse-mosquitto:2 \
  mosquitto_passwd -b -c /mosquitto/config/passwords <device_id> '<device_password>'

Для следующих пользователей запускайте команду без -c, чтобы не перезаписать файл:

docker run --rm -it \
  -v "$PWD/config:/mosquitto/config" \
  eclipse-mosquitto:2 \
  mosquitto_passwd /mosquitto/config/passwords <another_device_id>

Добавьте отдельного пользователя для бэкенд-приложений:

docker run --rm -it \
  -v "$PWD/config:/mosquitto/config" \
  eclipse-mosquitto:2 \
  mosquitto_passwd /mosquitto/config/passwords Backend

Вариант 2. Настройте клиентские сертификатыВариант 2. Настройте клиентские сертификаты

Если устройства уже используют X.509-сертификаты и вы хотите сохранить этот подход, настройте mTLS.

Создайте файл config/mosquitto.conf:

persistence true
persistence_location /mosquitto/data/
autosave_interval 60

log_dest stdout

listener 8883 0.0.0.0
protocol mqtt

certfile /mosquitto/config/certs/server.crt
keyfile /mosquitto/config/certs/server.key
cafile /mosquitto/config/certs/device-ca.crt

require_certificate true
use_identity_as_username true

allow_anonymous false
acl_file /mosquitto/config/acl

При такой настройке Mosquitto будет использовать Common Name из клиентского сертификата как имя пользователя. Убедитесь, что Common Name соответствует идентификатору устройства или другому стабильному идентификатору, который вы используете в ACL.

Настройте ACLНастройте ACL

Создайте файл config/acl.

Пример ACL, если имя пользователя совпадает с идентификатором устройства:

pattern write $devices/%u/events/#
pattern write $devices/%u/state/#
pattern read $devices/%u/commands/#
pattern read $devices/%u/config/#

pattern write devices/%u/events/#
pattern write devices/%u/state/#
pattern read devices/%u/commands/#
pattern read devices/%u/config/#

user Backend
topic read $devices/+/events/#
topic read $devices/+/state/#
topic write $devices/+/commands/#
topic write $devices/+/config/#

topic read devices/+/events/#
topic read devices/+/state/#
topic write devices/+/commands/#
topic write devices/+/config/#

В этом примере:

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

Если вы использовали топики реестров, добавьте правила для registries/... или сгенерируйте индивидуальные правила доступа для каждого устройства.

Запустите MosquittoЗапустите Mosquitto

Выполните команды:

docker compose up -d
docker compose logs -f mosquitto

Проверьте, что брокер слушает порт 8883:

docker compose ps

Альтернатива: установка Mosquitto через apt-пакетАльтернатива: установка Mosquitto через apt-пакет

Если в вашей инфраструктуре не используется Docker, Mosquitto можно установить как системный пакет Ubuntu. Этот путь проще для cloud-init, но требует осторожности с конфигурацией.

sudo apt-get update
sudo apt-get install -y mosquitto mosquitto-clients

Дополнительные настройки кладите в /etc/mosquitto/conf.d/<name>.conf, а не в основной /etc/mosquitto/mosquitto.conf.

persistence true
persistence_location /var/lib/mosquitto/

Error: Duplicate persistence_location value in configuration.

sudo grep -RhE '^(persistence|persistence_location|password_file|acl_file|listener)' \
  /etc/mosquitto/mosquitto.conf /etc/mosquitto/conf.d/

Если вы разворачиваете Mosquitto через cloud-init, учитывайте порядок выполнения модулей: write_files отрабатывает до packages, поэтому группы mosquitto еще может не быть. Записывайте файлы как root:root, а владельца меняйте в runcmd после установки пакета.

ACL с топиками $devices/... и $registries/... безопаснее передавать в cloud-init в Base64: так символ $ не будет интерпретирован shell-обвязкой при подготовке метаданных.

Файл /etc/mosquitto/passwd, который вы передаете в cloud-init, должен быть сгенерирован совместимой версией mosquitto_passwd. Если целевая ВМ использует Ubuntu-пакет Mosquitto 2.0.18, не подставляйте файл, созданный произвольной более новой локальной версией mosquitto_passwd: брокер может не прочитать хеш и будет отклонять пользователей.

Пример фрагмента cloud-init для apt-пакета:

#cloud-config
packages:
  - mosquitto
  - mosquitto-clients

write_files:
  - path: /etc/mosquitto/conf.d/iot-migration.conf
    owner: root:root
    permissions: "0644"
    content: |
      listener 8883 0.0.0.0
      protocol mqtt
      certfile /etc/mosquitto/certs/server.crt
      keyfile /etc/mosquitto/certs/server.key
      allow_anonymous false
      password_file /etc/mosquitto/passwd
      acl_file /etc/mosquitto/acl
  - path: /etc/mosquitto/acl
    owner: root:root
    permissions: "0640"
    encoding: b64
    content: <base64-encoded-acl>
  - path: /etc/mosquitto/passwd
    owner: root:root
    permissions: "0640"
    encoding: b64
    content: <base64-encoded-password-file>

runcmd:
  - mkdir -p /etc/mosquitto/certs
  - chown mosquitto:mosquitto /etc/mosquitto/acl /etc/mosquitto/passwd
  - chmod 0640 /etc/mosquitto/acl /etc/mosquitto/passwd
  - systemctl reset-failed mosquitto
  - systemctl restart mosquitto
  - systemctl is-active mosquitto

Пакет Mosquitto может попытаться запустить сервис сразу после установки, еще до выполнения runcmd, когда сертификаты и права на файлы доступа не подготовлены. Из-за нескольких неудачных стартов systemd может перевести сервис в состояние Start request repeated too quickly. Команда systemctl reset-failed mosquitto перед финальным restart сбрасывает этот start-limit.

Не добавляйте в этот conf.d-файл persistence и persistence_location: они уже есть в /etc/mosquitto/mosquitto.conf пакета Ubuntu по умолчанию.

После правки конфигурации выполните команды:

sudo systemctl restart mosquitto
sudo systemctl status mosquitto
sudo ss -tlnp | grep 8883

Для Ubuntu-пакета Mosquitto 2.0.18 не используйте mosquitto -t как проверку конфигурации: эта версия воспринимает -t как неизвестную опцию. Если сервис не стартует, посмотрите диагностику:

sudo journalctl -u mosquitto -n 100 --no-pager
sudo grep -RhE '^(persistence|persistence_location|password_file|acl_file|listener)' \
  /etc/mosquitto/mosquitto.conf /etc/mosquitto/conf.d/

При работе через apt не забудьте дать пользователю mosquitto права на каталоги с данными и логами (chown -R mosquitto:mosquitto /var/lib/mosquitto /var/log/mosquitto) и в группе безопасности открыть только 8883/TCP.

Проверьте подключениеПроверьте подключение

На рабочей или тестовой ВМ установите клиентские утилиты Mosquitto.

Подпишитесь на топик от имени бэкенд-пользователя:

mosquitto_sub \
  -h mqtt.example.com \
  -p 8883 \
  --cafile rootCA.crt \
  -u Backend \
  -P '<Backend_password>' \
  -t '$devices/+/events/#' \
  -q 1

Опубликуйте тестовое сообщение от имени устройства:

mosquitto_pub \
  -h mqtt.example.com \
  -p 8883 \
  --cafile rootCA.crt \
  -u <device_id> \
  -P '<device_password>' \
  -t '$devices/<device_id>/events' \
  -m '{"temperature":25}' \
  -q 1

Проверьте retained-сообщение для аналога постоянного топика state:

mosquitto_pub \
  -h mqtt.example.com \
  -p 8883 \
  --cafile rootCA.crt \
  -u <device_id> \
  -P '<device_password>' \
  -t '$devices/<device_id>/state' \
  -m '{"online":true}' \
  -q 1 \
  -r

Подпишитесь на этот топик новым клиентом. Он должен сразу получить последнее retained-сообщение:

mosquitto_sub \
  -h mqtt.example.com \
  -p 8883 \
  --cafile rootCA.crt \
  -u Backend \
  -P '<Backend_password>' \
  -t '$devices/<device_id>/state' \
  -q 1

Проверьте изоляцию устройств через ACLПроверьте изоляцию устройств через ACL

ACL — самая частая точка, в которой ошибки обнаруживаются уже после миграции. Обязательно проверьте, что одно устройство не может читать или писать в топики другого.

Сценарий 1. Чужие команды не доходят. Сначала от имени устройства <device_a> подпишитесь на топик чужого устройства <device_b>:

mosquitto_sub \
  -h mqtt.example.com \
  -p 8883 \
  --cafile rootCA.crt \
  -u <device_a> \
  -P '<device_a_password>' \
  -t '$devices/<device_b>/commands' \
  -q 1 \
  -W 5

Параметр -W 5 ограничивает время работы 5 секундами. Пока подписчик работает, опубликуйте сообщение в $devices/<device_b>/commands от бэкенда и убедитесь, что подписчик ничего не получает. В логах брокера для устройства <device_a> должна быть запись об отказе ACL.

Сценарий 2. Чужие события не публикуются. От имени устройства <device_a> попробуйте опубликовать в чужой топик:

mosquitto_pub \
  -h mqtt.example.com \
  -p 8883 \
  --cafile rootCA.crt \
  -u <device_a> \
  -P '<device_a_password>' \
  -t '$devices/<device_b>/events' \
  -m '{"spoofed":true}' \
  -q 1

Команда может выполниться без видимой ошибки на клиенте, но сообщение не должно дойти до бэкенд-подписки на $devices/<device_b>/events/#. Проверяйте именно отсутствие сообщения у бэкенд-подписчика и отказ ACL в логах брокера, а не только код ответа mosquitto_pub.

Сценарий 3. Анонимное подключение запрещено. Подключитесь без учетных данных:

mosquitto_sub \
  -h mqtt.example.com \
  -p 8883 \
  --cafile rootCA.crt \
  -t '$devices/<device_a>/events' \
  -W 5

Клиент должен получить ошибку Connection Refused: not authorised.

Замените триггерыЗамените триггеры

В Yandex IoT Core можно было настроить триггер, который вызывал функцию Cloud Functions или контейнер Serverless Containers при получении MQTT-сообщения. В пользовательском MQTT-брокере такую интеграцию нужно реализовать отдельным сервисом-подписчиком.

Ниже пример простого подписчика на Python. Он получает MQTT-сообщение и отправляет его в HTTP-обработчик.

Создайте файл subscriber.py:

import base64
import json
import os
import queue
import threading
import time
from urllib import request

import paho.mqtt.client as mqtt


MQTT_HOST = os.environ["MQTT_HOST"]
MQTT_PORT = int(os.getenv("MQTT_PORT", "8883"))
MQTT_USERNAME = os.environ["MQTT_USERNAME"]
MQTT_PASSWORD = os.environ["MQTT_PASSWORD"]
MQTT_TOPIC = os.getenv("MQTT_TOPIC", "$devices/+/events/#")
MQTT_CLIENT_ID = os.getenv("MQTT_CLIENT_ID", "iot-core-migration-subscriber")
MQTT_CA_FILE = os.getenv("MQTT_CA_FILE", "rootCA.crt")
WORKER_COUNT = int(os.getenv("WORKER_COUNT", "4"))

HANDLER_URL = os.environ["HANDLER_URL"]

# Очередь и пул воркеров развязывают прием сообщений из MQTT и их обработку.
# Если обрабатывать HTTP-запрос прямо в on_message, медленный обработчик
# заблокирует сетевой цикл MQTT, и клиент отключится по keepalive.
message_queue: queue.Queue = queue.Queue(maxsize=10000)


def build_event(msg: mqtt.MQTTMessage) -> dict:
    payload_base64 = base64.b64encode(msg.payload).decode("ascii")
    topic_parts = msg.topic.split("/")

    event = {
        "mqtt_topic": msg.topic,
        "payload": payload_base64,
        "is_base64": True,
        "qos": msg.qos,
        "retain": msg.retain,
    }

    if len(topic_parts) >= 2 and topic_parts[0] in ("$devices", "devices"):
        event["device_id"] = topic_parts[1]

    if len(topic_parts) >= 2 and topic_parts[0] in ("$registries", "registries"):
        event["registry_id"] = topic_parts[1]

    return event


def send_to_handler(event: dict) -> None:
    body = json.dumps(event).encode("utf-8")
    http_request = request.Request(
        HANDLER_URL,
        data=body,
        headers={"Content-Type": "application/json"},
        method="POST",
    )

    with request.urlopen(http_request, timeout=10) as response:
        response.read()


def on_connect(client, userdata, flags, rc):
    if rc != 0:
        print(f"MQTT connection failed with rc={rc}")
        return

    print(f"Connected to MQTT broker. Subscribing to {MQTT_TOPIC}")
    client.subscribe(MQTT_TOPIC, qos=1)


def worker() -> None:
    while True:
        event = message_queue.get()
        try:
            for attempt in range(3):
                try:
                    send_to_handler(event)
                    break
                except Exception as error:
                    print(f"Handler call failed, attempt={attempt + 1}, error={error}")
                    time.sleep(2 ** attempt)
            else:
                print(f"Message was not processed: topic={event.get('mqtt_topic')}")
        finally:
            message_queue.task_done()


def on_message(client, userdata, msg):
    # Кладем событие в очередь и сразу возвращаем управление сетевому циклу:
    # put() приостанавливает прием, только когда все воркеры заняты.
    message_queue.put(build_event(msg))


for _ in range(WORKER_COUNT):
    threading.Thread(target=worker, daemon=True).start()

client = mqtt.Client(client_id=MQTT_CLIENT_ID)
client.username_pw_set(MQTT_USERNAME, MQTT_PASSWORD)
client.tls_set(ca_certs=MQTT_CA_FILE)

client.on_connect = on_connect
client.on_message = on_message

client.connect(MQTT_HOST, MQTT_PORT, keepalive=60)
client.loop_forever()

Создайте файл requirements.txt:

paho-mqtt==1.6.1

Запустите подписчика:

export MQTT_HOST=mqtt.example.com
export MQTT_PORT=8883
export MQTT_USERNAME=Backend
export MQTT_PASSWORD='<Backend_password>'
export MQTT_TOPIC='$devices/+/events/#'
export MQTT_CA_FILE=./rootCA.crt
export HANDLER_URL='https://<your-handler-url>'

python3 -m pip install -r requirements.txt
python3 subscriber.py

Такого подписчика можно запустить:

• на виртуальной машине Compute Cloud;
• в кластере Managed Service for Kubernetes;
• в контейнере Serverless Containers, если приложение постоянно поддерживает MQTT-соединение;
• рядом с другими бэкенд-сервисами, которые обрабатывают телеметрию.

Замените экспорт в Yandex Data StreamsЗамените экспорт в Yandex Data Streams

Если вы использовали экспорт сообщений из Yandex IoT Core в потоки данных Yandex Data Streams, используйте один из вариантов:

• MQTT-подписчик читает сообщения из Mosquitto и отправляет их в поток данных Yandex Data Streams;
• Бэкенд-приложение одновременно публикует данные в MQTT и в поток данных Yandex Data Streams;
• Потоковая обработка выполняется в отдельном сервисе, который подписан на MQTT-топики.

Общий принцип:

Mosquitto → MQTT subscriber → Yandex Data Streams

При реализации учитывайте:

• размер сообщения;
• порядок сообщений;
• повторную отправку при ошибках;
• идемпотентность записи;
• мониторинг задержек и ошибок записи.

Перенесите устройстваПеренесите устройства

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

Рекомендуемый порядок:

1. Создайте новый DNS-адрес MQTT-брокера, например mqtt.example.com.
2. Настройте TLS-сертификат для этого адреса.
3. Создайте тестовое устройство или тестовую группу устройств.
4. Проверьте публикацию events и state.
5. Проверьте получение commands и config.
6. Проверьте, что бэкенд получает сообщения и корректно их обрабатывает.
7. Переведите небольшую пилотную группу устройств.
8. Сравните телеметрию в старом и новом контуре.
9. Переведите остальные устройства партиями.
10. После завершения миграции отключите публикацию в Yandex IoT Core.

Если прошивка устройств поддерживает одновременную публикацию в два MQTT-брокера, используйте период двойной отправки. Это поможет сравнить данные и снизить риск потери телеметрии.

Удалите тестовые ресурсыУдалите тестовые ресурсы

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

yc compute instance delete mqtt-broker
yc vpc security-group delete mqtt-broker-sg

Не добавляйте к этим командам флаг --quiet без проверки установленной версии Yandex Cloud CLI: в некоторых версиях он не поддерживается для yc compute instance delete. Если нужна неинтерактивная очистка, сначала проверьте актуальный синтаксис:

yc compute instance delete --help

Развертывание Mosquitto в Yandex Managed Service for KubernetesРазвертывание Mosquitto в Yandex Managed Service for Kubernetes

Mosquitto можно запустить в кластере Managed Service for Kubernetes. Для продакшен-окружения используйте:

• StatefulSet;
• PersistentVolumeClaim для каталога /mosquitto/data;
• Secret для TLS-ключей и паролей;
• ConfigMap для mosquitto.conf и ACL;
• Service типа LoadBalancer для TCP-порта 8883;
• PodDisruptionBudget;
• регулярные резервные копии persistent volume.

Пример Service:

apiVersion: v1
kind: Service
metadata:
  name: mosquitto
spec:
  type: LoadBalancer
  selector:
    app: mosquitto
  ports:
    - name: mqtts
      port: 8883
      targetPort: 8883
      protocol: TCP

Когда выбрать TBMQ Community EditionКогда выбрать TBMQ Community Edition

TBMQ Community Edition — MQTT-брокер с открытым исходным кодом, ориентированный на высокую нагрузку и кластерную архитектуру.

Рассмотрите TBMQ, если:

• один Mosquitto-брокер становится узким местом;
• нужна горизонтальная масштабируемость;
• важна долговременная доставка сообщений;
• нужна отказоустойчивость на уровне брокерского кластера;
• вы готовы сопровождать Apache Kafka®, базу данных и другие компоненты.

Типовая архитектура:

Устройства → TBMQ → Kafka → бэкенд / обработчики / хранилища

TBMQ сложнее Mosquitto в эксплуатации, но лучше подходит для сценариев, где MQTT-брокер является критичным высоконагруженным компонентом.

Базу данных для TBMQ или ThingsBoard можно вынести в сервис Yandex Managed Service for PostgreSQL, чтобы не администрировать СУБД самостоятельно, либо развернуть ее в кластере вместе с остальными компонентами.

Когда выбрать ThingsBoard Community EditionКогда выбрать ThingsBoard Community Edition

ThingsBoard Community Edition — IoT-платформа с открытым исходным кодом. Она включает MQTT API, управление устройствами, хранение телеметрии, дашборды и механизм rule engine.

Рассмотрите ThingsBoard CE, если:

• вам нужна не только передача MQTT-сообщений, но и IoT-платформа;
• пользователям нужен пользовательский интерфейс для устройств и телеметрии;
• нужны дашборды и механизм rule engine;
• вы хотите управлять данными учетных записей устройств в платформе;
• вы готовы адаптировать топики, payload и бэкенд-интеграции под ThingsBoard.

ThingsBoard CE не является drop-in-заменой Yandex IoT Core. Устройства и бэкенд-приложения обычно нужно адаптировать под API и модель данных ThingsBoard.

Типовая архитектура:

Устройства → ThingsBoard CE → Rule Engine → дашборд / бэкенд / хранилища

Также возможна архитектура, где Mosquitto остается MQTT-брокером, а ThingsBoard получает данные через интеграцию или API-шлюз:

Устройства → Mosquitto → ThingsBoard CE → дашборд / Rule Engine

Что не переносится автоматическиЧто не переносится автоматически

При миграции на опенсорс-решение не переносятся автоматически:

• реестры и устройства как управляемые ресурсы Yandex Cloud;
• алиасы топиков;
• пароли устройств и реестров;
• интеграции с IAM;
• публикация сообщений через gRPC-интерфейс Yandex IoT Core;
• триггеры;
• экспорт в Yandex Data Streams;
• мониторинговые топики Yandex IoT Core;
• поведение постоянных топиков без настройки retained-сообщений;
• роли и права доступа на уровне Yandex Cloud.

Для каждого пункта нужно выбрать замену:

• реестры и устройства — собственная база данных, конфигурация брокера или IoT-платформа.
• пароли — новый password_file, Dynamic Security plugin Mosquitto или данные учетных записей в IoT-платформе.
• сертификаты — собственный корневой сертификат и mTLS.
• алиасы — явные MQTT-топики или слой совместимости в бэкенде.
• публикация через gRPC-интерфейс — прямой совместимой замены нет, потому что в Yandex IoT Core использовалась IAM-авторизация. Если такой сценарий нужен, реализуйте отдельный бэкенд-адаптер с нужной схемой авторизации, который будет принимать gRPC/API-запросы от ваших систем и публиковать сообщения в Mosquitto;
• триггеры — MQTT-подписчик.
• экспорт — MQTT-подписчик, который пишет в потоки данных Yandex Data Streams или другое хранилище.
• мониторинг — метрики брокера, логи, проверки работоспособности и пользовательские heartbeat-топики.

Рекомендации по эксплуатацииРекомендации по эксплуатации

БезопасностьБезопасность

• Используйте MQTTS на порту 8883.
• Отключите анонимный доступ.
• Ограничивайте устройства через ACL.
• Не используйте один общий пароль для всех устройств.
• Храните пароли, приватные ключи и корневой сертификат в защищенном хранилище.
• Регулярно ротируйте учетные данные.
• Ограничьте входящий трафик группами безопасности.
• Не открывайте административные интерфейсы в интернет.

НадежностьНадежность

• Храните персистентные данные брокера на постоянном диске.
• Настройте резервное копирование диска или каталога данных.
• Документируйте процедуру восстановления.
• Проверяйте восстановление на тестовом стенде.
• Используйте QoS 1 для важных сообщений.
• Убедитесь, что для retained-сообщений включено сохранение.
• Для критичных сценариев рассмотрите active/passive-схему или кластерный брокер.

НаблюдаемостьНаблюдаемость

• Собирайте логи Mosquitto в Yandex Cloud Logging.
• Настройте алерты на недоступность порта 8883.
• Мониторьте CPU, RAM, диск и сетевой трафик ВМ или поды.
• Отслеживайте количество подключенных клиентов.
• Отслеживайте ошибки аутентификации.
• Добавьте метрики в бэкенд-подписчика: количество сообщений, ошибки обработки, задержку доставки.

ПроизводительностьПроизводительность

• Проведите нагрузочное тестирование до переноса всех устройств.
• Проверьте лимиты CPU, RAM и сетевой пропускной способности.
• Разделяйте топики так, чтобы бэкенд-подписчики не получали лишние сообщения.
• Не используйте слишком широкие wildcard-подписки без необходимости.
• Для больших инсталляций планируйте шардинг или переход на кластерный брокер.

План миграцииПлан миграции

Используйте этот чек-лист для контроля миграции:

• Выбрано целевое решение: Eclipse Mosquitto, TBMQ CE или ThingsBoard CE.
• Выгружен список реестров, устройств, сертификатов и алиасов.
• Сгенерированы новые пароли устройств и подготовлен план их доставки в прошивку.
• Создан новый эндпоинт MQTT.
• Настроен TLS.
• Настроена аутентификация устройств.
• Настроены ACL для обоих префиксов: $devices/... и devices/... (если поэтапно мигрируете часть устройств).
• Проверена публикация events.
• Проверена публикация state с retained-сообщениями.
• Проверено получение commands.
• Проверено получение config.
• Прогнан smoke-тест ACL-изоляции (устройство не читает и не пишет чужие топики, анонимное подключение отклоняется).
• Конфигурация Mosquitto проверена на отсутствие дубликатов опций (persistence, persistence_location, listener и т. п. встречаются ровно один раз во всех файлах конфигурации).
• Реализована замена триггеров.
• Реализована замена экспорта в Yandex Data Streams, если он использовался.
• Настроены логи, метрики и алерты.
• Настроены резервные копии.
• Проведен пилотный перенос небольшой группы устройств.
• Проведено сравнение данных в старом и новом контуре.
• Перенесены все устройства.
• Отключена зависимость продакшен-систем от Yandex IoT Core.

Полезные ссылкиПолезные ссылки

• Документация Eclipse Mosquitto
• Документация TBMQ
• Документация ThingsBoard
• Документация Yandex Compute Cloud
• Документация Yandex Managed Service for Kubernetes
• Документация Yandex Object Storage
• Документация Yandex Lockbox
• Документация Yandex Monitoring
• Документация Yandex Cloud Logging