diff --git a/cmd/api.go b/cmd/api.go index c14bda0..dc6e212 100644 --- a/cmd/api.go +++ b/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() diff --git a/go.mod b/go.mod index ca15b1d..9d48623 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index 41a78cc..e5c2c3f 100644 --- a/go.sum +++ b/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= diff --git a/internal/configs/configs.go b/internal/configs/configs.go index ddb8f26..45b0d6f 100644 --- a/internal/configs/configs.go +++ b/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. diff --git a/internal/configs/configs_test.go b/internal/configs/configs_test.go new file mode 100644 index 0000000..8798856 --- /dev/null +++ b/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") + } +} diff --git a/internal/configs/default.go b/internal/configs/default.go index 7e078e6..6e5a117 100644 --- a/internal/configs/default.go +++ b/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, } } diff --git a/internal/metrics/httpmetrics/metrics.go b/internal/metrics/httpmetrics/metrics.go new file mode 100644 index 0000000..b16be61 --- /dev/null +++ b/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 } diff --git a/internal/metrics/httpmetrics/middleware.go b/internal/metrics/httpmetrics/middleware.go new file mode 100644 index 0000000..501af8c --- /dev/null +++ b/internal/metrics/httpmetrics/middleware.go @@ -0,0 +1,72 @@ +package httpmetrics + +import ( + "strconv" + "time" + + "github.com/labstack/echo/v5" +) + +// Middleware returns an Echo middleware that records the configured HTTP RED +// metrics for every request. Routes are reported using Echo's route template +// (c.Path()); unmatched requests are bucketed as route="unknown". Status codes +// are collapsed into status classes (2xx, 3xx, 4xx, 5xx) to keep label +// cardinality bounded. +// +// Safe to call on a nil receiver: returns a passthrough middleware that +// records no metrics, matching the nil-safety contract of the other +// internal/metrics packages. +func (m *HTTPMetrics) Middleware() echo.MiddlewareFunc { + if m == nil { + return func(next echo.HandlerFunc) echo.HandlerFunc { + return next + } + } + return func(next echo.HandlerFunc) echo.HandlerFunc { + return func(c *echo.Context) error { + m.inFlight.Inc() + defer m.inFlight.Dec() + + start := time.Now() + err := next(c) + elapsed := time.Since(start).Seconds() + + route := c.Path() + if route == "" { + route = "unknown" + } + + method := c.Request().Method + // ResolveResponseStatus accounts for handlers that returned an + // error before writing a status (e.g. the default 404 path) by + // consulting the error's StatusCode, matching what Echo's + // HTTPErrorHandler is about to send to the wire. + _, status := echo.ResolveResponseStatus(c.Response(), err) + class := statusClass(status) + + m.requestsTotal.WithLabelValues(method, route, class).Inc() + m.requestDuration.WithLabelValues(method, route, class).Observe(elapsed) + + return err + } + } +} + +// statusClass returns "2xx" / "3xx" / "4xx" / "5xx" for a status code. +// Codes outside 100-599 are reported as their raw decimal form so that any +// unusual sentinel codes are still visible without inflating cardinality +// under normal operation. +func statusClass(status int) string { + switch { + case status >= 200 && status < 300: + return "2xx" + case status >= 300 && status < 400: + return "3xx" + case status >= 400 && status < 500: + return "4xx" + case status >= 500 && status < 600: + return "5xx" + default: + return strconv.Itoa(status) + } +} diff --git a/internal/metrics/httpmetrics/middleware_test.go b/internal/metrics/httpmetrics/middleware_test.go new file mode 100644 index 0000000..1f5d0fb --- /dev/null +++ b/internal/metrics/httpmetrics/middleware_test.go @@ -0,0 +1,102 @@ +package httpmetrics_test + +import ( + "net/http" + "net/http/httptest" + "testing" + + "github.com/amirhnajafiz/bedrock-api/internal/metrics/httpmetrics" + + "github.com/labstack/echo/v5" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/testutil" +) + +func setup(t *testing.T) (*httpmetrics.HTTPMetrics, *echo.Echo) { + t.Helper() + reg := prometheus.NewRegistry() + m := httpmetrics.New(reg) + + e := echo.New() + e.Use(m.Middleware()) + e.GET("/api/sessions/:id", func(c *echo.Context) error { + return c.String(http.StatusOK, "ok") + }) + e.POST("/api/sessions/:id", func(c *echo.Context) error { + return c.String(http.StatusInternalServerError, "boom") + }) + return m, e +} + +func TestMiddlewareCountsRequestsByRouteAndStatusClass(t *testing.T) { + m, e := setup(t) + + req := httptest.NewRequest(http.MethodGet, "/api/sessions/abc-123", nil) + rec := httptest.NewRecorder() + e.ServeHTTP(rec, req) + + got := testutil.ToFloat64(m.RequestsTotal().WithLabelValues(http.MethodGet, "/api/sessions/:id", "2xx")) + if got != 1 { + t.Errorf("requests_total{GET,/api/sessions/:id,2xx} = %v, want 1", got) + } +} + +func TestMiddlewareBucketsStatusClasses(t *testing.T) { + m, e := setup(t) + + req := httptest.NewRequest(http.MethodPost, "/api/sessions/x", nil) + rec := httptest.NewRecorder() + e.ServeHTTP(rec, req) + + got := testutil.ToFloat64(m.RequestsTotal().WithLabelValues(http.MethodPost, "/api/sessions/:id", "5xx")) + if got != 1 { + t.Errorf("requests_total{POST,/api/sessions/:id,5xx} = %v, want 1", got) + } +} + +func TestMiddlewareUsesRouteTemplateNotRawPath(t *testing.T) { + m, e := setup(t) + + for _, id := range []string{"abc", "def", "ghi"} { + req := httptest.NewRequest(http.MethodGet, "/api/sessions/"+id, nil) + e.ServeHTTP(httptest.NewRecorder(), req) + } + + // All three should collapse into the same label set. + got := testutil.ToFloat64(m.RequestsTotal().WithLabelValues(http.MethodGet, "/api/sessions/:id", "2xx")) + if got != 3 { + t.Errorf("requests_total = %v, want 3 (one label set, not three)", got) + } +} + +func TestMiddlewareBucketsUnknownRoutes(t *testing.T) { + m, e := setup(t) + + req := httptest.NewRequest(http.MethodGet, "/does/not/exist", nil) + rec := httptest.NewRecorder() + e.ServeHTTP(rec, req) + + got := testutil.ToFloat64(m.RequestsTotal().WithLabelValues(http.MethodGet, "unknown", "4xx")) + if got != 1 { + t.Errorf("requests_total{unknown,4xx} = %v, want 1", got) + } +} + +func TestMiddlewareIsNilSafe(t *testing.T) { + var m *httpmetrics.HTTPMetrics + mw := m.Middleware() + + e := echo.New() + e.Use(mw) + e.GET("/x", func(c *echo.Context) error { + return c.String(http.StatusOK, "ok") + }) + + req := httptest.NewRequest(http.MethodGet, "/x", nil) + rec := httptest.NewRecorder() + e.ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + t.Errorf("status = %d, want 200", rec.Code) + } +} diff --git a/internal/metrics/sessionmetrics/metrics.go b/internal/metrics/sessionmetrics/metrics.go new file mode 100644 index 0000000..aa4a0ba --- /dev/null +++ b/internal/metrics/sessionmetrics/metrics.go @@ -0,0 +1,144 @@ +package sessionmetrics + +import ( + "time" + + "github.com/amirhnajafiz/bedrock-api/pkg/enums" + + "github.com/prometheus/client_golang/prometheus" +) + +// SessionMetrics holds counters, histograms, and gauges describing the +// lifecycle of sessions and the DockerD worker pool. +type SessionMetrics struct { + sessionsByStatus *prometheus.GaugeVec + sessionsCreated prometheus.Counter + sessionsCompleted *prometheus.CounterVec + sessionDuration prometheus.Histogram + dockerdWorkers *prometheus.GaugeVec + dockerdSessions *prometheus.GaugeVec +} + +// sessionDurationBuckets covers TTLs from seconds to a day, since sessions are +// user-defined workloads with widely varying lifetimes (config.yaml shows TTLs +// like 10s ... 24h). +var sessionDurationBuckets = []float64{ + 1, 5, 15, 30, 60, 5 * 60, 15 * 60, 60 * 60, 6 * 60 * 60, 24 * 60 * 60, +} + +// New constructs and registers SessionMetrics. +func New(reg prometheus.Registerer) *SessionMetrics { + m := &SessionMetrics{ + sessionsByStatus: prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Namespace: "bedrock", + Subsystem: "api", + Name: "sessions_by_status", + Help: "Number of sessions currently in each status, reconciled periodically from KV.", + }, []string{"status"}), + sessionsCreated: prometheus.NewCounter(prometheus.CounterOpts{ + Namespace: "bedrock", + Subsystem: "api", + Name: "sessions_created_total", + Help: "Total sessions created.", + }), + sessionsCompleted: prometheus.NewCounterVec(prometheus.CounterOpts{ + Namespace: "bedrock", + Subsystem: "api", + Name: "sessions_completed_total", + Help: "Total sessions that reached a terminal status, by terminal_status.", + }, []string{"terminal_status"}), + sessionDuration: prometheus.NewHistogram(prometheus.HistogramOpts{ + Namespace: "bedrock", + Subsystem: "api", + Name: "session_duration_seconds", + Help: "Wall-clock duration of a session from creation to terminal transition.", + Buckets: sessionDurationBuckets, + }), + dockerdWorkers: prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Namespace: "bedrock", + Subsystem: "api", + Name: "dockerd_workers", + Help: "DockerD worker pool size, by state (registered/healthy).", + }, []string{"state"}), + dockerdSessions: prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Namespace: "bedrock", + Subsystem: "api", + Name: "dockerd_sessions", + Help: "Number of non-terminal sessions currently assigned to each DockerD instance.", + }, []string{"dockerd_id"}), + } + reg.MustRegister( + m.sessionsByStatus, + m.sessionsCreated, + m.sessionsCompleted, + m.sessionDuration, + m.dockerdWorkers, + m.dockerdSessions, + ) + return m +} + +// IncCreated records one newly-created session. Safe to call on a nil receiver. +func (m *SessionMetrics) IncCreated() { + if m == nil { + return + } + m.sessionsCreated.Inc() +} + +// ObserveCompleted records a session entering a terminal state with the +// elapsed wall-clock time since creation. Only terminal statuses produce +// observations; other statuses are silently ignored to keep callers simple. +// Safe to call on a nil receiver. +func (m *SessionMetrics) ObserveCompleted(status enums.SessionStatus, d time.Duration) { + if m == nil { + return + } + switch status { + case enums.SessionStatusFinished, enums.SessionStatusFailed, enums.SessionStatusStopped: + m.sessionsCompleted.WithLabelValues(string(status)).Inc() + m.sessionDuration.Observe(d.Seconds()) + } +} + +// SetByStatus sets the current count of sessions in the given status. +// Safe to call on a nil receiver. +func (m *SessionMetrics) SetByStatus(status enums.SessionStatus, count float64) { + if m == nil { + return + } + m.sessionsByStatus.WithLabelValues(string(status)).Set(count) +} + +// SetDockerdWorkers sets the count of DockerDs in the given state. +// state must be "registered" or "healthy". Safe to call on a nil receiver. +func (m *SessionMetrics) SetDockerdWorkers(state string, count float64) { + if m == nil { + return + } + m.dockerdWorkers.WithLabelValues(state).Set(count) +} + +// SetDockerdSessions sets the number of non-terminal sessions assigned to dockerdID. +// Safe to call on a nil receiver. +func (m *SessionMetrics) SetDockerdSessions(dockerdID string, count float64) { + if m == nil { + return + } + m.dockerdSessions.WithLabelValues(dockerdID).Set(count) +} + +// SessionsCreated exposes the counter for tests. +func (m *SessionMetrics) SessionsCreated() prometheus.Counter { return m.sessionsCreated } + +// SessionsCompleted exposes the counter for tests. +func (m *SessionMetrics) SessionsCompleted() *prometheus.CounterVec { return m.sessionsCompleted } + +// SessionsByStatus exposes the gauge for tests. +func (m *SessionMetrics) SessionsByStatus() *prometheus.GaugeVec { return m.sessionsByStatus } + +// DockerdWorkers exposes the gauge for tests. +func (m *SessionMetrics) DockerdWorkers() *prometheus.GaugeVec { return m.dockerdWorkers } + +// DockerdSessions exposes the gauge for tests. +func (m *SessionMetrics) DockerdSessions() *prometheus.GaugeVec { return m.dockerdSessions } diff --git a/internal/metrics/sessionmetrics/metrics_test.go b/internal/metrics/sessionmetrics/metrics_test.go new file mode 100644 index 0000000..fe0bdf0 --- /dev/null +++ b/internal/metrics/sessionmetrics/metrics_test.go @@ -0,0 +1,91 @@ +package sessionmetrics_test + +import ( + "testing" + "time" + + "github.com/amirhnajafiz/bedrock-api/internal/metrics/sessionmetrics" + "github.com/amirhnajafiz/bedrock-api/pkg/enums" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/testutil" +) + +func TestIncCreatedCounter(t *testing.T) { + reg := prometheus.NewRegistry() + m := sessionmetrics.New(reg) + + m.IncCreated() + m.IncCreated() + m.IncCreated() + + if got := testutil.ToFloat64(m.SessionsCreated()); got != 3 { + t.Errorf("sessions_created_total = %v, want 3", got) + } +} + +func TestObserveCompletedRecordsCounterAndHistogram(t *testing.T) { + reg := prometheus.NewRegistry() + m := sessionmetrics.New(reg) + + m.ObserveCompleted(enums.SessionStatusFinished, 12*time.Second) + m.ObserveCompleted(enums.SessionStatusFailed, 3*time.Second) + m.ObserveCompleted(enums.SessionStatusFinished, 7*time.Second) + + if got := testutil.ToFloat64(m.SessionsCompleted().WithLabelValues("finished")); got != 2 { + t.Errorf("sessions_completed_total{finished} = %v, want 2", got) + } + if got := testutil.ToFloat64(m.SessionsCompleted().WithLabelValues("failed")); got != 1 { + t.Errorf("sessions_completed_total{failed} = %v, want 1", got) + } +} + +func TestObserveCompletedIgnoresNonTerminalStatuses(t *testing.T) { + reg := prometheus.NewRegistry() + m := sessionmetrics.New(reg) + + m.ObserveCompleted(enums.SessionStatusPending, time.Second) + m.ObserveCompleted(enums.SessionStatusRunning, time.Second) + + // No counter for these statuses; no histogram observation either. + if got := testutil.ToFloat64(m.SessionsCompleted().WithLabelValues("pending")); got != 0 { + t.Errorf("sessions_completed_total{pending} = %v, want 0", got) + } +} + +func TestSetByStatusOverwritesGauge(t *testing.T) { + reg := prometheus.NewRegistry() + m := sessionmetrics.New(reg) + + m.SetByStatus(enums.SessionStatusRunning, 5) + m.SetByStatus(enums.SessionStatusRunning, 2) + + if got := testutil.ToFloat64(m.SessionsByStatus().WithLabelValues("running")); got != 2 { + t.Errorf("sessions_by_status{running} = %v, want 2", got) + } +} + +func TestSetDockerdWorkersAndSessions(t *testing.T) { + reg := prometheus.NewRegistry() + m := sessionmetrics.New(reg) + + m.SetDockerdWorkers("registered", 3) + m.SetDockerdWorkers("healthy", 2) + m.SetDockerdSessions("dockerd-1", 4) + + if got := testutil.ToFloat64(m.DockerdWorkers().WithLabelValues("healthy")); got != 2 { + t.Errorf("dockerd_workers{healthy} = %v, want 2", got) + } + if got := testutil.ToFloat64(m.DockerdSessions().WithLabelValues("dockerd-1")); got != 4 { + t.Errorf("dockerd_sessions{dockerd-1} = %v, want 4", got) + } +} + +func TestMethodsAreNilSafe(t *testing.T) { + var m *sessionmetrics.SessionMetrics + m.IncCreated() + m.ObserveCompleted(enums.SessionStatusFinished, time.Second) + m.SetByStatus(enums.SessionStatusRunning, 1) + m.SetDockerdWorkers("healthy", 1) + m.SetDockerdSessions("d1", 1) +} diff --git a/internal/metrics/storagemetrics/metrics.go b/internal/metrics/storagemetrics/metrics.go new file mode 100644 index 0000000..a1a575e --- /dev/null +++ b/internal/metrics/storagemetrics/metrics.go @@ -0,0 +1,95 @@ +package storagemetrics + +import ( + "time" + + "github.com/amirhnajafiz/bedrock-api/internal/storage" + "github.com/amirhnajafiz/bedrock-api/pkg/metrics" + + "github.com/prometheus/client_golang/prometheus" +) + +// StorageMetrics records KV storage operation counts and durations. +type StorageMetrics struct { + opsTotal *prometheus.CounterVec + opDuration *prometheus.HistogramVec +} + +// New constructs and registers StorageMetrics. +func New(reg prometheus.Registerer) *StorageMetrics { + m := &StorageMetrics{ + opsTotal: prometheus.NewCounterVec(prometheus.CounterOpts{ + Namespace: "bedrock", + Subsystem: "api", + Name: "storage_ops_total", + Help: "Total KV storage operations, by op (get/set/delete/list) and result (ok/err).", + }, []string{"op", "result"}), + opDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Namespace: "bedrock", + Subsystem: "api", + Name: "storage_op_duration_seconds", + Help: "Duration of KV storage operations in seconds, by op.", + Buckets: metrics.DefaultDurationBuckets, + }, []string{"op"}), + } + reg.MustRegister(m.opsTotal, m.opDuration) + return m +} + +// OpsTotal exposes the counter for tests. +func (m *StorageMetrics) OpsTotal() *prometheus.CounterVec { return m.opsTotal } + +// OpDuration exposes the histogram for tests. +func (m *StorageMetrics) OpDuration() *prometheus.HistogramVec { return m.opDuration } + +// Wrap returns a storage.KVStorage that records operation metrics into m +// before delegating to inner. If m is nil, inner is returned unchanged +// (no decorator, no overhead). +func Wrap(inner storage.KVStorage, m *StorageMetrics) storage.KVStorage { + if m == nil { + return inner + } + return &decorator{inner: inner, m: m} +} + +type decorator struct { + inner storage.KVStorage + m *StorageMetrics +} + +func (d *decorator) Set(key string, value []byte) error { + start := time.Now() + err := d.inner.Set(key, value) + d.observe("set", err, start) + return err +} + +func (d *decorator) Get(key string) ([]byte, error) { + start := time.Now() + v, err := d.inner.Get(key) + d.observe("get", err, start) + return v, err +} + +func (d *decorator) Delete(key string) error { + start := time.Now() + err := d.inner.Delete(key) + d.observe("delete", err, start) + return err +} + +func (d *decorator) List(wildcard string) ([][]byte, error) { + start := time.Now() + v, err := d.inner.List(wildcard) + d.observe("list", err, start) + return v, err +} + +func (d *decorator) observe(op string, err error, start time.Time) { + result := "ok" + if err != nil { + result = "err" + } + d.m.opsTotal.WithLabelValues(op, result).Inc() + d.m.opDuration.WithLabelValues(op).Observe(time.Since(start).Seconds()) +} diff --git a/internal/metrics/storagemetrics/metrics_test.go b/internal/metrics/storagemetrics/metrics_test.go new file mode 100644 index 0000000..13728ab --- /dev/null +++ b/internal/metrics/storagemetrics/metrics_test.go @@ -0,0 +1,63 @@ +package storagemetrics_test + +import ( + "errors" + "testing" + + "github.com/amirhnajafiz/bedrock-api/internal/metrics/storagemetrics" + "github.com/amirhnajafiz/bedrock-api/internal/storage" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/testutil" +) + +// fakeKV is an in-memory KVStorage that lets us drive specific success / error +// branches deterministically. +type fakeKV struct { + getErr error +} + +func (f *fakeKV) Set(key string, value []byte) error { return nil } +func (f *fakeKV) Get(key string) ([]byte, error) { return nil, f.getErr } +func (f *fakeKV) Delete(key string) error { return nil } +func (f *fakeKV) List(wildcard string) ([][]byte, error) { return nil, nil } + +var _ storage.KVStorage = (*fakeKV)(nil) + +func TestWrapCountsOpsByResult(t *testing.T) { + reg := prometheus.NewRegistry() + m := storagemetrics.New(reg) + + kv := storagemetrics.Wrap(&fakeKV{}, m) + + _ = kv.Set("k", []byte("v")) + _ = kv.Set("k", []byte("v")) + _, _ = kv.Get("k") + + if got := testutil.ToFloat64(m.OpsTotal().WithLabelValues("set", "ok")); got != 2 { + t.Errorf("ops_total{set,ok} = %v, want 2", got) + } + if got := testutil.ToFloat64(m.OpsTotal().WithLabelValues("get", "ok")); got != 1 { + t.Errorf("ops_total{get,ok} = %v, want 1", got) + } +} + +func TestWrapCountsErrors(t *testing.T) { + reg := prometheus.NewRegistry() + m := storagemetrics.New(reg) + + kv := storagemetrics.Wrap(&fakeKV{getErr: errors.New("boom")}, m) + _, _ = kv.Get("k") + + if got := testutil.ToFloat64(m.OpsTotal().WithLabelValues("get", "err")); got != 1 { + t.Errorf("ops_total{get,err} = %v, want 1", got) + } +} + +func TestWrapWithNilMetricsReturnsInner(t *testing.T) { + inner := &fakeKV{} + got := storagemetrics.Wrap(inner, nil) + if got != storage.KVStorage(inner) { + t.Errorf("Wrap(inner, nil) should return inner unchanged") + } +} diff --git a/internal/metrics/workermetrics/metrics.go b/internal/metrics/workermetrics/metrics.go new file mode 100644 index 0000000..6585286 --- /dev/null +++ b/internal/metrics/workermetrics/metrics.go @@ -0,0 +1,70 @@ +package workermetrics + +import ( + "time" + + "github.com/amirhnajafiz/bedrock-api/pkg/metrics" + + "github.com/prometheus/client_golang/prometheus" +) + +// WorkerMetrics records execution metrics for background ticker-based workers. +type WorkerMetrics struct { + ticksTotal *prometheus.CounterVec + errorsTotal *prometheus.CounterVec + tickDuration *prometheus.HistogramVec +} + +// New constructs and registers WorkerMetrics. The `worker` label takes one of +// a small bounded set of values defined by the caller (e.g. "expired_sessions", +// "dockerd_health", "cleanup"). +func New(reg prometheus.Registerer) *WorkerMetrics { + m := &WorkerMetrics{ + ticksTotal: prometheus.NewCounterVec(prometheus.CounterOpts{ + Namespace: "bedrock", + Subsystem: "api", + Name: "worker_ticks_total", + Help: "Total number of ticks executed by background workers, by worker name.", + }, []string{"worker"}), + errorsTotal: prometheus.NewCounterVec(prometheus.CounterOpts{ + Namespace: "bedrock", + Subsystem: "api", + Name: "worker_errors_total", + Help: "Total number of errors encountered by background workers, by worker name.", + }, []string{"worker"}), + tickDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Namespace: "bedrock", + Subsystem: "api", + Name: "worker_tick_duration_seconds", + Help: "Duration of a single worker tick in seconds, by worker name.", + Buckets: metrics.DefaultDurationBuckets, + }, []string{"worker"}), + } + reg.MustRegister(m.ticksTotal, m.errorsTotal, m.tickDuration) + return m +} + +// ObserveTick records a completed tick of the named worker. +// It is safe to call on a nil receiver; that no-ops. +func (m *WorkerMetrics) ObserveTick(worker string, d time.Duration) { + if m == nil { + return + } + m.ticksTotal.WithLabelValues(worker).Inc() + m.tickDuration.WithLabelValues(worker).Observe(d.Seconds()) +} + +// IncError increments the error counter for the named worker. +// It is safe to call on a nil receiver; that no-ops. +func (m *WorkerMetrics) IncError(worker string) { + if m == nil { + return + } + m.errorsTotal.WithLabelValues(worker).Inc() +} + +// TicksTotal exposes the counter for tests. +func (m *WorkerMetrics) TicksTotal() *prometheus.CounterVec { return m.ticksTotal } + +// ErrorsTotal exposes the counter for tests. +func (m *WorkerMetrics) ErrorsTotal() *prometheus.CounterVec { return m.errorsTotal } diff --git a/internal/metrics/workermetrics/metrics_test.go b/internal/metrics/workermetrics/metrics_test.go new file mode 100644 index 0000000..0c1ed7e --- /dev/null +++ b/internal/metrics/workermetrics/metrics_test.go @@ -0,0 +1,46 @@ +package workermetrics_test + +import ( + "testing" + "time" + + "github.com/amirhnajafiz/bedrock-api/internal/metrics/workermetrics" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/testutil" +) + +func TestObserveTickIncrementsCounterAndHistogram(t *testing.T) { + reg := prometheus.NewRegistry() + m := workermetrics.New(reg) + + m.ObserveTick("expired_sessions", 50*time.Millisecond) + m.ObserveTick("expired_sessions", 75*time.Millisecond) + + if got := testutil.ToFloat64(m.TicksTotal().WithLabelValues("expired_sessions")); got != 2 { + t.Errorf("ticks_total = %v, want 2", got) + } +} + +func TestIncErrorIncrementsErrorCounter(t *testing.T) { + reg := prometheus.NewRegistry() + m := workermetrics.New(reg) + + m.IncError("cleanup") + m.IncError("cleanup") + m.IncError("dockerd_health") + + if got := testutil.ToFloat64(m.ErrorsTotal().WithLabelValues("cleanup")); got != 2 { + t.Errorf("errors_total{cleanup} = %v, want 2", got) + } + if got := testutil.ToFloat64(m.ErrorsTotal().WithLabelValues("dockerd_health")); got != 1 { + t.Errorf("errors_total{dockerd_health} = %v, want 1", got) + } +} + +func TestMethodsAreNilSafe(t *testing.T) { + // Calling ObserveTick or IncError on a nil receiver must not panic. + var m *workermetrics.WorkerMetrics + m.ObserveTick("any", time.Millisecond) + m.IncError("any") +} diff --git a/internal/metrics/zmqmetrics/metrics.go b/internal/metrics/zmqmetrics/metrics.go new file mode 100644 index 0000000..1f892ea --- /dev/null +++ b/internal/metrics/zmqmetrics/metrics.go @@ -0,0 +1,58 @@ +package zmqmetrics + +import ( + "github.com/amirhnajafiz/bedrock-api/pkg/enums" + + "github.com/prometheus/client_golang/prometheus" +) + +// ZMQMetrics records counters for messages flowing through the ZMQ server. +type ZMQMetrics struct { + messagesTotal *prometheus.CounterVec + errorsTotal *prometheus.CounterVec +} + +// New constructs and registers ZMQMetrics. The `kind` label values are drawn +// from the bounded enums.EventType enum defined in pkg/enums/events.go, so +// cardinality is fixed at compile time. +func New(reg prometheus.Registerer) *ZMQMetrics { + m := &ZMQMetrics{ + messagesTotal: prometheus.NewCounterVec(prometheus.CounterOpts{ + Namespace: "bedrock", + Subsystem: "api", + Name: "zmq_messages_total", + Help: "Total ZMQ messages, by direction (in/out) and event kind.", + }, []string{"direction", "kind"}), + errorsTotal: prometheus.NewCounterVec(prometheus.CounterOpts{ + Namespace: "bedrock", + Subsystem: "api", + Name: "zmq_errors_total", + Help: "Total ZMQ errors, by direction (in/out).", + }, []string{"direction"}), + } + reg.MustRegister(m.messagesTotal, m.errorsTotal) + return m +} + +// IncMessage records one message of the given kind in the given direction. +// direction must be "in" or "out". Safe to call on a nil receiver. +func (m *ZMQMetrics) IncMessage(direction string, kind enums.EventType) { + if m == nil { + return + } + m.messagesTotal.WithLabelValues(direction, string(kind)).Inc() +} + +// IncError records one error in the given direction. Safe to call on a nil receiver. +func (m *ZMQMetrics) IncError(direction string) { + if m == nil { + return + } + m.errorsTotal.WithLabelValues(direction).Inc() +} + +// MessagesTotal exposes the counter for tests. +func (m *ZMQMetrics) MessagesTotal() *prometheus.CounterVec { return m.messagesTotal } + +// ErrorsTotal exposes the counter for tests. +func (m *ZMQMetrics) ErrorsTotal() *prometheus.CounterVec { return m.errorsTotal } diff --git a/internal/metrics/zmqmetrics/metrics_test.go b/internal/metrics/zmqmetrics/metrics_test.go new file mode 100644 index 0000000..3d72ded --- /dev/null +++ b/internal/metrics/zmqmetrics/metrics_test.go @@ -0,0 +1,47 @@ +package zmqmetrics_test + +import ( + "testing" + + "github.com/amirhnajafiz/bedrock-api/internal/metrics/zmqmetrics" + "github.com/amirhnajafiz/bedrock-api/pkg/enums" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/testutil" +) + +func TestIncMessageCountsByDirectionAndKind(t *testing.T) { + reg := prometheus.NewRegistry() + m := zmqmetrics.New(reg) + + m.IncMessage("in", enums.EventTypeSessionRunning) + m.IncMessage("in", enums.EventTypeSessionRunning) + m.IncMessage("out", enums.EventTypeSessionStart) + + if got := testutil.ToFloat64(m.MessagesTotal().WithLabelValues("in", string(enums.EventTypeSessionRunning))); got != 2 { + t.Errorf("messages_total{in,session_running} = %v, want 2", got) + } + if got := testutil.ToFloat64(m.MessagesTotal().WithLabelValues("out", string(enums.EventTypeSessionStart))); got != 1 { + t.Errorf("messages_total{out,session_start} = %v, want 1", got) + } +} + +func TestIncErrorCountsByDirection(t *testing.T) { + reg := prometheus.NewRegistry() + m := zmqmetrics.New(reg) + + m.IncError("in") + m.IncError("out") + m.IncError("out") + + if got := testutil.ToFloat64(m.ErrorsTotal().WithLabelValues("out")); got != 2 { + t.Errorf("errors_total{out} = %v, want 2", got) + } +} + +func TestMethodsAreNilSafe(t *testing.T) { + // Calling IncMessage or IncError on a nil receiver must not panic. + var m *zmqmetrics.ZMQMetrics + m.IncMessage("in", enums.EventTypeSessionRunning) + m.IncError("in") +} diff --git a/internal/ports/http/handlers.go b/internal/ports/http/handlers.go index 7b8a423..2ea4c42 100644 --- a/internal/ports/http/handlers.go +++ b/internal/ports/http/handlers.go @@ -62,6 +62,8 @@ func (h HTTPServer) createSession(c *echo.Context) error { return c.String(http.StatusInternalServerError, "failed to save session") } + h.sessionMetrics.IncCreated() + return c.JSON(http.StatusCreated, ToResponseSession(session)) } @@ -98,6 +100,8 @@ func (h HTTPServer) updateSession(c *echo.Context) error { return c.String(http.StatusInternalServerError, "failed to save session") } + h.sessionMetrics.ObserveCompleted(session.Status, time.Since(session.CreatedAt)) + return c.JSON(http.StatusOK, ToResponseSession(session)) } diff --git a/internal/ports/http/http.go b/internal/ports/http/http.go index 1b76f3c..323ac3d 100644 --- a/internal/ports/http/http.go +++ b/internal/ports/http/http.go @@ -6,6 +6,8 @@ import ( "github.com/amirhnajafiz/bedrock-api/internal/components/logs" "github.com/amirhnajafiz/bedrock-api/internal/components/sessions" zmqclient "github.com/amirhnajafiz/bedrock-api/internal/components/zmq_client" + "github.com/amirhnajafiz/bedrock-api/internal/metrics/httpmetrics" + "github.com/amirhnajafiz/bedrock-api/internal/metrics/sessionmetrics" "github.com/amirhnajafiz/bedrock-api/internal/scheduler" "github.com/amirhnajafiz/bedrock-api/internal/state_machine" "github.com/amirhnajafiz/bedrock-api/internal/storage" @@ -21,17 +23,21 @@ type HTTPServer struct { Logr *zap.Logger // private modules - address string - scheduler scheduler.Scheduler - sessionStore sessions.SessionStore - logStore logs.LogStore - zclient *zmqclient.ZMQClient - stateMachine *statemachine.StateMachine + address string + scheduler scheduler.Scheduler + sessionStore sessions.SessionStore + logStore logs.LogStore + zclient *zmqclient.ZMQClient + stateMachine *statemachine.StateMachine + httpMetrics *httpmetrics.HTTPMetrics + sessionMetrics *sessionmetrics.SessionMetrics } // NewHTTPServer creates and returns a new instance of HTTPServer. -func (h HTTPServer) Build(address, socketAddress string) *HTTPServer { +func (h HTTPServer) Build(address, socketAddress string, httpMetrics *httpmetrics.HTTPMetrics, sessionMetrics *sessionmetrics.SessionMetrics) *HTTPServer { h.address = address + h.httpMetrics = httpMetrics + h.sessionMetrics = sessionMetrics h.scheduler = scheduler.NewRoundRobin() h.sessionStore = sessions.NewSessionStore(storage.NewGoCache()) @@ -49,6 +55,10 @@ func (h HTTPServer) Serve() error { // set the health handler e.GET("/health", h.health) + // metrics middleware must be registered before other middleware so it + // observes the full handler duration including downstream middleware. + e.Use(h.httpMetrics.Middleware()) + // set the middlewares e.Use(middleware.RequestLoggerWithConfig(middleware.RequestLoggerConfig{ LogURI: true, diff --git a/internal/ports/zmq/handlers.go b/internal/ports/zmq/handlers.go index 757b26d..acef68b 100644 --- a/internal/ports/zmq/handlers.go +++ b/internal/ports/zmq/handlers.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "time" "github.com/amirhnajafiz/bedrock-api/pkg/enums" "github.com/amirhnajafiz/bedrock-api/pkg/models" @@ -31,6 +32,7 @@ func (z ZMQServer) socketReceiver(ctx context.Context, router *goczmq.Sock, chan if err != nil { if !errors.Is(err, goczmq.ErrTimeout) && !errors.Is(err, goczmq.ErrRecvFrame) { z.Logr.Warn("failed to receive message", zap.Error(err)) + z.zmqMetrics.IncError("in") } continue @@ -56,6 +58,7 @@ func (z ZMQServer) socketSender(ctx context.Context, router *goczmq.Sock, channe case event := <-channel: if err := router.SendMessage(event); err != nil { z.Logr.Warn("failed to send message", zap.Error(err)) + z.zmqMetrics.IncError("out") return fmt.Errorf("sender router failed: %v", err) } @@ -127,6 +130,8 @@ func (z ZMQServer) processPacketData(raw []byte) []byte { // read events from packet events and update session status in KV storage accordingly for _, event := range pkt.Events { + z.zmqMetrics.IncMessage("in", event.GetEventType()) + // get the session id from header sid := event.GetSessionId() @@ -157,7 +162,9 @@ func (z ZMQServer) processPacketData(raw []byte) []byte { } // use state machine for safe state transition - record.Status = z.stateMachine.Transition(record.Status, newSessionStatus) + priorStatus := record.Status + newStatus := z.stateMachine.Transition(record.Status, newSessionStatus) + record.Status = newStatus // update the session in KV storage if err := z.sessionStore.SaveSession(record); err != nil { @@ -169,6 +176,10 @@ func (z ZMQServer) processPacketData(raw []byte) []byte { ) continue } + + if newStatus != priorStatus { + z.sessionMetrics.ObserveCompleted(newStatus, time.Since(record.CreatedAt)) + } } // retrieve all sessions of the sender daemon from KV storage @@ -218,6 +229,10 @@ func (z ZMQServer) processPacketData(raw []byte) []byte { } } + for _, ev := range events { + z.zmqMetrics.IncMessage("out", ev.GetEventType()) + } + // send the response packet back to the sender return responsePkt.WithEvents(events...).ToBytes() } diff --git a/internal/ports/zmq/zmq.go b/internal/ports/zmq/zmq.go index 6008c11..c8a7518 100644 --- a/internal/ports/zmq/zmq.go +++ b/internal/ports/zmq/zmq.go @@ -5,6 +5,8 @@ import ( "fmt" "github.com/amirhnajafiz/bedrock-api/internal/components/sessions" + "github.com/amirhnajafiz/bedrock-api/internal/metrics/sessionmetrics" + "github.com/amirhnajafiz/bedrock-api/internal/metrics/zmqmetrics" statemachine "github.com/amirhnajafiz/bedrock-api/internal/state_machine" "github.com/amirhnajafiz/bedrock-api/internal/storage" @@ -21,16 +23,20 @@ type ZMQServer struct { Logr *zap.Logger // private modules - eventHandlers int - address string - sessionStore sessions.SessionStore - stateMachine *statemachine.StateMachine + eventHandlers int + address string + sessionStore sessions.SessionStore + stateMachine *statemachine.StateMachine + zmqMetrics *zmqmetrics.ZMQMetrics + sessionMetrics *sessionmetrics.SessionMetrics } // Build initializes the ZMQServer with the specified address and returns the server instance. -func (z ZMQServer) Build(address string, eventHandlers int) *ZMQServer { +func (z ZMQServer) Build(address string, eventHandlers int, zmqMetrics *zmqmetrics.ZMQMetrics, sessionMetrics *sessionmetrics.SessionMetrics) *ZMQServer { z.address = address z.eventHandlers = eventHandlers + z.zmqMetrics = zmqMetrics + z.sessionMetrics = sessionMetrics z.sessionStore = sessions.NewSessionStore(storage.NewGoCache()) z.stateMachine = statemachine.NewStateMachine() diff --git a/internal/storage/storage.go b/internal/storage/storage.go index 5bb7625..9bcad58 100644 --- a/internal/storage/storage.go +++ b/internal/storage/storage.go @@ -11,6 +11,7 @@ var ( // goCacheBackendInstance is a singleton instance of the go-cache backend for KVStorage. goCacheBackendInstance KVStorage glock sync.Mutex + metricsDecorator func(KVStorage) KVStorage ) // KVStorage represents a key-value storage backend. @@ -27,13 +28,31 @@ type KVStorage interface { List(wildcard string) ([][]byte, error) } -// NewGoCache returns a singleton instance of a KVStorage implementation using go-cache as the backend. +// SetMetricsDecorator registers a decorator applied to the singleton the +// first time NewGoCache is called. It is a no-op once the singleton has been +// materialized, so callers must invoke this during process startup before any +// other code reaches NewGoCache. Passing nil clears the decorator. +func SetMetricsDecorator(d func(KVStorage) KVStorage) { + glock.Lock() + defer glock.Unlock() + metricsDecorator = d +} + +// NewGoCache returns a singleton instance of a KVStorage implementation using +// go-cache as the backend. If a metrics decorator has been registered via +// SetMetricsDecorator before this is first called, the singleton is wrapped +// with that decorator. func NewGoCache() KVStorage { glock.Lock() defer glock.Unlock() if goCacheBackendInstance == nil { - goCacheBackendInstance = gocache.NewBackend(1 * time.Minute) + base := gocache.NewBackend(1 * time.Minute) + if metricsDecorator != nil { + goCacheBackendInstance = metricsDecorator(base) + } else { + goCacheBackendInstance = base + } } return goCacheBackendInstance diff --git a/internal/workers/workers.go b/internal/workers/workers.go index fe362fb..06a4fa5 100644 --- a/internal/workers/workers.go +++ b/internal/workers/workers.go @@ -5,6 +5,8 @@ import ( "time" "github.com/amirhnajafiz/bedrock-api/internal/components/sessions" + "github.com/amirhnajafiz/bedrock-api/internal/metrics/sessionmetrics" + "github.com/amirhnajafiz/bedrock-api/internal/metrics/workermetrics" "github.com/amirhnajafiz/bedrock-api/internal/scheduler" "github.com/amirhnajafiz/bedrock-api/internal/storage" "github.com/amirhnajafiz/bedrock-api/pkg/enums" @@ -13,11 +15,10 @@ import ( ) // WorkerCheckExpiredSessions continuously checks for expired sessions in the session store and removes them at regular intervals. -func WorkerCheckExpiredSessions(ctx context.Context, logr *zap.Logger, interval time.Duration) { - // get a reference to the session store instance +func WorkerCheckExpiredSessions(ctx context.Context, logr *zap.Logger, interval time.Duration, wm *workermetrics.WorkerMetrics, sm *sessionmetrics.SessionMetrics) { + const workerName = "expired_sessions" ss := sessions.NewSessionStore(storage.NewGoCache()) - // ticker is used to periodically check for expired sessions ticker := time.NewTicker(interval) defer ticker.Stop() @@ -28,47 +29,70 @@ func WorkerCheckExpiredSessions(ctx context.Context, logr *zap.Logger, interval case <-ctx.Done(): return case <-ticker.C: - timeSnapshot := time.Now() + tickStart := time.Now() + timeSnapshot := tickStart records, err := ss.ListSessions() if err != nil { logr.Error("failed to list sessions", zap.Error(err)) + wm.IncError(workerName) + wm.ObserveTick(workerName, time.Since(tickStart)) continue } - // loop through the sessions in the session store mark any that have expired for _, session := range records { - // only consider running sessions if session.Status != enums.SessionStatusRunning { continue } - - // check with the TTL if session.CreatedAt.Add(session.Spec.TTL).Before(timeSnapshot) { session.Status = enums.SessionStatusStopped - logr.Debug("session expired", zap.String("session_id", session.Id)) - - // update the session if err := ss.SaveSession(session); err != nil { logr.Error("failed to update session", zap.Error(err)) + wm.IncError(workerName) } } } + + // reconcile session and dockerd gauges from the freshly listed records. + // Iterating the canonical status list ensures unused statuses drop to 0. + byStatus := make(map[enums.SessionStatus]float64) + byDockerd := make(map[string]float64) + for _, s := range records { + if s.DeletedAt != nil { + continue + } + byStatus[s.Status]++ + if s.Status != enums.SessionStatusFinished && + s.Status != enums.SessionStatusFailed && + s.Status != enums.SessionStatusStopped { + byDockerd[s.DockerDId]++ + } + } + for _, st := range []enums.SessionStatus{ + enums.SessionStatusPending, + enums.SessionStatusRunning, + enums.SessionStatusStopped, + enums.SessionStatusFinished, + enums.SessionStatusFailed, + } { + sm.SetByStatus(st, byStatus[st]) + } + for dockerd, n := range byDockerd { + sm.SetDockerdSessions(dockerd, n) + } + + wm.ObserveTick(workerName, time.Since(tickStart)) } } } -// WorkerDockerDHealthCheck continuously checks the health status of Docker daemons by listening to an input channel -// for updates and using a ticker to periodically remove stale entries from the health map. -func WorkerDockerDHealthCheck(ctx context.Context, input chan string, logr *zap.Logger, interval time.Duration) { - // get a reference to the scheduler instance - scheduler := scheduler.NewRoundRobin() - - // healthMap keeps track of the last time a health update was received for each Docker daemon +// WorkerDockerDHealthCheck continuously checks the health status of Docker daemons. +func WorkerDockerDHealthCheck(ctx context.Context, input chan string, logr *zap.Logger, interval time.Duration, wm *workermetrics.WorkerMetrics, sm *sessionmetrics.SessionMetrics) { + const workerName = "dockerd_health" + sched := scheduler.NewRoundRobin() healthMap := make(map[string]time.Time) - // ticker is used to periodically check for stale entries in the healthMap ticker := time.NewTicker(interval) defer ticker.Stop() @@ -79,38 +103,35 @@ func WorkerDockerDHealthCheck(ctx context.Context, input chan string, logr *zap. case <-ctx.Done(): return case dockerd := <-input: - // update the healthMap with the current time for the received Docker daemon healthMap[dockerd] = time.Now() - scheduler.Append(dockerd) - + sched.Append(dockerd) logr.Debug("dockerd append", zap.String("dockerd_id", dockerd)) + sm.SetDockerdWorkers("registered", float64(len(healthMap))) + sm.SetDockerdWorkers("healthy", float64(len(healthMap))) case <-ticker.C: - timeSnapshot := time.Now() + tickStart := time.Now() + timeSnapshot := tickStart - // loop through the healthMap and remove any entries that haven't been updated within the interval for dockerd, lastUpdated := range healthMap { if timeSnapshot.Sub(lastUpdated) > interval { logr.Warn("removing stale Docker daemon from health map", zap.String("dockerd", dockerd)) - delete(healthMap, dockerd) - scheduler.Drop(dockerd) + sched.Drop(dockerd) } } + + sm.SetDockerdWorkers("healthy", float64(len(healthMap))) + wm.ObserveTick(workerName, time.Since(tickStart)) } } } -// WorkerRemoveFinishedSessions checks for finished sessions and marks them as deleted to avoid duplicate ZMQ cleanups. -// Note: This worker must see a finished session twice before it is marked as deleted, to ensure that the ZMQ server -// has processed the session status update and performed any necessary cleanups. -func WorkerRemoveFinishedSessions(ctx context.Context, logr *zap.Logger, interval time.Duration) { - // get a reference to the session store instance +// WorkerRemoveFinishedSessions checks for finished sessions and marks them as deleted. +func WorkerRemoveFinishedSessions(ctx context.Context, logr *zap.Logger, interval time.Duration, wm *workermetrics.WorkerMetrics) { + const workerName = "cleanup" ss := sessions.NewSessionStore(storage.NewGoCache()) - - // finishedSessions keeps track of the number of times a finished session has been seen finishedSessions := make(map[string]int) - // ticker is used to periodically check for finished sessions ticker := time.NewTicker(interval) defer ticker.Stop() @@ -121,34 +142,35 @@ func WorkerRemoveFinishedSessions(ctx context.Context, logr *zap.Logger, interva case <-ctx.Done(): return case <-ticker.C: + tickStart := time.Now() + records, err := ss.ListSessions() if err != nil { logr.Error("failed to list sessions", zap.Error(err)) + wm.IncError(workerName) + wm.ObserveTick(workerName, time.Since(tickStart)) continue } - // loop through the sessions in the session store and mark any that are finished as deleted for _, session := range records { if session.DeletedAt != nil { continue } - if session.Status == enums.SessionStatusFinished || session.Status == enums.SessionStatusFailed { finishedSessions[session.Id]++ - if finishedSessions[session.Id] >= 2 { tmp := time.Now() session.DeletedAt = &tmp - logr.Debug("marking session as deleted", zap.String("session_id", session.Id)) - - // update the session if err := ss.SaveSession(session); err != nil { logr.Error("failed to update session", zap.Error(err)) + wm.IncError(workerName) } } } } + + wm.ObserveTick(workerName, time.Since(tickStart)) } } } diff --git a/internal/workers/workers_test.go b/internal/workers/workers_test.go index 4ce5607..cc3dc02 100644 --- a/internal/workers/workers_test.go +++ b/internal/workers/workers_test.go @@ -7,11 +7,15 @@ import ( "github.com/amirhnajafiz/bedrock-api/internal/components/sessions" "github.com/amirhnajafiz/bedrock-api/internal/logger" + "github.com/amirhnajafiz/bedrock-api/internal/metrics/sessionmetrics" "github.com/amirhnajafiz/bedrock-api/internal/scheduler" "github.com/amirhnajafiz/bedrock-api/internal/storage" "github.com/amirhnajafiz/bedrock-api/internal/workers" "github.com/amirhnajafiz/bedrock-api/pkg/enums" "github.com/amirhnajafiz/bedrock-api/pkg/models" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/testutil" ) func TestWorkerCheckExpiredSessions(t *testing.T) { @@ -34,7 +38,7 @@ func TestWorkerCheckExpiredSessions(t *testing.T) { }) // start the WorkerCheckExpiredSessions in a separate goroutine - go workers.WorkerCheckExpiredSessions(ctx, logr.Named("session-worker"), 1*time.Second) + go workers.WorkerCheckExpiredSessions(ctx, logr.Named("session-worker"), 1*time.Second, nil, nil) // session must exist now if sess, err := ss.GetSession("1", "d1"); err != nil { @@ -73,7 +77,7 @@ func TestWorkerDockerDHealthCheck(t *testing.T) { sc := scheduler.NewRoundRobin() // start the WorkerDockerDHealthCheck in a separate goroutine - go workers.WorkerDockerDHealthCheck(ctx, input, logr.Named("dockerd-health"), 3*time.Second) + go workers.WorkerDockerDHealthCheck(ctx, input, logr.Named("dockerd-health"), 3*time.Second, nil, nil) // simulate sending health updates for Docker daemons input <- "dockerd1" @@ -129,7 +133,7 @@ func TestWorkerRemoveFinishedSessions(t *testing.T) { }) // start the WorkerRemoveFinishedSessions in a separate goroutine - go workers.WorkerRemoveFinishedSessions(ctx, logr.Named("remove-sessions"), 1*time.Second) + go workers.WorkerRemoveFinishedSessions(ctx, logr.Named("remove-sessions"), 1*time.Second, nil) time.Sleep(5 * time.Second) @@ -149,3 +153,41 @@ func TestWorkerRemoveFinishedSessions(t *testing.T) { // cancel the context to stop the worker cancel() } + +func TestWorkerCheckExpiredSessionsReconcilesSessionMetrics(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + logr := logger.New("info") + + reg := prometheus.NewRegistry() + sm := sessionmetrics.New(reg) + + ss := sessions.NewSessionStore(storage.NewGoCache()) + _ = ss.SaveSession(&models.Session{ + Id: "reconcile-a", + DockerDId: "d1", + CreatedAt: time.Now(), + Status: enums.SessionStatusRunning, + Spec: models.Spec{TTL: 10 * time.Minute}, + }) + _ = ss.SaveSession(&models.Session{ + Id: "reconcile-b", + DockerDId: "d1", + CreatedAt: time.Now(), + Status: enums.SessionStatusRunning, + Spec: models.Spec{TTL: 10 * time.Minute}, + }) + + go workers.WorkerCheckExpiredSessions(ctx, logr.Named("session-worker"), 500*time.Millisecond, nil, sm) + + // give the worker at least one full tick to reconcile + time.Sleep(1500 * time.Millisecond) + + if got := testutil.ToFloat64(sm.SessionsByStatus().WithLabelValues("running")); got < 2 { + t.Errorf("sessions_by_status{running} = %v, want >= 2", got) + } + if got := testutil.ToFloat64(sm.DockerdSessions().WithLabelValues("d1")); got < 2 { + t.Errorf("dockerd_sessions{d1} = %v, want >= 2", got) + } +} diff --git a/pkg/metrics/buckets.go b/pkg/metrics/buckets.go new file mode 100644 index 0000000..9cc9b2d --- /dev/null +++ b/pkg/metrics/buckets.go @@ -0,0 +1,9 @@ +package metrics + +// DefaultDurationBuckets are the Prometheus default histogram buckets for +// durations in seconds, suitable for HTTP request latencies and short +// internal operations. Defined as a shared preset so all components use +// the same buckets, which keeps recording-rule math consistent. +var DefaultDurationBuckets = []float64{ + 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, +} diff --git a/pkg/metrics/runtime.go b/pkg/metrics/runtime.go new file mode 100644 index 0000000..988e00a --- /dev/null +++ b/pkg/metrics/runtime.go @@ -0,0 +1,14 @@ +package metrics + +import ( + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/collectors" +) + +// RegisterRuntime adds the stock Go runtime and process collectors to reg. +// These yield goroutines, GC pauses, heap, file descriptors, CPU, and +// resident memory metrics without any custom code. +func RegisterRuntime(reg prometheus.Registerer) { + reg.MustRegister(collectors.NewGoCollector()) + reg.MustRegister(collectors.NewProcessCollector(collectors.ProcessCollectorOpts{})) +} diff --git a/pkg/metrics/server.go b/pkg/metrics/server.go new file mode 100644 index 0000000..6c32cf6 --- /dev/null +++ b/pkg/metrics/server.go @@ -0,0 +1,95 @@ +package metrics + +import ( + "context" + "errors" + "fmt" + "net" + "net/http" + "sync" + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promhttp" + + "go.uber.org/zap" +) + +// Server is a self-contained Prometheus metrics HTTP server. +// It owns its own prometheus.Registry; prometheus.DefaultRegisterer is never touched. +type Server struct { + listener net.Listener + registry *prometheus.Registry + logr *zap.Logger + once sync.Once + served bool +} + +// NewServer binds a TCP listener on addr (host:port) and constructs a Server. +// The listener is bound synchronously so Addr() is meaningful immediately. +// The server is not started until Serve is called. +func NewServer(addr string, logr *zap.Logger) (*Server, error) { + ln, err := net.Listen("tcp", addr) + if err != nil { + return nil, fmt.Errorf("metrics listen: %w", err) + } + return &Server{ + listener: ln, + registry: prometheus.NewRegistry(), + logr: logr, + }, nil +} + +// Registerer returns the registerer for this server. Pass to component +// metric struct constructors. The underlying registry is intentionally +// not exposed so callers cannot enumerate other components' metrics. +func (s *Server) Registerer() prometheus.Registerer { + return s.registry +} + +// Addr returns the actual listening address. +func (s *Server) Addr() string { + return s.listener.Addr().String() +} + +// Serve runs the HTTP server and returns when ctx is cancelled or the server +// fails. Shutdown is graceful with a 5s timeout. Calling Serve more than once +// returns ErrAlreadyServed; the first call owns the listener. +func (s *Server) Serve(ctx context.Context) error { + var ran bool + s.once.Do(func() { ran = true; s.served = true }) + if !ran { + return ErrAlreadyServed + } + + mux := http.NewServeMux() + mux.Handle("/metrics", promhttp.HandlerFor(s.registry, promhttp.HandlerOpts{ + Registry: s.registry, + })) + + httpSrv := &http.Server{Handler: mux} + + errCh := make(chan error, 1) + go func() { + s.logr.Info("metrics server started", zap.String("address", s.listener.Addr().String())) + errCh <- httpSrv.Serve(s.listener) + }() + + select { + case <-ctx.Done(): + shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := httpSrv.Shutdown(shutdownCtx); err != nil { + s.logr.Warn("metrics server shutdown error", zap.Error(err)) + } + return ctx.Err() + case err := <-errCh: + if errors.Is(err, http.ErrServerClosed) { + return nil + } + return err + } +} + +// ErrAlreadyServed is returned by Serve when called more than once on the same Server. +var ErrAlreadyServed = errors.New("metrics: Serve called more than once") diff --git a/pkg/metrics/server_test.go b/pkg/metrics/server_test.go new file mode 100644 index 0000000..0279a71 --- /dev/null +++ b/pkg/metrics/server_test.go @@ -0,0 +1,104 @@ +package metrics_test + +import ( + "context" + "errors" + "io" + "net/http" + "strings" + "testing" + + "github.com/amirhnajafiz/bedrock-api/pkg/metrics" + + "go.uber.org/zap" +) + +func TestServerExposesMetrics(t *testing.T) { + srv, err := metrics.NewServer("127.0.0.1:0", zap.NewNop()) + if err != nil { + t.Fatalf("NewServer: %v", err) + } + metrics.RegisterRuntime(srv.Registerer()) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + errCh := make(chan error, 1) + go func() { errCh <- srv.Serve(ctx) }() + + resp, err := http.Get("http://" + srv.Addr() + "/metrics") + if err != nil { + t.Fatalf("GET /metrics: %v", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d, want 200", resp.StatusCode) + } + + body, _ := io.ReadAll(resp.Body) + if !strings.Contains(string(body), "go_goroutines") { + t.Errorf("expected go_goroutines metric in output, got:\n%s", string(body)) + } + + cancel() + if err := <-errCh; err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, http.ErrServerClosed) { + t.Errorf("Serve returned unexpected error: %v", err) + } +} + +func TestRegistererIsIsolated(t *testing.T) { + a, err := metrics.NewServer("127.0.0.1:0", zap.NewNop()) + if err != nil { + t.Fatalf("NewServer a: %v", err) + } + b, err := metrics.NewServer("127.0.0.1:0", zap.NewNop()) + if err != nil { + t.Fatalf("NewServer b: %v", err) + } + + // Registering the same collector name into two different servers must not panic + // because each server owns its own registry. + metrics.RegisterRuntime(a.Registerer()) + metrics.RegisterRuntime(b.Registerer()) +} + +func TestServeTwiceReturnsError(t *testing.T) { + srv, err := metrics.NewServer("127.0.0.1:0", zap.NewNop()) + if err != nil { + t.Fatalf("NewServer: %v", err) + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + // First Serve runs until ctx cancel. + firstDone := make(chan error, 1) + go func() { firstDone <- srv.Serve(ctx) }() + + // Wait until the first Serve has consumed the once.Do and the HTTP server + // is actually accepting connections. A successful GET /metrics is our + // synchronization signal — it can only succeed once the goroutine inside + // the first Serve call is running httpSrv.Serve(s.listener). + url := "http://" + srv.Addr() + "/metrics" + var ready bool + for i := 0; i < 200; i++ { + resp, err := http.Get(url) + if err == nil { + resp.Body.Close() + ready = true + break + } + } + if !ready { + t.Fatalf("first Serve never started accepting connections") + } + + // A second Serve call must return ErrAlreadyServed without disturbing the first. + if secondErr := srv.Serve(ctx); !errors.Is(secondErr, metrics.ErrAlreadyServed) { + t.Errorf("second Serve returned %v, want ErrAlreadyServed", secondErr) + } + + cancel() + <-firstDone +}