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
This commit is contained in:
parent
533a1fba49
commit
f25f8e4d01
@ -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 {
|
||||
|
||||
@ -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()
|
||||
|
||||
@ -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()
|
||||
|
||||
@ -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 ---
|
||||
|
||||
@ -22,6 +22,7 @@ type Subscriber struct {
|
||||
sub *natsgo.Subscription
|
||||
engine *engine.Engine
|
||||
logger *filelog.Logger
|
||||
onNewNode func(nodeID, sipGateway string) // вызывается при первой метрике от новой ноды
|
||||
wg sync.WaitGroup
|
||||
}
|
||||
|
||||
@ -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-соединение активно.
|
||||
|
||||
@ -1,14 +1,18 @@
|
||||
<script lang="ts">
|
||||
import { browser } from '$app/environment';
|
||||
import { goto } from '$app/navigation';
|
||||
import { onMount } from 'svelte';
|
||||
import { isAuthenticated, getTrunks, createTrunk, updateTrunk, deleteTrunk, type Trunk } from '$lib/api';
|
||||
import { onMount, onDestroy } from 'svelte';
|
||||
import { isAuthenticated, getTrunks, createTrunk, updateTrunk, deleteTrunk, connectWebSocket, type Trunk } from '$lib/api';
|
||||
import { toast } from 'svelte-sonner';
|
||||
import { Plus, Trash2, Save, X, Cable } from 'lucide-svelte';
|
||||
|
||||
let trunks: Trunk[] = [];
|
||||
let loading = true;
|
||||
|
||||
// Gateway статусы с ESLa (up/down/unknown)
|
||||
let ws: WebSocket | null = null;
|
||||
let gatewayStatuses: Record<string, string> = {};
|
||||
|
||||
// Edit mode
|
||||
let editing: string | null = null; // trunk ID being edited
|
||||
let editData: Partial<Trunk> = {};
|
||||
@ -24,7 +28,19 @@
|
||||
|
||||
if (browser && !isAuthenticated()) { goto('/login'); }
|
||||
|
||||
onMount(loadTrunks);
|
||||
onMount(() => {
|
||||
loadTrunks();
|
||||
ws = connectWebSocket((msg) => {
|
||||
if (msg.type === 'gateway_update') {
|
||||
const { trunk_id, status } = msg.payload;
|
||||
gatewayStatuses = { ...gatewayStatuses, [trunk_id]: status };
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
onDestroy(() => {
|
||||
ws?.close();
|
||||
});
|
||||
|
||||
async function loadTrunks() {
|
||||
loading = true;
|
||||
@ -102,6 +118,13 @@
|
||||
}
|
||||
}
|
||||
|
||||
function gatewayIndicator(trunkID: string): string {
|
||||
const s = gatewayStatuses[trunkID] || 'unknown';
|
||||
if (s === 'up') return '🟢';
|
||||
if (s === 'down') return '🔴';
|
||||
return '⚪';
|
||||
}
|
||||
|
||||
function onFilterChange(e: Event) {
|
||||
typeFilter = (e.target as HTMLSelectElement).value;
|
||||
loadTrunks();
|
||||
@ -208,7 +231,11 @@
|
||||
<tr class="border-t hover:bg-secondary/50">
|
||||
<td class="px-4 py-3 font-medium">{t.name}</td>
|
||||
<td class="px-4 py-3"><span class="px-2 py-1 rounded text-xs font-medium {typeBadge(t.type)}">{typeLabel(t.type)}</span></td>
|
||||
<td class="px-4 py-3 font-mono text-xs">{t.gateway}</td>
|
||||
<td class="px-4 py-3 font-mono text-xs">
|
||||
{t.gateway}
|
||||
{#if t.type === 'ingress'}
|
||||
<span class="ml-2" title="Gateway: {gatewayStatuses[t.id] || 'unknown'}">{gatewayIndicator(t.id)}</span>
|
||||
{/if}
|
||||
<td class="px-4 py-3 font-mono text-xs">{t.node_id || '—'}</td>
|
||||
<td class="px-4 py-3">{t.enabled ? '🟢 Вкл' : '🔴 Выкл'}</td>
|
||||
<td class="px-4 py-3 text-right">
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user