Maksim Totmin 78785e54e1 feat: production-grade routing cache + stale detection fixes
Three-tier routing cache (ESL global -> file -> HTTP) eliminates HTTP
from call path for 2500-5000 concurrent calls. Lua reads cached route
in ~0.1us instead of blocking on api:execute('curl', ...).

Engine fixes:
- recalcBestLocked now skips stale nodes (was 5 min bug -> now ~10-15s)
- PickNodeForCall action_type='node' checks staleness for consistency
- onMetric callback pushes cache on every metric (no ticker delay)

Config:
- stale_threshold_sec default 20 -> 10 (industry standard)
- contrib/ included in Makefile deploy target
- route.lua paths updated for /opt/pulse-lets-go
2026-06-25 22:23:16 +07:00

237 lines
8.5 KiB
Go
Raw Permalink 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 (
"encoding/json"
"fmt"
"net/http"
"sync"
"github.com/pulse-lets-go/internal/config"
"github.com/pulse-lets-go/internal/engine"
"github.com/pulse-lets-go/internal/esl"
"github.com/pulse-lets-go/internal/models"
)
// API — главная структура HTTP API, агрегирует все зависимости.
type API struct {
engine *engine.Engine
configManager *config.Manager
jwtSecret string
monitoringAPIKey string
wsHub *wsHub
natsConnected func() bool // колбэк для проверки NATS-статуса (health)
eslClient *esl.Client // ESL-клиент (nil если не сконфигурирован)
logFormat string
routeLimiter *rateLimiter
apiLimiter *rateLimiter
// In-memory кэш gateway для /api/route (устраняет disk I/O в hot path)
trunkCacheMu sync.RWMutex
balanceGateways map[string]string // nodeID → gateway
fallbackGateway string
gatewayMu sync.RWMutex
gatewayStates map[string]string // trunkID → "up"/"down"
fsStats esl.FsStats
fsStatsMu sync.RWMutex
routeCachePath string // путь к файловому кэшу маршрута (для Lua)
}
// NewAPI создаёт новый HTTP API с заданными зависимостями.
func NewAPI(eng *engine.Engine, cfgMgr *config.Manager, jwtSecret, monitoringAPIKey string, natsFn func() bool, cfg *config.Config) *API {
a := &API{
engine: eng,
configManager: cfgMgr,
jwtSecret: jwtSecret,
monitoringAPIKey: monitoringAPIKey,
wsHub: newWSHub(),
natsConnected: natsFn,
logFormat: cfg.Log.Format,
gatewayStates: make(map[string]string),
balanceGateways: make(map[string]string),
}
if cfg.RateLimit.Enabled {
a.routeLimiter = newRateLimiter(cfg.RateLimit.RoutePerSec, cfg.RateLimit.RoutePerSec)
a.apiLimiter = newRateLimiter(cfg.RateLimit.APIPerSec, cfg.RateLimit.APIPerSec)
}
return a
}
// SetESLClient устанавливает ESL-клиент (вызывается после инициализации ESL).
func (a *API) SetESLClient(client *esl.Client) {
a.eslClient = client
}
// Handler возвращает http.Handler со всеми зарегистрированными маршрутами.
func (a *API) Handler() http.Handler {
mux := http.NewServeMux()
// --- Health endpoints (без авторизации) ---
mux.HandleFunc("GET /api/health", a.handleHealth)
mux.HandleFunc("GET /api/health/live", a.handleLiveness)
mux.HandleFunc("GET /api/health/ready", a.handleReadiness)
// --- Публичные маршруты (без авторизации) ---
mux.HandleFunc("POST /api/auth/login", a.handleLogin)
mux.HandleFunc("POST /api/auth/refresh", a.handleRefresh)
mux.HandleFunc("GET /api/route", a.handleRoute)
mux.HandleFunc("GET /ws/metrics", a.handleWS)
// Monitoring с API-key (без JWT)
mux.Handle("GET /api/monitoring/zabbix", a.apiKeyMiddleware(http.HandlerFunc(a.handleZabbix)))
mux.Handle("GET /api/monitoring/prometheus", a.apiKeyMiddleware(http.HandlerFunc(a.handlePrometheus)))
// --- Админка: Nodes (admin + viewer) ---
mux.Handle("GET /api/nodes", a.authMiddleware(http.HandlerFunc(a.handleGetNodes)))
mux.Handle("GET /api/nodes/{id}/metrics", a.authMiddleware(http.HandlerFunc(a.handleGetNodeMetrics)))
mux.Handle("PUT /api/nodes/{id}/toggle", a.authMiddleware(a.adminOnly(http.HandlerFunc(a.handleToggleNode))))
mux.Handle("PUT /api/route/mode", a.authMiddleware(a.adminOnly(http.HandlerFunc(a.handleSetRouteMode))))
// --- Админка: Trunks (admin only) ---
mux.Handle("GET /api/trunks", a.authMiddleware(a.adminOnly(http.HandlerFunc(a.handleGetTrunks))))
mux.Handle("POST /api/trunks", a.authMiddleware(a.adminOnly(http.HandlerFunc(a.handleCreateTrunk))))
mux.Handle("PUT /api/trunks/{id}", a.authMiddleware(a.adminOnly(http.HandlerFunc(a.handleUpdateTrunk))))
mux.Handle("DELETE /api/trunks/{id}", a.authMiddleware(a.adminOnly(http.HandlerFunc(a.handleDeleteTrunk))))
// --- Админка: Users (admin only) ---
mux.Handle("GET /api/users", a.authMiddleware(a.adminOnly(http.HandlerFunc(a.handleGetUsers))))
mux.Handle("POST /api/users", a.authMiddleware(a.adminOnly(http.HandlerFunc(a.handleCreateUser))))
mux.Handle("PUT /api/users/{id}", a.authMiddleware(a.adminOnly(http.HandlerFunc(a.handleUpdateUser))))
mux.Handle("DELETE /api/users/{id}", a.authMiddleware(a.adminOnly(http.HandlerFunc(a.handleDeleteUser))))
// --- Админка: Routing Rules (admin + viewer for GET, admin only for write) ---
mux.Handle("GET /api/routing/rules", a.authMiddleware(http.HandlerFunc(a.handleGetRoutingRules)))
mux.Handle("POST /api/routing/rules", a.authMiddleware(a.adminOnly(http.HandlerFunc(a.handleCreateRoutingRule))))
mux.Handle("PUT /api/routing/rules/{id}", a.authMiddleware(a.adminOnly(http.HandlerFunc(a.handleUpdateRoutingRule))))
mux.Handle("DELETE /api/routing/rules/{id}", a.authMiddleware(a.adminOnly(http.HandlerFunc(a.handleDeleteRoutingRule))))
mux.Handle("PUT /api/routing/rules/{id}/toggle", a.authMiddleware(a.adminOnly(http.HandlerFunc(a.handleToggleRoutingRule))))
mux.Handle("PUT /api/routing/rules/reorder", a.authMiddleware(a.adminOnly(http.HandlerFunc(a.handleReorderRoutingRules))))
// Логирование + CORS (навешиваем после mux)
var wrapped http.Handler = mux
wrapped = requestLoggingMiddleware(a.logFormat)(wrapped)
// Rate limiting
if a.routeLimiter != nil && a.apiLimiter != nil {
handlerBeforeRL := wrapped
wrapped = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path == "/api/route" {
rateLimitMiddleware(a.routeLimiter)(handlerBeforeRL).ServeHTTP(w, r)
return
}
if len(r.URL.Path) > 5 && r.URL.Path[:5] == "/api/" {
rateLimitMiddleware(a.apiLimiter)(handlerBeforeRL).ServeHTTP(w, r)
return
}
handlerBeforeRL.ServeHTTP(w, r)
})
}
return corsMiddleware(wrapped)
}
// corsMiddleware добавляет CORS-заголовки.
func corsMiddleware(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Access-Control-Allow-Origin", "*")
w.Header().Set("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS")
w.Header().Set("Access-Control-Allow-Headers", "Content-Type, Authorization, X-API-Key")
if r.Method == http.MethodOptions {
w.WriteHeader(http.StatusNoContent)
return
}
next.ServeHTTP(w, r)
})
}
// BroadcastGatewayEvent транслирует статус gateway всем WS-клиентам.
func (a *API) BroadcastGatewayEvent(gatewayName, trunkID, status string) {
a.gatewayMu.Lock()
a.gatewayStates[trunkID] = status
a.gatewayMu.Unlock()
msg := &models.WsMetricsMessage{
Type: "gateway_update",
Payload: map[string]string{
"gateway": gatewayName,
"trunk_id": trunkID,
"status": status,
},
}
a.wsHub.broadcast(msg)
}
func (a *API) gatewayStats() (total, up, down int) {
a.gatewayMu.RLock()
defer a.gatewayMu.RUnlock()
for _, s := range a.gatewayStates {
total++
switch s {
case "up":
up++
case "down":
down++
}
}
return
}
// RebuildTrunkCache перестраивает in-memory кэш gateway для /api/route.
// Вызывается при старте и после любого CRUD-оперирования с транками.
func (a *API) RebuildTrunkCache() {
trunks, err := a.configManager.ReadTrunks()
if err != nil {
return
}
a.trunkCacheMu.Lock()
defer a.trunkCacheMu.Unlock()
a.balanceGateways = make(map[string]string, len(trunks))
a.fallbackGateway = ""
for _, t := range trunks {
if t.Enabled {
switch t.Type {
case models.TrunkBalance:
if t.NodeID != "" {
a.balanceGateways[t.NodeID] = t.Gateway
}
case models.TrunkFallback:
if a.fallbackGateway == "" {
a.fallbackGateway = t.Gateway
}
}
}
}
}
// UpdateFsStats обновляет кэш метрик FreeSWITCH (вызывается из ticker).
func (a *API) UpdateFsStats(stats esl.FsStats) {
a.fsStatsMu.Lock()
a.fsStats = stats
a.fsStatsMu.Unlock()
}
// maxBodySize — максимальный размер тела запроса в байтах (1 MiB).
const maxBodySize = 1 << 20
// decodeJSON декодирует тело запроса в структуру v с ограничением размера.
func decodeJSON(r *http.Request, v interface{}) error {
defer r.Body.Close()
r.Body = http.MaxBytesReader(nil, r.Body, maxBodySize)
if err := json.NewDecoder(r.Body).Decode(v); err != nil {
return fmt.Errorf("декодирование JSON: %w", err)
}
return nil
}