From cc9da3ad7d36a618cdce4d192bb6717525b89cd1 Mon Sep 17 00:00:00 2001 From: Maksim Totmin Date: Thu, 25 Jun 2026 19:30:36 +0700 Subject: [PATCH] refactor: common util package, ESL/AMI/security fixes, Prometheus metrics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Backend stability and security improvements: * internal/util/ — common RandomHex helper, removed 3 duplicates * ESL: deduplicated readMessage (locked/unlocked), net.JoinHostPort for IPv6 * AMI: synchronous reconnect() in readEventsLoop, net.JoinHostPort for IPv6 * Auth: /api/auth/refresh accepts Authorization header only (no ?token=) * decodeJSON: http.MaxBytesReader(1<<20) body limit * Trunks: gatewayParams() uses configured ESL.GatewayPrefix * Config: jwt_secret_env env-var fallback * FSCollector: time.After → time.NewTimer with defer Stop * Monitoring: Prometheus counters (route_requests, nodes_total/healthy, uptime) * go fmt pass across all internal/ packages --- cmd/emulator/main.go | 4 +- .../agent/collector_freeswitch.go | 7 +- cmd/pulse-lets-go-agent/agent/config.go | 24 ++--- cmd/pulse-lets-go-agent/agent/publisher.go | 4 +- cmd/pulse-lets-go-agent/agent/system.go | 6 +- cmd/siptest/main.go | 32 +++--- internal/ami/client.go | 36 +++++-- internal/api/auth.go | 12 +-- internal/api/health.go | 10 +- internal/api/monitoring.go | 20 ++++ internal/api/ratelimit.go | 18 ++-- internal/api/route.go | 2 +- internal/api/router.go | 10 +- internal/api/trunks.go | 5 +- internal/api/users.go | 15 +-- internal/config/manager.go | 61 +++++++----- internal/engine/engine.go | 20 ++-- internal/engine/engine_test.go | 2 +- internal/esl/client.go | 75 ++++---------- internal/esl/events.go | 1 + internal/esl/gateway.go | 8 +- internal/log/logger.go | 8 +- internal/models/types.go | 98 +++++++++---------- internal/nats/subscriber.go | 2 +- internal/util/util.go | 18 ++++ 25 files changed, 261 insertions(+), 237 deletions(-) create mode 100644 internal/util/util.go diff --git a/cmd/emulator/main.go b/cmd/emulator/main.go index cabf50c..4402809 100644 --- a/cmd/emulator/main.go +++ b/cmd/emulator/main.go @@ -49,7 +49,7 @@ type valueSpec struct { isRand bool } -func fixed(v float64) valueSpec { return valueSpec{fixed: v} } +func fixed(v float64) valueSpec { return valueSpec{fixed: v} } func rrange(min, max float64) valueSpec { return valueSpec{min: min, max: max, isRand: true} } @@ -71,7 +71,7 @@ type nodeSpec struct { idleCPU valueSpec loadAvg valueSpec failRate valueSpec - sipGateway string // SIP-адрес для авто-создания транка + sipGateway string // SIP-адрес для авто-создания транка } func (ns *nodeSpec) generate() nodeMetric { diff --git a/cmd/pulse-lets-go-agent/agent/collector_freeswitch.go b/cmd/pulse-lets-go-agent/agent/collector_freeswitch.go index 8971812..1f0a0fa 100644 --- a/cmd/pulse-lets-go-agent/agent/collector_freeswitch.go +++ b/cmd/pulse-lets-go-agent/agent/collector_freeswitch.go @@ -32,12 +32,15 @@ func (fc *FSCollector) Connect() error { fc.client.ConnectWithRetry() // Ждём подключения с таймаутом + timer := time.NewTimer(100 * time.Millisecond) + defer timer.Stop() + for !fc.client.IsConnected() { + timer.Reset(100 * time.Millisecond) select { case <-ctx.Done(): return fmt.Errorf("esl connect timeout") - case <-time.After(100 * time.Millisecond): - // Выходим из цикла после ConnectWithRetry + case <-timer.C: return nil } } diff --git a/cmd/pulse-lets-go-agent/agent/config.go b/cmd/pulse-lets-go-agent/agent/config.go index 8693ac1..681dc60 100644 --- a/cmd/pulse-lets-go-agent/agent/config.go +++ b/cmd/pulse-lets-go-agent/agent/config.go @@ -27,18 +27,18 @@ import ( // Config — конфигурация агента (agent.json). type Config struct { - NodeID string `json:"node_id"` // уникальный ID ноды (uc06, ses-sip) - Type string `json:"type"` // "freeswitch" или "asterisk" - NatsURL string `json:"nats_url"` // NATS URL - NatsUser string `json:"nats_user,omitempty"` - NatsPassword string `json:"nats_password,omitempty"` - IntervalSec int `json:"interval_sec"` // интервал сбора (default: 5) - MaxCalls int `json:"max_calls"` // ёмкость ноды (default: 250) - SIPGateway string `json:"sip_gateway"` // SIP-адрес для auto-транка - SIPGatewayAuto bool `json:"sip_gateway_auto"` // авто-определить из PBX - FailureWindow int `json:"failure_window"` // окно для call_failure_rate (default: 1000) - ESL ESLCfg `json:"esl,omitempty"` - AMI AMICfg `json:"ami,omitempty"` + NodeID string `json:"node_id"` // уникальный ID ноды (uc06, ses-sip) + Type string `json:"type"` // "freeswitch" или "asterisk" + NatsURL string `json:"nats_url"` // NATS URL + NatsUser string `json:"nats_user,omitempty"` + NatsPassword string `json:"nats_password,omitempty"` + IntervalSec int `json:"interval_sec"` // интервал сбора (default: 5) + MaxCalls int `json:"max_calls"` // ёмкость ноды (default: 250) + SIPGateway string `json:"sip_gateway"` // SIP-адрес для auto-транка + SIPGatewayAuto bool `json:"sip_gateway_auto"` // авто-определить из PBX + FailureWindow int `json:"failure_window"` // окно для call_failure_rate (default: 1000) + ESL ESLCfg `json:"esl,omitempty"` + AMI AMICfg `json:"ami,omitempty"` } // ESLCfg — настройки подключения к FreeSWITCH ESL. diff --git a/cmd/pulse-lets-go-agent/agent/publisher.go b/cmd/pulse-lets-go-agent/agent/publisher.go index c6cf44b..142423d 100644 --- a/cmd/pulse-lets-go-agent/agent/publisher.go +++ b/cmd/pulse-lets-go-agent/agent/publisher.go @@ -12,8 +12,8 @@ import ( // Publisher публикует метрики в NATS. type Publisher struct { - nc *natsgo.Conn - nodeID string + nc *natsgo.Conn + nodeID string } // NewPublisher создаёт NATS publisher. diff --git a/cmd/pulse-lets-go-agent/agent/system.go b/cmd/pulse-lets-go-agent/agent/system.go index 3a615e4..dc7d603 100644 --- a/cmd/pulse-lets-go-agent/agent/system.go +++ b/cmd/pulse-lets-go-agent/agent/system.go @@ -15,9 +15,9 @@ func nowTS() int64 { // SystemStats содержит системные метрики (CPU, load). type SystemStats struct { - Load1 float64 // load average 1m - Load5 float64 // load average 5m - Load15 float64 // load average 15m + Load1 float64 // load average 1m + Load5 float64 // load average 5m + Load15 float64 // load average 15m IdleCPU float64 // процент простоя CPU } diff --git a/cmd/siptest/main.go b/cmd/siptest/main.go index 72c8135..8b1acf9 100644 --- a/cmd/siptest/main.go +++ b/cmd/siptest/main.go @@ -1,16 +1,16 @@ // SIP тестер для e2e проверки pulse-lets-go + FreeSWITCH. // Два режима: -// uas — отвечает 200 OK на SIP INVITE, симулирует PBX-ноду -// uac — шлёт SIP INVITE в FreeSWITCH, симулирует оператора +// +// uas — отвечает 200 OK на SIP INVITE, симулирует PBX-ноду +// uac — шлёт SIP INVITE в FreeSWITCH, симулирует оператора // // Примеры: -// ./siptest -mode uas (PBX-нода на порту 5090) -// ./siptest -mode uac -r 10 -l 100 (оператор, 10 CPS, до 100 одновременных) +// +// ./siptest -mode uas (PBX-нода на порту 5090) +// ./siptest -mode uac -r 10 -l 100 (оператор, 10 CPS, до 100 одновременных) package main import ( - "crypto/rand" - "encoding/hex" "flag" "fmt" "log" @@ -19,6 +19,8 @@ import ( "sync" "sync/atomic" "time" + + "github.com/pulse-lets-go/internal/util" ) const ( @@ -29,9 +31,9 @@ const ( // Статистика var ( - callsSent atomic.Int64 - callsOK atomic.Int64 - callsFailed atomic.Int64 + callsSent atomic.Int64 + callsOK atomic.Int64 + callsFailed atomic.Int64 ) func main() { @@ -103,7 +105,7 @@ func buildSIPResponse(invite, localAddr string) string { // Добавляем tag к To если его нет to = strings.TrimSpace(to) if !strings.Contains(to, ";tag=") { - to += ";tag=uas-" + randomHex(4) + to += ";tag=uas-" + util.RandomHex(4) } return fmt.Sprintf("SIP/2.0 200 OK\r\n"+ @@ -213,8 +215,8 @@ func runUAC(rate, limit, maxCalls int, dest, caller, host string) { } func buildSIPInvite(dest, caller string, callNum int, srcAddr, fsAddr string) string { - callID := fmt.Sprintf("call-%d-%s@%s", callNum, randomHex(4), "127.0.0.1") - branch := "z9hG4bK-" + randomHex(8) + callID := fmt.Sprintf("call-%d-%s@%s", callNum, util.RandomHex(4), "127.0.0.1") + branch := "z9hG4bK-" + util.RandomHex(8) return fmt.Sprintf("INVITE sip:%s@%s SIP/2.0\r\n"+ "Via: SIP/2.0/UDP %s;branch=%s\r\n"+ @@ -227,9 +229,3 @@ func buildSIPInvite(dest, caller string, callNum int, srcAddr, fsAddr string) st "\r\n", dest, fsAddr, srcAddr, branch, caller, callNum, dest, callID, caller, srcAddr) } - -func randomHex(n int) string { - b := make([]byte, n) - rand.Read(b) - return hex.EncodeToString(b)[:n] -} diff --git a/internal/ami/client.go b/internal/ami/client.go index 77feb58..84ae656 100644 --- a/internal/ami/client.go +++ b/internal/ami/client.go @@ -63,7 +63,7 @@ func NewClient(host string, port int, username, password string) *Client { func (c *Client) Connect() error { c.shouldReconn.Store(true) - addr := fmt.Sprintf("%s:%d", c.host, c.port) + addr := net.JoinHostPort(c.host, fmt.Sprintf("%d", c.port)) dialer := net.Dialer{Timeout: 5 * time.Second} conn, err := dialer.Dial("tcp", addr) if err != nil { @@ -268,11 +268,9 @@ func (c *Client) readEventsLoop() { if err != nil { if err == io.EOF || strings.Contains(err.Error(), "use of closed network connection") { c.connected.Store(false) - log.Printf("[ami] соединение разорвано") - if c.shouldReconn.Load() { - go c.ConnectWithRetry() - } - return + log.Printf("[ami] соединение разорвано, реконнект...") + c.reconnect() + continue } time.Sleep(100 * time.Millisecond) continue @@ -291,3 +289,29 @@ func (c *Client) readEventsLoop() { } } } + +// reconnect выполняет реконнект с backoff (без горутин — синхронный вызов). +func (c *Client) reconnect() { + if !c.shouldReconn.Load() { + return + } + log.Printf("[ami] реконнект через %v...", c.backoff) + time.Sleep(c.backoff) + + if c.conn != nil { + c.conn.Close() + c.conn = nil + } + + if err := c.Connect(); err != nil { + log.Printf("[ami] реконнект не удался: %v", err) + c.backoff *= 2 + if c.backoff > 60*time.Second { + c.backoff = 60 * time.Second + } + return + } + + c.backoff = 1 * time.Second + log.Printf("[ami] реконнект успешен") +} diff --git a/internal/api/auth.go b/internal/api/auth.go index 754ceab..30681d4 100644 --- a/internal/api/auth.go +++ b/internal/api/auth.go @@ -27,7 +27,7 @@ type claims struct { type contextKey string const ( - userContextKey contextKey = "user" + userContextKey contextKey = "user" refreshTokenKey contextKey = "refresh_token" ) @@ -39,8 +39,8 @@ type UserInfo struct { } const ( - jwtExpiry = 15 * time.Minute - refreshExpiry = 24 * time.Hour + jwtExpiry = 15 * time.Minute + refreshExpiry = 24 * time.Hour ) // generateJWT создаёт access+refresh токены. @@ -139,7 +139,7 @@ func (a *API) apiKeyMiddleware(next http.Handler) http.Handler { // 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 { + if err := decodeJSON(r, &req); err != nil { writeError(w, http.StatusBadRequest, "некорректный JSON") return } @@ -199,10 +199,6 @@ func (a *API) handleLogin(w http.ResponseWriter, r *http.Request) { // 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 diff --git a/internal/api/health.go b/internal/api/health.go index 295f898..4dc3cb2 100644 --- a/internal/api/health.go +++ b/internal/api/health.go @@ -17,12 +17,12 @@ func (a *API) handleHealth(w http.ResponseWriter, r *http.Request) { if a.eslClient != nil { eslStats := a.eslClient.GetStats() eslInfo = map[string]interface{}{ - "status": eslStats.Status, - "host": eslStats.Host, + "status": eslStats.Status, + "host": eslStats.Host, "uptime_seconds": eslStats.UptimeSec, - "reconnects": eslStats.Reconnects, - "gateway_ops": eslStats.GatewayOps, - "events_recv": eslStats.EventsRecv, + "reconnects": eslStats.Reconnects, + "gateway_ops": eslStats.GatewayOps, + "events_recv": eslStats.EventsRecv, } } diff --git a/internal/api/monitoring.go b/internal/api/monitoring.go index 74e6850..b2aded8 100644 --- a/internal/api/monitoring.go +++ b/internal/api/monitoring.go @@ -38,6 +38,26 @@ func (a *API) handlePrometheus(w http.ResponseWriter, r *http.Request) { var sb strings.Builder + // Общие метрики (counter gauge) + stats := a.engine.GetHealthStats() + fmt.Fprintf(&sb, "# HELP pulse_route_requests_total Total number of route requests.\n") + fmt.Fprintf(&sb, "# TYPE pulse_route_requests_total counter\n") + fmt.Fprintf(&sb, "pulse_route_requests_total %d\n", stats.RouteRequests) + + fmt.Fprintf(&sb, "# HELP pulse_nodes_total Total nodes registered.\n") + fmt.Fprintf(&sb, "# TYPE pulse_nodes_total gauge\n") + fmt.Fprintf(&sb, "pulse_nodes_total %d\n", stats.TotalNodes) + + fmt.Fprintf(&sb, "# HELP pulse_nodes_healthy Healthy nodes count.\n") + fmt.Fprintf(&sb, "# TYPE pulse_nodes_healthy gauge\n") + fmt.Fprintf(&sb, "pulse_nodes_healthy %d\n", stats.HealthyNodes) + + fmt.Fprintf(&sb, "# HELP pulse_uptime_seconds Uptime in seconds.\n") + fmt.Fprintf(&sb, "# TYPE pulse_uptime_seconds gauge\n") + fmt.Fprintf(&sb, "pulse_uptime_seconds %d\n", stats.UptimeSeconds) + + fmt.Fprintf(&sb, "\n") + for _, n := range nodes { id := sanitizePromLabel(n.NodeID) diff --git a/internal/api/ratelimit.go b/internal/api/ratelimit.go index 5e3afca..fc84232 100644 --- a/internal/api/ratelimit.go +++ b/internal/api/ratelimit.go @@ -9,12 +9,12 @@ import ( // rateLimiter — in-memory token bucket rate limiter. type rateLimiter struct { - mu sync.Mutex - buckets map[string]*tokenBucket - rate float64 // токенов в секунду - burst int // максимальный размер бакета + mu sync.Mutex + buckets map[string]*tokenBucket + rate float64 // токенов в секунду + burst int // максимальный размер бакета cleanupInterval time.Duration - lastCleanup time.Time + lastCleanup time.Time } type tokenBucket struct { @@ -28,11 +28,11 @@ func newRateLimiter(ratePerSec, burst int) *rateLimiter { burst = ratePerSec } return &rateLimiter{ - buckets: make(map[string]*tokenBucket), - rate: float64(ratePerSec), - burst: burst, + buckets: make(map[string]*tokenBucket), + rate: float64(ratePerSec), + burst: burst, cleanupInterval: 60 * time.Second, - lastCleanup: time.Now(), + lastCleanup: time.Now(), } } diff --git a/internal/api/route.go b/internal/api/route.go index aa68280..0972b41 100644 --- a/internal/api/route.go +++ b/internal/api/route.go @@ -27,7 +27,7 @@ func (a *API) handleRoute(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, models.RouteResponse{ Fallback: true, - SIPGateway: fallbackGW, + SIPGateway: fallbackGW, Reason: "all_nodes_unhealthy", Nodes: nodes, }) diff --git a/internal/api/router.go b/internal/api/router.go index e9382e3..98b329b 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -18,8 +18,8 @@ type API struct { jwtSecret string monitoringAPIKey string wsHub *wsHub - natsConnected func() bool // колбэк для проверки NATS-статуса (health) - eslClient *esl.Client // ESL-клиент (nil если не сконфигурирован) + natsConnected func() bool // колбэк для проверки NATS-статуса (health) + eslClient *esl.Client // ESL-клиент (nil если не сконфигурирован) logFormat string routeLimiter *rateLimiter apiLimiter *rateLimiter @@ -141,9 +141,13 @@ func (a *API) BroadcastGatewayEvent(gatewayName, trunkID, status string) { a.wsHub.broadcast(msg) } -// decodeJSON декодирует тело запроса в структуру v. +// maxBodySize — максимальный размер тела запроса в байтах (1 MiB). +const maxBodySize = 1 << 20 + +// decodeJSON декодирует тело запроса в структуру v с ограничением размера. func decodeJSON(r *http.Request, v interface{}) error { defer r.Body.Close() + r.Body = http.MaxBytesReader(nil, r.Body, maxBodySize) if err := json.NewDecoder(r.Body).Decode(v); err != nil { return fmt.Errorf("декодирование JSON: %w", err) } diff --git a/internal/api/trunks.go b/internal/api/trunks.go index b516ab7..cb6f113 100644 --- a/internal/api/trunks.go +++ b/internal/api/trunks.go @@ -268,8 +268,9 @@ func (a *API) eslPushGatewayDelete(trunk models.Trunk) { // 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) + cfg := a.readConfig() + profile = cfg.ESL.SofiaProfile + name = fmt.Sprintf("%s-ingress-%s", cfg.ESL.GatewayPrefix, trunk.ID) proxy = extractHost(trunk.Gateway) return } diff --git a/internal/api/users.go b/internal/api/users.go index 759d067..f5d06ca 100644 --- a/internal/api/users.go +++ b/internal/api/users.go @@ -1,13 +1,12 @@ package api import ( - "crypto/rand" - "encoding/hex" "fmt" "net/http" "time" "github.com/pulse-lets-go/internal/models" + "github.com/pulse-lets-go/internal/util" "golang.org/x/crypto/bcrypt" ) @@ -227,15 +226,5 @@ func findUserIndex(users []models.User, id string) int { // 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] + return fmt.Sprintf("%s-%s", prefix, util.RandomHex(6)) } diff --git a/internal/config/manager.go b/internal/config/manager.go index 083e5dc..a0964fe 100644 --- a/internal/config/manager.go +++ b/internal/config/manager.go @@ -27,9 +27,9 @@ type ScoringWeights struct { // ScoringConfig — пороги и веса для scoring engine. type ScoringConfig struct { - IdleCPUMin float64 `json:"idle_cpu_min"` - CallFailureRateLethal float64 `json:"call_failure_rate_lethal"` - Weights ScoringWeights `json:"weights"` + IdleCPUMin float64 `json:"idle_cpu_min"` + CallFailureRateLethal float64 `json:"call_failure_rate_lethal"` + Weights ScoringWeights `json:"weights"` } // TLSConfig — настройки HTTPS. @@ -54,11 +54,11 @@ type LogConfig struct { // ESLConfig — настройки подключения к FreeSWITCH ESL. type ESLConfig struct { - Host string `json:"host"` // если пустой — ESL не запускается - Port int `json:"port"` // порт ESL (по умолчанию 8021) + 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) + PasswordEnv string `json:"password_env"` // имя переменной окружения с паролем + SofiaProfile string `json:"sofia_profile"` // профиль Sofia (external/internal) GatewayPrefix string `json:"gateway_prefix"` // префикс gateway (pulse) } @@ -72,20 +72,31 @@ func (e *ESLConfig) GetPassword() string { return e.Password } +// GetJWTSecret возвращает JWT-секрет: из переменной окружения, если задан jwt_secret_env, иначе из поля jwt_secret. +func (c *Config) GetJWTSecret() string { + if c.JWTSecretEnv != "" { + if val := os.Getenv(c.JWTSecretEnv); val != "" { + return val + } + } + return c.JWTSecret +} + // 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"` + 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"` + JWTSecretEnv string `json:"jwt_secret_env"` // имя переменной окружения с JWT-секретом + 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 проверяет корректность всех полей конфига. @@ -168,8 +179,8 @@ func (c *Config) Validate() error { // Manager управляет чтением и атомарной записью JSON-конфигов. type Manager struct { - mu sync.RWMutex - dir string // путь к data/ + mu sync.RWMutex + dir string // путь к data/ } // NewManager создаёт новый менеджер конфигов из директории dataDir. @@ -302,9 +313,9 @@ func (m *Manager) saveUsersLocked(users []models.User) error { // --- Приватные методы --- -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") } +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. @@ -355,7 +366,7 @@ func (m *Manager) createDefaultConfig() error { MonitoringAPIKey: apiKey, StaleThresholdSec: 20, Scoring: ScoringConfig{ - IdleCPUMin: 5.0, + IdleCPUMin: 5.0, CallFailureRateLethal: 15.0, Weights: ScoringWeights{ CallScore: 0.40, diff --git a/internal/engine/engine.go b/internal/engine/engine.go index 2669c17..48ad32d 100644 --- a/internal/engine/engine.go +++ b/internal/engine/engine.go @@ -190,11 +190,11 @@ func (e *Engine) IncrementRouteRequests() { // 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"` + 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 (потокобезопасно). @@ -209,11 +209,11 @@ func (e *Engine) GetHealthStats() HealthStats { } } return HealthStats{ - TotalNodes: len(e.nodes), - HealthyNodes: healthy, - RouteRequests: e.routeRequests.Load(), - UptimeSeconds: int64(time.Since(e.startTime).Seconds()), - LastMetricTS: e.lastMetricTime.Load(), + TotalNodes: len(e.nodes), + HealthyNodes: healthy, + RouteRequests: e.routeRequests.Load(), + UptimeSeconds: int64(time.Since(e.startTime).Seconds()), + LastMetricTS: e.lastMetricTime.Load(), } } diff --git a/internal/engine/engine_test.go b/internal/engine/engine_test.go index d9bb31b..cf19683 100644 --- a/internal/engine/engine_test.go +++ b/internal/engine/engine_test.go @@ -12,7 +12,7 @@ func defaultCfg() *config.Config { return &config.Config{ StaleThresholdSec: 20, Scoring: config.ScoringConfig{ - IdleCPUMin: 5.0, + IdleCPUMin: 5.0, CallFailureRateLethal: 15.0, Weights: config.ScoringWeights{ CallScore: 0.40, diff --git a/internal/esl/client.go b/internal/esl/client.go index 48e743a..c0fd523 100644 --- a/internal/esl/client.go +++ b/internal/esl/client.go @@ -36,10 +36,10 @@ type Client struct { port int password string - conn net.Conn - reader *bufio.Reader - mu sync.Mutex // защищает отправку синхронных команд (auth, subscribe) - sendMu sync.Mutex // защищает запись в сокет (async gateway commands) + conn net.Conn + reader *bufio.Reader + mu sync.Mutex // защищает отправку синхронных команд (auth, subscribe) + sendMu sync.Mutex // защищает запись в сокет (async gateway commands) connected atomic.Bool eventsCh chan *Event @@ -109,7 +109,7 @@ func (c *Client) tryConnect() error { c.closeCh = make(chan struct{}) } - addr := fmt.Sprintf("%s:%d", c.host, c.port) + addr := net.JoinHostPort(c.host, fmt.Sprintf("%d", c.port)) dialer := net.Dialer{Timeout: 5 * time.Second} conn, err := dialer.Dial("tcp", addr) if err != nil { @@ -119,7 +119,7 @@ func (c *Client) tryConnect() error { c.reader = bufio.NewReader(conn) // 1. Читаем auth/request - headers, _, err := c.readMessageLocked() + headers, _, err := c.readMessage() if err != nil { conn.Close() return fmt.Errorf("чтение auth/request: %w", err) @@ -137,7 +137,7 @@ func (c *Client) tryConnect() error { } // 3. Читаем auth response - headers, _, err = c.readMessageLocked() + headers, _, err = c.readMessage() if err != nil { conn.Close() return fmt.Errorf("чтение auth response: %w", err) @@ -199,7 +199,7 @@ func (c *Client) Send(command string) (map[string]string, string, error) { } // Читаем ответ - return c.readMessageLocked() + return c.readMessage() } // Subscribe подписывается на события ESL. @@ -263,7 +263,7 @@ func (c *Client) readEventsLoop() { continue } - headers, body, err := c.readMessageUnlocked() + headers, body, err := c.readMessage() if err != nil { if err == io.EOF || strings.Contains(err.Error(), "use of closed network connection") { @@ -330,48 +330,9 @@ func (c *Client) reconnect() { // --- Приватные методы --- -// readMessageUnlocked читает одно ESL-сообщение: заголовки + тело. -// Должна вызываться только из readEventsLoop (одиночный читатель). -func (c *Client) readMessageUnlocked() (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 -} - -// readMessageLocked читает одно ESL-сообщение: заголовки + тело. -// Вызывающий держит c.mu (для синхронных команд Send). -func (c *Client) readMessageLocked() (map[string]string, string, error) { +// readMessage читает одно ESL-сообщение: заголовки + тело. +// Вызывающий должен гарантировать монопольный доступ к c.reader. +func (c *Client) readMessage() (map[string]string, string, error) { headers := make(map[string]string) for { @@ -412,12 +373,12 @@ func (c *Client) readMessageLocked() (map[string]string, string, error) { // 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"` + 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 возвращает текущую статистику. diff --git a/internal/esl/events.go b/internal/esl/events.go index 9bcbdb9..f8b4bc9 100644 --- a/internal/esl/events.go +++ b/internal/esl/events.go @@ -16,6 +16,7 @@ type EventHandler func(eventName string, headers map[string]string, body string) // - Вызывает 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" diff --git a/internal/esl/gateway.go b/internal/esl/gateway.go index f6a0791..3469b71 100644 --- a/internal/esl/gateway.go +++ b/internal/esl/gateway.go @@ -32,8 +32,8 @@ func (c *Client) GatewayDelete(profile, name string) error { return nil } -// GatewayList возвращает список имён gateway в заданном профиле. -func (c *Client) GatewayList(profile string) ([]string, error) { +// GatewayList возвращает список имён gateway в заданном профиле с указанным префиксом. +func (c *Client) GatewayList(profile, prefix string) ([]string, error) { cmd := fmt.Sprintf("api sofia profile %s gwlist", profile) _, body, err := c.Send(cmd) @@ -49,8 +49,8 @@ func (c *Client) GatewayList(profile string) ([]string, error) { if line == "" || strings.HasPrefix(line, "Name:") || strings.HasPrefix(line, "====") { continue } - // Ищем строки вида "pulse-ingress-xxxxx sip:..." - if strings.Contains(line, "pulse-") { + // Ищем строки вида "-ingress-xxxxx sip:..." + if strings.Contains(line, prefix+"-ingress-") { parts := strings.Fields(line) if len(parts) > 0 { gateways = append(gateways, parts[0]) diff --git a/internal/log/logger.go b/internal/log/logger.go index 4f4b441..9bdb292 100644 --- a/internal/log/logger.go +++ b/internal/log/logger.go @@ -15,10 +15,10 @@ import ( // Logger — потокобезопасный логгер метрик в ASCII-файл с дневной ротацией. type Logger struct { - mu sync.Mutex - dir string - file *os.File - today string // YYYY-MM-DD последнего открытого файла + mu sync.Mutex + dir string + file *os.File + today string // YYYY-MM-DD последнего открытого файла } // NewLogger создаёт логгер, записывающий в dataDir/. diff --git a/internal/models/types.go b/internal/models/types.go index 235cad1..5bf005f 100644 --- a/internal/models/types.go +++ b/internal/models/types.go @@ -7,15 +7,15 @@ import "time" // 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 % - SIPGateway string `json:"sip_gateway,omitempty"` // SIP-адрес ноды для авто-создания транка + 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 % + SIPGateway string `json:"sip_gateway,omitempty"` // SIP-адрес ноды для авто-создания транка } // --- In-memory состояние ноды в engine --- @@ -24,8 +24,8 @@ type NodeMetric struct { // Содержит данные последней метрики + поля ручного управления. type NodeState struct { NodeMetric - Disabled bool `json:"disabled"` - DisabledReason string `json:"disabled_reason"` + Disabled bool `json:"disabled"` + DisabledReason string `json:"disabled_reason"` Score float64 `json:"score"` LethalReason string `json:"lethal_reason,omitempty"` // причина, если score = -100 } @@ -46,7 +46,7 @@ type Trunk struct { ID string `json:"id"` Name string `json:"name"` Type TrunkType `json:"type"` - NodeID string `json:"node_id,omitempty"` // только для balance + NodeID string `json:"node_id,omitempty"` // только для balance Gateway string `json:"gateway"` Codecs []string `json:"codecs,omitempty"` Context string `json:"context,omitempty"` @@ -68,10 +68,10 @@ const ( // User — модель пользователя (из users.json). type User struct { - ID string `json:"id"` - Username string `json:"username"` - PasswordHash string `json:"password_hash"` - Role UserRole `json:"role"` + 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"` } @@ -96,27 +96,27 @@ type LoginResponse struct { // RouteResponse — ответ на запрос маршрутизации. type RouteResponse struct { // Успех - NodeID string `json:"node_id,omitempty"` - Score float64 `json:"score,omitempty"` - SIPGateway string `json:"sip_gateway,omitempty"` + 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"` + Fallback bool `json:"fallback,omitempty"` + Reason string `json:"reason,omitempty"` // Общая информация о нодах - Nodes []RouteNodeInfo `json:"nodes,omitempty"` + Nodes []RouteNodeInfo `json:"nodes,omitempty"` // Ошибка - Error string `json:"error,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"` + Reason string `json:"reason,omitempty"` + Status string `json:"status"` } // --- API: Nodes --- @@ -129,20 +129,20 @@ type ToggleNodeRequest struct { // 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"` + 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"` + 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 --- @@ -191,17 +191,17 @@ type UpdateTrunkRequest struct { // 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}"` + 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}"` + Score float64 `json:"{#SCORE}"` + Disabled bool `json:"{#DISABLED}"` + IsStale bool `json:"{#IS_STALE}"` + SecondsAgo float64 `json:"{#SECONDS_AGO}"` } // ZabbixResponse — ответ на запрос Zabbix LLD. @@ -213,6 +213,6 @@ type ZabbixResponse struct { // WsMetricsMessage — сообщение, отправляемое по WebSocket. type WsMetricsMessage struct { - Type string `json:"type"` // "nodes_update", "node_toggle", etc + Type string `json:"type"` // "nodes_update", "node_toggle", etc Payload interface{} `json:"payload"` } diff --git a/internal/nats/subscriber.go b/internal/nats/subscriber.go index fdcbb93..4f8f52a 100644 --- a/internal/nats/subscriber.go +++ b/internal/nats/subscriber.go @@ -9,9 +9,9 @@ import ( 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" + "github.com/pulse-lets-go/internal/models" ) const subject = "pulse.metrics.>" diff --git a/internal/util/util.go b/internal/util/util.go new file mode 100644 index 0000000..40ce30d --- /dev/null +++ b/internal/util/util.go @@ -0,0 +1,18 @@ +package util + +import ( + "crypto/rand" + "encoding/hex" + "fmt" + "time" +) + +// RandomHex генерирует случайную hex-строку из n байт. +// При ошибке crypto/rand использует time-based fallback. +func RandomHex(n int) string { + b := make([]byte, n) + if _, err := rand.Read(b); err != nil { + return fmt.Sprintf("%x", time.Now().UnixNano())[:n*2] + } + return hex.EncodeToString(b)[:n*2] +}