From f25f8e4d01d5191e1ba08f1889806ef4f5a3850a Mon Sep 17 00:00:00 2001 From: Maksim Totmin Date: Thu, 25 Jun 2026 16:56:49 +0700 Subject: [PATCH] feat: auto-trunk creation, SIP gateway in metrics, gateway WS broadcast - SIPGateway field in NodeMetric model - HasNode() engine method - OnNewNode callback in NATS subscriber - Auto-create balance trunk from new node metric - BroadcastGatewayEvent on WS hub - Gateway status indicators in trunks UI (green/red/gray) - SPA fallback for frontend routing --- cmd/pulse-lets-go/main.go | 69 +++++++++++++++++++++++++++++- internal/api/router.go | 14 ++++++ internal/engine/engine.go | 8 ++++ internal/models/types.go | 1 + internal/nats/subscriber.go | 24 ++++++++--- web/src/routes/trunks/+page.svelte | 35 +++++++++++++-- 6 files changed, 141 insertions(+), 10 deletions(-) diff --git a/cmd/pulse-lets-go/main.go b/cmd/pulse-lets-go/main.go index aeda2df..d14d816 100644 --- a/cmd/pulse-lets-go/main.go +++ b/cmd/pulse-lets-go/main.go @@ -5,6 +5,8 @@ package main import ( "context" + "crypto/rand" + "encoding/hex" "fmt" "log" "net/http" @@ -20,6 +22,7 @@ import ( "github.com/pulse-lets-go/internal/engine" "github.com/pulse-lets-go/internal/esl" filelog "github.com/pulse-lets-go/internal/log" + "github.com/pulse-lets-go/internal/models" "github.com/pulse-lets-go/internal/nats" ) @@ -73,6 +76,42 @@ func main() { } defer sub.Close() + // Авто-создание balance-транков при первой метрике от новой ноды + sub.SetOnNewNode(func(nodeID, sipGateway string) { + trunks, err := cfgMgr.ReadTrunks() + if err != nil { + log.Printf("[main] ошибка чтения транков для авто-создания: %v", err) + return + } + // Проверяем идемпотентность: транк уже существует? + for _, t := range trunks { + if t.Type == "balance" && t.NodeID == nodeID { + return + } + } + // Создаём новый balance-транк + now := time.Now().UTC() + trunkID := "trk-" + randomHex(6) + trunk := models.Trunk{ + ID: trunkID, + Name: fmt.Sprintf("%s баланс", nodeID), + Type: "balance", + NodeID: nodeID, + Gateway: sipGateway, + Codecs: []string{"PCMU", "PCMA"}, + Context: "default", + Enabled: true, + CreatedAt: now, + UpdatedAt: now, + } + trunks = append(trunks, trunk) + if err := cfgMgr.SaveTrunks(trunks); err != nil { + log.Printf("[main] ошибка сохранения авто-созданного транка: %v", err) + return + } + log.Printf("[nats] авто-создан balance-транк: %s → %s (узел: %s)", trunkID, sipGateway, nodeID) + }) + // 6. HTTP API apiHandler := api.NewAPI(eng, cfgMgr, cfg.JWTSecret, cfg.MonitoringAPIKey, sub.IsConnected, cfg) handler := apiHandler.Handler() @@ -105,7 +144,26 @@ func main() { eslClient.GatewaySyncAll(cfg.ESL.SofiaProfile, specs) }, func(eventName string, headers map[string]string, body string) { - log.Printf("[esl] событие: %s", eventName) + gwName := headers["Gateway-Name"] + if gwName == "" { + return + } + // Извлекаем trunk_id из имени gateway: "pulse-ingress-trk-X" → "trk-X" + prefix := cfg.ESL.GatewayPrefix + "-ingress-" + trunkID := strings.TrimPrefix(gwName, prefix) + if trunkID == gwName { + return // префикс не найден, не наш gateway + } + + status := "unknown" + switch eventName { + case "SOFIA::gateway_register": + status = "up" + case "SOFIA::gateway_unregister", "SOFIA::gateway_expire": + status = "down" + } + + apiHandler.BroadcastGatewayEvent(gwName, trunkID, status) }, ) cancel() @@ -236,6 +294,15 @@ func withSPA(webDir string, apiHandler http.Handler) http.Handler { }) } +// randomHex генерирует случайную hex-строку заданной длины. +func randomHex(n int) string { + b := make([]byte, n) + if _, err := rand.Read(b); err != nil { + return fmt.Sprintf("%x", time.Now().UnixNano())[:n] + } + return hex.EncodeToString(b)[:n] +} + // extractHostFromGateway извлекает хост из SIP URI. // "sip:mts-gw.lan:5060" → "mts-gw.lan" func extractHostFromGateway(gateway string) string { diff --git a/internal/api/router.go b/internal/api/router.go index 64b7e2a..e9382e3 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -8,6 +8,7 @@ import ( "github.com/pulse-lets-go/internal/config" "github.com/pulse-lets-go/internal/engine" "github.com/pulse-lets-go/internal/esl" + "github.com/pulse-lets-go/internal/models" ) // API — главная структура HTTP API, агрегирует все зависимости. @@ -127,6 +128,19 @@ func corsMiddleware(next http.Handler) http.Handler { }) } +// BroadcastGatewayEvent транслирует статус gateway всем WS-клиентам. +func (a *API) BroadcastGatewayEvent(gatewayName, trunkID, status string) { + msg := &models.WsMetricsMessage{ + Type: "gateway_update", + Payload: map[string]string{ + "gateway": gatewayName, + "trunk_id": trunkID, + "status": status, + }, + } + a.wsHub.broadcast(msg) +} + // decodeJSON декодирует тело запроса в структуру v. func decodeJSON(r *http.Request, v interface{}) error { defer r.Body.Close() diff --git a/internal/engine/engine.go b/internal/engine/engine.go index fa0d05e..2669c17 100644 --- a/internal/engine/engine.go +++ b/internal/engine/engine.go @@ -144,6 +144,14 @@ func (e *Engine) GetNodeHistory(nodeID string) []NodeSnapshot { return ring.snapshot() } +// HasNode возвращает true, если нода уже зарегистрирована в engine. +func (e *Engine) HasNode(nodeID string) bool { + e.mu.RLock() + defer e.mu.RUnlock() + _, exists := e.nodes[nodeID] + return exists +} + // ToggleNode включает/выключает ноду из распределения. func (e *Engine) ToggleNode(nodeID string, disabled bool, reason string) error { e.mu.Lock() diff --git a/internal/models/types.go b/internal/models/types.go index d7ae6cd..235cad1 100644 --- a/internal/models/types.go +++ b/internal/models/types.go @@ -15,6 +15,7 @@ type NodeMetric struct { 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 --- diff --git a/internal/nats/subscriber.go b/internal/nats/subscriber.go index 11f76f5..fdcbb93 100644 --- a/internal/nats/subscriber.go +++ b/internal/nats/subscriber.go @@ -18,11 +18,12 @@ 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 + nc *natsgo.Conn + sub *natsgo.Subscription + engine *engine.Engine + logger *filelog.Logger + onNewNode func(nodeID, sipGateway string) // вызывается при первой метрике от новой ноды + wg sync.WaitGroup } // NewSubscriber подключается к NATS и запускает подписку. @@ -57,6 +58,11 @@ func NewSubscriber(natsURL, user, password string, eng *engine.Engine, l *filelo return s, nil } +// SetOnNewNode устанавливает callback, вызываемый при первой метрике от новой ноды. +func (s *Subscriber) SetOnNewNode(fn func(nodeID, sipGateway string)) { + s.onNewNode = fn +} + // handleMetric — обработчик входящих сообщений NATS. func (s *Subscriber) handleMetric(msg *natsgo.Msg) { var metric models.NodeMetric @@ -74,8 +80,16 @@ func (s *Subscriber) handleMetric(msg *natsgo.Msg) { metric.MaxCalls = 250 // разумное значение по умолчанию } + // Проверка: это новая нода с sip_gateway? + isNew := !s.engine.HasNode(metric.NodeID) + ns := s.engine.UpdateMetric(&metric) s.logger.Log(ns) + + // Авто-создание balance-транка для новой ноды + if isNew && metric.SIPGateway != "" && s.onNewNode != nil { + s.onNewNode(metric.NodeID, metric.SIPGateway) + } } // IsConnected возвращает true, если NATS-соединение активно. diff --git a/web/src/routes/trunks/+page.svelte b/web/src/routes/trunks/+page.svelte index 8270c29..27ec85b 100644 --- a/web/src/routes/trunks/+page.svelte +++ b/web/src/routes/trunks/+page.svelte @@ -1,14 +1,18 @@