feat: перенос IoT managed service из sless в отдельную репу
Компоненты: - iot-operator: controller-manager (IoTDevice CRD) + REST API (порт 9090) - mqtt-bridge: MQTT (EMQX) → Kafka bridge - kafka-consumer: Kafka → Postgres pipeline Модуль: gitea.services.ngcloud.ru/Nail/IoT Все 3 бинарника собираются, import paths адаптированы.
This commit is contained in:
Naeel
@@ -0,0 +1,61 @@
|
||||
// Создано: 2026-04-12
|
||||
// Handler — контейнер зависимостей для IoT REST handlers.
|
||||
// Упрощённая версия из sless — только IoT-специфичные поля.
|
||||
//
|
||||
// Бизнес-логика:
|
||||
// - iot_device_handler.go — CRUD устройств + MQTT auth
|
||||
// - iot_telemetry_handler.go — чтение телеметрии
|
||||
// - iot_admin_stats_handler.go — админ-статистика
|
||||
|
||||
package handler
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
|
||||
"github.com/gorilla/mux"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"sigs.k8s.io/controller-runtime/pkg/client"
|
||||
|
||||
"gitea.services.ngcloud.ru/Nail/IoT/internal/storage/iotpg"
|
||||
)
|
||||
|
||||
// defaultNamespace — fallback namespace для dev/тестов.
|
||||
const defaultNamespace = "default"
|
||||
|
||||
// Handler содержит зависимости для всех IoT REST-обработчиков.
|
||||
type Handler struct {
|
||||
K8s client.Client
|
||||
Scheme *runtime.Scheme
|
||||
// IoTPG — хранилище IoT телеметрии (per-tenant Postgres). nil если IOT_PG_DSN не задан.
|
||||
IoTPG *iotpg.IoTPostgresStore
|
||||
// KafkaBrokers — адреса Kafka брокеров для чтения consumer lag на admin странице.
|
||||
KafkaBrokers string
|
||||
Log *slog.Logger
|
||||
}
|
||||
|
||||
// writeJSON отправляет JSON-ответ с указанным статусом.
|
||||
func writeJSON(w http.ResponseWriter, status int, v any) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(status)
|
||||
_ = json.NewEncoder(w).Encode(v)
|
||||
}
|
||||
|
||||
// errResp возвращает структуру ошибки для JSON.
|
||||
func errResp(msg string) map[string]string {
|
||||
return map[string]string{"error": msg}
|
||||
}
|
||||
|
||||
// pathVar читает переменную пути из gorilla/mux.
|
||||
func pathVar(r *http.Request, key string) string {
|
||||
return mux.Vars(r)[key]
|
||||
}
|
||||
|
||||
// namespace читает {namespace} из пути URL.
|
||||
func namespace(r *http.Request) string {
|
||||
if ns := mux.Vars(r)["namespace"]; ns != "" {
|
||||
return ns
|
||||
}
|
||||
return defaultNamespace
|
||||
}
|
||||
@@ -0,0 +1,214 @@
|
||||
// Создано: 2026-04-06
|
||||
// iot_admin_stats_handler.go — handler для страницы администратора IoT.
|
||||
//
|
||||
// Endpoints:
|
||||
// GET /iot-admin/stats — JSON с агрегированной статистикой (защищён ADMIN_STATS_TOKEN)
|
||||
//
|
||||
// Источники данных:
|
||||
// - PostgreSQL (IoTPG): counts per tenant, last 1h/24h, latest rows
|
||||
// - Kafka: consumer lag (latest offset - committed offset для group iot-pg-consumer)
|
||||
// - K8s: статус подов iot-mqtt-bridge и iot-kafka-consumer
|
||||
//
|
||||
// Авторизация: Bearer из env ADMIN_STATS_TOKEN.
|
||||
// Если ADMIN_STATS_TOKEN не задан — endpoint возвращает 503.
|
||||
|
||||
package handler
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"os"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
kafka "github.com/segmentio/kafka-go"
|
||||
corev1 "k8s.io/api/core/v1"
|
||||
"sigs.k8s.io/controller-runtime/pkg/client"
|
||||
)
|
||||
|
||||
// iotAdminPodStatus — краткая информация о k8s pod для страницы администратора.
|
||||
type iotAdminPodStatus struct {
|
||||
Name string `json:"name"`
|
||||
Phase string `json:"phase"`
|
||||
Ready bool `json:"ready"`
|
||||
Restarts int32 `json:"restarts"`
|
||||
Age string `json:"age"`
|
||||
}
|
||||
|
||||
// iotAdminKafkaStats — информация о Kafka топике и consumer lag.
|
||||
type iotAdminKafkaStats struct {
|
||||
LatestOffset int64 `json:"latest_offset"`
|
||||
CommittedOffset int64 `json:"committed_offset"`
|
||||
ConsumerLag int64 `json:"consumer_lag"`
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
// AdminStats обрабатывает GET /iot-admin/stats.
|
||||
// Проверяет Bearer-токен из ADMIN_STATS_TOKEN, затем собирает и возвращает статистику.
|
||||
func (h *Handler) AdminStats(w http.ResponseWriter, r *http.Request) {
|
||||
adminToken := os.Getenv("ADMIN_STATS_TOKEN")
|
||||
if adminToken == "" {
|
||||
writeJSON(w, http.StatusServiceUnavailable, errResp("admin stats not configured: ADMIN_STATS_TOKEN not set"))
|
||||
return
|
||||
}
|
||||
|
||||
authHeader := r.Header.Get("Authorization")
|
||||
if !strings.HasPrefix(authHeader, "Bearer ") || strings.TrimPrefix(authHeader, "Bearer ") != adminToken {
|
||||
writeJSON(w, http.StatusUnauthorized, errResp("unauthorized"))
|
||||
return
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(r.Context(), 15*time.Second)
|
||||
defer cancel()
|
||||
|
||||
result := map[string]any{
|
||||
"collected_at": time.Now().UTC(),
|
||||
}
|
||||
|
||||
// PostgreSQL: статистика по всем tenant
|
||||
if h.IoTPG != nil {
|
||||
pgStats, err := h.IoTPG.GetAdminStats(ctx)
|
||||
if err != nil {
|
||||
result["postgres"] = map[string]any{"reachable": false, "error": err.Error()}
|
||||
} else {
|
||||
result["postgres"] = pgStats
|
||||
}
|
||||
} else {
|
||||
result["postgres"] = map[string]any{"reachable": false, "error": "IoTPG not configured"}
|
||||
}
|
||||
|
||||
// Kafka: consumer lag для топика iot.telemetry / группы iot-pg-consumer
|
||||
result["kafka"] = h.collectIotKafkaLag(ctx)
|
||||
|
||||
// K8s: статус подов bridge и consumer
|
||||
result["pods"] = h.collectIotPodStatuses(ctx)
|
||||
|
||||
writeJSON(w, http.StatusOK, result)
|
||||
}
|
||||
|
||||
// collectIotKafkaLag получает latest offset топика и committed offset consumer group,
|
||||
// вычисляет lag = latest - committed.
|
||||
// Topic: "iot.telemetry", Consumer Group: "iot-pg-consumer".
|
||||
func (h *Handler) collectIotKafkaLag(ctx context.Context) iotAdminKafkaStats {
|
||||
if h.KafkaBrokers == "" {
|
||||
return iotAdminKafkaStats{Error: "KAFKA_BROKERS not configured"}
|
||||
}
|
||||
|
||||
brokers := strings.Split(h.KafkaBrokers, ",")
|
||||
brokerAddr := kafka.TCP(brokers...)
|
||||
|
||||
kc := &kafka.Client{
|
||||
Addr: brokerAddr,
|
||||
Timeout: 5 * time.Second,
|
||||
}
|
||||
|
||||
const topic = "iot.telemetry"
|
||||
const group = "iot-pg-consumer"
|
||||
|
||||
// Получаем latest offset (конец лога — сколько всего сообщений прошло)
|
||||
offsetsResp, err := kc.ListOffsets(ctx, &kafka.ListOffsetsRequest{
|
||||
Addr: brokerAddr,
|
||||
Topics: map[string][]kafka.OffsetRequest{
|
||||
topic: {kafka.LastOffsetOf(0)},
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
return iotAdminKafkaStats{Error: fmt.Sprintf("list offsets: %v", err)}
|
||||
}
|
||||
|
||||
var latestOffset int64
|
||||
if partitions, ok := offsetsResp.Topics[topic]; ok && len(partitions) > 0 {
|
||||
if partitions[0].Error == nil {
|
||||
latestOffset = partitions[0].LastOffset
|
||||
}
|
||||
}
|
||||
|
||||
// Получаем committed offset consumer group (что consumer уже обработал)
|
||||
fetchResp, err := kc.OffsetFetch(ctx, &kafka.OffsetFetchRequest{
|
||||
Addr: brokerAddr,
|
||||
GroupID: group,
|
||||
Topics: map[string][]int{topic: {0}},
|
||||
})
|
||||
if err != nil {
|
||||
return iotAdminKafkaStats{
|
||||
LatestOffset: latestOffset,
|
||||
Error: fmt.Sprintf("offset fetch: %v", err),
|
||||
}
|
||||
}
|
||||
|
||||
var committedOffset int64
|
||||
if partitions, ok := fetchResp.Topics[topic]; ok && len(partitions) > 0 {
|
||||
if partitions[0].Error == nil {
|
||||
committedOffset = partitions[0].CommittedOffset
|
||||
}
|
||||
}
|
||||
|
||||
lag := latestOffset - committedOffset
|
||||
if lag < 0 {
|
||||
lag = 0
|
||||
}
|
||||
|
||||
return iotAdminKafkaStats{
|
||||
LatestOffset: latestOffset,
|
||||
CommittedOffset: committedOffset,
|
||||
ConsumerLag: lag,
|
||||
}
|
||||
}
|
||||
|
||||
// collectIotPodStatuses собирает статус k8s pods для bridge и consumer по label app={name}.
|
||||
func (h *Handler) collectIotPodStatuses(ctx context.Context) map[string]any {
|
||||
result := map[string]any{}
|
||||
|
||||
for _, appLabel := range []string{"iot-mqtt-bridge", "iot-kafka-consumer"} {
|
||||
podList := &corev1.PodList{}
|
||||
if err := h.K8s.List(ctx, podList,
|
||||
client.InNamespace("sless"),
|
||||
client.MatchingLabels{"app": appLabel},
|
||||
); err != nil {
|
||||
result[appLabel] = map[string]any{"error": err.Error()}
|
||||
continue
|
||||
}
|
||||
if len(podList.Items) == 0 {
|
||||
result[appLabel] = map[string]any{"status": "not found"}
|
||||
continue
|
||||
}
|
||||
|
||||
pod := podList.Items[0]
|
||||
var restarts int32
|
||||
for _, cs := range pod.Status.ContainerStatuses {
|
||||
restarts += cs.RestartCount
|
||||
}
|
||||
ready := false
|
||||
for _, cond := range pod.Status.Conditions {
|
||||
if cond.Type == corev1.PodReady && cond.Status == corev1.ConditionTrue {
|
||||
ready = true
|
||||
}
|
||||
}
|
||||
|
||||
result[appLabel] = iotAdminPodStatus{
|
||||
Name: pod.Name,
|
||||
Phase: string(pod.Status.Phase),
|
||||
Ready: ready,
|
||||
Restarts: restarts,
|
||||
Age: iotFormatAge(pod.CreationTimestamp.Time),
|
||||
}
|
||||
}
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
// iotFormatAge возвращает человекочитаемый возраст (s/m/h/d) pod-а.
|
||||
func iotFormatAge(created time.Time) string {
|
||||
d := time.Since(created)
|
||||
switch {
|
||||
case d < time.Minute:
|
||||
return fmt.Sprintf("%ds", int(d.Seconds()))
|
||||
case d < time.Hour:
|
||||
return fmt.Sprintf("%dm", int(d.Minutes()))
|
||||
case d < 24*time.Hour:
|
||||
return fmt.Sprintf("%dh", int(d.Hours()))
|
||||
default:
|
||||
return fmt.Sprintf("%dd", int(d.Hours()/24))
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,454 @@
|
||||
// Создано: 2026-04-04
|
||||
// iot_device_handler.go — HTTP handlers для IoT-устройств (CRUD) и MQTT auth.
|
||||
//
|
||||
// Endpoints:
|
||||
// POST /internal/mqtt/auth → MQTTAuth (без JWT, для EMQX)
|
||||
// POST /v1/namespaces/{ns}/iot/devices → CreateIoTDevice
|
||||
// GET /v1/namespaces/{ns}/iot/devices → ListIoTDevices
|
||||
// GET /v1/namespaces/{ns}/iot/devices/{name} → GetIoTDevice (включает credentials)
|
||||
// DELETE /v1/namespaces/{ns}/iot/devices/{name} → DeleteIoTDevice
|
||||
// PATCH /v1/namespaces/{ns}/iot/devices/{name} → UpdateIoTDevice
|
||||
//
|
||||
// MQTTAuth вызывается EMQX при каждом MQTT CONNECT:
|
||||
// - всегда возвращает HTTP 200 (non-200 = EMQX игнорирует backend)
|
||||
// - {"result": "allow"|"deny"} в теле
|
||||
//
|
||||
// GetIoTDevice — единственный endpoint возвращающий mqtt-password.
|
||||
// ListIoTDevices — без паролей (security by design).
|
||||
|
||||
package handler
|
||||
|
||||
import (
|
||||
"crypto/subtle"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
corev1 "k8s.io/api/core/v1"
|
||||
"k8s.io/apimachinery/pkg/api/errors"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"sigs.k8s.io/controller-runtime/pkg/client"
|
||||
|
||||
iotv1alpha1 "gitea.services.ngcloud.ru/Nail/IoT/api/v1alpha1"
|
||||
)
|
||||
|
||||
// ——————————————————————————————————————————
|
||||
// Типы запросов / ответов
|
||||
// ——————————————————————————————————————————
|
||||
|
||||
// iotDeviceCreateRequest — тело POST при создании IoTDevice.
|
||||
type iotDeviceCreateRequest struct {
|
||||
// Name — имя k8s объекта IoTDevice (должно быть уникальным в namespace)
|
||||
Name string `json:"name"`
|
||||
// DeviceID — идентификатор устройства, используется в MQTT username и имени Secret
|
||||
DeviceID string `json:"device_id"`
|
||||
// Enabled — активно ли устройство с момента создания
|
||||
Enabled *bool `json:"enabled"`
|
||||
// Metadata — произвольные метаданные (модель, локация и т.д.)
|
||||
Metadata map[string]string `json:"metadata,omitempty"`
|
||||
}
|
||||
|
||||
// iotDeviceUpdateRequest — тело PATCH при обновлении IoTDevice.
|
||||
type iotDeviceUpdateRequest struct {
|
||||
// Enabled — включить/отключить устройство
|
||||
Enabled *bool `json:"enabled"`
|
||||
}
|
||||
|
||||
// iotDeviceResponse — ответ при чтении одного IoTDevice.
|
||||
// MQTTPassword заполняется только из GetIoTDevice (чтение из Secret).
|
||||
type iotDeviceResponse struct {
|
||||
Name string `json:"name"`
|
||||
Namespace string `json:"namespace"`
|
||||
DeviceID string `json:"device_id"`
|
||||
Enabled bool `json:"enabled"`
|
||||
Phase iotv1alpha1.IoTDevicePhase `json:"phase"`
|
||||
MQTTUsername string `json:"mqtt_username,omitempty"`
|
||||
MQTTPassword string `json:"mqtt_password,omitempty"` // только в GET /devices/{name}
|
||||
SecretName string `json:"secret_name,omitempty"`
|
||||
TopicPrefix string `json:"topic_prefix,omitempty"`
|
||||
LastConnected string `json:"last_connected,omitempty"`
|
||||
Message string `json:"message,omitempty"`
|
||||
Metadata map[string]string `json:"metadata,omitempty"`
|
||||
CreatedAt string `json:"created_at,omitempty"`
|
||||
}
|
||||
|
||||
// mqttAuthRequest — тело запроса от EMQX при MQTT CONNECT.
|
||||
// EMQX 5.x посылает JSON: username, password, clientid, peerhost.
|
||||
type mqttAuthRequest struct {
|
||||
Username string `json:"username"`
|
||||
Password string `json:"password"`
|
||||
ClientID string `json:"clientid"`
|
||||
PeerHost string `json:"peerhost"`
|
||||
}
|
||||
|
||||
// mqttAuthResponse — ответ для EMQX. Всегда HTTP 200.
|
||||
// result = "allow" | "deny"
|
||||
// ACL — список правил pub/sub, изолирует топики по устройству.
|
||||
type mqttAuthResponse struct {
|
||||
Result string `json:"result"`
|
||||
ACL []aclRule `json:"acl,omitempty"`
|
||||
}
|
||||
|
||||
// aclRule — одно правило ACL для EMQX HTTP auth plugin.
|
||||
// permission: "allow" | "deny"
|
||||
// action: "publish" | "subscribe" | "all"
|
||||
// topic: точный топик или wildcard (#, +)
|
||||
type aclRule struct {
|
||||
Permission string `json:"permission"`
|
||||
Action string `json:"action"`
|
||||
Topic string `json:"topic"`
|
||||
}
|
||||
|
||||
// ——————————————————————————————————————————
|
||||
// Вспомогательные функции
|
||||
// ——————————————————————————————————————————
|
||||
|
||||
// deviceToResponse конвертирует IoTDevice CRD в ответ API.
|
||||
// password передаётся отдельно — берётся из Secret только в GetIoTDevice.
|
||||
func deviceToResponse(d *iotv1alpha1.IoTDevice, password string) iotDeviceResponse {
|
||||
resp := iotDeviceResponse{
|
||||
Name: d.Name,
|
||||
Namespace: d.Namespace,
|
||||
DeviceID: d.Spec.DeviceID,
|
||||
Enabled: d.Spec.Enabled,
|
||||
Phase: d.Status.Phase,
|
||||
MQTTUsername: d.Status.MQTTUsername,
|
||||
MQTTPassword: password,
|
||||
SecretName: d.Status.SecretName,
|
||||
TopicPrefix: d.Status.TopicPrefix,
|
||||
Message: d.Status.Message,
|
||||
Metadata: d.Spec.Metadata,
|
||||
}
|
||||
if d.Status.LastConnected != nil && !d.Status.LastConnected.IsZero() {
|
||||
resp.LastConnected = d.Status.LastConnected.UTC().Format(time.RFC3339)
|
||||
}
|
||||
if !d.CreationTimestamp.IsZero() {
|
||||
resp.CreatedAt = d.CreationTimestamp.UTC().Format("2006-01-02 15:04:05 UTC")
|
||||
}
|
||||
return resp
|
||||
}
|
||||
|
||||
// ——————————————————————————————————————————
|
||||
// MQTT Auth — Этап 2
|
||||
// ——————————————————————————————————————————
|
||||
|
||||
// MQTTAuth — POST /internal/mqtt/auth
|
||||
// Вызывается EMQX при каждом MQTT CONNECT.
|
||||
// НЕ защищён JWT middleware — доступен только из кластера (путь /internal/).
|
||||
//
|
||||
// Логика аутентификации:
|
||||
// 1. Распарсить username → namespace + deviceId
|
||||
// 2. Получить Secret iot-{deviceId} в namespace
|
||||
// 3. Constant-time сравнение пароля (защита от timing attacks)
|
||||
// 4. Проверить что IoTDevice существует и enabled=true
|
||||
// 5. Обновить status.lastConnected в IoTDevice
|
||||
func (h *Handler) MQTTAuth(w http.ResponseWriter, r *http.Request) {
|
||||
var req mqttAuthRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
// Плохой JSON от EMQX — deny, но не 400 (EMQX игнорирует non-200)
|
||||
writeJSON(w, http.StatusOK, mqttAuthResponse{Result: "deny"})
|
||||
return
|
||||
}
|
||||
|
||||
// Парсим username: "{namespace}_{deviceId}"
|
||||
// Namespace содержит только [a-z0-9-], первый "_" — разделитель.
|
||||
idx := strings.Index(req.Username, "_")
|
||||
if idx < 0 {
|
||||
writeJSON(w, http.StatusOK, mqttAuthResponse{Result: "deny"})
|
||||
return
|
||||
}
|
||||
ns := req.Username[:idx]
|
||||
deviceID := req.Username[idx+1:]
|
||||
if ns == "" || deviceID == "" {
|
||||
writeJSON(w, http.StatusOK, mqttAuthResponse{Result: "deny"})
|
||||
return
|
||||
}
|
||||
|
||||
// Читаем Secret с MQTT credentials
|
||||
secretName := "iot-" + deviceID
|
||||
secret := &corev1.Secret{}
|
||||
if err := h.K8s.Get(r.Context(), client.ObjectKey{Namespace: ns, Name: secretName}, secret); err != nil {
|
||||
// Secret не найден или ошибка k8s — deny
|
||||
writeJSON(w, http.StatusOK, mqttAuthResponse{Result: "deny"})
|
||||
return
|
||||
}
|
||||
|
||||
// Constant-time сравнение пароля — защита от timing attacks
|
||||
storedPassword := secret.Data["mqtt-password"]
|
||||
if subtle.ConstantTimeCompare(storedPassword, []byte(req.Password)) != 1 {
|
||||
writeJSON(w, http.StatusOK, mqttAuthResponse{Result: "deny"})
|
||||
return
|
||||
}
|
||||
|
||||
// Проверяем что IoTDevice активно
|
||||
device := &iotv1alpha1.IoTDevice{}
|
||||
if err := h.K8s.Get(r.Context(), client.ObjectKey{Namespace: ns, Name: deviceID}, device); err != nil {
|
||||
// IoTDevice не найден (или удалён) — deny
|
||||
writeJSON(w, http.StatusOK, mqttAuthResponse{Result: "deny"})
|
||||
return
|
||||
}
|
||||
if !device.Spec.Enabled {
|
||||
writeJSON(w, http.StatusOK, mqttAuthResponse{Result: "deny"})
|
||||
return
|
||||
}
|
||||
|
||||
// Обновляем lastConnected в статусе устройства (best-effort, ошибка не критична)
|
||||
now := metav1.NewTime(time.Now().UTC())
|
||||
device.Status.LastConnected = &now
|
||||
if err := h.K8s.Status().Update(r.Context(), device); err != nil {
|
||||
h.Log.Warn("mqtt auth: failed to update lastConnected", "device", deviceID, "err", err)
|
||||
// Продолжаем — это некритично, устройство всё равно авторизовано
|
||||
}
|
||||
|
||||
// Формируем ACL правила для этого подключения.
|
||||
// Топик устройства: "{namespace}/telemetry/{deviceId}"
|
||||
// Это то что строит эмулятор: topicPrefix + "telemetry/" + device_id
|
||||
// topicPrefix = "{ns}/" → итого "{ns}/telemetry/{deviceId}"
|
||||
deviceTopic := ns + "/telemetry/" + deviceID
|
||||
|
||||
var aclRules []aclRule
|
||||
|
||||
if req.ClientID == "sless-iot-bridge" {
|
||||
// Bridge подписывается на "+/telemetry/+" (все тенанты) — разрешаем
|
||||
// Bridge НЕ публикует через MQTT — только читает
|
||||
aclRules = []aclRule{
|
||||
{Permission: "allow", Action: "subscribe", Topic: "+/telemetry/+"},
|
||||
{Permission: "deny", Action: "all", Topic: "#"},
|
||||
}
|
||||
} else {
|
||||
// Обычное IoT устройство: только свой топик
|
||||
aclRules = []aclRule{
|
||||
{Permission: "allow", Action: "publish", Topic: deviceTopic},
|
||||
{Permission: "allow", Action: "subscribe", Topic: deviceTopic},
|
||||
{Permission: "deny", Action: "all", Topic: "#"},
|
||||
}
|
||||
}
|
||||
|
||||
writeJSON(w, http.StatusOK, mqttAuthResponse{
|
||||
Result: "allow",
|
||||
ACL: aclRules,
|
||||
})
|
||||
}
|
||||
|
||||
// ——————————————————————————————————————————
|
||||
// IoT Device CRUD — Этап 4
|
||||
// ——————————————————————————————————————————
|
||||
|
||||
// CreateIoTDevice — POST /v1/namespaces/{namespace}/iot/devices
|
||||
// Создаёт IoTDevice CRD. Контроллер асинхронно сгенерирует MQTT credentials.
|
||||
// credentials доступны через GET /devices/{name} после reconcile (phase=Active).
|
||||
func (h *Handler) CreateIoTDevice(w http.ResponseWriter, r *http.Request) {
|
||||
ns := namespace(r)
|
||||
var req iotDeviceCreateRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
writeJSON(w, http.StatusBadRequest, errResp("invalid JSON: "+err.Error()))
|
||||
return
|
||||
}
|
||||
if req.Name == "" {
|
||||
writeJSON(w, http.StatusBadRequest, errResp("name is required"))
|
||||
return
|
||||
}
|
||||
if req.DeviceID == "" {
|
||||
writeJSON(w, http.StatusBadRequest, errResp("device_id is required"))
|
||||
return
|
||||
}
|
||||
|
||||
enabled := true
|
||||
if req.Enabled != nil {
|
||||
enabled = *req.Enabled
|
||||
}
|
||||
|
||||
device := &iotv1alpha1.IoTDevice{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: req.Name,
|
||||
Namespace: ns,
|
||||
},
|
||||
Spec: iotv1alpha1.IoTDeviceSpec{
|
||||
DeviceID: req.DeviceID,
|
||||
Enabled: enabled,
|
||||
Metadata: req.Metadata,
|
||||
},
|
||||
}
|
||||
|
||||
if err := h.K8s.Create(r.Context(), device); err != nil {
|
||||
if errors.IsAlreadyExists(err) {
|
||||
writeJSON(w, http.StatusConflict, errResp("iot device already exists"))
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusInternalServerError, errResp(err.Error()))
|
||||
return
|
||||
}
|
||||
|
||||
writeJSON(w, http.StatusCreated, deviceToResponse(device, ""))
|
||||
}
|
||||
|
||||
// ListIoTDevices — GET /v1/namespaces/{namespace}/iot/devices
|
||||
// Возвращает список устройств БЕЗ паролей (security by design).
|
||||
func (h *Handler) ListIoTDevices(w http.ResponseWriter, r *http.Request) {
|
||||
ns := namespace(r)
|
||||
list := &iotv1alpha1.IoTDeviceList{}
|
||||
if err := h.K8s.List(r.Context(), list, client.InNamespace(ns)); err != nil {
|
||||
writeJSON(w, http.StatusInternalServerError, errResp(err.Error()))
|
||||
return
|
||||
}
|
||||
result := make([]iotDeviceResponse, 0, len(list.Items))
|
||||
for i := range list.Items {
|
||||
result = append(result, deviceToResponse(&list.Items[i], ""))
|
||||
}
|
||||
writeJSON(w, http.StatusOK, result)
|
||||
}
|
||||
|
||||
// GetIoTDevice — GET /v1/namespaces/{namespace}/iot/devices/{name}
|
||||
// Возвращает устройство включая mqtt_password из Secret.
|
||||
// mqtt_password нужен пользователю для конфигурации физического устройства.
|
||||
func (h *Handler) GetIoTDevice(w http.ResponseWriter, r *http.Request) {
|
||||
ns := namespace(r)
|
||||
name := pathVar(r, "name")
|
||||
|
||||
device := &iotv1alpha1.IoTDevice{}
|
||||
if err := h.K8s.Get(r.Context(), client.ObjectKey{Namespace: ns, Name: name}, device); err != nil {
|
||||
if errors.IsNotFound(err) {
|
||||
writeJSON(w, http.StatusNotFound, errResp("iot device not found"))
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusInternalServerError, errResp(err.Error()))
|
||||
return
|
||||
}
|
||||
|
||||
// Читаем пароль из Secret — если ещё не создан (phase=Pending), password будет пустым
|
||||
password := ""
|
||||
if device.Status.SecretName != "" {
|
||||
secret := &corev1.Secret{}
|
||||
err := h.K8s.Get(r.Context(), client.ObjectKey{Namespace: ns, Name: device.Status.SecretName}, secret)
|
||||
if err == nil {
|
||||
password = string(secret.Data["mqtt-password"])
|
||||
}
|
||||
// Если Secret не найден — просто передаём пустой пароль (устройство ещё provisioning)
|
||||
}
|
||||
|
||||
writeJSON(w, http.StatusOK, deviceToResponse(device, password))
|
||||
}
|
||||
|
||||
// DeleteIoTDevice — DELETE /v1/namespaces/{namespace}/iot/devices/{name}
|
||||
// Удаляет IoTDevice CRD. Контроллер через finalizer удалит Secret каскадно.
|
||||
func (h *Handler) DeleteIoTDevice(w http.ResponseWriter, r *http.Request) {
|
||||
ns := namespace(r)
|
||||
name := pathVar(r, "name")
|
||||
|
||||
device := &iotv1alpha1.IoTDevice{}
|
||||
if err := h.K8s.Get(r.Context(), client.ObjectKey{Namespace: ns, Name: name}, device); err != nil {
|
||||
if errors.IsNotFound(err) {
|
||||
writeJSON(w, http.StatusNotFound, errResp("iot device not found"))
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusInternalServerError, errResp(err.Error()))
|
||||
return
|
||||
}
|
||||
|
||||
if err := h.K8s.Delete(r.Context(), device); err != nil {
|
||||
writeJSON(w, http.StatusInternalServerError, errResp(err.Error()))
|
||||
return
|
||||
}
|
||||
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
}
|
||||
|
||||
// UpdateIoTDevice — PATCH /v1/namespaces/{namespace}/iot/devices/{name}
|
||||
// Позволяет включить/отключить устройство (spec.enabled).
|
||||
// Контроллер увидит изменение и обновит status.phase.
|
||||
func (h *Handler) UpdateIoTDevice(w http.ResponseWriter, r *http.Request) {
|
||||
ns := namespace(r)
|
||||
name := pathVar(r, "name")
|
||||
|
||||
var req iotDeviceUpdateRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
writeJSON(w, http.StatusBadRequest, errResp("invalid JSON: "+err.Error()))
|
||||
return
|
||||
}
|
||||
if req.Enabled == nil {
|
||||
writeJSON(w, http.StatusBadRequest, errResp("enabled field is required"))
|
||||
return
|
||||
}
|
||||
|
||||
device := &iotv1alpha1.IoTDevice{}
|
||||
if err := h.K8s.Get(r.Context(), client.ObjectKey{Namespace: ns, Name: name}, device); err != nil {
|
||||
if errors.IsNotFound(err) {
|
||||
writeJSON(w, http.StatusNotFound, errResp("iot device not found"))
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusInternalServerError, errResp(err.Error()))
|
||||
return
|
||||
}
|
||||
|
||||
device.Spec.Enabled = *req.Enabled
|
||||
if err := h.K8s.Update(r.Context(), device); err != nil {
|
||||
writeJSON(w, http.StatusInternalServerError, errResp(err.Error()))
|
||||
return
|
||||
}
|
||||
|
||||
writeJSON(w, http.StatusOK, deviceToResponse(device, ""))
|
||||
}
|
||||
|
||||
// ——————————————————————————————————————————
|
||||
// MQTT ACL — авторизация pub/sub
|
||||
// ——————————————————————————————————————————
|
||||
|
||||
// mqttAclRequest — тело запроса от EMQX при каждом publish/subscribe.
|
||||
type mqttAclRequest struct {
|
||||
Username string `json:"username"`
|
||||
ClientID string `json:"clientid"`
|
||||
Action string `json:"action"` // "publish" | "subscribe"
|
||||
Topic string `json:"topic"`
|
||||
}
|
||||
|
||||
// MQTTAcl — POST /internal/mqtt/acl
|
||||
// Вызывается EMQX для каждого pub/sub действия.
|
||||
// НЕ защищён JWT — доступен только из кластера.
|
||||
//
|
||||
// Логика разрешений:
|
||||
// 1. Bridge clientid "sless-iot-bridge" — subscribe на любой топик (нужен для "+/telemetry/+")
|
||||
// 2. IoT Device (username "{ns}_{deviceId}") — publish/subscribe на "{ns}/telemetry/{deviceId}"
|
||||
func (h *Handler) MQTTAcl(w http.ResponseWriter, r *http.Request) {
|
||||
var req mqttAclRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
writeJSON(w, http.StatusOK, mqttAuthResponse{Result: "deny"})
|
||||
return
|
||||
}
|
||||
|
||||
// Специальный случай: mqtt-bridge подписывается на "+/telemetry/+" (все тенанты).
|
||||
// Публикация bridge НЕ разрешена — только чтение.
|
||||
if req.ClientID == "sless-iot-bridge" {
|
||||
if req.Action == "subscribe" {
|
||||
writeJSON(w, http.StatusOK, mqttAuthResponse{Result: "allow"})
|
||||
} else {
|
||||
writeJSON(w, http.StatusOK, mqttAuthResponse{Result: "deny"})
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// Парсим username → namespace + deviceId (формат: "{ns}_{deviceId}")
|
||||
// strings.Index находит ПЕРВЫЙ '_' — namespace содержит только дефисы
|
||||
idx := strings.Index(req.Username, "_")
|
||||
if idx < 0 {
|
||||
writeJSON(w, http.StatusOK, mqttAuthResponse{Result: "deny"})
|
||||
return
|
||||
}
|
||||
ns := req.Username[:idx]
|
||||
deviceID := req.Username[idx+1:]
|
||||
if ns == "" || deviceID == "" {
|
||||
writeJSON(w, http.StatusOK, mqttAuthResponse{Result: "deny"})
|
||||
return
|
||||
}
|
||||
|
||||
// Разрешённый топик: "{ns}/telemetry/{deviceId}"
|
||||
// Это то что эмулятор строит как: topicPrefix + "telemetry/" + device_id
|
||||
// topicPrefix = "{ns}/" → итого "{ns}/telemetry/{deviceId}"
|
||||
allowedTopic := ns + "/telemetry/" + deviceID
|
||||
if req.Topic == allowedTopic {
|
||||
writeJSON(w, http.StatusOK, mqttAuthResponse{Result: "allow"})
|
||||
return
|
||||
}
|
||||
|
||||
writeJSON(w, http.StatusOK, mqttAuthResponse{Result: "deny"})
|
||||
}
|
||||
@@ -0,0 +1,55 @@
|
||||
// Создано: 2026-04-05
|
||||
// iot_telemetry_handler.go — REST handler для чтения IoT телеметрии.
|
||||
//
|
||||
// Endpoint:
|
||||
// GET /v1/namespaces/{namespace}/iot/telemetry?device={id}&limit={n}
|
||||
//
|
||||
// Авторизация: Bearer JWT → namespace validation (как все /v1/ маршруты).
|
||||
// Данные берутся из per-tenant Postgres DB через IoTPostgresStore.
|
||||
// Если IoTPG не инициализирован (IOT_PG_DSN не задан) — возвращает 503.
|
||||
|
||||
package handler
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"strconv"
|
||||
|
||||
"github.com/gorilla/mux"
|
||||
|
||||
"gitea.services.ngcloud.ru/Nail/IoT/internal/storage/iotpg"
|
||||
)
|
||||
|
||||
// ListIoTTelemetry обрабатывает GET /v1/namespaces/{namespace}/iot/telemetry.
|
||||
// Параметры: device (опционально), limit (default 50, max 1000).
|
||||
func (h *Handler) ListIoTTelemetry(w http.ResponseWriter, r *http.Request) {
|
||||
if h.IoTPG == nil {
|
||||
writeJSON(w, http.StatusServiceUnavailable, errResp("IoT telemetry storage not configured"))
|
||||
return
|
||||
}
|
||||
|
||||
ns := mux.Vars(r)["namespace"]
|
||||
deviceID := r.URL.Query().Get("device")
|
||||
limit := 50
|
||||
if ls := r.URL.Query().Get("limit"); ls != "" {
|
||||
if n, err := strconv.Atoi(ls); err == nil && n > 0 {
|
||||
limit = n
|
||||
}
|
||||
}
|
||||
|
||||
rows, err := h.IoTPG.QueryTelemetry(r.Context(), ns, deviceID, limit)
|
||||
if err != nil {
|
||||
h.Log.Error("query IoT telemetry", "namespace", ns, "device", deviceID, "err", err)
|
||||
writeJSON(w, http.StatusInternalServerError, errResp("failed to query telemetry"))
|
||||
return
|
||||
}
|
||||
|
||||
// Возвращаем пустой массив вместо null — удобнее для JS
|
||||
if rows == nil {
|
||||
rows = []iotpg.TelemetryRow{}
|
||||
}
|
||||
|
||||
writeJSON(w, http.StatusOK, map[string]any{
|
||||
"items": rows,
|
||||
"count": len(rows),
|
||||
})
|
||||
}
|
||||
Reference in New Issue
Block a user