Nail/IoT

Files

T

Repinoid 110d49effe docs(thinking): IoT↔SQS handoff — сводка находок и порядок работ

2026-10-05 20:04:20 +03:00

15 KiB

Raw Blame History

IoT ↔ SQS: сводка всех находок и мыслей (handoff для нового чата/репы)

Дата: 2026-10-05. Назначение: полностью самодостаточный документ — чтобы, открыв его в НОВОЙ репе или новом чате, можно было продолжить без чтения истории. Вся информация ниже взята из реального кода/доков репозитория IoT (/home/naeel/nubes/IoT), а НЕ из памяти и НЕ из чужих заметок.

0. КЛЮЧЕВОЙ РАЗВОРОТ: что в какой репе

Есть ДВА разных проекта, в ДВУХ разных репах:

Проект	Репа	Что делает	Состояние
shared-sqs	отдельная репа (SQS-service)	SQS-совместимая очередь (AWS API, Go, форк GoAWS), multi-tenant	ГОТОВ, работает на проде v0.1.35 (все тесты зелёные, нагрузка без ошибок)
IoT	эта репа (`/home/naeel/nubes/IoT`)	телеметрия устройств: MQTT → SQS → PostgreSQL → REST	MVP, не production-ready (есть подтверждённые блокеры)

Зависимость: IoT потребляет shared-sqs как внешний сервис (очередь телеметрии iot-telemetry). shared-sqs от IoT не зависит.

Вывод о порядке (отвечая на вопрос «что сначала — SQS или IoT»):

shared-sqs фактически уже готов и работает раньше, чем IoT дотянет до production. Его «доделывать» не надо для старта работ по IoT.
Порядок работ по сути такой: SQS уже есть (работает) → теперь доработки IoT, который на него опирается.
Если планируется создавать оба «с нуля» — логический порядок архитектурно: сначала SQS, потом IoT (т.к. IoT от него зависит).

ВНИМАНИЕ про ветки IoT-репы (важно, легко запутаться):

Актуальный код IoT (монолит iot-service, internal/service/, internal/storage/iotpg/) лежит на ветке feature/replace-kafka-with-sqs. Имя ветки — ПЕРЕЖИТОК старой миграции Kafka→SQS (апрель 2026), к текущей задаче отношения не имеет. Не путать имя с сутью: это текущая ветка IoT.
Ветка master — СТАРАЯ, без монолита (там легаси: internal/api/, CRD, k8s-манифесты, deployments/, examples/, config/crd/). НЕ актуальна.
На master ошибочно закоммичены 3 файла документации (см. раздел 5).

1. Что такое текущий IoT (архитектура)

Монолит iot-service — один Go-бинарник с тремя ролями в одном процессе:

REST API + MQTT auth/acl на порту :9090 (устройства хранятся в PostgreSQL);
bridge — MQTT-подписка на EMQX → отправка в shared-SQS;
consumer — чтение из shared-SQS (long-poll) → запись в per-tenant PostgreSQL.

Конвейер телеметрии:

устройство → wss → EMQX (authn+authz напрямую в PG) → бридж → shared-SQS
  → consumer → per-tenant PostgreSQL → REST API (/v1/namespaces/{ns}/iot/telemetry)

Стек:

Go 1.25, gorilla/mux, paho.mqtt.golang, aws-sdk-go-v2 (SQS), lib/pq.
EMQX 5.5.1 (образ naeel/iot-emqx), shared-SQS (SQS-совместимая очередь), PostgreSQL managed (per-tenant БД).
Модуль: gitea.services.ngcloud.ru/Nail/IoT.

Ключевые файлы актуального кода (на ветке feature):

cmd/iot-service/main.go — точка входа: config → PG → SQS → API + горутины bridge/consumer.
internal/service/config/config.go, defaults.go — env-конфигурация.
internal/service/api/router.go, middleware/auth.go — маршруты + JWT.
internal/service/api/handler/*.go — devices, mqtt_auth, telemetry, admin, sqsstats.
internal/service/store/*.go — устройства в PG (iot_devices).
internal/storage/iotpg/iot_telemetry_store.go — запись телеметрии (batch).
internal/service/bridge/*.go — bridge → SQS.
internal/service/consumer/*.go — SQS → PG.
internal/service/sqsclient/sqsclient.go — SQS-клиент (endpoint override).
emqx/emqx.conf — конфиг EMQX (auth/ACL — напрямую в PostgreSQL).

2. Как IoT использует shared-sqs (факты)

Внутренний endpoint shared-sqs (тот же кластер Nubes, realm iot-naeel): http://containerk8s.f1ffb134-7d16-45bd-8bef-69f6ec8ab33c.svc.cluster.local:4100 (вшит дефолтом в config/defaults.go).
Очередь: iot-telemetry, регион us-east-1.
Тенант IoT: t-fec713a719ad0b33; креды детерминированы из email: tenantID t-+sha256(email)[:8], AccessKey SSAK-+sha256(email)[:12], SecretKey sha256(email+":shared-sqs:secret-key:v1").
Лимиты shared-sqs: 50 очередей на тенанта, сообщение до 256 КБ, Receive 1–10.
В бридже: неблокирующий SQS-диспетчер (канал 10000 + 4 worker'а, ретраи 1/2/4с), дроп при переполнении буфера и при oversize >250 КБ.
В consumer: long-poll 20с, VisibilityTimeout 120с (в коде), backoff при сбое PG (1с→60с), DeleteMessage после обработки.

3. ПОДТВЕРЖДЁННЫЕ БЛОКЕРЫ IoT (для доработок)

Источник: два независимых анализа + ревью Opus + моя перепроверка построчно. Все — по КОДУ, с точными ссылками. Сгруппировано по приоритету.

P0 — компрометация/потеря данных или доступ к чужим данным

P0-1. Tenant-изоляция не работает.

internal/service/api/middleware/auth.go (validateJWT, ~стр. 62, 120): проверяет только наличие sub и (необязательно) exp, а sub НИКАК не сверяется с {namespace} из URL.
Обработчики берут namespace из пути (handler/devices.go:91, handler/telemetry.go:19) без сверки с токеном.
Итог: любой валидный токен читает/пишет ЛЮБОЙ namespace — горизонтальный доступ к чужим устройствам и телеметрии.

P0-2. Проверка подписи JWT необязательна.

auth.go (~стр. 70): при пустом JWT_HMAC_SECRET подпись НЕ проверяется — только структура токена. Токен подделывается вручную ({"sub":"x","exp":...}).
config/config.go (~стр. 118): дефолт секрета пустой, секрет не в списке required env.

P0-3. Пароли устройств в открытом виде.

store/open.go:21: mqtt_password TEXT — хранится открытым текстом.
handler/devices.go:189: пароль отдаётся повторно через GET /devices/{name}.
emqx/emqx.conf:46: password_hash_algorithm name=plain.
Компрометация БД = все пароли устройств + возможность выдать их повторно.

P0-7. Потеря телеметрии: SQS-удаление до фактической записи.

consumer/consumer.go:78: удаляет сообщение из SQS после processTelemetry==nil.
iotpg/iot_telemetry_store.go:356-376: InsertTelemetry при неполном батче кладёт строку В ПАМЯТЬ и возвращает nil (запись ещё не сделана).
iotpg/iot_telemetry_store.go:129-135 (flushAll): подменяет буфер на пустой ДО записи в PG, при ошибке — только логирует.
Итог: краш между «кладём в буфер» и «flush» → сообщение уже удалено из SQS, в PG не попало → безвозвратная потеря.

P0-8. Нет идемпотентности (дубли).

iot_telemetry таблица: id BIGSERIAL, без ключа дедупликации.
bridge/handler.go:19: envelope без message-id.
At-least-once SQS → молчаливые дубли при повторе сообщения.

P1 — важные риски эксплуатации

P1-4. Лишние HTTP-auth/ACL-эндпоинты.

router.go:50-51: /internal/mqtt/auth и /internal/mqtt/acl висят на публичном порту 9090 БЕЗ аутентификации.
Но EMQX ходит по PG-auth напрямую (emqx/emqx.conf, backend=postgresql), эти эндпоинты — НЕиспользуемый fallback. Рекомендация: удалить, а не защищать.
Риск: кто достучался до порта — перебирает пароли устройств и получает ACL.

P1-5. AUTH_TEST_MODE=true на проде.

Был включён для диагностики DevOps (HISTORY §38); любой Bearer-токен проходит.
Нужно выставить false + задать JWT_HMAC_SECRET.

P1-9. /health не отражает состояние.

router.go:72: healthHandler всегда {"status":"ok"}, не проверяет bridge/consumer/PG/SQS.
main.go: ошибки bridge.Run/consumer.Run только логируются; горутина завершается, а liveness остаётся зелёным.

P1-10. Дропы в бридже без метрик.

bridge/sender.go:73: дроп при переполнении RAM-очереди (10000).
bridge/sender.go:112: дроп после исчерпания ретраев.

P2 — обвязка/гигиена

P2-6. TLS. sslmode=require (без verify-full) к PG; EMQX verify=verify_none (emqx.conf:40).

P2-11/P2-12. Нет тестов и CI. *_test.go отсутствует; .github/workflows нет (в .github/ только правила).

P2-13. /metrics заявлен, но не зарегистрирован.

config/defaults.go:6: в комментарии упомянут /metrics, но роут в router.go НЕ создан. prometheus — только indirect dep. Rate-limiting нет.

P2-14. Легаси не вычищено.

Остались cmd/iot-operator, controllers/, internal/api/ (старый k8s-код), cmd/mqtt-bridge, cmd/sqs-consumer; k8s/controller-runtime в go.mod (~12 строк); бинарники в bin/, ws-probe/ws-probe.

P2-15. Дрейф доков.

doc/deployment-nubes-production.md — образы v0.1.2; Makefile — v0.1.13.
loadtests/README.md — visibility «30с» (в коде 120с), описывает уже исправленную блокировку MQTT-обработчика.

P2-16. emqx.conf: cookie = "iot-emqx-cookie-mvp" захардкожен.

ПЛАТФОРМА (вне нашего кода, но блокирует прод)

wss-обрывы ~150с на внешнем шлюзе Nubes (детерминированно, доказано с 3 независимых точек: станция, Vultr; внутри кластера и с ВМ — чисто). Тикет Nubes подготовлен (doc/thinking/nubes-devops-report-2026-08-16.md).
MSS/MTU-баг — плавающий обрыв ответов >~15–64 КБ (баг нестабилен по маршруту/времени). Обход в коде: offset-пагинация API, внутренний endpoint SQS.

4. Что уже ХОРОШО (не трогать без нужды)

Конвейер end-to-end работает и проверен на проде (HISTORY: baseline 2390/2390 без потерь/дублей, ACL-нарушения отсекаются, reconnect-storm чистый).
Уже исправлены (по итогам ревью 2026-08-16): асинхронный SQS-диспатчер, VisibilityTimeout 120с, backoff, graceful shutdown, batch-INSERT) — НЕ регрессировать это.
EMQX auth/ACL — напрямую в PostgreSQL, без зависимости от монолита.
MQTT-reconnect с повторной подпиской (OnConnectHandler).
Документация: деплой без kubectl, интеграция с SQS, 10 нагрузочных сценариев.

5. Что уже записано в этой репе (файлы документации)

doc/thinking/readiness-assessment-2026-10-05.md — два анализа (5/10 и 60–65%) + моё мнение.
doc/thinking/remediation-plan-2026-10-05.md — план доработок (4 трека, P0/P1/P2).
doc/thinking/opus-review-prompt.md — промпт для ревью (allowlist файлов).
ВНИМАНИЕ: эти 3 файла лежат в ветке master (коммит a4e8fd5, запушен), а НЕ на актуальной feature/replace-kafka-with-sqs, где живёт код. doc/sqs-integration.md, doc/deployment-nubes-production.md, HISTORY/ — на feature-ветке.

Backup-ветки (сохранение состояний):

backup/2026-08-17-feature-sqs (старая точка).
backup/2026-10-05-feature-sqs (сделана 2026-10-05, на 12db7d3).

6. Рекомендуемый порядок работ (по плану)

P0-1 + P0-2 — tenant-граница (связать sub с namespace) + обязательная подпись JWT.
P0-7 + P0-8 — доставка без потерь (ack только после успешного flush) + dedup (message-id).
P0-5 — выключить AUTH_TEST_MODE.
P0-3 + P1-4 — пароли (не отдавать повторно/хешировать) + удалить /internal/mqtt/*.
P1-9 + P1-10 — честный /health + метрики дропов.
P1-11 + P1-12 — Go-тесты + CI.
P2 (гигиена: легаси, доки, cookie, TLS).

15 KiB Raw Blame History Unescape Escape