// Package engine содержит scoring engine и router выбора лучшей ноды. // Потокобезопасен: RWMutex разделяет чтение (роутинг) и запись (обновление метрик). package engine import ( "fmt" "log" "math/rand/v2" "sync" "sync/atomic" "time" "github.com/pulse-lets-go/internal/config" "github.com/pulse-lets-go/internal/models" ) const defaultRingCapacity = 360 // 30 минут истории при метриках каждые 5 сек // Engine управляет состоянием нод и вычислением скора. type Engine struct { mu sync.RWMutex nodes map[string]*models.NodeState rings map[string]*MetricsRingBuffer // история метрик на каждую ноду cfg *config.Config // Кэш для O(1) доступа из /api/route bestNodeID string bestScore float64 fallbackActive bool // true, если все ноды имеют score < 0 // Статистика для health endpoint startTime time.Time routeRequests atomic.Int64 routeFallbacks atomic.Int64 lastMetricTime atomic.Int64 // unix ts последней полученной метрики // Управление eviction stale-нод evictionStopCh chan struct{} onNodeEvicted func(nodeID string) // callback после удаления stale-ноды // Режим балансировки balanceMode string // "best" или "weighted_random" } // NewEngine создаёт новый engine с заданной конфигурацией. func NewEngine(cfg *config.Config) *Engine { mode := cfg.BalanceMode if mode == "" { mode = "weighted_random" } return &Engine{ nodes: make(map[string]*models.NodeState), rings: make(map[string]*MetricsRingBuffer), cfg: cfg, bestScore: -1, startTime: time.Now(), balanceMode: mode, } } // UpdateConfig обновляет конфигурацию скоринга (при изменении через API). func (e *Engine) UpdateConfig(cfg *config.Config) { e.mu.Lock() defer e.mu.Unlock() e.cfg = cfg e.recalculateAll() } // UpdateMetric принимает метрику из NATS, обновляет состояние ноды и пересчитывает score. // disabled-флаг НЕ трогается — он управляется только через ToggleNode. func (e *Engine) UpdateMetric(m *models.NodeMetric) *models.NodeState { e.mu.Lock() defer e.mu.Unlock() ns, exists := e.nodes[m.NodeID] firstMetric := !exists if !exists { ns = &models.NodeState{NodeMetric: *m} e.nodes[m.NodeID] = ns } else { // Сохраняем admin-флаги disabled := ns.Disabled disabledReason := ns.DisabledReason ns.NodeMetric = *m ns.Disabled = disabled ns.DisabledReason = disabledReason } e.scoreNodeLocked(ns, firstMetric) e.recalcBestLocked() // Обновляем время последней метрики e.lastMetricTime.Store(m.TS) // Добавляем снапшот в кольцевой буфер истории ring, ok := e.rings[m.NodeID] if !ok { ring = newMetricsRingBuffer(defaultRingCapacity) e.rings[m.NodeID] = ring } ring.push(NodeSnapshot{ ActiveCalls: ns.ActiveCalls, MaxCalls: ns.MaxCalls, IdleCPU: ns.IdleCPU, LoadAvg: ns.LoadAvg, CallFailureRate: ns.CallFailureRate, Score: ns.Score, TS: ns.TS, }) return ns } // GetBestNode возвращает ID лучшей ноды и fallback-статус. // O(1) — читает кэш, не сканирует все ноды. func (e *Engine) GetBestNode() (nodeID string, score float64, fallback bool) { e.mu.RLock() defer e.mu.RUnlock() return e.pickBestNodeLocked() } // PickNode — единый фасад маршрутизации вызова. // Выбирает ноду согласно активной стратегии балансировки (best / weighted_random). func (e *Engine) PickNode() (nodeID string, score float64, fallback bool) { e.mu.RLock() defer e.mu.RUnlock() switch e.balanceMode { case "weighted_random": return e.pickWeightedRandomLocked() default: return e.pickBestNodeLocked() } } // pickBestNodeLocked — O(1) из кэша (вызывающий держит RLock). func (e *Engine) pickBestNodeLocked() (nodeID string, score float64, fallback bool) { if len(e.nodes) == 0 { return "", 0, true } return e.bestNodeID, e.bestScore, e.fallbackActive } // pickWeightedRandomLocked — O(n) weighted random выбор (вызывающий держит RLock). func (e *Engine) pickWeightedRandomLocked() (nodeID string, score float64, fallback bool) { if len(e.nodes) == 0 { return "", 0, true } now := time.Now().Unix() staleThreshold := int64(e.cfg.StaleThresholdSec) type candidate struct { id string score float64 } candidates := make([]candidate, 0, len(e.nodes)) var totalScore float64 for id, ns := range e.nodes { if ns.Disabled { continue } if ns.Score < 0 { continue } if now-ns.TS > staleThreshold { continue } candidates = append(candidates, candidate{id: id, score: ns.Score}) totalScore += ns.Score } if len(candidates) == 0 { return e.bestNodeID, e.bestScore, true } if totalScore <= 0 { idx := rand.IntN(len(candidates)) return candidates[idx].id, 0, false } pick := rand.Float64() * totalScore for _, c := range candidates { pick -= c.score if pick <= 0 { return c.id, c.score, false } } return candidates[len(candidates)-1].id, candidates[len(candidates)-1].score, false } // GetBalancingMode возвращает текущий режим балансировки. func (e *Engine) GetBalancingMode() string { e.mu.RLock() defer e.mu.RUnlock() return e.balanceMode } // SetBalancingMode переключает режим балансировки (best / weighted_random). func (e *Engine) SetBalancingMode(mode string) { e.mu.Lock() defer e.mu.Unlock() e.balanceMode = mode } // GetNodeInfo возвращает информацию о конкретной ноде. func (e *Engine) GetNodeInfo(nodeID string) (*models.NodeInfo, bool) { e.mu.RLock() defer e.mu.RUnlock() ns, exists := e.nodes[nodeID] if !exists { return nil, false } return stateToNodeInfo(ns, e.cfg.StaleThresholdSec, time.Now()), true } // GetAllNodes возвращает массив NodeInfo для всех нод. func (e *Engine) GetAllNodes() []*models.NodeInfo { e.mu.RLock() defer e.mu.RUnlock() now := time.Now() result := make([]*models.NodeInfo, 0, len(e.nodes)) for _, ns := range e.nodes { result = append(result, stateToNodeInfo(ns, e.cfg.StaleThresholdSec, now)) } return result } // GetNodeHistory возвращает историю метрик ноды из ring buffer. func (e *Engine) GetNodeHistory(nodeID string) []NodeSnapshot { e.mu.RLock() defer e.mu.RUnlock() ring, ok := e.rings[nodeID] if !ok { return nil } return ring.snapshot() } // HasNode возвращает true, если нода уже зарегистрирована в engine. func (e *Engine) HasNode(nodeID string) bool { e.mu.RLock() defer e.mu.RUnlock() _, exists := e.nodes[nodeID] return exists } // ToggleNode включает/выключает ноду из распределения. func (e *Engine) ToggleNode(nodeID string, disabled bool, reason string) error { e.mu.Lock() defer e.mu.Unlock() ns, exists := e.nodes[nodeID] if !exists { return fmt.Errorf("нода %s не найдена", nodeID) } ns.Disabled = disabled if disabled { ns.DisabledReason = reason } else { ns.DisabledReason = "" } e.scoreNodeLocked(ns, false) e.recalcBestLocked() return nil } // HasRecentMetrics возвращает true, если была получена хотя бы одна метрика за последние N секунд. func (e *Engine) HasRecentMetrics(seconds int64) bool { lastTS := e.lastMetricTime.Load() if lastTS == 0 { return false } return time.Now().Unix()-lastTS <= seconds } // IncrementRouteRequests увеличивает счётчик запросов /api/route. func (e *Engine) IncrementRouteRequests() { e.routeRequests.Add(1) } // IncrementRouteFallbacks увеличивает счётчик fallback-запросов. func (e *Engine) IncrementRouteFallbacks() { e.routeFallbacks.Add(1) } // HealthStats возвращает статистику для health endpoint. type HealthStats struct { TotalNodes int `json:"total_nodes"` HealthyNodes int `json:"healthy_nodes"` RouteRequests int64 `json:"route_requests_total"` RouteFallbacks int64 `json:"route_fallbacks_total"` UptimeSeconds int64 `json:"uptime_seconds"` LastMetricTS int64 `json:"last_metric_ts"` } // GetHealthStats возвращает текущую HealthStats (потокобезопасно). func (e *Engine) GetHealthStats() HealthStats { e.mu.RLock() defer e.mu.RUnlock() healthy := 0 for _, ns := range e.nodes { if ns.Score >= 0 && !ns.Disabled { healthy++ } } return HealthStats{ TotalNodes: len(e.nodes), HealthyNodes: healthy, RouteRequests: e.routeRequests.Load(), RouteFallbacks: e.routeFallbacks.Load(), UptimeSeconds: int64(time.Since(e.startTime).Seconds()), LastMetricTS: e.lastMetricTime.Load(), } } // StartEvictionLoop запускает периодическую очистку нод без метрик. // maxStale — время, после которого нода считается мёртвой (напр. 5 мин). func (e *Engine) StartEvictionLoop(interval, maxStale time.Duration) { e.evictionStopCh = make(chan struct{}) ticker := time.NewTicker(interval) go func() { defer ticker.Stop() for { select { case <-ticker.C: evicted := e.evictStaleNodes(maxStale) for _, id := range evicted { if e.onNodeEvicted != nil { e.onNodeEvicted(id) } } case <-e.evictionStopCh: return } } }() } // StopEvictionLoop останавливает цикл эвикции. func (e *Engine) StopEvictionLoop() { if e.evictionStopCh != nil { close(e.evictionStopCh) } } // SetOnNodeEvicted устанавливает callback, вызываемый при удалении stale-ноды. // Callback выполняется вне engine-блокировки и может безопасно делать file I/O. func (e *Engine) SetOnNodeEvicted(fn func(nodeID string)) { e.onNodeEvicted = fn } // evictStaleNodes удаляет ноды, от которых нет метрик дольше maxStale. // Возвращает список ID удалённых нод. func (e *Engine) evictStaleNodes(maxStale time.Duration) []string { e.mu.Lock() defer e.mu.Unlock() now := time.Now() 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) evicted = append(evicted, id) } } if len(evicted) > 0 { log.Printf("[engine] eviction: удалены ноды %v (stale > %v)", evicted, maxStale) e.recalcBestLocked() } return evicted } // --- Приватные методы --- // scoreNodeLocked вычисляет score для одной ноды. // Вызывающий держит write lock. // firstMetric — true при первом замере метрики (без EWMA-сглаживания). func (e *Engine) scoreNodeLocked(ns *models.NodeState, firstMetric bool) { sc := &e.cfg.Scoring // 1. Проверка lethal-условий if ns.Disabled { ns.Score = -100 ns.LethalReason = fmt.Sprintf("disabled: %s", ns.DisabledReason) return } if ns.Status != "ok" { ns.Score = -100 ns.LethalReason = fmt.Sprintf("status: %s", ns.Status) return } now := time.Now().Unix() staleSec := int(now - ns.TS) if staleSec > e.cfg.StaleThresholdSec { ns.Score = -100 ns.LethalReason = fmt.Sprintf("stale: %d сек без метрик", staleSec) return } if ns.ActiveCalls >= ns.MaxCalls && ns.MaxCalls > 0 { ns.Score = -100 ns.LethalReason = "active_calls >= max_calls" return } if ns.IdleCPU < sc.IdleCPUMin { ns.Score = -100 ns.LethalReason = fmt.Sprintf("idle_cpu %.1f < %.0f", ns.IdleCPU, sc.IdleCPUMin) return } if ns.CallFailureRate > sc.CallFailureRateLethal { ns.Score = -100 ns.LethalReason = fmt.Sprintf("call_failure_rate %.1f > %.0f", ns.CallFailureRate, sc.CallFailureRateLethal) return } // 2. Weighted score ns.LethalReason = "" loadMultiplier := sc.LoadAvgMultiplier if loadMultiplier <= 0 { loadMultiplier = 50.0 } callScore := clamp(100.0-(float64(ns.ActiveCalls)/float64(ns.MaxCalls)*100.0), 0, 100) loadScore := clamp(100.0-(ns.LoadAvg*loadMultiplier), 0, 100) idleScore := clamp(ns.IdleCPU, 0, 100) failScore := clamp(100.0-ns.CallFailureRate, 0, 100) w := &sc.Weights rawScore := callScore*w.CallScore + loadScore*w.LoadScore + idleScore*w.IdleScore + failScore*w.FailScore // EWMA-сглаживание: предотвращает резкие скачки score от одиночных выбросов метрики if sc.SmoothingFactor > 0 && !firstMetric { ns.Score = ns.Score*(1-sc.SmoothingFactor) + rawScore*sc.SmoothingFactor } else { ns.Score = rawScore } } // recalcBestLocked пересчитывает кэш лучшей ноды. // Вызывающий держит write lock. func (e *Engine) recalcBestLocked() { bestID := "" bestScore := -999.0 anyHealthy := false for id, ns := range e.nodes { if ns.Score >= 0 { anyHealthy = true } if ns.Score > bestScore { bestScore = ns.Score bestID = id } } e.bestNodeID = bestID e.bestScore = bestScore e.fallbackActive = !anyHealthy } // recalculateAll пересчитывает score всех нод и обновляет кэш. func (e *Engine) recalculateAll() { for _, ns := range e.nodes { e.scoreNodeLocked(ns, false) } e.recalcBestLocked() } // stateToNodeInfo конвертирует NodeState в NodeInfo для API. func stateToNodeInfo(ns *models.NodeState, staleThresholdSec int, now time.Time) *models.NodeInfo { secondsAgo := now.Sub(time.Unix(ns.TS, 0)).Seconds() isStale := int(secondsAgo) > staleThresholdSec return &models.NodeInfo{ NodeID: ns.NodeID, TS: ns.TS, Status: ns.Status, ActiveCalls: ns.ActiveCalls, MaxCalls: ns.MaxCalls, IdleCPU: ns.IdleCPU, LoadAvg: ns.LoadAvg, CallFailureRate: ns.CallFailureRate, Disabled: ns.Disabled, DisabledReason: ns.DisabledReason, Score: ns.Score, LethalReason: ns.LethalReason, IsStale: isStale, SecondsAgo: secondsAgo, } } // clamp ограничивает значение диапазоном [minVal, maxVal]. func clamp(v, minVal, maxVal float64) float64 { if v < minVal { return minVal } if v > maxVal { return maxVal } return v }