237 lines
8.3 KiB
Go
237 lines
8.3 KiB
Go
// Создано: 2026-04-04
|
|
// Изменено: 2026-04-06 (заменён RabbitMQ на Kafka)
|
|
// mqtt-bridge/main.go — сервис-мост: MQTT (EMQX) → Kafka.
|
|
//
|
|
// Роль в архитектуре:
|
|
// IoT Device → MQTT PUBLISH → EMQX → [mqtt-bridge подписан на "+/telemetry/+"]
|
|
// → Kafka топик "iot.telemetry"
|
|
// → iot-kafka-consumer → Postgres (история телеметрии)
|
|
// → event-dispatcher → Serverless Functions (триггеры)
|
|
//
|
|
// Логика:
|
|
// 1. Подключиться к EMQX как MQTT клиент (credentials из env)
|
|
// 2. Подписаться на топик "+/telemetry/+" (any namespace / telemetry / any device)
|
|
// 3. При получении сообщения — опубликовать в Kafka топик "iot.telemetry"
|
|
// 4. Payload оборачивается в envelope с метаданными (namespace, device_id, ts)
|
|
//
|
|
// Конфигурация через env vars:
|
|
// MQTT_BROKER_URL — tcp://emqx.sless.svc:1883
|
|
// MQTT_USERNAME — username для подключения bridge к EMQX
|
|
// MQTT_PASSWORD — пароль bridge клиента
|
|
// KAFKA_BROKERS — kafka.sless.svc.cluster.local:9092 (заменить на managed в prod)
|
|
//
|
|
// Для возврата к Postgres напрямую: см. git история, коммиты до 2026-04-06.
|
|
|
|
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"log/slog"
|
|
"os"
|
|
"os/signal"
|
|
"strings"
|
|
"syscall"
|
|
"time"
|
|
|
|
mqtt "github.com/eclipse/paho.mqtt.golang"
|
|
kafka "github.com/segmentio/kafka-go"
|
|
)
|
|
|
|
// mqttBridgeConfig — конфигурация сервиса из env vars.
|
|
type mqttBridgeConfig struct {
|
|
MQTTBrokerURL string
|
|
MQTTUsername string
|
|
MQTTPassword string
|
|
KafkaBrokers string
|
|
}
|
|
|
|
// iotTelemetryMessage — envelope сообщения публикуемого в Kafka.
|
|
// Потребители (iot-kafka-consumer, event-dispatcher) читают этот формат.
|
|
type iotTelemetryMessage struct {
|
|
// Namespace — k8s namespace тенанта (из MQTT topic, первый сегмент)
|
|
Namespace string `json:"namespace"`
|
|
// DeviceID — идентификатор устройства (из MQTT topic, третий сегмент)
|
|
DeviceID string `json:"device_id"`
|
|
// Topic — оригинальный MQTT topic
|
|
Topic string `json:"topic"`
|
|
// Payload — данные от устройства (JSON as-is, или строка если не JSON)
|
|
Payload json.RawMessage `json:"payload"`
|
|
// ReceivedAt — время получения сообщения мостом (UTC, RFC3339)
|
|
ReceivedAt string `json:"received_at"`
|
|
}
|
|
|
|
// iotTelemetryTopic — Kafka топик для IoT телеметрии.
|
|
// Все устройства всех тенантов пишут в один топик, изоляция — по полю Namespace в payload.
|
|
const iotTelemetryTopic = "iot.telemetry"
|
|
|
|
func main() {
|
|
log := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo}))
|
|
|
|
cfg := loadBridgeConfig()
|
|
|
|
ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
|
|
defer cancel()
|
|
|
|
log.Info("starting iot-mqtt-bridge",
|
|
"mqtt_broker", cfg.MQTTBrokerURL,
|
|
"mqtt_username", cfg.MQTTUsername,
|
|
"kafka_brokers", cfg.KafkaBrokers,
|
|
)
|
|
|
|
// Kafka writer — асинхронный, с автоматическим созданием топика.
|
|
kafkaWriter := &kafka.Writer{
|
|
Addr: kafka.TCP(strings.Split(cfg.KafkaBrokers, ",")...),
|
|
Topic: iotTelemetryTopic,
|
|
Balancer: &kafka.LeastBytes{},
|
|
// Позволяет продолжать работу при временной недоступности Kafka (буфер в памяти)
|
|
Async: false,
|
|
RequiredAcks: kafka.RequireOne,
|
|
}
|
|
defer kafkaWriter.Close()
|
|
|
|
// Создаём MQTT клиент
|
|
mqttClient, err := connectMQTT(cfg, log)
|
|
if err != nil {
|
|
log.Error("failed to connect to MQTT broker", "err", err)
|
|
os.Exit(1)
|
|
}
|
|
defer mqttClient.Disconnect(500)
|
|
|
|
// Функция-обработчик MQTT сообщений
|
|
messageHandler := buildMQTTMessageHandler(ctx, kafkaWriter, log)
|
|
|
|
// Подписываемся на все telemetry топики всех namespace
|
|
// "+/telemetry/+" = {любой namespace}/telemetry/{любой deviceId}
|
|
const telemetryTopicFilter = "+/telemetry/+"
|
|
token := mqttClient.Subscribe(telemetryTopicFilter, 1, messageHandler)
|
|
token.Wait()
|
|
if token.Error() != nil {
|
|
log.Error("mqtt subscribe failed", "topic", telemetryTopicFilter, "err", token.Error())
|
|
os.Exit(1)
|
|
}
|
|
log.Info("subscribed to MQTT topic", "filter", telemetryTopicFilter)
|
|
|
|
<-ctx.Done()
|
|
log.Info("shutting down iot-mqtt-bridge")
|
|
}
|
|
|
|
// loadBridgeConfig читает конфигурацию из env vars.
|
|
func loadBridgeConfig() mqttBridgeConfig {
|
|
required := func(key string) string {
|
|
v := os.Getenv(key)
|
|
if v == "" {
|
|
slog.Error("required env var not set", "key", key)
|
|
os.Exit(1)
|
|
}
|
|
return v
|
|
}
|
|
|
|
return mqttBridgeConfig{
|
|
MQTTBrokerURL: getEnvOrDefault("MQTT_BROKER_URL", "tcp://emqx.sless.svc:1883"),
|
|
MQTTUsername: required("MQTT_USERNAME"),
|
|
MQTTPassword: required("MQTT_PASSWORD"),
|
|
KafkaBrokers: getEnvOrDefault("KAFKA_BROKERS", "kafka.sless.svc.cluster.local:9092"),
|
|
}
|
|
}
|
|
|
|
func getEnvOrDefault(key, defaultVal string) string {
|
|
if v := os.Getenv(key); v != "" {
|
|
return v
|
|
}
|
|
return defaultVal
|
|
}
|
|
|
|
// connectMQTT устанавливает подключение к EMQX брокеру.
|
|
func connectMQTT(cfg mqttBridgeConfig, log *slog.Logger) (mqtt.Client, error) {
|
|
opts := mqtt.NewClientOptions()
|
|
opts.AddBroker(cfg.MQTTBrokerURL)
|
|
opts.SetClientID("sless-iot-bridge")
|
|
opts.SetUsername(cfg.MQTTUsername)
|
|
opts.SetPassword(cfg.MQTTPassword)
|
|
opts.SetAutoReconnect(true)
|
|
opts.SetConnectRetry(true)
|
|
opts.SetConnectRetryInterval(5 * time.Second)
|
|
opts.SetKeepAlive(30 * time.Second)
|
|
opts.SetCleanSession(false)
|
|
|
|
opts.SetConnectionLostHandler(func(_ mqtt.Client, err error) {
|
|
log.Warn("MQTT connection lost, reconnecting...", "err", err)
|
|
})
|
|
opts.SetReconnectingHandler(func(_ mqtt.Client, _ *mqtt.ClientOptions) {
|
|
log.Info("MQTT reconnecting...")
|
|
})
|
|
opts.SetOnConnectHandler(func(_ mqtt.Client) {
|
|
log.Info("MQTT connected to broker")
|
|
})
|
|
|
|
client := mqtt.NewClient(opts)
|
|
token := client.Connect()
|
|
if !token.WaitTimeout(30 * time.Second) {
|
|
return nil, fmt.Errorf("MQTT connect timeout")
|
|
}
|
|
if token.Error() != nil {
|
|
return nil, fmt.Errorf("MQTT connect: %w", token.Error())
|
|
}
|
|
return client, nil
|
|
}
|
|
|
|
// buildMQTTMessageHandler возвращает обработчик MQTT сообщений.
|
|
// При получении сообщения — публикует envelope в Kafka топик "iot.telemetry".
|
|
// Ключ сообщения Kafka = namespace, для партиционирования по тенанту.
|
|
func buildMQTTMessageHandler(ctx context.Context, w *kafka.Writer, log *slog.Logger) mqtt.MessageHandler {
|
|
return func(_ mqtt.Client, msg mqtt.Message) {
|
|
topic := msg.Topic()
|
|
payload := msg.Payload()
|
|
|
|
// Топик: "{namespace}/telemetry/{deviceId}"
|
|
parts := strings.SplitN(topic, "/", 3)
|
|
if len(parts) != 3 {
|
|
log.Warn("unexpected MQTT topic format, skipping", "topic", topic)
|
|
return
|
|
}
|
|
ns := parts[0]
|
|
deviceID := parts[2]
|
|
|
|
// Нормализуем payload: если не JSON — оборачиваем в строку
|
|
rawPayload := json.RawMessage(payload)
|
|
if !json.Valid(payload) {
|
|
quotedBytes, _ := json.Marshal(string(payload))
|
|
rawPayload = json.RawMessage(quotedBytes)
|
|
}
|
|
|
|
envelope := iotTelemetryMessage{
|
|
Namespace: ns,
|
|
DeviceID: deviceID,
|
|
Topic: topic,
|
|
Payload: rawPayload,
|
|
ReceivedAt: time.Now().UTC().Format(time.RFC3339),
|
|
}
|
|
|
|
body, err := json.Marshal(envelope)
|
|
if err != nil {
|
|
log.Error("marshal telemetry message", "topic", topic, "err", err)
|
|
return
|
|
}
|
|
|
|
// Ключ = namespace — Kafka будет группировать сообщения одного тенанта
|
|
// на одну партицию (для упорядоченной обработки на consumer side)
|
|
err = w.WriteMessages(ctx, kafka.Message{
|
|
Key: []byte(ns),
|
|
Value: body,
|
|
})
|
|
if err != nil {
|
|
log.Error("publish to Kafka", "topic", iotTelemetryTopic, "namespace", ns, "err", err)
|
|
return
|
|
}
|
|
|
|
log.Info("forwarded IoT telemetry to Kafka",
|
|
"mqtt_topic", topic,
|
|
"namespace", ns,
|
|
"device", deviceID,
|
|
"kafka_topic", iotTelemetryTopic,
|
|
)
|
|
}
|
|
}
|