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

37
Makefile
View File

@ -4,9 +4,9 @@
APP := pulse-lets-go
BINDIR := bin
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-node-down emulate-stale emulate-chaos build-emulator \
build-siptest siptest-uas siptest-uac siptest-full
@ -22,17 +22,13 @@ build-web:
@echo "→ Сборка SvelteKit..."
cd $(WEB_DIR) && npm run build
# Сборка Go бинарника (с авто-копированием статики)
# Сборка Go бинарника
build-go:
@mkdir -p $(BINDIR)
@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 (без фронта, без статики)
# Быстрая сборка только Go (без фронта)
build-go-only:
@mkdir -p $(BINDIR)
go build -o $(BINDIR)/$(APP) ./cmd/$(APP)
@ -47,26 +43,23 @@ run:
build-prod:
@mkdir -p $(BINDIR)
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)
# Деплой в PREFIX: бинарник + статика + data + log
deploy: build-prod
@echo "→ Деплой в $(PREFIX)"
install -d $(PREFIX)/bin $(PREFIX)/data $(PREFIX)/web $(PREFIX)/log
install -m 755 $(BINDIR)/$(APP) $(PREFIX)/bin/
cp -r $(WEB_DIR)/build/. $(PREFIX)/web/
@echo "✓ Готово: $(PREFIX)"
@echo " $(PREFIX)/bin/$(APP)"
@echo " $(PREFIX)/web/ (SvelteKit статика)"
@echo " $(PREFIX)/data/ (config.json, trunks.json, users.json)"
@echo " $(PREFIX)/log/ (ASCII-лог метрик)"
# Установка: копирует бинарник и статику в /usr/local
install: build
@echo "→ Установка в /usr/local"
mkdir -p /usr/local/share/$(APP)/web
cp -r $(WEB_DIR)/build /usr/local/share/$(APP)/web/
cp $(BINDIR)/$(APP) /usr/local/bin/$(APP)
chmod +x /usr/local/bin/$(APP)
@echo "✓ Установлен в /usr/local/bin/$(APP)"
# Установка systemd unit
install-systemd:
install -m 644 deploy/$(APP).service /etc/systemd/system/
@echo "→ Установка systemd unit"
cp deploy/$(APP).service /etc/systemd/system/
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
```bash
make deploy # полный деплой в /opt/pulse-lets-go
make install # копирует бинарник + web-статику
make install-systemd # устанавливает systemd unit
```
@ -436,15 +436,16 @@ After=network.target
[Service]
User=pulse
Group=pulse
ExecStart=/opt/pulse-lets-go/bin/pulse-lets-go
WorkingDirectory=/opt/pulse-lets-go
WorkingDirectory=/usr/local/share/pulse-lets-go
Environment=PULSE_DATA_DIR=/etc/pulse-lets-go
ExecStart=/usr/local/bin/pulse-lets-go
Restart=always
RestartSec=5
NoNewPrivileges=true
PrivateTmp=true
ProtectSystem=strict
ProtectHome=yes
ReadWritePaths=/opt/pulse-lets-go/data /opt/pulse-lets-go/log
ReadWritePaths=/etc/pulse-lets-go /var/log/pulse-lets-go
[Install]
WantedBy=multi-user.target
@ -453,15 +454,14 @@ WantedBy=multi-user.target
### Структура директорий
```
/opt/pulse-lets-go/
├── bin/pulse-lets-go # Бинарник
├── data/
│ ├── config.json # Конфигурация
│ ├── trunks.json # Транки
│ └── users.json # Пользователи
├── web/ # SvelteKit статика
└── log/
└── metrics.YYYY-MM-DD.log # ASCII-лог метрик
/usr/local/share/pulse-lets-go/
├── web/build/ # SvelteKit статика
/etc/pulse-lets-go/
├── config.json # Конфигурация
├── trunks.json # Транки
├── users.json # Пользователи
/var/log/pulse-lets-go/
└── metrics.YYYY-MM-DD.log # ASCII-лог метрик
```
### Security
@ -497,13 +497,13 @@ fs_cli -x "load mod_curl"
# <load module="mod_curl"/>
```
### Порядок деплоя (8 шагов)
### Порядок деплоя (7 шагов)
**Шаг 1 — Деплой на сервер**
**Шаг 1 — Скопировать бинарник и конфиги**
```bash
make deploy
scp -r /opt/pulse-lets-go devadmin@10.101.60.81:/opt/
scp contrib/route.lua devadmin@10.101.60.81:/tmp/
make build-prod
scp bin/pulse-lets-go devadmin@10.101.60.81:/usr/local/bin/
scp -r contrib/route.lua devadmin@10.101.60.81:/tmp/
```
**Шаг 2 — NATS сервер** (если не установлен)
@ -562,10 +562,11 @@ curl -X POST .../api/trunks
**Шаг 7 — Запуск pulse-lets-go**
```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": {"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 — Проверка**

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

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

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

@ -5,6 +5,8 @@ package main
import (
"context"
"crypto/rand"
"encoding/hex"
"fmt"
"log"
"net/http"
@ -22,17 +24,15 @@ import (
filelog "github.com/pulse-lets-go/internal/log"
"github.com/pulse-lets-go/internal/models"
"github.com/pulse-lets-go/internal/nats"
"github.com/pulse-lets-go/internal/util"
)
func main() {
// Определяем рабочую директорию
// По умолчанию data/ — родительская папка от бинарника.
// Пример: /opt/pulse-lets-go/bin/pulse-lets-go → /opt/pulse-lets-go/data
exe, _ := os.Executable()
exeDir := filepath.Dir(exe)
dataDir := filepath.Join(filepath.Dir(exeDir), "data")
dataDir := filepath.Join(exeDir, "data")
// Разрешаем переопределение через переменную окружения
if envDir := os.Getenv("PULSE_DATA_DIR"); envDir != "" {
dataDir = envDir
}
@ -91,7 +91,7 @@ func main() {
}
// Создаём новый balance-транк
now := time.Now().UTC()
trunkID := "trk-" + util.RandomHex(6)
trunkID := "trk-" + randomHex(6)
trunk := models.Trunk{
ID: trunkID,
Name: fmt.Sprintf("%s баланс", nodeID),
@ -113,7 +113,7 @@ func main() {
})
// 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()
// 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 статики, если директория существует
webDir := findWebDir(exeDir)
if webDir != "" {
@ -243,19 +230,20 @@ func main() {
log.Printf("[main] ошибка остановки HTTP сервера: %v", err)
}
sub.Close()
metricsLogger.Close()
log.Println("[main] pulse-lets-go остановлен")
}
// findWebDir ищет директорию со SvelteKit сборкой (web/build/).
// Пробует по порядку: PULSE_WEB_DIR, CWD, ../web/build, рядом с бинарником.
// Пробует по порядку: переменная окружения, рядом с бинарником, в корне проекта, CWD.
func findWebDir(exeDir string) string {
candidates := []string{
os.Getenv("PULSE_WEB_DIR"),
"web/build", // ./web/build (CWD — dev, systemd working dir)
filepath.Join(filepath.Dir(exeDir), "web", "build"), // ../web/build (корень проекта, dev с bin/)
filepath.Join(exeDir, "web", "build"), // рядом с бинарником (production install, make deploy)
filepath.Join(exeDir, "web", "build"), // bin/web/build
filepath.Join(filepath.Dir(exeDir), "web", "build"), // ../web/build (корень проекта)
"web/build", // ./web/build (CWD)
}
for _, d := range candidates {
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.
// "sip:mts-gw.lan:5060" → "mts-gw.lan"
func extractHostFromGateway(gateway string) string {

18
cmd/siptest/main.go
View File

@ -1,16 +1,16 @@
// SIP тестер для e2e проверки pulse-lets-go + FreeSWITCH.
// Два режима:
//
// uas — отвечает 200 OK на SIP INVITE, симулирует PBX-ноду
// uac — шлёт SIP INVITE в FreeSWITCH, симулирует оператора
//
// Примеры:
//
// ./siptest -mode uas (PBX-нода на порту 5090)
// ./siptest -mode uac -r 10 -l 100 (оператор, 10 CPS, до 100 одновременных)
package main
import (
"crypto/rand"
"encoding/hex"
"flag"
"fmt"
"log"
@ -19,8 +19,6 @@ import (
"sync"
"sync/atomic"
"time"
"github.com/pulse-lets-go/internal/util"
)
const (
@ -105,7 +103,7 @@ func buildSIPResponse(invite, localAddr string) string {
// Добавляем tag к To если его нет
to = strings.TrimSpace(to)
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"+
@ -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 {
callID := fmt.Sprintf("call-%d-%s@%s", callNum, util.RandomHex(4), "127.0.0.1")
branch := "z9hG4bK-" + util.RandomHex(8)
callID := fmt.Sprintf("call-%d-%s@%s", callNum, randomHex(4), "127.0.0.1")
branch := "z9hG4bK-" + randomHex(8)
return fmt.Sprintf("INVITE sip:%s@%s SIP/2.0\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",
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]
Description=pulse-lets-go — Telephone Load Balancer
Documentation=https://github.com/pulse-lets-go
After=network.target
After=network.target nats.service
Wants=nats.service
[Service]
Type=simple
User=pulse
Group=pulse
ExecStart=/opt/pulse-lets-go/bin/pulse-lets-go
WorkingDirectory=/opt/pulse-lets-go
ExecStart=/usr/local/bin/pulse-lets-go
WorkingDirectory=/usr/local/share/pulse-lets-go
Environment=PULSE_DATA_DIR=/etc/pulse-lets-go
Restart=always
RestartSec=5
@ -17,7 +19,7 @@ NoNewPrivileges=yes
PrivateTmp=yes
ProtectSystem=strict
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

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 {
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}
conn, err := dialer.Dial("tcp", addr)
if err != nil {
@ -268,9 +268,11 @@ func (c *Client) readEventsLoop() {
if err != nil {
if err == io.EOF || strings.Contains(err.Error(), "use of closed network connection") {
c.connected.Store(false)
log.Printf("[ami] соединение разорвано, реконнект...")
c.reconnect()
continue
log.Printf("[ami] соединение разорвано")
if c.shouldReconn.Load() {
go c.ConnectWithRetry()
}
return
}
time.Sleep(100 * time.Millisecond)
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] реконнект успешен")
}

6
internal/api/auth.go
View File

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

11
internal/api/logging.go
View File

@ -1,13 +1,10 @@
package api
import (
"bufio"
"crypto/rand"
"encoding/hex"
"encoding/json"
"fmt"
"log"
"net"
"net/http"
"time"
)
@ -85,11 +82,3 @@ func (w *loggingResponseWriter) WriteHeader(code int) {
func (w *loggingResponseWriter) Write(b []byte) (int, error) {
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
// Общие метрики (counter gauge)
stats := a.engine.GetHealthStats()
fmt.Fprintf(&sb, "# HELP pulse_route_requests_total Total number of route requests.\n")
fmt.Fprintf(&sb, "# TYPE pulse_route_requests_total counter\n")
fmt.Fprintf(&sb, "pulse_route_requests_total %d\n", stats.RouteRequests)
fmt.Fprintf(&sb, "# HELP pulse_nodes_total Total nodes registered.\n")
fmt.Fprintf(&sb, "# TYPE pulse_nodes_total gauge\n")
fmt.Fprintf(&sb, "pulse_nodes_total %d\n", stats.TotalNodes)
fmt.Fprintf(&sb, "# HELP pulse_nodes_healthy Healthy nodes count.\n")
fmt.Fprintf(&sb, "# TYPE pulse_nodes_healthy gauge\n")
fmt.Fprintf(&sb, "pulse_nodes_healthy %d\n", stats.HealthyNodes)
fmt.Fprintf(&sb, "# HELP pulse_uptime_seconds Uptime in seconds.\n")
fmt.Fprintf(&sb, "# TYPE pulse_uptime_seconds gauge\n")
fmt.Fprintf(&sb, "pulse_uptime_seconds %d\n", stats.UptimeSeconds)
fmt.Fprintf(&sb, "\n")
for _, n := range nodes {
id := sanitizePromLabel(n.NodeID)

2
internal/api/route.go
View File

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

39
internal/api/router.go
View File

@ -4,7 +4,6 @@ import (
"encoding/json"
"fmt"
"net/http"
"sync"
"github.com/pulse-lets-go/internal/config"
"github.com/pulse-lets-go/internal/engine"
@ -24,11 +23,6 @@ type API struct {
logFormat string
routeLimiter *rateLimiter
apiLimiter *rateLimiter
gatewayMu sync.RWMutex
gatewayStates map[string]string // trunkID → "up"/"down"
fsStats esl.FsStats
fsStatsMu sync.RWMutex
}
// NewAPI создаёт новый HTTP API с заданными зависимостями.
@ -41,7 +35,6 @@ func NewAPI(eng *engine.Engine, cfgMgr *config.Manager, jwtSecret, monitoringAPI
wsHub: newWSHub(),
natsConnected: natsFn,
logFormat: cfg.Log.Format,
gatewayStates: make(map[string]string),
}
if cfg.RateLimit.Enabled {
a.routeLimiter = newRateLimiter(cfg.RateLimit.RoutePerSec, cfg.RateLimit.RoutePerSec)
@ -137,10 +130,6 @@ func corsMiddleware(next http.Handler) http.Handler {
// BroadcastGatewayEvent транслирует статус gateway всем WS-клиентам.
func (a *API) BroadcastGatewayEvent(gatewayName, trunkID, status string) {
a.gatewayMu.Lock()
a.gatewayStates[trunkID] = status
a.gatewayMu.Unlock()
msg := &models.WsMetricsMessage{
Type: "gateway_update",
Payload: map[string]string{
@ -152,35 +141,9 @@ func (a *API) BroadcastGatewayEvent(gatewayName, trunkID, status string) {
a.wsHub.broadcast(msg)
}
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 с ограничением размера.
// decodeJSON декодирует тело запроса в структуру v.
func decodeJSON(r *http.Request, v interface{}) error {
defer r.Body.Close()
r.Body = http.MaxBytesReader(nil, r.Body, maxBodySize)
if err := json.NewDecoder(r.Body).Decode(v); err != nil {
return fmt.Errorf("декодирование JSON: %w", err)
}

5
internal/api/trunks.go
View File

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

15
internal/api/users.go
View File

@ -1,12 +1,13 @@
package api
import (
"crypto/rand"
"encoding/hex"
"fmt"
"net/http"
"time"
"github.com/pulse-lets-go/internal/models"
"github.com/pulse-lets-go/internal/util"
"golang.org/x/crypto/bcrypt"
)
@ -226,5 +227,15 @@ func findUserIndex(users []models.User, id string) int {
// generateID генерирует ID вида "prefix-XXXXX".
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() {
a.BroadcastAll()
nodes := a.engine.GetAllNodes()
msg := &models.WsMetricsMessage{
Type: "nodes_update",
Payload: nodes,
}
a.wsHub.broadcast(msg)
}()
// Читаем из вебсокета (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() {
nodes := a.engine.GetAllNodes()
msg := &models.WsMetricsMessage{
@ -130,46 +135,6 @@ func (a *API) BroadcastMetrics() {
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.
func (a *API) validateToken(tokenStr string) (*UserInfo, error) {
token, err := parseJWT(tokenStr, a.jwtSecret)

11
internal/config/manager.go
View File

@ -72,16 +72,6 @@ func (e *ESLConfig) GetPassword() string {
return e.Password
}
// GetJWTSecret возвращает JWT-секрет: из переменной окружения, если задан jwt_secret_env, иначе из поля jwt_secret.
func (c *Config) GetJWTSecret() string {
if c.JWTSecretEnv != "" {
if val := os.Getenv(c.JWTSecretEnv); val != "" {
return val
}
}
return c.JWTSecret
}
// Config — корневая конфигурация приложения (config.json).
type Config struct {
NatsURL string `json:"nats_url"`
@ -89,7 +79,6 @@ type Config struct {
NatsPassword string `json:"nats_password"`
ListenAddr string `json:"listen_addr"`
JWTSecret string `json:"jwt_secret"`
JWTSecretEnv string `json:"jwt_secret_env"` // имя переменной окружения с JWT-секретом
MonitoringAPIKey string `json:"monitoring_api_key"`
StaleThresholdSec int `json:"stale_threshold_sec"`
Scoring ScoringConfig `json:"scoring"`

8
internal/engine/engine.go
View File

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

122
internal/esl/client.go
View File

@ -109,7 +109,7 @@ func (c *Client) tryConnect() error {
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}
conn, err := dialer.Dial("tcp", addr)
if err != nil {
@ -119,7 +119,7 @@ func (c *Client) tryConnect() error {
c.reader = bufio.NewReader(conn)
// 1. Читаем auth/request
headers, _, err := c.readMessage()
headers, _, err := c.readMessageLocked()
if err != nil {
conn.Close()
return fmt.Errorf("чтение auth/request: %w", err)
@ -137,7 +137,7 @@ func (c *Client) tryConnect() error {
}
// 3. Читаем auth response
headers, _, err = c.readMessage()
headers, _, err = c.readMessageLocked()
if err != nil {
conn.Close()
return fmt.Errorf("чтение auth response: %w", err)
@ -199,7 +199,7 @@ func (c *Client) Send(command string) (map[string]string, string, error) {
}
// Читаем ответ
return c.readMessage()
return c.readMessageLocked()
}
// Subscribe подписывается на события ESL.
@ -263,7 +263,7 @@ func (c *Client) readEventsLoop() {
continue
}
headers, body, err := c.readMessage()
headers, body, err := c.readMessageUnlocked()
if err != nil {
if err == io.EOF || strings.Contains(err.Error(), "use of closed network connection") {
@ -330,9 +330,48 @@ func (c *Client) reconnect() {
// --- Приватные методы ---
// readMessage читает одно ESL-сообщение: заголовки + тело.
// Вызывающий должен гарантировать монопольный доступ к c.reader.
func (c *Client) readMessage() (map[string]string, string, error) {
// readMessageUnlocked читает одно ESL-сообщение: заголовки + тело.
// Должна вызываться только из readEventsLoop (одиночный читатель).
func (c *Client) readMessageUnlocked() (map[string]string, string, error) {
headers := make(map[string]string)
for {
line, err := c.reader.ReadString('\n')
if err != nil {
return nil, "", err
}
line = strings.TrimRight(line, "\r\n")
if line == "" {
break // конец заголовков
}
parts := strings.SplitN(line, ":", 2)
if len(parts) == 2 {
headers[strings.TrimSpace(parts[0])] = strings.TrimSpace(parts[1])
}
}
// Читаем тело согласно Content-Length
cl := headers["Content-Length"]
if cl == "" {
return headers, "", nil
}
length, err := strconv.Atoi(cl)
if err != nil {
return headers, "", fmt.Errorf("некорректный Content-Length: %s", cl)
}
body := make([]byte, length)
if _, err := io.ReadFull(c.reader, body); err != nil {
return headers, "", fmt.Errorf("чтение тела (Content-Length=%d): %w", length, err)
}
return headers, strings.TrimSpace(string(body)), nil
}
// readMessageLocked читает одно ESL-сообщение: заголовки + тело.
// Вызывающий держит c.mu (для синхронных команд Send).
func (c *Client) readMessageLocked() (map[string]string, string, error) {
headers := make(map[string]string)
for {
@ -401,70 +440,3 @@ func (c *Client) GetStats() Stats {
func (c *Client) IncGatewayOps() {
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)
// - Подписывается на события SOFIA::gateway_register/unregister
// - Запускает чтение асинхронных событий
//
// При разрыве соединения — автоматический реконнект.
func (c *Client) StartEventLoop(ctx context.Context, profile string, onConnect func(), onEvent EventHandler) {
const subscribeEvents = "SOFIA::gateway_register SOFIA::gateway_unregister SOFIA::gateway_expire"

8
internal/esl/gateway.go
View File

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

21
internal/models/types.go
View File

@ -209,29 +209,10 @@ type ZabbixResponse struct {
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 ---
// WsMetricsMessage — сообщение, отправляемое по WebSocket.
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"`
}

2
internal/nats/subscriber.go
View File

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

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 ---
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 {
const token = getToken();
const proto = window.location.protocol === 'https:' ? 'wss' : 'ws';

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

@ -1,40 +1,31 @@
<script lang="ts">
import { browser } from '$app/environment';
import { page } from '$app/stores';
import { goto } from '$app/navigation';
import '../app.css';
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 user: { username: string; role: string } | null = null;
function updateAuth() {
token = typeof localStorage !== 'undefined' ? localStorage.getItem('token') : null;
if (token) {
try { user = JSON.parse(atob(token.split('.')[1])); } catch { user = null; }
} else {
user = null;
}
}
let token: string | null = browser ? localStorage.getItem('token') : null;
let user: { username: string; role: string } | null = token
? (() => { try { return JSON.parse(atob(token.split('.')[1])); } catch { return null; } })()
: null;
function logout() {
if (browser) {
localStorage.removeItem('token');
localStorage.removeItem('refreshToken');
updateAuth();
goto('/login');
}
token = null;
user = null;
}
let pathname = $derived($page.url.pathname);
$effect(() => {
$page.url.pathname;
updateAuth();
});
$: pathname = browser ? window.location.pathname : '';
$: isLogin = pathname === '/login';
</script>
<Toaster position="top-right" />
{#if pathname !== '/login'}
{#if !isLogin}
<div class="flex min-h-screen">
<!-- Sidebar -->
<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 {
getNodes, connectWebSocket, isAuthenticated,
type NodeInfo, type BalancerInfo
type NodeInfo
} 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 balancer: BalancerInfo | null = null;
let ws: WebSocket | null = null;
let error: string | null = null;
@ -22,8 +21,6 @@
ws = connectWebSocket((msg) => {
if (msg.type === 'nodes_update') {
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;
$: 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;
$: 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>
{#if error}
@ -103,51 +79,6 @@
</div>
{/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 -->
<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">