From 110d49effe830d0296840ee541b88d65ee215b87 Mon Sep 17 00:00:00 2001 From: Repinoid Date: Mon, 5 Oct 2026 20:04:20 +0300 Subject: [PATCH] =?UTF-8?q?docs(thinking):=20IoT=E2=86=94SQS=20handoff=20?= =?UTF-8?q?=E2=80=94=20=D1=81=D0=B2=D0=BE=D0=B4=D0=BA=D0=B0=20=D0=BD=D0=B0?= =?UTF-8?q?=D1=85=D0=BE=D0=B4=D0=BE=D0=BA=20=D0=B8=20=D0=BF=D0=BE=D1=80?= =?UTF-8?q?=D1=8F=D0=B4=D0=BE=D0=BA=20=D1=80=D0=B0=D0=B1=D0=BE=D1=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- doc/thinking/iot-sqs-handoff-2026-10-05.md | 230 +++++++++++++++++++++ 1 file changed, 230 insertions(+) create mode 100644 doc/thinking/iot-sqs-handoff-2026-10-05.md diff --git a/doc/thinking/iot-sqs-handoff-2026-10-05.md b/doc/thinking/iot-sqs-handoff-2026-10-05.md new file mode 100644 index 0000000..dbd02d3 --- /dev/null +++ b/doc/thinking/iot-sqs-handoff-2026-10-05.md @@ -0,0 +1,230 @@ +# 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-бинарник с тремя ролями в одном процессе: +1. **REST API** + MQTT auth/acl на порту `:9090` (устройства хранятся в PostgreSQL); +2. **bridge** — MQTT-подписка на EMQX → отправка в shared-SQS; +3. **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"` захардкожен.** + +### ПЛАТФОРМА (вне нашего кода, но блокирует прод) + +1. **wss-обрывы ~150с** на внешнем шлюзе Nubes (детерминированно, доказано с + 3 независимых точек: станция, Vultr; внутри кластера и с ВМ — чисто). + Тикет Nubes подготовлен (`doc/thinking/nubes-devops-report-2026-08-16.md`). +2. **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. Рекомендуемый порядок работ (по плану) + +1. P0-1 + P0-2 — tenant-граница (связать `sub` с namespace) + обязательная подпись JWT. +2. P0-7 + P0-8 — доставка без потерь (ack только после успешного flush) + dedup (message-id). +3. P0-5 — выключить `AUTH_TEST_MODE`. +4. P0-3 + P1-4 — пароли (не отдавать повторно/хешировать) + удалить `/internal/mqtt/*`. +5. P1-9 + P1-10 — честный `/health` + метрики дропов. +6. P1-11 + P1-12 — Go-тесты + CI. +7. P2 (гигиена: легаси, доки, cookie, TLS).