feat(split): data-plane (публичный SQS API) и control-plane (админка/метрики/UI на localhost) — Шаг 2

This commit is contained in:
Repinoid
2026-10-10 16:00:57 +03:00
parent a66b908689
commit a26e6da8b7
2 changed files with 82 additions and 39 deletions
+33 -11
View File
@@ -29,12 +29,14 @@ func main() {
var configFile string var configFile string
var adminToken string var adminToken string
var port string var port string
var controlPort string
var debug bool var debug bool
var loglevel string var loglevel string
flag.StringVar(&configFile, "config", "", "config file location") flag.StringVar(&configFile, "config", "", "config file location")
flag.StringVar(&adminToken, "admin-token", "", "admin API bearer token") 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.BoolVar(&debug, "debug", false, "set debug log level")
flag.StringVar(&loglevel, "loglevel", "info", "log level (info, debug, warn, error)") flag.StringVar(&loglevel, "loglevel", "info", "log level (info, debug, warn, error)")
flag.Parse() flag.Parse()
@@ -130,8 +132,10 @@ func main() {
// Billing: подключение к PostgreSQL для учёта использования. // Billing: подключение к PostgreSQL для учёта использования.
// Если BILLING_PG_HOST не задан — billing отключён, SQS работает без него. // Если BILLING_PG_HOST не задан — billing отключён, SQS работает без него.
billing.Init() billing.Init()
// Роутер с tenant auth и admin API // Разбиение data-plane/control-plane (Шаг 2):
r := router.New(tenantStore, adminToken) // data-plane — публичный SQS API; control-plane — админка/метрики/UI (только localhost).
dataRouter := router.NewDataPlane(tenantStore)
controlRouter := router.NewControlPlane(tenantStore, adminToken)
// PeriodicTasks — visibility timeout, DLQ, deduplication // PeriodicTasks — visibility timeout, DLQ, deduplication
quit := make(chan bool) quit := make(chan bool)
@@ -148,20 +152,35 @@ func main() {
// - IdleTimeout 90с — keep-alive; больше интервала ретраев клиентских SDK. // - IdleTimeout 90с — keep-alive; больше интервала ретраев клиентских SDK.
// - MaxHeaderBytes 8192 — защита от гигантских заголовков // - MaxHeaderBytes 8192 — защита от гигантских заголовков
// (платформа режет >16KB, наш лимит строже). // (платформа режет >16KB, наш лимит строже).
srv := &http.Server{ dataSrv := &http.Server{
Addr: "0.0.0.0:" + port, 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, ReadHeaderTimeout: 10 * time.Second,
WriteTimeout: 35 * time.Second, WriteTimeout: 35 * time.Second,
IdleTimeout: 90 * time.Second, IdleTimeout: 90 * time.Second,
MaxHeaderBytes: 8192, MaxHeaderBytes: 8192,
} }
// Запуск в горутине для graceful shutdown // Запуск в горутинах для graceful shutdown
serverErr := make(chan error, 1) serverErr := make(chan error, 2)
go func() { go func() {
log.Infof("shared-sqs listening on 0.0.0.0:%s", port) log.Infof("shared-sqs data-plane listening on 0.0.0.0:%s", port)
if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { 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 serverErr <- err
} }
}() }()
@@ -186,8 +205,11 @@ func main() {
// Дать 10 секунд на завершение текущих HTTP запросов // Дать 10 секунд на завершение текущих HTTP запросов
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel() defer cancel()
if err := srv.Shutdown(ctx); err != nil { if err := dataSrv.Shutdown(ctx); err != nil {
log.Errorf("Server shutdown error: %v", err) 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") log.Info("shared-sqs stopped")
} }
+49 -28
View File
@@ -28,8 +28,13 @@ import (
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
) )
// New — создаёт HTTP router с tenant auth и admin API // NewDataPlane — публичный SQS API (data-plane): SQS actions, /health, статика, описание.
func New(tenantStore *tenant.TenantStore, adminToken string) http.Handler { // НЕ содержит админ-консоль, /metrics и /ui — они в control-plane (NewControlPlane).
//
// Разбиение data-plane/control-plane (Шаг 2): публично доступен только этот роутер.
// Пока — два листенера в одном процессе (общее in-memory состояние); отдельные процессы
// станут возможны после выноса состояния в Redis (Шаг 3).
func NewDataPlane(tenantStore *tenant.TenantStore) http.Handler {
r := mux.NewRouter() r := mux.NewRouter()
// Fix #3.5: глобальный лимит размера тела HTTP запроса — 3MB // Fix #3.5: глобальный лимит размера тела HTTP запроса — 3MB
@@ -42,37 +47,11 @@ func New(tenantStore *tenant.TenantStore, adminToken string) http.Handler {
// /static — логотип, favicon для публичных страниц // /static — логотип, favicon для публичных страниц
r.PathPrefix("/static").Handler(ui.StaticHandler()).Methods("GET") 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) { r.HandleFunc("/examples", func(w http.ResponseWriter, req *http.Request) {
ui.ExamplesHandler(models.Version).ServeHTTP(w, req) ui.ExamplesHandler(models.Version).ServeHTTP(w, req)
}).Methods("GET") }).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 отдельно. // SQS API — tenant auth middleware оборачивает каждый handler отдельно.
// r.NewRoute().Subrouter() с Use() некорректно работает в gorilla/mux v1.8.0 // r.NewRoute().Subrouter() с Use() некорректно работает в gorilla/mux v1.8.0
// при пустом prefix — ответы теряются. Поэтому используем явную обёртку. // при пустом prefix — ответы теряются. Поэтому используем явную обёртку.
@@ -94,6 +73,48 @@ func New(tenantStore *tenant.TenantStore, adminToken string) http.Handler {
return r 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. // requestBodyLimitMiddleware — ограничивает размер тела HTTP запроса до 3MB.
// Защита от OOM при отправке гигантских messages/batch (Fix #3.5). // Защита от OOM при отправке гигантских messages/batch (Fix #3.5).
func requestBodyLimitMiddleware(next http.Handler) http.Handler { func requestBodyLimitMiddleware(next http.Handler) http.Handler {