Ft prometheus metrics

cmd/api.go

@@ -8,10 +8,18 @@ import (
 
 	"github.com/amirhnajafiz/bedrock-api/internal/configs"
 	"github.com/amirhnajafiz/bedrock-api/internal/logger"
+	"github.com/amirhnajafiz/bedrock-api/internal/metrics/httpmetrics"
+	"github.com/amirhnajafiz/bedrock-api/internal/metrics/sessionmetrics"
+	"github.com/amirhnajafiz/bedrock-api/internal/metrics/storagemetrics"
+	"github.com/amirhnajafiz/bedrock-api/internal/metrics/workermetrics"
+	"github.com/amirhnajafiz/bedrock-api/internal/metrics/zmqmetrics"
 	"github.com/amirhnajafiz/bedrock-api/internal/ports/http"
 	"github.com/amirhnajafiz/bedrock-api/internal/ports/zmq"
+	"github.com/amirhnajafiz/bedrock-api/internal/storage"
 	"github.com/amirhnajafiz/bedrock-api/internal/workers"
+	"github.com/amirhnajafiz/bedrock-api/pkg/metrics"
 
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/spf13/cobra"
 	"go.uber.org/zap"
 	"golang.org/x/sync/errgroup"
@@ -69,29 +77,74 @@ func StartAPI(ctx context.Context, cfg *configs.APIConfig) error {
 	// create a new logger instance
 	logr := logger.New(cfg.LogLevel)
 
+	// build the metrics server (skipped when disabled)
+	var metricsSrv *metrics.Server
+	if cfg.MetricsEnabled {
+		metricsAddr := fmt.Sprintf("%s:%d", cfg.MetricsHost, cfg.MetricsPort)
+		var err error
+		metricsSrv, err = metrics.NewServer(metricsAddr, logr.Named("metrics"))
+		if err != nil {
+			return fmt.Errorf("metrics server init: %w", err)
+		}
+		metrics.RegisterRuntime(metricsSrv.Registerer())
+	}
+
 	// create an errgroup with the provided context
 	erg, ectx := errgroup.WithContext(ctx)
 
+	// start the metrics server if enabled
+	if metricsSrv != nil {
+		erg.Go(func() error {
+			err := metricsSrv.Serve(ectx)
+			logr.Info("metrics server stopped", zap.Error(err))
+			return err
+		})
+	}
+
 	// parse durations
 	dockerdHealthCheckInterval, _ := time.ParseDuration(cfg.DockerDHealthCheckInterval)
 	sessionStatusCheckInterval, _ := time.ParseDuration(cfg.SessionStatusCheckInterval)
 
+	// resolve a registerer for component metric structs: real registry when
+	// the metrics server is on, throwaway registry otherwise.
+	var componentReg prometheus.Registerer
+	if metricsSrv != nil {
+		componentReg = metricsSrv.Registerer()
+	} else {
+		componentReg = prometheus.NewRegistry()
+	}
+
+	// build component metric structs
+	httpMetrics := httpmetrics.New(componentReg)
+	workerMetrics := workermetrics.New(componentReg)
+	zmqMetrics := zmqmetrics.New(componentReg)
+
+	// install the storage metrics decorator before any code path reaches
+	// storage.NewGoCache (workers / zmq.Build / http.Build all crystallize
+	// the singleton).
+	storageMetrics := storagemetrics.New(componentReg)
+	storage.SetMetricsDecorator(func(s storage.KVStorage) storage.KVStorage {
+		return storagemetrics.Wrap(s, storageMetrics)
+	})
+
+	sessionMetrics := sessionmetrics.New(componentReg)
+
 	// start the workers
 	dockerdHealthChannel := make(chan string)
 	erg.Go(func() error {
-		workers.WorkerDockerDHealthCheck(ectx, dockerdHealthChannel, logr.Named("dockerd-health"), dockerdHealthCheckInterval)
+		workers.WorkerDockerDHealthCheck(ectx, dockerdHealthChannel, logr.Named("dockerd-health"), dockerdHealthCheckInterval, workerMetrics, sessionMetrics)
 		logr.Info("docker daemon health check worker stopped")
 
 		return nil
 	})
 	erg.Go(func() error {
-		workers.WorkerCheckExpiredSessions(ectx, logr.Named("session-worker"), sessionStatusCheckInterval)
+		workers.WorkerCheckExpiredSessions(ectx, logr.Named("session-worker"), sessionStatusCheckInterval, workerMetrics, sessionMetrics)
 		logr.Info("session status check worker stopped")
 
 		return nil
 	})
 	erg.Go(func() error {
-		workers.WorkerRemoveFinishedSessions(ectx, logr.Named("cleanup-worker"), sessionStatusCheckInterval)
+		workers.WorkerRemoveFinishedSessions(ectx, logr.Named("cleanup-worker"), sessionStatusCheckInterval, workerMetrics)
 		logr.Info("cleanup worker stopped")
 
 		return nil
@@ -105,6 +158,8 @@ func StartAPI(ctx context.Context, cfg *configs.APIConfig) error {
 	}.Build(
 		zmqAddress,
 		cfg.SocketHandlers,
+		zmqMetrics,
+		sessionMetrics,
 	)
 	erg.Go(func() error {
 		err := zmqServer.Serve(ectx)
@@ -119,6 +174,8 @@ func StartAPI(ctx context.Context, cfg *configs.APIConfig) error {
 	}.Build(
 		fmt.Sprintf("%s:%d", cfg.HTTPHost, cfg.HTTPPort),
 		zmqAddress,
+		httpMetrics,
+		sessionMetrics,
 	)
 	erg.Go(func() error {
 		err := httpServer.Serve()

go.mod

@@ -8,6 +8,7 @@ require (
 	github.com/eko/gocache/lib/v4 v4.2.3
 	github.com/eko/gocache/store/go_cache/v4 v4.2.4
 	github.com/go-playground/validator/v10 v10.30.2
+	github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510
 	github.com/google/uuid v1.6.0
 	github.com/knadh/koanf/parsers/yaml v1.1.0
 	github.com/knadh/koanf/providers/file v1.2.1
@@ -16,6 +17,7 @@ require (
 	github.com/labstack/echo/v5 v5.1.0
 	github.com/opencontainers/image-spec v1.1.1
 	github.com/patrickmn/go-cache v2.1.0+incompatible
+	github.com/prometheus/client_golang v1.23.2
 	github.com/spf13/cobra v1.10.2
 	github.com/stretchr/testify v1.11.1
 	github.com/zeromq/goczmq v4.1.0+incompatible
@@ -42,9 +44,9 @@ require (
 	github.com/go-playground/locales v0.14.1 // indirect
 	github.com/go-playground/universal-translator v0.18.1 // indirect
 	github.com/go-viper/mapstructure/v2 v2.4.0 // indirect
-	github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/knadh/koanf/maps v0.1.2 // indirect
+	github.com/kylelemons/godebug v1.1.0 // indirect
 	github.com/leodido/go-urn v1.4.0 // indirect
 	github.com/mitchellh/copystructure v1.2.0 // indirect
 	github.com/mitchellh/reflectwalk v1.0.2 // indirect
@@ -56,7 +58,6 @@ require (
 	github.com/opencontainers/go-digest v1.0.0 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
-	github.com/prometheus/client_golang v1.23.2 // indirect
 	github.com/prometheus/client_model v0.6.2 // indirect
 	github.com/prometheus/common v0.67.4 // indirect
 	github.com/prometheus/procfs v0.19.2 // indirect

go.sum

@@ -62,6 +62,8 @@ github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 h1:HWRh5R2+9EifMyIHV7ZV+MIZqgz
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0/go.mod h1:JfhWUomR1baixubs02l85lZYYOm7LV6om4ceouMv45c=
 github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
 github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
+github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
+github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
 github.com/knadh/koanf/maps v0.1.2 h1:RBfmAW5CnZT+PJ1CVc1QSJKf4Xu9kxfQgYVQSu8hpbo=
 github.com/knadh/koanf/maps v0.1.2/go.mod h1:npD/QZY3V6ghQDdcQzl1W4ICNVTkohC8E73eI2xW4yI=
 github.com/knadh/koanf/parsers/yaml v1.1.0 h1:3ltfm9ljprAHt4jxgeYLlFPmUaunuCgu1yILuTXRdM4=

internal/configs/configs.go

@@ -24,6 +24,9 @@ type APIConfig struct {
 	DockerDHealthCheckInterval string `koanf:"dockerd_health_check_interval" validate:"duration"`
 	SessionStatusCheckInterval string `koanf:"session_status_check_interval" validate:"duration"`
 	BedrockTracerImage         string `koanf:"bedrock_tracer_image"`
+	MetricsEnabled             bool   `koanf:"metrics_enabled"`
+	MetricsHost                string `koanf:"metrics_host" validate:"omitempty,ip"`
+	MetricsPort                int    `koanf:"metrics_port" validate:"min=1,max=65535"`
 }
 
 // DockerdConfig represents the configuration for the Docker Daemon.

internal/configs/configs_test.go

@@ -0,0 +1,107 @@
+package configs_test
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/amirhnajafiz/bedrock-api/internal/configs"
+)
+
+func TestAPIConfigMetricsDefaults(t *testing.T) {
+	cfg := configs.DefaultAPIConfig()
+	if !cfg.MetricsEnabled {
+		t.Errorf("MetricsEnabled default = false, want true")
+	}
+	if cfg.MetricsHost != "127.0.0.1" {
+		t.Errorf("MetricsHost default = %q, want 127.0.0.1", cfg.MetricsHost)
+	}
+	if cfg.MetricsPort != 9090 {
+		t.Errorf("MetricsPort default = %d, want 9090", cfg.MetricsPort)
+	}
+}
+
+func TestLoadConfigParsesMetricsFields(t *testing.T) {
+	tmp := t.TempDir()
+	path := filepath.Join(tmp, "c.yaml")
+	contents := []byte(`api:
+  log_level: info
+  http_host: 127.0.0.1
+  http_port: 8080
+  socket_host: 127.0.0.1
+  socket_port: 8081
+  socket_handlers: 1
+  dockerd_health_check_interval: 15s
+  session_status_check_interval: 10s
+  metrics_enabled: false
+  metrics_host: 0.0.0.0
+  metrics_port: 9191
+`)
+	if err := os.WriteFile(path, contents, 0o644); err != nil {
+		t.Fatal(err)
+	}
+
+	cfg, err := configs.LoadConfig(path)
+	if err != nil {
+		t.Fatalf("LoadConfig: %v", err)
+	}
+	if cfg.API.MetricsEnabled {
+		t.Errorf("MetricsEnabled = true, want false")
+	}
+	if cfg.API.MetricsHost != "0.0.0.0" {
+		t.Errorf("MetricsHost = %q, want 0.0.0.0", cfg.API.MetricsHost)
+	}
+	if cfg.API.MetricsPort != 9191 {
+		t.Errorf("MetricsPort = %d, want 9191", cfg.API.MetricsPort)
+	}
+}
+
+func TestLoadConfigRejectsInvalidMetricsPort(t *testing.T) {
+	tmp := t.TempDir()
+	path := filepath.Join(tmp, "c.yaml")
+	contents := []byte(`api:
+  log_level: info
+  http_host: 127.0.0.1
+  http_port: 8080
+  socket_host: 127.0.0.1
+  socket_port: 8081
+  socket_handlers: 1
+  dockerd_health_check_interval: 15s
+  session_status_check_interval: 10s
+  metrics_enabled: true
+  metrics_host: 127.0.0.1
+  metrics_port: 70000
+`)
+	if err := os.WriteFile(path, contents, 0o644); err != nil {
+		t.Fatal(err)
+	}
+
+	if _, err := configs.LoadConfig(path); err == nil {
+		t.Fatal("LoadConfig: want error for out-of-range metrics_port, got nil")
+	}
+}
+
+func TestLoadConfigRejectsInvalidMetricsHost(t *testing.T) {
+	tmp := t.TempDir()
+	path := filepath.Join(tmp, "c.yaml")
+	contents := []byte(`api:
+  log_level: info
+  http_host: 127.0.0.1
+  http_port: 8080
+  socket_host: 127.0.0.1
+  socket_port: 8081
+  socket_handlers: 1
+  dockerd_health_check_interval: 15s
+  session_status_check_interval: 10s
+  metrics_enabled: true
+  metrics_host: not-an-ip
+  metrics_port: 9090
+`)
+	if err := os.WriteFile(path, contents, 0o644); err != nil {
+		t.Fatal(err)
+	}
+
+	if _, err := configs.LoadConfig(path); err == nil {
+		t.Fatal("LoadConfig: want error for invalid metrics_host, got nil")
+	}
+}

internal/configs/default.go

@@ -21,6 +21,9 @@ func DefaultAPIConfig() *APIConfig {
 		DockerDHealthCheckInterval: "1m",
 		SessionStatusCheckInterval: "30s",
 		BedrockTracerImage:         "ghcr.io/amirhnajafiz/bedrock-tracer:v0.0.7-beta",
+		MetricsEnabled:             true,
+		MetricsHost:                "127.0.0.1",
+		MetricsPort:                9090,
 	}
 }
 

internal/metrics/httpmetrics/metrics.go

@@ -0,0 +1,50 @@
+package httpmetrics
+
+import (
+	"github.com/amirhnajafiz/bedrock-api/pkg/metrics"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+// HTTPMetrics holds RED-style metrics for the Echo HTTP server.
+type HTTPMetrics struct {
+	requestsTotal   *prometheus.CounterVec
+	requestDuration *prometheus.HistogramVec
+	inFlight        prometheus.Gauge
+}
+
+// New constructs an HTTPMetrics, registering each collector into reg.
+func New(reg prometheus.Registerer) *HTTPMetrics {
+	m := &HTTPMetrics{
+		requestsTotal: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Namespace: "bedrock",
+			Subsystem: "api",
+			Name:      "http_requests_total",
+			Help:      "Total HTTP requests handled, by method, route template, and status class.",
+		}, []string{"method", "route", "status_class"}),
+
+		requestDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{
+			Namespace: "bedrock",
+			Subsystem: "api",
+			Name:      "http_request_duration_seconds",
+			Help:      "HTTP request duration in seconds, by method, route template, and status class.",
+			Buckets:   metrics.DefaultDurationBuckets,
+		}, []string{"method", "route", "status_class"}),
+
+		inFlight: prometheus.NewGauge(prometheus.GaugeOpts{
+			Namespace: "bedrock",
+			Subsystem: "api",
+			Name:      "http_requests_in_flight",
+			Help:      "Current number of HTTP requests being handled.",
+		}),
+	}
+
+	reg.MustRegister(m.requestsTotal, m.requestDuration, m.inFlight)
+	return m
+}
+
+// RequestsTotal exposes the counter for tests.
+func (m *HTTPMetrics) RequestsTotal() *prometheus.CounterVec { return m.requestsTotal }
+
+// RequestDuration exposes the histogram for tests.
+func (m *HTTPMetrics) RequestDuration() *prometheus.HistogramVec { return m.requestDuration }