78785e54e1
Three-tier routing cache (ESL global -> file -> HTTP) eliminates HTTP
from call path for 2500-5000 concurrent calls. Lua reads cached route
in ~0.1us instead of blocking on api:execute('curl', ...).
Engine fixes:
- recalcBestLocked now skips stale nodes (was 5 min bug -> now ~10-15s)
- PickNodeForCall action_type='node' checks staleness for consistency
- onMetric callback pushes cache on every metric (no ticker delay)
Config:
- stale_threshold_sec default 20 -> 10 (industry standard)
- contrib/ included in Makefile deploy target
- route.lua paths updated for /opt/pulse-lets-go
123 lines
3.5 KiB
Go
123 lines
3.5 KiB
Go
// Package nats содержит подписчика на NATS-метрики от нижестоящих АТС.
|
||
package nats
|
||
|
||
import (
|
||
"encoding/json"
|
||
"fmt"
|
||
"log"
|
||
"sync"
|
||
|
||
natsgo "github.com/nats-io/nats.go"
|
||
|
||
"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.>"
|
||
|
||
// Subscriber подписывается на NATS и обновляет engine + пишет лог.
|
||
type Subscriber struct {
|
||
nc *natsgo.Conn
|
||
sub *natsgo.Subscription
|
||
engine *engine.Engine
|
||
logger *filelog.Logger
|
||
onNewNode func(nodeID, sipGateway string) // вызывается при первой метрике от новой ноды
|
||
onMetric func() // вызывается после каждой обработанной метрики
|
||
wg sync.WaitGroup
|
||
}
|
||
|
||
// NewSubscriber подключается к NATS и запускает подписку.
|
||
func NewSubscriber(natsURL, user, password string, eng *engine.Engine, l *filelog.Logger) (*Subscriber, error) {
|
||
opts := []natsgo.Option{
|
||
natsgo.ReconnectWait(natsgo.DefaultReconnectWait),
|
||
natsgo.MaxReconnects(-1),
|
||
}
|
||
if user != "" && password != "" {
|
||
opts = append(opts, natsgo.UserInfo(user, password))
|
||
}
|
||
|
||
nc, err := natsgo.Connect(natsURL, opts...)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("nats connect: %w", err)
|
||
}
|
||
|
||
s := &Subscriber{
|
||
nc: nc,
|
||
engine: eng,
|
||
logger: l,
|
||
}
|
||
|
||
sub, err := nc.Subscribe(subject, s.handleMetric)
|
||
if err != nil {
|
||
nc.Close()
|
||
return nil, fmt.Errorf("nats subscribe: %w", err)
|
||
}
|
||
s.sub = sub
|
||
|
||
log.Printf("[nats] подписан на %s, подключён к %s", subject, natsURL)
|
||
return s, nil
|
||
}
|
||
|
||
// SetOnNewNode устанавливает callback, вызываемый при первой метрике от новой ноды.
|
||
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
|
||
if err := json.Unmarshal(msg.Data, &metric); err != nil {
|
||
log.Printf("[nats] ошибка парсинга метрики: %v", err)
|
||
return
|
||
}
|
||
|
||
// Валидация минимальных полей
|
||
if metric.NodeID == "" {
|
||
log.Printf("[nats] метрика без node_id, игнорируется")
|
||
return
|
||
}
|
||
if metric.MaxCalls <= 0 {
|
||
metric.MaxCalls = 250 // разумное значение по умолчанию
|
||
}
|
||
|
||
// Проверка: это новая нода с sip_gateway?
|
||
isNew := !s.engine.HasNode(metric.NodeID)
|
||
|
||
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)
|
||
}
|
||
}
|
||
|
||
// IsConnected возвращает true, если NATS-соединение активно.
|
||
func (s *Subscriber) IsConnected() bool {
|
||
if s.nc == nil {
|
||
return false
|
||
}
|
||
return s.nc.IsConnected()
|
||
}
|
||
|
||
// Close останавливает подписку и закрывает соединение.
|
||
func (s *Subscriber) Close() {
|
||
if s.sub != nil {
|
||
s.sub.Unsubscribe()
|
||
}
|
||
if s.nc != nil {
|
||
s.nc.Close()
|
||
}
|
||
s.wg.Wait()
|
||
}
|