Naeel
d4f7c5abc8
- architecture/overview.md: добавлена актуальная схема (SQS, managed PG) - deployment.md: добавлена процедура деплоя в iot-naeel - progress.md: добавлен блок 2026-04-12 (Kafka→SQS, деплой) - api/endpoints.md: добавлен актуальный формат /iot-admin/stats - thinking/2026-04-12.md: дописан ход реализации и деплоя - copilot-instructions.md: добавлено правило «документацию не стирать» Старая документация (Kafka, RabbitMQ) сохранена как база знаний.
125 lines
7.5 KiB
Markdown
125 lines
7.5 KiB
Markdown
# 2026-04-12 — GitHub Copilot (Claude Opus 4.6)
|
||
|
||
## Задача
|
||
Пользователь просит:
|
||
1. Просмотреть всё в репе IoT
|
||
2. Оценить замену Kafka и RabbitMQ на собственный SQS-service (из ~/terra/SQS-service)
|
||
3. Контекст: проект переехал из ~/terra/sless в отдельную репу ~/remote_dev/IoT
|
||
|
||
## План
|
||
1. Изучить текущую структуру IoT репы — что есть, как используется Kafka
|
||
2. Посмотреть SQS-service — что он из себя представляет
|
||
3. Посмотреть старую репу sless — что оттуда переехало, что осталось
|
||
4. Дать краткую оценку замены Kafka/RabbitMQ → SQS
|
||
|
||
## Ход работы
|
||
|
||
### Анализ: что нашёл
|
||
|
||
**IoT — текущее использование Kafka:**
|
||
- `cmd/mqtt-bridge/main.go` — MQTT → Kafka writer (топик `iot.telemetry`, библиотека `segmentio/kafka-go`)
|
||
- `cmd/kafka-consumer/main.go` — Kafka reader → Postgres (consumer group `iot-pg-consumer`)
|
||
- `internal/api/handler/iot_admin_stats_handler.go` — admin UI: показывает Kafka consumer lag
|
||
- `cmd/iot-operator/main.go` — передаёт KAFKA_BROKERS в handler
|
||
- `deployments/k8s/kafka.yaml` — деплоймент Kafka в кластере
|
||
- В документации (deployment.md, iot-mvp-plan.md) упоминается RabbitMQ как изначальный план, но в коде **RabbitMQ нет** — в реальности уже Kafka
|
||
|
||
**SQS-service (shared-sqs):**
|
||
- AWS SQS-совместимый сервис на Go (модуль `shared-sqs`)
|
||
- HTTP API с SQS-протоколом: CreateQueue, SendMessage, ReceiveMessage, DeleteMessage и т.д.
|
||
- Multi-tenant: TenantStore с auth
|
||
- Persistence: Redis (write-through) + in-memory
|
||
- Billing через PostgreSQL
|
||
- Порт 4100, gorilla/mux
|
||
- **Не Kafka-совместимый** — это pull-based очередь (SQS-like), не streaming platform
|
||
|
||
**RabbitMQ в IoT:** только в документах (plan). В коде не используется.
|
||
|
||
**shared-SQS (из Gitea README):**
|
||
- AWS SQS-совместимый managed сервис — 17 операций SQS
|
||
- Уже задеплоен: endpoint `https://qu.kube5s.ru`, Web UI `https://qu.kube5s.ru/ui/`
|
||
- Multi-tenant с auth (JWT nubes → auto-create tenant → Access Key / Secret Key)
|
||
- Можно использовать через стандартные AWS SDK (Go, Python boto3, Node.js @aws-sdk/client-sqs)
|
||
- Persistence: Redis + in-memory
|
||
- Billing: PostgreSQL
|
||
- Benchmark: конкурентоспособен с Yandex MQ
|
||
|
||
### Выводы по замене Kafka → shared-SQS
|
||
|
||
**Что реально нужно сделать в IoT:**
|
||
1. `mqtt-bridge`: заменить `kafka.Writer` → AWS SDK SQS `SendMessage` (endpoint = qu.kube5s.ru или внутрикластерный)
|
||
2. `kafka-consumer` → переименовать в `sqs-consumer`: заменить `kafka.Reader` → SQS `ReceiveMessage` + `DeleteMessage` polling loop
|
||
3. `admin stats handler`: вместо Kafka lag → SQS `GetQueueAttributes` (ApproximateNumberOfMessages)
|
||
4. Удалить `deployments/k8s/kafka.yaml`
|
||
5. Убрать `segmentio/kafka-go` из go.mod, добавить `aws-sdk-go-v2`
|
||
6. Обновить env vars: KAFKA_BROKERS → SQS_ENDPOINT + SQS credentials
|
||
7. Обновить документацию
|
||
|
||
**Плюс:** AWS SDK for Go — стандартная библиотека, код станет проще. shared-SQS уже живой.
|
||
**Минус:** polling latency (ReceiveMessage WaitTimeSeconds до 20s) vs Kafka push. Для IoT телеметрии — приемлемо.
|
||
|
||
---
|
||
|
||
## Продолжение — GitHub Copilot (Claude Opus 4.6)
|
||
|
||
### Реализация замены Kafka → SQS
|
||
|
||
**Что сделано:**
|
||
|
||
1. **mqtt-bridge** — полностью переписан:
|
||
- Убран `segmentio/kafka-go`, добавлен `aws-sdk-go-v2` (sqs, config, credentials)
|
||
- `kafka.Writer` → `sqs.Client.SendMessage`
|
||
- При старте: `GetQueueUrl` для резолва URL очереди "iot-telemetry"
|
||
- Env vars: `SQS_ENDPOINT`, `SQS_ACCESS_KEY`, `SQS_SECRET_KEY`, `SQS_QUEUE_NAME`, `SQS_REGION`
|
||
- Формат сообщения (MessageBody JSON) не изменился: `{namespace, device_id, topic, payload, received_at}`
|
||
|
||
2. **sqs-consumer** — создан с нуля (заменяет kafka-consumer):
|
||
- Long polling: `ReceiveMessage(WaitTimeSeconds=20)` — минимизирует запросы при пустой очереди
|
||
- At-least-once: `DeleteMessage` только после успешной записи в Postgres
|
||
- Использует `iotpg.IoTPostgresStore` — тот же механизм per-tenant DB что и kafka-consumer
|
||
|
||
3. **admin stats handler** — переписан:
|
||
- Вместо Kafka consumer lag → `GetQueueAttributes(ApproximateNumberOfMessages, ApproximateNumberOfMessagesNotVisible)`
|
||
- Pod labels для consumer: `iot-kafka-consumer` → `iot-sqs-consumer`
|
||
|
||
4. **Баг .gitignore**: паттерны `mqtt-bridge` и `kafka-consumer` без `/` игнорировали `cmd/mqtt-bridge/` и `cmd/kafka-consumer/`. Исправлено добавлением `/` префикса.
|
||
|
||
5. **Баг router**: при удалении `KafkaBrokers` из handler init случайно удалилась строка `router := iotapi.NewRouter(h, log)`. Восстановлена.
|
||
|
||
### Деплой в новый кластер iot-naeel
|
||
|
||
**Обнаружения при деплое:**
|
||
|
||
1. Кластер **полностью новый** — namespace `sless` не существовал, ничего не задеплоено.
|
||
|
||
2. **Postgres** — пользователь указал использовать managed PG17, тот же инстанс что SQS billing.
|
||
Credentials в `/home/naeel/terra/SQS-service/secrets/iot_pg.md`.
|
||
Самодеплоенный postgres:16-alpine из YAML заменён на DSN к managed PG.
|
||
|
||
3. **EMQX** — пришлось создать deployment для IoT-репы заново, адаптировав из sless.
|
||
Ключевое изменение: auth URL `sless-operator.sless.svc:9090` → `iot-operator.sless.svc:9090`.
|
||
|
||
4. **iot-operator deployment** — его не было в IoT-репе! Создан новый:
|
||
- ServiceAccount + ClusterRole (iotdevices CRD, secrets, events, namespaces, leases)
|
||
- ClusterRoleBinding
|
||
- Deployment + Service :9090
|
||
|
||
5. **kubectl токен** истекал за 24 часа — пользователь обновлял вручную.
|
||
|
||
6. **Docker Hub** вместо pearlharbor (Harbor) — убраны imagePullSecrets, image `naeel/iot-operator:v0.2.0`.
|
||
|
||
7. **shared-SQS tenant** создан через API:
|
||
- Tenant: `iot-service`, ID: `t-96afe7e9f781f6ca`
|
||
- Queue: `iot-telemetry`
|
||
- Admin token из Secret `shared-sqs-admin` в namespace `shared-sqs`
|
||
|
||
### Результат
|
||
|
||
Все 4 пода Running 1/1:
|
||
- `iot-operator` — controller работает, MQTT auth/acl обрабатывает запросы
|
||
- `emqx` — MQTT брокер, подключает IoT устройства
|
||
- `iot-mqtt-bridge` — подписан на EMQX, SQS queue resolved
|
||
- `iot-sqs-consumer` — подключён к PG и SQS, polling loop активен
|
||
|
||
TLS сертификат для `iot.kube5s.ru` выпускается cert-manager.
|