From c713a4f142d3a85714f040e10fef0015c4023497 Mon Sep 17 00:00:00 2001 From: Maksim Totmin Date: Thu, 25 Jun 2026 16:59:10 +0700 Subject: [PATCH] feat: metrics agent for FreeSWITCH and Asterisk nodes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - pulse-lets-go-agent: collects metrics from PBX nodes via ESL (FreeSWITCH) or AMI (Asterisk) - AMI client (raw TCP) for Asterisk Manager Interface - Collector interface: collect, connect, subscribe hangup events, close - FreeSWITCH collector: show channels, max_sessions, idle_cpu - Asterisk collector: CoreShowChannels, CoreSettings - System collector: /proc/loadavg, /proc/stat (CPU idle) - Failure tracker: sliding window for call_failure_rate - NATS publisher: pulse.metrics. every 5 seconds - Graceful shutdown, retry logic (5 failures → status=down) - Template configs for UC nodes (Asterisk) and ses-sip (FreeSWITCH) --- cmd/pulse-lets-go-agent/agent/agent.go | 174 +++++++++++ cmd/pulse-lets-go-agent/agent/collector.go | 70 +++++ .../agent/collector_asterisk.go | 107 +++++++ .../agent/collector_freeswitch.go | 134 ++++++++ cmd/pulse-lets-go-agent/agent/config.go | 124 ++++++++ cmd/pulse-lets-go-agent/agent/failtrack.go | 52 ++++ cmd/pulse-lets-go-agent/agent/publisher.go | 57 ++++ cmd/pulse-lets-go-agent/agent/system.go | 84 +++++ cmd/pulse-lets-go-agent/main.go | 44 +++ contrib/agent-sessip.json | 15 + contrib/agent-uc.json | 16 + internal/ami/client.go | 293 ++++++++++++++++++ 12 files changed, 1170 insertions(+) create mode 100644 cmd/pulse-lets-go-agent/agent/agent.go create mode 100644 cmd/pulse-lets-go-agent/agent/collector.go create mode 100644 cmd/pulse-lets-go-agent/agent/collector_asterisk.go create mode 100644 cmd/pulse-lets-go-agent/agent/collector_freeswitch.go create mode 100644 cmd/pulse-lets-go-agent/agent/config.go create mode 100644 cmd/pulse-lets-go-agent/agent/failtrack.go create mode 100644 cmd/pulse-lets-go-agent/agent/publisher.go create mode 100644 cmd/pulse-lets-go-agent/agent/system.go create mode 100644 cmd/pulse-lets-go-agent/main.go create mode 100644 contrib/agent-sessip.json create mode 100644 contrib/agent-uc.json create mode 100644 internal/ami/client.go diff --git a/cmd/pulse-lets-go-agent/agent/agent.go b/cmd/pulse-lets-go-agent/agent/agent.go new file mode 100644 index 0000000..42535e1 --- /dev/null +++ b/cmd/pulse-lets-go-agent/agent/agent.go @@ -0,0 +1,174 @@ +package agent + +import ( + "fmt" + "log" + "os" + "os/signal" + "sync/atomic" + "syscall" + "time" + + "github.com/pulse-lets-go/internal/models" +) + +const maxConsecutiveFailures = 5 + +// Agent — главный цикл сбора метрик и публикации в NATS. +type Agent struct { + cfg *Config + publisher *Publisher + collector Collector + failtrack *FailureTracker + pubCount atomic.Int64 +} + +// NewAgent создаёт агента на основе конфигурации. +func NewAgent(cfg *Config) (*Agent, error) { + if err := cfg.Validate(); err != nil { + return nil, err + } + + // Publisher + pub, err := NewPublisher(cfg.NatsURL, cfg.NatsUser, cfg.NatsPassword, cfg.NodeID) + if err != nil { + return nil, err + } + + // Collector — в зависимости от типа PBX + var col Collector + switch cfg.Type { + case "freeswitch": + col = NewFSCollector(cfg.ESL.Host, cfg.ESL.Port, cfg.ESL.Password) + case "asterisk": + col = NewAsteriskCollector(cfg.AMI.Host, cfg.AMI.Port, cfg.AMI.Username, cfg.AMI.Password) + default: + pub.Close() + return nil, fmt.Errorf("неподдерживаемый тип PBX: %s", cfg.Type) + } + + if err := col.Connect(); err != nil { + pub.Close() + return nil, fmt.Errorf("подключение к %s: %w", cfg.Type, err) + } + + // Failure rate tracker + ft := NewFailureTracker(cfg.FailureWindow) + + // Подписка на hangup-события + col.ListenHangups(func(success bool) { + ft.Record(success) + }) + + return &Agent{ + cfg: cfg, + publisher: pub, + collector: col, + failtrack: ft, + }, nil +} + +// Start запускает главный цикл агента: сбор метрик → публикация. +// Блокирует до получения сигнала завершения (SIGINT, SIGTERM). +func (a *Agent) Start() { + log.Printf("[agent] запущен: node=%s, type=%s, interval=%ds", + a.cfg.NodeID, a.cfg.Type, a.cfg.IntervalSec) + + ticker := time.NewTicker(time.Duration(a.cfg.IntervalSec) * time.Second) + defer ticker.Stop() + + quit := make(chan os.Signal, 1) + signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM) + + // Счётчик последовательных ошибок сбора + consecutiveFailures := 0 + + // Первый сбор сразу + a.collectAndPublish(&consecutiveFailures) + + for { + select { + case <-ticker.C: + a.collectAndPublish(&consecutiveFailures) + case <-quit: + log.Println("[agent] получен сигнал завершения") + a.Shutdown() + return + } + } +} + +// collectAndPublish выполняет один цикл: сбор метрик → публикация. +func (a *Agent) collectAndPublish(failures *int) { + // 1. Системные метрики (CPU, load) + sys := CollectSystem() + + // 2. Метрики PBX (calls, max_calls, статус) + pbx, err := a.collector.Collect() + if err != nil { + *failures++ + log.Printf("[agent] ошибка сбора метрик (попытка %d): %v", *failures, err) + + if *failures >= maxConsecutiveFailures { + // После N неудач — считаем ноду "down" + pbx = &PBXResult{ + Status: "down", + } + } else { + // Пробуем ещё раз через секунду + time.Sleep(1 * time.Second) + pbx, err = a.collector.Collect() + if err != nil { + return // пропускаем этот тик + } + *failures = 0 + } + } else { + *failures = 0 + } + + // 3. Failure rate + failRate := a.failtrack.Rate() + + // 4. SIP gateway + sipGW := a.cfg.GetSIPGateway("") + + // 5. Формируем метрику + metric := ToMetric(a.cfg.NodeID, pbx, sys, failRate, sipGW, a.cfg.MaxCalls) + + // 6. Публикуем + if err := a.publisher.Publish(metric); err != nil { + log.Printf("[agent] ошибка публикации: %v", err) + return + } + + a.pubCount.Add(1) + log.Printf("[agent] опубликована метрика: calls=%d/%d load=%.2f fail=%.1f%% cpu=%.0f%% status=%s", + metric.ActiveCalls, metric.MaxCalls, metric.LoadAvg, metric.CallFailureRate, metric.IdleCPU, metric.Status) +} + +// Shutdown gracefully shuts down the agent. +func (a *Agent) Shutdown() { + log.Printf("[agent] остановка, всего опубликовано метрик: %d", a.pubCount.Load()) + if a.collector != nil { + a.collector.Close() + } + if a.publisher != nil { + a.publisher.Close() + } +} + +// CollectAndPublish экспортированный метод для ручного вызова (тесты, дебаг). +func (a *Agent) CollectAndPublish() *models.NodeMetric { + sys := CollectSystem() + pbx, err := a.collector.Collect() + if err != nil { + pbx = &PBXResult{Status: "down"} + } + failRate := a.failtrack.Rate() + sipGW := a.cfg.GetSIPGateway("") + metric := ToMetric(a.cfg.NodeID, pbx, sys, failRate, sipGW, a.cfg.MaxCalls) + a.publisher.Publish(metric) + a.pubCount.Add(1) + return metric +} diff --git a/cmd/pulse-lets-go-agent/agent/collector.go b/cmd/pulse-lets-go-agent/agent/collector.go new file mode 100644 index 0000000..eee61ca --- /dev/null +++ b/cmd/pulse-lets-go-agent/agent/collector.go @@ -0,0 +1,70 @@ +package agent + +import "github.com/pulse-lets-go/internal/models" + +// PBXResult — результат сбора метрик от PBX (FS или Asterisk). +type PBXResult struct { + ActiveCalls int // текущее количество звонков + MaxCalls int // ёмкость (0 — не удалось получить) + IdleCPU float64 // idle CPU из PBX (0 — не удалось, fallback на system) + Status string // "ok" — PBX отвечает нормально + SIPGateway string // авто-определённый SIP-адрес +} + +// Collector — интерфейс сбора метрик с PBX. +type Collector interface { + // Type возвращает тип PBX: "freeswitch" или "asterisk". + Type() string + + // Connect подключается к PBX (ESL auth / AMI login). + Connect() error + + // Collect собирает метрики с PBX. + // Возвращает PBXResult и ошибку, если сбор не удался. + Collect() (*PBXResult, error) + + // ListenHangups подписывается на события завершения звонков + // для расчёта call_failure_rate. + ListenHangups(onHangup func(success bool)) + + // Close закрывает соединение с PBX. + Close() +} + +// HangupEvent — событие завершения звонка. +type HangupEvent struct { + Success bool // true — звонок успешен, false — ошибка + Cause string // код причины (NORMAL_CLEARING, BUSY, и т.д.) +} + +// ToMetric формирует models.NodeMetric из результатов сбора. +func ToMetric(nodeID string, pbx *PBXResult, sys *SystemStats, failureRate float64, sipGateway string, maxCallsDefault int) *models.NodeMetric { + // Определяем max_calls: приоритет PBX → конфиг + maxCalls := pbx.MaxCalls + if maxCalls <= 0 { + maxCalls = maxCallsDefault + } + // Определяем idle_cpu: приоритет PBX → system + idleCPU := pbx.IdleCPU + if idleCPU <= 0 { + idleCPU = sys.IdleCPU + } + + // Определяем статус + status := pbx.Status + if pbx.Status == "ok" && sys.IdleCPU < 5 { + status = "degraded" + } + + return &models.NodeMetric{ + NodeID: nodeID, + TS: nowTS(), + Status: status, + ActiveCalls: pbx.ActiveCalls, + MaxCalls: maxCalls, + IdleCPU: idleCPU, + LoadAvg: sys.Load1, + CallFailureRate: failureRate, + SIPGateway: sipGateway, + } +} diff --git a/cmd/pulse-lets-go-agent/agent/collector_asterisk.go b/cmd/pulse-lets-go-agent/agent/collector_asterisk.go new file mode 100644 index 0000000..6d8c157 --- /dev/null +++ b/cmd/pulse-lets-go-agent/agent/collector_asterisk.go @@ -0,0 +1,107 @@ +package agent + +import ( + "fmt" + "strconv" + "strings" + + "github.com/pulse-lets-go/internal/ami" +) + +// AsteriskCollector собирает метрики с Asterisk через AMI. +type AsteriskCollector struct { + client *ami.Client +} + +// NewAsteriskCollector создаёт Asterisk-коллектор. +func NewAsteriskCollector(host string, port int, username, password string) *AsteriskCollector { + return &AsteriskCollector{ + client: ami.NewClient(host, port, username, password), + } +} + +func (ac *AsteriskCollector) Type() string { return "asterisk" } + +func (ac *AsteriskCollector) Connect() error { + return ac.client.Connect() +} + +func (ac *AsteriskCollector) Collect() (*PBXResult, error) { + if !ac.client.IsConnected() { + return nil, fmt.Errorf("ami не подключён") + } + + result := &PBXResult{Status: "ok"} + + // active_calls: CoreShowChannels определяет количество активных каналов + if active, err := ac.coreShowChannelsCount(); err == nil { + result.ActiveCalls = active + } + + // max_calls: CoreSettings → CoreMaxCalls + if max, err := ac.coreSettingsInt("CoreMaxCalls"); err == nil { + result.MaxCalls = max + } + + // idle_cpu: Asterisk не даёт FS-шного idle_cpu, используем системный + result.IdleCPU = 0 + + return result, nil +} + +func (ac *AsteriskCollector) ListenHangups(onHangup func(success bool)) { + // Подписываемся на события Hangup через AMI + ac.client.Subscribe("call") + + go func() { + for evt := range ac.client.Events() { + if evt.Headers["Event"] == "Hangup" { + cause := evt.Headers["Cause"] + // Считаем успешным только NORMAL_CLEARING (код 16) + success := cause == "16" || strings.Contains(cause, "NORMAL_CLEARING") + onHangup(success) + } + } + }() +} + +func (ac *AsteriskCollector) Close() { + ac.client.Disconnect() +} + +// --- helpers --- + +// coreShowChannelsCount получает количество активных каналов. +func (ac *AsteriskCollector) coreShowChannelsCount() (int, error) { + cmd := "Action: Command\r\nCommand: core show channels count\r\n" + headers, body, err := ac.client.SendCommandBody(cmd) + if err != nil { + return 0, err + } + _ = headers + + // Ответ выглядит как: "1 active call" или "5 active calls" + body = strings.TrimSpace(body) + fields := strings.Fields(body) + if len(fields) >= 3 { + // Первое слово — число + n, err := strconv.Atoi(fields[0]) + if err == nil { + return n, nil + } + } + return 0, fmt.Errorf("не удалось распарсить количество каналов из: %s", body) +} + +// coreSettingsInt получает целочисленное значение из CoreSettings. +func (ac *AsteriskCollector) coreSettingsInt(key string) (int, error) { + headers, _, err := ac.client.SendCommandBody("Action: CoreSettings\r\n") + if err != nil { + return 0, err + } + val := headers[key] + if val == "" { + return 0, fmt.Errorf("поле %s не найдено", key) + } + return strconv.Atoi(strings.TrimSpace(val)) +} diff --git a/cmd/pulse-lets-go-agent/agent/collector_freeswitch.go b/cmd/pulse-lets-go-agent/agent/collector_freeswitch.go new file mode 100644 index 0000000..8971812 --- /dev/null +++ b/cmd/pulse-lets-go-agent/agent/collector_freeswitch.go @@ -0,0 +1,134 @@ +package agent + +import ( + "context" + "fmt" + "strconv" + "strings" + "time" + + "github.com/pulse-lets-go/internal/esl" +) + +// FSCollector собирает метрики с FreeSWITCH через ESL. +type FSCollector struct { + client *esl.Client +} + +// NewFSCollector создаёт FreeSWITCH-коллектор. +func NewFSCollector(host string, port int, password string) *FSCollector { + return &FSCollector{ + client: esl.NewClient(host, port, password), + } +} + +func (fc *FSCollector) Type() string { return "freeswitch" } + +func (fc *FSCollector) Connect() error { + // ESL client подключается синхронно через tryConnect + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + fc.client.ConnectWithRetry() + + // Ждём подключения с таймаутом + for !fc.client.IsConnected() { + select { + case <-ctx.Done(): + return fmt.Errorf("esl connect timeout") + case <-time.After(100 * time.Millisecond): + // Выходим из цикла после ConnectWithRetry + return nil + } + } + return nil +} + +func (fc *FSCollector) Collect() (*PBXResult, error) { + if !fc.client.IsConnected() { + return nil, fmt.Errorf("esl не подключён") + } + + result := &PBXResult{Status: "ok"} + + // active_calls: api show channels count + if active, err := fc.apiInt("show channels count"); err == nil { + result.ActiveCalls = active + } + + // max_calls: api json status → max_sessions + if max, err := fc.apiJSONInt("json", "status", "max_sessions"); err == nil { + result.MaxCalls = max + } + + // idle_cpu: eval $${idle_cpu} — FS-специфичный + if idle, err := fc.apiFloat("eval $${idle_cpu}"); err == nil && idle > 0 { + result.IdleCPU = idle + } + + return result, nil +} + +func (fc *FSCollector) ListenHangups(onHangup func(success bool)) { + // Подписываемся на CHANNEL_HANGUP через ESL + fc.client.Subscribe("CHANNEL_HANGUP") + + go func() { + // Для FS используем обработку событий через eventsCh + // Упрощённо: не реализуем полноценную подписку на первом этапе + // failure rate остаётся на нуле + time.Sleep(1 * time.Second) + }() +} + +func (fc *FSCollector) Close() { + fc.client.Disconnect() +} + +// --- helpers --- + +func (fc *FSCollector) apiInt(cmd string) (int, error) { + headers, body, err := fc.client.Send("api " + cmd) + if err != nil { + return 0, err + } + _ = headers + val := strings.TrimSpace(body) + n, err := strconv.Atoi(val) + if err != nil { + // иногда ответ содержит текст помимо числа + n, _ = strconv.Atoi(strings.Fields(val)[0]) + } + return n, nil +} + +func (fc *FSCollector) apiFloat(cmd string) (float64, error) { + _, body, err := fc.client.Send("api " + cmd) + if err != nil { + return 0, err + } + val := strings.TrimSpace(body) + f, _ := strconv.ParseFloat(val, 64) + return f, nil +} + +// apiJSONInt парсит JSON-ответ от FS api: api json status → field1.field2 +func (fc *FSCollector) apiJSONInt(cmd string, field string, subfield string) (int, error) { + _, body, err := fc.client.Send("api " + cmd) + if err != nil { + return 0, err + } + // Быстрый парсинг: ищем "subfield": число + search := fmt.Sprintf(`"%s":`, subfield) + idx := strings.Index(body, search) + if idx < 0 { + return 0, fmt.Errorf("поле %s не найдено в ответе", subfield) + } + rest := body[idx+len(search):] + rest = strings.TrimLeft(rest, " \t") + end := strings.IndexAny(rest, ",\n\r}") + if end > 0 { + rest = rest[:end] + } + return strconv.Atoi(strings.TrimSpace(rest)) +} diff --git a/cmd/pulse-lets-go-agent/agent/config.go b/cmd/pulse-lets-go-agent/agent/config.go new file mode 100644 index 0000000..deeb23c --- /dev/null +++ b/cmd/pulse-lets-go-agent/agent/config.go @@ -0,0 +1,124 @@ +// Package agent реализует агент сбора метрик для pulse-lets-go. +// Агент работает на каждой PBX-ноде (FreeSWITCH или Asterisk), +// собирает метрики и публикует их в NATS каждые 5 секунд. +// +// Поддерживаемые типы PBX: +// - "freeswitch" — сбор через ESL (Event Socket Library) +// - "asterisk" — сбор через AMI (Asterisk Manager Interface) +// +// Пример agent.json: +// +// { +// "node_id": "uc06", +// "type": "asterisk", +// "nats_url": "nats://balancer.lan:4222", +// "sip_gateway": "sip:uc-pbx.lan:5060", +// "interval_sec": 5, +// "max_calls": 150, +// "ami": { "host": "127.0.0.1", "port": 6154, "username": "admin", "password": "changeme" } +// } +package agent + +import ( + "encoding/json" + "fmt" + "os" +) + +// Config — конфигурация агента (agent.json). +type Config struct { + NodeID string `json:"node_id"` // уникальный ID ноды (uc06, ses-sip) + Type string `json:"type"` // "freeswitch" или "asterisk" + NatsURL string `json:"nats_url"` // NATS URL + NatsUser string `json:"nats_user,omitempty"` + NatsPassword string `json:"nats_password,omitempty"` + IntervalSec int `json:"interval_sec"` // интервал сбора (default: 5) + MaxCalls int `json:"max_calls"` // ёмкость ноды (default: 250) + SIPGateway string `json:"sip_gateway"` // SIP-адрес для auto-транка + SIPGatewayAuto bool `json:"sip_gateway_auto"` // авто-определить из PBX + FailureWindow int `json:"failure_window"` // окно для call_failure_rate (default: 1000) + ESL ESLCfg `json:"esl,omitempty"` + AMI AMICfg `json:"ami,omitempty"` +} + +// ESLCfg — настройки подключения к FreeSWITCH ESL. +type ESLCfg struct { + Host string `json:"host"` + Port int `json:"port"` + Password string `json:"password"` +} + +// AMICfg — настройки подключения к Asterisk AMI. +type AMICfg struct { + Host string `json:"host"` + Port int `json:"port"` + Username string `json:"username"` + Password string `json:"password"` +} + +// LoadConfig загружает конфигурацию агента из JSON-файла. +func LoadConfig(path string) (*Config, error) { + data, err := os.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("чтение %s: %w", path, err) + } + + var cfg Config + if err := json.Unmarshal(data, &cfg); err != nil { + return nil, fmt.Errorf("парсинг %s: %w", path, err) + } + + // Defaults + if cfg.IntervalSec <= 0 { + cfg.IntervalSec = 5 + } + if cfg.MaxCalls <= 0 { + cfg.MaxCalls = 250 + } + if cfg.FailureWindow <= 0 { + cfg.FailureWindow = 1000 + } + if cfg.ESL.Port == 0 { + cfg.ESL.Port = 8021 + } + if cfg.AMI.Port == 0 { + cfg.AMI.Port = 5038 + } + + return &cfg, nil +} + +// Validate проверяет корректность конфигурации. +func (c *Config) Validate() error { + if c.NodeID == "" { + return fmt.Errorf("node_id обязателен") + } + if c.NatsURL == "" { + return fmt.Errorf("nats_url обязателен") + } + switch c.Type { + case "freeswitch": + if c.ESL.Host == "" { + return fmt.Errorf("esl.host обязателен для type=freeswitch") + } + case "asterisk": + if c.AMI.Host == "" { + return fmt.Errorf("ami.host обязателен для type=asterisk") + } + default: + return fmt.Errorf("type должен быть freeswitch или asterisk, получен: %s", c.Type) + } + return nil +} + +// GetSIPGateway возвращает SIP-адрес для авто-транка. +// Если SIPGatewayAuto = true — вернётся "" (collector определит сам). +func (c *Config) GetSIPGateway(autoDetected string) string { + if c.SIPGateway != "" { + return c.SIPGateway + } + if c.SIPGatewayAuto { + return autoDetected + } + return "" +} diff --git a/cmd/pulse-lets-go-agent/agent/failtrack.go b/cmd/pulse-lets-go-agent/agent/failtrack.go new file mode 100644 index 0000000..5ac44e2 --- /dev/null +++ b/cmd/pulse-lets-go-agent/agent/failtrack.go @@ -0,0 +1,52 @@ +package agent + +import "sync" + +// FailureTracker — скользящее окно для расчёта call_failure_rate. +// Хранит последние N результатов звонков (success/failure), +// вычисляет процент неудачных звонков в реальном времени. +type FailureTracker struct { + mu sync.Mutex + window []bool // true=success, false=failure + position int // позиция следующей записи + size int // текущий размер окна + capacity int // максимальный размер окна +} + +// NewFailureTracker создаёт трекер с окном заданного размера. +func NewFailureTracker(capacity int) *FailureTracker { + return &FailureTracker{ + window: make([]bool, capacity), + capacity: capacity, + } +} + +// Record записывает результат звонка (true=success, false=failure). +func (ft *FailureTracker) Record(success bool) { + ft.mu.Lock() + defer ft.mu.Unlock() + + ft.window[ft.position] = success + ft.position = (ft.position + 1) % ft.capacity + if ft.size < ft.capacity { + ft.size++ + } +} + +// Rate возвращает процент неудачных звонков (0.0..100.0). +func (ft *FailureTracker) Rate() float64 { + ft.mu.Lock() + defer ft.mu.Unlock() + + if ft.size == 0 { + return 0 + } + + failures := 0 + for i := 0; i < ft.size; i++ { + if !ft.window[i] { + failures++ + } + } + return float64(failures) / float64(ft.size) * 100.0 +} diff --git a/cmd/pulse-lets-go-agent/agent/publisher.go b/cmd/pulse-lets-go-agent/agent/publisher.go new file mode 100644 index 0000000..c6cf44b --- /dev/null +++ b/cmd/pulse-lets-go-agent/agent/publisher.go @@ -0,0 +1,57 @@ +package agent + +import ( + "encoding/json" + "fmt" + "log" + + natsgo "github.com/nats-io/nats.go" + + "github.com/pulse-lets-go/internal/models" +) + +// Publisher публикует метрики в NATS. +type Publisher struct { + nc *natsgo.Conn + nodeID string +} + +// NewPublisher создаёт NATS publisher. +func NewPublisher(natsURL, user, password, nodeID string) (*Publisher, 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) + } + + log.Printf("[publisher] NATS подключён к %s", natsURL) + return &Publisher{nc: nc, nodeID: nodeID}, nil +} + +// Publish отправляет метрику в NATS: pulse.metrics.. +func (p *Publisher) Publish(m *models.NodeMetric) error { + subject := fmt.Sprintf("pulse.metrics.%s", p.nodeID) + data, err := json.Marshal(m) + if err != nil { + return fmt.Errorf("marshal metric: %w", err) + } + + if err := p.nc.Publish(subject, data); err != nil { + return fmt.Errorf("publish: %w", err) + } + return nil +} + +// Close закрывает NATS-соединение. +func (p *Publisher) Close() { + if p.nc != nil { + p.nc.Close() + } +} diff --git a/cmd/pulse-lets-go-agent/agent/system.go b/cmd/pulse-lets-go-agent/agent/system.go new file mode 100644 index 0000000..3a615e4 --- /dev/null +++ b/cmd/pulse-lets-go-agent/agent/system.go @@ -0,0 +1,84 @@ +package agent + +import ( + "bufio" + "os" + "strconv" + "strings" + "time" +) + +// nowTS возвращает текущий unix timestamp. +func nowTS() int64 { + return time.Now().Unix() +} + +// SystemStats содержит системные метрики (CPU, load). +type SystemStats struct { + Load1 float64 // load average 1m + Load5 float64 // load average 5m + Load15 float64 // load average 15m + IdleCPU float64 // процент простоя CPU +} + +// CollectSystem собирает системные метрики из /proc. +func CollectSystem() *SystemStats { + load1, load5, load15 := loadAvg() + return &SystemStats{ + Load1: load1, + Load5: load5, + Load15: load15, + IdleCPU: cpuIdle(), + } +} + +// loadAvg читает /proc/loadavg. +func loadAvg() (float64, float64, float64) { + data, err := os.ReadFile("/proc/loadavg") + if err != nil { + return 0, 0, 0 + } + parts := strings.Fields(string(data)) + if len(parts) < 3 { + return 0, 0, 0 + } + load1, _ := strconv.ParseFloat(parts[0], 64) + load5, _ := strconv.ParseFloat(parts[1], 64) + load15, _ := strconv.ParseFloat(parts[2], 64) + return load1, load5, load15 +} + +// cpuIdle вычисляет процент простоя CPU из /proc/stat. +func cpuIdle() float64 { + file, err := os.Open("/proc/stat") + if err != nil { + return 0 + } + defer file.Close() + + scanner := bufio.NewScanner(file) + for scanner.Scan() { + line := scanner.Text() + if !strings.HasPrefix(line, "cpu ") { + continue + } + fields := strings.Fields(line) + if len(fields) < 8 { + return 0 + } + // cpu user nice system idle iowait irq softirq steal ... + var total, idle float64 + for i, f := range fields[1:] { + val, _ := strconv.ParseFloat(f, 64) + total += val + if i == 3 || i == 4 { // idle + iowait + idle += val + } + } + if total > 0 { + return (idle / total) * 100.0 + } + return 0 + } + return 0 +} diff --git a/cmd/pulse-lets-go-agent/main.go b/cmd/pulse-lets-go-agent/main.go new file mode 100644 index 0000000..67c5ca7 --- /dev/null +++ b/cmd/pulse-lets-go-agent/main.go @@ -0,0 +1,44 @@ +// pulse-lets-go-agent — агент сбора метрик для pulse-lets-go. +// Работает на каждой PBX-ноде (FreeSWITCH или Asterisk), +// собирает метрики и публикует их в NATS каждые 5 секунд. +// +// Использование: +// +// pulse-lets-go-agent -config /etc/pulse-lets-go-agent/agent.json +// +// Пример agent.json: +// +// { +// "node_id": "uc06", +// "type": "asterisk", +// "nats_url": "nats://balancer.lan:4222", +// "sip_gateway": "sip:uc-pbx.lan:5060", +// "interval_sec": 5, +// "max_calls": 150, +// "ami": { "host": "127.0.0.1", "port": 6154, "username": "admin", "password": "changeme" } +// } +package main + +import ( + "flag" + "log" + + "github.com/pulse-lets-go/cmd/pulse-lets-go-agent/agent" +) + +func main() { + configPath := flag.String("config", "agent.json", "путь к agent.json") + flag.Parse() + + cfg, err := agent.LoadConfig(*configPath) + if err != nil { + log.Fatalf("ошибка загрузки конфига: %v", err) + } + + a, err := agent.NewAgent(cfg) + if err != nil { + log.Fatalf("ошибка создания агента: %v", err) + } + + a.Start() +} diff --git a/contrib/agent-sessip.json b/contrib/agent-sessip.json new file mode 100644 index 0000000..d929352 --- /dev/null +++ b/contrib/agent-sessip.json @@ -0,0 +1,15 @@ +{ + "node_id": "ses-sip", + "type": "freeswitch", + "nats_url": "nats://balancer.lan:4222", + "interval_sec": 5, + "max_calls": 250, + "sip_gateway": "sip:ses-pbx.lan:5060", + "sip_gateway_auto": false, + "failure_window": 1000, + "esl": { + "host": "127.0.0.1", + "port": 8021, + "password": "ClueCon" + } +} diff --git a/contrib/agent-uc.json b/contrib/agent-uc.json new file mode 100644 index 0000000..3256809 --- /dev/null +++ b/contrib/agent-uc.json @@ -0,0 +1,16 @@ +{ + "node_id": "uc06", + "type": "asterisk", + "nats_url": "nats://balancer.lan:4222", + "interval_sec": 5, + "max_calls": 150, + "sip_gateway": "sip:uc-pbx.lan:5060", + "sip_gateway_auto": false, + "failure_window": 1000, + "ami": { + "host": "127.0.0.1", + "port": 6154, + "username": "admin", + "password": "changeme" + } +} diff --git a/internal/ami/client.go b/internal/ami/client.go new file mode 100644 index 0000000..77feb58 --- /dev/null +++ b/internal/ami/client.go @@ -0,0 +1,293 @@ +// Package ami реализует клиент Asterisk Manager Interface (AMI). +// Использует сырой TCP — протокол AMI текстовый (Key: Value\r\n), +// аналогичен ESL по простоте, не нужна внешняя библиотека. +// +// Клиент обеспечивает: +// - Подключение и аутентификацию (Action: Login) +// - Синхронные команды (Action: Command) +// - Подписку на события (Action: Events) +// - Автоматический реконнект с exponential backoff +// - Потокобезопасность (mutex на отправку команд) +package ami + +import ( + "bufio" + "fmt" + "io" + "log" + "net" + "strings" + "sync" + "sync/atomic" + "time" +) + +// Event — событие, полученное от Asterisk AMI. +type Event struct { + Headers map[string]string // все заголовки события (Event:, Channel:, Cause:, ...) +} + +// Client — AMI-клиент для взаимодействия с Asterisk. +type Client struct { + host string + port int + username string + password string + + conn net.Conn + reader *bufio.Reader + sendMu sync.Mutex // защищает запись в сокет + + connected atomic.Bool + eventsCh chan *Event + closeCh chan struct{} + + shouldReconn atomic.Bool + backoff time.Duration +} + +// NewClient создаёт нового AMI-клиента. +func NewClient(host string, port int, username, password string) *Client { + return &Client{ + host: host, + port: port, + username: username, + password: password, + eventsCh: make(chan *Event, 100), + closeCh: make(chan struct{}), + backoff: 1 * time.Second, + } +} + +// Connect подключается к Asterisk AMI и выполняет аутентификацию. +func (c *Client) Connect() error { + c.shouldReconn.Store(true) + + 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 { + return fmt.Errorf("ami dial %s: %w", addr, err) + } + c.conn = conn + c.reader = bufio.NewReader(conn) + + // 1. Читаем приветственное сообщение Asterisk (одна строка, без пустой строки) + line, err := c.reader.ReadString('\n') + if err != nil { + conn.Close() + return fmt.Errorf("чтение приветствия AMI: %w", err) + } + _ = line // "Asterisk Call Manager/5.0.5\r\n" + + // 2. Отправляем логин + loginCmd := fmt.Sprintf("Action: Login\r\nUsername: %s\r\nSecret: %s\r\n\r\n", c.username, c.password) + if _, err := conn.Write([]byte(loginCmd)); err != nil { + conn.Close() + return fmt.Errorf("отправка Login: %w", err) + } + + // 3. Читаем ответ + headers, err := c.readResponse() + if err != nil { + conn.Close() + return fmt.Errorf("чтение ответа Login: %w", err) + } + if resp := headers["Response"]; resp != "Success" { + conn.Close() + return fmt.Errorf("ami auth rejected: %s", headers["Message"]) + } + + c.connected.Store(true) + c.backoff = 1 * time.Second + + log.Printf("[ami] подключён к %s", addr) + return nil +} + +// Disconnect отключает клиент. +func (c *Client) Disconnect() { + c.shouldReconn.Store(false) + c.connected.Store(false) + if c.conn != nil { + c.conn.Close() + c.conn = nil + } + select { + case <-c.closeCh: + default: + close(c.closeCh) + } +} + +// IsConnected возвращает статус подключения. +func (c *Client) IsConnected() bool { + return c.connected.Load() +} + +// SendCommand отправляет синхронную команду и возвращает заголовки ответа. +func (c *Client) SendCommand(cmd string) (map[string]string, error) { + c.sendMu.Lock() + defer c.sendMu.Unlock() + + if !c.connected.Load() || c.conn == nil { + return nil, fmt.Errorf("ami не подключён") + } + + fullCmd := cmd + "\r\n\r\n" + if _, err := c.conn.Write([]byte(fullCmd)); err != nil { + return nil, fmt.Errorf("ami send: %w", err) + } + + return c.readResponse() +} + +// SendCommandBody отправляет команду и возвращает заголовки + тело ответа. +// Нужно для команд, возвращающих многострочный вывод (напр. Command). +func (c *Client) SendCommandBody(cmd string) (map[string]string, string, error) { + c.sendMu.Lock() + defer c.sendMu.Unlock() + + if !c.connected.Load() || c.conn == nil { + return nil, "", fmt.Errorf("ami не подключён") + } + + fullCmd := cmd + "\r\n\r\n" + if _, err := c.conn.Write([]byte(fullCmd)); err != nil { + return nil, "", fmt.Errorf("ami send: %w", err) + } + + // Читаем заголовки до пустой строки + headers, err := c.readResponse() + if err != nil { + return nil, "", fmt.Errorf("ami read headers: %w", err) + } + + // Если это ответ на Command — читаем тело до --END COMMAND-- + if headers["Response"] == "Follows" { + var bodyLines []string + for { + line, err := c.reader.ReadString('\n') + if err != nil { + break + } + line = strings.TrimRight(line, "\r\n") + if line == "--END COMMAND--" { + break + } + bodyLines = append(bodyLines, line) + } + return headers, strings.Join(bodyLines, "\n"), nil + } + + return headers, "", nil +} + +// Subscribe подписывается на события AMI. +func (c *Client) Subscribe(eventMask string) error { + cmd := fmt.Sprintf("Action: Events\r\nEventMask: %s\r\n", eventMask) + _, _, err := c.SendCommandBody(cmd) + if err != nil { + return fmt.Errorf("подписка на события: %w", err) + } + log.Printf("[ami] подписан на события: %s", eventMask) + return nil +} + +// Events возвращает канал с входящими событиями. +func (c *Client) Events() <-chan *Event { + return c.eventsCh +} + +// StartReadingEvents запускает горутину чтения асинхронных событий. +func (c *Client) StartReadingEvents() { + go c.readEventsLoop() +} + +// ConnectWithRetry подключается в фоне с бесконечными попытками. +func (c *Client) ConnectWithRetry() { + c.shouldReconn.Store(true) + for { + if err := c.Connect(); err == nil { + return + } + if !c.shouldReconn.Load() { + return + } + log.Printf("[ami] ошибка подключения, повтор через %v", c.backoff) + time.Sleep(c.backoff) + c.backoff *= 2 + if c.backoff > 60*time.Second { + c.backoff = 60 * time.Second + } + } +} + +// --- Приватные методы --- + +// readResponse читает один AMI-ответ: заголовки Key: Value до пустой строки. +func (c *Client) readResponse() (map[string]string, error) { + headers := make(map[string]string) + + for { + line, err := c.reader.ReadString('\n') + if err != nil { + if err == io.EOF { + return headers, nil + } + return headers, err + } + line = strings.TrimRight(line, "\r\n") + + if line == "" { + return headers, nil + } + + parts := strings.SplitN(line, ": ", 2) + if len(parts) == 2 { + headers[parts[0]] = parts[1] + } + } +} + +// readEventsLoop читает асинхронные события от Asterisk. +func (c *Client) readEventsLoop() { + for { + select { + case <-c.closeCh: + return + default: + } + + if !c.connected.Load() || c.conn == nil { + time.Sleep(500 * time.Millisecond) + continue + } + + headers, err := c.readResponse() + if err != nil { + if err == io.EOF || strings.Contains(err.Error(), "use of closed network connection") { + c.connected.Store(false) + log.Printf("[ami] соединение разорвано") + if c.shouldReconn.Load() { + go c.ConnectWithRetry() + } + return + } + time.Sleep(100 * time.Millisecond) + continue + } + + eventName := headers["Event"] + if eventName == "" { + continue + } + + evt := &Event{Headers: headers} + select { + case c.eventsCh <- evt: + default: + log.Printf("[ami] канал событий переполнен, пропускаем %s", eventName) + } + } +}