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 }