40b8afb150
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
113 lines
3.2 KiB
Go
113 lines
3.2 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) // вызывается при первой метрике от новой ноды
|
||
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
|
||
}
|
||
|
||
// 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)
|
||
|
||
// Авто-создание 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()
|
||
}
|