// Создано: 2026-04-06 // Изменено: 2026-04-12 (замена Kafka lag → SQS queue stats) // 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 // - SQS: approximate message count (ApproximateNumberOfMessages) // - K8s: статус подов iot-mqtt-bridge и iot-sqs-consumer // // Авторизация: Bearer из env ADMIN_STATS_TOKEN. // Если ADMIN_STATS_TOKEN не задан — endpoint возвращает 503. package handler import ( "context" "fmt" "net/http" "os" "strconv" "strings" "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" sqstypes "github.com/aws/aws-sdk-go-v2/service/sqs/types" 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"` } // iotAdminSQSStats — информация о SQS очереди для мониторинга. type iotAdminSQSStats struct { ApproximateMessages int64 `json:"approximate_messages"` ApproximateMessagesNotVisible int64 `json:"approximate_messages_not_visible"` 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"} } // SQS: approximate message count для очереди iot-telemetry result["sqs"] = h.collectIotSQSStats(ctx) // K8s: статус подов bridge и consumer result["pods"] = h.collectIotPodStatuses(ctx) writeJSON(w, http.StatusOK, result) } // collectIotSQSStats получает ApproximateNumberOfMessages из SQS очереди. // Использует SQS_ENDPOINT, SQS_ACCESS_KEY, SQS_SECRET_KEY, SQS_QUEUE_NAME из env. func (h *Handler) collectIotSQSStats(ctx context.Context) iotAdminSQSStats { endpoint := os.Getenv("SQS_ENDPOINT") accessKey := os.Getenv("SQS_ACCESS_KEY") secretKey := os.Getenv("SQS_SECRET_KEY") queueName := os.Getenv("SQS_QUEUE_NAME") if queueName == "" { queueName = "iot-telemetry" } region := os.Getenv("SQS_REGION") if region == "" { region = "us-east-1" } if endpoint == "" || accessKey == "" || secretKey == "" { return iotAdminSQSStats{Error: "SQS credentials not configured (SQS_ENDPOINT, SQS_ACCESS_KEY, SQS_SECRET_KEY)"} } sqsClient := sqs.New(sqs.Options{ Region: region, Credentials: credentials.NewStaticCredentialsProvider( accessKey, secretKey, "", ), BaseEndpoint: aws.String(endpoint), }) // Получаем URL очереди queueUrlOut, err := sqsClient.GetQueueUrl(ctx, &sqs.GetQueueUrlInput{ QueueName: aws.String(queueName), }) if err != nil { return iotAdminSQSStats{Error: fmt.Sprintf("GetQueueUrl: %v", err)} } // Запрашиваем атрибуты очереди — approximate message counts attrsOut, err := sqsClient.GetQueueAttributes(ctx, &sqs.GetQueueAttributesInput{ QueueUrl: queueUrlOut.QueueUrl, AttributeNames: []sqstypes.QueueAttributeName{ sqstypes.QueueAttributeNameApproximateNumberOfMessages, sqstypes.QueueAttributeNameApproximateNumberOfMessagesNotVisible, }, }) if err != nil { return iotAdminSQSStats{Error: fmt.Sprintf("GetQueueAttributes: %v", err)} } approxMsg, _ := strconv.ParseInt(attrsOut.Attributes["ApproximateNumberOfMessages"], 10, 64) approxNotVisible, _ := strconv.ParseInt(attrsOut.Attributes["ApproximateNumberOfMessagesNotVisible"], 10, 64) return iotAdminSQSStats{ ApproximateMessages: approxMsg, ApproximateMessagesNotVisible: approxNotVisible, } } // 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-sqs-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)) } }