first commit

This commit is contained in:
Maksim Totmin
2026-06-25 14:27:41 +07:00
commit 44f5fa88e5
44 changed files with 8664 additions and 0 deletions
+271
View File
@@ -0,0 +1,271 @@
// Package api реализует HTTP API pulse-lets-go.
package api
import (
"context"
"encoding/json"
"log"
"net/http"
"strings"
"time"
"github.com/golang-jwt/jwt/v5"
"golang.org/x/crypto/bcrypt"
"github.com/pulse-lets-go/internal/models"
)
// JWT claims
type claims struct {
UserID string `json:"user_id"`
Username string `json:"username"`
Role string `json:"role"`
jwt.RegisteredClaims
}
// contextKey — тип для ключей context.
type contextKey string
const (
userContextKey contextKey = "user"
refreshTokenKey contextKey = "refresh_token"
)
// UserInfo — информация о пользователе, извлекаемая из JWT.
type UserInfo struct {
UserID string
Username string
Role string
}
const (
jwtExpiry = 15 * time.Minute
refreshExpiry = 24 * time.Hour
)
// generateJWT создаёт access+refresh токены.
func (a *API) generateJWT(user *models.User) (*models.LoginResponse, error) {
now := time.Now()
// Access token
accessClaims := claims{
UserID: user.ID,
Username: user.Username,
Role: string(user.Role),
RegisteredClaims: jwt.RegisteredClaims{
ExpiresAt: jwt.NewNumericDate(now.Add(jwtExpiry)),
IssuedAt: jwt.NewNumericDate(now),
},
}
accessToken, err := jwt.NewWithClaims(jwt.SigningMethodHS256, accessClaims).SignedString([]byte(a.jwtSecret))
if err != nil {
return nil, err
}
// Refresh token
refreshClaims := claims{
UserID: user.ID,
Username: user.Username,
Role: string(user.Role),
RegisteredClaims: jwt.RegisteredClaims{
ExpiresAt: jwt.NewNumericDate(now.Add(refreshExpiry)),
IssuedAt: jwt.NewNumericDate(now),
},
}
refreshToken, err := jwt.NewWithClaims(jwt.SigningMethodHS256, refreshClaims).SignedString([]byte(a.jwtSecret))
if err != nil {
return nil, err
}
return &models.LoginResponse{
AccessToken: accessToken,
RefreshToken: refreshToken,
ExpiresIn: int64(jwtExpiry.Seconds()),
}, nil
}
// authMiddleware — проверяет JWT в заголовке Authorization.
func (a *API) authMiddleware(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
tokenStr := extractBearerToken(r)
if tokenStr == "" {
writeError(w, http.StatusUnauthorized, "требуется авторизация")
return
}
c := &claims{}
token, err := jwt.ParseWithClaims(tokenStr, c, func(token *jwt.Token) (interface{}, error) {
return []byte(a.jwtSecret), nil
})
if err != nil || !token.Valid {
writeError(w, http.StatusUnauthorized, "недействительный токен")
return
}
user := &UserInfo{
UserID: c.UserID,
Username: c.Username,
Role: c.Role,
}
ctx := context.WithValue(r.Context(), userContextKey, user)
next.ServeHTTP(w, r.WithContext(ctx))
})
}
// adminOnly — middleware проверки роли admin.
func (a *API) adminOnly(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
user := GetUserFromContext(r.Context())
if user == nil || user.Role != string(models.RoleAdmin) {
writeError(w, http.StatusForbidden, "требуется роль admin")
return
}
next.ServeHTTP(w, r)
})
}
// apiKeyMiddleware — проверяет API-key для monitoring endpoint.
func (a *API) apiKeyMiddleware(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
key := r.Header.Get("X-API-Key")
if key == "" || key != a.monitoringAPIKey {
writeError(w, http.StatusUnauthorized, "недействительный API-ключ")
return
}
next.ServeHTTP(w, r)
})
}
// handleLogin обрабатывает POST /api/auth/login.
func (a *API) handleLogin(w http.ResponseWriter, r *http.Request) {
var req models.LoginRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, "некорректный JSON")
return
}
if req.Username == "" || req.Password == "" {
writeError(w, http.StatusBadRequest, "имя и пароль обязательны")
return
}
users, err := a.configManager.ReadUsers()
if err != nil {
log.Printf("[auth] ошибка чтения users.json: %v", err)
writeError(w, http.StatusInternalServerError, "ошибка сервера")
return
}
var found *models.User
for i := range users {
if users[i].Username == req.Username {
found = &users[i]
break
}
}
if found == nil {
writeError(w, http.StatusUnauthorized, "неверное имя или пароль")
return
}
// Если пароль не задан (плейсхолдер), разрешаем вход с паролем "admin"
if found.PasswordHash == "" {
if req.Password != "admin" {
writeError(w, http.StatusUnauthorized, "неверное имя или пароль")
return
}
// Хешируем пароль и сохраняем
hash, _ := bcrypt.GenerateFromPassword([]byte(req.Password), bcrypt.DefaultCost)
found.PasswordHash = string(hash)
a.configManager.SaveUsers(users)
} else {
if err := bcrypt.CompareHashAndPassword([]byte(found.PasswordHash), []byte(req.Password)); err != nil {
writeError(w, http.StatusUnauthorized, "неверное имя или пароль")
return
}
}
resp, err := a.generateJWT(found)
if err != nil {
log.Printf("[auth] ошибка генерации JWT: %v", err)
writeError(w, http.StatusInternalServerError, "ошибка сервера")
return
}
writeJSON(w, http.StatusOK, resp)
}
// handleRefresh выдаёт новый access токен по refresh токену.
func (a *API) handleRefresh(w http.ResponseWriter, r *http.Request) {
tokenStr := extractBearerToken(r)
if tokenStr == "" {
// пробуем получить из query param (для WebSocket)
tokenStr = r.URL.Query().Get("token")
}
if tokenStr == "" {
writeError(w, http.StatusUnauthorized, "требуется токен")
return
}
c := &claims{}
token, err := jwt.ParseWithClaims(tokenStr, c, func(token *jwt.Token) (interface{}, error) {
return []byte(a.jwtSecret), nil
})
if err != nil || !token.Valid {
writeError(w, http.StatusUnauthorized, "недействительный refresh токен")
return
}
user := &models.User{
ID: c.UserID,
Username: c.Username,
Role: models.UserRole(c.Role),
}
resp, err := a.generateJWT(user)
if err != nil {
writeError(w, http.StatusInternalServerError, "ошибка сервера")
return
}
writeJSON(w, http.StatusOK, resp)
}
// --- Хелперы ---
// extractBearerToken извлекает JWT из заголовка Authorization: Bearer <token>.
func extractBearerToken(r *http.Request) string {
auth := r.Header.Get("Authorization")
if auth == "" {
return ""
}
parts := strings.SplitN(auth, " ", 2)
if len(parts) != 2 || !strings.EqualFold(parts[0], "bearer") {
return ""
}
return parts[1]
}
// GetUserFromContext извлекает UserInfo из контекста запроса.
func GetUserFromContext(ctx context.Context) *UserInfo {
user, ok := ctx.Value(userContextKey).(*UserInfo)
if !ok {
return nil
}
return user
}
// writeJSON отправляет JSON-ответ.
func writeJSON(w http.ResponseWriter, status int, v interface{}) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(status)
json.NewEncoder(w).Encode(v)
}
// writeError отправляет JSON-ошибку.
func writeError(w http.ResponseWriter, status int, msg string) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(status)
json.NewEncoder(w).Encode(map[string]string{"error": msg})
}
+63
View File
@@ -0,0 +1,63 @@
package api
import (
"net/http"
)
// handleHealth — GET /api/health — возвращает статус всех зависимостей.
func (a *API) handleHealth(w http.ResponseWriter, r *http.Request) {
stats := a.engine.GetHealthStats()
natsStatus := "disconnected"
if a.natsConnected != nil && a.natsConnected() {
natsStatus = "connected"
}
eslInfo := map[string]interface{}{"status": "disabled"}
if a.eslClient != nil {
eslStats := a.eslClient.GetStats()
eslInfo = map[string]interface{}{
"status": eslStats.Status,
"host": eslStats.Host,
"uptime_seconds": eslStats.UptimeSec,
"reconnects": eslStats.Reconnects,
"gateway_ops": eslStats.GatewayOps,
"events_recv": eslStats.EventsRecv,
}
}
writeJSON(w, http.StatusOK, map[string]interface{}{
"status": "ok",
"version": "1.0.0",
"connections": map[string]interface{}{
"nats": natsStatus,
"esl": eslInfo,
},
"stats": stats,
})
}
// handleLiveness — GET /api/health/live — всегда 200, если процесс жив.
func (a *API) handleLiveness(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, map[string]string{"status": "alive"})
}
// handleReadiness — GET /api/health/ready — 200 когда NATS подключён, метрики поступают.
// ESL опционален — не требуется для готовности.
func (a *API) handleReadiness(w http.ResponseWriter, r *http.Request) {
if a.natsConnected != nil && !a.natsConnected() {
writeJSON(w, http.StatusServiceUnavailable, map[string]string{
"status": "not_ready",
"reason": "nats_disconnected",
})
return
}
if !a.engine.HasRecentMetrics(60) {
writeJSON(w, http.StatusServiceUnavailable, map[string]string{
"status": "not_ready",
"reason": "no_recent_metrics",
})
return
}
writeJSON(w, http.StatusOK, map[string]string{"status": "ready"})
}
+84
View File
@@ -0,0 +1,84 @@
package api
import (
"crypto/rand"
"encoding/hex"
"encoding/json"
"log"
"net/http"
"time"
)
// requestLoggingMiddleware логирует каждый HTTP-запрос с Request ID, методом, путём, статусом и длительностью.
func requestLoggingMiddleware(logFormat string) func(http.Handler) http.Handler {
return func(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
// Не логируем health-check запросы и OPTIONS
if r.URL.Path == "/api/health/live" || r.URL.Path == "/api/health/ready" || r.Method == "OPTIONS" {
next.ServeHTTP(w, r)
return
}
reqID := generateRequestID()
r.Header.Set("X-Request-ID", reqID)
w.Header().Set("X-Request-ID", reqID)
start := time.Now()
lrw := &loggingResponseWriter{ResponseWriter: w, statusCode: http.StatusOK}
next.ServeHTTP(lrw, r)
duration := time.Since(start)
// Извлекаем пользователя из контекста (может быть nil)
userStr := "-"
if user := GetUserFromContext(r.Context()); user != nil {
userStr = user.Username
}
switch logFormat {
case "json":
entry := map[string]interface{}{
"ts": start.UTC().Format(time.RFC3339),
"id": reqID,
"method": r.Method,
"path": r.URL.Path,
"status": lrw.statusCode,
"duration_ms": float64(duration.Microseconds()) / 1000.0,
"remote_addr": r.RemoteAddr,
"user": userStr,
}
data, _ := json.Marshal(entry)
log.Printf("%s", string(data))
default:
log.Printf("[%s] %s %s %d %.2fms %s user=%s",
reqID, r.Method, r.URL.Path, lrw.statusCode,
float64(duration.Microseconds())/1000.0, r.RemoteAddr, userStr)
}
})
}
}
// generateRequestID генерирует случайный Request ID: req-XXXXXXXX.
func generateRequestID() string {
b := make([]byte, 6)
if _, err := rand.Read(b); err != nil {
return "req-" + hex.EncodeToString([]byte{0, 0, 0, 0, 0, 0})
}
return "req-" + hex.EncodeToString(b)
}
// loggingResponseWriter оборачивает http.ResponseWriter для перехвата статус-кода.
type loggingResponseWriter struct {
http.ResponseWriter
statusCode int
}
func (w *loggingResponseWriter) WriteHeader(code int) {
w.statusCode = code
w.ResponseWriter.WriteHeader(code)
}
func (w *loggingResponseWriter) Write(b []byte) (int, error) {
return w.ResponseWriter.Write(b)
}
+78
View File
@@ -0,0 +1,78 @@
package api
import (
"fmt"
"net/http"
"strings"
"github.com/pulse-lets-go/internal/models"
)
// handleZabbix — GET /api/monitoring/zabbix — возвращает JSON для Zabbix LLD + items.
func (a *API) handleZabbix(w http.ResponseWriter, r *http.Request) {
nodes := a.engine.GetAllNodes()
data := make([]models.ZabbixNode, 0, len(nodes))
for _, n := range nodes {
data = append(data, models.ZabbixNode{
NodeID: n.NodeID,
Status: n.Status,
ActiveCalls: n.ActiveCalls,
MaxCalls: n.MaxCalls,
IdleCPU: n.IdleCPU,
LoadAvg: n.LoadAvg,
CallFailureRate: n.CallFailureRate,
Score: n.Score,
Disabled: n.Disabled,
IsStale: n.IsStale,
SecondsAgo: n.SecondsAgo,
})
}
writeJSON(w, http.StatusOK, models.ZabbixResponse{Data: data})
}
// handlePrometheus — GET /api/monitoring/prometheus — text/plain метрики для Prometheus.
func (a *API) handlePrometheus(w http.ResponseWriter, r *http.Request) {
nodes := a.engine.GetAllNodes()
var sb strings.Builder
for _, n := range nodes {
id := sanitizePromLabel(n.NodeID)
fmt.Fprintf(&sb, "pulse_node_active_calls{node=\"%s\"} %d\n", id, n.ActiveCalls)
fmt.Fprintf(&sb, "pulse_node_max_calls{node=\"%s\"} %d\n", id, n.MaxCalls)
fmt.Fprintf(&sb, "pulse_node_idle_cpu{node=\"%s\"} %.2f\n", id, n.IdleCPU)
fmt.Fprintf(&sb, "pulse_node_load_avg{node=\"%s\"} %.2f\n", id, n.LoadAvg)
fmt.Fprintf(&sb, "pulse_node_call_failure_rate{node=\"%s\"} %.2f\n", id, n.CallFailureRate)
fmt.Fprintf(&sb, "pulse_node_score{node=\"%s\"} %.2f\n", id, n.Score)
fmt.Fprintf(&sb, "pulse_node_seconds_ago{node=\"%s\"} %.2f\n", id, n.SecondsAgo)
disabledVal := 0
staleVal := 0
statusVal := 1 // ok
if n.Disabled {
disabledVal = 1
}
if n.IsStale {
staleVal = 1
}
if n.Status != "ok" {
statusVal = 0
}
fmt.Fprintf(&sb, "pulse_node_disabled{node=\"%s\"} %d\n", id, disabledVal)
fmt.Fprintf(&sb, "pulse_node_stale{node=\"%s\"} %d\n", id, staleVal)
fmt.Fprintf(&sb, "pulse_node_status_ok{node=\"%s\"} %d\n", id, statusVal)
}
w.Header().Set("Content-Type", "text/plain; charset=utf-8")
w.WriteHeader(http.StatusOK)
w.Write([]byte(sb.String()))
}
// sanitizePromLabel заменяет символы, недопустимые в Prometheus-лейблах.
func sanitizePromLabel(s string) string {
return strings.ReplaceAll(strings.ReplaceAll(s, ".", "_"), "-", "_")
}
+52
View File
@@ -0,0 +1,52 @@
package api
import (
"net/http"
"github.com/pulse-lets-go/internal/models"
)
// handleGetNodes — GET /api/nodes — список всех нод с текущими score.
func (a *API) handleGetNodes(w http.ResponseWriter, r *http.Request) {
nodes := a.engine.GetAllNodes()
writeJSON(w, http.StatusOK, nodes)
}
// handleGetNodeMetrics — GET /api/nodes/{id}/metrics — история метрик ноды из ring buffer.
func (a *API) handleGetNodeMetrics(w http.ResponseWriter, r *http.Request) {
nodeID := r.PathValue("id")
if nodeID == "" {
writeError(w, http.StatusBadRequest, "не указан id ноды")
return
}
snapshots := a.engine.GetNodeHistory(nodeID)
writeJSON(w, http.StatusOK, map[string]interface{}{
"node_id": nodeID,
"snapshots": snapshots,
"count": len(snapshots),
})
}
// handleToggleNode — PUT /api/nodes/{id}/toggle — включение/выключение ноды.
func (a *API) handleToggleNode(w http.ResponseWriter, r *http.Request) {
nodeID := r.PathValue("id")
if nodeID == "" {
writeError(w, http.StatusBadRequest, "не указан id ноды")
return
}
var req models.ToggleNodeRequest
if err := decodeJSON(r, &req); err != nil {
writeError(w, http.StatusBadRequest, "некорректный JSON")
return
}
if err := a.engine.ToggleNode(nodeID, req.Disabled, req.Reason); err != nil {
writeError(w, http.StatusNotFound, err.Error())
return
}
info, _ := a.engine.GetNodeInfo(nodeID)
writeJSON(w, http.StatusOK, info)
}
+109
View File
@@ -0,0 +1,109 @@
package api
import (
"net"
"net/http"
"sync"
"time"
)
// rateLimiter — in-memory token bucket rate limiter.
type rateLimiter struct {
mu sync.Mutex
buckets map[string]*tokenBucket
rate float64 // токенов в секунду
burst int // максимальный размер бакета
cleanupInterval time.Duration
lastCleanup time.Time
}
type tokenBucket struct {
tokens float64
lastSeen time.Time
}
// newRateLimiter создаёт rate limiter с заданными параметрами.
func newRateLimiter(ratePerSec, burst int) *rateLimiter {
if burst <= 0 {
burst = ratePerSec
}
return &rateLimiter{
buckets: make(map[string]*tokenBucket),
rate: float64(ratePerSec),
burst: burst,
cleanupInterval: 60 * time.Second,
lastCleanup: time.Now(),
}
}
// allow проверяет, разрешён ли запрос для данного ключа.
// Возвращает true если разрешён, false если превышен лимит.
func (rl *rateLimiter) allow(key string) bool {
rl.mu.Lock()
defer rl.mu.Unlock()
now := time.Now()
// Периодическая очистка устаревших бакетов
if now.Sub(rl.lastCleanup) > rl.cleanupInterval {
for k, b := range rl.buckets {
if now.Sub(b.lastSeen) > rl.cleanupInterval {
delete(rl.buckets, k)
}
}
rl.lastCleanup = now
}
b, exists := rl.buckets[key]
if !exists {
// Новый бакет с полным запасом токенов
b = &tokenBucket{tokens: float64(rl.burst), lastSeen: now}
rl.buckets[key] = b
b.tokens-- // расходуем один токен
return true
}
// Пополняем токены
elapsed := now.Sub(b.lastSeen).Seconds()
b.tokens += elapsed * rl.rate
if b.tokens > float64(rl.burst) {
b.tokens = float64(rl.burst)
}
b.lastSeen = now
if b.tokens >= 1.0 {
b.tokens--
return true
}
return false
}
// rateLimitMiddleware создаёт middleware с rate limiting по IP.
// Ключ — RemoteAddr (или X-Forwarded-For, если за проксей).
func rateLimitMiddleware(limiter *rateLimiter) func(http.Handler) http.Handler {
return func(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
key := stripPort(r.RemoteAddr)
if fwd := r.Header.Get("X-Forwarded-For"); fwd != "" {
key = fwd
}
if !limiter.allow(key) {
w.Header().Set("Retry-After", "1")
writeError(w, http.StatusTooManyRequests, "слишком много запросов")
return
}
next.ServeHTTP(w, r)
})
}
}
// stripPort убирает порт из адреса (напр. "[::1]:58544" → "::1", "127.0.0.1:1234" → "127.0.0.1").
func stripPort(addr string) string {
host, _, err := net.SplitHostPort(addr)
if err != nil {
return addr // нет порта
}
return host
}
+102
View File
@@ -0,0 +1,102 @@
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.GetBestNode()
// Нет зарегистрированных нод
if nodeID == "" {
writeJSON(w, http.StatusServiceUnavailable, models.RouteResponse{
Error: "no_nodes_registered",
})
return
}
// Все ноды unhealthy — fallback
if fallback || score < 0 {
fallbackGW, _ := a.findFallbackGateway()
nodes := a.getRouteNodeInfo()
writeJSON(w, http.StatusOK, models.RouteResponse{
Fallback: true,
SIPGateway: fallbackGW,
Reason: "all_nodes_unhealthy",
Nodes: nodes,
})
return
}
// Нормальный маршрут
gw, ok := a.findBalanceGateway(nodeID)
if !ok {
writeError(w, http.StatusInternalServerError, "gateway не найден для ноды")
return
}
writeJSON(w, http.StatusOK, models.RouteResponse{
NodeID: nodeID,
Score: score,
SIPGateway: gw,
})
}
// findBalanceGateway ищет gateway для balance-транка, привязанного к nodeID.
func (a *API) findBalanceGateway(nodeID string) (string, bool) {
trunks, err := a.configManager.ReadTrunks()
if err != nil {
return "", false
}
for _, t := range trunks {
if t.Type == models.TrunkBalance && t.NodeID == nodeID && t.Enabled {
return t.Gateway, true
}
}
return "", false
}
// findFallbackGateway ищет gateway для fallback-транка.
func (a *API) findFallbackGateway() (string, bool) {
trunks, err := a.configManager.ReadTrunks()
if err != nil {
return "", false
}
for _, t := range trunks {
if t.Type == models.TrunkFallback && t.Enabled {
return t.Gateway, true
}
}
return "", false
}
// 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
}
+137
View File
@@ -0,0 +1,137 @@
package api
import (
"encoding/json"
"fmt"
"net/http"
"github.com/pulse-lets-go/internal/config"
"github.com/pulse-lets-go/internal/engine"
"github.com/pulse-lets-go/internal/esl"
)
// 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
}
// 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,
}
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))))
// --- Админка: 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))))
// Логирование + 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)
})
}
// decodeJSON декодирует тело запроса в структуру v.
func decodeJSON(r *http.Request, v interface{}) error {
defer r.Body.Close()
if err := json.NewDecoder(r.Body).Decode(v); err != nil {
return fmt.Errorf("декодирование JSON: %w", err)
}
return nil
}
+306
View File
@@ -0,0 +1,306 @@
package api
import (
"fmt"
"log"
"net/http"
"sort"
"strings"
"time"
"github.com/pulse-lets-go/internal/config"
"github.com/pulse-lets-go/internal/models"
)
// handleGetTrunks — GET /api/trunks — список всех транков с опциональным фильтром.
func (a *API) handleGetTrunks(w http.ResponseWriter, r *http.Request) {
trunks, err := a.configManager.ReadTrunks()
if err != nil {
writeError(w, http.StatusInternalServerError, "ошибка чтения транков")
return
}
// Фильтр по типу
typeFilter := r.URL.Query().Get("type")
if typeFilter != "" {
filtered := make([]models.Trunk, 0)
for _, t := range trunks {
if string(t.Type) == typeFilter {
filtered = append(filtered, t)
}
}
trunks = filtered
}
// Сортировка по created_at
sort.Slice(trunks, func(i, j int) bool {
return trunks[i].CreatedAt.Before(trunks[j].CreatedAt)
})
writeJSON(w, http.StatusOK, trunks)
}
// handleCreateTrunk — POST /api/trunks — создание нового транка.
func (a *API) handleCreateTrunk(w http.ResponseWriter, r *http.Request) {
var req models.CreateTrunkRequest
if err := decodeJSON(r, &req); err != nil {
writeError(w, http.StatusBadRequest, "некорректный JSON")
return
}
if err := validateTrunk(req.Name, req.Type, req.Gateway); err != nil {
writeError(w, http.StatusBadRequest, err.Error())
return
}
// Проверяем: fallback может быть только один
trunks, err := a.configManager.ReadTrunks()
if err != nil {
writeError(w, http.StatusInternalServerError, "ошибка чтения транков")
return
}
if req.Type == models.TrunkFallback {
for _, t := range trunks {
if t.Type == models.TrunkFallback {
writeError(w, http.StatusConflict, "fallback-транк уже существует (должен быть ровно 1)")
return
}
}
}
now := time.Now().UTC()
trunk := models.Trunk{
ID: generateID("trk"),
Name: req.Name,
Type: req.Type,
NodeID: req.NodeID,
Gateway: req.Gateway,
Codecs: req.Codecs,
Context: req.Context,
Enabled: req.Enabled,
Description: req.Description,
CreatedAt: now,
UpdatedAt: now,
}
trunks = append(trunks, trunk)
if err := a.configManager.SaveTrunks(trunks); err != nil {
writeError(w, http.StatusInternalServerError, "ошибка сохранения транка")
return
}
// ESL push: создаём gateway на FS для ingress-транка
a.eslPushGatewayAdd(trunk)
writeJSON(w, http.StatusCreated, trunk)
}
// handleUpdateTrunk — PUT /api/trunks/{id} — обновление транка.
func (a *API) handleUpdateTrunk(w http.ResponseWriter, r *http.Request) {
trunkID := r.PathValue("id")
if trunkID == "" {
writeError(w, http.StatusBadRequest, "не указан id транка")
return
}
var req models.UpdateTrunkRequest
if err := decodeJSON(r, &req); err != nil {
writeError(w, http.StatusBadRequest, "некорректный JSON")
return
}
trunks, err := a.configManager.ReadTrunks()
if err != nil {
writeError(w, http.StatusInternalServerError, "ошибка чтения транков")
return
}
idx := findTrunkIndex(trunks, trunkID)
if idx < 0 {
writeError(w, http.StatusNotFound, "транк не найден")
return
}
// Применяем частичные обновления
t := &trunks[idx]
if req.Name != nil {
t.Name = *req.Name
}
if req.Type != nil {
// Проверка fallback-уникальности
if *req.Type == models.TrunkFallback && t.Type != models.TrunkFallback {
for _, other := range trunks {
if other.Type == models.TrunkFallback && other.ID != trunkID {
writeError(w, http.StatusConflict, "fallback-транк уже существует")
return
}
}
}
t.Type = *req.Type
}
if req.NodeID != nil {
t.NodeID = *req.NodeID
}
if req.Gateway != nil {
t.Gateway = *req.Gateway
}
if req.Codecs != nil {
t.Codecs = req.Codecs
}
if req.Context != nil {
t.Context = *req.Context
}
if req.Enabled != nil {
t.Enabled = *req.Enabled
}
if req.Description != nil {
t.Description = *req.Description
}
t.UpdatedAt = time.Now().UTC()
if err := a.configManager.SaveTrunks(trunks); err != nil {
writeError(w, http.StatusInternalServerError, "ошибка сохранения транка")
return
}
// ESL push: обновляем gateway на FS
a.eslPushGatewayUpdate(*t)
writeJSON(w, http.StatusOK, t)
}
// handleDeleteTrunk — DELETE /api/trunks/{id} — удаление транка.
func (a *API) handleDeleteTrunk(w http.ResponseWriter, r *http.Request) {
trunkID := r.PathValue("id")
if trunkID == "" {
writeError(w, http.StatusBadRequest, "не указан id транка")
return
}
trunks, err := a.configManager.ReadTrunks()
if err != nil {
writeError(w, http.StatusInternalServerError, "ошибка чтения транков")
return
}
idx := findTrunkIndex(trunks, trunkID)
if idx < 0 {
writeError(w, http.StatusNotFound, "транк не найден")
return
}
deletedTrunk := trunks[idx]
trunks = append(trunks[:idx], trunks[idx+1:]...)
// ESL push: удаляем gateway на FS
a.eslPushGatewayDelete(deletedTrunk)
if err := a.configManager.SaveTrunks(trunks); err != nil {
writeError(w, http.StatusInternalServerError, "ошибка сохранения транков")
return
}
writeJSON(w, http.StatusNoContent, nil)
}
// --- helpers ---
func validateTrunk(name string, trunkType models.TrunkType, gateway string) error {
if name == "" {
return fmt.Errorf("имя транка обязательно")
}
if gateway == "" {
return fmt.Errorf("gateway обязателен")
}
switch trunkType {
case models.TrunkIngress, models.TrunkBalance, models.TrunkFallback:
break
default:
return fmt.Errorf("некорректный тип транка: %s", trunkType)
}
return nil
}
func findTrunkIndex(trunks []models.Trunk, id string) int {
for i := range trunks {
if trunks[i].ID == id {
return i
}
}
return -1
}
// --- ESL push helpers ---
// eslPushGatewayAdd создаёт gateway на FS для ingress-транка.
func (a *API) eslPushGatewayAdd(trunk models.Trunk) {
if a.eslClient == nil || !a.eslClient.IsConnected() || trunk.Type != models.TrunkIngress {
return
}
profile, name, proxy := a.gatewayParams(trunk)
if err := a.eslClient.GatewayAdd(profile, name, proxy); err != nil {
log.Printf("[esl] GatewayAdd %s: %v", name, err)
}
}
// eslPushGatewayUpdate обновляет gateway на FS (удаляет старый + создаёт новый).
func (a *API) eslPushGatewayUpdate(trunk models.Trunk) {
if a.eslClient == nil || !a.eslClient.IsConnected() || trunk.Type != models.TrunkIngress {
return
}
profile, name, proxy := a.gatewayParams(trunk)
_ = a.eslClient.GatewayDelete(profile, name) // игнорируем ошибку — может не существовать
if err := a.eslClient.GatewayAdd(profile, name, proxy); err != nil {
log.Printf("[esl] GatewayUpdate %s: %v", name, err)
}
}
// eslPushGatewayDelete удаляет gateway на FS для ingress-транка.
func (a *API) eslPushGatewayDelete(trunk models.Trunk) {
if a.eslClient == nil || !a.eslClient.IsConnected() || trunk.Type != models.TrunkIngress {
return
}
profile, name, _ := a.gatewayParams(trunk)
if err := a.eslClient.GatewayDelete(profile, name); err != nil {
log.Printf("[esl] GatewayDelete %s: %v", name, err)
}
}
// gatewayParams возвращает параметры для создания FS gateway на основе транка.
func (a *API) gatewayParams(trunk models.Trunk) (profile, name, proxy string) {
profile = a.readConfig().ESL.SofiaProfile
name = fmt.Sprintf("pulse-ingress-%s", trunk.ID)
proxy = extractHost(trunk.Gateway)
return
}
// extractHost извлекает хост из SIP URI.
// "sip:mts-gw.lan:5060" → "mts-gw.lan"
func extractHost(gateway string) string {
s := strings.TrimPrefix(gateway, "sip:")
s = strings.TrimPrefix(s, "sips:")
if idx := strings.LastIndex(s, ":"); idx > 0 {
// Проверяем, порт ли это (все цифры после двоеточия)
port := s[idx+1:]
isNumeric := true
for _, c := range port {
if c < '0' || c > '9' {
isNumeric = false
break
}
}
if isNumeric {
s = s[:idx]
}
}
return s
}
// readConfig возвращает текущий конфиг (для получения esl.sofia_profile).
func (a *API) readConfig() *config.Config {
cfg, _ := a.configManager.ReadConfig()
if cfg == nil {
return &config.Config{ESL: config.ESLConfig{SofiaProfile: "external"}}
}
return cfg
}
+241
View File
@@ -0,0 +1,241 @@
package api
import (
"crypto/rand"
"encoding/hex"
"fmt"
"net/http"
"time"
"github.com/pulse-lets-go/internal/models"
"golang.org/x/crypto/bcrypt"
)
// handleGetUsers — GET /api/users — список пользователей.
func (a *API) handleGetUsers(w http.ResponseWriter, r *http.Request) {
users, err := a.configManager.ReadUsers()
if err != nil {
writeError(w, http.StatusInternalServerError, "ошибка чтения пользователей")
return
}
// Не возвращаем хэши паролей
type safeUser struct {
ID string `json:"id"`
Username string `json:"username"`
Role models.UserRole `json:"role"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
result := make([]safeUser, len(users))
for i, u := range users {
result[i] = safeUser{
ID: u.ID,
Username: u.Username,
Role: u.Role,
CreatedAt: u.CreatedAt,
UpdatedAt: u.UpdatedAt,
}
}
writeJSON(w, http.StatusOK, result)
}
// handleCreateUser — POST /api/users — создание пользователя.
func (a *API) handleCreateUser(w http.ResponseWriter, r *http.Request) {
var req models.CreateUserRequest
if err := decodeJSON(r, &req); err != nil {
writeError(w, http.StatusBadRequest, "некорректный JSON")
return
}
if req.Username == "" || req.Password == "" {
writeError(w, http.StatusBadRequest, "имя и пароль обязательны")
return
}
if req.Role != models.RoleAdmin && req.Role != models.RoleViewer {
writeError(w, http.StatusBadRequest, "роль должна быть admin или viewer")
return
}
users, err := a.configManager.ReadUsers()
if err != nil {
writeError(w, http.StatusInternalServerError, "ошибка чтения пользователей")
return
}
// Проверка уникальности имени
for _, u := range users {
if u.Username == req.Username {
writeError(w, http.StatusConflict, "пользователь с таким именем уже существует")
return
}
}
hash, err := bcrypt.GenerateFromPassword([]byte(req.Password), bcrypt.DefaultCost)
if err != nil {
writeError(w, http.StatusInternalServerError, "ошибка хеширования пароля")
return
}
now := time.Now().UTC()
user := models.User{
ID: generateID("user"),
Username: req.Username,
PasswordHash: string(hash),
Role: req.Role,
CreatedAt: now,
UpdatedAt: now,
}
users = append(users, user)
if err := a.configManager.SaveUsers(users); err != nil {
writeError(w, http.StatusInternalServerError, "ошибка сохранения пользователя")
return
}
writeJSON(w, http.StatusCreated, map[string]string{
"id": user.ID,
"username": user.Username,
"role": string(user.Role),
})
}
// handleUpdateUser — PUT /api/users/{id} — обновление пользователя.
func (a *API) handleUpdateUser(w http.ResponseWriter, r *http.Request) {
userID := r.PathValue("id")
if userID == "" {
writeError(w, http.StatusBadRequest, "не указан id пользователя")
return
}
var req models.UpdateUserRequest
if err := decodeJSON(r, &req); err != nil {
writeError(w, http.StatusBadRequest, "некорректный JSON")
return
}
users, err := a.configManager.ReadUsers()
if err != nil {
writeError(w, http.StatusInternalServerError, "ошибка чтения пользователей")
return
}
idx := findUserIndex(users, userID)
if idx < 0 {
writeError(w, http.StatusNotFound, "пользователь не найден")
return
}
u := &users[idx]
if req.Username != nil {
// Проверка уникальности
for i, other := range users {
if other.Username == *req.Username && i != idx {
writeError(w, http.StatusConflict, "пользователь с таким именем уже существует")
return
}
}
u.Username = *req.Username
}
if req.Password != nil {
hash, _ := bcrypt.GenerateFromPassword([]byte(*req.Password), bcrypt.DefaultCost)
u.PasswordHash = string(hash)
}
if req.Role != nil {
if *req.Role != models.RoleAdmin && *req.Role != models.RoleViewer {
writeError(w, http.StatusBadRequest, "роль должна быть admin или viewer")
return
}
// Защита первичного администратора от понижения роли
if userID == "user-01" && *req.Role != models.RoleAdmin {
writeError(w, http.StatusConflict, "роль первичного администратора нельзя изменить")
return
}
u.Role = *req.Role
}
u.UpdatedAt = time.Now().UTC()
if err := a.configManager.SaveUsers(users); err != nil {
writeError(w, http.StatusInternalServerError, "ошибка сохранения пользователя")
return
}
writeJSON(w, http.StatusOK, map[string]string{
"id": u.ID,
"username": u.Username,
"role": string(u.Role),
})
}
// handleDeleteUser — DELETE /api/users/{id} — удаление пользователя.
func (a *API) handleDeleteUser(w http.ResponseWriter, r *http.Request) {
userID := r.PathValue("id")
if userID == "" {
writeError(w, http.StatusBadRequest, "не указан id пользователя")
return
}
users, err := a.configManager.ReadUsers()
if err != nil {
writeError(w, http.StatusInternalServerError, "ошибка чтения пользователей")
return
}
idx := findUserIndex(users, userID)
if idx < 0 {
writeError(w, http.StatusNotFound, "пользователь не найден")
return
}
// Защита первичного администратора от удаления
if userID == "user-01" {
writeError(w, http.StatusConflict, "первичного администратора нельзя удалить")
return
}
// Не даём удалить последнего админа
adminCount := 0
for _, u := range users {
if u.Role == models.RoleAdmin {
adminCount++
}
}
if users[idx].Role == models.RoleAdmin && adminCount <= 1 {
writeError(w, http.StatusConflict, "нельзя удалить последнего администратора")
return
}
users = append(users[:idx], users[idx+1:]...)
if err := a.configManager.SaveUsers(users); err != nil {
writeError(w, http.StatusInternalServerError, "ошибка сохранения пользователей")
return
}
writeJSON(w, http.StatusNoContent, nil)
}
// findUserIndex находит индекс пользователя по ID.
func findUserIndex(users []models.User, id string) int {
for i := range users {
if users[i].ID == id {
return i
}
}
return -1
}
// generateID генерирует ID вида "prefix-XXXXX".
func generateID(prefix string) string {
return fmt.Sprintf("%s-%s", prefix, randomHex(6))
}
// randomHex генерирует случайную hex-строку заданной длины.
func randomHex(n int) string {
b := make([]byte, n)
if _, err := rand.Read(b); err != nil {
// fallback на time-based, если crypto/rand недоступен
return fmt.Sprintf("%x", time.Now().UnixNano())
}
return hex.EncodeToString(b)[:n]
}
+161
View File
@@ -0,0 +1,161 @@
package api
import (
"fmt"
"log"
"net/http"
"sync"
"github.com/golang-jwt/jwt/v5"
"github.com/gorilla/websocket"
"github.com/pulse-lets-go/internal/models"
)
var upgrader = websocket.Upgrader{
ReadBufferSize: 1024,
WriteBufferSize: 1024,
CheckOrigin: func(r *http.Request) bool {
return true // Разрешаем все origins для dev
},
}
// wsClient — соединение WebSocket одного клиента.
type wsClient struct {
conn *websocket.Conn
mu sync.Mutex
}
// wsHub управляет подключёнными WebSocket-клиентами и рассылает метрики.
type wsHub struct {
mu sync.RWMutex
clients map[*wsClient]bool
}
// newWSHub создаёт хаб WebSocket.
func newWSHub() *wsHub {
return &wsHub{
clients: make(map[*wsClient]bool),
}
}
// add добавляет клиента в хаб.
func (h *wsHub) add(c *wsClient) {
h.mu.Lock()
h.clients[c] = true
h.mu.Unlock()
}
// remove удаляет клиента из хаба.
func (h *wsHub) remove(c *wsClient) {
h.mu.Lock()
if h.clients[c] {
c.conn.Close()
delete(h.clients, c)
}
h.mu.Unlock()
}
// broadcast рассылает метрики всем подключённым клиентам.
func (h *wsHub) broadcast(msg *models.WsMetricsMessage) {
h.mu.RLock()
defer h.mu.RUnlock()
for client := range h.clients {
client.mu.Lock()
err := client.conn.WriteJSON(msg)
client.mu.Unlock()
if err != nil {
log.Printf("[ws] ошибка отправки: %v", err)
go h.remove(client)
}
}
}
// handleWS — WebSocket /ws/metrics.
func (a *API) handleWS(w http.ResponseWriter, r *http.Request) {
token := r.URL.Query().Get("token")
if token == "" {
// пробуем заголовок
token = extractBearerToken(r)
}
if token == "" {
writeError(w, http.StatusUnauthorized, "требуется токен")
return
}
// Валидация JWT
userInfo, err := a.validateToken(token)
if err != nil {
writeError(w, http.StatusUnauthorized, "недействительный токен")
return
}
conn, err := upgrader.Upgrade(w, r, nil)
if err != nil {
log.Printf("[ws] ошибка upgrade: %v", err)
return
}
client := &wsClient{conn: conn}
a.wsHub.add(client)
log.Printf("[ws] клиент подключён: %s (%s)", userInfo.Username, r.RemoteAddr)
// Отправляем текущее состояние при подключении
go func() {
nodes := a.engine.GetAllNodes()
msg := &models.WsMetricsMessage{
Type: "nodes_update",
Payload: nodes,
}
a.wsHub.broadcast(msg)
}()
// Читаем из вебсокета (ping/pong + закрытие)
go func() {
defer a.wsHub.remove(client)
for {
_, _, err := conn.ReadMessage()
if err != nil {
log.Printf("[ws] клиент отключён: %s", r.RemoteAddr)
return
}
}
}()
}
// BroadcastMetrics рассылает метрики всем WS-клиентам (вызывается engine при обновлении).
func (a *API) BroadcastMetrics() {
nodes := a.engine.GetAllNodes()
msg := &models.WsMetricsMessage{
Type: "nodes_update",
Payload: nodes,
}
a.wsHub.broadcast(msg)
}
// validateToken проверяет JWT и возвращает UserInfo.
func (a *API) validateToken(tokenStr string) (*UserInfo, error) {
token, err := parseJWT(tokenStr, a.jwtSecret)
if err != nil {
return nil, err
}
return token, nil
}
// parseJWT парсит JWT токен.
func parseJWT(tokenStr string, secret string) (*UserInfo, error) {
c := &claims{}
token, err := jwt.ParseWithClaims(tokenStr, c, func(token *jwt.Token) (interface{}, error) {
return []byte(secret), nil
})
if err != nil || !token.Valid {
return nil, fmt.Errorf("недействительный токен: %w", err)
}
return &UserInfo{
UserID: c.UserID,
Username: c.Username,
Role: c.Role,
}, nil
}
+420
View File
@@ -0,0 +1,420 @@
// Package config управляет JSON-конфигами приложения: config.json, trunks.json, users.json.
// Обеспечивает атомарную запись (tmp → rename) и thread-safe доступ через sync.RWMutex.
package config
import (
"crypto/rand"
"encoding/hex"
"encoding/json"
"fmt"
"math"
"os"
"path/filepath"
"strings"
"sync"
"time"
"github.com/pulse-lets-go/internal/models"
)
// ScoringWeights — веса компонентов скора.
type ScoringWeights struct {
CallScore float64 `json:"call_score"`
LoadScore float64 `json:"load_score"`
IdleScore float64 `json:"idle_score"`
FailScore float64 `json:"fail_score"`
}
// ScoringConfig — пороги и веса для scoring engine.
type ScoringConfig struct {
IdleCPUMin float64 `json:"idle_cpu_min"`
CallFailureRateLethal float64 `json:"call_failure_rate_lethal"`
Weights ScoringWeights `json:"weights"`
}
// TLSConfig — настройки HTTPS.
type TLSConfig struct {
Enabled bool `json:"enabled"`
CertFile string `json:"cert_file"`
KeyFile string `json:"key_file"`
}
// RateLimitConfig — ограничения частоты запросов.
type RateLimitConfig struct {
Enabled bool `json:"enabled"`
RoutePerSec int `json:"route_per_sec"`
APIPerSec int `json:"api_per_sec"`
}
// LogConfig — настройки логирования.
type LogConfig struct {
Level string `json:"level"` // debug, info, warn, error
Format string `json:"format"` // text, json
}
// ESLConfig — настройки подключения к FreeSWITCH ESL.
type ESLConfig struct {
Host string `json:"host"` // если пустой — ESL не запускается
Port int `json:"port"` // порт ESL (по умолчанию 8021)
Password string `json:"password"`
PasswordEnv string `json:"password_env"` // имя переменной окружения с паролем
SofiaProfile string `json:"sofia_profile"` // профиль Sofia (external/internal)
GatewayPrefix string `json:"gateway_prefix"` // префикс gateway (pulse)
}
// GetPassword возвращает ESL-пароль: из переменной окружения, если задан password_env, иначе из поля password.
func (e *ESLConfig) GetPassword() string {
if e.PasswordEnv != "" {
if val := os.Getenv(e.PasswordEnv); val != "" {
return val
}
}
return e.Password
}
// Config — корневая конфигурация приложения (config.json).
type Config struct {
NatsURL string `json:"nats_url"`
NatsUser string `json:"nats_user"`
NatsPassword string `json:"nats_password"`
ListenAddr string `json:"listen_addr"`
JWTSecret string `json:"jwt_secret"`
MonitoringAPIKey string `json:"monitoring_api_key"`
StaleThresholdSec int `json:"stale_threshold_sec"`
Scoring ScoringConfig `json:"scoring"`
TLS TLSConfig `json:"tls"`
RateLimit RateLimitConfig `json:"rate_limit"`
Log LogConfig `json:"log"`
ESL ESLConfig `json:"esl"`
}
// Validate проверяет корректность всех полей конфига.
// Возвращает читаемое описание всех найденных ошибок.
func (c *Config) Validate() error {
var errs []string
if c.NatsURL == "" {
errs = append(errs, "nats_url обязателен")
}
if c.ListenAddr == "" {
errs = append(errs, "listen_addr обязателен")
}
if c.StaleThresholdSec <= 0 {
errs = append(errs, "stale_threshold_sec должен быть > 0")
}
if c.JWTSecret == "" {
errs = append(errs, "jwt_secret обязателен")
}
if c.MonitoringAPIKey == "" {
errs = append(errs, "monitoring_api_key обязателен")
}
if c.Scoring.IdleCPUMin < 0 {
errs = append(errs, "scoring.idle_cpu_min должен быть >= 0")
}
if c.Scoring.CallFailureRateLethal < 0 || c.Scoring.CallFailureRateLethal > 100 {
errs = append(errs, "scoring.call_failure_rate_lethal должен быть 0..100")
}
w := &c.Scoring.Weights
sum := w.CallScore + w.LoadScore + w.IdleScore + w.FailScore
if math.Abs(sum-1.0) > 0.01 {
errs = append(errs, fmt.Sprintf("scoring.weights сумма должна быть ≈1.0 (сейчас %.2f)", sum))
}
if c.TLS.Enabled {
if c.TLS.CertFile == "" || c.TLS.KeyFile == "" {
errs = append(errs, "tls.cert_file и tls.key_file обязательны при tls.enabled=true")
}
}
if c.RateLimit.Enabled {
if c.RateLimit.RoutePerSec <= 0 {
errs = append(errs, "rate_limit.route_per_sec должен быть > 0")
}
if c.RateLimit.APIPerSec <= 0 {
errs = append(errs, "rate_limit.api_per_sec должен быть > 0")
}
}
switch c.Log.Level {
case "debug", "info", "warn", "error", "":
default:
errs = append(errs, "log.level должен быть debug/info/warn/error")
}
switch c.Log.Format {
case "text", "json", "":
default:
errs = append(errs, "log.format должен быть text/json")
}
// ESL validation
if c.ESL.Host != "" {
if c.ESL.Port <= 0 || c.ESL.Port > 65535 {
errs = append(errs, "esl.port должен быть 1-65535")
}
if c.ESL.SofiaProfile == "" {
errs = append(errs, "esl.sofia_profile обязателен при esl.host")
}
if c.ESL.Password == "" && c.ESL.PasswordEnv == "" {
errs = append(errs, "esl.password или esl.password_env обязателен при esl.host")
}
}
if len(errs) > 0 {
return fmt.Errorf("ошибки конфигурации:\n - %s", strings.Join(errs, "\n - "))
}
return nil
}
// Manager управляет чтением и атомарной записью JSON-конфигов.
type Manager struct {
mu sync.RWMutex
dir string // путь к data/
}
// NewManager создаёт новый менеджер конфигов из директории dataDir.
// Если файлы отсутствуют — создаёт их с дефолтными значениями.
func NewManager(dataDir string) (*Manager, error) {
// Убеждаемся, что директория существует
if err := os.MkdirAll(dataDir, 0750); err != nil {
return nil, fmt.Errorf("создание data/: %w", err)
}
m := &Manager{dir: dataDir}
// Инициализируем конфиги при первом запуске
if _, err := os.Stat(m.configPath()); os.IsNotExist(err) {
if err := m.createDefaultConfig(); err != nil {
return nil, fmt.Errorf("создание config.json: %w", err)
}
}
if _, err := os.Stat(m.trunksPath()); os.IsNotExist(err) {
if err := m.saveFile(m.trunksPath(), []models.Trunk{}); err != nil {
return nil, fmt.Errorf("создание trunks.json: %w", err)
}
}
if _, err := os.Stat(m.usersPath()); os.IsNotExist(err) {
if err := m.createDefaultUsers(); err != nil {
return nil, fmt.Errorf("создание users.json: %w", err)
}
}
return m, nil
}
// ReadConfig читает config.json. Потокобезопасно.
func (m *Manager) ReadConfig() (*Config, error) {
m.mu.RLock()
defer m.mu.RUnlock()
data, err := os.ReadFile(m.configPath())
if err != nil {
return nil, fmt.Errorf("чтение config.json: %w", err)
}
var cfg Config
if err := json.Unmarshal(data, &cfg); err != nil {
return nil, fmt.Errorf("парсинг config.json: %w", err)
}
return &cfg, nil
}
// SaveConfig атомарно сохраняет config.json.
func (m *Manager) SaveConfig(cfg *Config) error {
m.mu.Lock()
defer m.mu.Unlock()
data, err := json.MarshalIndent(cfg, "", " ")
if err != nil {
return fmt.Errorf("сериализация config.json: %w", err)
}
return m.saveFile(m.configPath(), data)
}
// ReadTrunks читает trunks.json. Потокобезопасно.
func (m *Manager) ReadTrunks() ([]models.Trunk, error) {
m.mu.RLock()
defer m.mu.RUnlock()
return m.readTrunksLocked()
}
// readTrunksLocked — внутреннее чтение без блокировки (вызывающий держит lock).
func (m *Manager) readTrunksLocked() ([]models.Trunk, error) {
data, err := os.ReadFile(m.trunksPath())
if err != nil {
return nil, fmt.Errorf("чтение trunks.json: %w", err)
}
var trunks []models.Trunk
if err := json.Unmarshal(data, &trunks); err != nil {
return nil, fmt.Errorf("парсинг trunks.json: %w", err)
}
return trunks, nil
}
// SaveTrunks атомарно сохраняет trunks.json.
func (m *Manager) SaveTrunks(trunks []models.Trunk) error {
m.mu.Lock()
defer m.mu.Unlock()
return m.saveTrunksLocked(trunks)
}
func (m *Manager) saveTrunksLocked(trunks []models.Trunk) error {
data, err := json.MarshalIndent(trunks, "", " ")
if err != nil {
return fmt.Errorf("сериализация trunks.json: %w", err)
}
return m.saveFile(m.trunksPath(), data)
}
// ReadUsers читает users.json. Потокобезопасно.
func (m *Manager) ReadUsers() ([]models.User, error) {
m.mu.RLock()
defer m.mu.RUnlock()
return m.readUsersLocked()
}
func (m *Manager) readUsersLocked() ([]models.User, error) {
data, err := os.ReadFile(m.usersPath())
if err != nil {
return nil, fmt.Errorf("чтение users.json: %w", err)
}
var users []models.User
if err := json.Unmarshal(data, &users); err != nil {
return nil, fmt.Errorf("парсинг users.json: %w", err)
}
return users, nil
}
// SaveUsers атомарно сохраняет users.json.
func (m *Manager) SaveUsers(users []models.User) error {
m.mu.Lock()
defer m.mu.Unlock()
return m.saveUsersLocked(users)
}
func (m *Manager) saveUsersLocked(users []models.User) error {
data, err := json.MarshalIndent(users, "", " ")
if err != nil {
return fmt.Errorf("сериализация users.json: %w", err)
}
return m.saveFile(m.usersPath(), data)
}
// --- Приватные методы ---
func (m *Manager) configPath() string { return filepath.Join(m.dir, "config.json") }
func (m *Manager) trunksPath() string { return filepath.Join(m.dir, "trunks.json") }
func (m *Manager) usersPath() string { return filepath.Join(m.dir, "users.json") }
// saveFile атомарно сохраняет данные в файл: tmp → rename.
// Вызывающий держит write lock.
func (m *Manager) saveFile(path string, data interface{}) error {
var raw []byte
var err error
switch v := data.(type) {
case []byte:
raw = v
default:
raw, err = json.MarshalIndent(data, "", " ")
if err != nil {
return fmt.Errorf("сериализация: %w", err)
}
}
tmpPath := path + ".tmp"
if err := os.WriteFile(tmpPath, raw, 0640); err != nil {
return fmt.Errorf("запись временного файла: %w", err)
}
if err := os.Rename(tmpPath, path); err != nil {
os.Remove(tmpPath) // clean up
return fmt.Errorf("атомарный rename: %w", err)
}
return nil
}
// createDefaultConfig генерирует config.json со значениями по умолчанию.
func (m *Manager) createDefaultConfig() error {
jwtSecret, err := generateRandomHex(32)
if err != nil {
return err
}
apiKey, err := generateRandomHex(32)
if err != nil {
return err
}
cfg := Config{
NatsURL: "nats://localhost:4222",
NatsUser: "",
NatsPassword: "",
ListenAddr: ":8080",
JWTSecret: jwtSecret,
MonitoringAPIKey: apiKey,
StaleThresholdSec: 20,
Scoring: ScoringConfig{
IdleCPUMin: 5.0,
CallFailureRateLethal: 15.0,
Weights: ScoringWeights{
CallScore: 0.40,
LoadScore: 0.30,
IdleScore: 0.20,
FailScore: 0.10,
},
},
TLS: TLSConfig{
Enabled: false,
CertFile: "",
KeyFile: "",
},
RateLimit: RateLimitConfig{
Enabled: true,
RoutePerSec: 5000,
APIPerSec: 100,
},
Log: LogConfig{
Level: "info",
Format: "text",
},
ESL: ESLConfig{
Host: "",
Port: 8021,
Password: "",
PasswordEnv: "",
SofiaProfile: "external",
GatewayPrefix: "pulse",
},
}
return m.SaveConfig(&cfg)
}
// createDefaultUsers создаёт users.json с дефолтным админом (admin/admin).
// Пароль хранится в bcrypt. На первом запуске запрашивается смена.
func (m *Manager) createDefaultUsers() error {
// Используем плейсхолдер-хэш "admin" — при первом логине админ обязан сменить пароль.
// Хэш будет заменён на bcrypt при создании/обновлении через API.
defaultTime := time.Now().UTC()
users := []models.User{
{
ID: "user-01",
Username: "admin",
Role: models.RoleAdmin,
CreatedAt: defaultTime,
UpdatedAt: defaultTime,
// PasswordHash будет установлен при первом логине через API
},
}
return m.saveUsersLocked(users)
}
// generateRandomHex генерирует случайную hex-строку заданной длины в байтах.
func generateRandomHex(bytes int) (string, error) {
buf := make([]byte, bytes)
if _, err := rand.Read(buf); err != nil {
return "", fmt.Errorf("генерация случайных байт: %w", err)
}
return hex.EncodeToString(buf), nil
}
+331
View File
@@ -0,0 +1,331 @@
// Package engine содержит scoring engine и router выбора лучшей ноды.
// Потокобезопасен: RWMutex разделяет чтение (роутинг) и запись (обновление метрик).
package engine
import (
"fmt"
"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
lastMetricTime atomic.Int64 // unix ts последней полученной метрики
}
// NewEngine создаёт новый engine с заданной конфигурацией.
func NewEngine(cfg *config.Config) *Engine {
return &Engine{
nodes: make(map[string]*models.NodeState),
rings: make(map[string]*MetricsRingBuffer),
cfg: cfg,
bestScore: -1,
startTime: time.Now(),
}
}
// 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]
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)
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()
if len(e.nodes) == 0 {
return "", 0, true
}
return e.bestNodeID, e.bestScore, e.fallbackActive
}
// 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()
}
// 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)
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)
}
// HealthStats возвращает статистику для health endpoint.
type HealthStats struct {
TotalNodes int `json:"total_nodes"`
HealthyNodes int `json:"healthy_nodes"`
RouteRequests int64 `json:"route_requests_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(),
UptimeSeconds: int64(time.Since(e.startTime).Seconds()),
LastMetricTS: e.lastMetricTime.Load(),
}
}
// --- Приватные методы ---
// scoreNodeLocked вычисляет score для одной ноды.
// Вызывающий держит write lock.
func (e *Engine) scoreNodeLocked(ns *models.NodeState) {
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 = ""
callScore := clamp(100.0-(float64(ns.ActiveCalls)/float64(ns.MaxCalls)*100.0), 0, 100)
loadScore := clamp(100.0-(ns.LoadAvg*50.0), 0, 100)
idleScore := clamp(ns.IdleCPU, 0, 100)
failScore := clamp(100.0-ns.CallFailureRate, 0, 100)
w := &sc.Weights
ns.Score = callScore*w.CallScore + loadScore*w.LoadScore + idleScore*w.IdleScore + failScore*w.FailScore
}
// 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)
}
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
}
+340
View File
@@ -0,0 +1,340 @@
package engine
import (
"testing"
"time"
"github.com/pulse-lets-go/internal/config"
"github.com/pulse-lets-go/internal/models"
)
func defaultCfg() *config.Config {
return &config.Config{
StaleThresholdSec: 20,
Scoring: config.ScoringConfig{
IdleCPUMin: 5.0,
CallFailureRateLethal: 15.0,
Weights: config.ScoringWeights{
CallScore: 0.40,
LoadScore: 0.30,
IdleScore: 0.20,
FailScore: 0.10,
},
},
}
}
func freshMetric(nodeID string) *models.NodeMetric {
return &models.NodeMetric{
NodeID: nodeID,
TS: time.Now().Unix(),
Status: "ok",
ActiveCalls: 42,
MaxCalls: 250,
IdleCPU: 35,
LoadAvg: 0.85,
CallFailureRate: 0.5,
}
}
func TestNewEngine(t *testing.T) {
eng := NewEngine(defaultCfg())
if eng == nil {
t.Fatal("ожидался engine, получен nil")
}
}
func TestUpdateMetric_CreatesNode(t *testing.T) {
eng := NewEngine(defaultCfg())
m := freshMetric("pbx-01")
ns := eng.UpdateMetric(m)
if ns.NodeID != "pbx-01" {
t.Errorf("ожидался node_id=pbx-01, получен %s", ns.NodeID)
}
if len(eng.nodes) != 1 {
t.Errorf("ожидалась 1 нода, получено %d", len(eng.nodes))
}
}
func TestUpdateMetric_PreservesDisabled(t *testing.T) {
eng := NewEngine(defaultCfg())
m := freshMetric("pbx-01")
eng.UpdateMetric(m)
// Отключаем ноду вручную
eng.ToggleNode("pbx-01", true, "test")
eng.UpdateMetric(m)
ns, _ := eng.GetNodeInfo("pbx-01")
if !ns.Disabled {
t.Error("ожидалось disabled=true")
}
if ns.DisabledReason != "test" {
t.Errorf("ожидалась причина 'test', получена '%s'", ns.DisabledReason)
}
// Включаем обратно
eng.ToggleNode("pbx-01", false, "")
eng.UpdateMetric(m)
ns, _ = eng.GetNodeInfo("pbx-01")
if ns.Disabled {
t.Error("ожидалось disabled=false")
}
}
func TestScore_LethalDisabled(t *testing.T) {
eng := NewEngine(defaultCfg())
m := freshMetric("pbx-01")
eng.UpdateMetric(m)
eng.ToggleNode("pbx-01", true, "maintenance")
ns, _ := eng.GetNodeInfo("pbx-01")
if ns.Score != -100 {
t.Errorf("disabled нода должна иметь score=-100, получен %.0f", ns.Score)
}
}
func TestScore_LethalStatusNotOk(t *testing.T) {
eng := NewEngine(defaultCfg())
m := freshMetric("pbx-01")
m.Status = "down"
eng.UpdateMetric(m)
ns, _ := eng.GetNodeInfo("pbx-01")
if ns.Score != -100 {
t.Errorf("status==down нода должна иметь score=-100, получен %.0f", ns.Score)
}
}
func TestScore_LethalStale(t *testing.T) {
eng := NewEngine(defaultCfg())
m := freshMetric("pbx-01")
// Устанавливаем ts на 30 секунд назад (stale_threshold=20)
m.TS = time.Now().Unix() - 30
eng.UpdateMetric(m)
ns, _ := eng.GetNodeInfo("pbx-01")
if ns.Score != -100 {
t.Errorf("stale нода должна иметь score=-100, получен %.0f", ns.Score)
}
}
func TestScore_LethalMaxCalls(t *testing.T) {
eng := NewEngine(defaultCfg())
m := freshMetric("pbx-01")
m.ActiveCalls = 250
m.MaxCalls = 250
eng.UpdateMetric(m)
ns, _ := eng.GetNodeInfo("pbx-01")
if ns.Score != -100 {
t.Errorf("нода с исчерпанной ёмкостью должна иметь score=-100, получен %.0f", ns.Score)
}
}
func TestScore_LethalIdleCPU(t *testing.T) {
eng := NewEngine(defaultCfg())
m := freshMetric("pbx-01")
m.IdleCPU = 2.0
eng.UpdateMetric(m)
ns, _ := eng.GetNodeInfo("pbx-01")
if ns.Score != -100 {
t.Errorf("нода с idle_cpu < 5 должна иметь score=-100, получен %.0f", ns.Score)
}
}
func TestScore_LethalFailureRate(t *testing.T) {
eng := NewEngine(defaultCfg())
m := freshMetric("pbx-01")
m.CallFailureRate = 20.0
eng.UpdateMetric(m)
ns, _ := eng.GetNodeInfo("pbx-01")
if ns.Score != -100 {
t.Errorf("нода с call_failure_rate > 15 должна иметь score=-100, получен %.0f", ns.Score)
}
}
func TestScore_Healthy(t *testing.T) {
eng := NewEngine(defaultCfg())
m := freshMetric("pbx-01")
ns := eng.UpdateMetric(m)
if ns.Score <= 0 {
t.Errorf("здоровая нода должна иметь положительный score, получен %.0f", ns.Score)
}
}
func TestScore_HealthyFormula(t *testing.T) {
eng := NewEngine(defaultCfg())
m := freshMetric("pbx-01")
m.ActiveCalls = 0
m.MaxCalls = 100
m.IdleCPU = 100
m.LoadAvg = 0
m.CallFailureRate = 0
ns := eng.UpdateMetric(m)
// Все компоненты = 100
// score = 100*0.4 + 100*0.3 + 100*0.2 + 100*0.1 = 100
if ns.Score != 100.0 {
t.Errorf("идеальная нода должна иметь score=100, получен %.1f", ns.Score)
}
}
func TestScore_WeightedFormula(t *testing.T) {
eng := NewEngine(defaultCfg())
m := freshMetric("pbx-01")
m.ActiveCalls = 50
m.MaxCalls = 100
m.IdleCPU = 50
m.LoadAvg = 1.0
m.CallFailureRate = 5.0
ns := eng.UpdateMetric(m)
// call_score = 100 - (50/100 * 100) = 50
// load_score = 100 - (1.0 * 50) = 50
// idle_score = 50
// fail_score = 100 - 5 = 95
// score = 50*0.4 + 50*0.3 + 50*0.2 + 95*0.1 = 20+15+10+9.5 = 54.5
expected := 54.5
if ns.Score < expected-0.1 || ns.Score > expected+0.1 {
t.Errorf("ожидался score=%.1f, получен %.1f", expected, ns.Score)
}
}
func TestGetBestNode_NoNodes(t *testing.T) {
eng := NewEngine(defaultCfg())
_, _, fallback := eng.GetBestNode()
if !fallback {
t.Error("без нод должен быть fallback")
}
}
func TestGetBestNode_OneNode(t *testing.T) {
eng := NewEngine(defaultCfg())
eng.UpdateMetric(freshMetric("pbx-01"))
id, score, fallback := eng.GetBestNode()
if fallback {
t.Error("не должно быть fallback с одной здоровой нодой")
}
if id != "pbx-01" {
t.Errorf("ожидалась pbx-01, получена %s", id)
}
if score <= 0 {
t.Errorf("ожидался положительный score, получен %.0f", score)
}
}
func TestGetBestNode_AllUnhealthy(t *testing.T) {
eng := NewEngine(defaultCfg())
m1 := freshMetric("pbx-01")
m1.Status = "down"
eng.UpdateMetric(m1)
m2 := freshMetric("pbx-02")
m2.ActiveCalls = 500
m2.MaxCalls = 500
eng.UpdateMetric(m2)
_, _, fallback := eng.GetBestNode()
if !fallback {
t.Error("все ноды unhealthy — должен быть fallback")
}
}
func TestGetBestNode_MultipleNodes(t *testing.T) {
eng := NewEngine(defaultCfg())
// pbx-01: 80% загружена (меньше score)
m1 := freshMetric("pbx-01")
m1.ActiveCalls = 200
m1.MaxCalls = 250
eng.UpdateMetric(m1)
// pbx-02: 20% загружена (выше score)
m2 := freshMetric("pbx-02")
m2.ActiveCalls = 20
m2.MaxCalls = 250
eng.UpdateMetric(m2)
id, _, _ := eng.GetBestNode()
if id != "pbx-02" {
t.Errorf("ожидалась pbx-02 (меньше загрузки), получена %s", id)
}
}
func TestGetNodeHistory(t *testing.T) {
eng := NewEngine(defaultCfg())
eng.UpdateMetric(freshMetric("pbx-01"))
// Повторяем обновление для накопления истории
for i := 0; i < 5; i++ {
m := freshMetric("pbx-01")
m.ActiveCalls = i * 10
eng.UpdateMetric(m)
}
history := eng.GetNodeHistory("pbx-01")
if len(history) < 3 {
t.Errorf("ожидалось минимум 3 снапшота, получено %d", len(history))
}
}
func TestGetNodeHistory_NoNode(t *testing.T) {
eng := NewEngine(defaultCfg())
history := eng.GetNodeHistory("nonexistent")
if history != nil {
t.Error("для несуществующей ноды ожидался nil")
}
}
func TestToggleNode_NotFound(t *testing.T) {
eng := NewEngine(defaultCfg())
err := eng.ToggleNode("nonexistent", true, "test")
if err == nil {
t.Error("ожидалась ошибка для несуществующей ноды")
}
}
func TestUpdateConfig(t *testing.T) {
eng := NewEngine(defaultCfg())
m := freshMetric("pbx-01")
eng.UpdateMetric(m)
// Меняем конфиг на более строгий
newCfg := defaultCfg()
newCfg.Scoring.CallFailureRateLethal = 0.1
newCfg.Scoring.IdleCPUMin = 50
eng.UpdateConfig(newCfg)
ns, _ := eng.GetNodeInfo("pbx-01")
if ns.Score != -100 {
t.Errorf("после ужесточения конфига нода должна стать unhealthy, получен score=%.0f", ns.Score)
}
}
func TestClamp(t *testing.T) {
tests := []struct {
v, min, max, expected float64
}{
{50, 0, 100, 50},
{-10, 0, 100, 0},
{150, 0, 100, 100},
{0, 0, 0, 0},
}
for _, tt := range tests {
got := clamp(tt.v, tt.min, tt.max)
if got != tt.expected {
t.Errorf("clamp(%.0f, %.0f, %.0f) = %.0f, ожидалось %.0f",
tt.v, tt.min, tt.max, got, tt.expected)
}
}
}
+52
View File
@@ -0,0 +1,52 @@
package engine
// MetricsRingBuffer — кольцевой буфер истории метрик для одной ноды.
// Хранит до `capacity` последних снапшотов NodeState.
// Потокобезопасен: используется внутри Engine (защищён engine.mu).
type MetricsRingBuffer struct {
buf []NodeSnapshot
capacity int
head int // позиция следующей записи
size int // текущее количество элементов
}
// NodeSnapshot — снапшот метрик на момент времени.
type NodeSnapshot struct {
ActiveCalls int
MaxCalls int
IdleCPU float64
LoadAvg float64
CallFailureRate float64
Score float64
TS int64
}
// newMetricsRingBuffer создаёт кольцевой буфер заданной ёмкости.
func newMetricsRingBuffer(capacity int) *MetricsRingBuffer {
return &MetricsRingBuffer{
buf: make([]NodeSnapshot, capacity),
capacity: capacity,
}
}
// push добавляет снапшот в буфер. Если буфер заполнен, затирает самый старый.
func (b *MetricsRingBuffer) push(snapshot NodeSnapshot) {
b.buf[b.head] = snapshot
b.head = (b.head + 1) % b.capacity
if b.size < b.capacity {
b.size++
}
}
// snapshot возвращает копию всех снапшотов в хронологическом порядке.
func (b *MetricsRingBuffer) snapshot() []NodeSnapshot {
result := make([]NodeSnapshot, b.size)
start := b.head - b.size
if start < 0 {
start += b.capacity
}
for i := 0; i < b.size; i++ {
result[i] = b.buf[(start+i)%b.capacity]
}
return result
}
+377
View File
@@ -0,0 +1,377 @@
// Package esl реализует клиент FreeSWITCH Event Socket Library (ESL).
// Использует сырой TCP — протокол ESL прост (HTTP-подобные заголовки +
// Content-Length тело), не нужна внешняя библиотека.
//
// Клиент обеспечивает:
// - Подключение и аутентификацию
// - Синхронные команды (api)
// - Подписку на события
// - Автоматический реконнект с exponential backoff
// - Потокобезопасность (mutex на отправку команд)
package esl
import (
"bufio"
"fmt"
"io"
"log"
"net"
"strconv"
"strings"
"sync"
"sync/atomic"
"time"
)
// Event — событие, полученное от FreeSWITCH ESL.
type Event struct {
Name string
Headers map[string]string
Body string
}
// Client — ESL-клиент для управления FreeSWITCH.
type Client struct {
host string
port int
password string
conn net.Conn
reader *bufio.Reader
mu sync.Mutex // защищает отправку команд
connected atomic.Bool
eventsCh chan *Event
closeCh chan struct{}
// Переподключение
shouldReconn atomic.Bool
backoff time.Duration
// Статистика
reconnects atomic.Int64
gatewayOps atomic.Int64
eventsRecv atomic.Int64
startTime time.Time
}
// NewClient создаёт нового ESL-клиента.
func NewClient(host string, port int, password string) *Client {
return &Client{
host: host,
port: port,
password: password,
eventsCh: make(chan *Event, 100),
closeCh: make(chan struct{}),
backoff: 1 * time.Second,
}
}
// Connect подключается к FreeSWITCH и запускает фоновый реконнект.
// Блокируется до первой успешной аутентификации.
func (c *Client) Connect() error {
c.shouldReconn.Store(true)
if err := c.tryConnect(); err != nil {
return fmt.Errorf("esl connect: %w", err)
}
return nil
}
// ConnectWithRetry подключается в фоне с бесконечными попытками.
func (c *Client) ConnectWithRetry() {
c.shouldReconn.Store(true)
for {
err := c.tryConnect()
if err == nil {
return // успех
}
if !c.shouldReconn.Load() {
return // остановка
}
log.Printf("[esl] ошибка подключения: %v, повтор через %v", err, c.backoff)
time.Sleep(c.backoff)
c.backoff *= 2
if c.backoff > 60*time.Second {
c.backoff = 60 * time.Second
}
}
}
// tryConnect выполняет одну попытку подключения и аутентификации.
func (c *Client) tryConnect() error {
c.mu.Lock()
defer c.mu.Unlock()
// Закрываем предыдущее соединение если было
if c.conn != nil {
c.conn.Close()
c.closeCh = make(chan struct{})
}
addr := fmt.Sprintf("%s:%d", c.host, c.port)
dialer := net.Dialer{Timeout: 5 * time.Second}
conn, err := dialer.Dial("tcp", addr)
if err != nil {
return fmt.Errorf("dial %s: %w", addr, err)
}
c.conn = conn
c.reader = bufio.NewReader(conn)
// 1. Читаем auth/request
headers, _, err := c.readMessageLocked()
if err != nil {
conn.Close()
return fmt.Errorf("чтение auth/request: %w", err)
}
if headers["Content-Type"] != "auth/request" {
conn.Close()
return fmt.Errorf("ожидался auth/request, получен %s", headers["Content-Type"])
}
// 2. Отправляем пароль
authCmd := fmt.Sprintf("auth %s\n\n", c.password)
if _, err := conn.Write([]byte(authCmd)); err != nil {
conn.Close()
return fmt.Errorf("отправка auth: %w", err)
}
// 3. Читаем auth response
headers, _, err = c.readMessageLocked()
if err != nil {
conn.Close()
return fmt.Errorf("чтение auth response: %w", err)
}
if reply := headers["Reply-Text"]; !strings.HasPrefix(reply, "+OK") {
conn.Close()
return fmt.Errorf("auth rejected: %s", reply)
}
c.connected.Store(true)
c.startTime = time.Now()
c.backoff = 1 * time.Second // сброс backoff после успешного коннекта
endpoint := fmt.Sprintf("%s:%d", c.host, c.port)
log.Printf("[esl] подключён к %s", endpoint)
return nil
}
// Disconnect отключает клиент и останавливает реконнект.
func (c *Client) Disconnect() {
c.shouldReconn.Store(false)
c.mu.Lock()
defer c.mu.Unlock()
c.connected.Store(false)
if c.conn != nil {
c.conn.Close()
c.conn = nil
}
close(c.closeCh)
}
// IsConnected возвращает статус подключения.
func (c *Client) IsConnected() bool {
return c.connected.Load()
}
// Send отправляет синхронную команду и возвращает тело ответа.
// Пример: Send("api sofia profile external gwlist") → список gateway.
func (c *Client) Send(command string) (map[string]string, string, error) {
c.mu.Lock()
defer c.mu.Unlock()
if !c.connected.Load() || c.conn == nil {
return nil, "", fmt.Errorf("esl не подключён")
}
// Отправляем команду
fullCmd := command + "\n\n"
if _, err := c.conn.Write([]byte(fullCmd)); err != nil {
c.connected.Store(false)
return nil, "", fmt.Errorf("отправка команды: %w", err)
}
// Читаем ответ
return c.readMessageLocked()
}
// Subscribe подписывается на события ESL.
// events — имена событий для фильтра: "SOFIA::gateway_register SOFIA::gateway_unregister".
func (c *Client) Subscribe(events string) error {
cmd := fmt.Sprintf("event plain %s", events)
_, _, err := c.Send(cmd)
if err != nil {
return fmt.Errorf("подписка на события: %w", err)
}
log.Printf("[esl] подписан на события: %s", events)
return nil
}
// Events возвращает канал с входящими событиями.
func (c *Client) Events() <-chan *Event {
return c.eventsCh
}
// StartReadingEvents запускает горутину чтения асинхронных событий FS.
// Должна вызываться после Connect().
func (c *Client) StartReadingEvents() {
if !c.shouldReconn.Load() {
return
}
go c.readEventsLoop()
}
// readEventsLoop читает асинхронные сообщения (события) от FS.
func (c *Client) readEventsLoop() {
for {
select {
case <-c.closeCh:
return
default:
}
if !c.connected.Load() || c.conn == nil {
time.Sleep(500 * time.Millisecond)
continue
}
c.mu.Lock()
headers, body, err := c.readMessageLocked()
c.mu.Unlock()
if err != nil {
if err == io.EOF || strings.Contains(err.Error(), "use of closed network connection") {
c.connected.Store(false)
log.Printf("[esl] соединение разорвано, реконнект...")
c.reconnects.Add(1)
c.reconnect()
continue
}
log.Printf("[esl] ошибка чтения события: %v", err)
time.Sleep(100 * time.Millisecond)
continue
}
eventName := headers["Event-Name"]
if eventName == "" {
continue // не событие (например, ответ на команду)
}
evt := &Event{
Name: eventName,
Headers: headers,
Body: body,
}
c.eventsRecv.Add(1)
select {
case c.eventsCh <- evt:
default:
log.Printf("[esl] канал событий переполнен, пропускаем %s", eventName)
}
}
}
// reconnect выполняет реконнект с backoff.
func (c *Client) reconnect() {
if !c.shouldReconn.Load() {
return
}
log.Printf("[esl] реконнект через %v...", c.backoff)
time.Sleep(c.backoff)
c.mu.Lock()
if c.conn != nil {
c.conn.Close()
c.conn = nil
}
c.mu.Unlock()
if err := c.tryConnect(); err != nil {
log.Printf("[esl] реконнект не удался: %v", err)
c.backoff *= 2
if c.backoff > 60*time.Second {
c.backoff = 60 * time.Second
}
return
}
// Успешный реконнект
c.backoff = 1 * time.Second
log.Printf("[esl] реконнект успешен, переподписываемся на события")
// TODO: gatewaySyncAll() и resubscribe на события вызываются снаружи
}
// --- Приватные методы ---
// readMessageLocked читает одно ESL-сообщение: заголовки + тело.
// Вызывающий держит c.mu.
func (c *Client) readMessageLocked() (map[string]string, string, error) {
headers := make(map[string]string)
for {
line, err := c.reader.ReadString('\n')
if err != nil {
return nil, "", err
}
line = strings.TrimRight(line, "\r\n")
if line == "" {
break // конец заголовков
}
parts := strings.SplitN(line, ":", 2)
if len(parts) == 2 {
headers[strings.TrimSpace(parts[0])] = strings.TrimSpace(parts[1])
}
}
// Читаем тело согласно Content-Length
cl := headers["Content-Length"]
if cl == "" {
return headers, "", nil
}
length, err := strconv.Atoi(cl)
if err != nil {
return headers, "", fmt.Errorf("некорректный Content-Length: %s", cl)
}
body := make([]byte, length)
if _, err := io.ReadFull(c.reader, body); err != nil {
return headers, "", fmt.Errorf("чтение тела (Content-Length=%d): %w", length, err)
}
return headers, strings.TrimSpace(string(body)), nil
}
// --- Метрики ---
// Stats возвращает статистику ESL-клиента для health endpoint.
type Stats struct {
Status string `json:"status"`
Host string `json:"host"`
Reconnects int64 `json:"reconnects"`
GatewayOps int64 `json:"gateway_ops_total"`
EventsRecv int64 `json:"events_received"`
UptimeSec int64 `json:"uptime_seconds"`
}
// GetStats возвращает текущую статистику.
func (c *Client) GetStats() Stats {
st := "connected"
if !c.connected.Load() {
st = "disconnected"
}
return Stats{
Status: st,
Host: fmt.Sprintf("%s:%d", c.host, c.port),
Reconnects: c.reconnects.Load(),
GatewayOps: c.gatewayOps.Load(),
EventsRecv: c.eventsRecv.Load(),
UptimeSec: int64(time.Since(c.startTime).Seconds()),
}
}
// IncGatewayOps увеличивает счётчик операций gateway.
func (c *Client) IncGatewayOps() {
c.gatewayOps.Add(1)
}
+91
View File
@@ -0,0 +1,91 @@
package esl
import (
"context"
"log"
"time"
)
// EventHandler — callback для обработки событий ESL.
// Принимает имя события и заголовки, вызывается из readEventsLoop.
type EventHandler func(eventName string, headers map[string]string, body string)
// StartEventLoop запускает основной цикл реконнекта и чтения событий.
// При успешном подключении:
// - Вызывает onConnect (например, для GatewaySyncAll)
// - Подписывается на события SOFIA::gateway_register/unregister
// - Запускает чтение асинхронных событий
// При разрыве соединения — автоматический реконнект.
func (c *Client) StartEventLoop(ctx context.Context, profile string, onConnect func(), onEvent EventHandler) {
const subscribeEvents = "SOFIA::gateway_register SOFIA::gateway_unregister SOFIA::gateway_expire"
go func() {
for {
select {
case <-ctx.Done():
c.Disconnect()
return
default:
}
// Пытаемся подключиться
log.Printf("[esl] попытка подключения к %s:%d", c.host, c.port)
if err := c.Connect(); err != nil {
log.Printf("[esl] ошибка подключения: %v, повтор через %v", err, c.backoff)
time.Sleep(c.backoff)
c.backoff *= 2
if c.backoff > 60*time.Second {
c.backoff = 60 * time.Second
}
continue
}
// Успешное подключение
c.backoff = 1 * time.Second
// Подписываемся на события
if err := c.Subscribe(subscribeEvents); err != nil {
log.Printf("[esl] ошибка подписки: %v", err)
c.Disconnect()
continue
}
// Вызываем callback (GatewaySyncAll)
if onConnect != nil {
onConnect()
}
// Читаем события пока соединение живо
readEvents := make(chan struct{})
go func() {
defer close(readEvents)
c.readEventsLoop()
// Не ждём реконнект внутри readEventsLoop
}()
// Основной цикл обработки событий
for {
select {
case <-ctx.Done():
c.Disconnect()
return
case <-readEvents:
// Соединение разорвалось — идём на реконнект
log.Printf("[esl] чтение событий остановлено")
goto reconnect
case evt, ok := <-c.Events():
if !ok {
goto reconnect
}
if onEvent != nil {
onEvent(evt.Name, evt.Headers, evt.Body)
}
}
}
reconnect:
c.Disconnect()
time.Sleep(c.backoff)
}
}()
}
+107
View File
@@ -0,0 +1,107 @@
package esl
import (
"fmt"
"log"
"strings"
)
// GatewayAdd создаёт gateway на FreeSWITCH через Sofia профиль.
// Возвращает nil если gateway создан или уже существует (идемпотентность).
func (c *Client) GatewayAdd(profile, name, proxy string) error {
cmd := fmt.Sprintf("api sofia profile %s gwadd %s %s", profile, name, proxy)
headers, body, err := c.Send(cmd)
c.IncGatewayOps()
if err != nil {
log.Printf("[esl] GatewayAdd %s: ошибка команды: %v", name, err)
return err
}
reply := headers["Reply-Text"]
if strings.HasPrefix(reply, "+OK") {
log.Printf("[esl] GatewayAdd %s: создан (%s → %s)", name, profile, proxy)
return nil
}
// Gateway уже существует — не ошибка
if strings.Contains(body, "GATEWAY ALREADY") || strings.Contains(reply, "GATEWAY ALREADY") {
log.Printf("[esl] GatewayAdd %s: уже существует, ок", name)
return nil
}
log.Printf("[esl] GatewayAdd %s: ошибка FS — Reply: %s, Body: %s", name, reply, body)
return fmt.Errorf("GatewayAdd %s: %s (%s)", name, reply, body)
}
// GatewayDelete удаляет gateway на FreeSWITCH.
// Возвращает nil если gateway удалён или не найден (идемпотентность).
func (c *Client) GatewayDelete(profile, name string) error {
cmd := fmt.Sprintf("api sofia profile %s gwdel %s", profile, name)
headers, body, err := c.Send(cmd)
c.IncGatewayOps()
if err != nil {
log.Printf("[esl] GatewayDelete %s: ошибка команды: %v", name, err)
return err
}
reply := headers["Reply-Text"]
if strings.HasPrefix(reply, "+OK") {
log.Printf("[esl] GatewayDelete %s: удалён", name)
return nil
}
// Gateway не найден — не ошибка
if strings.Contains(body, "GATEWAY NOT FOUND") || strings.Contains(reply, "GATEWAY NOT FOUND") {
log.Printf("[esl] GatewayDelete %s: не найден, ок", name)
return nil
}
log.Printf("[esl] GatewayDelete %s: ошибка FS — Reply: %s, Body: %s", name, reply, body)
return fmt.Errorf("GatewayDelete %s: %s (%s)", name, reply, body)
}
// GatewayList возвращает список имён gateway в заданном профиле.
func (c *Client) GatewayList(profile string) ([]string, error) {
cmd := fmt.Sprintf("api sofia profile %s gwlist", profile)
_, body, err := c.Send(cmd)
if err != nil {
return nil, fmt.Errorf("GatewayList: %w", err)
}
// Парсим вывод gwlist — имена gateway на отдельных строках
lines := strings.Split(body, "\n")
var gateways []string
for _, line := range lines {
line = strings.TrimSpace(line)
if line == "" || strings.HasPrefix(line, "Name:") || strings.HasPrefix(line, "====") {
continue
}
// Ищем строки вида "pulse-ingress-xxxxx sip:..."
if strings.Contains(line, "pulse-") {
parts := strings.Fields(line)
if len(parts) > 0 {
gateways = append(gateways, parts[0])
}
}
}
return gateways, nil
}
// GatewaySyncAll синхронизирует все ingress-транки как gateway на FS.
// Вызывается после реконнекта.
func (c *Client) GatewaySyncAll(profile string, gateways []GatewaySpec) {
for _, g := range gateways {
if err := c.GatewayAdd(profile, g.Name, g.Proxy); err != nil {
log.Printf("[esl] GatewaySyncAll: ошибка %s: %v", g.Name, err)
}
}
}
// GatewaySpec описывает gateway для синхронизации.
type GatewaySpec struct {
Name string
Proxy string
}
+93
View File
@@ -0,0 +1,93 @@
// Package log реализует ASCII-логгер метрик с ротацией по дням.
// Формат: "2026-06-25T10:00:00Z pbx-01 calls=42 load=0.85 idle_cpu=35 fail=0.5 score=73"
package log
import (
"fmt"
"os"
"path/filepath"
"strings"
"sync"
"time"
"github.com/pulse-lets-go/internal/models"
)
// Logger — потокобезопасный логгер метрик в ASCII-файл с дневной ротацией.
type Logger struct {
mu sync.Mutex
dir string
file *os.File
today string // YYYY-MM-DD последнего открытого файла
}
// NewLogger создаёт логгер, записывающий в dataDir/.
// Файлы именуются как metrics.YYYY-MM-DD.log.
func NewLogger(dataDir string) (*Logger, error) {
if err := os.MkdirAll(dataDir, 0750); err != nil {
return nil, fmt.Errorf("создание data/: %w", err)
}
l := &Logger{dir: dataDir}
if err := l.ensureFile(); err != nil {
return nil, err
}
return l, nil
}
// Log записывает снапшот метрик ноды в лог-файл.
func (l *Logger) Log(ns *models.NodeState) {
l.mu.Lock()
defer l.mu.Unlock()
if err := l.ensureFile(); err != nil {
return // тихая ошибка — не блокируем метрики из-за логгера
}
line := l.formatLine(ns)
l.file.WriteString(line + "\n")
}
// formatLine форматирует строку лога.
func (l *Logger) formatLine(ns *models.NodeState) string {
t := time.Unix(ns.TS, 0).UTC().Format(time.RFC3339)
return fmt.Sprintf("%s %s calls=%d load=%.2f idle_cpu=%.1f fail=%.1f score=%.1f disabled=%t",
t, ns.NodeID, ns.ActiveCalls, ns.LoadAvg, ns.IdleCPU, ns.CallFailureRate, ns.Score, ns.Disabled)
}
// Close закрывает файл лога.
func (l *Logger) Close() error {
l.mu.Lock()
defer l.mu.Unlock()
if l.file != nil {
return l.file.Close()
}
return nil
}
// ensureFile открывает файл на текущий день, создавая новый, если день сменился.
// Вызывающий держит mu.
func (l *Logger) ensureFile() error {
today := time.Now().UTC().Format("2006-01-02")
if l.file != nil && l.today == today {
return nil
}
if l.file != nil {
l.file.Close()
}
path := filepath.Join(l.dir, "metrics."+today+".log")
f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0640)
if err != nil {
return fmt.Errorf("открытие лог-файла %s: %w", path, err)
}
l.file = f
l.today = today
return nil
}
// TrimPrefix обрезает "metrics." и ".log" из имени файла (для API истории).
func TrimPrefix(filename string) string {
return strings.TrimPrefix(strings.TrimSuffix(filename, ".log"), "metrics.")
}
+217
View File
@@ -0,0 +1,217 @@
// Package models содержит общие типы данных pulse-lets-go.
package models
import "time"
// --- NATS: метрика от нижестоящей АТС ---
// NodeMetric — сообщение, публикуемое нодой в NATS каждые 5 сек.
type NodeMetric struct {
NodeID string `json:"node_id"`
TS int64 `json:"ts"` // unix timestamp
Status string `json:"status"` // "ok", "degraded", "down"
ActiveCalls int `json:"active_calls"`
MaxCalls int `json:"max_calls"`
IdleCPU float64 `json:"idle_cpu"` // 0..100 %
LoadAvg float64 `json:"load_avg"` // system load average (1m)
CallFailureRate float64 `json:"call_failure_rate"` // 0.0..100.0 %
}
// --- In-memory состояние ноды в engine ---
// NodeState — полное состояние ноды в памяти балансировщика.
// Содержит данные последней метрики + поля ручного управления.
type NodeState struct {
NodeMetric
Disabled bool `json:"disabled"`
DisabledReason string `json:"disabled_reason"`
Score float64 `json:"score"`
LethalReason string `json:"lethal_reason,omitempty"` // причина, если score = -100
}
// --- Trunk ---
// TrunkType — тип транка.
type TrunkType string
const (
TrunkIngress TrunkType = "ingress"
TrunkBalance TrunkType = "balance"
TrunkFallback TrunkType = "fallback"
)
// Trunk — модель транка (из trunks.json).
type Trunk struct {
ID string `json:"id"`
Name string `json:"name"`
Type TrunkType `json:"type"`
NodeID string `json:"node_id,omitempty"` // только для balance
Gateway string `json:"gateway"`
Codecs []string `json:"codecs,omitempty"`
Context string `json:"context,omitempty"`
Enabled bool `json:"enabled"`
Description string `json:"description,omitempty"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
// --- User ---
// UserRole — роль пользователя.
type UserRole string
const (
RoleAdmin UserRole = "admin"
RoleViewer UserRole = "viewer"
)
// User — модель пользователя (из users.json).
type User struct {
ID string `json:"id"`
Username string `json:"username"`
PasswordHash string `json:"password_hash"`
Role UserRole `json:"role"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
// --- Auth ---
// LoginRequest — тело запроса на /api/auth/login.
type LoginRequest struct {
Username string `json:"username"`
Password string `json:"password"`
}
// LoginResponse — тело ответа /api/auth/login.
type LoginResponse struct {
AccessToken string `json:"access_token"`
RefreshToken string `json:"refresh_token"`
ExpiresIn int64 `json:"expires_in"`
}
// --- /api/route ---
// RouteResponse — ответ на запрос маршрутизации.
type RouteResponse struct {
// Успех
NodeID string `json:"node_id,omitempty"`
Score float64 `json:"score,omitempty"`
SIPGateway string `json:"sip_gateway,omitempty"`
// Fallback
Fallback bool `json:"fallback,omitempty"`
Reason string `json:"reason,omitempty"`
// Общая информация о нодах
Nodes []RouteNodeInfo `json:"nodes,omitempty"`
// Ошибка
Error string `json:"error,omitempty"`
}
// RouteNodeInfo — информация о ноде в ответе маршрутизации.
type RouteNodeInfo struct {
ID string `json:"id"`
Score float64 `json:"score"`
Reason string `json:"reason,omitempty"`
Status string `json:"status"`
}
// --- API: Nodes ---
// ToggleNodeRequest — запрос на включение/выключение ноды.
type ToggleNodeRequest struct {
Disabled bool `json:"disabled"`
Reason string `json:"reason,omitempty"`
}
// NodeInfo — информация о ноде для фронта.
type NodeInfo struct {
NodeID string `json:"node_id"`
TS int64 `json:"ts"`
Status string `json:"status"`
ActiveCalls int `json:"active_calls"`
MaxCalls int `json:"max_calls"`
IdleCPU float64 `json:"idle_cpu"`
LoadAvg float64 `json:"load_avg"`
CallFailureRate float64 `json:"call_failure_rate"`
Disabled bool `json:"disabled"`
DisabledReason string `json:"disabled_reason,omitempty"`
Score float64 `json:"score"`
LethalReason string `json:"lethal_reason,omitempty"`
IsStale bool `json:"is_stale"`
SecondsAgo float64 `json:"seconds_ago"`
}
// --- API: Users ---
// CreateUserRequest — запрос на создание пользователя.
type CreateUserRequest struct {
Username string `json:"username"`
Password string `json:"password"`
Role UserRole `json:"role"`
}
// UpdateUserRequest — запрос на обновление пользователя.
type UpdateUserRequest struct {
Username *string `json:"username,omitempty"`
Password *string `json:"password,omitempty"`
Role *UserRole `json:"role,omitempty"`
}
// --- API: Trunks ---
// CreateTrunkRequest — запрос на создание транка.
type CreateTrunkRequest struct {
Name string `json:"name"`
Type TrunkType `json:"type"`
NodeID string `json:"node_id,omitempty"`
Gateway string `json:"gateway"`
Codecs []string `json:"codecs,omitempty"`
Context string `json:"context,omitempty"`
Enabled bool `json:"enabled"`
Description string `json:"description,omitempty"`
}
// UpdateTrunkRequest — запрос на обновление транка.
type UpdateTrunkRequest struct {
Name *string `json:"name,omitempty"`
Type *TrunkType `json:"type,omitempty"`
NodeID *string `json:"node_id,omitempty"`
Gateway *string `json:"gateway,omitempty"`
Codecs []string `json:"codecs,omitempty"`
Context *string `json:"context,omitempty"`
Enabled *bool `json:"enabled,omitempty"`
Description *string `json:"description,omitempty"`
}
// --- Monitoring ---
// ZabbixNode — элемент LLD (Low-Level Discovery) для Zabbix.
type ZabbixNode struct {
NodeID string `json:"{#NODE_ID}"`
Status string `json:"{#STATUS}"`
ActiveCalls int `json:"{#ACTIVE_CALLS}"`
MaxCalls int `json:"{#MAX_CALLS}"`
IdleCPU float64 `json:"{#IDLE_CPU}"`
LoadAvg float64 `json:"{#LOAD_AVG}"`
CallFailureRate float64 `json:"{#CALL_FAILURE_RATE}"`
Score float64 `json:"{#SCORE}"`
Disabled bool `json:"{#DISABLED}"`
IsStale bool `json:"{#IS_STALE}"`
SecondsAgo float64 `json:"{#SECONDS_AGO}"`
}
// ZabbixResponse — ответ на запрос Zabbix LLD.
type ZabbixResponse struct {
Data []ZabbixNode `json:"data"`
}
// --- WebSocket ---
// WsMetricsMessage — сообщение, отправляемое по WebSocket.
type WsMetricsMessage struct {
Type string `json:"type"` // "nodes_update", "node_toggle", etc
Payload interface{} `json:"payload"`
}
+98
View File
@@ -0,0 +1,98 @@
// Package nats содержит подписчика на NATS-метрики от нижестоящих АТС.
package nats
import (
"encoding/json"
"fmt"
"log"
"sync"
natsgo "github.com/nats-io/nats.go"
"github.com/pulse-lets-go/internal/models"
"github.com/pulse-lets-go/internal/engine"
filelog "github.com/pulse-lets-go/internal/log"
)
const subject = "pulse.metrics.>"
// Subscriber подписывается на NATS и обновляет engine + пишет лог.
type Subscriber struct {
nc *natsgo.Conn
sub *natsgo.Subscription
engine *engine.Engine
logger *filelog.Logger
wg sync.WaitGroup
}
// NewSubscriber подключается к NATS и запускает подписку.
func NewSubscriber(natsURL, user, password string, eng *engine.Engine, l *filelog.Logger) (*Subscriber, error) {
opts := []natsgo.Option{
natsgo.ReconnectWait(natsgo.DefaultReconnectWait),
natsgo.MaxReconnects(-1),
}
if user != "" && password != "" {
opts = append(opts, natsgo.UserInfo(user, password))
}
nc, err := natsgo.Connect(natsURL, opts...)
if err != nil {
return nil, fmt.Errorf("nats connect: %w", err)
}
s := &Subscriber{
nc: nc,
engine: eng,
logger: l,
}
sub, err := nc.Subscribe(subject, s.handleMetric)
if err != nil {
nc.Close()
return nil, fmt.Errorf("nats subscribe: %w", err)
}
s.sub = sub
log.Printf("[nats] подписан на %s, подключён к %s", subject, natsURL)
return s, nil
}
// handleMetric — обработчик входящих сообщений NATS.
func (s *Subscriber) handleMetric(msg *natsgo.Msg) {
var metric models.NodeMetric
if err := json.Unmarshal(msg.Data, &metric); err != nil {
log.Printf("[nats] ошибка парсинга метрики: %v", err)
return
}
// Валидация минимальных полей
if metric.NodeID == "" {
log.Printf("[nats] метрика без node_id, игнорируется")
return
}
if metric.MaxCalls <= 0 {
metric.MaxCalls = 250 // разумное значение по умолчанию
}
ns := s.engine.UpdateMetric(&metric)
s.logger.Log(ns)
}
// IsConnected возвращает true, если NATS-соединение активно.
func (s *Subscriber) IsConnected() bool {
if s.nc == nil {
return false
}
return s.nc.IsConnected()
}
// Close останавливает подписку и закрывает соединение.
func (s *Subscriber) Close() {
if s.sub != nil {
s.sub.Unsubscribe()
}
if s.nc != nil {
s.nc.Close()
}
s.wg.Wait()
}