diff --git a/Makefile b/Makefile
index b1de363..19bf3b8 100644
--- a/Makefile
+++ b/Makefile
@@ -53,14 +53,16 @@ build-prod:
# Деплой в PREFIX: бинарник + статика + data + log
deploy: build-prod
@echo "→ Деплой в $(PREFIX)"
- install -d $(PREFIX)/bin $(PREFIX)/data $(PREFIX)/web $(PREFIX)/log
+ install -d $(PREFIX)/bin $(PREFIX)/data $(PREFIX)/web $(PREFIX)/log $(PREFIX)/contrib
install -m 755 $(BINDIR)/$(APP) $(PREFIX)/bin/
cp -r $(WEB_DIR)/build/. $(PREFIX)/web/
+ cp contrib/*.lua contrib/*.json $(PREFIX)/contrib/ 2>/dev/null || true
@echo "✓ Готово: $(PREFIX)"
@echo " $(PREFIX)/bin/$(APP)"
@echo " $(PREFIX)/web/ (SvelteKit статика)"
@echo " $(PREFIX)/data/ (config.json, trunks.json, users.json)"
@echo " $(PREFIX)/log/ (ASCII-лог метрик)"
+ @echo " $(PREFIX)/contrib/ (route.lua, конфиги агентов)"
# Установка systemd unit
install-systemd:
diff --git a/cmd/pulse-lets-go/main.go b/cmd/pulse-lets-go/main.go
index e0379fe..45000e0 100644
--- a/cmd/pulse-lets-go/main.go
+++ b/cmd/pulse-lets-go/main.go
@@ -139,8 +139,13 @@ func main() {
// 6. HTTP API
apiHandler = api.NewAPI(eng, cfgMgr, cfg.GetJWTSecret(), cfg.MonitoringAPIKey, sub.IsConnected, cfg)
+ apiHandler.SetRouteCachePath(filepath.Join(dataDir, "route_cache.json"))
apiHandler.RebuildTrunkCache() // первичное построение in-memory кэша gateway
+ sub.SetOnMetric(func() {
+ apiHandler.PushRouteCache()
+ })
+
// При эвикции stale-ноды — удаляем её balance-транк из trunks.json
eng.SetOnNodeEvicted(func(nodeID string) {
trunks, err := cfgMgr.ReadTrunks()
@@ -232,6 +237,7 @@ func main() {
go func() {
ticker := time.NewTicker(5 * time.Second)
for range ticker.C {
+ apiHandler.PushRouteCache()
if eslClient != nil && eslClient.IsConnected() {
if stats, err := eslClient.FetchFsStats(); err == nil {
apiHandler.UpdateFsStats(stats)
diff --git a/contrib/route.lua b/contrib/route.lua
index 925d76e..4efc8db 100644
--- a/contrib/route.lua
+++ b/contrib/route.lua
@@ -3,7 +3,7 @@
Телефонный балансировщик нагрузки на базе FreeSWITCH + NATS.
┌─ Размещение ─────────────────────────────────────────────────────┐
- │ /etc/freeswitch/scripts/route.lua │
+ │ /opt/pulse-lets-go/contrib/route.lua │
│ │
│ ┌─ Использование в dialplan ────────────────────────────────────┐│
│ │ ││
@@ -13,24 +13,30 @@
│ │ ││
│ └───────────────────────────────────────────────────────────────┘│
│ │
+ │ ┌─ Кэширование (защита от блокировки dialplan-тредов) ───────────┐│
+ │ │ Уровень 1: FS global variable (ESL push от balancer, ~0.1µs) ││
+ │ │ Уровень 2: File cache (/tmp/pulse_route_cache.json, ~50µs) ││
+ │ │ Уровень 3: HTTP API (/api/route, ~1-5ms, safety net) ││
+ │ └───────────────────────────────────────────────────────────────┘│
+ │ │
│ ┌─ Переменные канала (опционально) ─────────────────────────────┐│
- │ │ balancer_url — URL pulse-lets-go (default: localhost:8080) ││
- │ │ ingress_trunk — ID входящего транка ││
+ │ │ balancer_url — URL pulse-lets-go (default: localhost:8080)││
+ │ │ ingress_trunk — ID входящего транка ││
+ │ │ route_timeout — таймаут HTTP fallback (default: 3с) ││
+ │ │ route_cache_path — путь к файловому кэшу (default: /data/…) ││
│ └───────────────────────────────────────────────────────────────┘│
│ │
│ ┌─ Требования FreeSWITCH ───────────────────────────────────────┐│
- │ │ mod_curl — в modules.conf.xml ││
+ │ │ mod_curl — (только для уровня 3) ││
│ │ cjson — входит в стандартную поставку FS Lua ││
│ └───────────────────────────────────────────────────────────────┘│
└───────────────────────────────────────────────────────────────────┘
Логика:
- 1. Извлечь caller_id + destination из переменных сессии
- 2. HTTP GET /api/route?caller_id=X&dest=Y&ingress_trunk=Z
- 3. Разобрать JSON-ответ
- 4. При fallback=true → bridge на fallback gateway (оператор)
- 5. При ошибке → кладём трубку с NORMAL_TEMPORARY_FAILURE
- 6. Нормальный маршрут → bridge на sip_gateway выбранной ноды
+ 1. Попробовать FS global variable pulse_route_json (ESL push)
+ 2. Попробовать файловый кэш
+ 3. HTTP API /api/route (fallback)
+ 4. Разобрать JSON → bridge на sip_gateway
]]--
-- ===================================================================
@@ -40,13 +46,16 @@
-- URL балансировщика (переопределяется через переменную канала)
local BALANCER_URL = session:getVariable("balancer_url") or "http://localhost:8080"
--- Таймаут HTTP-запроса в секундах (через параметр curl)
+-- Таймаут HTTP-запроса в секундах
local HTTP_TIMEOUT = tonumber(session:getVariable("route_timeout") or "3")
+-- Путь к файловому кэшу маршрута
+local ROUTE_CACHE_PATH = session:getVariable("route_cache_path") or "/opt/pulse-lets-go/data/route_cache.json"
+
-- JSON-парсер
local cjson = require("cjson")
--- API FreeSWITCH для вызова mod_curl
+-- API FreeSWITCH для вызова curl
local api = freeswitch.API()
-- ===================================================================
@@ -58,26 +67,6 @@ local function log(level, msg)
freeswitch.consoleLog(level, "[route] " .. msg .. "\n")
end
--- Выполнить HTTP GET и вернуть тело ответа (без заголовков).
--- mod_curl API возвращает сырой HTTP-ответ: заголовки + \r?\n\r?\n + тело.
-local function http_get(url)
- local cmd = "curl " .. url
- local raw = api:execute("curl", cmd)
-
- if not raw or raw == "" then
- return nil, "пустой ответ от curl"
- end
-
- -- Отделяем заголовки от тела
- -- Ищем двойной перенос строки (CRLF или LF)
- local body = raw:match("\r?\n\r?\n(.+)")
- if not body then
- -- Если нет заголовков — весь ответ это тело
- return raw, nil
- end
- return body, nil
-end
-
-- Безопасная отправка трубки с кодом причины
local function safe_hangup(cause)
if session:ready() then
@@ -85,8 +74,95 @@ local function safe_hangup(cause)
end
end
+-- Попробовать получить JSON-маршрут из FS global variable (уровень 1).
+local function try_global_cache()
+ -- session:getVariable с префиксом global_ читает глобальные переменные FS
+ local val = session:getVariable("global_pulse_route_json")
+ if val and val ~= "" then
+ return val
+ end
+ return nil
+end
+
+-- Попробовать получить JSON-маршрут из файлового кэша (уровень 2).
+local function try_file_cache()
+ local f, err = io.open(ROUTE_CACHE_PATH, "r")
+ if not f then
+ return nil
+ end
+ local content = f:read("*a")
+ f:close()
+ if content and content ~= "" then
+ return content
+ end
+ return nil
+end
+
+-- Попробовать получить JSON-маршрут через HTTP API (уровень 3, fallback).
+local function try_http(caller_id, dest, ing_trunk)
+ local query = string.format("caller_id=%s&dest=%s", caller_id, dest)
+ if ing_trunk ~= "" then
+ query = query .. "&ingress_trunk=" .. ing_trunk
+ end
+ local url = BALANCER_URL .. "/api/route?" .. query
+
+ local cmd = "curl " .. url
+ local raw = api:execute("curl", cmd)
+
+ if not raw or raw == "" then
+ return nil, "пустой ответ от curl"
+ end
+
+ -- Отделяем заголовки от тела (mod_curl возвращает сырой HTTP)
+ local body = raw:match("\r?\n\r?\n(.+)")
+ if not body then
+ body = raw
+ end
+ return body, nil
+end
+
+-- Разобрать JSON и выполнить bridge.
+-- Возвращает true если звонок обработан (bridge или error).
+local function process_route_json(json_str, source)
+ if not json_str then
+ return false
+ end
+
+ local ok, route = pcall(cjson.decode, json_str)
+ if not ok or type(route) ~= "table" then
+ log("warn", source .. ": некорректный JSON")
+ return false
+ end
+
+ -- Ошибка балансировщика (нет зарегистрированных нод)
+ if route.error then
+ log("err", source .. ": " .. tostring(route.error))
+ safe_hangup()
+ return true
+ end
+
+ -- Фолбэк (все ноды unhealthy)
+ if route.fallback and route.sip_gateway and route.sip_gateway ~= "" then
+ log("info", string.format("%s: FALLBACK → %s (причина: %s)",
+ source, route.sip_gateway, route.reason or "all_nodes_unhealthy"))
+ session:execute("bridge", route.sip_gateway)
+ return true
+ end
+
+ -- Успешный маршрут
+ if route.sip_gateway and route.sip_gateway ~= "" then
+ log("info", string.format("%s: ROUTE %s score=%s → %s",
+ source, route.node_id or "?", tostring(route.score or 0), route.sip_gateway))
+ session:execute("bridge", route.sip_gateway)
+ return true
+ end
+
+ log("warn", source .. ": нет sip_gateway в ответе")
+ return false
+end
+
-- ===================================================================
--- Основная логика
+-- Основная логика (трёхуровневый кэш)
-- ===================================================================
-- Извлекаем параметры вызова
@@ -94,63 +170,29 @@ local caller_id = session:getVariable("caller_id_number") or ""
local dest = session:getVariable("destination_number") or ""
local ing_trunk = session:getVariable("ingress_trunk") or ""
--- Собираем URL
-local query = string.format("caller_id=%s&dest=%s", caller_id, dest)
-if ing_trunk ~= "" then
- query = query .. "&ingress_trunk=" .. ing_trunk
+-- Уровень 1: FS global variable (ESL push, sub-µs)
+local json_str = try_global_cache()
+if process_route_json(json_str, "global") then
+ return
end
-local url = BALANCER_URL .. "/api/route?" .. query
-log("info", "запрос маршрута: " .. url)
+-- Уровень 2: файловый кэш (локальный, ~50µs)
+json_str = try_file_cache()
+if process_route_json(json_str, "file") then
+ return
+end
--- Запрашиваем маршрут у pulse-lets-go
-local body, curl_err = http_get(url)
+-- Уровень 3: HTTP API (fallback, ~1-5ms)
+log("info", "запрос маршрута HTTP: " .. BALANCER_URL .. "/api/route?caller_id=" .. caller_id .. "&dest=" .. dest)
+
+local body, http_err = try_http(caller_id, dest, ing_trunk)
if not body then
- log("err", "ошибка HTTP: " .. (curl_err or "unknown"))
+ log("err", "все уровни кэша недоступны: " .. (http_err or "unknown"))
safe_hangup()
return
end
--- Парсим JSON
-local ok, route = pcall(cjson.decode, body)
-if not ok or type(route) ~= "table" then
- log("err", "ошибка парсинга JSON: " .. tostring(route))
+if not process_route_json(body, "http") then
+ log("err", "непредвиденный формат ответа балансировщика")
safe_hangup()
- return
end
-
--- ── Фолбэк (все ноды unhealthy) ──────────────────────────────────
-if route.fallback and route.sip_gateway then
- log("info", string.format("FALLBACK → %s (причина: %s)",
- route.sip_gateway, route.reason or "all_nodes_unhealthy"))
-
- if route.nodes then
- for _, n in ipairs(route.nodes) do
- log("info", string.format(" %s: score=%d reason=%s",
- n.id or "?", n.score or 0, n.reason or "-"))
- end
- end
-
- session:execute("bridge", route.sip_gateway)
- return
-end
-
--- ── Ошибка API (нет зарегистрированных нод) ──────────────────────
-if route.error then
- log("err", "ошибка балансировщика: " .. route.error)
- safe_hangup()
- return
-end
-
--- ── Успешный маршрут ─────────────────────────────────────────────
-if route.sip_gateway then
- log("info", string.format("ROUTE: %s score=%d → %s",
- route.node_id or "?", route.score or 0, route.sip_gateway))
-
- session:execute("bridge", route.sip_gateway)
- return
-end
-
--- ── Непредвиденная ситуация ──────────────────────────────────────
-log("err", "нет sip_gateway в ответе балансировщика")
-safe_hangup()
diff --git a/internal/api/nodes.go b/internal/api/nodes.go
index 61acf32..3f48d91 100644
--- a/internal/api/nodes.go
+++ b/internal/api/nodes.go
@@ -47,6 +47,8 @@ func (a *API) handleToggleNode(w http.ResponseWriter, r *http.Request) {
return
}
+ a.PushRouteCache()
+
info, _ := a.engine.GetNodeInfo(nodeID)
writeJSON(w, http.StatusOK, info)
}
diff --git a/internal/api/route.go b/internal/api/route.go
index 022a118..771a7c3 100644
--- a/internal/api/route.go
+++ b/internal/api/route.go
@@ -1,8 +1,11 @@
package api
import (
+ "encoding/json"
"net/http"
+ "os"
+ "github.com/pulse-lets-go/internal/esl"
"github.com/pulse-lets-go/internal/models"
)
@@ -90,6 +93,45 @@ func (a *API) handleSetRouteMode(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, map[string]string{"mode": req.Mode})
}
+// GetBalanceGateway возвращает gateway для ноды из in-memory кэша (O(1)).
+func (a *API) GetBalanceGateway(nodeID string) (string, bool) {
+ return a.cachedBalanceGateway(nodeID)
+}
+
+// GetFallbackGateway возвращает fallback gateway из in-memory кэша (O(1)).
+func (a *API) GetFallbackGateway() string {
+ return a.cachedFallbackGateway()
+}
+
+// SetRouteCachePath устанавливает путь к файловому кэшу маршрута.
+func (a *API) SetRouteCachePath(path string) {
+ a.routeCachePath = path
+}
+
+// PushRouteCache читает best-node из engine, резолвит gateway из trunk cache,
+// атомарно пишет файловый кэш и пушит маршрут в FreeSWITCH через ESL.
+func (a *API) PushRouteCache() {
+ nodeID, score, fallback := a.engine.GetBestNode()
+
+ var gw string
+ if nodeID != "" && !fallback {
+ gw, _ = a.cachedBalanceGateway(nodeID)
+ }
+ if gw == "" {
+ gw = a.cachedFallbackGateway()
+ }
+
+ entry := buildRouteCacheEntry(nodeID, gw, score, fallback)
+
+ if a.routeCachePath != "" {
+ writeRouteCacheFile(a.routeCachePath, entry)
+ }
+
+ if a.eslClient != nil && a.eslClient.IsConnected() {
+ esl.PushRoute(a.eslClient, entry)
+ }
+}
+
// cachedBalanceGateway возвращает gateway для balance-транка из in-memory кэша (O(1)).
func (a *API) cachedBalanceGateway(nodeID string) (string, bool) {
a.trunkCacheMu.RLock()
@@ -105,6 +147,36 @@ func (a *API) cachedFallbackGateway() string {
return a.fallbackGateway
}
+// buildRouteCacheEntry строит запись для кэширования в ESL global / файл.
+func buildRouteCacheEntry(nodeID, gw string, score float64, fallback bool) esl.RouteCacheEntry {
+ entry := esl.RouteCacheEntry{
+ SIPGateway: gw,
+ Score: score,
+ Fallback: fallback,
+ }
+ if nodeID == "" {
+ entry.Error = "no_nodes_registered"
+ } else if fallback {
+ entry.Reason = "all_nodes_unhealthy"
+ } else {
+ entry.NodeID = nodeID
+ }
+ return entry
+}
+
+// writeRouteCacheFile атомарно записывает файловый кэш маршрута.
+func writeRouteCacheFile(path string, entry esl.RouteCacheEntry) {
+ b, err := json.Marshal(entry)
+ if err != nil {
+ return
+ }
+ tmpPath := path + ".tmp"
+ if err := os.WriteFile(tmpPath, b, 0644); err != nil {
+ return
+ }
+ os.Rename(tmpPath, path)
+}
+
// getRouteNodeInfo возвращает список нод с их скорами для ответа маршрутизации.
func (a *API) getRouteNodeInfo() []models.RouteNodeInfo {
nodes := a.engine.GetAllNodes()
diff --git a/internal/api/router.go b/internal/api/router.go
index 0c33f22..44631f9 100644
--- a/internal/api/router.go
+++ b/internal/api/router.go
@@ -34,6 +34,8 @@ type API struct {
gatewayStates map[string]string // trunkID → "up"/"down"
fsStats esl.FsStats
fsStatsMu sync.RWMutex
+
+ routeCachePath string // путь к файловому кэшу маршрута (для Lua)
}
// NewAPI создаёт новый HTTP API с заданными зависимостями.
diff --git a/internal/config/manager.go b/internal/config/manager.go
index 06af859..d374015 100644
--- a/internal/config/manager.go
+++ b/internal/config/manager.go
@@ -420,7 +420,7 @@ func (m *Manager) createDefaultConfig() error {
ListenAddr: ":8080",
JWTSecret: jwtSecret,
MonitoringAPIKey: apiKey,
- StaleThresholdSec: 20,
+ StaleThresholdSec: 10,
Scoring: ScoringConfig{
IdleCPUMin: 5.0,
CallFailureRateLethal: 15.0,
diff --git a/internal/engine/engine.go b/internal/engine/engine.go
index c795da7..83a6131 100644
--- a/internal/engine/engine.go
+++ b/internal/engine/engine.go
@@ -456,12 +456,18 @@ func (e *Engine) scoreNodeLocked(ns *models.NodeState, firstMetric bool) {
// recalcBestLocked пересчитывает кэш лучшей ноды.
// Вызывающий держит write lock.
+// Проверяет staleness (now - ns.TS > staleThreshold) — исключает ноды без свежих метрик.
func (e *Engine) recalcBestLocked() {
bestID := ""
bestScore := -999.0
anyHealthy := false
+ now := time.Now().Unix()
+ staleThreshold := int64(e.cfg.StaleThresholdSec)
for id, ns := range e.nodes {
+ if now-ns.TS > staleThreshold {
+ continue
+ }
if ns.Score >= 0 {
anyHealthy = true
}
diff --git a/internal/engine/router.go b/internal/engine/router.go
index b82cfd2..cff4c17 100644
--- a/internal/engine/router.go
+++ b/internal/engine/router.go
@@ -318,8 +318,10 @@ func (e *Engine) PickNodeForCall(callerID, dest, ingress string) (nodeID string,
case "node":
e.mu.RLock()
ns, ok := e.nodes[cr.ActionNodeID]
+ now := time.Now().Unix()
+ staleThreshold := int64(e.cfg.StaleThresholdSec)
e.mu.RUnlock()
- if ok && ns.Score >= 0 {
+ if ok && ns.Score >= 0 && now-ns.TS <= staleThreshold {
nodeID, score, matchedRule = cr.ActionNodeID, 100, cr.Name
resolved = true
}
diff --git a/internal/esl/routepush.go b/internal/esl/routepush.go
new file mode 100644
index 0000000..8d5b6d3
--- /dev/null
+++ b/internal/esl/routepush.go
@@ -0,0 +1,36 @@
+package esl
+
+import (
+ "encoding/json"
+ "fmt"
+ "time"
+)
+
+// RouteCacheEntry — запись для кэширования маршрута в FS global variable и файле.
+type RouteCacheEntry struct {
+ NodeID string `json:"node_id,omitempty"`
+ SIPGateway string `json:"sip_gateway,omitempty"`
+ Score float64 `json:"score,omitempty"`
+ Fallback bool `json:"fallback,omitempty"`
+ Reason string `json:"reason,omitempty"`
+ Error string `json:"error,omitempty"`
+ TS int64 `json:"ts"`
+}
+
+// PushRoute пушит текущий маршрут в FreeSWITCH через global variable.
+// Если клиент не подключён — возвращает ошибку молча (вызывающий игнорирует).
+func PushRoute(c *Client, entry RouteCacheEntry) error {
+ if c == nil || !c.IsConnected() {
+ return fmt.Errorf("esl не подключён")
+ }
+
+ entry.TS = time.Now().Unix()
+
+ jsonBytes, err := json.Marshal(entry)
+ if err != nil {
+ return fmt.Errorf("маршалинг route cache: %w", err)
+ }
+
+ cmd := fmt.Sprintf("api set_global pulse_route_json=%s", string(jsonBytes))
+ return c.SendAsync(cmd)
+}
diff --git a/internal/nats/subscriber.go b/internal/nats/subscriber.go
index 4f8f52a..379eb4c 100644
--- a/internal/nats/subscriber.go
+++ b/internal/nats/subscriber.go
@@ -23,6 +23,7 @@ type Subscriber struct {
engine *engine.Engine
logger *filelog.Logger
onNewNode func(nodeID, sipGateway string) // вызывается при первой метрике от новой ноды
+ onMetric func() // вызывается после каждой обработанной метрики
wg sync.WaitGroup
}
@@ -63,6 +64,11 @@ func (s *Subscriber) SetOnNewNode(fn func(nodeID, sipGateway string)) {
s.onNewNode = fn
}
+// SetOnMetric устанавливает callback, вызываемый после каждой обработанной метрики.
+func (s *Subscriber) SetOnMetric(fn func()) {
+ s.onMetric = fn
+}
+
// handleMetric — обработчик входящих сообщений NATS.
func (s *Subscriber) handleMetric(msg *natsgo.Msg) {
var metric models.NodeMetric
@@ -86,6 +92,10 @@ func (s *Subscriber) handleMetric(msg *natsgo.Msg) {
ns := s.engine.UpdateMetric(&metric)
s.logger.Log(ns)
+ if s.onMetric != nil {
+ s.onMetric()
+ }
+
// Авто-создание balance-транка для новой ноды
if isNew && metric.SIPGateway != "" && s.onNewNode != nil {
s.onNewNode(metric.NodeID, metric.SIPGateway)