Compare commits

..

5 Commits

Author SHA1 Message Date
Maksim Totmin
ff037f9c1e feat: balancer dashboard metrics + fix WebSocket Hijacker 2026-06-25 19:59:26 +07:00
Maksim Totmin
273e316af9 refactor: deploy model — all files under /opt/pulse-lets-go
Single prefix deployment: /opt/pulse-lets-go/ contains
bin/, data/, web/, log/ — no more split across
/usr/local/bin, /etc, /usr/local/share, /var/log.

* Makefile: install → deploy target with PREFIX=/opt/pulse-lets-go
* deploy/pulse-lets-go.service: all paths under /opt/pulse-lets-go,
  removed stale nats.service dependency
* main.go: default dataDir = parent of exeDir (PREFIX/bin → PREFIX/data)
* README: updated all paths, step 1 uses make deploy
2026-06-25 19:35:54 +07:00
Maksim Totmin
0d8a56e73f fix: build system — findWebDir order, auto-copy web static to bin
* findWebDir: CWD and project root checked before binary-adjacent path,
  so dev always serves fresh web/build/ over stale bin/web/build/
* build-go: auto-copies web/build/ to bin/web/ before Go compile
* build-prod: same auto-copy for production builds
* Proper refresh of frontend static on every make build
2026-06-25 19:30:49 +07:00
Maksim Totmin
78e53f42bf fix: SvelteKit UI reactivity with Svelte 5 runes
* / replaces legacy $: syntax
* Active menu highlight via .url.pathname (/stores)
* Logout navigates to /login via goto()
* Token/user footer updates reactively via
2026-06-25 19:30:42 +07:00
Maksim Totmin
cc9da3ad7d refactor: common util package, ESL/AMI/security fixes, Prometheus metrics
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
2026-06-25 19:30:36 +07:00
34 changed files with 617 additions and 317 deletions

View File

@ -4,9 +4,9 @@
APP := pulse-lets-go APP := pulse-lets-go
BINDIR := bin BINDIR := bin
WEB_DIR := web WEB_DIR := web
DATA_DIR := data PREFIX := /opt/pulse-lets-go
.PHONY: all build build-go build-web clean run test install \ .PHONY: all build build-go build-web clean run test deploy \
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,13 +22,17 @@ 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)
@ -43,23 +47,26 @@ 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)
# Установка: копирует бинарник и статику в /usr/local # Деплой в PREFIX: бинарник + статика + data + log
install: build deploy: build-prod
@echo "→ Установка в /usr/local" @echo "→ Деплой в $(PREFIX)"
mkdir -p /usr/local/share/$(APP)/web install -d $(PREFIX)/bin $(PREFIX)/data $(PREFIX)/web $(PREFIX)/log
cp -r $(WEB_DIR)/build /usr/local/share/$(APP)/web/ install -m 755 $(BINDIR)/$(APP) $(PREFIX)/bin/
cp $(BINDIR)/$(APP) /usr/local/bin/$(APP) cp -r $(WEB_DIR)/build/. $(PREFIX)/web/
chmod +x /usr/local/bin/$(APP) @echo "✓ Готово: $(PREFIX)"
@echo "✓ Установлен в /usr/local/bin/$(APP)" @echo " $(PREFIX)/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:
@echo "→ Установка systemd unit" install -m 644 deploy/$(APP).service /etc/systemd/system/
cp deploy/$(APP).service /etc/systemd/system/
systemctl daemon-reload systemctl daemon-reload
@echo "✓ systemd unit установлен. Запуск: systemctl start $(APP)" @echo "✓ systemd unit установлен. Запуск: sudo systemctl start $(APP)"
# ============================================================ # ============================================================
# Эмулятор метрик # Эмулятор метрик

View File

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

View File

@ -32,12 +32,15 @@ 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 <-time.After(100 * time.Millisecond): case <-timer.C:
// Выходим из цикла после ConnectWithRetry
return nil return nil
} }
} }

View File

@ -5,8 +5,6 @@ package main
import ( import (
"context" "context"
"crypto/rand"
"encoding/hex"
"fmt" "fmt"
"log" "log"
"net/http" "net/http"
@ -24,15 +22,17 @@ 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(exeDir, "data") dataDir := filepath.Join(filepath.Dir(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-" + randomHex(6) trunkID := "trk-" + util.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.JWTSecret, cfg.MonitoringAPIKey, sub.IsConnected, cfg) apiHandler := api.NewAPI(eng, cfgMgr, cfg.GetJWTSecret(), cfg.MonitoringAPIKey, sub.IsConnected, cfg)
handler := apiHandler.Handler() handler := apiHandler.Handler()
// 6.1 ESL-клиент (опционально — если сконфигурирован) // 6.1 ESL-клиент (опционально — если сконфигурирован)
@ -177,6 +177,19 @@ 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 != "" {
@ -230,20 +243,19 @@ 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/).
// Пробует по порядку: переменная окружения, рядом с бинарником, в корне проекта, CWD. // Пробует по порядку: PULSE_WEB_DIR, CWD, ../web/build, рядом с бинарником.
func findWebDir(exeDir string) string { func findWebDir(exeDir string) string {
candidates := []string{ candidates := []string{
os.Getenv("PULSE_WEB_DIR"), os.Getenv("PULSE_WEB_DIR"),
filepath.Join(exeDir, "web", "build"), // bin/web/build "web/build", // ./web/build (CWD — dev, systemd working dir)
filepath.Join(filepath.Dir(exeDir), "web", "build"), // ../web/build (корень проекта) filepath.Join(filepath.Dir(exeDir), "web", "build"), // ../web/build (корень проекта, dev с bin/)
"web/build", // ./web/build (CWD) filepath.Join(exeDir, "web", "build"), // рядом с бинарником (production install, make deploy)
} }
for _, d := range candidates { for _, d := range candidates {
if d == "" { if d == "" {
@ -294,15 +306,6 @@ 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 {

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,6 +19,8 @@ import (
"sync" "sync"
"sync/atomic" "sync/atomic"
"time" "time"
"github.com/pulse-lets-go/internal/util"
) )
const ( const (
@ -103,7 +105,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-" + randomHex(4) to += ";tag=uas-" + util.RandomHex(4)
} }
return fmt.Sprintf("SIP/2.0 200 OK\r\n"+ 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 { 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") callID := fmt.Sprintf("call-%d-%s@%s", callNum, util.RandomHex(4), "127.0.0.1")
branch := "z9hG4bK-" + randomHex(8) branch := "z9hG4bK-" + util.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"+
@ -227,9 +229,3 @@ 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]
}

View File

@ -1,16 +1,14 @@
[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 nats.service After=network.target
Wants=nats.service
[Service] [Service]
Type=simple Type=simple
User=pulse User=pulse
Group=pulse Group=pulse
ExecStart=/usr/local/bin/pulse-lets-go 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
Restart=always Restart=always
RestartSec=5 RestartSec=5
@ -19,7 +17,7 @@ NoNewPrivileges=yes
PrivateTmp=yes PrivateTmp=yes
ProtectSystem=strict ProtectSystem=strict
ProtectHome=yes ProtectHome=yes
ReadWritePaths=/etc/pulse-lets-go /var/log/pulse-lets-go ReadWritePaths=/opt/pulse-lets-go/data /opt/pulse-lets-go/log
# Логирование # Логирование
StandardOutput=journal StandardOutput=journal

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 := 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} 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,11 +268,9 @@ 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] соединение разорвано, реконнект...")
if c.shouldReconn.Load() { c.reconnect()
go c.ConnectWithRetry() continue
}
return
} }
time.Sleep(100 * time.Millisecond) time.Sleep(100 * time.Millisecond)
continue 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] реконнект успешен")
}

View File

@ -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 := json.NewDecoder(r.Body).Decode(&req); err != nil { if err := decodeJSON(r, &req); err != nil {
writeError(w, http.StatusBadRequest, "некорректный JSON") writeError(w, http.StatusBadRequest, "некорректный JSON")
return return
} }
@ -199,10 +199,6 @@ 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

View File

@ -1,10 +1,13 @@
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"
) )
@ -82,3 +85,11 @@ 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()
}

View File

@ -38,6 +38,26 @@ 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)

View File

@ -14,6 +14,7 @@ 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",
}) })
@ -22,6 +23,7 @@ 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()

View File

@ -4,6 +4,7 @@ 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"
@ -23,6 +24,11 @@ type API struct {
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 с заданными зависимостями.
@ -35,6 +41,7 @@ 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)
@ -130,6 +137,10 @@ 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{
@ -141,9 +152,35 @@ func (a *API) BroadcastGatewayEvent(gatewayName, trunkID, status string) {
a.wsHub.broadcast(msg) a.wsHub.broadcast(msg)
} }
// decodeJSON декодирует тело запроса в структуру v. func (a *API) gatewayStats() (total, up, down int) {
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)
} }

View File

@ -268,8 +268,9 @@ 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) {
profile = a.readConfig().ESL.SofiaProfile cfg := a.readConfig()
name = fmt.Sprintf("pulse-ingress-%s", trunk.ID) profile = cfg.ESL.SofiaProfile
name = fmt.Sprintf("%s-ingress-%s", cfg.ESL.GatewayPrefix, trunk.ID)
proxy = extractHost(trunk.Gateway) proxy = extractHost(trunk.Gateway)
return return
} }

View File

@ -1,13 +1,12 @@
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"
) )
@ -227,15 +226,5 @@ 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, randomHex(6)) return fmt.Sprintf("%s-%s", prefix, util.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]
} }

View File

@ -104,12 +104,7 @@ func (a *API) handleWS(w http.ResponseWriter, r *http.Request) {
// Отправляем текущее состояние при подключении // Отправляем текущее состояние при подключении
go func() { go func() {
nodes := a.engine.GetAllNodes() a.BroadcastAll()
msg := &models.WsMetricsMessage{
Type: "nodes_update",
Payload: nodes,
}
a.wsHub.broadcast(msg)
}() }()
// Читаем из вебсокета (ping/pong + закрытие) // Читаем из вебсокета (ping/pong + закрытие)
@ -125,7 +120,7 @@ func (a *API) handleWS(w http.ResponseWriter, r *http.Request) {
}() }()
} }
// BroadcastMetrics рассылает метрики всем WS-клиентам (вызывается engine при обновлении). // BroadcastMetrics рассылает метрики нод всем WS-клиентам.
func (a *API) BroadcastMetrics() { func (a *API) BroadcastMetrics() {
nodes := a.engine.GetAllNodes() nodes := a.engine.GetAllNodes()
msg := &models.WsMetricsMessage{ msg := &models.WsMetricsMessage{
@ -135,6 +130,46 @@ 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)

View File

@ -72,6 +72,16 @@ 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"`
@ -79,6 +89,7 @@ type Config struct {
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"`

View File

@ -29,6 +29,7 @@ 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 последней полученной метрики
} }
@ -188,11 +189,17 @@ 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"`
} }
@ -212,6 +219,7 @@ 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(),
} }

View File

@ -109,7 +109,7 @@ func (c *Client) tryConnect() error {
c.closeCh = make(chan struct{}) 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} 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.readMessageLocked() headers, _, err := c.readMessage()
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.readMessageLocked() headers, _, err = c.readMessage()
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.readMessageLocked() return c.readMessage()
} }
// Subscribe подписывается на события ESL. // Subscribe подписывается на события ESL.
@ -263,7 +263,7 @@ func (c *Client) readEventsLoop() {
continue continue
} }
headers, body, err := c.readMessageUnlocked() headers, body, err := c.readMessage()
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,48 +330,9 @@ func (c *Client) reconnect() {
// --- Приватные методы --- // --- Приватные методы ---
// readMessageUnlocked читает одно ESL-сообщение: заголовки + тело. // readMessage читает одно ESL-сообщение: заголовки + тело.
// Должна вызываться только из readEventsLoop (одиночный читатель). // Вызывающий должен гарантировать монопольный доступ к c.reader.
func (c *Client) readMessageUnlocked() (map[string]string, string, error) { func (c *Client) readMessage() (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 {
@ -440,3 +401,70 @@ 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 ""
}

View File

@ -16,6 +16,7 @@ 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"

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 string) ([]string, error) { func (c *Client) GatewayList(profile, prefix 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 string) ([]string, error) {
if line == "" || strings.HasPrefix(line, "Name:") || strings.HasPrefix(line, "====") { if line == "" || strings.HasPrefix(line, "Name:") || strings.HasPrefix(line, "====") {
continue continue
} }
// Ищем строки вида "pulse-ingress-xxxxx sip:..." // Ищем строки вида "<prefix>-ingress-xxxxx sip:..."
if strings.Contains(line, "pulse-") { if strings.Contains(line, prefix+"-ingress-") {
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])

View File

@ -209,10 +209,29 @@ 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", "node_toggle", etc Type string `json:"type"` // "nodes_update", "balancer_update", "gateway_update", etc
Payload interface{} `json:"payload"` Payload interface{} `json:"payload"`
} }

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 Normal file
View File

@ -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]
}

View File

@ -184,6 +184,22 @@ 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';

View File

@ -1,31 +1,40 @@
<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 { LogIn, LayoutDashboard, Phone, Cable, Users } from 'lucide-svelte'; import { LayoutDashboard, Phone, Cable, Users } from 'lucide-svelte';
let token: string | null = browser ? localStorage.getItem('token') : null; let token: string | null = null;
let user: { username: string; role: string } | null = token let user: { username: string; role: string } | null = null;
? (() => { try { return JSON.parse(atob(token.split('.')[1])); } catch { return null; } })()
: null;
function logout() { function updateAuth() {
if (browser) { token = typeof localStorage !== 'undefined' ? localStorage.getItem('token') : null;
localStorage.removeItem('token'); if (token) {
localStorage.removeItem('refreshToken'); try { user = JSON.parse(atob(token.split('.')[1])); } catch { user = null; }
} } else {
token = null;
user = null; user = null;
} }
}
$: pathname = browser ? window.location.pathname : ''; function logout() {
$: isLogin = pathname === '/login'; localStorage.removeItem('token');
localStorage.removeItem('refreshToken');
updateAuth();
goto('/login');
}
let pathname = $derived($page.url.pathname);
$effect(() => {
$page.url.pathname;
updateAuth();
});
</script> </script>
<Toaster position="top-right" /> <Toaster position="top-right" />
{#if !isLogin} {#if pathname !== '/login'}
<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">

View File

@ -4,11 +4,12 @@
import { onMount, onDestroy } from 'svelte'; import { onMount, onDestroy } from 'svelte';
import { import {
getNodes, connectWebSocket, isAuthenticated, getNodes, connectWebSocket, isAuthenticated,
type NodeInfo type NodeInfo, type BalancerInfo
} from '$lib/api'; } from '$lib/api';
import { Activity, Phone, Zap, AlertTriangle, Wifi, Server } from 'lucide-svelte'; import { Activity, Phone, Zap, AlertTriangle, Wifi, Server, Plug, Globe, ArrowLeftRight, Network } 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;
@ -21,6 +22,8 @@
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;
} }
}); });
}); });
@ -67,10 +70,31 @@
} }
} }
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}
@ -79,6 +103,51 @@
</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">