shared-sqs: Этапы 5+6 — Admin API, auth middleware, graceful shutdown

2026-04-09 12:59:53 +03:00
parent 08053ca8e5
commit 073683250c
3 changed files with 351 additions and 80 deletions
@@ -0,0 +1,211 @@
+// app/admin/admin.go
+// Admin API handlers for shared-sqs management
+// Created: 2026-04-09
+package admin
+
+import (
+"encoding/json"
+"net/http"
+"strings"
+"time"
+
+"shared-sqs/app/models"
+"shared-sqs/app/tenant"
+
+"github.com/gorilla/mux"
+log "github.com/sirupsen/logrus"
+)
+
+// Handler — admin API handler, holds TenantStore и admin token
+type Handler struct {
+store      *tenant.TenantStore
+adminToken string
+}
+
+// NewHandler — создаёт admin handler
+func NewHandler(store *tenant.TenantStore, adminToken string) *Handler {
+return &Handler{store: store, adminToken: adminToken}
+}
+
+// RegisterRoutes — регистрирует admin маршруты на переданном router
+func (h *Handler) RegisterRoutes(r *mux.Router) {
+adminRouter := r.PathPrefix("/admin").Subrouter()
+adminRouter.Use(h.bearerAuthMiddleware)
+adminRouter.HandleFunc("/tenants", h.createTenant).Methods("POST")
+adminRouter.HandleFunc("/tenants", h.listTenants).Methods("GET")
+adminRouter.HandleFunc("/tenants/{id}", h.getTenant).Methods("GET")
+adminRouter.HandleFunc("/tenants/{id}", h.deleteTenant).Methods("DELETE")
+adminRouter.HandleFunc("/health", h.detailedHealth).Methods("GET")
+}
+
+// bearerAuthMiddleware — проверяет Bearer token для admin API (Trap #12)
+func (h *Handler) bearerAuthMiddleware(next http.Handler) http.Handler {
+return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+authHeader := r.Header.Get("Authorization")
+expected := "Bearer " + h.adminToken
+if authHeader != expected {
+w.Header().Set("Content-Type", "application/json")
+w.WriteHeader(http.StatusUnauthorized)
+json.NewEncoder(w).Encode(map[string]string{"error": "unauthorized"})
+return
+}
+next.ServeHTTP(w, r)
+})
+}
+
+// createTenantRequest — тело запроса POST /admin/tenants
+type createTenantRequest struct {
+Name      string `json:"name"`
+MaxQueues int    `json:"max_queues"`
+}
+
+// tenantCreateResponse — ответ с secret_key (показывается ТОЛЬКО при создании)
+type tenantCreateResponse struct {
+ID        string    `json:"id"`
+Name      string    `json:"name"`
+AccessKey string    `json:"access_key"`
+SecretKey string    `json:"secret_key"`
+MaxQueues int       `json:"max_queues"`
+CreatedAt time.Time `json:"created_at"`
+Active    bool      `json:"active"`
+}
+
+// tenantListItem — данные тенанта без secret_key (для List/Get)
+type tenantListItem struct {
+ID        string    `json:"id"`
+Name      string    `json:"name"`
+AccessKey string    `json:"access_key"`
+MaxQueues int       `json:"max_queues"`
+CreatedAt time.Time `json:"created_at"`
+Active    bool      `json:"active"`
+}
+
+// createTenant — POST /admin/tenants
+func (h *Handler) createTenant(w http.ResponseWriter, r *http.Request) {
+var req createTenantRequest
+if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+w.Header().Set("Content-Type", "application/json")
+w.WriteHeader(http.StatusBadRequest)
+json.NewEncoder(w).Encode(map[string]string{"error": "invalid request body"})
+return
+}
+if req.Name == "" {
+w.Header().Set("Content-Type", "application/json")
+w.WriteHeader(http.StatusBadRequest)
+json.NewEncoder(w).Encode(map[string]string{"error": "name is required"})
+return
+}
+t, err := h.store.Create(req.Name, req.MaxQueues)
+if err != nil {
+log.Errorf("admin: failed to create tenant: %v", err)
+w.Header().Set("Content-Type", "application/json")
+w.WriteHeader(http.StatusInternalServerError)
+json.NewEncoder(w).Encode(map[string]string{"error": "failed to create tenant"})
+return
+}
+w.Header().Set("Content-Type", "application/json")
+w.WriteHeader(http.StatusCreated)
+json.NewEncoder(w).Encode(tenantCreateResponse{
+ID:        t.ID,
+Name:      t.Name,
+AccessKey: t.AccessKey,
+SecretKey: t.SecretKey,
+MaxQueues: t.MaxQueues,
+CreatedAt: t.CreatedAt,
+Active:    t.Active,
+})
+}
+
+// listTenants — GET /admin/tenants
+func (h *Handler) listTenants(w http.ResponseWriter, r *http.Request) {
+tenants := h.store.List()
+items := make([]tenantListItem, 0, len(tenants))
+for _, t := range tenants {
+items = append(items, tenantListItem{
+ID:        t.ID,
+Name:      t.Name,
+AccessKey: t.AccessKey,
+MaxQueues: t.MaxQueues,
+CreatedAt: t.CreatedAt,
+Active:    t.Active,
+})
+}
+w.Header().Set("Content-Type", "application/json")
+json.NewEncoder(w).Encode(items)
+}
+
+// getTenant — GET /admin/tenants/{id}
+func (h *Handler) getTenant(w http.ResponseWriter, r *http.Request) {
+vars := mux.Vars(r)
+id := vars["id"]
+t, ok := h.store.GetByID(id)
+if !ok {
+w.Header().Set("Content-Type", "application/json")
+w.WriteHeader(http.StatusNotFound)
+json.NewEncoder(w).Encode(map[string]string{"error": "tenant not found"})
+return
+}
+w.Header().Set("Content-Type", "application/json")
+json.NewEncoder(w).Encode(tenantListItem{
+ID:        t.ID,
+Name:      t.Name,
+AccessKey: t.AccessKey,
+MaxQueues: t.MaxQueues,
+CreatedAt: t.CreatedAt,
+Active:    t.Active,
+})
+}
+
+// deleteTenant — DELETE /admin/tenants/{id}
+// Удаляет тенанта И ВСЕ его очереди из SyncQueues (Trap #11: иначе memory leak)
+func (h *Handler) deleteTenant(w http.ResponseWriter, r *http.Request) {
+vars := mux.Vars(r)
+id := vars["id"]
+t, ok := h.store.GetByID(id)
+if !ok {
+w.Header().Set("Content-Type", "application/json")
+w.WriteHeader(http.StatusNotFound)
+json.NewEncoder(w).Encode(map[string]string{"error": "tenant not found"})
+return
+}
+// Удаляем все очереди тенанта из SyncQueues
+prefix := t.AccessKey + ":"
+models.SyncQueues.Lock()
+for key := range models.SyncQueues.Queues {
+if strings.HasPrefix(key, prefix) {
+delete(models.SyncQueues.Queues, key)
+}
+}
+models.SyncQueues.Unlock()
+
+h.store.Delete(id)
+w.WriteHeader(http.StatusNoContent)
+}
+
+// adminHealthDetail — ответ GET /admin/health
+type adminHealthDetail struct {
+Status       string `json:"status"`
+TenantCount  int    `json:"tenant_count"`
+QueueCount   int    `json:"queue_count"`
+MessageCount int    `json:"message_count"`
+}
+
+// detailedHealth — GET /admin/health
+func (h *Handler) detailedHealth(w http.ResponseWriter, r *http.Request) {
+tenants := h.store.List()
+models.SyncQueues.RLock()
+queueCount := len(models.SyncQueues.Queues)
+msgCount := 0
+for _, q := range models.SyncQueues.Queues {
+msgCount += len(q.Messages)
+}
+models.SyncQueues.RUnlock()
+
+w.Header().Set("Content-Type", "application/json")
+json.NewEncoder(w).Encode(adminHealthDetail{
+Status:       "ok",
+TenantCount:  len(tenants),
+QueueCount:   queueCount,
+MessageCount: msgCount,
+})
+}
@@ -1,80 +1,127 @@
+// app/cmd/goaws.go
+// Entry point — shared-sqs server
+// Updated: 2026-04-09 — добавлены TenantStore, admin token, graceful shutdown (Trap #13)
 package main

 import (
-	"flag"
-	"net/http"
-	"os"
-	"time"
+"context"
+"flag"
+"net/http"
+"os"
+"os/signal"
+"syscall"
+"time"

-	"shared-sqs/app/models"
+"shared-sqs/app/conf"
+"shared-sqs/app/gosqs"
+"shared-sqs/app/models"
+"shared-sqs/app/router"
+"shared-sqs/app/tenant"

-	log "github.com/sirupsen/logrus"
-
-	"shared-sqs/app/conf"
-	"shared-sqs/app/gosqs"
-	"shared-sqs/app/router"
+log "github.com/sirupsen/logrus"
 )

 func main() {
-	var filename string
-	var debug bool
-	var loglevel string
-	flag.StringVar(&filename, "config", "", "config file location + name")
-	flag.BoolVar(&debug, "debug", false, "set debug log level")
-	flag.StringVar(&loglevel, "loglevel", "info", "log level (default info)")
-	flag.Parse()
+var configFile string
+var adminToken string
+var port string
+var debug bool
+var loglevel string

-	log.SetFormatter(&log.JSONFormatter{})
-	log.SetOutput(os.Stdout)
+flag.StringVar(&configFile, "config", "", "config file location")
+flag.StringVar(&adminToken, "admin-token", "", "admin API bearer token")
+flag.StringVar(&port, "port", "4100", "listen port")
+flag.BoolVar(&debug, "debug", false, "set debug log level")
+flag.StringVar(&loglevel, "loglevel", "info", "log level (info, debug, warn, error)")
+flag.Parse()

-	if debug {
-		log.SetLevel(log.DebugLevel)
-	} else {
-		level, err := log.ParseLevel(loglevel)
-		if err != nil {
-			log.SetLevel(log.InfoLevel)
-			log.Warnf("Failed to parse loglevel %v, defaulting to info", loglevel)
-		} else {
-			log.SetLevel(level)
-		}
-	}
+log.SetFormatter(&log.JSONFormatter{})
+log.SetOutput(os.Stdout)

-	env := "Local"
-	if flag.NArg() > 0 {
-		env = flag.Arg(0)
-	}
-
-	portNumbers := conf.LoadYamlConfig(filename, env)
-
-	if models.CurrentEnvironment.LogToFile {
-		filename := models.CurrentEnvironment.LogFile
-		file, err := os.OpenFile(filename, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0666)
-		if err == nil {
-			log.SetOutput(file)
-		} else {
-			log.Infof("Failed to log to file: %s, using default stderr", filename)
-		}
-	}
-
-	r := router.New()
-
-	quit := make(chan bool, 0)
-	go gosqs.PeriodicTasks(1*time.Second, quit)
-
-	if len(portNumbers) == 1 {
-		log.Warnf("GoAws listening on: 0.0.0.0:%s", portNumbers[0])
-		err := http.ListenAndServe("0.0.0.0:"+portNumbers[0], r)
-		log.Fatal(err)
-	} else if len(portNumbers) == 2 {
-		go func() {
-			log.Warnf("GoAws listening on: 0.0.0.0:%s", portNumbers[0])
-			err := http.ListenAndServe("0.0.0.0:"+portNumbers[0], r)
-			log.Fatal(err)
-		}()
-		log.Warnf("GoAws listening on: 0.0.0.0:%s", portNumbers[1])
-		err := http.ListenAndServe("0.0.0.0:"+portNumbers[1], r)
-		log.Fatal(err)
-	} else {
-		log.Fatal("Not enough or too many ports defined to start GoAws.")
-	}
+if debug {
+log.SetLevel(log.DebugLevel)
+} else {
+level, err := log.ParseLevel(loglevel)
+if err != nil {
+log.SetLevel(log.InfoLevel)
+log.Warnf("Failed to parse loglevel %v, defaulting to info", loglevel)
+} else {
+log.SetLevel(level)
+}
+}
+
+// Admin token: flag > env SHARED_SQS_ADMIN_TOKEN > fatal (Trap #13)
+if adminToken == "" {
+adminToken = os.Getenv("SHARED_SQS_ADMIN_TOKEN")
+}
+if adminToken == "" {
+log.Fatal("admin token required: use --admin-token flag or SHARED_SQS_ADMIN_TOKEN env var")
+}
+
+// Загрузить конфиг (очереди, env — без SNS)
+env := "Local"
+if flag.NArg() > 0 {
+env = flag.Arg(0)
+}
+conf.LoadYamlConfig(configFile, env)
+
+if models.CurrentEnvironment.LogToFile {
+filename := models.CurrentEnvironment.LogFile
+file, err := os.OpenFile(filename, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0666)
+if err == nil {
+log.SetOutput(file)
+} else {
+log.Infof("Failed to log to file: %s, using default stdout", filename)
+}
+}
+
+// Инициализация in-memory TenantStore
+tenantStore := tenant.NewTenantStore()
+
+// Роутер с tenant auth и admin API
+r := router.New(tenantStore, adminToken)
+
+// PeriodicTasks — visibility timeout, DLQ, deduplication
+quit := make(chan bool)
+go gosqs.PeriodicTasks(1*time.Second, quit)
+
+// HTTP сервер с таймаутами
+srv := &http.Server{
+Addr:         "0.0.0.0:" + port,
+Handler:      r,
+ReadTimeout:  30 * time.Second,
+WriteTimeout: 35 * time.Second, // чуть больше чем max WaitTimeSeconds (20s)
+IdleTimeout:  60 * time.Second,
+}
+
+// Запуск в горутине для graceful shutdown
+serverErr := make(chan error, 1)
+go func() {
+log.Infof("shared-sqs listening on 0.0.0.0:%s", port)
+if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+serverErr <- err
+}
+}()
+
+// Graceful shutdown по SIGTERM/SIGINT (Trap #13)
+sigCh := make(chan os.Signal, 1)
+signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT)
+
+select {
+case sig := <-sigCh:
+log.Infof("Received signal %s, shutting down", sig)
+case err := <-serverErr:
+log.Fatalf("Server error: %v", err)
+}
+
+// Остановить PeriodicTasks
+close(quit)
+
+// Дать 10 секунд на завершение текущих HTTP запросов
+ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+defer cancel()
+if err := srv.Shutdown(ctx); err != nil {
+log.Errorf("Server shutdown error: %v", err)
+}
+log.Info("shared-sqs stopped")
 }
@@ -1,3 +1,6 @@
+// app/router/router.go
+// HTTP router для shared-sqs
+// Updated: 2026-04-09 — добавлены TenantStore, admin API, auth middleware
 package router

 import (
@@ -8,23 +11,34 @@ import (
 "net/http"
 "strings"

+"shared-sqs/app/admin"
+"shared-sqs/app/auth"
 "shared-sqs/app/interfaces"
-
-log "github.com/sirupsen/logrus"
-
 sqs "shared-sqs/app/gosqs"
+"shared-sqs/app/tenant"
+
 "github.com/gorilla/mux"
+log "github.com/sirupsen/logrus"
 )

-// New returns a new router
-func New() http.Handler {
+// New — создаёт HTTP router с tenant auth и admin API
+func New(tenantStore *tenant.TenantStore, adminToken string) http.Handler {
 r := mux.NewRouter()

-r.HandleFunc("/", actionHandler).Methods("GET", "POST")
+// /health — публичный, без auth
 r.HandleFunc("/health", health).Methods("GET")
-r.HandleFunc("/{account}", actionHandler).Methods("GET", "POST")
-r.HandleFunc("/queue/{queueName}", actionHandler).Methods("GET", "POST")
-r.HandleFunc("/{account}/{queueName}", actionHandler).Methods("GET", "POST")
+
+// Admin API — Bearer token auth, регистрируется через AdminHandler
+adminHandler := admin.NewHandler(tenantStore, adminToken)
+adminHandler.RegisterRoutes(r)
+
+// SQS API — tenant auth middleware
+sqsRouter := r.NewRoute().Subrouter()
+sqsRouter.Use(auth.AuthMiddleware(tenantStore))
+sqsRouter.HandleFunc("/", actionHandler).Methods("GET", "POST")
+sqsRouter.HandleFunc("/{account}", actionHandler).Methods("GET", "POST")
+sqsRouter.HandleFunc("/queue/{queueName}", actionHandler).Methods("GET", "POST")
+sqsRouter.HandleFunc("/{account}/{queueName}", actionHandler).Methods("GET", "POST")

 return r
 }
@@ -56,7 +70,7 @@ _, _ = w.Write(result)
 }
 }

-// routingTableV1 содержит только SQS actions — SNS удалён
+// routingTableV1 — только SQS actions (SNS удалён)
 var routingTableV1 = map[string]func(r *http.Request) (int, interfaces.AbstractResponseBody){
 "CreateQueue":             sqs.CreateQueueV1,
 "ListQueues":              sqs.ListQueuesV1,
@@ -80,8 +94,7 @@ fmt.Fprint(w, "OK")

 func actionHandler(w http.ResponseWriter, req *http.Request) {
 action := extractAction(req)
-log.WithFields(
-log.Fields{
+log.WithFields(log.Fields{
 "action": action,
 "url":    req.URL,
 }).Debug("Handling URL request")