diff --git a/app/cmd/goaws.go b/app/cmd/goaws.go index 335f2ab..fda1bb6 100644 --- a/app/cmd/goaws.go +++ b/app/cmd/goaws.go @@ -7,6 +7,7 @@ import ( "context" "encoding/json" "flag" + "fmt" "net/http" "os" "os/signal" @@ -16,6 +17,7 @@ import ( "shared-sqs/app/billing" "shared-sqs/app/conf" "shared-sqs/app/gosqs" + "shared-sqs/app/leader" "shared-sqs/app/metrics" "shared-sqs/app/models" "shared-sqs/app/persistence" @@ -32,6 +34,9 @@ func main() { var controlPort string var debug bool var loglevel string + var leaderElection bool + var leaderTTL time.Duration + var leaderInterval time.Duration flag.StringVar(&configFile, "config", "", "config file location") flag.StringVar(&adminToken, "admin-token", "", "admin API bearer token") @@ -39,6 +44,11 @@ func main() { 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)") + // Leader-election (active/passive HA): выключено по умолчанию — поведение + // одиночного инстанса не меняется. Подробно — doc/thinking/state-extraction-and-ha-2026-10-10.md + flag.BoolVar(&leaderElection, "leader-election", false, "enable leader election via Redis lock (active/passive HA); off = always ready") + flag.DurationVar(&leaderTTL, "leader-ttl", 10*time.Second, "leader lock TTL (should be > 3x leader-interval)") + flag.DurationVar(&leaderInterval, "leader-interval", 3*time.Second, "leader lock refresh/acquire interval") flag.Parse() log.SetFormatter(&log.JSONFormatter{}) @@ -147,6 +157,35 @@ func main() { dataRouter := router.NewDataPlane(tenantStore) controlRouter := router.NewControlPlane(tenantStore, adminToken) + // Leader-election (active/passive, Шаг 3 — вариант C). + // Включается флагом --leader-election или SHARED_SQS_LEADER_ELECTION=true. + // Только лидер отдаёт /ready → 200; пассивный — 503 → LB/failover держит + // трафик на активном. Без флага ReadyFunc = nil (всегда готов) — одиночный + // инстанс работает ровно как раньше. + // + // Ограничение (осознанное, см. док): in-memory состояние НЕ шарится. Пока + // второй инстанс запущен «горячим», очереди живут в Redis, но изменения + // второго (пассивного) инстанса не увидит активный до рестарта. Поэтому + // на фазе C пассивный НЕ должен принимать SQS-трафик: nginx/LB обязан + // проверять /ready и не отправлять API-запросы на 503. + if !leaderElection && os.Getenv("SHARED_SQS_LEADER_ELECTION") == "true" { + leaderElection = true + } + var elector *leader.Elector + if leaderElection { + hostname, _ := os.Hostname() + if hostname == "" { + hostname = "unknown" + } + instanceID := fmt.Sprintf("%s-%d", hostname, os.Getpid()) + elector = leader.New("ssq:leader", instanceID, leaderTTL, leaderInterval) + elector.Start() + router.ReadyFunc = elector.IsLeader + log.Infof("leader-election включён (id=%s, ttl=%s, interval=%s)", instanceID, leaderTTL, leaderInterval) + } else { + log.Info("leader-election выключен — инстанс всегда ready (одиночный режим)") + } + // PeriodicTasks — visibility timeout, DLQ, deduplication quit := make(chan bool) go gosqs.PeriodicTasks(1*time.Second, quit) @@ -209,6 +248,12 @@ func main() { // Остановить PeriodicTasks close(quit) + // Отпустить лидерский лок (если были лидером) до закрытия слушателей, + // чтобы пассивный взял его быстрее, чем через TTL. + if elector != nil { + elector.Stop() + } + // Закрыть billing (если был подключён) billing.Close() diff --git a/app/leader/leader.go b/app/leader/leader.go new file mode 100644 index 0000000..cb2666c --- /dev/null +++ b/app/leader/leader.go @@ -0,0 +1,126 @@ +// app/leader/leader.go +// Простейший leader-election для active/passive (Шаг 3, вариант C). +// +// Лидер держит Redis-лок (ssq:leader) с TTL и продлевает его каждые interval. +// Пассивные периодически пытаются взять лок; когда лидер падает, TTL истекает и +// один из пассивных становится лидером. Готовность отдаётся через IsLeader(). +// +// Почему так, а не полный вынос состояния (вариант A): дешевле в разы и не требует +// переписывания ядра очередей; снимает главную боль «одна ВМ = SPOF». Потолок +// (горизонтальное масштабирование) при этом остаётся — см. +// doc/thinking/state-extraction-and-ha-2026-10-10.md. +// +// ВАЖНО: при memory-only (нет Redis) — считаем себя лидером (мы единственные). +package leader + +import ( + "sync" + "time" + + "shared-sqs/app/persistence" + + log "github.com/sirupsen/logrus" +) + +// Elector — выборы лидера через Redis-лок. +type Elector struct { + key string + id string + ttl time.Duration + interval time.Duration + + mu sync.RWMutex + isLeader bool + stopCh chan struct{} + stopped bool +} + +// New — создаёт Elector. key — ключ лока, id — уникальный id инстанса. +func New(key, id string, ttl, interval time.Duration) *Elector { + return &Elector{ + key: key, + id: id, + ttl: ttl, + interval: interval, + stopCh: make(chan struct{}), + } +} + +// Start — запускает цикл выборов в фоне (первая попытка — сразу). +func (e *Elector) Start() { + log.Infof("leader: старт выборов (key=%s, id=%s, ttl=%s)", e.key, e.id, e.ttl) + go e.loop() +} + +// IsLeader — текущий статус (готовность обслуживать). +func (e *Elector) IsLeader() bool { + e.mu.RLock() + defer e.mu.RUnlock() + return e.isLeader +} + +// Stop — останавливает цикл и освобождает лок (если были лидером). +func (e *Elector) Stop() { + e.mu.Lock() + if e.stopped { + e.mu.Unlock() + return + } + e.stopped = true + close(e.stopCh) + was := e.isLeader + e.mu.Unlock() + if was { + persistence.ReleaseLock(e.key, e.id) + } +} + +func (e *Elector) setLeader(v bool) { + e.mu.Lock() + changed := e.isLeader != v + e.isLeader = v + e.mu.Unlock() + if changed { + if v { + log.Info("leader: инстанс стал ЛИДЕРОМ (активный)") + } else { + log.Warn("leader: потерян лидерский лок (пассивный)") + } + } +} + +func (e *Elector) loop() { + t := time.NewTicker(e.interval) + defer t.Stop() + e.step() + for { + select { + case <-e.stopCh: + return + case <-t.C: + e.step() + } + } +} + +func (e *Elector) step() { + if e.IsLeader() { + ok, err := persistence.RefreshLock(e.key, e.id, e.ttl) + if err != nil { + log.Errorf("leader: refresh: %v", err) + return + } + if !ok { + e.setLeader(false) + } + return + } + ok, err := persistence.TryAcquireLock(e.key, e.id, e.ttl) + if err != nil { + log.Errorf("leader: acquire: %v", err) + return + } + if ok { + e.setLeader(true) + } +} diff --git a/app/persistence/redis.go b/app/persistence/redis.go index 8ae3c62..1e9edfe 100644 --- a/app/persistence/redis.go +++ b/app/persistence/redis.go @@ -653,3 +653,63 @@ func LoadAllTenantsRaw() (map[string][]byte, error) { log.Infof("persistence: загружено %d тенантов из Redis", len(result)) return result, nil } + +// ---- Leader election (active/passive) ---- +// +// Простейший HA-механизм (вариант C): один инстанс — активный (держит лок), +// остальные — пассивные. При падении активного лок истекает по TTL и пассивный +// его перехватывает (см. app/leader). Инстанс сообщает готовность через /ready. +// +// ВАЖНО: при Client == nil (memory-only) — считаем себя лидером (мы единственные). + +// TryAcquireLock — пытается взять лок (SETNX с TTL). true — лок наш. +func TryAcquireLock(key, value string, ttl time.Duration) (bool, error) { + if Client == nil { + return true, nil + } + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + return Client.SetNX(ctx, key, value, ttl).Result() +} + +// refreshLockScript — продлить TTL, ТОЛЬКО если лок всё ещё наш (атомарно). +var refreshLockScript = redis.NewScript(` +if redis.call("GET", KEYS[1]) == ARGV[1] then + return redis.call("PEXPIRE", KEYS[1], ARGV[2]) +else + return 0 +end`) + +// releaseLockScript — удалить лок, ТОЛЬКО если он наш. +var releaseLockScript = redis.NewScript(` +if redis.call("GET", KEYS[1]) == ARGV[1] then + return redis.call("DEL", KEYS[1]) +else + return 0 +end`) + +// RefreshLock — продлевает TTL лока, если он всё ещё наш. false — лок потерян. +func RefreshLock(key, value string, ttl time.Duration) (bool, error) { + if Client == nil { + return true, nil + } + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + n, err := refreshLockScript.Run(ctx, Client, []string{key}, value, ttl.Milliseconds()).Int() + if err != nil { + return false, err + } + return n == 1, nil +} + +// ReleaseLock — освобождает лок (если он наш). Вызывается при graceful shutdown. +func ReleaseLock(key, value string) { + if Client == nil { + return + } + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + if _, err := releaseLockScript.Run(ctx, Client, []string{key}, value).Int(); err != nil { + log.Errorf("persistence: release lock %q: %v", key, err) + } +} diff --git a/app/router/router.go b/app/router/router.go index 3b0fa08..876ed4e 100644 --- a/app/router/router.go +++ b/app/router/router.go @@ -44,6 +44,9 @@ func NewDataPlane(tenantStore *tenant.TenantStore) http.Handler { // /health — публичный, без auth r.HandleFunc("/health", health).Methods("GET") + // /ready — готовность инстанса (для failover/балансировщика) + r.HandleFunc("/ready", readyHandler).Methods("GET") + // /static — логотип, favicon для публичных страниц r.PathPrefix("/static").Handler(ui.StaticHandler()).Methods("GET") @@ -83,6 +86,9 @@ func NewControlPlane(tenantStore *tenant.TenantStore, adminToken string) http.Ha // /health — для проверки control-plane локально r.HandleFunc("/health", health).Methods("GET") + // /ready — готовность (лидер ли) + r.HandleFunc("/ready", readyHandler).Methods("GET") + // /static — нужен UI-консоли (логотип/favicon) r.PathPrefix("/static").Handler(ui.StaticHandler()).Methods("GET") @@ -205,6 +211,22 @@ func health(w http.ResponseWriter, req *http.Request) { fmt.Fprintf(w, `{"status":"ok","version":%q}`, models.Version) } +// ReadyFunc — если задана, /ready отдаёт 200 при true и 503 иначе (для failover/LB). +// Задаётся в main при включённом leader-election. nil → всегда готовы. +var ReadyFunc func() bool + +// readyHandler — готовность инстанса обслуживать трафик (для балансировщика/failover). +// Пассивный инстанс отдаёт 503, чтобы LB не направлял на него трафик. +func readyHandler(w http.ResponseWriter, req *http.Request) { + if ReadyFunc != nil && !ReadyFunc() { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"status":"ready"}`)) +} + // metricsAuth — оборачивает /metrics в проверку Bearer admin-токена. // В отличие от публичного /health, метрики содержат per-tenant данные // (accessKey, число очередей/сообщений) и не должны быть доступны без auth.