From 073683250c9287d0f8f31f023a0370cfb6ee9d5a Mon Sep 17 00:00:00 2001 From: Naeel Date: Thu, 9 Apr 2026 12:59:53 +0300 Subject: [PATCH] =?UTF-8?q?shared-sqs:=20=D0=AD=D1=82=D0=B0=D0=BF=D1=8B=20?= =?UTF-8?q?5+6=20=E2=80=94=20Admin=20API,=20auth=20middleware,=20graceful?= =?UTF-8?q?=20shutdown?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- shared-sqs/app/admin/admin.go | 211 ++++++++++++++++++++++++++++++++ shared-sqs/app/cmd/goaws.go | 183 +++++++++++++++++---------- shared-sqs/app/router/router.go | 37 ++++-- 3 files changed, 351 insertions(+), 80 deletions(-) create mode 100644 shared-sqs/app/admin/admin.go diff --git a/shared-sqs/app/admin/admin.go b/shared-sqs/app/admin/admin.go new file mode 100644 index 0000000..0457b1d --- /dev/null +++ b/shared-sqs/app/admin/admin.go @@ -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, +}) +} diff --git a/shared-sqs/app/cmd/goaws.go b/shared-sqs/app/cmd/goaws.go index ed1c67e..4029e5e 100644 --- a/shared-sqs/app/cmd/goaws.go +++ b/shared-sqs/app/cmd/goaws.go @@ -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") } diff --git a/shared-sqs/app/router/router.go b/shared-sqs/app/router/router.go index 39a611c..aa84645 100644 --- a/shared-sqs/app/router/router.go +++ b/shared-sqs/app/router/router.go @@ -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")