feat(metrics): migrate to Prometheus + fix graceful shutdown

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"]

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:

README.md

@@ -13,17 +13,16 @@ 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
 
 - 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:

cmd/main.go

@@ -2,23 +2,19 @@ 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 (
-	Version   = "dev"
-	BuildTime = "unknown"
-)
+var Version = "dev"
 
 func main() {
 	if err := run(); err != nil {
@@ -29,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)
@@ -59,18 +55,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)

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

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=

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 выводит справку по использованию.

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()
 }

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
 }
 

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)
-	}
-}