diff --git a/app/cmd/goaws.go b/app/cmd/goaws.go index f2d6282..a9b47c2 100644 --- a/app/cmd/goaws.go +++ b/app/cmd/goaws.go @@ -29,12 +29,14 @@ func main() { var configFile string var adminToken string var port string + var controlPort string var debug bool var loglevel string flag.StringVar(&configFile, "config", "", "config file location") flag.StringVar(&adminToken, "admin-token", "", "admin API bearer token") - flag.StringVar(&port, "port", "4100", "listen port") + flag.StringVar(&port, "port", "4100", "data-plane listen port") + flag.StringVar(&controlPort, "control-port", "4101", "control-plane listen port (localhost only: admin/metrics/ui)") flag.BoolVar(&debug, "debug", false, "set debug log level") flag.StringVar(&loglevel, "loglevel", "info", "log level (info, debug, warn, error)") flag.Parse() @@ -130,8 +132,10 @@ func main() { // Billing: подключение к PostgreSQL для учёта использования. // Если BILLING_PG_HOST не задан — billing отключён, SQS работает без него. billing.Init() - // Роутер с tenant auth и admin API - r := router.New(tenantStore, adminToken) + // Разбиение data-plane/control-plane (Шаг 2): + // data-plane — публичный SQS API; control-plane — админка/метрики/UI (только localhost). + dataRouter := router.NewDataPlane(tenantStore) + controlRouter := router.NewControlPlane(tenantStore, adminToken) // PeriodicTasks — visibility timeout, DLQ, deduplication quit := make(chan bool) @@ -148,20 +152,35 @@ func main() { // - IdleTimeout 90с — keep-alive; больше интервала ретраев клиентских SDK. // - MaxHeaderBytes 8192 — защита от гигантских заголовков // (платформа режет >16KB, наш лимит строже). - srv := &http.Server{ + dataSrv := &http.Server{ Addr: "0.0.0.0:" + port, - Handler: r, + Handler: dataRouter, + ReadHeaderTimeout: 10 * time.Second, + WriteTimeout: 35 * time.Second, + IdleTimeout: 90 * time.Second, + MaxHeaderBytes: 8192, + } + // control-plane слушает ТОЛЬКО localhost — админка/метрики/UI не должны быть публичными. + controlSrv := &http.Server{ + Addr: "127.0.0.1:" + controlPort, + Handler: controlRouter, ReadHeaderTimeout: 10 * time.Second, WriteTimeout: 35 * time.Second, IdleTimeout: 90 * time.Second, MaxHeaderBytes: 8192, } - // Запуск в горутине для graceful shutdown - serverErr := make(chan error, 1) + // Запуск в горутинах для graceful shutdown + serverErr := make(chan error, 2) go func() { - log.Infof("shared-sqs listening on 0.0.0.0:%s", port) - if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { + log.Infof("shared-sqs data-plane listening on 0.0.0.0:%s", port) + if err := dataSrv.ListenAndServe(); err != nil && err != http.ErrServerClosed { + serverErr <- err + } + }() + go func() { + log.Infof("shared-sqs control-plane listening on 127.0.0.1:%s", controlPort) + if err := controlSrv.ListenAndServe(); err != nil && err != http.ErrServerClosed { serverErr <- err } }() @@ -186,8 +205,11 @@ func main() { // Дать 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) + if err := dataSrv.Shutdown(ctx); err != nil { + log.Errorf("Data-plane shutdown error: %v", err) + } + if err := controlSrv.Shutdown(ctx); err != nil { + log.Errorf("Control-plane shutdown error: %v", err) } log.Info("shared-sqs stopped") } diff --git a/app/router/router.go b/app/router/router.go index 322622f..3b0fa08 100644 --- a/app/router/router.go +++ b/app/router/router.go @@ -28,8 +28,13 @@ import ( log "github.com/sirupsen/logrus" ) -// New — создаёт HTTP router с tenant auth и admin API -func New(tenantStore *tenant.TenantStore, adminToken string) http.Handler { +// NewDataPlane — публичный SQS API (data-plane): SQS actions, /health, статика, описание. +// НЕ содержит админ-консоль, /metrics и /ui — они в control-plane (NewControlPlane). +// +// Разбиение data-plane/control-plane (Шаг 2): публично доступен только этот роутер. +// Пока — два листенера в одном процессе (общее in-memory состояние); отдельные процессы +// станут возможны после выноса состояния в Redis (Шаг 3). +func NewDataPlane(tenantStore *tenant.TenantStore) http.Handler { r := mux.NewRouter() // Fix #3.5: глобальный лимит размера тела HTTP запроса — 3MB @@ -42,37 +47,11 @@ func New(tenantStore *tenant.TenantStore, adminToken string) http.Handler { // /static — логотип, favicon для публичных страниц r.PathPrefix("/static").Handler(ui.StaticHandler()).Methods("GET") - // /metrics — Prometheus endpoint для Victoria Metrics scrape. - // Защищён Bearer admin-токеном: метрики содержат per-tenant данные - // (accessKey/число очередей и сообщений), которые нельзя светить наружу. - r.Handle("/metrics", metricsAuth(adminToken, promhttp.Handler())).Methods("GET") - // Страница примеров команд — публично r.HandleFunc("/examples", func(w http.ResponseWriter, req *http.Request) { ui.ExamplesHandler(models.Version).ServeHTTP(w, req) }).Methods("GET") - // Админ-консоль — статика. Регистрируем ДО admin subrouter, - // иначе PathPrefix("/admin") с bearer-auth перехватит GET /admin. - r.HandleFunc("/admin", func(w http.ResponseWriter, req *http.Request) { - ui.AdminHandler(models.Version).ServeHTTP(w, req) - }).Methods("GET") - r.HandleFunc("/admin/", func(w http.ResponseWriter, req *http.Request) { - ui.AdminHandler(models.Version).ServeHTTP(w, req) - }).Methods("GET") - - // Admin API — Bearer token auth, регистрируется через AdminHandler - adminHandler := admin.NewHandler(tenantStore, adminToken) - adminHandler.RegisterRoutes(r) - - // UI public API — JWT auth, для встроенной console - // ВАЖНО: RegisterPublicRoutes ПЕРЕД static handler — иначе PathPrefix("/ui") перехватит /ui/api/* - adminHandler.RegisterPublicRoutes(r) - - // UI console — встроенный SPA, публичный доступ - // НЕ ловит /ui/api/* — mux сначала проверит более специфичные маршруты выше - r.PathPrefix("/ui").Handler(http.StripPrefix("/ui", ui.Handler())) - // SQS API — tenant auth middleware оборачивает каждый handler отдельно. // r.NewRoute().Subrouter() с Use() некорректно работает в gorilla/mux v1.8.0 // при пустом prefix — ответы теряются. Поэтому используем явную обёртку. @@ -94,6 +73,48 @@ func New(tenantStore *tenant.TenantStore, adminToken string) http.Handler { return r } +// NewControlPlane — админка, метрики и UI (control-plane). +// ВАЖНО: этот роутер должен слушать ТОЛЬКО localhost (не проксируется наружу). +func NewControlPlane(tenantStore *tenant.TenantStore, adminToken string) http.Handler { + r := mux.NewRouter() + + r.Use(requestBodyLimitMiddleware) + + // /health — для проверки control-plane локально + r.HandleFunc("/health", health).Methods("GET") + + // /static — нужен UI-консоли (логотип/favicon) + r.PathPrefix("/static").Handler(ui.StaticHandler()).Methods("GET") + + // /metrics — Prometheus endpoint для Victoria Metrics scrape. + // Защищён Bearer admin-токеном: метрики содержат per-tenant данные + // (accessKey/число очередей и сообщений), которые нельзя светить наружу. + r.Handle("/metrics", metricsAuth(adminToken, promhttp.Handler())).Methods("GET") + + // Админ-консоль — статика. Регистрируем ДО admin subrouter, + // иначе PathPrefix("/admin") с bearer-auth перехватит GET /admin. + r.HandleFunc("/admin", func(w http.ResponseWriter, req *http.Request) { + ui.AdminHandler(models.Version).ServeHTTP(w, req) + }).Methods("GET") + r.HandleFunc("/admin/", func(w http.ResponseWriter, req *http.Request) { + ui.AdminHandler(models.Version).ServeHTTP(w, req) + }).Methods("GET") + + // Admin API — Bearer token auth, регистрируется через AdminHandler + adminHandler := admin.NewHandler(tenantStore, adminToken) + adminHandler.RegisterRoutes(r) + + // UI public API — JWT auth, для встроенной console + // ВАЖНО: RegisterPublicRoutes ПЕРЕД static handler — иначе PathPrefix("/ui") перехватит /ui/api/* + adminHandler.RegisterPublicRoutes(r) + + // UI console — встроенный SPA + // НЕ ловит /ui/api/* — mux сначала проверит более специфичные маршруты выше + r.PathPrefix("/ui").Handler(http.StripPrefix("/ui", ui.Handler())) + + return r +} + // requestBodyLimitMiddleware — ограничивает размер тела HTTP запроса до 3MB. // Защита от OOM при отправке гигантских messages/batch (Fix #3.5). func requestBodyLimitMiddleware(next http.Handler) http.Handler {