From 54d56690c660f2324ae93adb226ca039cf8608f8 Mon Sep 17 00:00:00 2001 From: Maksim Totmin Date: Thu, 25 Jun 2026 21:01:58 +0700 Subject: [PATCH] feat: eviction trunk cleanup + gateway auto-update on re-registration --- cmd/pulse-lets-go/main.go | 43 ++++++++++++++++++++++--- internal/engine/engine.go | 31 ++++++++++++------ internal/engine/engine_test.go | 57 ++++++++++++++++++++++++++++++++++ 3 files changed, 118 insertions(+), 13 deletions(-) diff --git a/cmd/pulse-lets-go/main.go b/cmd/pulse-lets-go/main.go index bef677a..b23e7ae 100644 --- a/cmd/pulse-lets-go/main.go +++ b/cmd/pulse-lets-go/main.go @@ -64,7 +64,7 @@ func main() { eng := engine.NewEngine(cfg) // 3.1 Запускаем эвикцию stale-нод (очистка нод без метрик > 5 минут, проверка каждые 60 сек) - eng.StartEvictionLoop(60*time.Second, 5*time.Minute, nil) + eng.StartEvictionLoop(60*time.Second, 5*time.Minute) // 4. ASCII логгер метрик metricsLogger, err := filelog.NewLogger(dataDir) @@ -81,16 +81,26 @@ func main() { var apiHandler *api.API - // Авто-создание balance-транков при первой метрике от новой ноды + // Авто-создание или обновление balance-транка при метрике от ноды sub.SetOnNewNode(func(nodeID, sipGateway string) { trunks, err := cfgMgr.ReadTrunks() if err != nil { log.Printf("[main] ошибка чтения транков для авто-создания: %v", err) return } - // Проверяем идемпотентность: транк уже существует? - for _, t := range trunks { + // Ищем существующий balance-транк для этой ноды + for i, t := range trunks { if t.Type == "balance" && t.NodeID == nodeID { + if t.Gateway != sipGateway { + trunks[i].Gateway = sipGateway + trunks[i].UpdatedAt = time.Now().UTC() + if err := cfgMgr.SaveTrunks(trunks); err != nil { + log.Printf("[main] ошибка обновления gateway транка для %s: %v", nodeID, err) + return + } + apiHandler.RebuildTrunkCache() + log.Printf("[nats] обновлён gateway транка для ноды %s → %s", nodeID, sipGateway) + } return } } @@ -121,6 +131,31 @@ func main() { // 6. HTTP API apiHandler = api.NewAPI(eng, cfgMgr, cfg.GetJWTSecret(), cfg.MonitoringAPIKey, sub.IsConnected, cfg) apiHandler.RebuildTrunkCache() // первичное построение in-memory кэша gateway + + // При эвикции stale-ноды — удаляем её balance-транк из trunks.json + eng.SetOnNodeEvicted(func(nodeID string) { + trunks, err := cfgMgr.ReadTrunks() + if err != nil { + log.Printf("[main] ошибка чтения транков для очистки evicted ноды %s: %v", nodeID, err) + return + } + filtered := make([]models.Trunk, 0, len(trunks)) + for _, t := range trunks { + if t.Type == models.TrunkBalance && t.NodeID == nodeID { + continue + } + filtered = append(filtered, t) + } + if len(filtered) == len(trunks) { + return + } + if err := cfgMgr.SaveTrunks(filtered); err != nil { + log.Printf("[main] ошибка сохранения транков после eviction %s: %v", nodeID, err) + return + } + apiHandler.RebuildTrunkCache() + log.Printf("[main] удалён balance-транк для evicted ноды %s", nodeID) + }) handler := apiHandler.Handler() // 6.1 ESL-клиент (опционально — если сконфигурирован) diff --git a/internal/engine/engine.go b/internal/engine/engine.go index f329d30..e31e53a 100644 --- a/internal/engine/engine.go +++ b/internal/engine/engine.go @@ -4,6 +4,7 @@ package engine import ( "fmt" + "log" "math/rand/v2" "sync" "sync/atomic" @@ -35,6 +36,7 @@ type Engine struct { // Управление eviction stale-нод evictionStopCh chan struct{} + onNodeEvicted func(nodeID string) // callback после удаления stale-ноды // Режим балансировки balanceMode string // "best" или "weighted_random" @@ -320,8 +322,7 @@ func (e *Engine) GetHealthStats() HealthStats { // StartEvictionLoop запускает периодическую очистку нод без метрик. // maxStale — время, после которого нода считается мёртвой (напр. 5 мин). -// stopCh — канал для остановки цикла (graceful shutdown). -func (e *Engine) StartEvictionLoop(interval, maxStale time.Duration, stopCh <-chan struct{}) { +func (e *Engine) StartEvictionLoop(interval, maxStale time.Duration) { e.evictionStopCh = make(chan struct{}) ticker := time.NewTicker(interval) @@ -330,9 +331,12 @@ func (e *Engine) StartEvictionLoop(interval, maxStale time.Duration, stopCh <-ch for { select { case <-ticker.C: - e.evictStaleNodes(maxStale) - case <-stopCh: - return + evicted := e.evictStaleNodes(maxStale) + for _, id := range evicted { + if e.onNodeEvicted != nil { + e.onNodeEvicted(id) + } + } case <-e.evictionStopCh: return } @@ -347,25 +351,34 @@ func (e *Engine) StopEvictionLoop() { } } +// SetOnNodeEvicted устанавливает callback, вызываемый при удалении stale-ноды. +// Callback выполняется вне engine-блокировки и может безопасно делать file I/O. +func (e *Engine) SetOnNodeEvicted(fn func(nodeID string)) { + e.onNodeEvicted = fn +} + // evictStaleNodes удаляет ноды, от которых нет метрик дольше maxStale. -func (e *Engine) evictStaleNodes(maxStale time.Duration) { +// Возвращает список ID удалённых нод. +func (e *Engine) evictStaleNodes(maxStale time.Duration) []string { e.mu.Lock() defer e.mu.Unlock() now := time.Now() - changed := false + var evicted []string for id, ns := range e.nodes { if now.Sub(time.Unix(ns.TS, 0)) > maxStale { delete(e.nodes, id) delete(e.rings, id) - changed = true + evicted = append(evicted, id) } } - if changed { + if len(evicted) > 0 { + log.Printf("[engine] eviction: удалены ноды %v (stale > %v)", evicted, maxStale) e.recalcBestLocked() } + return evicted } // --- Приватные методы --- diff --git a/internal/engine/engine_test.go b/internal/engine/engine_test.go index 22529ea..f423cf4 100644 --- a/internal/engine/engine_test.go +++ b/internal/engine/engine_test.go @@ -488,6 +488,63 @@ func TestEvictAllNodes_RecalcBest(t *testing.T) { } } +func TestEvictionReturnsEvictedIDs(t *testing.T) { + eng := NewEngine(defaultCfg()) + + m1 := freshMetric("pbx-01") + m1.TS = time.Now().Add(-10 * time.Minute).Unix() + eng.UpdateMetric(m1) + + m2 := freshMetric("pbx-02") + eng.UpdateMetric(m2) + + evicted := eng.evictStaleNodes(5 * time.Minute) + if len(evicted) != 1 { + t.Errorf("ожидался 1 evicted ID, получено %d: %v", len(evicted), evicted) + } + if evicted[0] != "pbx-01" { + t.Errorf("ожидался pbx-01, получен %s", evicted[0]) + } +} + +func TestSetOnNodeEvictedCallback(t *testing.T) { + eng := NewEngine(defaultCfg()) + + var evictedIDs []string + eng.SetOnNodeEvicted(func(nodeID string) { + evictedIDs = append(evictedIDs, nodeID) + }) + + m1 := freshMetric("pbx-01") + m1.TS = time.Now().Add(-10 * time.Minute).Unix() + eng.UpdateMetric(m1) + + m2 := freshMetric("pbx-02") + m2.TS = time.Now().Add(-8 * time.Minute).Unix() + eng.UpdateMetric(m2) + + m3 := freshMetric("pbx-03") + eng.UpdateMetric(m3) + + evicted := eng.evictStaleNodes(5 * time.Minute) + for _, id := range evicted { + if eng.onNodeEvicted != nil { + eng.onNodeEvicted(id) + } + } + + if len(evictedIDs) != 2 { + t.Errorf("ожидалось 2 callback вызова, получено %d: %v", len(evictedIDs), evictedIDs) + } + if evictedIDs[0] != "pbx-01" || evictedIDs[1] != "pbx-02" { + t.Errorf("неверный порядок evicted ID: %v", evictedIDs) + } + // pbx-03 должна остаться + if _, exists := eng.nodes["pbx-03"]; !exists { + t.Error("pbx-03 должна остаться после эвикции") + } +} + func cfgWithMode(mode string) *config.Config { cfg := defaultCfg() cfg.Scoring.LoadAvgMultiplier = 50.0