@@ -1026,6 +1048,14 @@ function emuSendOnce() {
mqttPublish(topic, payload.trim());
}
+// generateSensorPayload — случайный payload: температура [18-28°C], влажность [40-80%]
+// Используется кнопкой «Случайные данные» в эмуляторе
+function generateSensorPayload() {
+ const temp = parseFloat((18 + Math.random() * 10).toFixed(1));
+ const hum = Math.round(40 + Math.random() * 40);
+ return JSON.stringify({ temperature: temp, humidity: hum, ts: new Date().toISOString() });
+}
+
// toggleAuto — включить/выключить авто-отправку
function toggleAuto() {
const btn = document.getElementById('auto-btn');
@@ -1044,18 +1074,126 @@ function toggleAuto() {
// ВКЛАДКА — ТЕЛЕМЕТРИЯ
// =============================================================
function telemetryTab() {
+ const deviceOptions = S.devices.map(d =>
+ `
`
+ : S.telemetryRows.map(r => {
+ const ts = r.ts ? new Date(r.ts).toLocaleString('ru') : '—';
+ const dev = h(r.device_id || '—');
+ let payloadStr = '';
+ try {
+ payloadStr = typeof r.payload === 'object'
+ ? JSON.stringify(r.payload)
+ : String(r.payload);
+ } catch (_) { payloadStr = String(r.payload); }
+ return `
-
-
📊
-
Телеметрия
-
Скоро — хранение данных и API чтения
-
- В разработке — появится в следующем обновлении
-
+
Телеметрия
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ | Время |
+ Устройство |
+ Данные |
+
+
+ ${rows}
+
`;
}
+// loadTelemetry — загружает историю телеметрии с API.
+// URL: GET /v1/namespaces/{ns}/iot/telemetry?limit=100[&device=...]
+async function loadTelemetry() {
+ if (!S.ns || !S.token) return;
+ const status = document.getElementById('telem-status');
+ if (status) status.textContent = 'Загрузка...';
+
+ // Читаем текущий фильтр из DOM (если вкладка открыта) или из state
+ const sel = document.getElementById('telem-dev-filter');
+ if (sel) S.telemetryDeviceFilter = sel.value;
+
+ try {
+ let url = `${S.apiBase}/v1/namespaces/${encodeURIComponent(S.ns)}/iot/telemetry?limit=100`;
+ if (S.telemetryDeviceFilter) url += `&device=${encodeURIComponent(S.telemetryDeviceFilter)}`;
+
+ const resp = await fetch(url, { headers: { Authorization: 'Bearer ' + S.token } });
+ if (!resp.ok) throw new Error(`HTTP ${resp.status}`);
+ const data = await resp.json();
+ S.telemetryRows = data.items || [];
+
+ // Обновляем только tbody — не перерисовываем весь таб (сохраняем фильтр/чекбокс)
+ const tbody = document.getElementById('telem-tbody');
+ if (tbody) {
+ if (S.telemetryRows.length === 0) {
+ tbody.innerHTML = '
| Нет данных |
';
+ } else {
+ tbody.innerHTML = S.telemetryRows.map(r => {
+ const ts = r.ts ? new Date(r.ts).toLocaleString('ru') : '—';
+ const dev = h(r.device_id || '—');
+ let payloadStr = '';
+ try {
+ payloadStr = typeof r.payload === 'object'
+ ? JSON.stringify(r.payload)
+ : String(r.payload);
+ } catch (_) { payloadStr = String(r.payload); }
+ return `
+ | ${h(ts)} |
+ ${dev} |
+ ${h(payloadStr)} |
+
`;
+ }).join('');
+ }
+ }
+ if (status) status.textContent = `Обновлено в ${new Date().toLocaleTimeString('ru')}`;
+ } catch(e) {
+ if (status) status.textContent = 'Ошибка: ' + e.message;
+ }
+}
+
+// toggleTelemetryAutoRefresh — включить/выключить авто-обновление каждые 5 секунд
+function toggleTelemetryAutoRefresh(enabled) {
+ S.telemetryAutoRefresh = enabled;
+ clearInterval(S.telemetryRefreshTimer);
+ S.telemetryRefreshTimer = null;
+ if (enabled) {
+ // Немедленная загрузка + интервал
+ loadTelemetry();
+ S.telemetryRefreshTimer = setInterval(loadTelemetry, 5000);
+ }
+}
+
// =============================================================
// NAVBAR
// =============================================================
diff --git a/internal/storage/iotpg/iot_telemetry_store.go b/internal/storage/iotpg/iot_telemetry_store.go
new file mode 100644
index 0000000..574805b
--- /dev/null
+++ b/internal/storage/iotpg/iot_telemetry_store.go
@@ -0,0 +1,288 @@
+// Создано: 2026-04-05
+// iot_telemetry_store.go — управление per-tenant PostgreSQL databases для IoT телеметрии.
+//
+// Архитектура (принято 2026-04-04, см. doc/decisions/iot-telemetry-storage-2026-04-04.md):
+// - Один Postgres инстанс (iot-postgres.sless.svc) — отдельный от sless postgres
+// - Отдельная DATABASE per tenant: tenant_{namespace} (дефисы → подчёркивания)
+// - Suперюзер iot_admin управляет всеми DBs; клиенты читают только через REST API
+// - Пароли tenant хранятся в таблице tenant_credentials в management DB iot_platform
+//
+// Почему tenant_credentials в БД, а не в k8s Secret:
+// mqtt-bridge вызывает InsertTelemetry в горячем пути MQTT.
+// k8s API round-trip на каждое сообщение — неприемлемо.
+//
+// Подключение к tenant DB: суперюзер iot_admin, DSN строится заменой db name в adminDSN.
+// Кэширование: sync.Map для *sql.DB per tenant (lazy init при первом обращении).
+
+package iotpg
+
+import (
+ "context"
+ "database/sql"
+ "encoding/json"
+ "fmt"
+ "log/slog"
+ "os"
+ "strings"
+ "sync"
+ "time"
+
+ "github.com/google/uuid"
+ _ "github.com/lib/pq"
+)
+
+// IoTPostgresStore управляет per-tenant Postgres databases для IoT телеметрии.
+type IoTPostgresStore struct {
+ adminDB *sql.DB
+ adminDSN string
+ tenants sync.Map
+ log *slog.Logger
+}
+
+// TelemetryRow — одна запись телеметрии из таблицы iot_telemetry.
+type TelemetryRow struct {
+ ID int64 `json:"id"`
+ DeviceID string `json:"device_id"`
+ Ts time.Time `json:"ts"`
+ Payload json.RawMessage `json:"payload"`
+}
+
+// New подключается к management DB (iot_platform) и создаёт служебные таблицы.
+func New(adminDSN string, log *slog.Logger) (*IoTPostgresStore, error) {
+ db, err := sql.Open("postgres", adminDSN)
+ if err != nil {
+ return nil, fmt.Errorf("iotpg: open admin DB: %w", err)
+ }
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ if err := db.PingContext(ctx); err != nil {
+ db.Close()
+ return nil, fmt.Errorf("iotpg: ping admin DB: %w", err)
+ }
+ db.SetMaxOpenConns(5)
+ db.SetMaxIdleConns(2)
+ db.SetConnMaxLifetime(5 * time.Minute)
+
+ store := &IoTPostgresStore{adminDB: db, adminDSN: adminDSN, log: log}
+ if err := store.initManagementSchema(ctx); err != nil {
+ db.Close()
+ return nil, fmt.Errorf("iotpg: init management schema: %w", err)
+ }
+ log.Info("iotpg: connected to IoT Postgres management DB")
+ return store, nil
+}
+
+// NewFromEnv создаёт store из env var IOT_PG_DSN.
+// Возвращает (nil, nil) если переменная не задана — IoT Postgres опционален.
+func NewFromEnv(log *slog.Logger) (*IoTPostgresStore, error) {
+ dsn := os.Getenv("IOT_PG_DSN")
+ if dsn == "" {
+ log.Info("iotpg: IOT_PG_DSN not set, IoT telemetry disabled")
+ return nil, nil
+ }
+ return New(dsn, log)
+}
+
+// initManagementSchema создаёт таблицу tenant_credentials в iot_platform.
+func (s *IoTPostgresStore) initManagementSchema(ctx context.Context) error {
+ _, err := s.adminDB.ExecContext(ctx, `
+CREATE TABLE IF NOT EXISTS tenant_credentials (
+namespace TEXT PRIMARY KEY,
+pg_password TEXT NOT NULL,
+created_at TIMESTAMPTZ DEFAULT now()
+)
+`)
+ return err
+}
+
+// EnsureTenantDB создаёт DATABASE, USER и таблицу iot_telemetry для namespace.
+// Идемпотентен — повторный вызов безопасен.
+// Вызывается mqtt-bridge при первом сообщении от нового tenant.
+func (s *IoTPostgresStore) EnsureTenantDB(ctx context.Context, namespace string) error {
+ dbName := tenantDBName(namespace)
+ userName := dbName
+
+ var exists bool
+ err := s.adminDB.QueryRowContext(ctx,
+ `SELECT EXISTS(SELECT 1 FROM pg_database WHERE datname = $1)`, dbName,
+ ).Scan(&exists)
+ if err != nil {
+ return fmt.Errorf("iotpg: check tenant DB %s: %w", dbName, err)
+ }
+
+ if !exists {
+ password := uuid.New().String()
+
+ // CREATE USER через DO block — pg не поддерживает CREATE USER IF NOT EXISTS
+ _, err = s.adminDB.ExecContext(ctx, fmt.Sprintf(
+ `DO $$ BEGIN
+ IF NOT EXISTS (SELECT FROM pg_roles WHERE rolname = '%s') THEN
+ CREATE USER %s WITH PASSWORD '%s';
+ END IF;
+ END $$`, userName, userName, password,
+ ))
+ if err != nil {
+ return fmt.Errorf("iotpg: create user %s: %w", userName, err)
+ }
+
+ // CREATE DATABASE нельзя в транзакции
+ if _, err = s.adminDB.ExecContext(ctx,
+ fmt.Sprintf(`CREATE DATABASE %s OWNER %s`, dbName, userName),
+ ); err != nil {
+ return fmt.Errorf("iotpg: create database %s: %w", dbName, err)
+ }
+
+ if _, err = s.adminDB.ExecContext(ctx,
+ `INSERT INTO tenant_credentials (namespace, pg_password) VALUES ($1, $2)
+ ON CONFLICT (namespace) DO NOTHING`,
+ namespace, password,
+ ); err != nil {
+ return fmt.Errorf("iotpg: save credentials %s: %w", namespace, err)
+ }
+ s.log.Info("iotpg: created tenant DB", "namespace", namespace, "db", dbName)
+ }
+
+ // Создаём таблицу в tenant DB (суперюзер имеет доступ)
+ tenantDB, err := s.getTenantDB(ctx, namespace)
+ if err != nil {
+ return err
+ }
+ _, err = tenantDB.ExecContext(ctx, `
+CREATE TABLE IF NOT EXISTS iot_telemetry (
+id BIGSERIAL PRIMARY KEY,
+device_id TEXT NOT NULL,
+ts TIMESTAMPTZ NOT NULL DEFAULT now(),
+payload JSONB NOT NULL
+);
+CREATE INDEX IF NOT EXISTS idx_iot_telemetry_device_ts
+ON iot_telemetry (device_id, ts DESC);
+`)
+ return err
+}
+
+// InsertTelemetry записывает строку телеметрии в tenant DB.
+func (s *IoTPostgresStore) InsertTelemetry(ctx context.Context, namespace, deviceID string, payload json.RawMessage) error {
+ tenantDB, err := s.getTenantDB(ctx, namespace)
+ if err != nil {
+ return fmt.Errorf("iotpg: get tenant DB for insert: %w", err)
+ }
+ _, err = tenantDB.ExecContext(ctx,
+ `INSERT INTO iot_telemetry (device_id, payload) VALUES ($1, $2)`,
+ deviceID, []byte(payload),
+ )
+ return err
+}
+
+// QueryTelemetry читает телеметрию из tenant DB (ts DESC).
+// deviceID — фильтр (пустая строка = все устройства). limit — max записей (50..1000).
+func (s *IoTPostgresStore) QueryTelemetry(ctx context.Context, namespace, deviceID string, limit int) ([]TelemetryRow, error) {
+ if limit <= 0 {
+ limit = 50
+ }
+ if limit > 1000 {
+ limit = 1000
+ }
+ tenantDB, err := s.getTenantDB(ctx, namespace)
+ if err != nil {
+ return nil, fmt.Errorf("iotpg: get tenant DB for query: %w", err)
+ }
+
+ var rows *sql.Rows
+ if deviceID != "" {
+ rows, err = tenantDB.QueryContext(ctx,
+ `SELECT id, device_id, ts, payload FROM iot_telemetry
+ WHERE device_id = $1 ORDER BY ts DESC LIMIT $2`,
+ deviceID, limit,
+ )
+ } else {
+ rows, err = tenantDB.QueryContext(ctx,
+ `SELECT id, device_id, ts, payload FROM iot_telemetry
+ ORDER BY ts DESC LIMIT $1`,
+ limit,
+ )
+ }
+ if err != nil {
+ return nil, fmt.Errorf("iotpg: query telemetry for %s: %w", namespace, err)
+ }
+ defer rows.Close()
+
+ var result []TelemetryRow
+ for rows.Next() {
+ var r TelemetryRow
+ var rawPayload []byte
+ if err := rows.Scan(&r.ID, &r.DeviceID, &r.Ts, &rawPayload); err != nil {
+ return nil, fmt.Errorf("iotpg: scan row: %w", err)
+ }
+ r.Payload = json.RawMessage(rawPayload)
+ result = append(result, r)
+ }
+ return result, rows.Err()
+}
+
+// Close закрывает все подключения (admin + tenant кэш).
+func (s *IoTPostgresStore) Close() error {
+ s.tenants.Range(func(_, value any) bool {
+ if db, ok := value.(*sql.DB); ok {
+ db.Close()
+ }
+ return true
+ })
+ return s.adminDB.Close()
+}
+
+// getTenantDB возвращает *sql.DB для tenant DB из кэша или открывает новый.
+func (s *IoTPostgresStore) getTenantDB(ctx context.Context, namespace string) (*sql.DB, error) {
+ if cached, ok := s.tenants.Load(namespace); ok {
+ return cached.(*sql.DB), nil
+ }
+ dsn := replaceDSNDatabase(s.adminDSN, tenantDBName(namespace))
+ db, err := sql.Open("postgres", dsn)
+ if err != nil {
+ return nil, fmt.Errorf("iotpg: open tenant DB %s: %w", tenantDBName(namespace), err)
+ }
+ pingCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
+ defer cancel()
+ if err := db.PingContext(pingCtx); err != nil {
+ db.Close()
+ return nil, fmt.Errorf("iotpg: ping tenant DB %s: %w", tenantDBName(namespace), err)
+ }
+ db.SetMaxOpenConns(5)
+ db.SetMaxIdleConns(2)
+ db.SetConnMaxLifetime(5 * time.Minute)
+
+ actual, loaded := s.tenants.LoadOrStore(namespace, db)
+ if loaded {
+ db.Close()
+ return actual.(*sql.DB), nil
+ }
+ return db, nil
+}
+
+// replaceDSNDatabase заменяет имя базы данных в DSN.
+// Вход: "postgresql://user:pass@host:5432/iot_platform?sslmode=disable"
+// Выход: "postgresql://user:pass@host:5432/tenant_abc?sslmode=disable"
+func replaceDSNDatabase(dsn, newDBName string) string {
+ schemeEnd := strings.Index(dsn, "://")
+ if schemeEnd < 0 {
+ return dsn
+ }
+ hostPart := dsn[schemeEnd+3:]
+ slashIdx := strings.LastIndex(hostPart, "/")
+ if slashIdx < 0 {
+ return dsn
+ }
+ afterSlash := hostPart[slashIdx+1:]
+ suffix := ""
+ if qIdx := strings.Index(afterSlash, "?"); qIdx >= 0 {
+ suffix = afterSlash[qIdx:]
+ }
+ prefix := dsn[:schemeEnd+3+slashIdx+1]
+ return prefix + newDBName + suffix
+}
+
+// tenantDBName возвращает имя Postgres DATABASE для namespace.
+// Дефисы заменяются на подчёркивания (pg не поддерживает дефисы в unquoted именах).
+// Пример: "sless-abc123" → "tenant_sless_abc123"
+func tenantDBName(namespace string) string {
+ return "tenant_" + strings.ReplaceAll(namespace, "-", "_")
+}
diff --git a/iot/cmd/mqtt-bridge/main.go b/iot/cmd/mqtt-bridge/main.go
index 392e4bf..29420db 100644
--- a/iot/cmd/mqtt-bridge/main.go
+++ b/iot/cmd/mqtt-bridge/main.go
@@ -1,4 +1,5 @@
// Создано: 2026-04-04
+// Изменено: 2026-04-05 (добавлен INSERT в IoT Postgres)
// mqtt-bridge/main.go — сервис-мост: MQTT (EMQX) → RabbitMQ.
//
// Роль в архитектуре:
@@ -39,6 +40,8 @@ import (
mqtt "github.com/eclipse/paho.mqtt.golang"
amqp "github.com/rabbitmq/amqp091-go"
+
+ "gitea-naeel.giteak8s.services.ngcloud.ru/naeel/sless/internal/storage/iotpg"
)
// mqttBridgeConfig — конфигурация сервиса из env vars.
@@ -92,6 +95,18 @@ func main() {
}
defer rabbitCh.Close()
+ // IoT Postgres — сохранение телеметрии (per-tenant DB).
+ // Опционально: если IOT_PG_DSN не задан — продолжаем работать без Postgres (только RabbitMQ)
+ iotPGStore, err := iotpg.NewFromEnv(log)
+ if err != nil {
+ log.Error("failed to connect to IoT Postgres", "err", err)
+ os.Exit(1)
+ }
+ if iotPGStore != nil {
+ defer iotPGStore.Close()
+ log.Info("connected to IoT Postgres for telemetry storage")
+ }
+
// Создаём MQTT клиент
mqttClient, err := connectMQTT(cfg, log)
if err != nil {
@@ -102,7 +117,7 @@ func main() {
// Функция-обработчик MQTT сообщений
// Вызывается в goroutine paho при каждом сообщении
- messageHandler := buildMQTTMessageHandler(rabbitCh, log)
+ messageHandler := buildMQTTMessageHandler(ctx, rabbitCh, iotPGStore, log)
// Подписываемся на все telemetry топики всех namespace
// "+/telemetry/+" = {любой namespace}/telemetry/{любой deviceId}
@@ -203,8 +218,13 @@ func connectRabbitMQWithRetry(ctx context.Context, url string, log *slog.Logger)
}
// buildMQTTMessageHandler возвращает функцию-обработчик MQTT сообщений.
-// Замыкание над rabbitCh (RabbitMQ channel) и logger.
-func buildMQTTMessageHandler(rabbitCh *amqp.Channel, log *slog.Logger) mqtt.MessageHandler {
+// Замыкание над rabbitCh (RabbitMQ channel), iotStore (может быть nil) и logger.
+// Порядок действий при получении сообщения:
+// 1. INSERT в IoT Postgres (tenant DB) — если iotStore != nil
+// 2. Publish в RabbitMQ — всегда (для event-dispatcher → function triggers)
+//
+// Ошибка INSERT не блокирует RabbitMQ publish — разные failure domain.
+func buildMQTTMessageHandler(ctx context.Context, rabbitCh *amqp.Channel, iotStore *iotpg.IoTPostgresStore, log *slog.Logger) mqtt.MessageHandler {
return func(_ mqtt.Client, msg mqtt.Message) {
topic := msg.Topic()
payload := msg.Payload()
@@ -219,15 +239,28 @@ func buildMQTTMessageHandler(rabbitCh *amqp.Channel, log *slog.Logger) mqtt.Mess
ns := parts[0]
deviceID := parts[2]
- // Формируем envelope — оборачиваем payload в JSON с метаданными
- // Payload от устройства может быть любым JSON или строкой
+ // Нормализуем payload: если это не JSON — оборачиваем в строку
rawPayload := json.RawMessage(payload)
if !json.Valid(payload) {
- // Если payload не JSON — упаковываем в строку
quotedBytes, _ := json.Marshal(string(payload))
rawPayload = json.RawMessage(quotedBytes)
}
+ // ШАГ 1: INSERT в IoT Postgres — сохраняем телеметрию в per-tenant DB
+ // EnsureTenantDB идемпотентен: кэшируется после первого вызова
+ if iotStore != nil {
+ if err := iotStore.EnsureTenantDB(ctx, ns); err != nil {
+ log.Error("ensure tenant DB", "namespace", ns, "err", err)
+ // НЕ возвращаемся — продолжаем RabbitMQ publish
+ } else if err := iotStore.InsertTelemetry(ctx, ns, deviceID, rawPayload); err != nil {
+ log.Error("insert telemetry", "topic", topic, "err", err)
+ // НЕ возвращаемся — RabbitMQ не должен зависеть от Postgres
+ } else {
+ log.Debug("telemetry saved to Postgres", "namespace", ns, "device", deviceID)
+ }
+ }
+
+ // ШАГ 2: Publish в RabbitMQ (для event-dispatcher → function triggers)
envelope := iotTelemetryMessage{
Namespace: ns,
DeviceID: deviceID,
diff --git a/main.go b/main.go
index 70d45b1..cf4c5d1 100644
--- a/main.go
+++ b/main.go
@@ -1,4 +1,4 @@
-// Изменено: 2026-03-20 (function-service-split: добавлена регистрация ServiceReconciler)
+// Изменено: 2026-04-05 (добавлена инициализация IoT Postgres для телеметрии)
// main.go — точка входа. Запускает operator manager и REST API сервер параллельно.
// Operator manager управляет Function/Trigger CRD через reconcile loop.
// REST API (gorilla/mux) принимает запросы от Terraform provider.
@@ -32,6 +32,7 @@ import (
"gitea-naeel.giteak8s.services.ngcloud.ru/naeel/sless/internal/builder"
"gitea-naeel.giteak8s.services.ngcloud.ru/naeel/sless/internal/config"
"gitea-naeel.giteak8s.services.ngcloud.ru/naeel/sless/internal/harbor"
+ "gitea-naeel.giteak8s.services.ngcloud.ru/naeel/sless/internal/storage/iotpg"
"gitea-naeel.giteak8s.services.ngcloud.ru/naeel/sless/internal/storage/postgres"
"gitea-naeel.giteak8s.services.ngcloud.ru/naeel/sless/internal/storage/s3"
iotv1alpha1 "gitea-naeel.giteak8s.services.ngcloud.ru/naeel/sless/iot/api/v1alpha1"
@@ -213,12 +214,24 @@ func main() {
os.Exit(1)
}
+ // IoT Postgres — подключение к per-tenant storage для телеметрии
+ // Опционально: если IOT_PG_DSN не задан — телеметрия недоступна (503), остальное работает
+ iotPGStore, err := iotpg.NewFromEnv(log)
+ if err != nil {
+ log.Error("connect IoT Postgres", "err", err)
+ os.Exit(1)
+ }
+ if iotPGStore != nil {
+ defer iotPGStore.Close()
+ }
+
// REST API сервер — запускается параллельно с operator manager
apiHandler := slessapi.NewRouter(&handler.Handler{
K8s: mgr.GetClient(),
Scheme: mgr.GetScheme(),
S3: s3Client,
PG: pg,
+ IoTPG: iotPGStore,
Log: log,
}, log)