Nail/SQS-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(ha): leader-election via Redis lock + /ready endpoint (active/passive, off by default)

Browse Source 
- app/leader: Elector (SETNX lock + PEXPIRE heartbeat via Lua, IsLeader)
- app/persistence: TryAcquireLock/RefreshLock/ReleaseLock (no-op when Redis absent)
- app/router: /ready -> 200 if ReadyFunc nil/true, else 503 (both planes)
- app/cmd: --leader-election / SHARED_SQS_LEADER_ELECTION, --leader-ttl, --leader-interval;
  ReadyFunc wired only when enabled, lock released on graceful shutdown
- default OFF: single-instance behaviour unchanged
This commit is contained in:
Repinoid
2026-10-10 16:30:15 +03:00
parent 607aebb8ef
commit d7b3c653e0
4 changed files with 253 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/cmd/goaws.go
+45
View File
@@ -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()
app/leader/leader.go
+126
View File
@@ -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)
}
}
app/persistence/redis.go
+60
View File
@@ -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)
}
}
app/router/router.go
+22
View File
@@ -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.