feat(iot): Этапы 2-7 — MQTT auth, IoT API, EMQX, mqtt-bridge, E2E demo

Этап 2+4: internal/api/handler/iot_device_handler.go - MQTTAuth: POST /internal/mqtt/auth (без JWT, для EMQX) - CreateIoTDevice, ListIoTDevices, GetIoTDevice (c password), DeleteIoTDevice, UpdateIoTDevice - crypto/subtle.ConstantTimeCompare против timing attacks Этап 4: internal/api/router.go - /v1/namespaces/{ns}/iot/devices CRUD - /internal/mqtt/auth (без JWT middleware) Этап 3: deployments/k8s/emqx.yaml - EMQX 5.5.1, emqx.conf (HOCON) с HTTP auth backend - Сервис exposure: 1883 (MQTT), 8083 (WS), 18083 (Dashboard) Этап 3: iot/cmd/mqtt-bridge/main.go - paho.mqtt.golang: подписка на +/telemetry/+ - amqp091-go: publish в iot.{namespace}.telemetry - deployments/k8s/iot-mqtt-bridge.yaml Этап 7: examples/IOT/ — E2E demo (main.tf, handler.py, README.md) go.mod: добавлен github.com/eclipse/paho.mqtt.golang v1.5.1 go build ./... — ошибок нет
2026-04-04 09:45:23 +03:00
parent 716efafda8
commit 1e53766c46
14 changed files with 1209 additions and 151 deletions
@@ -0,0 +1,83 @@
+# IoT MVP — E2E Demo
+
+## Что делает этот пример
+
+Показывает полную цепочку:
+
+```
+IoT Device (mosquitto_pub)
+  → MQTT PUBLISH → EMQX (HTTP auth → sless-operator)
+    → [iot-mqtt-bridge подписан на "+/telemetry/+"]
+      → RabbitMQ queue "iot.{namespace}.telemetry"
+        → event-dispatcher
+          → POST → serverless function (handler.py)
+```
+
+## Предусловия
+
+1. EMQX запущен: `kubectl apply -f deployments/k8s/emqx.yaml`
+2. iot-mqtt-bridge запущен: `kubectl apply -f deployments/k8s/iot-mqtt-bridge.yaml`
+3. event-dispatcher запущен (уже должен работать)
+
+## Запуск
+
+```bash
+# Установить переменные
+export API_TOKEN="your-jwt-token"
+export NAMESPACE="sless-abc123def456"  # твой namespace
+
+# Инициализировать
+terraform init
+terraform apply \
+  -var="namespace=${NAMESPACE}" \
+  -var="api_token=${API_TOKEN}"
+
+# Получить credentials
+MQTT_USER=$(terraform output -raw mqtt_username)
+MQTT_PASS=$(terraform output -raw mqtt_password)
+MQTT_TOPIC=$(terraform output -raw mqtt_topic)
+
+echo "MQTT user: ${MQTT_USER}"
+echo "MQTT topic: ${MQTT_TOPIC}"
+```
+
+## Отправить тестовое сообщение
+
+```bash
+# Через mosquitto_pub (из пода внутри кластера)
+kubectl run mqtt-test --rm -i --image=eclipse-mosquitto --restart=Never -- \
+  mosquitto_pub \
+  -h emqx.sless.svc \
+  -p 1883 \
+  -u "${MQTT_USER}" \
+  -P "${MQTT_PASS}" \
+  -t "${MQTT_TOPIC}" \
+  -m '{"temperature": 22.5, "humidity": 65, "unit": "celsius"}'
+```
+
+## Проверить что функция вызвалась
+
+```bash
+# Логи event-dispatcher
+kubectl logs -n sless deployment/event-dispatcher -f
+
+# Логи функции (через invocations API)
+curl -H "Authorization: Bearer ${API_TOKEN}" \
+  https://sless.kube5s.ru/v1/namespaces/${NAMESPACE}/functions/iot-telemetry-handler/invocations
+```
+
+## Структура файлов
+
+```
+examples/IOT/
+  main.tf      # Terraform: function + trigger + iot_device
+  handler.py   # Python обработчик телеметрии
+  README.md    # Этот файл
+```
+
+## Известные ограничения MVP
+
+- `sless_iot_device` Terraform ресурс требует реализации в terraform-provider-sless (Этап 6)
+- EMQX TLS отключён — включить для prod (настроить cert-manager secret)
+- iot-mqtt-bridge credentials создаются вручную (автоматизировать в будущем)
+- Нет обратного канала: Cloud → Device команды (Device Shadow — вне MVP)
@@ -0,0 +1,57 @@
+"""
+Создано: 2026-04-04
+handler.py — обработчик IoT-телеметрии для демонстрации IoT MVP.
+
+Вызывается event-dispatcher при каждом MQTT сообщении от устройства.
+Входящий event.body содержит JSON сформированный mqtt-bridge:
+{
+  "namespace": "sless-abc123",
+  "device_id": "temp-sensor-01",
+  "topic": "sless-abc123/telemetry/temp-sensor-01",
+  "payload": {"temperature": 22.5, "humidity": 65},
+  "received_at": "2026-04-04T12:00:00Z"
+}
+"""
+
+import json
+import os
+
+
+def handle(event, context):
+    """Обработчик телеметрии IoT-устройства.
+    
+    Логирует данные и возвращает подтверждение.
+    В реальном сценарии здесь: сохранение в БД, алертинг, управляющие команды.
+    """
+    log_level = os.getenv("LOG_LEVEL", "INFO")
+
+    try:
+        body = json.loads(event.get("body", "{}"))
+    except json.JSONDecodeError as e:
+        return {
+            "statusCode": 400,
+            "body": json.dumps({"error": f"invalid JSON: {e}"})
+        }
+
+    namespace = body.get("namespace", "unknown")
+    device_id = body.get("device_id", "unknown")
+    payload = body.get("payload", {})
+    received_at = body.get("received_at", "")
+
+    if log_level == "INFO":
+        print(f"[IoT] namespace={namespace} device={device_id} at={received_at}")
+        print(f"[IoT] payload={json.dumps(payload)}")
+
+    # Здесь добавить бизнес-логику:
+    # - Запись в PostgreSQL (через POSTGRES_DSN из env)
+    # - Проверка порогов и алертинг
+    # - Публикация управляющей команды обратно на устройство
+
+    return {
+        "statusCode": 200,
+        "body": json.dumps({
+            "processed": True,
+            "device_id": device_id,
+            "namespace": namespace,
+        })
+    }
@@ -0,0 +1,102 @@
+# Создано: 2026-04-04
+# E2E Demo: IoT Device → MQTT → RabbitMQ → Serverless Function
+#
+# Порядок применения:
+#   1. terraform init
+#   2. terraform apply
+#   3. Получить credentials: terraform output mqtt_password
+#   4. Отправить тестовое MQTT сообщение (см. README.md ниже)
+
+terraform {
+  required_providers {
+    sless = {
+      source  = "kube5s.ru/naeel/sless"
+      version = ">= 0.1"
+    }
+  }
+}
+
+# Адрес API sless оператора
+provider "sless" {
+  api_url = "https://sless.kube5s.ru"
+}
+
+# Переменные
+variable "namespace" {
+  description = "Namespace пользователя (создаётся через EnsureNamespace)"
+  type        = string
+}
+
+variable "api_token" {
+  description = "JWT токен для аутентификации в sless API"
+  type        = string
+  sensitive   = true
+}
+
+# Python функция-обработчик IoT-телеметрии
+resource "sless_function" "iot_telemetry_handler" {
+  namespace   = var.namespace
+  name        = "iot-telemetry-handler"
+  runtime     = "python3.11"
+  entrypoint  = "handler.handle"
+  memory_mb   = 128
+  timeout_sec = 30
+
+  env_vars = {
+    LOG_LEVEL = "INFO"
+  }
+}
+
+# Event Trigger: подписка на IoT telemetry queue
+# event-dispatcher читает из этой очереди и вызывает функцию
+resource "sless_trigger" "iot_telemetry_trigger" {
+  namespace    = var.namespace
+  name         = "iot-telemetry-events"
+  type         = "event"
+  function_ref = sless_function.iot_telemetry_handler.name
+  # queue = "iot.{namespace}.telemetry" — формируется mqtt-bridge автоматически
+  queue        = "iot.${var.namespace}.telemetry"
+  enabled      = true
+}
+
+# IoT устройство — температурный датчик
+resource "sless_iot_device" "temperature_sensor" {
+  namespace = var.namespace
+  name      = "temperature-sensor"
+  device_id = "temp-sensor-01"
+  enabled   = true
+
+  metadata = {
+    model    = "DHT22"
+    location = "server-room"
+    owner    = "ops-team"
+  }
+}
+
+# ——— Outputs ———
+
+output "mqtt_broker" {
+  value       = "emqx.sless.svc:1883"
+  description = "MQTT broker адрес (доступен внутри кластера)"
+}
+
+output "mqtt_username" {
+  value       = sless_iot_device.temperature_sensor.mqtt_username
+  description = "MQTT username для устройства"
+}
+
+output "mqtt_password" {
+  value       = sless_iot_device.temperature_sensor.mqtt_password
+  sensitive   = true
+  description = "MQTT пароль для устройства (sensitive)"
+}
+
+output "mqtt_topic" {
+  value       = "${var.namespace}/telemetry/temp-sensor-01"
+  description = "MQTT topic для публикации телеметрии"
+}
+
+output "iot_device_phase" {
+  value       = sless_iot_device.temperature_sensor.phase
+  description = "Статус IoT устройства (Active/Pending/Disabled/Error)"
+}
@@ -0,0 +1,21 @@
+# Terraform provider plugins
+.terraform/
+.terraform.lock.hcl
+
+# Terraform state
+terraform.tfstate
+terraform.tfstate.backup
+*.tfstate
+*.tfstate.backup
+
+# Sensitive data
+terraform.tfvars
+!terraform.tfvars.example
+
+# Backup files
+*.bak
+*.bak_db
+*.bak_*
+
+# Test artifacts
+test_*.log
@@ -1,94 +0,0 @@
-// 2026-04-01 — postgres.tf: Managed PostgreSQL инстанс, пользователь и база данных.
-//
-// Порядок создания:
-//   1. nubes_postgres      — сам инстанс PostgreSQL
-//   2. nubes_postgres_user — пользователь; пароль автоматически попадает в vault_secrets
-//   3. nubes_postgres_database — база данных с owner = созданный пользователь
-//
-// Важно: vault_secrets["users"] появляется только ПОСЛЕ первого apply (нет пользователя — нет ключа).
-//        try() в locals страхует от ошибки на первом прогоне.
-
-// ── Locals: credentials из vault ─────────────────────────────────────────────
-
-locals {
-  # Карта username→{password, username} из vault_secrets, который Nubes заполняет после
-  # создания пользователя. try() нужен для первого apply, когда ключа ещё нет.
-  pg_creds_map = try(
-    jsondecode(lookup(nubes_postgres.pg_test_instance.vault_secrets, "users", "{}")),
-    {}
-  )
-  pg_password = try(local.pg_creds_map[var.pg_username]["password"], "")
-
-  # Адрес master-ноды (внутренний — для подключения из кластера).
-  pg_host = nubes_postgres.pg_test_instance.state_out_flat["internalConnect.master"]
-  pg_port = 5432
-}
-
-// ── Инстанс PostgreSQL ────────────────────────────────────────────────────────
-
-resource "nubes_postgres" "pg_test_instance" {
-  resource_name  = var.pg_resource_name
-  s3_uid         = var.s3_uid
-  resource_realm = var.realm
-
-  # Минимальные ресурсы — достаточно для тестирования.
-  resource_instances = 1
-  resource_memory    = 512 # MiB
-  resource_c_p_u     = 500 # millicores
-  resource_disk      = "1" # GiB
-  app_version        = "17"
-
-  # json_parameters убран — при передаче пустого объекта API возвращает "Invalid JSON String".
-  # Если нужны кастомные параметры PG — добавить после диагностики.
-
-  # Pooler не нужен для тестов — упрощает топологию.
-  enable_pg_pooler_master = false
-  enable_pg_pooler_slave  = false
-
-  allow_no_s_s_l         = false
-  auto_scale             = false
-  auto_scale_percentage  = 10
-  auto_scale_tech_window = 0
-  auto_scale_quota_gb    = "1"
-
-  # Внешний адрес не нужен — подключаемся изнутри кластера.
-  need_external_address_master = false
-
-  operation_timeout = "11m"
-
-  # Позволяет импортировать уже существующий инстанс с тем же именем, не падая
-  # с "already exists" — удобно при повторном apply после ручного создания.
-  adopt_existing_on_create = true
-}
-
-// ── Пользователь ──────────────────────────────────────────────────────────────
-
-resource "nubes_postgres_user" "pg_test_user" {
-  postgres_id = nubes_postgres.pg_test_instance.id
-  username    = var.pg_username
-  role        = var.pg_role
-
-  # Не падать если пользователь с таким именем уже существует.
-  adopt_existing_on_create = true
-}
-
-resource "nubes_postgres_user" "pg_test_user3" {
-  postgres_id = nubes_postgres.pg_test_instance.id
-  username    = "u3"
-  role        = var.pg_role
-
-  depends_on = [nubes_postgres_user.pg_test_user]
-  # Не падать если пользователь с таким именем уже существует.
-  adopt_existing_on_create = true
-}
-
-// ── База данных ───────────────────────────────────────────────────────────────
-
-resource "nubes_postgres_database" "pg_test_db" {
-  postgres_id = nubes_postgres.pg_test_instance.id
-  db_name     = var.pg_db_name
-  db_owner    = nubes_postgres_user.pg_test_user.username
-
-  # Не падать если БД уже существует.
-  adopt_existing_on_create = true
-}
@@ -1,56 +0,0 @@
-// 2026-04-01 — postgres_extra.tf: дополнительные пользователи и базы данных для lifecycle-тестов.
-//
-// ВАЖНО: ресурсы создаются строго последовательно через depends_on.
-// Параллельное создание нескольких пользователей в одном инстансе вызывает
-// ошибки API ("Секрет не был создан", "key doesn't exist") — race condition на стороне Nubes.
-//
-// ИЗВЕСТНОЕ ОГРАНИЧЕНИЕ: если apply упал в середине создания пользователя —
-// этот пользователь может "зависнуть" в промежуточном состоянии в Nubes API.
-// adopt_existing_on_create не спасает. Решение: использовать имена без истории,
-// либо ждать очистки на стороне Nubes.
-
-// ── Пользователь 1 ───────────────────────────────────────────────────────────
-// Первый в extra-цепочке. Ждёт pg_test_db (postgres.tf).
-
-resource "nubes_postgres_user" "test_extra_user1" {
-  postgres_id              = nubes_postgres.pg_test_instance.id
-  username                 = "test_eu1"
-  role                     = "ddl_user"
-  adopt_existing_on_create = true
-
-  # Ждём pg_test_db — иначе параллельный старт с созданием БД ломает API.
-  depends_on = [nubes_postgres_database.pg_test_db]
-}
-
-// ── Пользователь 2 ───────────────────────────────────────────────────────────
-
-resource "nubes_postgres_user" "test_extra_user2" {
-  postgres_id              = nubes_postgres.pg_test_instance.id
-  username                 = "test_eu2"
-  role                     = "ddl_user"
-  adopt_existing_on_create = true
-
-  depends_on = [nubes_postgres_user.test_extra_user1]
-}
-
-// ── База данных 1 (owner = test_eu1) ─────────────────────────────────────────
-
-resource "nubes_postgres_database" "test_extra_db1" {
-  postgres_id              = nubes_postgres.pg_test_instance.id
-  db_name                  = "test_edb1"
-  db_owner                 = nubes_postgres_user.test_extra_user1.username
-  adopt_existing_on_create = true
-
-  depends_on = [nubes_postgres_user.test_extra_user2]
-}
-
-// ── База данных 2 (owner = test_eu2) ─────────────────────────────────────────
-
-resource "nubes_postgres_database" "test_extra_db2" {
-  postgres_id              = nubes_postgres.pg_test_instance.id
-  db_name                  = "test_edb2"
-  db_owner                 = nubes_postgres_user.test_extra_user2.username
-  adopt_existing_on_create = true
-
-  depends_on = [nubes_postgres_database.test_extra_db1]
-}