From f3c635e233a027e2b7886d3390c74778e86a1016 Mon Sep 17 00:00:00 2001 From: "n.shabanov" Date: Wed, 14 Jan 2026 13:12:11 +0300 Subject: [PATCH 1/2] feat: implement prometheus metrics --- Dockerfile | 2 + README.md | 8 +- cmd/main.go | 20 +- go.mod | 15 +- go.sum | 20 ++ internal/lifecycle/manager.go | 20 +- internal/lifecycle/manager_test.go | 46 +-- internal/metrics/metrics.go | 360 +++----------------- internal/metrics/metrics_test.go | 195 ----------- internal/monitoring/monitoring.go | 70 ---- internal/monitoring/monitoring_test.go | 87 ----- internal/network/tcp_pool.go | 6 - internal/pipeline/factory/factory.go | 15 +- internal/pipeline/factory/factory_test.go | 15 +- internal/pipeline/stages/common.go | 17 - internal/pipeline/stages/generation.go | 24 +- internal/pipeline/stages/generation_test.go | 36 -- internal/pipeline/stages/sending.go | 64 +--- internal/pipeline/stages/sending_test.go | 96 +----- internal/workers/workerpool_test.go | 23 +- internal/workers/workers.go | 137 ++------ internal/workers/workers_bench_test.go | 6 +- 22 files changed, 170 insertions(+), 1112 deletions(-) delete mode 100644 internal/metrics/metrics_test.go delete mode 100644 internal/monitoring/monitoring.go delete mode 100644 internal/monitoring/monitoring_test.go diff --git a/Dockerfile b/Dockerfile index c283b18..bd78909 100644 --- a/Dockerfile +++ b/Dockerfile @@ -23,4 +23,6 @@ COPY --from=builder /build/pulsar /app/pulsar LABEL maintainer="nik.shabanov2018@yandex.com" LABEL description="Pulsar: CLI tool for high EPS event generation" +EXPOSE 9090 + ENTRYPOINT ["/app/pulsar"] diff --git a/README.md b/README.md index 6b33093..a98299b 100644 --- a/README.md +++ b/README.md @@ -13,7 +13,7 @@ Generate realistic logs and network flows at scale up to **150000 events per sec - **Lock-free pipeline engine**: Zero-lock worker pools with atomic queues for sustained 150K+ EPS - **CLI-first**: Optionla config files - everything tunable via flags or env vars. Perfect for scripts and CI -- **Built-in observability**: Metrics for throughput, drops, and latency out of the box +- **Built-in observability**: Prometheus metrics for events, workers, queue depth and Go runtime. Exposed on `:9090/metrics` ## Known problems @@ -21,9 +21,8 @@ Perfect for scripts and CI - TCP connection pool is basic (no smart retries; errors on shutdown) - Event types and data are hardcoded - no support for custom format - Potential data loss on shutdown under high load (visible with -race flag) -- Incomplete metrics (network/processing stats not fully implemented) - Suboptimal memory usage at high EPS (OOM risk on constrained systems) -- No build-time information (`-v` shows placeholder) + ## Usage ### Run from source (for development) @@ -50,9 +49,8 @@ go build -o pulsar . | `-v, --version` | - | Show version information | - | | `-h, --help` | - | Show this help | - | -### Docker -> ⚠️**Wrk in progress** docker-compose is just an exampl and not optimal. Public image and stable version coming soon. +### Docker You can run multiple generators via `docker-compose.yml` (netflow + syslog, as example) Build and run both generators for 5 minutes: diff --git a/cmd/main.go b/cmd/main.go index 8b7a777..a971d3e 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -2,17 +2,16 @@ package main import ( "fmt" + "net/http" "os" "strings" - "time" "github.com/NGRsoftlab/pulsar/internal/cli" "github.com/NGRsoftlab/pulsar/internal/config" "github.com/NGRsoftlab/pulsar/internal/lifecycle" "github.com/NGRsoftlab/pulsar/internal/logger" - "github.com/NGRsoftlab/pulsar/internal/metrics" - "github.com/NGRsoftlab/pulsar/internal/monitoring" "github.com/NGRsoftlab/pulsar/internal/pipeline/factory" + "github.com/prometheus/client_golang/prometheus/promhttp" ) var ( @@ -59,18 +58,23 @@ func run() error { fmt.Println("Configuration loaded successfully!") fmt.Println("Next step: create pipeline...") + go func() { + http.Handle("/metrics", promhttp.Handler()) + err := http.ListenAndServe(":9090", nil) + if err != nil && err != http.ErrServerClosed { + log.Error("Metrics server failed: %v", err) + } + }() + // Создаем pipeline - factory := factory.NewPipelineFactory(cfg, metrics.GetGlobalMetrics()) + factory := factory.NewPipelineFactory(cfg) pipeline, err := factory.CreatePipeline() if err != nil { return fmt.Errorf("failed to build pipeline: %v", err) } - // Создаем monitoring - monitoring := monitoring.NewMonitor(10*time.Second, log) - // Создаем и запускаем приложение - manager := lifecycle.NewManager(pipeline, monitoring, log) + manager := lifecycle.NewManager(pipeline, log) if err := manager.Run(cfg.Generator.Duration); err != nil { return fmt.Errorf("application failed: %v", err) diff --git a/go.mod b/go.mod index 6ef6287..4561154 100644 --- a/go.mod +++ b/go.mod @@ -1,4 +1,4 @@ -module github.com/NGRsoftlab/pulsar +module github.com/NGRsoftlab/pulsar go 1.24.0 @@ -7,6 +7,19 @@ require ( gopkg.in/yaml.v3 v3.0.1 ) +require ( + github.com/beorn7/perks v1.0.1 // indirect + github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect + github.com/prometheus/client_golang v1.23.2 // indirect + github.com/prometheus/client_model v0.6.2 // indirect + github.com/prometheus/common v0.66.1 // indirect + github.com/prometheus/procfs v0.16.1 // indirect + go.yaml.in/yaml/v2 v2.4.2 // indirect + golang.org/x/sys v0.35.0 // indirect + google.golang.org/protobuf v1.36.8 // indirect +) + require ( github.com/davecgh/go-spew v1.1.1 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect diff --git a/go.sum b/go.sum index c7f3ba3..a25e43b 100644 --- a/go.sum +++ b/go.sum @@ -1,13 +1,33 @@ +github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= +github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/prometheus/client_golang v1.23.2 h1:Je96obch5RDVy3FDMndoUsjAhG5Edi49h0RJWRi/o0o= +github.com/prometheus/client_golang v1.23.2/go.mod h1:Tb1a6LWHB3/SPIzCoaDXI4I8UHKeFTEQ1YCr+0Gyqmg= +github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk= +github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE= +github.com/prometheus/common v0.66.1 h1:h5E0h5/Y8niHc5DlaLlWLArTQI7tMrsfQjHV+d9ZoGs= +github.com/prometheus/common v0.66.1/go.mod h1:gcaUsgf3KfRSwHY4dIMXLPV0K/Wg1oZ8+SbZk/HH/dA= +github.com/prometheus/procfs v0.16.1 h1:hZ15bTNuirocR6u0JZ6BAHHmwS1p8B4P6MRqxtzMyRg= +github.com/prometheus/procfs v0.16.1/go.mod h1:teAbpZRB1iIAJYREa1LsoWUXykVXA1KlTmWl8x/U+Is= github.com/stretchr/objx v0.5.2 h1:xuMeJ0Sdp5ZMRXx/aWO6RZxdr3beISkG5/G/aIRr3pY= github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI= +go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU= +golang.org/x/sys v0.35.0 h1:vz1N37gP5bs89s7He8XuIYXpyY0+QlsKmzipCbUtyxI= +golang.org/x/sys v0.35.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= golang.org/x/time v0.13.0 h1:eUlYslOIt32DgYD6utsuUeHs4d7AsEYLuIAdg7FlYgI= golang.org/x/time v0.13.0/go.mod h1:eL/Oa2bBBK0TkX57Fyni+NgnyQQN4LitPmob2Hjnqw4= +google.golang.org/protobuf v1.36.8 h1:xHScyCOEuuwZEc6UtSOvPbAT4zRh0xcNRYekJwfqyMc= +google.golang.org/protobuf v1.36.8/go.mod h1:fuxRtAxBytpl4zzqUh6/eyUujkJdNiuEkXntxiD/uRU= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= diff --git a/internal/lifecycle/manager.go b/internal/lifecycle/manager.go index 77c2fbb..de15bdb 100644 --- a/internal/lifecycle/manager.go +++ b/internal/lifecycle/manager.go @@ -17,21 +17,15 @@ type pipeline interface { Stop() error } -type monitor interface { - Start(ctx context.Context) - Stop() error -} - type Manager struct { pipeline pipeline - monitor monitor logger logger.Logger cancel context.CancelFunc mu sync.Mutex } -func NewManager(p pipeline, m monitor, log logger.Logger) *Manager { - return &Manager{pipeline: p, monitor: m, logger: log} +func NewManager(p pipeline, log logger.Logger) *Manager { + return &Manager{pipeline: p, logger: log} } func (m *Manager) Run(duration time.Duration) error { @@ -49,8 +43,6 @@ func (m *Manager) Run(duration time.Duration) error { errCh := make(chan error, 1) - go m.monitor.Start(ctx) - m.logger.Info("Starting pipeline...") go func() { @@ -77,14 +69,6 @@ func (m *Manager) Run(duration time.Duration) error { } } - m.logger.Info("Stopping monitor...") - if err := m.monitor.Stop(); err != nil { - m.logger.Error("Error stopping monitor: %v", err) - if runErr == nil { - runErr = err - } - } - return runErr } diff --git a/internal/lifecycle/manager_test.go b/internal/lifecycle/manager_test.go index b0641ae..42677a7 100644 --- a/internal/lifecycle/manager_test.go +++ b/internal/lifecycle/manager_test.go @@ -119,10 +119,9 @@ func (m *mockMonitorWithError) WasStopped() bool { func TestManager_Run_Success(t *testing.T) { p := &mockPipeline{} - m := &mockMonitor{} log := logger.NewStdLogger() - manager := NewManager(p, m, log) + manager := NewManager(p, log) // Запускаем на 10 мс — этого достаточно для graceful shutdown err := manager.Run(10 * time.Millisecond) @@ -139,21 +138,14 @@ func TestManager_Run_Success(t *testing.T) { if !p.WasStopped() { t.Error("pipeline.Stop was not called") } - if !m.WasStarted() { - t.Error("monitor.Start was not called") - } - if !m.WasStopped() { - t.Error("monitor.Stop was not called") - } } func TestManager_Run_PipelineError(t *testing.T) { expectedErr := errors.New("pipeline failed") p := &mockPipeline{startErr: expectedErr} - m := &mockMonitor{} log := logger.NewStdLogger() - manager := NewManager(p, m, log) + manager := NewManager(p, log) err := manager.Run(0) if err == nil || err.Error() != expectedErr.Error() { @@ -163,17 +155,13 @@ func TestManager_Run_PipelineError(t *testing.T) { if !p.WasStopped() { t.Error("pipeline.Stop should be called even on error") } - if !m.WasStopped() { - t.Error("monitor.Stop should be called even on error") - } } func TestManager_Run_DurationTimeout(t *testing.T) { p := &mockPipeline{} - m := &mockMonitor{} log := logger.NewStdLogger() - manager := NewManager(p, m, log) + manager := NewManager(p, log) err := manager.Run(10 * time.Millisecond) if err != nil { @@ -188,17 +176,13 @@ func TestManager_Run_DurationTimeout(t *testing.T) { if !p.WasStopped() { t.Error("pipeline should have been stopped after timeout") } - if !m.WasStopped() { - t.Error("monitor should have been stopped after timeout") - } } func TestManager_Stop_ExternalCancel(t *testing.T) { p := &mockPipeline{} - m := &mockMonitor{} log := logger.NewStdLogger() - manager := NewManager(p, m, log) + manager := NewManager(p, log) var runErr error done := make(chan struct{}) @@ -220,38 +204,18 @@ func TestManager_Stop_ExternalCancel(t *testing.T) { if !p.WasStopped() { t.Error("pipeline should be stopped after external Stop()") } - if !m.WasStopped() { - t.Error("monitor should be stopped after external Stop()") - } } func TestManager_Run_PipelineStopErrorPriority(t *testing.T) { pipelineStopErr := errors.New("pipeline stop failed") - monitorStopErr := errors.New("monitor stop failed") p := &mockPipeline{stopErr: pipelineStopErr} - mon := &mockMonitorWithError{stopErr: monitorStopErr} log := logger.NewStdLogger() - manager := NewManager(p, mon, log) + manager := NewManager(p, log) err := manager.Run(50 * time.Millisecond) if err == nil || err.Error() != pipelineStopErr.Error() { t.Errorf("expected pipeline stop error, got %v", err) } } - -func TestManager_Run_MonitorStopErrorWhenPipelineOK(t *testing.T) { - monitorStopErr := errors.New("monitor stop failed") - - p := &mockPipeline{} - m := &mockMonitorWithError{stopErr: monitorStopErr} - log := logger.NewStdLogger() - - manager := NewManager(p, m, log) - - err := manager.Run(50 * time.Millisecond) - if err == nil || err.Error() != monitorStopErr.Error() { - t.Errorf("expected monitor stop error, got %v", err) - } -} diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 6cf2b70..96060da 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -1,327 +1,43 @@ -// Package metrics сбор и подсчет метрик пайплайна +// Package metrics exports Prometheus metrics for the Pulsar load generator package metrics import ( - "fmt" - "runtime" - "sync/atomic" - "time" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" ) -// PerformanceMetrics - полная система метрик для UEBA генератора -type PerformanceMetrics struct { - // === СЧЕТЧИКИ СОБЫТИЙ === - generatedEvents uint64 // Сколько событий сгенерировано - sentEvents uint64 // Сколько событий отправлено - failedEvents uint64 // Сколько событий завершилось ошибкой - droppedEvents uint64 // Сколько событий отброшено (queue full) - - // === СЕТЕВЫЕ МЕТРИКИ === - networkConnections uint64 // Количество активных соединений - networkReconnects uint64 // Количество переподключений - networkTimeouts uint64 // Количество таймаутов - - // === WORKER POOL МЕТРИКИ === - activeWorkers uint64 // Активные worker'ы - queuedJobs uint64 // Задач в очереди - completedJobs uint64 // Выполненных задач - rejectedJobs uint64 // Отклоненных задач (queue full) - - // === ВРЕМЕННЫЕ МЕТРИКИ === - startTime time.Time - lastResetTime time.Time - lastEpsCalculation int64 - lastGeneratedCount uint64 // НОВОЕ ПОЛЕ для расчета текущего EPS - - // === ПРОИЗВОДИТЕЛЬНОСТЬ === - totalProcessingTime uint64 // Общее время обработки в наносекундах - maxProcessingTime uint64 // Максимальное время обработки события - minProcessingTime uint64 // Минимальное время обработки события - - // === MEMORY METRICS === - // totalAllocations uint64 // Общее количество аллокаций - // currentMemoryUsage uint64 // Текущее потребление памяти -} - -func NewPerformanceMetrics() *PerformanceMetrics { - now := time.Now() - nowNano := time.Now().UnixNano() - return &PerformanceMetrics{ - startTime: now, - lastResetTime: now, - lastEpsCalculation: nowNano, - minProcessingTime: ^uint64(0), // Максимальное значение uint64 - } -} - -// === МЕТОДЫ ДЛЯ СОБЫТИЙ === - -func (m *PerformanceMetrics) IncrementGenerated() { - atomic.AddUint64(&m.generatedEvents, 1) -} - -func (m *PerformanceMetrics) IncrementSent() { - atomic.AddUint64(&m.sentEvents, 1) -} - -func (m *PerformanceMetrics) IncrementFailed() { - atomic.AddUint64(&m.failedEvents, 1) -} - -func (m *PerformanceMetrics) IncrementDropped() { - atomic.AddUint64(&m.droppedEvents, 1) -} - -// === МЕТОДЫ ДЛЯ СЕТИ === - -func (m *PerformanceMetrics) IncrementConnections() { - atomic.AddUint64(&m.networkConnections, 1) -} - -func (m *PerformanceMetrics) DecrementConnections() { - atomic.AddUint64(&m.networkConnections, ^uint64(0)) // -1 в unsigned арифметике -} - -func (m *PerformanceMetrics) IncrementReconnects() { - atomic.AddUint64(&m.networkReconnects, 1) -} - -func (m *PerformanceMetrics) IncrementTimeouts() { - atomic.AddUint64(&m.networkTimeouts, 1) -} - -// === МЕТОДЫ ДЛЯ WORKER POOL === - -func (m *PerformanceMetrics) IncrementActiveWorkers() { - atomic.AddUint64(&m.activeWorkers, 1) -} - -func (m *PerformanceMetrics) DecrementActiveWorkers() { - atomic.AddUint64(&m.activeWorkers, ^uint64(0)) -} - -func (m *PerformanceMetrics) SetQueuedJobs(count uint64) { - atomic.StoreUint64(&m.queuedJobs, count) -} - -func (m *PerformanceMetrics) IncrementCompletedJobs() { - atomic.AddUint64(&m.completedJobs, 1) -} - -func (m *PerformanceMetrics) IncrementRejectedJobs() { - atomic.AddUint64(&m.rejectedJobs, 1) -} - -// === МЕТОДЫ ДЛЯ ПРОИЗВОДИТЕЛЬНОСТИ === - -func (m *PerformanceMetrics) RecordProcessingTime(duration time.Duration) { - nanos := uint64(duration.Nanoseconds()) - - // Проверяем валидность времени - if nanos == 0 || duration < 0 { - return // Игнорируем некорректные значения - } - - // Обновляем общее время - atomic.AddUint64(&m.totalProcessingTime, nanos) - - // Обновляем максимум - for { - current := atomic.LoadUint64(&m.maxProcessingTime) - if nanos <= current { - break - } - if atomic.CompareAndSwapUint64(&m.maxProcessingTime, current, nanos) { - break - } - } - - for { - current := atomic.LoadUint64(&m.minProcessingTime) - // Если текущий минимум еще не установлен (максимальное значение uint64) - if current == ^uint64(0) { - if atomic.CompareAndSwapUint64(&m.minProcessingTime, current, nanos) { - break - } - continue - } - // Если новое значение меньше текущего минимума - if nanos >= current { - break - } - if atomic.CompareAndSwapUint64(&m.minProcessingTime, current, nanos) { - break - } - } -} - -// calculateCurrentEPS рассчитывает EPS за последний период -func (m *PerformanceMetrics) calculateCurrentEPS() float64 { - now := time.Now() - nowNano := now.UnixNano() - - // Атомарно загружаем предыдущие значения - lastCalcNano := atomic.LoadInt64(&m.lastEpsCalculation) - lastCount := atomic.LoadUint64(&m.lastGeneratedCount) - - currentGenerated := atomic.LoadUint64(&m.generatedEvents) - - // Первый вызов? - if lastCount == 0 && lastCalcNano == 0 { - atomic.StoreUint64(&m.lastGeneratedCount, currentGenerated) - atomic.StoreInt64(&m.lastEpsCalculation, nowNano) - return 0.0 - } - - durationSec := float64(nowNano-lastCalcNano) / 1e9 - if durationSec < 1.0 { - return 0.0 - } - - eventsDiff := currentGenerated - lastCount - currentEPS := float64(eventsDiff) / durationSec - - atomic.StoreUint64(&m.lastGeneratedCount, currentGenerated) - atomic.StoreInt64(&m.lastEpsCalculation, nowNano) - - return currentEPS -} - -// === МЕТОДЫ ПОЛУЧЕНИЯ СТАТИСТИКИ === - -func (m *PerformanceMetrics) GetStats() (generated, sent, failed uint64, eps float64) { - generated = atomic.LoadUint64(&m.generatedEvents) - sent = atomic.LoadUint64(&m.sentEvents) - failed = atomic.LoadUint64(&m.failedEvents) - - duration := time.Since(m.startTime).Seconds() - if duration > 0 { - eps = float64(generated) / duration - } - - return generated, sent, failed, eps -} - -// DetailedStats возвращает подробную статистику -type DetailedStats struct { - // События - Generated uint64 `json:"generated"` - Sent uint64 `json:"sent"` - Failed uint64 `json:"failed"` - Dropped uint64 `json:"dropped"` - - // Производительность - EPS float64 `json:"eps"` - CurrentEPS float64 `json:"current_eps"` - AvgProcessingTime time.Duration `json:"avg_processing_time"` - MaxProcessingTime time.Duration `json:"max_processing_time"` - MinProcessingTime time.Duration `json:"min_processing_time"` - - // Сеть - NetworkConnections uint64 `json:"network_connections"` - NetworkReconnects uint64 `json:"network_reconnects"` - NetworkTimeouts uint64 `json:"network_timeouts"` - - // Worker Pool - ActiveWorkers uint64 `json:"active_workers"` - QueuedJobs uint64 `json:"queued_jobs"` - CompletedJobs uint64 `json:"completed_jobs"` - RejectedJobs uint64 `json:"rejected_jobs"` - - // Система - Runtime time.Duration `json:"runtime"` - MemoryUsage uint64 `json:"memory_usage_mb"` - Goroutines int `json:"goroutines"` - GCPauses uint32 `json:"gc_pauses"` -} - -func (m *PerformanceMetrics) GetDetailedStats() DetailedStats { - generated := atomic.LoadUint64(&m.generatedEvents) - sent := atomic.LoadUint64(&m.sentEvents) - failed := atomic.LoadUint64(&m.failedEvents) - dropped := atomic.LoadUint64(&m.droppedEvents) - - totalProcessing := atomic.LoadUint64(&m.totalProcessingTime) - maxProcessing := atomic.LoadUint64(&m.maxProcessingTime) - minProcessing := atomic.LoadUint64(&m.minProcessingTime) - - // ИСПРАВЛЕНИЕ: переименовываем переменную - uptime := time.Since(m.startTime) // Было: runtime := time.Since(m.startTime) - uptimeSeconds := uptime.Seconds() // Было: runtimeSeconds := runtime.Seconds() - - // Текущий EPS (за последние 10 секунд) - currentEPS := m.calculateCurrentEPS() - - // Средняя EPS за все время - var avgEPS float64 - if uptimeSeconds > 0 { - avgEPS = float64(generated) / uptimeSeconds - } - - // Среднее время обработки - var avgProcessingTime time.Duration - if generated > 0 { - avgProcessingTime = time.Duration(totalProcessing / generated) - } - - // ИСПРАВЛЕНИЕ: правильное использование пакета runtime - var memStats runtime.MemStats // Было: runtime.memStats - runtime.ReadMemStats(&memStats) - - return DetailedStats{ - Generated: generated, - Sent: sent, - Failed: failed, - Dropped: dropped, - EPS: avgEPS, - CurrentEPS: currentEPS, - AvgProcessingTime: avgProcessingTime, - MaxProcessingTime: time.Duration(maxProcessing), - MinProcessingTime: time.Duration(minProcessing), - NetworkConnections: atomic.LoadUint64(&m.networkConnections), - NetworkReconnects: atomic.LoadUint64(&m.networkReconnects), - NetworkTimeouts: atomic.LoadUint64(&m.networkTimeouts), - ActiveWorkers: atomic.LoadUint64(&m.activeWorkers), - QueuedJobs: atomic.LoadUint64(&m.queuedJobs), - CompletedJobs: atomic.LoadUint64(&m.completedJobs), - RejectedJobs: atomic.LoadUint64(&m.rejectedJobs), - Runtime: uptime, // Было: Runtime: runtime - MemoryUsage: memStats.Alloc / 1024 / 1024, // MB - Goroutines: runtime.NumGoroutine(), - GCPauses: memStats.NumGC, - } -} - -// String тоже нужно обновить -func (m *PerformanceMetrics) String() string { - stats := m.GetDetailedStats() - - return fmt.Sprintf(` -📊 === UEBA GENERATOR METRICS === -📈 Events: Generated=%d, Sent=%d, Failed=%d, Dropped=%d -⚡ Performance: %.1f EPS (avg), %.1f EPS (current) -⏱️ Processing: avg=%v, max=%v, min=%v -🌐 Network: Connections=%d, Reconnects=%d, Timeouts=%d -👷 Workers: Active=%d, Queued=%d, Completed=%d, Rejected=%d -💻 System: Uptime=%v, Memory=%dMB, Goroutines=%d, GC=%d -`, - stats.Generated, stats.Sent, stats.Failed, stats.Dropped, - stats.EPS, stats.CurrentEPS, - stats.AvgProcessingTime, stats.MaxProcessingTime, stats.MinProcessingTime, - stats.NetworkConnections, stats.NetworkReconnects, stats.NetworkTimeouts, - stats.ActiveWorkers, stats.QueuedJobs, stats.CompletedJobs, stats.RejectedJobs, - stats.Runtime.Truncate(time.Second), stats.MemoryUsage, stats.Goroutines, stats.GCPauses, - ) -} - -// === ГЛОБАЛЬНЫЙ РЕЕСТР МЕТРИК === - -var globalMetrics *PerformanceMetrics - -// GetGlobalMetrics возвращает глобальный экземпляр метрик -func GetGlobalMetrics() *PerformanceMetrics { - if globalMetrics == nil { - globalMetrics = NewPerformanceMetrics() - } - return globalMetrics -} +var ( + GeneratedEvents = promauto.NewCounter(prometheus.CounterOpts{ + Name: "pulsar_events_generated_total", + Help: "Total number of events generated by the generator", + }) + SentEvents = promauto.NewCounter(prometheus.CounterOpts{ + Name: "pulsar_events_sent_total", + Help: "Total number of events successfullly delivered to destination", + }) + FailedEvents = promauto.NewCounter(prometheus.CounterOpts{ + Name: "pulsar_events_failed_total", + Help: "Total number of events that failed to be sent due to errors", + }) + DroppedEvents = promauto.NewCounter(prometheus.CounterOpts{ + Name: "pulsar_events_dropped_total", + Help: "Total number of events dropped to internal resourse constraints", + }) + + ActiveWorkers = promauto.NewGauge(prometheus.GaugeOpts{ + Name: "pulsar_workers_active", + Help: "Current number of active worker goroutines", + }) + QueuedJobs = promauto.NewGauge(prometheus.GaugeOpts{ + Name: "pulsar_jobs_queued", + Help: "Current number of jobs waiting in queue to be processed by workers", + }) + CompletedJobs = promauto.NewCounter(prometheus.CounterOpts{ + Name: "pulsar_jobs_completed_total", + Help: "Total number of jobs successfully processed by the worker pool", + }) + RejectedJobs = promauto.NewCounter(prometheus.CounterOpts{ + Name: "pulsar_jobs_rejected_total", + Help: "Total number of jobs rejected because the job queue was at capacity", + }) +) diff --git a/internal/metrics/metrics_test.go b/internal/metrics/metrics_test.go deleted file mode 100644 index bfa9ffe..0000000 --- a/internal/metrics/metrics_test.go +++ /dev/null @@ -1,195 +0,0 @@ -package metrics - -import ( - "testing" - "time" -) - -func TestNewPerformanceMetrics(t *testing.T) { - m := NewPerformanceMetrics() - if m == nil { - t.Fatal("Expected non-nil metrics") - } - - if m.minProcessingTime != ^uint64(0) { - t.Errorf("Expected minProcessingTime to be max uint64, got %d", m.minProcessingTime) - } -} - -func TestEventCounters(t *testing.T) { - m := NewPerformanceMetrics() - - m.IncrementGenerated() - m.IncrementSent() - m.IncrementFailed() - m.IncrementDropped() - - g, s, f, _ := m.GetStats() - if g != 1 || s != 1 || f != 1 { - t.Errorf("Expected (1,1,1), got (%d,%d,%d)", g, s, f) - } - - d := atomicLoad(&m.droppedEvents) - if d != 1 { - t.Errorf("Expected dropped=1, got %d", d) - } -} - -func TestNetworkCounters(t *testing.T) { - m := NewPerformanceMetrics() - - m.IncrementConnections() - m.IncrementReconnects() - m.IncrementTimeouts() - m.DecrementConnections() // should go back to 0 - - conns := atomicLoad(&m.networkConnections) - if conns != 0 { - t.Errorf("Expected connections=0 after inc+dec, got %d", conns) - } - - if atomicLoad(&m.networkReconnects) != 1 || - atomicLoad(&m.networkTimeouts) != 1 { - t.Error("Reconnects or timeouts not incremented correctly") - } -} - -func TestWorkerPoolCounters(t *testing.T) { - m := NewPerformanceMetrics() - - m.IncrementActiveWorkers() - m.IncrementActiveWorkers() - m.DecrementActiveWorkers() - m.SetQueuedJobs(5) - m.IncrementCompletedJobs() - m.IncrementRejectedJobs() - - active := atomicLoad(&m.activeWorkers) - queued := atomicLoad(&m.queuedJobs) - completed := atomicLoad(&m.completedJobs) - rejected := atomicLoad(&m.rejectedJobs) - - if active != 1 { - t.Errorf("Expected active workers = 1, got %d", active) - } - if queued != 5 { - t.Errorf("Expected queued jobs = 5, got %d", queued) - } - if completed != 1 || rejected != 1 { - t.Errorf("Completed/rejected mismatch: %d / %d", completed, rejected) - } -} - -func TestRecordProcessingTime(t *testing.T) { - m := NewPerformanceMetrics() - - // Record valid durations - m.RecordProcessingTime(100 * time.Millisecond) - m.RecordProcessingTime(200 * time.Millisecond) - m.RecordProcessingTime(50 * time.Millisecond) - - total := atomicLoad(&m.totalProcessingTime) - max := atomicLoad(&m.maxProcessingTime) - min := atomicLoad(&m.minProcessingTime) - - expectedTotal := uint64((100 + 200 + 50) * 1e6) // ns - if total != expectedTotal { - t.Errorf("Expected total=%d, got %d", expectedTotal, total) - } - if max != uint64(200*1e6) { - t.Errorf("Expected max=200ms in ns, got %d", max) - } - if min != uint64(50*1e6) { - t.Errorf("Expected min=50ms in ns, got %d", min) - } - - // Ignore invalid durations - m.RecordProcessingTime(-10 * time.Millisecond) - m.RecordProcessingTime(0) - - // Values should not change - if atomicLoad(&m.totalProcessingTime) != expectedTotal { - t.Error("Invalid duration changed total") - } -} - -func TestEPSAndCurrentEPS(t *testing.T) { - m := NewPerformanceMetrics() - - // Simulate some events over time - m.IncrementGenerated() - time.Sleep(100 * time.Millisecond) - m.IncrementGenerated() - m.IncrementGenerated() - - // Get average EPS (over full runtime) - _, _, _, avgEPS := m.GetStats() - if avgEPS <= 0 { - t.Errorf("Expected positive avg EPS, got %.2f", avgEPS) - } - - // First call to current EPS → returns 0 and initializes baseline - currentEPS := m.calculateCurrentEPS() - if currentEPS != 0 { - t.Errorf("First current EPS should be 0, got %.2f", currentEPS) - } - - // Wait and generate more - time.Sleep(1100 * time.Millisecond) - m.IncrementGenerated() - m.IncrementGenerated() - - // Now current EPS should reflect recent rate - currentEPS = m.calculateCurrentEPS() - if currentEPS <= 0 { - t.Errorf("Expected positive current EPS after second batch, got %.2f", currentEPS) - } -} - -func TestGetDetailedStats(t *testing.T) { - m := NewPerformanceMetrics() - - m.IncrementGenerated() - m.IncrementSent() - m.RecordProcessingTime(100 * time.Millisecond) - m.IncrementConnections() - m.IncrementActiveWorkers() - m.SetQueuedJobs(3) - - stats := m.GetDetailedStats() - - if stats.Generated != 1 || - stats.Sent != 1 || - stats.ActiveWorkers != 1 || - stats.QueuedJobs != 3 { - t.Error("Detailed stats mismatch") - } - - if stats.AvgProcessingTime != 100*time.Millisecond { - t.Errorf("Expected avg processing = 100ms, got %v", stats.AvgProcessingTime) - } - - if stats.Goroutines <= 0 { - t.Error("Goroutines count should be > 0") - } -} - -func TestGlobalMetricsSingleton(t *testing.T) { - m1 := GetGlobalMetrics() - m2 := GetGlobalMetrics() - - if m1 != m2 { - t.Error("Global metrics should be singleton") - } - - m1.IncrementGenerated() - g2, _, _, _ := m2.GetStats() - if g2 != 1 { - t.Error("Global instance not shared") - } -} - -// Helper for cleaner atomic reads in tests -func atomicLoad(ptr *uint64) uint64 { - return *ptr // safe in single-threaded test context -} diff --git a/internal/monitoring/monitoring.go b/internal/monitoring/monitoring.go deleted file mode 100644 index 896ccd3..0000000 --- a/internal/monitoring/monitoring.go +++ /dev/null @@ -1,70 +0,0 @@ -package monitoring - -import ( - "context" - "time" - - "github.com/NGRsoftlab/pulsar/internal/logger" - "github.com/NGRsoftlab/pulsar/internal/metrics" -) - -type Monitor interface { - Start(ctx context.Context) - Stop() error -} - -// Monitor периодически выводит статистику -type MonitorImp struct { - interval time.Duration - metrics *metrics.PerformanceMetrics - stopChan chan struct{} - logger logger.Logger -} - -func NewMonitor(interval time.Duration, log logger.Logger) *MonitorImp { - return &MonitorImp{ - interval: interval, - metrics: metrics.GetGlobalMetrics(), - stopChan: make(chan struct{}), - logger: log, - } -} - -func (m *MonitorImp) Start(ctx context.Context) { - ticker := time.NewTicker(m.interval) - defer ticker.Stop() - - m.logger.Info("Monitor запущен: интервал %v", m.interval) - - for { - select { - case <-ticker.C: - m.logger.Info("%s", m.metrics.String()) - - case <-ctx.Done(): - m.logger.Info("Monitor остановлен") - m.printFinalStats() - return - - case <-m.stopChan: - m.logger.Info("Monitor остановлен (stop signal)") - m.printFinalStats() - return - } - } -} - -func (m *MonitorImp) Stop() error { - select { - case <-m.stopChan: - - default: - close(m.stopChan) - } - return nil -} - -func (m *MonitorImp) printFinalStats() { - m.logger.Info("=== ФИНАЛЬНАЯ СТАТИСТИКА ===") - m.logger.Info("%s", m.metrics.String()) -} diff --git a/internal/monitoring/monitoring_test.go b/internal/monitoring/monitoring_test.go deleted file mode 100644 index 1a8dfed..0000000 --- a/internal/monitoring/monitoring_test.go +++ /dev/null @@ -1,87 +0,0 @@ -package monitoring_test - -import ( - "context" - "testing" - "time" - - "github.com/NGRsoftlab/pulsar/internal/monitoring" - "github.com/stretchr/testify/mock" - "github.com/stretchr/testify/require" -) - -// MockLogger реализует мок логгера -type MockLogger struct { - mock.Mock -} - -func (m *MockLogger) Debug(msg string, args ...any) { - m.Called(append([]any{msg}, args...)...) -} - -func (m *MockLogger) Info(msg string, args ...any) { - m.Called(append([]any{msg}, args...)...) -} - -func (m *MockLogger) Warn(msg string, args ...any) { - m.Called(append([]any{msg}, args...)...) -} - -func (m *MockLogger) Error(msg string, args ...any) { - m.Called(append([]any{msg}, args...)...) -} - -func TestMonitor_StartAndStopWithContext(t *testing.T) { - log := new(MockLogger) - monitor := monitoring.NewMonitor(50*time.Millisecond, log) - - // ожидаем старт - log.On("Info", mock.Anything, mock.Anything).Return().Maybe() - log.On("Info", mock.Anything).Return().Maybe() - - ctx, cancel := context.WithTimeout(context.Background(), 120*time.Millisecond) - defer cancel() - - go monitor.Start(ctx) - - // ждём завершения контекста - <-ctx.Done() - - // Stop можно вызвать, но он не должен паниковать - require.NoError(t, monitor.Stop()) - - log.AssertExpectations(t) -} - -func TestMonitor_StopSignal(t *testing.T) { - log := new(MockLogger) - monitor := monitoring.NewMonitor(50*time.Millisecond, log) - - log.On("Info", mock.Anything, mock.Anything).Return().Maybe() - log.On("Info", mock.Anything).Return().Maybe() - - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - - go monitor.Start(ctx) - - // подождём немного, чтобы тикер успел один раз отработать - time.Sleep(70 * time.Millisecond) - - // теперь остановим через Stop - require.NoError(t, monitor.Stop()) - - // дадим горутине завершиться - time.Sleep(30 * time.Millisecond) - - log.AssertExpectations(t) -} - -func TestMonitor_StopIdempotent(t *testing.T) { - log := new(MockLogger) - monitor := monitoring.NewMonitor(100*time.Millisecond, log) - - // несколько вызовов Stop не должны паниковать - require.NoError(t, monitor.Stop()) - require.NoError(t, monitor.Stop()) -} diff --git a/internal/network/tcp_pool.go b/internal/network/tcp_pool.go index bc6cf72..e571e9c 100644 --- a/internal/network/tcp_pool.go +++ b/internal/network/tcp_pool.go @@ -8,8 +8,6 @@ import ( "sync" "sync/atomic" "time" - - "github.com/NGRsoftlab/pulsar/internal/metrics" ) type TCPConnection struct { @@ -91,8 +89,6 @@ func (pool *TCPConnectionPool) createConnection(id int) (*TCPConnection, error) } } - metrics.GetGlobalMetrics().IncrementConnections() - return &TCPConnection{ conn: conn, id: id, @@ -138,8 +134,6 @@ func (conn *TCPConnection) MarkUnhealthy() { conn.isHealthy = false if conn.conn != nil { conn.conn.Close() - metrics.GetGlobalMetrics().DecrementConnections() - metrics.GetGlobalMetrics().IncrementReconnects() } log.Printf("TCP connection %d marked as unhealthy", conn.id) } diff --git a/internal/pipeline/factory/factory.go b/internal/pipeline/factory/factory.go index 28a5434..a09b496 100644 --- a/internal/pipeline/factory/factory.go +++ b/internal/pipeline/factory/factory.go @@ -18,12 +18,11 @@ import ( ) type PipelineFactory struct { - cfg *config.Config - metrics stages.MetricsCollector + cfg *config.Config } -func NewPipelineFactory(cfg *config.Config, metrics stages.MetricsCollector) *PipelineFactory { - return &PipelineFactory{cfg: cfg, metrics: metrics} +func NewPipelineFactory(cfg *config.Config) *PipelineFactory { + return &PipelineFactory{cfg: cfg} } func (f *PipelineFactory) calculateBufferSize() int { @@ -87,7 +86,7 @@ func (f *PipelineFactory) createGenerationStage() (coordinator.Stage, error) { queueSize := max(f.cfg.Generator.EventsPerSecond*2, 1000) workerPool := workers.NewWorkerPool(0, queueSize, func() types.JobBatch { return stages.NewEventGenerationJobBatch(50) - }, f.metrics.(workers.WorkerMetrics)) + }) workerPool.SetPoolType("generation") // 4. Создаём этап с ЧИСТЫМ ИНТЕРФЕЙСОМ @@ -97,7 +96,6 @@ func (f *PipelineFactory) createGenerationStage() (coordinator.Stage, error) { serializationMode, packetMode, workerPool, - f.metrics, ), nil } @@ -131,19 +129,16 @@ func (f *PipelineFactory) createSendingStage() (coordinator.Stage, error) { return nil, fmt.Errorf("failed to create sender: %w", err) } - wm := f.metrics.(workers.WorkerMetrics) - workerPool := workers.NewWorkerPool( 0, 5000, func() types.JobBatch { return stages.NewNetworkSendJobBatch(50) }, - wm, ) workerPool.SetPoolType("network") - sendStage := stages.NewNetworkSendingStage(f.metrics, workerPool, sender) + sendStage := stages.NewNetworkSendingStage(workerPool, sender) if err := sendStage.SetDestinations(f.cfg.Sender.Destinations); err != nil { return nil, fmt.Errorf("failed to set destinations: %w", err) diff --git a/internal/pipeline/factory/factory_test.go b/internal/pipeline/factory/factory_test.go index 34f3754..41ca5c1 100644 --- a/internal/pipeline/factory/factory_test.go +++ b/internal/pipeline/factory/factory_test.go @@ -6,7 +6,6 @@ import ( "github.com/stretchr/testify/assert" "github.com/NGRsoftlab/pulsar/internal/config" - "github.com/NGRsoftlab/pulsar/internal/metrics" ) func TestPipelineFactory_ParseEventTypes(t *testing.T) { @@ -49,7 +48,7 @@ func TestPipelineFactory_ParseEventTypes(t *testing.T) { EventTypes: tt.eventTypes, }, } - factory := NewPipelineFactory(cfg, metrics.GetGlobalMetrics()) + factory := NewPipelineFactory(cfg) _, err := factory.ParseEventTypes() @@ -70,7 +69,7 @@ func TestPipelineFactory_createGenerationStage(t *testing.T) { EventTypes: []string{"netflow"}, }, } - factory := NewPipelineFactory(cfg, metrics.GetGlobalMetrics()) + factory := NewPipelineFactory(cfg) stage, err := factory.createGenerationStage() assert.NoError(t, err) @@ -86,7 +85,7 @@ func TestPipelineFactory_createSendingStage(t *testing.T) { Protocol: "udp", }, } - factory := NewPipelineFactory(cfg, metrics.GetGlobalMetrics()) + factory := NewPipelineFactory(cfg) stage, err := factory.createSendingStage() assert.NoError(t, err) @@ -99,7 +98,7 @@ func TestPipelineFactory_createSendingStage(t *testing.T) { Protocol: "tcp", }, } - factory := NewPipelineFactory(cfg, metrics.GetGlobalMetrics()) + factory := NewPipelineFactory(cfg) stage, err := factory.createSendingStage() assert.Error(t, err) @@ -136,7 +135,7 @@ func TestPipelineFactory_bufferSizeCalculation(t *testing.T) { Protocol: "udp", }, } - factory := NewPipelineFactory(cfg, metrics.GetGlobalMetrics()) + factory := NewPipelineFactory(cfg) pipeline, err := factory.CreatePipeline() assert.NoError(t, err) @@ -161,7 +160,7 @@ func TestPipelineFactory_CreatePipeline(t *testing.T) { }, } - factory := NewPipelineFactory(cfg, metrics.GetGlobalMetrics()) + factory := NewPipelineFactory(cfg) pipeline, err := factory.CreatePipeline() assert.NoError(t, err) @@ -231,7 +230,7 @@ func TestPipelineFactory_calculateBufferSize(t *testing.T) { EventsPerSecond: tt.eventsPerSecond, }, } - factory := NewPipelineFactory(cfg, metrics.GetGlobalMetrics()) + factory := NewPipelineFactory(cfg) actual := factory.calculateBufferSize() diff --git a/internal/pipeline/stages/common.go b/internal/pipeline/stages/common.go index 8b91f64..d34c860 100644 --- a/internal/pipeline/stages/common.go +++ b/internal/pipeline/stages/common.go @@ -5,7 +5,6 @@ package stages import ( "context" - "time" "github.com/NGRsoftlab/pulsar/internal/types" ) @@ -17,22 +16,6 @@ type StageConfig struct { BufferSize int `yaml:"buffer_size" json:"buffer_size"` } -// StageMetrics базовые метрики стадии -type StageMetrics struct { - Name string `json:"name"` - ProcessedEvents uint64 `json:"processed_events"` - ErrorCount uint64 `json:"error_count"` - LastActivity time.Time `json:"last_activity"` -} - -type MetricsCollector interface { - IncrementGenerated() - IncrementSent() - IncrementFailed() - IncrementDropped() - GetStats() (generated, sent, failed uint64, dropped float64) -} - type Pool interface { Start(ctx context.Context) Stop() diff --git a/internal/pipeline/stages/generation.go b/internal/pipeline/stages/generation.go index af798fd..7c5ff00 100644 --- a/internal/pipeline/stages/generation.go +++ b/internal/pipeline/stages/generation.go @@ -9,6 +9,7 @@ import ( "golang.org/x/time/rate" "github.com/NGRsoftlab/pulsar/internal/domain/event" + "github.com/NGRsoftlab/pulsar/internal/metrics" ) // EventGenerationJobBatch - пакет событий для WorkerPool @@ -40,13 +41,13 @@ func (jb *EventGenerationJobBatch) ExecuteBatch() error { if netflowEvt, ok := evt.(*event.NetflowEvent); ok { data, err := netflowEvt.ToBinaryNetFlow() if err != nil { - jb.stage.metrics.IncrementFailed() + metrics.FailedEvents.Inc() continue } serializedData = NewSerializedData(data, evt.Type(), evt.GetID(), jb.stage.serializationMode) } else { // Binary mode не поддерживается для syslog — пропускаем или ошибку? - jb.stage.metrics.IncrementFailed() + metrics.FailedEvents.Inc() continue } } else { @@ -54,14 +55,14 @@ func (jb *EventGenerationJobBatch) ExecuteBatch() error { serializedData, err = NewSerializedDataFromEvent(evt, jb.stage.serializationMode) if err != nil { log.Printf("Serialization failed for event %s: %v", evt.GetID(), err) - jb.stage.metrics.IncrementFailed() + metrics.FailedEvents.Inc() continue } } select { case jb.out <- serializedData: - jb.stage.metrics.IncrementGenerated() + metrics.GeneratedEvents.Inc() case <-context.Background().Done(): return context.Canceled } @@ -79,7 +80,6 @@ type EventGenerationStage struct { serializationMode SerializationMode packetMode bool workerPool Pool - metrics MetricsCollector } // NewEventGenerationStage создаёт стадию генерации @@ -89,7 +89,6 @@ func NewEventGenerationStage( serializationMode SerializationMode, packetMode bool, workerPool Pool, - metrics MetricsCollector, ) *EventGenerationStage { return &EventGenerationStage{ eventType: eventType, @@ -97,7 +96,6 @@ func NewEventGenerationStage( serializationMode: serializationMode, packetMode: packetMode, workerPool: workerPool, - metrics: metrics, } } @@ -196,15 +194,3 @@ func (g *EventGenerationStage) generateEvent() (event.Event, error) { return nil, fmt.Errorf("unsupported event type: %v", g.eventType) } } - -// Конфигурационные методы - -func (g *EventGenerationStage) GetGeneratedCount() uint64 { - generated, _, _, _ := g.metrics.GetStats() - return generated -} - -func (g *EventGenerationStage) GetFailedCount() uint64 { - _, _, failed, _ := g.metrics.GetStats() - return failed -} diff --git a/internal/pipeline/stages/generation_test.go b/internal/pipeline/stages/generation_test.go index 277d27f..d59c48c 100644 --- a/internal/pipeline/stages/generation_test.go +++ b/internal/pipeline/stages/generation_test.go @@ -60,23 +60,6 @@ func (wp *mockWorkerPool) GetJob() types.JobBatch { return job } -// mockMetricsCollector собирает метрики для проверки в тестах -type mockMetricsCollector struct { - mu sync.Mutex - generated, failed, dropped uint64 - sent float64 -} - -func (m *mockMetricsCollector) IncrementGenerated() { m.mu.Lock(); m.generated++; m.mu.Unlock() } -func (m *mockMetricsCollector) IncrementFailed() { m.mu.Lock(); m.failed++; m.mu.Unlock() } -func (m *mockMetricsCollector) IncrementDropped() { m.mu.Lock(); m.dropped++; m.mu.Unlock() } -func (m *mockMetricsCollector) IncrementSent() { m.mu.Lock(); m.sent++; m.mu.Unlock() } -func (m *mockMetricsCollector) GetStats() (uint64, uint64, uint64, float64) { - m.mu.Lock() - defer m.mu.Unlock() - return m.generated, m.dropped, m.failed, m.sent -} - // Тест создания нового батча func TestNewEventGenerationJobBatch(t *testing.T) { batch := NewEventGenerationJobBatch(10) @@ -96,11 +79,9 @@ func TestEventGenerationJobBatch_ExecuteBatch_Nil(t *testing.T) { // Тест ExecuteBatch с событием NetflowEvent func TestEventGenerationJobBatch_ExecuteBatch_Netflow(t *testing.T) { - metrics := &mockMetricsCollector{} stage := &EventGenerationStage{ serializationMode: SerializationModeBinary, packetMode: false, - metrics: metrics, } out := make(chan *SerializedData, 10) @@ -124,10 +105,6 @@ func TestEventGenerationJobBatch_ExecuteBatch_Netflow(t *testing.T) { default: assert.Fail(t, "expected serialized data in output channel") } - - generated, _, failed, _ := metrics.GetStats() - assert.Equal(t, uint64(1), generated) - assert.Equal(t, uint64(0), failed) } // fakeEvent — минимальная реализация event.Event для тестов @@ -165,12 +142,10 @@ func (f *fakeEvent) GetDestinationIP() netip.Addr { // Тест генерации NetflowEvent func TestEventGenerationStage_generateEvent(t *testing.T) { - metrics := &mockMetricsCollector{} stage := &EventGenerationStage{ eventType: event.EventTypeNetflow, serializationMode: SerializationModeBinary, packetMode: false, - metrics: metrics, } evt, err := stage.generateEvent() @@ -181,7 +156,6 @@ func TestEventGenerationStage_generateEvent(t *testing.T) { // Тест Run с ограничением скорости и обработкой через mock pool func TestEventGenerationStage_Run(t *testing.T) { - metrics := &mockMetricsCollector{} wp := newMockWorkerPool() defer wp.Stop() @@ -191,7 +165,6 @@ func TestEventGenerationStage_Run(t *testing.T) { SerializationModeBinary, true, // packetMode wp, - metrics, ) ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) @@ -232,15 +205,11 @@ func TestEventGenerationStage_Run(t *testing.T) { close(out) received = <-receivedCh - generated, _, failed, _ := metrics.GetStats() - assert.GreaterOrEqual(t, generated, uint64(1)) - assert.Equal(t, uint64(0), failed) assert.GreaterOrEqual(t, received, 1) } // Тест остановки по контексту и отправки оставшегося батча func TestEventGenerationStage_Run_CancelWithPendingBatch(t *testing.T) { - metrics := &mockMetricsCollector{} wp := newMockWorkerPool() defer wp.Stop() @@ -250,7 +219,6 @@ func TestEventGenerationStage_Run_CancelWithPendingBatch(t *testing.T) { SerializationModeBinary, true, // packetMode = true → использует ToBinaryNetFlow() wp, - metrics, ) ctx, cancel := context.WithCancel(context.Background()) @@ -286,8 +254,4 @@ func TestEventGenerationStage_Run_CancelWithPendingBatch(t *testing.T) { // Закрываем out, чтобы завершить consumer close(out) <-done - - generated, _, failed, _ := metrics.GetStats() - assert.GreaterOrEqual(t, generated, uint64(1)) - assert.Equal(t, uint64(0), failed) } diff --git a/internal/pipeline/stages/sending.go b/internal/pipeline/stages/sending.go index 61bee92..df2dbeb 100644 --- a/internal/pipeline/stages/sending.go +++ b/internal/pipeline/stages/sending.go @@ -8,6 +8,7 @@ import ( "time" "github.com/NGRsoftlab/pulsar/internal/domain/event" + "github.com/NGRsoftlab/pulsar/internal/metrics" ) type Sender interface { @@ -34,12 +35,10 @@ type NetworkSendingStage struct { sender Sender workerPool Pool - metrics MetricsCollector input chan event.Event } func NewNetworkSendingStage( - metrics MetricsCollector, workerPool Pool, sender Sender, ) *NetworkSendingStage { @@ -49,7 +48,6 @@ func NewNetworkSendingStage( timeout: 5 * time.Second, sender: sender, workerPool: workerPool, - metrics: metrics, input: make(chan event.Event, 1000), } } @@ -121,7 +119,7 @@ func (s *NetworkSendingStage) Run(ctx context.Context, in <-chan *SerializedData if len(currentBatch.data) >= batchSize { if !s.workerPool.Submit(currentBatch) { - s.metrics.IncrementDropped() + metrics.DroppedEvents.Inc() } currentBatch = nil if timer != nil { @@ -134,7 +132,7 @@ func (s *NetworkSendingStage) Run(ctx context.Context, in <-chan *SerializedData case <-timerC: if currentBatch != nil && len(currentBatch.data) > 0 { if !s.workerPool.Submit(currentBatch) { - s.metrics.IncrementDropped() + metrics.DroppedEvents.Inc() } currentBatch = nil } @@ -158,11 +156,11 @@ func (s *NetworkSendingStage) SendData(data *SerializedData) error { err := s.sender.Send(destination, data.Data) if err != nil { - s.metrics.IncrementFailed() + metrics.FailedEvents.Inc() return err } - s.metrics.IncrementSent() + metrics.SentEvents.Inc() return nil } @@ -202,61 +200,9 @@ func (s *NetworkSendingStage) RecreateUnhealthyConnections() int { return 0 } -func (s *NetworkSendingStage) GetSentCount() uint64 { - _, sent, _, _ := s.metrics.GetStats() - return sent -} - -func (s *NetworkSendingStage) GetFailedCount() uint64 { - _, _, failed, _ := s.metrics.GetStats() - return failed -} - -func (s *NetworkSendingStage) GetStageStats() map[string]any { - _, sent, failed, dropped := s.metrics.GetStats() - - stats := map[string]any{ - "protocol": s.protocol, - "destinations": len(s.destinations), - "events_sent": sent, - "events_failed": failed, - "events_dropped": dropped, - "worker_pool_healthy": s.workerPool != nil, - } - - return stats -} - func (s *NetworkSendingStage) IsHealthy() (bool, string) { if s.workerPool == nil { return false, "worker pool not initialized" } return s.sender.IsHealthy() } - -func (s *NetworkSendingStage) GetOptimizationRecommendations() []string { - var recommendations []string - _, dropped, failed, sent := s.metrics.GetStats() - - if dropped > 0 { - dropRate := float64(dropped) / (float64(sent) + float64(failed) + float64(dropped)) * 100.0 - if dropRate > 1.0 { - recommendations = append(recommendations, - fmt.Sprintf("High drop rate (%.1f%%) - consider increasing worker pool queue size", dropRate)) - } - } - - if failed > 0 { - failRate := float64(failed)/float64(sent) + float64(failed)*100 - if failRate > 5.0 { - recommendations = append(recommendations, - fmt.Sprintf("High failure rate (%.1f%%) - check network connectivity", failRate)) - } - } - - if len(recommendations) == 0 { - recommendations = append(recommendations, "Stage operating optimally - no recommendations") - } - - return recommendations -} diff --git a/internal/pipeline/stages/sending_test.go b/internal/pipeline/stages/sending_test.go index 5bebb6f..a67f402 100644 --- a/internal/pipeline/stages/sending_test.go +++ b/internal/pipeline/stages/sending_test.go @@ -132,11 +132,10 @@ func newTestSerializedData(dest string, data []byte) *SerializedData { // ================ func TestNewNetworkSendingStage(t *testing.T) { - metrics := &mockMetricsCollector{} wp := newMockWorkerPoolForNetwork() sender := newMockSender() - stage := NewNetworkSendingStage(metrics, wp, sender) + stage := NewNetworkSendingStage(wp, sender) assert.NotNil(t, stage) assert.Equal(t, []string{"127.0.0.1:514"}, stage.destinations) @@ -154,11 +153,9 @@ func TestNetworkSendJobBatch_ExecuteBatch_Nil(t *testing.T) { func TestNetworkSendJobBatch_ExecuteBatch(t *testing.T) { sender := newMockSender() - metrics := &mockMetricsCollector{} stage := &NetworkSendingStage{ destinations: []string{"127.0.0.1:9999"}, sender: sender, - metrics: metrics, protocol: "udp", } @@ -178,22 +175,14 @@ func TestNetworkSendJobBatch_ExecuteBatch(t *testing.T) { assert.Len(t, sent, 2) assert.Equal(t, "127.0.0.1:9999", sent[0].destination) // из destinations assert.Equal(t, "192.0.2.1:1234", sent[1].destination) // из data.Destination - - generated, dropped, failed, sentCount := metrics.GetStats() - assert.Equal(t, uint64(0), generated) - assert.Equal(t, uint64(0), dropped) - assert.Equal(t, uint64(0), failed) - assert.Equal(t, float64(2), sentCount) } func TestNetworkSendJobBatch_ExecuteBatch_SendFailure(t *testing.T) { sender := newMockSender() sender.SetFailSend(true) - metrics := &mockMetricsCollector{} stage := &NetworkSendingStage{ destinations: []string{"127.0.0.1:9999"}, sender: sender, - metrics: metrics, protocol: "udp", } @@ -206,18 +195,13 @@ func TestNetworkSendJobBatch_ExecuteBatch_SendFailure(t *testing.T) { err := batch.ExecuteBatch() require.NoError(t, err) // ошибка отправки не прерывает batch - - _, _, failed, sent := metrics.GetStats() - assert.Equal(t, uint64(1), failed) - assert.Equal(t, float64(0), sent) } func TestNetworkSendingStage_Run(t *testing.T) { - metrics := &mockMetricsCollector{} wp := newMockWorkerPoolForNetwork() sender := newMockSender() - stage := NewNetworkSendingStage(metrics, wp, sender) + stage := NewNetworkSendingStage(wp, sender) stage.SetDestinations([]string{"127.0.0.1:1234"}) ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) @@ -256,19 +240,13 @@ func TestNetworkSendingStage_Run(t *testing.T) { sent := sender.GetSent() assert.Len(t, sent, 2) assert.Equal(t, "127.0.0.1:1234", sent[0].destination) - - _, dropped, failed, sentCount := metrics.GetStats() - assert.Equal(t, uint64(0), dropped) - assert.Equal(t, uint64(0), failed) - assert.Equal(t, float64(2), sentCount) } func TestNetworkSendingStage_Run_CancelWithPendingBatch(t *testing.T) { - metrics := &mockMetricsCollector{} wp := newMockWorkerPoolForNetwork() sender := newMockSender() - stage := NewNetworkSendingStage(metrics, wp, sender) + stage := NewNetworkSendingStage(wp, sender) stage.SetDestinations([]string{"127.0.0.1:1234"}) ctx, cancel := context.WithCancel(context.Background()) @@ -304,15 +282,10 @@ func TestNetworkSendingStage_Run_CancelWithPendingBatch(t *testing.T) { sent := sender.GetSent() assert.Len(t, sent, 3) - - _, dropped, failed, sentCount := metrics.GetStats() - assert.Equal(t, uint64(0), dropped) - assert.Equal(t, uint64(0), failed) - assert.Equal(t, float64(3), sentCount) } func TestNetworkSendingStage_SetDestinations(t *testing.T) { - stage := NewNetworkSendingStage(&mockMetricsCollector{}, newMockWorkerPoolForNetwork(), newMockSender()) + stage := NewNetworkSendingStage(newMockWorkerPoolForNetwork(), newMockSender()) err := stage.SetDestinations([]string{"192.168.1.1:514", "10.0.0.1:514"}) require.NoError(t, err) @@ -326,7 +299,7 @@ func TestNetworkSendingStage_SetDestinations(t *testing.T) { } func TestNetworkSendingStage_SetProtocol(t *testing.T) { - stage := NewNetworkSendingStage(&mockMetricsCollector{}, newMockWorkerPoolForNetwork(), newMockSender()) + stage := NewNetworkSendingStage(newMockWorkerPoolForNetwork(), newMockSender()) err := stage.SetProtocol("tcp") require.NoError(t, err) @@ -341,7 +314,7 @@ func TestNetworkSendingStage_SetProtocol(t *testing.T) { } func TestNetworkSendingStage_ResizeConnectionPool(t *testing.T) { - stage := NewNetworkSendingStage(&mockMetricsCollector{}, newMockWorkerPoolForNetwork(), newMockSender()) + stage := NewNetworkSendingStage(newMockWorkerPoolForNetwork(), newMockSender()) err := stage.ResizeConnectionPool(10) assert.Error(t, err) @@ -352,23 +325,9 @@ func TestNetworkSendingStage_ResizeConnectionPool(t *testing.T) { assert.NotContains(t, err.Error(), "only supported for TCP") } -func TestNetworkSendingStage_GetStageStats(t *testing.T) { - metrics := &mockMetricsCollector{} - sender := newMockSender() - stage := NewNetworkSendingStage(metrics, newMockWorkerPoolForNetwork(), sender) - - stats := stage.GetStageStats() - expectedKeys := []string{"protocol", "destinations", "events_sent", "events_failed", "events_dropped", "worker_pool_healthy"} - for _, key := range expectedKeys { - assert.Contains(t, stats, key) - } - assert.Equal(t, "udp", stats["protocol"]) - assert.Equal(t, 1, stats["destinations"]) -} - func TestNetworkSendingStage_IsHealthy(t *testing.T) { sender := newMockSender() - stage := NewNetworkSendingStage(&mockMetricsCollector{}, newMockWorkerPoolForNetwork(), sender) + stage := NewNetworkSendingStage(newMockWorkerPoolForNetwork(), sender) healthy, msg := stage.IsHealthy() assert.True(t, healthy) @@ -379,44 +338,3 @@ func TestNetworkSendingStage_IsHealthy(t *testing.T) { assert.False(t, healthy) assert.Equal(t, "mock unhealthy", msg) } - -func TestNetworkSendingStage_GetOptimizationRecommendations(t *testing.T) { - t.Run("Optimal", func(t *testing.T) { - metrics := &mockMetricsCollector{} - sender := newMockSender() - stage := NewNetworkSendingStage(metrics, newMockWorkerPoolForNetwork(), sender) - - recommendations := stage.GetOptimizationRecommendations() - assert.Contains(t, recommendations[0], "operating optimally") - }) - - t.Run("HighDropRate", func(t *testing.T) { - metrics := &mockMetricsCollector{} - sender := newMockSender() - stage := NewNetworkSendingStage(metrics, newMockWorkerPoolForNetwork(), sender) - - metrics.IncrementSent() - metrics.IncrementSent() - metrics.IncrementDropped() - metrics.IncrementDropped() - metrics.IncrementDropped() - - recommendations := stage.GetOptimizationRecommendations() - assert.Contains(t, recommendations[0], "High drop rate") - }) - - t.Run("HighFailureRate", func(t *testing.T) { - metrics := &mockMetricsCollector{} - sender := newMockSender() - stage := NewNetworkSendingStage(metrics, newMockWorkerPoolForNetwork(), sender) - - metrics.IncrementSent() - metrics.IncrementSent() - metrics.IncrementSent() - metrics.IncrementFailed() - metrics.IncrementFailed() - - recommendations := stage.GetOptimizationRecommendations() - assert.Contains(t, recommendations[0], "High failure rate") - }) -} diff --git a/internal/workers/workerpool_test.go b/internal/workers/workerpool_test.go index d3666d2..878bed3 100644 --- a/internal/workers/workerpool_test.go +++ b/internal/workers/workerpool_test.go @@ -12,24 +12,15 @@ import ( "github.com/NGRsoftlab/pulsar/internal/types" ) -type mockWorkerMetrics struct{} - -func (m *mockWorkerMetrics) IncrementActiveWorkers() {} -func (m *mockWorkerMetrics) DecrementActiveWorkers() {} -func (m *mockWorkerMetrics) IncrementCompletedJobs() {} -func (m *mockWorkerMetrics) IncrementRejectedJobs() {} -func (m *mockWorkerMetrics) SetQueuedJobs(count uint64) {} -func (m *mockWorkerMetrics) RecordProcessingTime(duration time.Duration) {} - func TestWorkerPool_SubmitNil(t *testing.T) { - wp := NewWorkerPool(1, 1, func() types.JobBatch { return &mockJob{} }, &mockWorkerMetrics{}) + wp := NewWorkerPool(1, 1, func() types.JobBatch { return &mockJob{} }) success := wp.Submit(nil) assert.False(t, success, "ожидалось false при отправке nil") } func TestWorkerPool_Submit_ValidJob(t *testing.T) { - wp := NewWorkerPool(1, 2, func() types.JobBatch { return &mockJob{} }, &mockWorkerMetrics{}) + wp := NewWorkerPool(1, 2, func() types.JobBatch { return &mockJob{} }) job := &mockJob{} success := wp.Submit(job) @@ -47,7 +38,7 @@ func TestWorkerPool_GetJob_CallFactory(t *testing.T) { wp := NewWorkerPool(1, 1, func() types.JobBatch { called = true return exceptedJob - }, &mockWorkerMetrics{}) + }) job := wp.GetJob() @@ -66,7 +57,7 @@ func (c countingJob) ExecuteBatch() error { func TestWorkerPool_Integration_SingleJob(t *testing.T) { var wg sync.WaitGroup - wp := NewWorkerPool(1, 2, func() types.JobBatch { return &mockJob{} }, &mockWorkerMetrics{}) + wp := NewWorkerPool(1, 2, func() types.JobBatch { return &mockJob{} }) wg.Add(1) job := countingJob{ @@ -102,7 +93,7 @@ func TestWorkerPool_Integration_MultitipleJobs(t *testing.T) { var completed int var mu sync.Mutex - wp := NewWorkerPool(1, 10, func() types.JobBatch { return &mockJob{} }, &mockWorkerMetrics{}) + wp := NewWorkerPool(1, 10, func() types.JobBatch { return &mockJob{} }) jobs := make([]*countingJob, jobCount) for i := range jobs { @@ -136,7 +127,7 @@ func TestWorkerPool_ContextCancel_StopsAcceptingNewWork(t *testing.T) { var mu sync.Mutex var completed int - wp := NewWorkerPool(2, 10, func() types.JobBatch { return &mockJob{} }, &mockWorkerMetrics{}) + wp := NewWorkerPool(2, 10, func() types.JobBatch { return &mockJob{} }) longJob := &countingJob{ fn: func() { @@ -171,7 +162,7 @@ func TestWorkerPool_Integration_Concurrency(t *testing.T) { var completed int64 var accepted int64 - wp := NewWorkerPool(workerCount, queueSize, func() types.JobBatch { return &mockJob{} }, &mockWorkerMetrics{}) + wp := NewWorkerPool(workerCount, queueSize, func() types.JobBatch { return &mockJob{} }) ctx, cancel := context.WithCancel(context.Background()) diff --git a/internal/workers/workers.go b/internal/workers/workers.go index fc5b21d..a9f7437 100644 --- a/internal/workers/workers.go +++ b/internal/workers/workers.go @@ -6,20 +6,11 @@ import ( "runtime" "sync" "sync/atomic" - "time" + "github.com/NGRsoftlab/pulsar/internal/metrics" "github.com/NGRsoftlab/pulsar/internal/types" ) -type WorkerMetrics interface { - IncrementActiveWorkers() - DecrementActiveWorkers() - IncrementCompletedJobs() - IncrementRejectedJobs() - SetQueuedJobs(count uint64) - RecordProcessingTime(duration time.Duration) -} - // LockFreeQueue использует sync/atomic.Value для безопасного доступа к буферу type LockFreeQueue struct { // Массив хранит слайсы с types.JobBatch для минимизации конкуренции @@ -108,18 +99,16 @@ func (q *LockFreeQueue) WaitPop(ctx context.Context) types.JobBatch { // WorkerPool управляет пулом воркеров с lock-free очередью type WorkerPool struct { workerCount int - metrics WorkerMetrics queue *LockFreeQueue - quit atomic.Bool wg sync.WaitGroup poolType string newJobFunc func() types.JobBatch - metricsChan chan func() started atomic.Bool + cancel context.CancelFunc } // NewWorkerPool создаёт пул с пользовательской фабрикой задач -func NewWorkerPool(workerCount, queueSize int, newJobFunc func() types.JobBatch, metrics WorkerMetrics) *WorkerPool { +func NewWorkerPool(workerCount, queueSize int, newJobFunc func() types.JobBatch) *WorkerPool { if workerCount <= 0 { workerCount = runtime.NumCPU() * 2 } @@ -128,8 +117,6 @@ func NewWorkerPool(workerCount, queueSize int, newJobFunc func() types.JobBatch, workerCount: workerCount, queue: NewLockFreeQueue(queueSize), newJobFunc: newJobFunc, - metricsChan: make(chan func(), 1000), - metrics: metrics, } return wp @@ -145,8 +132,7 @@ func (wp *WorkerPool) Start(ctx context.Context) { return } - wp.wg.Add(1) - go wp.metricsWorker(ctx) + ctx, wp.cancel = context.WithCancel(ctx) for i := 0; i < wp.workerCount; i++ { wp.wg.Add(1) @@ -157,129 +143,76 @@ func (wp *WorkerPool) Start(ctx context.Context) { // worker — рабочая горутина func (wp *WorkerPool) worker(ctx context.Context) { defer wp.wg.Done() - wp.metrics.IncrementActiveWorkers() - defer wp.metrics.DecrementActiveWorkers() + metrics.ActiveWorkers.Inc() + defer metrics.ActiveWorkers.Dec() localCompleted := uint64(0) for { - // Простая проверка в начале - if wp.quit.Load() { - // Простой drain без retry - for { - job := wp.queue.TryPop() - if job == nil { - return - } - start := time.Now() - _ = job.ExecuteBatch() - localCompleted++ - - if localCompleted%100 == 0 { - duration := time.Since(start) - select { - case wp.metricsChan <- func() { - wp.metrics.IncrementCompletedJobs() - wp.metrics.RecordProcessingTime(duration) - }: - default: - } - localCompleted = 0 - } - } - } - job := wp.queue.TryPop() if job == nil { job = wp.queue.WaitPop(ctx) if job == nil { - continue + break } } - start := time.Now() _ = job.ExecuteBatch() localCompleted++ if localCompleted%100 == 0 { - duration := time.Since(start) - select { - case wp.metricsChan <- func() { - wp.metrics.IncrementCompletedJobs() - wp.metrics.RecordProcessingTime(duration) - }: - default: - } + metrics.CompletedJobs.Add(100) + localCompleted = 0 + } + } + + for { + job := wp.queue.TryPop() + if job == nil { + break + } + _ = job.ExecuteBatch() + localCompleted++ + if localCompleted%100 == 0 { + metrics.CompletedJobs.Add(100) localCompleted = 0 } } + + if localCompleted > 0 { + metrics.CompletedJobs.Add(float64(localCompleted)) + } } // Submit отправляет задачу в очередь func (wp *WorkerPool) Submit(job types.JobBatch) bool { if job == nil { - select { - case wp.metricsChan <- func() { - wp.metrics.IncrementRejectedJobs() - }: - default: - } + metrics.RejectedJobs.Inc() return false } if !wp.queue.TryPush(job) { - wp.metricsChan <- func() { - wp.metrics.IncrementRejectedJobs() - } + metrics.RejectedJobs.Inc() return false } - select { - case wp.metricsChan <- func() { - head := wp.queue.head.Load() - tail := wp.queue.tail.Load() - wp.metrics.SetQueuedJobs(tail - head) - }: - default: - } + head := wp.queue.head.Load() + tail := wp.queue.tail.Load() + metrics.QueuedJobs.Set(float64(tail - head)) + return true } // Stop останавливает пул воркеров func (wp *WorkerPool) Stop() { - wp.quit.Store(true) + if wp.cancel != nil { + wp.cancel() + } + wp.wg.Wait() - time.Sleep(5 * time.Millisecond) - close(wp.metricsChan) } // GetJob возвращает новую задачу от фабрики func (wp *WorkerPool) GetJob() types.JobBatch { return wp.newJobFunc() } - -// metricsWorker обрабатывает метрики -func (wp *WorkerPool) metricsWorker(ctx context.Context) { - defer wp.wg.Done() - for { - select { - case f, ok := <-wp.metricsChan: - if !ok { - return - } - f() - case <-ctx.Done(): - for { - select { - case f, ok := <-wp.metricsChan: - if !ok { - return - } - f() - default: - return - } - } - } - } -} diff --git a/internal/workers/workers_bench_test.go b/internal/workers/workers_bench_test.go index b393e81..d38ce45 100644 --- a/internal/workers/workers_bench_test.go +++ b/internal/workers/workers_bench_test.go @@ -43,7 +43,7 @@ func BenchmarkLockFreeQueue_Concurrent(b *testing.B) { // ===== WORKERPOOL ===== func BenchmarkWorkerPool_Submit(b *testing.B) { - wp := NewWorkerPool(8, 1024, func() types.JobBatch { return &mockJob{} }, &mockWorkerMetrics{}) + wp := NewWorkerPool(8, 1024, func() types.JobBatch { return &mockJob{} }) ctx, cancel := context.WithCancel(context.Background()) wp.Start(ctx) defer func() { @@ -60,7 +60,7 @@ func BenchmarkWorkerPool_Submit(b *testing.B) { } func BenchmarkWorkerPool_Throughput(b *testing.B) { - wp := NewWorkerPool(8, 1024, func() types.JobBatch { return &mockJob{} }, &mockWorkerMetrics{}) + wp := NewWorkerPool(8, 1024, func() types.JobBatch { return &mockJob{} }) ctx, cancel := context.WithCancel(context.Background()) wp.Start(ctx) defer func() { @@ -85,7 +85,7 @@ func BenchmarkWorkerPool_Throughput(b *testing.B) { } func BenchmarkWorkerPool_HighContention(b *testing.B) { - wp := NewWorkerPool(2, 256, func() types.JobBatch { return &mockJob{} }, &mockWorkerMetrics{}) + wp := NewWorkerPool(2, 256, func() types.JobBatch { return &mockJob{} }) ctx, cancel := context.WithCancel(context.Background()) wp.Start(ctx) defer func() { From 74468303bc5d3b380a8655c67be64e5620ece79f Mon Sep 17 00:00:00 2001 From: "n.shabanov" Date: Wed, 14 Jan 2026 13:21:53 +0300 Subject: [PATCH 2/2] fix: remove BuildTime var --- Makefile | 3 +-- cmd/main.go | 7 ++----- internal/cli/parser.go | 17 +++++++---------- internal/cli/parser_test.go | 6 +++--- 4 files changed, 13 insertions(+), 20 deletions(-) diff --git a/Makefile b/Makefile index f4b749f..62c917b 100644 --- a/Makefile +++ b/Makefile @@ -11,10 +11,9 @@ ifeq ($(OS),Windows_NT) endif BINARY_PATH=./cmd VERSION ?= $(shell git describe --tags --always --dirty="-dev") -BUILD_TIME ?= $(shell date /t) build: - $(GOBUILD) -ldflags="-X main.Version=$(VERSION) -X main.BuildTime=$(BUILD_TIME)" -o $(BINARY_NAME) $(BINARY_PATH) + $(GOBUILD) -ldflags="-X main.Version=$(VERSION)" -o $(BINARY_NAME) $(BINARY_PATH) # Run tests all tests from root dir test-all: diff --git a/cmd/main.go b/cmd/main.go index a971d3e..85b485a 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -14,10 +14,7 @@ import ( "github.com/prometheus/client_golang/prometheus/promhttp" ) -var ( - Version = "dev" - BuildTime = "unknown" -) +var Version = "dev" func main() { if err := run(); err != nil { @@ -28,7 +25,7 @@ func main() { func run() error { // Парсим CLI флаги - parser := cli.NewParser(Version, BuildTime) + parser := cli.NewParser(Version) flags, err := parser.Parse(os.Args[1:]) if err != nil { return fmt.Errorf("failde to parse CLI flags: %v", err) diff --git a/internal/cli/parser.go b/internal/cli/parser.go index e3599df..9d86b7d 100644 --- a/internal/cli/parser.go +++ b/internal/cli/parser.go @@ -22,14 +22,13 @@ type Flags struct { // Parser инкапсулирует логику парсинга CLI-аргументов. type Parser struct { - version string - buildTime string - flagSet *flag.FlagSet - flags *Flags + version string + flagSet *flag.FlagSet + flags *Flags } // NewParser создаёт новый парсер с заданной версией и временем сборки. -func NewParser(version, buildTime string) *Parser { +func NewParser(version string) *Parser { fs := flag.NewFlagSet("event-generator", flag.ContinueOnError) flags := &Flags{} @@ -61,10 +60,9 @@ func NewParser(version, buildTime string) *Parser { fs.DurationVar(&flags.Duration, "t", 0, "Duration (shorthand)") return &Parser{ - version: version, - buildTime: buildTime, - flagSet: fs, - flags: flags, + version: version, + flagSet: fs, + flags: flags, } } @@ -83,7 +81,6 @@ func (p *Parser) Parse(args []string) (*Flags, error) { func (p *Parser) ShowVersionInfo() { fmt.Printf("Pulsar\n") fmt.Printf("Version: %s\n", p.version) - fmt.Printf("Build Time: %s\n", p.buildTime) } // ShowHelpInfo выводит справку по использованию. diff --git a/internal/cli/parser_test.go b/internal/cli/parser_test.go index 0a79817..d7cab94 100644 --- a/internal/cli/parser_test.go +++ b/internal/cli/parser_test.go @@ -110,7 +110,7 @@ func TestParser_Parse(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - parser := NewParser("test-version", "test-build-time") + parser := NewParser("test-version") parser.flagSet.SetOutput(new(bytes.Buffer)) flags, err := parser.Parse(tt.args) @@ -133,11 +133,11 @@ func TestParser_Parse(t *testing.T) { // Smoke-тесты для функций вывода (они не должны паниковать) func TestParser_ShowVersionInfo(t *testing.T) { - parser := NewParser("1.0.0", "2025-01-01T00:00:00Z") + parser := NewParser("1.0.0") parser.ShowVersionInfo() } func TestParser_ShowHelpInfo(t *testing.T) { - parser := NewParser("1.0.0", "2025-01-01T00:00:00Z") + parser := NewParser("1.0.0") parser.ShowHelpInfo() }