209 lines
8.0 KiB
Go
209 lines
8.0 KiB
Go
// iot-service — монолит IoT на платформе Nubes (контейнер «Простой HTTP»).
|
||
//
|
||
// Три роли в одном процессе:
|
||
// - REST API + MQTT auth/acl (:9090, устройства в PostgreSQL);
|
||
// - bridge — MQTT-подписка EMQX → shared-SQS;
|
||
// - consumer — shared-SQS → per-tenant PostgreSQL.
|
||
//
|
||
// Наследие старого IoT (k8s, оператор, CRD) НЕ используется.
|
||
package main
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"log/slog"
|
||
"net/http"
|
||
"os"
|
||
"os/signal"
|
||
"sync"
|
||
"sync/atomic"
|
||
"syscall"
|
||
"time"
|
||
|
||
"gitea.services.ngcloud.ru/Nail/IoT/internal/service/api"
|
||
"gitea.services.ngcloud.ru/Nail/IoT/internal/service/api/handler"
|
||
"gitea.services.ngcloud.ru/Nail/IoT/internal/service/bridge"
|
||
"gitea.services.ngcloud.ru/Nail/IoT/internal/service/config"
|
||
"gitea.services.ngcloud.ru/Nail/IoT/internal/service/consumer"
|
||
"gitea.services.ngcloud.ru/Nail/IoT/internal/service/sqsclient"
|
||
"gitea.services.ngcloud.ru/Nail/IoT/internal/service/store"
|
||
"gitea.services.ngcloud.ru/Nail/IoT/internal/storage/iotpg"
|
||
)
|
||
|
||
// shutdownGrace — максимум ожидания остановки подсистем при завершении. Если
|
||
// что-то зависло — не блокируем выход (ревью 2026-10-10, «ограничить wg.Wait»).
|
||
const shutdownGrace = 15 * time.Second
|
||
|
||
// waitWithTimeout ждёт wg не дольше d; false = не дождались.
|
||
func waitWithTimeout(wg *sync.WaitGroup, d time.Duration) bool {
|
||
done := make(chan struct{})
|
||
go func() { wg.Wait(); close(done) }()
|
||
select {
|
||
case <-done:
|
||
return true
|
||
case <-time.After(d):
|
||
return false
|
||
}
|
||
}
|
||
|
||
// version — подставляется через ldflags при сборке.
|
||
var version = "dev"
|
||
|
||
// main — тонкая обёртка: os.Exit обходит defer-ы, поэтому вся логика в run(),
|
||
// которая возвращает код выхода и выполняет свои defer-ы (devices/iotStore
|
||
// Close, финальный flush) ДО выхода (ревью 🔴5: «os.Exit обходит defer-ы»).
|
||
func main() {
|
||
os.Exit(run())
|
||
}
|
||
|
||
func run() int {
|
||
cfg, err := config.Load()
|
||
if err != nil {
|
||
slog.Error("config", "err", err)
|
||
return 1
|
||
}
|
||
|
||
log := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: cfg.LogLevel}))
|
||
slog.SetDefault(log)
|
||
|
||
log.Info("iot-service starting", "version", version, "port", cfg.APIPort)
|
||
|
||
ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
|
||
defer cancel()
|
||
|
||
// PostgreSQL — устройства и телеметрия.
|
||
storeCtx, storeCancel := context.WithTimeout(ctx, 15*time.Second)
|
||
devices, err := store.Open(storeCtx, cfg.IOTPGDSN)
|
||
storeCancel()
|
||
if err != nil {
|
||
log.Error("devices store", "err", err)
|
||
return 1
|
||
}
|
||
defer devices.Close()
|
||
log.Info("devices store ready")
|
||
|
||
// Служебная строка бриджа в iot_devices — для PG-auth EMQX (хвост 3).
|
||
ensureCtx, ensureCancel := context.WithTimeout(ctx, 10*time.Second)
|
||
err = devices.EnsureBridgeDevice(ensureCtx, cfg.MQTTUsername, cfg.MQTTPassword)
|
||
ensureCancel()
|
||
if err != nil {
|
||
log.Error("ensure bridge device", "err", err)
|
||
return 1
|
||
}
|
||
log.Info("bridge device ensured", "username", cfg.MQTTUsername)
|
||
|
||
iotStore, err := iotpg.New(cfg.IOTPGDSN, log)
|
||
if err != nil {
|
||
log.Error("iotpg store", "err", err)
|
||
return 1
|
||
}
|
||
defer iotStore.Close()
|
||
log.Info("telemetry store ready")
|
||
|
||
sqsClient := sqsclient.New(cfg)
|
||
|
||
h := &handler.Handler{
|
||
Devices: devices,
|
||
IoTPG: iotStore,
|
||
SQS: sqsClient,
|
||
Cfg: cfg,
|
||
Log: log,
|
||
Version: version,
|
||
StartedAt: time.Now(),
|
||
}
|
||
|
||
// HTTP API — в основном потоке.
|
||
// Readiness: /ready отдаёт 503, пока bridge/consumer не подняты.
|
||
ready := &api.Readiness{}
|
||
router := api.NewRouter(h, log, cfg.AuthTestMode, cfg.JwtHMACSecret, version, ready)
|
||
server := &http.Server{
|
||
Addr: ":" + cfg.APIPort,
|
||
Handler: router,
|
||
ReadHeaderTimeout: 10 * time.Second,
|
||
ReadTimeout: 15 * time.Second,
|
||
// WriteTimeout с запасом: admin-stats/telemetry отвечают через SQS/PG
|
||
// (внутренние таймауты до 15с). IdleTimeout — для keep-alive платформы.
|
||
WriteTimeout: 60 * time.Second,
|
||
IdleTimeout: 120 * time.Second,
|
||
}
|
||
|
||
// bridge + consumer — в фоне.
|
||
var wg sync.WaitGroup
|
||
wg.Add(2)
|
||
|
||
// shutdownDone — закрывается после завершения server.Shutdown (дренаж
|
||
// in-flight HTTP-запросов). run() ждёт его перед выходом, чтобы deferred
|
||
// Close (iotStore/devices) не сработали, пока запросы ещё дренируются.
|
||
shutdownDone := make(chan struct{})
|
||
|
||
// fatal — подсистема (bridge/consumer) завершилась с ПОСТОЯННОЙ ошибкой.
|
||
// Тогда после штатной остановки выходим с кодом ≠ 0, чтобы платформа
|
||
// перезапустила контейнер (ревью 🔴5: раньше процесс оставался жив, а
|
||
// /health отдавал 200).
|
||
var fatal atomic.Bool
|
||
|
||
go func() {
|
||
defer wg.Done()
|
||
// ready.SetBridge управляется самим bridge: true только после успешной
|
||
// подписки (находка #3). Иначе «соединён, но не подписан» давал /ready=200
|
||
// и молча терял телеметрию.
|
||
defer ready.SetBridge(false)
|
||
if err := bridge.Run(ctx, cfg, sqsClient, log, ready.SetBridge); err != nil && !errors.Is(err, context.Canceled) {
|
||
log.Error("bridge stopped", "err", err)
|
||
fatal.Store(true)
|
||
cancel() // штатный shutdown, затем выход ≠ 0
|
||
}
|
||
}()
|
||
|
||
go func() {
|
||
defer wg.Done()
|
||
// ready.SetConsumer управляется consumer'ом: true после успешного
|
||
// ReceiveMessage, false при ошибке SQS (находка «readiness»).
|
||
defer ready.SetConsumer(false)
|
||
if err := consumer.Run(ctx, cfg, sqsClient, iotStore, log, ready.SetConsumer); err != nil && !errors.Is(err, context.Canceled) {
|
||
log.Error("consumer stopped", "err", err)
|
||
fatal.Store(true)
|
||
cancel()
|
||
}
|
||
}()
|
||
|
||
// Сервер гаснет ПОСЛЕДНИМ: сначала останавливаются bridge+consumer
|
||
// (ctx → disconnect MQTT → flush SQS), потом HTTP. Иначе платформа
|
||
// может убить контейнер по liveness, пока bridge/consumer ещё работают
|
||
// (фикс MEDIUM из ревью 2026-08-16).
|
||
go func() {
|
||
<-ctx.Done()
|
||
// Ждём остановки подсистем, но не бесконечно (зависшую не ждём).
|
||
if !waitWithTimeout(&wg, shutdownGrace) {
|
||
log.Warn("iot-service: подсистемы не завершились за отведённое время", "grace", shutdownGrace)
|
||
}
|
||
shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||
defer shutdownCancel()
|
||
_ = server.Shutdown(shutdownCtx)
|
||
close(shutdownDone)
|
||
}()
|
||
|
||
log.Info("iot-service listening", "addr", server.Addr)
|
||
if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
|
||
log.Error("http server", "err", err)
|
||
cancel() // останавливаем bridge/consumer
|
||
waitWithTimeout(&wg, shutdownGrace)
|
||
<-shutdownDone
|
||
return 1
|
||
}
|
||
|
||
// Ждём завершения Shutdown (дренаж in-flight HTTP) до deferred Close.
|
||
<-shutdownDone
|
||
if !waitWithTimeout(&wg, shutdownGrace) {
|
||
log.Warn("iot-service: подсистемы не завершились за отведённое время", "grace", shutdownGrace)
|
||
}
|
||
if fatal.Load() {
|
||
// Не os.Exit: возвращаем код, чтобы отработали defer-ы (iotStore.Close →
|
||
// финальный flush последнего батча). os.Exit их бы обошёл.
|
||
log.Error("iot-service: подсистема завершилась с ошибкой — выход ≠ 0")
|
||
return 1
|
||
}
|
||
log.Info("iot-service stopped")
|
||
return 0
|
||
}
|