From 18e4352a479d261deac985e950861f72fb95aa6d Mon Sep 17 00:00:00 2001 From: Repinoid Date: Wed, 7 Oct 2026 07:50:29 +0300 Subject: [PATCH] =?UTF-8?q?chore(P2-14/=D0=BF.4):=20=D1=83=D0=B4=D0=B0?= =?UTF-8?q?=D0=BB=D1=91=D0=BD=20legacy=20cmd/sqs-consumer?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Не собирался Dockerfile/Makefile (только ./cmd/iot-service/) → в образ/деплой не попадал; логика поглощена internal/service/consumer/. Был источником пустого message_id (обход дедупа). message_id оставлен nullable — единственный пишущий путь (bridge) всегда ставит UUID. LEGACY.md обновлён. Плюс исправлена неточность в HISTORY про поведение без DLQ. --- HISTORY/2026-10-07-session-log.md | 18 ++- LEGACY.md | 8 +- cmd/sqs-consumer/main.go | 211 ------------------------------ 3 files changed, 23 insertions(+), 214 deletions(-) delete mode 100644 cmd/sqs-consumer/main.go diff --git a/HISTORY/2026-10-07-session-log.md b/HISTORY/2026-10-07-session-log.md index e435404..c25304d 100644 --- a/HISTORY/2026-10-07-session-log.md +++ b/HISTORY/2026-10-07-session-log.md @@ -148,5 +148,21 @@ удалить legacy (P2-14). Минимальный риск, но формально «не NOT NULL». - Нужно решение владельца. +### Решено (п.4) — удалён legacy `cmd/sqs-consumer` +- Проверено: `Dockerfile`/`Makefile` собирают ТОЛЬКО `./cmd/iot-service/` → + `cmd/sqs-consumer` не попадает в образ/деплой (пустой `message_id` до прода + не доходил). Логика поглощена `internal/service/consumer/`. +- Удалён (`git rm -r cmd/sqs-consumer`); сборка/vet зелёные. `message_id` + оставлен nullable — единственный пишущий путь (bridge) всегда ставит UUID. +- LEGACY.md обновлён: `cmd/sqs-consumer` перенесён в «удалено». +- Примечание-противоречие: LEGACY.md до этого называл `cmd/sqs-consumer` «НЕ + легаси, переиспользуется»; при сверке с Dockerfile/Makefile подтверждено, что + standalone-бинарник не используется → удаление корректно. + +### Исправление неточности (моей) +- В коммите п.3 и в этой HISTORY я написал «без DLQ — НЕ подтверждаем». По коду + НЕ так: даже без DLQ битый envelope ПОДТВЕРЖДАЕТСЯ (удаляется с логом) — и это + правильно (иначе он зациклился бы). Формулировка была неверной, исправлено. + ## Проверки -- `go build ./...` и `go vet ./internal/...` — зелёные на каждом шаге. +- `go build ./...` и `go vet ./...` — зелёные на каждом шаге. diff --git a/LEGACY.md b/LEGACY.md index 679e16c..14dc83d 100644 --- a/LEGACY.md +++ b/LEGACY.md @@ -43,5 +43,9 @@ - `HISTORY/2026-08-16-session-log.md` - `doc/sqs-integration.md` - `ws-probe/` -- `cmd/mqtt-bridge/`, `cmd/sqs-consumer/`, `internal/` — Go-код, переиспользуется - в новом монолите (после адаптации под платформу) +- `cmd/mqtt-bridge/`, `internal/` — Go-код, переиспользуется в новом монолите + (после адаптации под платформу) +- ⛔ `cmd/sqs-consumer/` — **УДАЛЁН 2026-10-07**: standalone-консьюмер, который + НЕ собирался Dockerfile/Makefile (те билдят только `./cmd/iot-service/`); его + логика поглощена `internal/service/consumer/`. Был источником пустого + `message_id` (обход дедупа). diff --git a/cmd/sqs-consumer/main.go b/cmd/sqs-consumer/main.go deleted file mode 100644 index ad41f41..0000000 --- a/cmd/sqs-consumer/main.go +++ /dev/null @@ -1,211 +0,0 @@ -// Создано: 2026-04-12 (замена kafka-consumer → sqs-consumer) -// sqs-consumer/main.go — iot-sqs-consumer: читает IoT телеметрию из shared-SQS → пишет в Postgres. -// -// Роль в архитектуре: -// SQS очередь "iot-telemetry" → iot-sqs-consumer → IoT Postgres (per-tenant DB) -// -// Long polling (WaitTimeSeconds=20) — минимизирует пустые запросы к SQS. -// При временной недоступности Postgres — сообщения остаются в очереди (visibility timeout). -// DeleteMessage вызывается только после успешной записи в Postgres. -// -// Конфигурация через env vars: -// SQS_ENDPOINT — https://qu.kube5s.ru (или внутрикластерный endpoint) -// SQS_ACCESS_KEY — Access Key для shared-SQS tenant -// SQS_SECRET_KEY — Secret Key для shared-SQS tenant -// SQS_QUEUE_NAME — имя очереди (default: iot-telemetry) -// SQS_REGION — регион (default: us-east-1) -// IOT_PG_DSN — postgres://user:pass@host:5432/iotdb (master DSN для IoT Postgres) - -package main - -import ( - "context" - "encoding/json" - "log/slog" - "os" - "os/signal" - "syscall" - "time" - - "github.com/aws/aws-sdk-go-v2/aws" - "github.com/aws/aws-sdk-go-v2/credentials" - "github.com/aws/aws-sdk-go-v2/service/sqs" - - "gitea.services.ngcloud.ru/Nail/IoT/internal/storage/iotpg" -) - -// sqsConsumerConfig — конфигурация из env vars. -type sqsConsumerConfig struct { - SQSEndpoint string - SQSAccessKey string - SQSSecretKey string - SQSQueueName string - SQSRegion string -} - -// iotTelemetryMessage — envelope из SQS (идентичен bridge). -type iotTelemetryMessage struct { - Namespace string `json:"namespace"` - DeviceID string `json:"device_id"` - Topic string `json:"topic"` - Payload json.RawMessage `json:"payload"` - ReceivedAt string `json:"received_at"` -} - -func main() { - log := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo})) - - cfg := loadConsumerConfig() - - ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT) - defer cancel() - - log.Info("starting iot-sqs-consumer", - "sqs_endpoint", cfg.SQSEndpoint, - "sqs_queue", cfg.SQSQueueName, - ) - - // IoT Postgres — обязательный компонент для этого сервиса - iotStore, err := iotpg.NewFromEnv(log) - if err != nil || iotStore == nil { - log.Error("failed to connect to IoT Postgres — IOT_PG_DSN required", "err", err) - os.Exit(1) - } - defer iotStore.Close() - log.Info("connected to IoT Postgres") - - // SQS клиент - sqsClient := sqs.New(sqs.Options{ - Region: cfg.SQSRegion, - Credentials: credentials.NewStaticCredentialsProvider( - cfg.SQSAccessKey, cfg.SQSSecretKey, "", - ), - BaseEndpoint: aws.String(cfg.SQSEndpoint), - }) - - // Получаем QueueUrl по имени один раз при старте - queueUrlOut, err := sqsClient.GetQueueUrl(ctx, &sqs.GetQueueUrlInput{ - QueueName: aws.String(cfg.SQSQueueName), - }) - if err != nil { - log.Error("failed to get SQS queue URL — очередь должна существовать", "queue", cfg.SQSQueueName, "err", err) - os.Exit(1) - } - queueURL := *queueUrlOut.QueueUrl - log.Info("SQS queue resolved", "queue_url", queueURL) - - log.Info("sqs consumer ready, starting polling loop...") - - // Polling loop — long polling (WaitTimeSeconds=20) минимизирует пустые запросы. - // При ошибке SQS — backoff 5 сек и продолжаем. - for { - if ctx.Err() != nil { - break - } - - resp, err := sqsClient.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{ - QueueUrl: aws.String(queueURL), - MaxNumberOfMessages: 10, - WaitTimeSeconds: 20, // long polling — SQS держит соединение до 20 сек - }) - if err != nil { - if ctx.Err() != nil { - break // штатное завершение - } - log.Error("SQS ReceiveMessage failed", "err", err) - // Backoff при ошибках SQS — не спамим запросами - select { - case <-ctx.Done(): - case <-time.After(5 * time.Second): - } - continue - } - - for _, sqsMsg := range resp.Messages { - if sqsMsg.Body == nil { - continue - } - - if err := processSQSTelemetry(ctx, *sqsMsg.Body, iotStore, log); err != nil { - log.Error("process telemetry message", "err", err, "message_id", derefStr(sqsMsg.MessageId)) - // НЕ удаляем сообщение — оно вернётся в очередь после visibility timeout - continue - } - - // Удаляем сообщение из SQS только после успешной записи в Postgres - _, err := sqsClient.DeleteMessage(ctx, &sqs.DeleteMessageInput{ - QueueUrl: aws.String(queueURL), - ReceiptHandle: sqsMsg.ReceiptHandle, - }) - if err != nil { - log.Error("SQS DeleteMessage failed", "err", err, "message_id", derefStr(sqsMsg.MessageId)) - } - } - } - - log.Info("shutting down iot-sqs-consumer") -} - -// processSQSTelemetry десериализует envelope из SQS и записывает в Postgres. -func processSQSTelemetry(ctx context.Context, body string, store *iotpg.IoTPostgresStore, log *slog.Logger) error { - var envelope iotTelemetryMessage - if err := json.Unmarshal([]byte(body), &envelope); err != nil { - // Битое сообщение — логируем и пропускаем (не блокируем очередь) - log.Warn("failed to unmarshal telemetry envelope, skipping", "err", err, "raw", body) - return nil - } - - // EnsureTenantDB идемпотентен — кэшируется после первого вызова - if err := store.EnsureTenantDB(ctx, envelope.Namespace); err != nil { - return err - } - - receivedAt, perr := time.Parse(time.RFC3339, envelope.ReceivedAt) - if perr != nil { - receivedAt = time.Time{} - } - if err := store.InsertTelemetry(ctx, envelope.Namespace, envelope.DeviceID, "", receivedAt, envelope.Payload); err != nil { - return err - } - - log.Info("telemetry saved to Postgres", - "namespace", envelope.Namespace, - "device", envelope.DeviceID, - ) - return nil -} - -// loadConsumerConfig читает конфигурацию из env vars. -func loadConsumerConfig() sqsConsumerConfig { - 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 sqsConsumerConfig{ - SQSEndpoint: required("SQS_ENDPOINT"), - SQSAccessKey: required("SQS_ACCESS_KEY"), - SQSSecretKey: required("SQS_SECRET_KEY"), - SQSQueueName: getEnvOrDefault("SQS_QUEUE_NAME", "iot-telemetry"), - SQSRegion: getEnvOrDefault("SQS_REGION", "us-east-1"), - } -} - -func getEnvOrDefault(key, defaultVal string) string { - if v := os.Getenv(key); v != "" { - return v - } - return defaultVal -} - -// derefStr — безопасная разыменовка строкового указателя. -func derefStr(s *string) string { - if s == nil { - return "" - } - return *s -}