Maksim Totmin b0697da8e4 feat: weighted random routing + mode switching + EWMA smoothing + stale eviction
- Weighted random routing (math/rand/v2 per-goroutine ChaCha8)
- PUT /api/route/mode for runtime mode switch (best / weighted_random)
- EWMA score smoothing (configurable smoothing_factor, default 0.3)
- Periodic stale node eviction (5 min threshold, 60s interval)
- Non-blocking WS broadcast (per-client buffered channel + writePump)
- Background rate limiter cleanup goroutine
- In-memory trunk gateway cache (zero disk I/O in hot path)
- Configurable load_avg multiplier (default 50.0, backward compat)
- UI mode indicator + admin Switch button in dashboard
- 42 tests: weighted random, mode switching, EWMA, eviction, API integration
2026-06-25 20:42:46 +07:00

123 lines
3.3 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

package api
import (
"net/http"
"github.com/pulse-lets-go/internal/models"
)
// handleRoute — GET /api/route — главный эндпоинт маршрутизации вызова.
func (a *API) handleRoute(w http.ResponseWriter, r *http.Request) {
a.engine.IncrementRouteRequests()
nodeID, score, fallback := a.engine.PickNode()
// Нет зарегистрированных нод
if nodeID == "" {
a.engine.IncrementRouteFallbacks()
writeJSON(w, http.StatusServiceUnavailable, models.RouteResponse{
Error: "no_nodes_registered",
})
return
}
// Все ноды unhealthy — fallback
if fallback || score < 0 {
a.engine.IncrementRouteFallbacks()
fallbackGW := a.cachedFallbackGateway()
nodes := a.getRouteNodeInfo()
writeJSON(w, http.StatusOK, models.RouteResponse{
Fallback: true,
SIPGateway: fallbackGW,
Reason: "all_nodes_unhealthy",
Nodes: nodes,
})
return
}
// Нормальный маршрут — O(1) из in-memory кэша
gw, ok := a.cachedBalanceGateway(nodeID)
if !ok {
writeError(w, http.StatusInternalServerError, "gateway не найден для ноды")
return
}
writeJSON(w, http.StatusOK, models.RouteResponse{
NodeID: nodeID,
Score: score,
SIPGateway: gw,
})
}
// handleSetRouteMode — PUT /api/route/mode (admin only).
func (a *API) handleSetRouteMode(w http.ResponseWriter, r *http.Request) {
var req struct {
Mode string `json:"mode"`
}
if err := decodeJSON(r, &req); err != nil {
writeError(w, http.StatusBadRequest, "некорректный JSON")
return
}
if req.Mode != "best" && req.Mode != "weighted_random" {
writeError(w, http.StatusBadRequest, "mode должен быть 'best' или 'weighted_random'")
return
}
// Сохраняем в config.json атомарно
if a.configManager != nil {
cfg, err := a.configManager.ReadConfig()
if err == nil {
cfg.BalanceMode = req.Mode
a.configManager.SaveConfig(cfg)
}
}
a.engine.SetBalancingMode(req.Mode)
// Немедленный WS-push для обновления UI
go a.broadcastBalancerUpdate()
writeJSON(w, http.StatusOK, map[string]string{"mode": req.Mode})
}
// cachedBalanceGateway возвращает gateway для balance-транка из in-memory кэша (O(1)).
func (a *API) cachedBalanceGateway(nodeID string) (string, bool) {
a.trunkCacheMu.RLock()
defer a.trunkCacheMu.RUnlock()
gw, ok := a.balanceGateways[nodeID]
return gw, ok
}
// cachedFallbackGateway возвращает fallback gateway из in-memory кэша (O(1)).
func (a *API) cachedFallbackGateway() string {
a.trunkCacheMu.RLock()
defer a.trunkCacheMu.RUnlock()
return a.fallbackGateway
}
// getRouteNodeInfo возвращает список нод с их скорами для ответа маршрутизации.
func (a *API) getRouteNodeInfo() []models.RouteNodeInfo {
nodes := a.engine.GetAllNodes()
result := make([]models.RouteNodeInfo, 0, len(nodes))
for _, n := range nodes {
info := models.RouteNodeInfo{
ID: n.NodeID,
Score: n.Score,
Reason: n.LethalReason,
Status: n.Status,
}
if n.Disabled {
info.Status = "disabled"
} else if n.IsStale {
info.Status = "stale"
} else if n.Score < 0 {
info.Status = "unhealthy"
} else {
info.Status = "healthy"
}
result = append(result, info)
}
return result
}