en2zmax/pulse-lets-go
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: en2zmax:ff037f9c1e3afad401d07a657b11bbcb2bc62e54
Branches Tags
en2zmax:main
..
compare: en2zmax:e66ac27dd985cb86f158ea394c77db92e5f9330c
Branches Tags
en2zmax:main

No commits in common. "ff037f9c1e3afad401d07a657b11bbcb2bc62e54" and "e66ac27dd985cb86f158ea394c77db92e5f9330c" have entirely different histories.
ff037f9c1e ... e66ac27dd9

34 changed files with 317 additions and 617 deletions
Show Stats Expand all files Collapse all files

41
Makefile
View File

@ -1,12 +1,12 @@
# pulse-lets-go — Telephone Load Balancer # pulse-lets-go — Telephone Load Balancer
# Сборка Go backend + SvelteKit фронта в единый бинарник. # Сборка Go backend + SvelteKit фронта в единый бинарник.
APP := pulse-lets-go APP := pulse-lets-go
BINDIR := bin BINDIR := bin
WEB_DIR := web WEB_DIR := web
PREFIX := /opt/pulse-lets-go DATA_DIR := data
.PHONY: all build build-go build-web clean run test deploy \ .PHONY: all build build-go build-web clean run test install \
emulate emulate-normal emulate-overload emulate-cpu-low emulate-fail-high \ emulate emulate-normal emulate-overload emulate-cpu-low emulate-fail-high \
emulate-node-down emulate-stale emulate-chaos build-emulator \ emulate-node-down emulate-stale emulate-chaos build-emulator \
build-siptest siptest-uas siptest-uac siptest-full build-siptest siptest-uas siptest-uac siptest-full
@ -22,17 +22,13 @@ build-web:
@echo "→ Сборка SvelteKit..." @echo "→ Сборка SvelteKit..."
cd $(WEB_DIR) && npm run build cd $(WEB_DIR) && npm run build
# Сборка Go бинарника (с авто-копированием статики) # Сборка Go бинарника
build-go: build-go:
@mkdir -p $(BINDIR) @mkdir -p $(BINDIR)
@echo "→ Сборка Go backend..." @echo "→ Сборка Go backend..."
@if [ -d $(WEB_DIR)/build ]; then \
rm -rf $(BINDIR)/web && cp -r $(WEB_DIR)/build $(BINDIR)/web; \
echo " статика скопирована в $(BINDIR)/web/"; \
fi
go build -o $(BINDIR)/$(APP) ./cmd/$(APP) go build -o $(BINDIR)/$(APP) ./cmd/$(APP)
# Быстрая сборка только Go (без фронта, без статики) # Быстрая сборка только Go (без фронта)
build-go-only: build-go-only:
@mkdir -p $(BINDIR) @mkdir -p $(BINDIR)
go build -o $(BINDIR)/$(APP) ./cmd/$(APP) go build -o $(BINDIR)/$(APP) ./cmd/$(APP)
@ -47,26 +43,23 @@ run:
build-prod: build-prod:
@mkdir -p $(BINDIR) @mkdir -p $(BINDIR)
cd $(WEB_DIR) && npm run build cd $(WEB_DIR) && npm run build
rm -rf $(BINDIR)/web && cp -r $(WEB_DIR)/build $(BINDIR)/web
CGO_ENABLED=0 go build -ldflags="-s -w" -o $(BINDIR)/$(APP) ./cmd/$(APP) CGO_ENABLED=0 go build -ldflags="-s -w" -o $(BINDIR)/$(APP) ./cmd/$(APP)
# Деплой в PREFIX: бинарник + статика + data + log # Установка: копирует бинарник и статику в /usr/local
deploy: build-prod install: build
@echo "→ Деплой в $(PREFIX)" @echo "→ Установка в /usr/local"
install -d $(PREFIX)/bin $(PREFIX)/data $(PREFIX)/web $(PREFIX)/log mkdir -p /usr/local/share/$(APP)/web
install -m 755 $(BINDIR)/$(APP) $(PREFIX)/bin/ cp -r $(WEB_DIR)/build /usr/local/share/$(APP)/web/
cp -r $(WEB_DIR)/build/. $(PREFIX)/web/ cp $(BINDIR)/$(APP) /usr/local/bin/$(APP)
@echo "✓ Готово: $(PREFIX)" chmod +x /usr/local/bin/$(APP)
@echo " $(PREFIX)/bin/$(APP)" @echo "✓ Установлен в /usr/local/bin/$(APP)"
@echo " $(PREFIX)/web/ (SvelteKit статика)"
@echo " $(PREFIX)/data/ (config.json, trunks.json, users.json)"
@echo " $(PREFIX)/log/ (ASCII-лог метрик)"
# Установка systemd unit # Установка systemd unit
install-systemd: install-systemd:
install -m 644 deploy/$(APP).service /etc/systemd/system/ @echo "→ Установка systemd unit"
cp deploy/$(APP).service /etc/systemd/system/
systemctl daemon-reload systemctl daemon-reload
@echo "✓ systemd unit установлен. Запуск: sudo systemctl start $(APP)" @echo "✓ systemd unit установлен. Запуск: systemctl start $(APP)"
# ============================================================ # ============================================================
# Эмулятор метрик # Эмулятор метрик

41
README.md
View File

@ -423,7 +423,7 @@ end
### systemd ### systemd
```bash ```bash
make deploy # полный деплой в /opt/pulse-lets-go make install # копирует бинарник + web-статику
make install-systemd # устанавливает systemd unit make install-systemd # устанавливает systemd unit
``` ```
@ -436,15 +436,16 @@ After=network.target
[Service] [Service]
User=pulse User=pulse
Group=pulse Group=pulse
ExecStart=/opt/pulse-lets-go/bin/pulse-lets-go WorkingDirectory=/usr/local/share/pulse-lets-go
WorkingDirectory=/opt/pulse-lets-go Environment=PULSE_DATA_DIR=/etc/pulse-lets-go
ExecStart=/usr/local/bin/pulse-lets-go
Restart=always Restart=always
RestartSec=5 RestartSec=5
NoNewPrivileges=true NoNewPrivileges=true
PrivateTmp=true PrivateTmp=true
ProtectSystem=strict ProtectSystem=strict
ProtectHome=yes ProtectHome=yes
ReadWritePaths=/opt/pulse-lets-go/data /opt/pulse-lets-go/log ReadWritePaths=/etc/pulse-lets-go /var/log/pulse-lets-go
[Install] [Install]
WantedBy=multi-user.target WantedBy=multi-user.target
@ -453,15 +454,14 @@ WantedBy=multi-user.target
### Структура директорий ### Структура директорий
``` ```
/opt/pulse-lets-go/ /usr/local/share/pulse-lets-go/
├── bin/pulse-lets-go # Бинарник ├── web/build/ # SvelteKit статика
├── data/ /etc/pulse-lets-go/
│ ├── config.json # Конфигурация ├── config.json # Конфигурация
│ ├── trunks.json # Транки ├── trunks.json # Транки
│ └── users.json # Пользователи ├── users.json # Пользователи
├── web/ # SvelteKit статика /var/log/pulse-lets-go/
└── log/ └── metrics.YYYY-MM-DD.log # ASCII-лог метрик
└── metrics.YYYY-MM-DD.log # ASCII-лог метрик
``` ```
### Security ### Security
@ -497,13 +497,13 @@ fs_cli -x "load mod_curl"
# <load module="mod_curl"/> # <load module="mod_curl"/>
``` ```
### Порядок деплоя (8 шагов) ### Порядок деплоя (7 шагов)
**Шаг 1 — Деплой на сервер** **Шаг 1 — Скопировать бинарник и конфиги**
```bash ```bash
make deploy make build-prod
scp -r /opt/pulse-lets-go devadmin@10.101.60.81:/opt/ scp bin/pulse-lets-go devadmin@10.101.60.81:/usr/local/bin/
scp contrib/route.lua devadmin@10.101.60.81:/tmp/ scp -r contrib/route.lua devadmin@10.101.60.81:/tmp/
``` ```
**Шаг 2 — NATS сервер** (если не установлен) **Шаг 2 — NATS сервер** (если не установлен)
@ -562,10 +562,11 @@ curl -X POST .../api/trunks
**Шаг 7 — Запуск pulse-lets-go** **Шаг 7 — Запуск pulse-lets-go**
```bash ```bash
# config.json создастся автоматически при первом запуске в /opt/pulse-lets-go/data/ sudo mkdir -p /etc/pulse-lets-go /var/log/pulse-lets-go
# config.json создастся автоматически при первом запуске
# Добавить esl блок в config.json: # Добавить esl блок в config.json:
# "esl": {"host": "127.0.0.1", "port": 8021, "password": "ClueCon"} # "esl": {"host": "127.0.0.1", "port": 8021, "password": "ClueCon"}
/opt/pulse-lets-go/bin/pulse-lets-go PULSE_DATA_DIR=/etc/pulse-lets-go ./pulse-lets-go
``` ```
**Шаг 8 — Проверка** **Шаг 8 — Проверка**

4
cmd/emulator/main.go
View File

@ -49,7 +49,7 @@ type valueSpec struct {
isRand bool 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 { func rrange(min, max float64) valueSpec {
return valueSpec{min: min, max: max, isRand: true} return valueSpec{min: min, max: max, isRand: true}
} }
@ -71,7 +71,7 @@ type nodeSpec struct {
idleCPU valueSpec idleCPU valueSpec
loadAvg valueSpec loadAvg valueSpec
failRate valueSpec failRate valueSpec
sipGateway string // SIP-адрес для авто-создания транка sipGateway string // SIP-адрес для авто-создания транка
} }
func (ns *nodeSpec) generate() nodeMetric { func (ns *nodeSpec) generate() nodeMetric {

7
cmd/pulse-lets-go-agent/agent/collector_freeswitch.go
View File

@ -32,15 +32,12 @@ func (fc *FSCollector) Connect() error {
fc.client.ConnectWithRetry() fc.client.ConnectWithRetry()
// Ждём подключения с таймаутом // Ждём подключения с таймаутом
timer := time.NewTimer(100 * time.Millisecond)
defer timer.Stop()
for !fc.client.IsConnected() { for !fc.client.IsConnected() {
timer.Reset(100 * time.Millisecond)
select { select {
case <-ctx.Done(): case <-ctx.Done():
return fmt.Errorf("esl connect timeout") return fmt.Errorf("esl connect timeout")
case <-timer.C: case <-time.After(100 * time.Millisecond):
// Выходим из цикла после ConnectWithRetry
return nil return nil
} }
} }

24
cmd/pulse-lets-go-agent/agent/config.go
View File

@ -27,18 +27,18 @@ import (
// Config — конфигурация агента (agent.json). // Config — конфигурация агента (agent.json).
type Config struct { type Config struct {
NodeID string `json:"node_id"` // уникальный ID ноды (uc06, ses-sip) NodeID string `json:"node_id"` // уникальный ID ноды (uc06, ses-sip)
Type string `json:"type"` // "freeswitch" или "asterisk" Type string `json:"type"` // "freeswitch" или "asterisk"
NatsURL string `json:"nats_url"` // NATS URL NatsURL string `json:"nats_url"` // NATS URL
NatsUser string `json:"nats_user,omitempty"` NatsUser string `json:"nats_user,omitempty"`
NatsPassword string `json:"nats_password,omitempty"` NatsPassword string `json:"nats_password,omitempty"`
IntervalSec int `json:"interval_sec"` // интервал сбора (default: 5) IntervalSec int `json:"interval_sec"` // интервал сбора (default: 5)
MaxCalls int `json:"max_calls"` // ёмкость ноды (default: 250) MaxCalls int `json:"max_calls"` // ёмкость ноды (default: 250)
SIPGateway string `json:"sip_gateway"` // SIP-адрес для auto-транка SIPGateway string `json:"sip_gateway"` // SIP-адрес для auto-транка
SIPGatewayAuto bool `json:"sip_gateway_auto"` // авто-определить из PBX SIPGatewayAuto bool `json:"sip_gateway_auto"` // авто-определить из PBX
FailureWindow int `json:"failure_window"` // окно для call_failure_rate (default: 1000) FailureWindow int `json:"failure_window"` // окно для call_failure_rate (default: 1000)
ESL ESLCfg `json:"esl,omitempty"` ESL ESLCfg `json:"esl,omitempty"`
AMI AMICfg `json:"ami,omitempty"` AMI AMICfg `json:"ami,omitempty"`
} }
// ESLCfg — настройки подключения к FreeSWITCH ESL. // ESLCfg — настройки подключения к FreeSWITCH ESL.

4
cmd/pulse-lets-go-agent/agent/publisher.go
View File

@ -12,8 +12,8 @@ import (
// Publisher публикует метрики в NATS. // Publisher публикует метрики в NATS.
type Publisher struct { type Publisher struct {
nc *natsgo.Conn nc *natsgo.Conn
nodeID string nodeID string
} }
// NewPublisher создаёт NATS publisher. // NewPublisher создаёт NATS publisher.

6
cmd/pulse-lets-go-agent/agent/system.go
View File

@ -15,9 +15,9 @@ func nowTS() int64 {
// SystemStats содержит системные метрики (CPU, load). // SystemStats содержит системные метрики (CPU, load).
type SystemStats struct { type SystemStats struct {
Load1 float64 // load average 1m Load1 float64 // load average 1m
Load5 float64 // load average 5m Load5 float64 // load average 5m
Load15 float64 // load average 15m Load15 float64 // load average 15m
IdleCPU float64 // процент простоя CPU IdleCPU float64 // процент простоя CPU
} }

43
cmd/pulse-lets-go/main.go
View File

@ -5,6 +5,8 @@ package main
import ( import (
"context" "context"
"crypto/rand"
"encoding/hex"
"fmt" "fmt"
"log" "log"
"net/http" "net/http"
@ -22,17 +24,15 @@ import (
filelog "github.com/pulse-lets-go/internal/log" filelog "github.com/pulse-lets-go/internal/log"
"github.com/pulse-lets-go/internal/models" "github.com/pulse-lets-go/internal/models"
"github.com/pulse-lets-go/internal/nats" "github.com/pulse-lets-go/internal/nats"
"github.com/pulse-lets-go/internal/util"
) )
func main() { func main() {
// Определяем рабочую директорию // Определяем рабочую директорию
// По умолчанию data/ — родительская папка от бинарника.
// Пример: /opt/pulse-lets-go/bin/pulse-lets-go → /opt/pulse-lets-go/data
exe, _ := os.Executable() exe, _ := os.Executable()
exeDir := filepath.Dir(exe) exeDir := filepath.Dir(exe)
dataDir := filepath.Join(filepath.Dir(exeDir), "data") dataDir := filepath.Join(exeDir, "data")
// Разрешаем переопределение через переменную окружения
if envDir := os.Getenv("PULSE_DATA_DIR"); envDir != "" { if envDir := os.Getenv("PULSE_DATA_DIR"); envDir != "" {
dataDir = envDir dataDir = envDir
} }
@ -91,7 +91,7 @@ func main() {
} }
// Создаём новый balance-транк // Создаём новый balance-транк
now := time.Now().UTC() now := time.Now().UTC()
trunkID := "trk-" + util.RandomHex(6) trunkID := "trk-" + randomHex(6)
trunk := models.Trunk{ trunk := models.Trunk{
ID: trunkID, ID: trunkID,
Name: fmt.Sprintf("%s баланс", nodeID), Name: fmt.Sprintf("%s баланс", nodeID),
@ -113,7 +113,7 @@ func main() {
}) })
// 6. HTTP API // 6. HTTP API
apiHandler := api.NewAPI(eng, cfgMgr, cfg.GetJWTSecret(), cfg.MonitoringAPIKey, sub.IsConnected, cfg) apiHandler := api.NewAPI(eng, cfgMgr, cfg.JWTSecret, cfg.MonitoringAPIKey, sub.IsConnected, cfg)
handler := apiHandler.Handler() handler := apiHandler.Handler()
// 6.1 ESL-клиент (опционально — если сконфигурирован) // 6.1 ESL-клиент (опционально — если сконфигурирован)
@ -177,19 +177,6 @@ func main() {
} }
}() }()
// 6.2 Periodic WS broadcast + FS stats poll
go func() {
ticker := time.NewTicker(5 * time.Second)
for range ticker.C {
if eslClient != nil && eslClient.IsConnected() {
if stats, err := eslClient.FetchFsStats(); err == nil {
apiHandler.UpdateFsStats(stats)
}
}
apiHandler.BroadcastAll()
}
}()
// Подключаем раздачу SvelteKit статики, если директория существует // Подключаем раздачу SvelteKit статики, если директория существует
webDir := findWebDir(exeDir) webDir := findWebDir(exeDir)
if webDir != "" { if webDir != "" {
@ -243,19 +230,20 @@ func main() {
log.Printf("[main] ошибка остановки HTTP сервера: %v", err) log.Printf("[main] ошибка остановки HTTP сервера: %v", err)
} }
sub.Close()
metricsLogger.Close() metricsLogger.Close()
log.Println("[main] pulse-lets-go остановлен") log.Println("[main] pulse-lets-go остановлен")
} }
// findWebDir ищет директорию со SvelteKit сборкой (web/build/). // findWebDir ищет директорию со SvelteKit сборкой (web/build/).
// Пробует по порядку: PULSE_WEB_DIR, CWD, ../web/build, рядом с бинарником. // Пробует по порядку: переменная окружения, рядом с бинарником, в корне проекта, CWD.
func findWebDir(exeDir string) string { func findWebDir(exeDir string) string {
candidates := []string{ candidates := []string{
os.Getenv("PULSE_WEB_DIR"), os.Getenv("PULSE_WEB_DIR"),
"web/build", // ./web/build (CWD — dev, systemd working dir) filepath.Join(exeDir, "web", "build"), // bin/web/build
filepath.Join(filepath.Dir(exeDir), "web", "build"), // ../web/build (корень проекта, dev с bin/) filepath.Join(filepath.Dir(exeDir), "web", "build"), // ../web/build (корень проекта)
filepath.Join(exeDir, "web", "build"), // рядом с бинарником (production install, make deploy) "web/build", // ./web/build (CWD)
} }
for _, d := range candidates { for _, d := range candidates {
if d == "" { if d == "" {
@ -306,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. // extractHostFromGateway извлекает хост из SIP URI.
// "sip:mts-gw.lan:5060" → "mts-gw.lan" // "sip:mts-gw.lan:5060" → "mts-gw.lan"
func extractHostFromGateway(gateway string) string { func extractHostFromGateway(gateway string) string {

32
cmd/siptest/main.go
View File

@ -1,16 +1,16 @@
// SIP тестер для e2e проверки pulse-lets-go + FreeSWITCH. // SIP тестер для e2e проверки pulse-lets-go + FreeSWITCH.
// Два режима: // Два режима:
// // uas — отвечает 200 OK на SIP INVITE, симулирует PBX-ноду
// uas — отвечает 200 OK на SIP INVITE, симулирует PBX-ноду // uac — шлёт SIP INVITE в FreeSWITCH, симулирует оператора
// uac — шлёт SIP INVITE в FreeSWITCH, симулирует оператора
// //
// Примеры: // Примеры:
// // ./siptest -mode uas (PBX-нода на порту 5090)
// ./siptest -mode uas (PBX-нода на порту 5090) // ./siptest -mode uac -r 10 -l 100 (оператор, 10 CPS, до 100 одновременных)
// ./siptest -mode uac -r 10 -l 100 (оператор, 10 CPS, до 100 одновременных)
package main package main
import ( import (
"crypto/rand"
"encoding/hex"
"flag" "flag"
"fmt" "fmt"
"log" "log"
@ -19,8 +19,6 @@ import (
"sync" "sync"
"sync/atomic" "sync/atomic"
"time" "time"
"github.com/pulse-lets-go/internal/util"
) )
const ( const (
@ -31,9 +29,9 @@ const (
// Статистика // Статистика
var ( var (
callsSent atomic.Int64 callsSent atomic.Int64
callsOK atomic.Int64 callsOK atomic.Int64
callsFailed atomic.Int64 callsFailed atomic.Int64
) )
func main() { func main() {
@ -105,7 +103,7 @@ func buildSIPResponse(invite, localAddr string) string {
// Добавляем tag к To если его нет // Добавляем tag к To если его нет
to = strings.TrimSpace(to) to = strings.TrimSpace(to)
if !strings.Contains(to, ";tag=") { if !strings.Contains(to, ";tag=") {
to += ";tag=uas-" + util.RandomHex(4) to += ";tag=uas-" + randomHex(4)
} }
return fmt.Sprintf("SIP/2.0 200 OK\r\n"+ return fmt.Sprintf("SIP/2.0 200 OK\r\n"+
@ -215,8 +213,8 @@ func runUAC(rate, limit, maxCalls int, dest, caller, host string) {
} }
func buildSIPInvite(dest, caller string, callNum int, srcAddr, fsAddr string) string { func buildSIPInvite(dest, caller string, callNum int, srcAddr, fsAddr string) string {
callID := fmt.Sprintf("call-%d-%s@%s", callNum, util.RandomHex(4), "127.0.0.1") callID := fmt.Sprintf("call-%d-%s@%s", callNum, randomHex(4), "127.0.0.1")
branch := "z9hG4bK-" + util.RandomHex(8) branch := "z9hG4bK-" + randomHex(8)
return fmt.Sprintf("INVITE sip:%s@%s SIP/2.0\r\n"+ return fmt.Sprintf("INVITE sip:%s@%s SIP/2.0\r\n"+
"Via: SIP/2.0/UDP %s;branch=%s\r\n"+ "Via: SIP/2.0/UDP %s;branch=%s\r\n"+
@ -229,3 +227,9 @@ func buildSIPInvite(dest, caller string, callNum int, srcAddr, fsAddr string) st
"\r\n", "\r\n",
dest, fsAddr, srcAddr, branch, caller, callNum, dest, callID, caller, srcAddr) 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]
}

10
deploy/pulse-lets-go.service
View File

@ -1,14 +1,16 @@
[Unit] [Unit]
Description=pulse-lets-go — Telephone Load Balancer Description=pulse-lets-go — Telephone Load Balancer
Documentation=https://github.com/pulse-lets-go Documentation=https://github.com/pulse-lets-go
After=network.target After=network.target nats.service
Wants=nats.service
[Service] [Service]
Type=simple Type=simple
User=pulse User=pulse
Group=pulse Group=pulse
ExecStart=/opt/pulse-lets-go/bin/pulse-lets-go ExecStart=/usr/local/bin/pulse-lets-go
WorkingDirectory=/opt/pulse-lets-go WorkingDirectory=/usr/local/share/pulse-lets-go
Environment=PULSE_DATA_DIR=/etc/pulse-lets-go
Restart=always Restart=always
RestartSec=5 RestartSec=5
@ -17,7 +19,7 @@ NoNewPrivileges=yes
PrivateTmp=yes PrivateTmp=yes
ProtectSystem=strict ProtectSystem=strict
ProtectHome=yes ProtectHome=yes
ReadWritePaths=/opt/pulse-lets-go/data /opt/pulse-lets-go/log ReadWritePaths=/etc/pulse-lets-go /var/log/pulse-lets-go
# Логирование # Логирование
StandardOutput=journal StandardOutput=journal

36
internal/ami/client.go
View File

@ -63,7 +63,7 @@ func NewClient(host string, port int, username, password string) *Client {
func (c *Client) Connect() error { func (c *Client) Connect() error {
c.shouldReconn.Store(true) c.shouldReconn.Store(true)
addr := net.JoinHostPort(c.host, fmt.Sprintf("%d", c.port)) addr := fmt.Sprintf("%s:%d", c.host, c.port)
dialer := net.Dialer{Timeout: 5 * time.Second} dialer := net.Dialer{Timeout: 5 * time.Second}
conn, err := dialer.Dial("tcp", addr) conn, err := dialer.Dial("tcp", addr)
if err != nil { if err != nil {
@ -268,9 +268,11 @@ func (c *Client) readEventsLoop() {
if err != nil { if err != nil {
if err == io.EOF || strings.Contains(err.Error(), "use of closed network connection") { if err == io.EOF || strings.Contains(err.Error(), "use of closed network connection") {
c.connected.Store(false) c.connected.Store(false)
log.Printf("[ami] соединение разорвано, реконнект...") log.Printf("[ami] соединение разорвано")
c.reconnect() if c.shouldReconn.Load() {
continue go c.ConnectWithRetry()
}
return
} }
time.Sleep(100 * time.Millisecond) time.Sleep(100 * time.Millisecond)
continue continue
@ -289,29 +291,3 @@ 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] реконнект успешен")
}

12
internal/api/auth.go
View File

@ -27,7 +27,7 @@ type claims struct {
type contextKey string type contextKey string
const ( const (
userContextKey contextKey = "user" userContextKey contextKey = "user"
refreshTokenKey contextKey = "refresh_token" refreshTokenKey contextKey = "refresh_token"
) )
@ -39,8 +39,8 @@ type UserInfo struct {
} }
const ( const (
jwtExpiry = 15 * time.Minute jwtExpiry = 15 * time.Minute
refreshExpiry = 24 * time.Hour refreshExpiry = 24 * time.Hour
) )
// generateJWT создаёт access+refresh токены. // generateJWT создаёт access+refresh токены.
@ -139,7 +139,7 @@ func (a *API) apiKeyMiddleware(next http.Handler) http.Handler {
// handleLogin обрабатывает POST /api/auth/login. // handleLogin обрабатывает POST /api/auth/login.
func (a *API) handleLogin(w http.ResponseWriter, r *http.Request) { func (a *API) handleLogin(w http.ResponseWriter, r *http.Request) {
var req models.LoginRequest var req models.LoginRequest
if err := decodeJSON(r, &req); err != nil { if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, "некорректный JSON") writeError(w, http.StatusBadRequest, "некорректный JSON")
return return
} }
@ -199,6 +199,10 @@ func (a *API) handleLogin(w http.ResponseWriter, r *http.Request) {
// handleRefresh выдаёт новый access токен по refresh токену. // handleRefresh выдаёт новый access токен по refresh токену.
func (a *API) handleRefresh(w http.ResponseWriter, r *http.Request) { func (a *API) handleRefresh(w http.ResponseWriter, r *http.Request) {
tokenStr := extractBearerToken(r) tokenStr := extractBearerToken(r)
if tokenStr == "" {
// пробуем получить из query param (для WebSocket)
tokenStr = r.URL.Query().Get("token")
}
if tokenStr == "" { if tokenStr == "" {
writeError(w, http.StatusUnauthorized, "требуется токен") writeError(w, http.StatusUnauthorized, "требуется токен")
return return

10
internal/api/health.go
View File

@ -17,12 +17,12 @@ func (a *API) handleHealth(w http.ResponseWriter, r *http.Request) {
if a.eslClient != nil { if a.eslClient != nil {
eslStats := a.eslClient.GetStats() eslStats := a.eslClient.GetStats()
eslInfo = map[string]interface{}{ eslInfo = map[string]interface{}{
"status": eslStats.Status, "status": eslStats.Status,
"host": eslStats.Host, "host": eslStats.Host,
"uptime_seconds": eslStats.UptimeSec, "uptime_seconds": eslStats.UptimeSec,
"reconnects": eslStats.Reconnects, "reconnects": eslStats.Reconnects,
"gateway_ops": eslStats.GatewayOps, "gateway_ops": eslStats.GatewayOps,
"events_recv": eslStats.EventsRecv, "events_recv": eslStats.EventsRecv,
} }
} }

11
internal/api/logging.go
View File

@ -1,13 +1,10 @@
package api package api
import ( import (
"bufio"
"crypto/rand" "crypto/rand"
"encoding/hex" "encoding/hex"
"encoding/json" "encoding/json"
"fmt"
"log" "log"
"net"
"net/http" "net/http"
"time" "time"
) )
@ -85,11 +82,3 @@ func (w *loggingResponseWriter) WriteHeader(code int) {
func (w *loggingResponseWriter) Write(b []byte) (int, error) { func (w *loggingResponseWriter) Write(b []byte) (int, error) {
return w.ResponseWriter.Write(b) return w.ResponseWriter.Write(b)
} }
func (w *loggingResponseWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) {
hijacker, ok := w.ResponseWriter.(http.Hijacker)
if !ok {
return nil, nil, fmt.Errorf("loggingResponseWriter: underlying ResponseWriter does not implement http.Hijacker")
}
return hijacker.Hijack()
}

20
internal/api/monitoring.go
View File

@ -38,26 +38,6 @@ func (a *API) handlePrometheus(w http.ResponseWriter, r *http.Request) {
var sb strings.Builder 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 { for _, n := range nodes {
id := sanitizePromLabel(n.NodeID) id := sanitizePromLabel(n.NodeID)

18
internal/api/ratelimit.go
View File

@ -9,12 +9,12 @@ import (
// rateLimiter — in-memory token bucket rate limiter. // rateLimiter — in-memory token bucket rate limiter.
type rateLimiter struct { type rateLimiter struct {
mu sync.Mutex mu sync.Mutex
buckets map[string]*tokenBucket buckets map[string]*tokenBucket
rate float64 // токенов в секунду rate float64 // токенов в секунду
burst int // максимальный размер бакета burst int // максимальный размер бакета
cleanupInterval time.Duration cleanupInterval time.Duration
lastCleanup time.Time lastCleanup time.Time
} }
type tokenBucket struct { type tokenBucket struct {
@ -28,11 +28,11 @@ func newRateLimiter(ratePerSec, burst int) *rateLimiter {
burst = ratePerSec burst = ratePerSec
} }
return &rateLimiter{ return &rateLimiter{
buckets: make(map[string]*tokenBucket), buckets: make(map[string]*tokenBucket),
rate: float64(ratePerSec), rate: float64(ratePerSec),
burst: burst, burst: burst,
cleanupInterval: 60 * time.Second, cleanupInterval: 60 * time.Second,
lastCleanup: time.Now(), lastCleanup: time.Now(),
} }
} }

4
internal/api/route.go
View File

@ -14,7 +14,6 @@ func (a *API) handleRoute(w http.ResponseWriter, r *http.Request) {
// Нет зарегистрированных нод // Нет зарегистрированных нод
if nodeID == "" { if nodeID == "" {
a.engine.IncrementRouteFallbacks()
writeJSON(w, http.StatusServiceUnavailable, models.RouteResponse{ writeJSON(w, http.StatusServiceUnavailable, models.RouteResponse{
Error: "no_nodes_registered", Error: "no_nodes_registered",
}) })
@ -23,13 +22,12 @@ func (a *API) handleRoute(w http.ResponseWriter, r *http.Request) {
// Все ноды unhealthy — fallback // Все ноды unhealthy — fallback
if fallback || score < 0 { if fallback || score < 0 {
a.engine.IncrementRouteFallbacks()
fallbackGW, _ := a.findFallbackGateway() fallbackGW, _ := a.findFallbackGateway()
nodes := a.getRouteNodeInfo() nodes := a.getRouteNodeInfo()
writeJSON(w, http.StatusOK, models.RouteResponse{ writeJSON(w, http.StatusOK, models.RouteResponse{
Fallback: true, Fallback: true,
SIPGateway: fallbackGW, SIPGateway: fallbackGW,
Reason: "all_nodes_unhealthy", Reason: "all_nodes_unhealthy",
Nodes: nodes, Nodes: nodes,
}) })

43
internal/api/router.go
View File

@ -4,7 +4,6 @@ import (
"encoding/json" "encoding/json"
"fmt" "fmt"
"net/http" "net/http"
"sync"
"github.com/pulse-lets-go/internal/config" "github.com/pulse-lets-go/internal/config"
"github.com/pulse-lets-go/internal/engine" "github.com/pulse-lets-go/internal/engine"
@ -19,16 +18,11 @@ type API struct {
jwtSecret string jwtSecret string
monitoringAPIKey string monitoringAPIKey string
wsHub *wsHub wsHub *wsHub
natsConnected func() bool // колбэк для проверки NATS-статуса (health) natsConnected func() bool // колбэк для проверки NATS-статуса (health)
eslClient *esl.Client // ESL-клиент (nil если не сконфигурирован) eslClient *esl.Client // ESL-клиент (nil если не сконфигурирован)
logFormat string logFormat string
routeLimiter *rateLimiter routeLimiter *rateLimiter
apiLimiter *rateLimiter apiLimiter *rateLimiter
gatewayMu sync.RWMutex
gatewayStates map[string]string // trunkID → "up"/"down"
fsStats esl.FsStats
fsStatsMu sync.RWMutex
} }
// NewAPI создаёт новый HTTP API с заданными зависимостями. // NewAPI создаёт новый HTTP API с заданными зависимостями.
@ -41,7 +35,6 @@ func NewAPI(eng *engine.Engine, cfgMgr *config.Manager, jwtSecret, monitoringAPI
wsHub: newWSHub(), wsHub: newWSHub(),
natsConnected: natsFn, natsConnected: natsFn,
logFormat: cfg.Log.Format, logFormat: cfg.Log.Format,
gatewayStates: make(map[string]string),
} }
if cfg.RateLimit.Enabled { if cfg.RateLimit.Enabled {
a.routeLimiter = newRateLimiter(cfg.RateLimit.RoutePerSec, cfg.RateLimit.RoutePerSec) a.routeLimiter = newRateLimiter(cfg.RateLimit.RoutePerSec, cfg.RateLimit.RoutePerSec)
@ -137,10 +130,6 @@ func corsMiddleware(next http.Handler) http.Handler {
// BroadcastGatewayEvent транслирует статус gateway всем WS-клиентам. // BroadcastGatewayEvent транслирует статус gateway всем WS-клиентам.
func (a *API) BroadcastGatewayEvent(gatewayName, trunkID, status string) { func (a *API) BroadcastGatewayEvent(gatewayName, trunkID, status string) {
a.gatewayMu.Lock()
a.gatewayStates[trunkID] = status
a.gatewayMu.Unlock()
msg := &models.WsMetricsMessage{ msg := &models.WsMetricsMessage{
Type: "gateway_update", Type: "gateway_update",
Payload: map[string]string{ Payload: map[string]string{
@ -152,35 +141,9 @@ func (a *API) BroadcastGatewayEvent(gatewayName, trunkID, status string) {
a.wsHub.broadcast(msg) a.wsHub.broadcast(msg)
} }
func (a *API) gatewayStats() (total, up, down int) { // decodeJSON декодирует тело запроса в структуру v.
a.gatewayMu.RLock()
defer a.gatewayMu.RUnlock()
for _, s := range a.gatewayStates {
total++
switch s {
case "up":
up++
case "down":
down++
}
}
return
}
// UpdateFsStats обновляет кэш метрик FreeSWITCH (вызывается из ticker).
func (a *API) UpdateFsStats(stats esl.FsStats) {
a.fsStatsMu.Lock()
a.fsStats = stats
a.fsStatsMu.Unlock()
}
// maxBodySize — максимальный размер тела запроса в байтах (1 MiB).
const maxBodySize = 1 << 20
// decodeJSON декодирует тело запроса в структуру v с ограничением размера.
func decodeJSON(r *http.Request, v interface{}) error { func decodeJSON(r *http.Request, v interface{}) error {
defer r.Body.Close() defer r.Body.Close()
r.Body = http.MaxBytesReader(nil, r.Body, maxBodySize)
if err := json.NewDecoder(r.Body).Decode(v); err != nil { if err := json.NewDecoder(r.Body).Decode(v); err != nil {
return fmt.Errorf("декодирование JSON: %w", err) return fmt.Errorf("декодирование JSON: %w", err)
} }

5
internal/api/trunks.go
View File

@ -268,9 +268,8 @@ func (a *API) eslPushGatewayDelete(trunk models.Trunk) {
// gatewayParams возвращает параметры для создания FS gateway на основе транка. // gatewayParams возвращает параметры для создания FS gateway на основе транка.
func (a *API) gatewayParams(trunk models.Trunk) (profile, name, proxy string) { func (a *API) gatewayParams(trunk models.Trunk) (profile, name, proxy string) {
cfg := a.readConfig() profile = a.readConfig().ESL.SofiaProfile
profile = cfg.ESL.SofiaProfile name = fmt.Sprintf("pulse-ingress-%s", trunk.ID)
name = fmt.Sprintf("%s-ingress-%s", cfg.ESL.GatewayPrefix, trunk.ID)
proxy = extractHost(trunk.Gateway) proxy = extractHost(trunk.Gateway)
return return
} }

15
internal/api/users.go
View File

@ -1,12 +1,13 @@
package api package api
import ( import (
"crypto/rand"
"encoding/hex"
"fmt" "fmt"
"net/http" "net/http"
"time" "time"
"github.com/pulse-lets-go/internal/models" "github.com/pulse-lets-go/internal/models"
"github.com/pulse-lets-go/internal/util"
"golang.org/x/crypto/bcrypt" "golang.org/x/crypto/bcrypt"
) )
@ -226,5 +227,15 @@ func findUserIndex(users []models.User, id string) int {
// generateID генерирует ID вида "prefix-XXXXX". // generateID генерирует ID вида "prefix-XXXXX".
func generateID(prefix string) string { func generateID(prefix string) string {
return fmt.Sprintf("%s-%s", prefix, util.RandomHex(6)) 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]
} }

49
internal/api/ws.go
View File

@ -104,7 +104,12 @@ func (a *API) handleWS(w http.ResponseWriter, r *http.Request) {
// Отправляем текущее состояние при подключении // Отправляем текущее состояние при подключении
go func() { go func() {
a.BroadcastAll() nodes := a.engine.GetAllNodes()
msg := &models.WsMetricsMessage{
Type: "nodes_update",
Payload: nodes,
}
a.wsHub.broadcast(msg)
}() }()
// Читаем из вебсокета (ping/pong + закрытие) // Читаем из вебсокета (ping/pong + закрытие)
@ -120,7 +125,7 @@ func (a *API) handleWS(w http.ResponseWriter, r *http.Request) {
}() }()
} }
// BroadcastMetrics рассылает метрики нод всем WS-клиентам. // BroadcastMetrics рассылает метрики всем WS-клиентам (вызывается engine при обновлении).
func (a *API) BroadcastMetrics() { func (a *API) BroadcastMetrics() {
nodes := a.engine.GetAllNodes() nodes := a.engine.GetAllNodes()
msg := &models.WsMetricsMessage{ msg := &models.WsMetricsMessage{
@ -130,46 +135,6 @@ func (a *API) BroadcastMetrics() {
a.wsHub.broadcast(msg) a.wsHub.broadcast(msg)
} }
// BroadcastAll рассылает ноды и состояние балансировщика.
func (a *API) BroadcastAll() {
a.BroadcastMetrics()
a.broadcastBalancerUpdate()
}
func (a *API) broadcastBalancerUpdate() {
stats := a.engine.GetHealthStats()
gwTotal, gwUp, gwDown := a.gatewayStats()
info := models.BalancerInfo{
NatsConnected: a.natsConnected != nil && a.natsConnected(),
GatewaysTotal: gwTotal,
GatewaysUp: gwUp,
GatewaysDown: gwDown,
RouteTotal: stats.RouteRequests,
RouteFallbacks: stats.RouteFallbacks,
UptimeSec: stats.UptimeSeconds,
}
if a.eslClient != nil {
eslStats := a.eslClient.GetStats()
info.EslConnected = eslStats.Status == "connected"
info.EslUptimeSec = eslStats.UptimeSec
info.EslReconnects = eslStats.Reconnects
a.fsStatsMu.RLock()
info.FsActiveCalls = a.fsStats.ActiveCalls
info.FsMaxSessions = a.fsStats.MaxSessions
info.FsUptime = a.fsStats.FsUptime
a.fsStatsMu.RUnlock()
}
msg := &models.WsMetricsMessage{
Type: "balancer_update",
Payload: info,
}
a.wsHub.broadcast(msg)
}
// validateToken проверяет JWT и возвращает UserInfo. // validateToken проверяет JWT и возвращает UserInfo.
func (a *API) validateToken(tokenStr string) (*UserInfo, error) { func (a *API) validateToken(tokenStr string) (*UserInfo, error) {
token, err := parseJWT(tokenStr, a.jwtSecret) token, err := parseJWT(tokenStr, a.jwtSecret)

61
internal/config/manager.go
View File

@ -27,9 +27,9 @@ type ScoringWeights struct {
// ScoringConfig — пороги и веса для scoring engine. // ScoringConfig — пороги и веса для scoring engine.
type ScoringConfig struct { type ScoringConfig struct {
IdleCPUMin float64 `json:"idle_cpu_min"` IdleCPUMin float64 `json:"idle_cpu_min"`
CallFailureRateLethal float64 `json:"call_failure_rate_lethal"` CallFailureRateLethal float64 `json:"call_failure_rate_lethal"`
Weights ScoringWeights `json:"weights"` Weights ScoringWeights `json:"weights"`
} }
// TLSConfig — настройки HTTPS. // TLSConfig — настройки HTTPS.
@ -54,11 +54,11 @@ type LogConfig struct {
// ESLConfig — настройки подключения к FreeSWITCH ESL. // ESLConfig — настройки подключения к FreeSWITCH ESL.
type ESLConfig struct { type ESLConfig struct {
Host string `json:"host"` // если пустой — ESL не запускается Host string `json:"host"` // если пустой — ESL не запускается
Port int `json:"port"` // порт ESL (по умолчанию 8021) Port int `json:"port"` // порт ESL (по умолчанию 8021)
Password string `json:"password"` Password string `json:"password"`
PasswordEnv string `json:"password_env"` // имя переменной окружения с паролем PasswordEnv string `json:"password_env"` // имя переменной окружения с паролем
SofiaProfile string `json:"sofia_profile"` // профиль Sofia (external/internal) SofiaProfile string `json:"sofia_profile"` // профиль Sofia (external/internal)
GatewayPrefix string `json:"gateway_prefix"` // префикс gateway (pulse) GatewayPrefix string `json:"gateway_prefix"` // префикс gateway (pulse)
} }
@ -72,31 +72,20 @@ func (e *ESLConfig) GetPassword() string {
return e.Password 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). // Config — корневая конфигурация приложения (config.json).
type Config struct { type Config struct {
NatsURL string `json:"nats_url"` NatsURL string `json:"nats_url"`
NatsUser string `json:"nats_user"` NatsUser string `json:"nats_user"`
NatsPassword string `json:"nats_password"` NatsPassword string `json:"nats_password"`
ListenAddr string `json:"listen_addr"` ListenAddr string `json:"listen_addr"`
JWTSecret string `json:"jwt_secret"` JWTSecret string `json:"jwt_secret"`
JWTSecretEnv string `json:"jwt_secret_env"` // имя переменной окружения с JWT-секретом MonitoringAPIKey string `json:"monitoring_api_key"`
MonitoringAPIKey string `json:"monitoring_api_key"` StaleThresholdSec int `json:"stale_threshold_sec"`
StaleThresholdSec int `json:"stale_threshold_sec"` Scoring ScoringConfig `json:"scoring"`
Scoring ScoringConfig `json:"scoring"` TLS TLSConfig `json:"tls"`
TLS TLSConfig `json:"tls"` RateLimit RateLimitConfig `json:"rate_limit"`
RateLimit RateLimitConfig `json:"rate_limit"` Log LogConfig `json:"log"`
Log LogConfig `json:"log"` ESL ESLConfig `json:"esl"`
ESL ESLConfig `json:"esl"`
} }
// Validate проверяет корректность всех полей конфига. // Validate проверяет корректность всех полей конфига.
@ -179,8 +168,8 @@ func (c *Config) Validate() error {
// Manager управляет чтением и атомарной записью JSON-конфигов. // Manager управляет чтением и атомарной записью JSON-конфигов.
type Manager struct { type Manager struct {
mu sync.RWMutex mu sync.RWMutex
dir string // путь к data/ dir string // путь к data/
} }
// NewManager создаёт новый менеджер конфигов из директории dataDir. // NewManager создаёт новый менеджер конфигов из директории dataDir.
@ -313,9 +302,9 @@ func (m *Manager) saveUsersLocked(users []models.User) error {
// --- Приватные методы --- // --- Приватные методы ---
func (m *Manager) configPath() string { return filepath.Join(m.dir, "config.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) trunksPath() string { return filepath.Join(m.dir, "trunks.json") }
func (m *Manager) usersPath() string { return filepath.Join(m.dir, "users.json") } func (m *Manager) usersPath() string { return filepath.Join(m.dir, "users.json") }
// saveFile атомарно сохраняет данные в файл: tmp → rename. // saveFile атомарно сохраняет данные в файл: tmp → rename.
// Вызывающий держит write lock. // Вызывающий держит write lock.
@ -366,7 +355,7 @@ func (m *Manager) createDefaultConfig() error {
MonitoringAPIKey: apiKey, MonitoringAPIKey: apiKey,
StaleThresholdSec: 20, StaleThresholdSec: 20,
Scoring: ScoringConfig{ Scoring: ScoringConfig{
IdleCPUMin: 5.0, IdleCPUMin: 5.0,
CallFailureRateLethal: 15.0, CallFailureRateLethal: 15.0,
Weights: ScoringWeights{ Weights: ScoringWeights{
CallScore: 0.40, CallScore: 0.40,

12
internal/engine/engine.go
View File

@ -29,7 +29,6 @@ type Engine struct {
// Статистика для health endpoint // Статистика для health endpoint
startTime time.Time startTime time.Time
routeRequests atomic.Int64 routeRequests atomic.Int64
routeFallbacks atomic.Int64
lastMetricTime atomic.Int64 // unix ts последней полученной метрики lastMetricTime atomic.Int64 // unix ts последней полученной метрики
} }
@ -189,17 +188,11 @@ func (e *Engine) IncrementRouteRequests() {
e.routeRequests.Add(1) e.routeRequests.Add(1)
} }
// IncrementRouteFallbacks увеличивает счётчик fallback-запросов.
func (e *Engine) IncrementRouteFallbacks() {
e.routeFallbacks.Add(1)
}
// HealthStats возвращает статистику для health endpoint. // HealthStats возвращает статистику для health endpoint.
type HealthStats struct { type HealthStats struct {
TotalNodes int `json:"total_nodes"` TotalNodes int `json:"total_nodes"`
HealthyNodes int `json:"healthy_nodes"` HealthyNodes int `json:"healthy_nodes"`
RouteRequests int64 `json:"route_requests_total"` RouteRequests int64 `json:"route_requests_total"`
RouteFallbacks int64 `json:"route_fallbacks_total"`
UptimeSeconds int64 `json:"uptime_seconds"` UptimeSeconds int64 `json:"uptime_seconds"`
LastMetricTS int64 `json:"last_metric_ts"` LastMetricTS int64 `json:"last_metric_ts"`
} }
@ -219,7 +212,6 @@ func (e *Engine) GetHealthStats() HealthStats {
TotalNodes: len(e.nodes), TotalNodes: len(e.nodes),
HealthyNodes: healthy, HealthyNodes: healthy,
RouteRequests: e.routeRequests.Load(), RouteRequests: e.routeRequests.Load(),
RouteFallbacks: e.routeFallbacks.Load(),
UptimeSeconds: int64(time.Since(e.startTime).Seconds()), UptimeSeconds: int64(time.Since(e.startTime).Seconds()),
LastMetricTS: e.lastMetricTime.Load(), LastMetricTS: e.lastMetricTime.Load(),
} }

2
internal/engine/engine_test.go
View File

@ -12,7 +12,7 @@ func defaultCfg() *config.Config {
return &config.Config{ return &config.Config{
StaleThresholdSec: 20, StaleThresholdSec: 20,
Scoring: config.ScoringConfig{ Scoring: config.ScoringConfig{
IdleCPUMin: 5.0, IdleCPUMin: 5.0,
CallFailureRateLethal: 15.0, CallFailureRateLethal: 15.0,
Weights: config.ScoringWeights{ Weights: config.ScoringWeights{
CallScore: 0.40, CallScore: 0.40,

142
internal/esl/client.go
View File

@ -36,10 +36,10 @@ type Client struct {
port int port int
password string password string
conn net.Conn conn net.Conn
reader *bufio.Reader reader *bufio.Reader
mu sync.Mutex // защищает отправку синхронных команд (auth, subscribe) mu sync.Mutex // защищает отправку синхронных команд (auth, subscribe)
sendMu sync.Mutex // защищает запись в сокет (async gateway commands) sendMu sync.Mutex // защищает запись в сокет (async gateway commands)
connected atomic.Bool connected atomic.Bool
eventsCh chan *Event eventsCh chan *Event
@ -109,7 +109,7 @@ func (c *Client) tryConnect() error {
c.closeCh = make(chan struct{}) c.closeCh = make(chan struct{})
} }
addr := net.JoinHostPort(c.host, fmt.Sprintf("%d", c.port)) addr := fmt.Sprintf("%s:%d", c.host, c.port)
dialer := net.Dialer{Timeout: 5 * time.Second} dialer := net.Dialer{Timeout: 5 * time.Second}
conn, err := dialer.Dial("tcp", addr) conn, err := dialer.Dial("tcp", addr)
if err != nil { if err != nil {
@ -119,7 +119,7 @@ func (c *Client) tryConnect() error {
c.reader = bufio.NewReader(conn) c.reader = bufio.NewReader(conn)
// 1. Читаем auth/request // 1. Читаем auth/request
headers, _, err := c.readMessage() headers, _, err := c.readMessageLocked()
if err != nil { if err != nil {
conn.Close() conn.Close()
return fmt.Errorf("чтение auth/request: %w", err) return fmt.Errorf("чтение auth/request: %w", err)
@ -137,7 +137,7 @@ func (c *Client) tryConnect() error {
} }
// 3. Читаем auth response // 3. Читаем auth response
headers, _, err = c.readMessage() headers, _, err = c.readMessageLocked()
if err != nil { if err != nil {
conn.Close() conn.Close()
return fmt.Errorf("чтение auth response: %w", err) return fmt.Errorf("чтение auth response: %w", err)
@ -199,7 +199,7 @@ func (c *Client) Send(command string) (map[string]string, string, error) {
} }
// Читаем ответ // Читаем ответ
return c.readMessage() return c.readMessageLocked()
} }
// Subscribe подписывается на события ESL. // Subscribe подписывается на события ESL.
@ -263,7 +263,7 @@ func (c *Client) readEventsLoop() {
continue continue
} }
headers, body, err := c.readMessage() headers, body, err := c.readMessageUnlocked()
if err != nil { if err != nil {
if err == io.EOF || strings.Contains(err.Error(), "use of closed network connection") { if err == io.EOF || strings.Contains(err.Error(), "use of closed network connection") {
@ -330,9 +330,48 @@ func (c *Client) reconnect() {
// --- Приватные методы --- // --- Приватные методы ---
// readMessage читает одно ESL-сообщение: заголовки + тело. // readMessageUnlocked читает одно ESL-сообщение: заголовки + тело.
// Вызывающий должен гарантировать монопольный доступ к c.reader. // Должна вызываться только из readEventsLoop (одиночный читатель).
func (c *Client) readMessage() (map[string]string, string, error) { 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) {
headers := make(map[string]string) headers := make(map[string]string)
for { for {
@ -373,12 +412,12 @@ func (c *Client) readMessage() (map[string]string, string, error) {
// Stats возвращает статистику ESL-клиента для health endpoint. // Stats возвращает статистику ESL-клиента для health endpoint.
type Stats struct { type Stats struct {
Status string `json:"status"` Status string `json:"status"`
Host string `json:"host"` Host string `json:"host"`
Reconnects int64 `json:"reconnects"` Reconnects int64 `json:"reconnects"`
GatewayOps int64 `json:"gateway_ops_total"` GatewayOps int64 `json:"gateway_ops_total"`
EventsRecv int64 `json:"events_received"` EventsRecv int64 `json:"events_received"`
UptimeSec int64 `json:"uptime_seconds"` UptimeSec int64 `json:"uptime_seconds"`
} }
// GetStats возвращает текущую статистику. // GetStats возвращает текущую статистику.
@ -401,70 +440,3 @@ func (c *Client) GetStats() Stats {
func (c *Client) IncGatewayOps() { func (c *Client) IncGatewayOps() {
c.gatewayOps.Add(1) c.gatewayOps.Add(1)
} }
// --- FS stats ---
// FsStats — показатели FreeSWITCH балансировщика.
type FsStats struct {
ActiveCalls int
MaxSessions int
FsUptime string
}
// FetchFsStats запрашивает метрики FreeSWITCH через api команды.
func (c *Client) FetchFsStats() (FsStats, error) {
if !c.connected.Load() {
return FsStats{}, fmt.Errorf("esl не подключён")
}
var stats FsStats
_, body, err := c.Send("api show calls count")
if err != nil {
return FsStats{}, err
}
stats.ActiveCalls = parseFirstInt(body)
_, body, err = c.Send("api status")
if err != nil {
return stats, err
}
stats.MaxSessions = parseMaxSessions(body)
stats.FsUptime = parseUptime(body)
return stats, nil
}
func parseFirstInt(s string) int {
s = strings.TrimSpace(s)
for i, c := range s {
if c < '0' || c > '9' {
if i == 0 {
return 0
}
v, err := strconv.Atoi(s[:i])
if err != nil {
return 0
}
return v
}
}
return 0
}
func parseMaxSessions(body string) int {
for _, line := range strings.Split(body, "\n") {
if strings.Contains(line, "session(s) max") {
return parseFirstInt(line)
}
}
return 0
}
func parseUptime(body string) string {
for _, line := range strings.Split(body, "\n") {
if strings.HasPrefix(line, "UP ") {
return strings.TrimPrefix(line, "UP ")
}
}
return ""
}

1
internal/esl/events.go
View File

@ -16,7 +16,6 @@ type EventHandler func(eventName string, headers map[string]string, body string)
// - Вызывает onConnect (например, для GatewaySyncAll) // - Вызывает onConnect (например, для GatewaySyncAll)
// - Подписывается на события SOFIA::gateway_register/unregister // - Подписывается на события SOFIA::gateway_register/unregister
// - Запускает чтение асинхронных событий // - Запускает чтение асинхронных событий
//
// При разрыве соединения — автоматический реконнект. // При разрыве соединения — автоматический реконнект.
func (c *Client) StartEventLoop(ctx context.Context, profile string, onConnect func(), onEvent EventHandler) { func (c *Client) StartEventLoop(ctx context.Context, profile string, onConnect func(), onEvent EventHandler) {
const subscribeEvents = "SOFIA::gateway_register SOFIA::gateway_unregister SOFIA::gateway_expire" const subscribeEvents = "SOFIA::gateway_register SOFIA::gateway_unregister SOFIA::gateway_expire"

8
internal/esl/gateway.go
View File

@ -32,8 +32,8 @@ func (c *Client) GatewayDelete(profile, name string) error {
return nil return nil
} }
// GatewayList возвращает список имён gateway в заданном профиле с указанным префиксом. // GatewayList возвращает список имён gateway в заданном профиле.
func (c *Client) GatewayList(profile, prefix string) ([]string, error) { func (c *Client) GatewayList(profile string) ([]string, error) {
cmd := fmt.Sprintf("api sofia profile %s gwlist", profile) cmd := fmt.Sprintf("api sofia profile %s gwlist", profile)
_, body, err := c.Send(cmd) _, body, err := c.Send(cmd)
@ -49,8 +49,8 @@ func (c *Client) GatewayList(profile, prefix string) ([]string, error) {
if line == "" || strings.HasPrefix(line, "Name:") || strings.HasPrefix(line, "====") { if line == "" || strings.HasPrefix(line, "Name:") || strings.HasPrefix(line, "====") {
continue continue
} }
// Ищем строки вида "<prefix>-ingress-xxxxx sip:..." // Ищем строки вида "pulse-ingress-xxxxx sip:..."
if strings.Contains(line, prefix+"-ingress-") { if strings.Contains(line, "pulse-") {
parts := strings.Fields(line) parts := strings.Fields(line)
if len(parts) > 0 { if len(parts) > 0 {
gateways = append(gateways, parts[0]) gateways = append(gateways, parts[0])

8
internal/log/logger.go
View File

@ -15,10 +15,10 @@ import (
// Logger — потокобезопасный логгер метрик в ASCII-файл с дневной ротацией. // Logger — потокобезопасный логгер метрик в ASCII-файл с дневной ротацией.
type Logger struct { type Logger struct {
mu sync.Mutex mu sync.Mutex
dir string dir string
file *os.File file *os.File
today string // YYYY-MM-DD последнего открытого файла today string // YYYY-MM-DD последнего открытого файла
} }
// NewLogger создаёт логгер, записывающий в dataDir/. // NewLogger создаёт логгер, записывающий в dataDir/.

117
internal/models/types.go
View File

@ -7,15 +7,15 @@ import "time"
// NodeMetric — сообщение, публикуемое нодой в NATS каждые 5 сек. // NodeMetric — сообщение, публикуемое нодой в NATS каждые 5 сек.
type NodeMetric struct { type NodeMetric struct {
NodeID string `json:"node_id"` NodeID string `json:"node_id"`
TS int64 `json:"ts"` // unix timestamp TS int64 `json:"ts"` // unix timestamp
Status string `json:"status"` // "ok", "degraded", "down" Status string `json:"status"` // "ok", "degraded", "down"
ActiveCalls int `json:"active_calls"` ActiveCalls int `json:"active_calls"`
MaxCalls int `json:"max_calls"` MaxCalls int `json:"max_calls"`
IdleCPU float64 `json:"idle_cpu"` // 0..100 % IdleCPU float64 `json:"idle_cpu"` // 0..100 %
LoadAvg float64 `json:"load_avg"` // system load average (1m) LoadAvg float64 `json:"load_avg"` // system load average (1m)
CallFailureRate float64 `json:"call_failure_rate"` // 0.0..100.0 % CallFailureRate float64 `json:"call_failure_rate"` // 0.0..100.0 %
SIPGateway string `json:"sip_gateway,omitempty"` // SIP-адрес ноды для авто-создания транка SIPGateway string `json:"sip_gateway,omitempty"` // SIP-адрес ноды для авто-создания транка
} }
// --- In-memory состояние ноды в engine --- // --- In-memory состояние ноды в engine ---
@ -24,8 +24,8 @@ type NodeMetric struct {
// Содержит данные последней метрики + поля ручного управления. // Содержит данные последней метрики + поля ручного управления.
type NodeState struct { type NodeState struct {
NodeMetric NodeMetric
Disabled bool `json:"disabled"` Disabled bool `json:"disabled"`
DisabledReason string `json:"disabled_reason"` DisabledReason string `json:"disabled_reason"`
Score float64 `json:"score"` Score float64 `json:"score"`
LethalReason string `json:"lethal_reason,omitempty"` // причина, если score = -100 LethalReason string `json:"lethal_reason,omitempty"` // причина, если score = -100
} }
@ -46,7 +46,7 @@ type Trunk struct {
ID string `json:"id"` ID string `json:"id"`
Name string `json:"name"` Name string `json:"name"`
Type TrunkType `json:"type"` Type TrunkType `json:"type"`
NodeID string `json:"node_id,omitempty"` // только для balance NodeID string `json:"node_id,omitempty"` // только для balance
Gateway string `json:"gateway"` Gateway string `json:"gateway"`
Codecs []string `json:"codecs,omitempty"` Codecs []string `json:"codecs,omitempty"`
Context string `json:"context,omitempty"` Context string `json:"context,omitempty"`
@ -68,10 +68,10 @@ const (
// User — модель пользователя (из users.json). // User — модель пользователя (из users.json).
type User struct { type User struct {
ID string `json:"id"` ID string `json:"id"`
Username string `json:"username"` Username string `json:"username"`
PasswordHash string `json:"password_hash"` PasswordHash string `json:"password_hash"`
Role UserRole `json:"role"` Role UserRole `json:"role"`
CreatedAt time.Time `json:"created_at"` CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"` UpdatedAt time.Time `json:"updated_at"`
} }
@ -96,27 +96,27 @@ type LoginResponse struct {
// RouteResponse — ответ на запрос маршрутизации. // RouteResponse — ответ на запрос маршрутизации.
type RouteResponse struct { type RouteResponse struct {
// Успех // Успех
NodeID string `json:"node_id,omitempty"` NodeID string `json:"node_id,omitempty"`
Score float64 `json:"score,omitempty"` Score float64 `json:"score,omitempty"`
SIPGateway string `json:"sip_gateway,omitempty"` SIPGateway string `json:"sip_gateway,omitempty"`
// Fallback // Fallback
Fallback bool `json:"fallback,omitempty"` Fallback bool `json:"fallback,omitempty"`
Reason string `json:"reason,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 — информация о ноде в ответе маршрутизации. // RouteNodeInfo — информация о ноде в ответе маршрутизации.
type RouteNodeInfo struct { type RouteNodeInfo struct {
ID string `json:"id"` ID string `json:"id"`
Score float64 `json:"score"` Score float64 `json:"score"`
Reason string `json:"reason,omitempty"` Reason string `json:"reason,omitempty"`
Status string `json:"status"` Status string `json:"status"`
} }
// --- API: Nodes --- // --- API: Nodes ---
@ -129,20 +129,20 @@ type ToggleNodeRequest struct {
// NodeInfo — информация о ноде для фронта. // NodeInfo — информация о ноде для фронта.
type NodeInfo struct { type NodeInfo struct {
NodeID string `json:"node_id"` NodeID string `json:"node_id"`
TS int64 `json:"ts"` TS int64 `json:"ts"`
Status string `json:"status"` Status string `json:"status"`
ActiveCalls int `json:"active_calls"` ActiveCalls int `json:"active_calls"`
MaxCalls int `json:"max_calls"` MaxCalls int `json:"max_calls"`
IdleCPU float64 `json:"idle_cpu"` IdleCPU float64 `json:"idle_cpu"`
LoadAvg float64 `json:"load_avg"` LoadAvg float64 `json:"load_avg"`
CallFailureRate float64 `json:"call_failure_rate"` CallFailureRate float64 `json:"call_failure_rate"`
Disabled bool `json:"disabled"` Disabled bool `json:"disabled"`
DisabledReason string `json:"disabled_reason,omitempty"` DisabledReason string `json:"disabled_reason,omitempty"`
Score float64 `json:"score"` Score float64 `json:"score"`
LethalReason string `json:"lethal_reason,omitempty"` LethalReason string `json:"lethal_reason,omitempty"`
IsStale bool `json:"is_stale"` IsStale bool `json:"is_stale"`
SecondsAgo float64 `json:"seconds_ago"` SecondsAgo float64 `json:"seconds_ago"`
} }
// --- API: Users --- // --- API: Users ---
@ -191,17 +191,17 @@ type UpdateTrunkRequest struct {
// ZabbixNode — элемент LLD (Low-Level Discovery) для Zabbix. // ZabbixNode — элемент LLD (Low-Level Discovery) для Zabbix.
type ZabbixNode struct { type ZabbixNode struct {
NodeID string `json:"{#NODE_ID}"` NodeID string `json:"{#NODE_ID}"`
Status string `json:"{#STATUS}"` Status string `json:"{#STATUS}"`
ActiveCalls int `json:"{#ACTIVE_CALLS}"` ActiveCalls int `json:"{#ACTIVE_CALLS}"`
MaxCalls int `json:"{#MAX_CALLS}"` MaxCalls int `json:"{#MAX_CALLS}"`
IdleCPU float64 `json:"{#IDLE_CPU}"` IdleCPU float64 `json:"{#IDLE_CPU}"`
LoadAvg float64 `json:"{#LOAD_AVG}"` LoadAvg float64 `json:"{#LOAD_AVG}"`
CallFailureRate float64 `json:"{#CALL_FAILURE_RATE}"` CallFailureRate float64 `json:"{#CALL_FAILURE_RATE}"`
Score float64 `json:"{#SCORE}"` Score float64 `json:"{#SCORE}"`
Disabled bool `json:"{#DISABLED}"` Disabled bool `json:"{#DISABLED}"`
IsStale bool `json:"{#IS_STALE}"` IsStale bool `json:"{#IS_STALE}"`
SecondsAgo float64 `json:"{#SECONDS_AGO}"` SecondsAgo float64 `json:"{#SECONDS_AGO}"`
} }
// ZabbixResponse — ответ на запрос Zabbix LLD. // ZabbixResponse — ответ на запрос Zabbix LLD.
@ -209,29 +209,10 @@ type ZabbixResponse struct {
Data []ZabbixNode `json:"data"` Data []ZabbixNode `json:"data"`
} }
// --- Balancer ---
// BalancerInfo — состояние балансировщика для дашборда.
type BalancerInfo struct {
EslConnected bool `json:"esl_connected"`
EslUptimeSec int64 `json:"esl_uptime_sec,omitempty"`
EslReconnects int64 `json:"esl_reconnects,omitempty"`
NatsConnected bool `json:"nats_connected"`
GatewaysTotal int `json:"gateways_total"`
GatewaysUp int `json:"gateways_up"`
GatewaysDown int `json:"gateways_down"`
FsActiveCalls int `json:"fs_active_calls"`
FsMaxSessions int `json:"fs_max_sessions"`
FsUptime string `json:"fs_uptime"`
RouteTotal int64 `json:"route_total"`
RouteFallbacks int64 `json:"route_fallbacks"`
UptimeSec int64 `json:"uptime_sec"`
}
// --- WebSocket --- // --- WebSocket ---
// WsMetricsMessage — сообщение, отправляемое по WebSocket. // WsMetricsMessage — сообщение, отправляемое по WebSocket.
type WsMetricsMessage struct { type WsMetricsMessage struct {
Type string `json:"type"` // "nodes_update", "balancer_update", "gateway_update", etc Type string `json:"type"` // "nodes_update", "node_toggle", etc
Payload interface{} `json:"payload"` Payload interface{} `json:"payload"`
} }

2
internal/nats/subscriber.go
View File

@ -9,9 +9,9 @@ import (
natsgo "github.com/nats-io/nats.go" natsgo "github.com/nats-io/nats.go"
"github.com/pulse-lets-go/internal/models"
"github.com/pulse-lets-go/internal/engine" "github.com/pulse-lets-go/internal/engine"
filelog "github.com/pulse-lets-go/internal/log" filelog "github.com/pulse-lets-go/internal/log"
"github.com/pulse-lets-go/internal/models"
) )
const subject = "pulse.metrics.>" const subject = "pulse.metrics.>"

18
internal/util/util.go
View File

@ -1,18 +0,0 @@
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]
}

16
web/src/lib/api.ts
View File

@ -184,22 +184,6 @@ export async function deleteUser(id: string) {
} }
// --- WebSocket --- // --- WebSocket ---
export interface BalancerInfo {
esl_connected: boolean;
esl_uptime_sec: number;
esl_reconnects: number;
nats_connected: boolean;
gateways_total: number;
gateways_up: number;
gateways_down: number;
fs_active_calls: number;
fs_max_sessions: number;
fs_uptime: string;
route_total: number;
route_fallbacks: number;
uptime_sec: number;
}
export function connectWebSocket(onMessage: (data: any) => void): WebSocket { export function connectWebSocket(onMessage: (data: any) => void): WebSocket {
const token = getToken(); const token = getToken();
const proto = window.location.protocol === 'https:' ? 'wss' : 'ws'; const proto = window.location.protocol === 'https:' ? 'wss' : 'ws';

39
web/src/routes/+layout.svelte
View File

@ -1,40 +1,31 @@
<script lang="ts"> <script lang="ts">
import { browser } from '$app/environment';
import { page } from '$app/stores'; import { page } from '$app/stores';
import { goto } from '$app/navigation';
import '../app.css'; import '../app.css';
import { Toaster } from 'svelte-sonner'; import { Toaster } from 'svelte-sonner';
import { LayoutDashboard, Phone, Cable, Users } from 'lucide-svelte'; import { LogIn, LayoutDashboard, Phone, Cable, Users } from 'lucide-svelte';
let token: string | null = null; let token: string | null = browser ? localStorage.getItem('token') : null;
let user: { username: string; role: string } | null = null; let user: { username: string; role: string } | null = token
? (() => { try { return JSON.parse(atob(token.split('.')[1])); } catch { return null; } })()
function updateAuth() { : null;
token = typeof localStorage !== 'undefined' ? localStorage.getItem('token') : null;
if (token) {
try { user = JSON.parse(atob(token.split('.')[1])); } catch { user = null; }
} else {
user = null;
}
}
function logout() { function logout() {
localStorage.removeItem('token'); if (browser) {
localStorage.removeItem('refreshToken'); localStorage.removeItem('token');
updateAuth(); localStorage.removeItem('refreshToken');
goto('/login'); }
token = null;
user = null;
} }
let pathname = $derived($page.url.pathname); $: pathname = browser ? window.location.pathname : '';
$: isLogin = pathname === '/login';
$effect(() => {
$page.url.pathname;
updateAuth();
});
</script> </script>
<Toaster position="top-right" /> <Toaster position="top-right" />
{#if pathname !== '/login'} {#if !isLogin}
<div class="flex min-h-screen"> <div class="flex min-h-screen">
<!-- Sidebar --> <!-- Sidebar -->
<aside class="w-64 border-r bg-card flex flex-col"> <aside class="w-64 border-r bg-card flex flex-col">

73
web/src/routes/+page.svelte
View File

@ -4,12 +4,11 @@
import { onMount, onDestroy } from 'svelte'; import { onMount, onDestroy } from 'svelte';
import { import {
getNodes, connectWebSocket, isAuthenticated, getNodes, connectWebSocket, isAuthenticated,
type NodeInfo, type BalancerInfo type NodeInfo
} from '$lib/api'; } from '$lib/api';
import { Activity, Phone, Zap, AlertTriangle, Wifi, Server, Plug, Globe, ArrowLeftRight, Network } from 'lucide-svelte'; import { Activity, Phone, Zap, AlertTriangle, Wifi, Server } from 'lucide-svelte';
let nodes: NodeInfo[] = []; let nodes: NodeInfo[] = [];
let balancer: BalancerInfo | null = null;
let ws: WebSocket | null = null; let ws: WebSocket | null = null;
let error: string | null = null; let error: string | null = null;
@ -22,8 +21,6 @@
ws = connectWebSocket((msg) => { ws = connectWebSocket((msg) => {
if (msg.type === 'nodes_update') { if (msg.type === 'nodes_update') {
nodes = msg.payload; nodes = msg.payload;
} else if (msg.type === 'balancer_update') {
balancer = msg.payload;
} }
}); });
}); });
@ -70,31 +67,10 @@
} }
} }
function dotColor(ok: boolean, degrade?: boolean): string {
if (ok) return 'bg-green-500';
if (degrade) return 'bg-yellow-500';
return 'bg-red-500';
}
function fmtUptime(sec: number): string {
if (sec < 60) return `${sec}с`;
if (sec < 3600) return `${Math.floor(sec / 60)}м`;
if (sec < 86400) return `${Math.floor(sec / 3600)}ч`;
return `${Math.floor(sec / 86400)}д`;
}
function fmtNum(n: number): string {
if (n >= 1000) return (n / 1000).toFixed(1) + 'K';
return String(n);
}
$: healthyCount = nodes.filter(n => n.score >= 0 && !n.disabled && !n.is_stale).length; $: healthyCount = nodes.filter(n => n.score >= 0 && !n.disabled && !n.is_stale).length;
$: totalCalls = nodes.reduce((s, n) => s + n.active_calls, 0); $: totalCalls = nodes.reduce((s, n) => s + n.active_calls, 0);
$: avgScore = nodes.length > 0 ? nodes.reduce((s, n) => s + n.score, 0) / nodes.length : 0; $: avgScore = nodes.length > 0 ? nodes.reduce((s, n) => s + n.score, 0) / nodes.length : 0;
$: unhealthyList = nodes.filter(n => n.score < 0 || n.disabled || n.is_stale); $: unhealthyList = nodes.filter(n => n.score < 0 || n.disabled || n.is_stale);
$: fallbackPct = balancer && balancer.route_total > 0
? (balancer.route_fallbacks / balancer.route_total * 100).toFixed(1)
: '0';
</script> </script>
{#if error} {#if error}
@ -103,51 +79,6 @@
</div> </div>
{/if} {/if}
<!-- Balancer Status Bar -->
{#if balancer}
<div class="flex flex-wrap items-center gap-1 text-xs text-muted-foreground mb-4 px-3 py-2 border rounded-md bg-muted/30">
<span class="flex items-center gap-1 mr-2">
<span class="w-1.5 h-1.5 rounded-full {dotColor(balancer.esl_connected)}"></span>
<Plug class="h-3 w-3" />
ESL {balancer.esl_connected ? 'Online' : 'Offline'}
{#if balancer.esl_uptime_sec > 0}{fmtUptime(balancer.esl_uptime_sec)}{/if}
</span>
<span class="opacity-30">|</span>
<span class="flex items-center gap-1 mr-2">
<span class="w-1.5 h-1.5 rounded-full {dotColor(balancer.nats_connected)}"></span>
<Network class="h-3 w-3" />
NATS {balancer.nats_connected ? 'Online' : 'Offline'}
</span>
<span class="opacity-30">|</span>
<span class="flex items-center gap-1 mr-2">
<span class="w-1.5 h-1.5 rounded-full {balancer.gateways_down > 0 ? 'bg-red-500' : 'bg-green-500'}"></span>
<Globe class="h-3 w-3" />
GW {balancer.gateways_up}/{balancer.gateways_total}
{#if balancer.gateways_down > 0}
<span class="text-red-500">({balancer.gateways_down} down)</span>
{/if}
</span>
<span class="opacity-30">|</span>
<span class="flex items-center gap-1">
<ArrowLeftRight class="h-3 w-3" />
R {fmtNum(balancer.route_total)}
{#if balancer.route_fallbacks > 0}
<span class="text-orange-500">({fallbackPct}% fb)</span>
{/if}
</span>
{#if balancer.esl_connected && balancer.fs_max_sessions > 0}
<span class="opacity-30">|</span>
<span class="flex items-center gap-1">
<Phone class="h-3 w-3" />
FS {balancer.fs_active_calls}·{balancer.fs_max_sessions}
{#if balancer.fs_uptime}
<span class="opacity-50">{balancer.fs_uptime}</span>
{/if}
</span>
{/if}
</div>
{/if}
<!-- Summary Cards --> <!-- Summary Cards -->
<div class="grid grid-cols-1 sm:grid-cols-2 lg:grid-cols-4 gap-4 mb-6"> <div class="grid grid-cols-1 sm:grid-cols-2 lg:grid-cols-4 gap-4 mb-6">
<div class="border rounded-lg p-4 bg-card"> <div class="border rounded-lg p-4 bg-card">