Add Prometheus metrics for HTTP, votes, cache, and WebSocket

cmd/server/main.go

@@ -17,6 +17,7 @@ import (
 	"github.com/jackc/pgx/v5/pgxpool"
 	"github.com/pscheid92/chatpulse/internal/adapter/eventpublisher"
 	"github.com/pscheid92/chatpulse/internal/adapter/httpserver"
+	"github.com/pscheid92/chatpulse/internal/adapter/metrics"
 	"github.com/pscheid92/chatpulse/internal/adapter/postgres"
 	"github.com/pscheid92/chatpulse/internal/adapter/redis"
 	"github.com/pscheid92/chatpulse/internal/adapter/twitch"
@@ -125,8 +126,8 @@ type realtimeInfra struct {
 	presence  *websocket.PresenceChecker
 }
 
-func setupWebSocket(streamerRepo domain.StreamerRepository, redisAddr string, configCache domain.ConfigSource) (realtimeInfra, error) {
-	node, err := websocket.NewNode(streamerRepo)
+func setupWebSocket(streamerRepo domain.StreamerRepository, redisAddr string, configCache domain.ConfigSource, wsMetrics *metrics.WebSocketMetrics) (realtimeInfra, error) {
+	node, err := websocket.NewNode(streamerRepo, wsMetrics)
 	if err != nil {
 		return realtimeInfra{}, fmt.Errorf("create Centrifuge node: %w", err)
 	}
@@ -142,7 +143,7 @@ func setupWebSocket(streamerRepo domain.StreamerRepository, redisAddr string, co
 
 	return realtimeInfra{
 		node:      node,
-		publisher: websocket.NewPublisher(node, configCache),
+		publisher: websocket.NewPublisher(node, configCache, wsMetrics),
 		presence:  websocket.NewPresenceChecker(node),
 	}, nil
 }
@@ -163,8 +164,8 @@ func initEventSubManager(cfg *config.Config, eventSubRepo *postgres.EventSubRepo
 	return mgr, nil
 }
 
-func initWebhookHandler(cfg *config.Config, ovl *app.Overlay, presence twitch.ViewerPresence, publisher domain.EventPublisher, onVoteApplied func(string)) *twitch.WebhookHandler {
-	return twitch.NewWebhookHandler(cfg.WebhookSecret, ovl, presence, publisher, onVoteApplied)
+func initWebhookHandler(cfg *config.Config, ovl *app.Overlay, presence twitch.ViewerPresence, publisher domain.EventPublisher, onVoteApplied func(string), voteMetrics *metrics.VoteMetrics) *twitch.WebhookHandler {
+	return twitch.NewWebhookHandler(cfg.WebhookSecret, ovl, presence, publisher, onVoteApplied, voteMetrics)
 }
 
 func runGracefulShutdown(srv *httpserver.Server, node *centrifuge.Node, conduitMgr *twitch.EventSubManager) <-chan struct{} {
@@ -229,9 +230,17 @@ func main() {
 	defer rc.stopEviction()
 	defer rc.cancelPubsub()
 
+	// Metrics setup
+	reg := metrics.NewRegistry()
+	httpMetrics := metrics.NewHTTPMetrics(reg)
+	voteMetrics := metrics.NewVoteMetrics(reg)
+	cacheMetrics := metrics.NewCacheMetrics(reg)
+	wsMetrics := metrics.NewWebSocketMetrics(reg)
+	rc.configCache.SetCacheMetrics(cacheMetrics)
+
 	ovl := app.NewOverlay(rc.configCache, rc.sentiment, rc.debouncer)
 
-	ws, err := setupWebSocket(db.streamerRepo, rc.client.Options().Addr, rc.configCache)
+	ws, err := setupWebSocket(db.streamerRepo, rc.client.Options().Addr, rc.configCache, wsMetrics)
 	if err != nil {
 		slog.Error("WebSocket setup failed", "error", err)
 		os.Exit(1)
@@ -252,7 +261,7 @@ func main() {
 	}
 	slog.Info("EventSub configured", "callback_url", cfg.WebhookCallbackURL)
 
-	webhookHdlr := initWebhookHandler(cfg, ovl, ws.presence, publisher, ticker.Track)
+	webhookHdlr := initWebhookHandler(cfg, ovl, ws.presence, publisher, ticker.Track, voteMetrics)
 	appSvc := app.NewService(db.streamerRepo, db.configRepo, ovl, eventsubMgr, publisher)
 
 	healthChecks := slices.Concat(db.healthChecks, rc.healthChecks)
@@ -261,7 +270,7 @@ func main() {
 		CheckOrigin: websocket.NewCheckOrigin(cfg.WebhookCallbackURL, cfg.AppEnv == "development"),
 	})
 
-	srv, err := httpserver.NewServer(cfg, appSvc, ws.presence, wsHandler, http.HandlerFunc(webhookHdlr.HandleEventSub), eventsubMgr, healthChecks)
+	srv, err := httpserver.NewServer(cfg, appSvc, ws.presence, wsHandler, http.HandlerFunc(webhookHdlr.HandleEventSub), eventsubMgr, healthChecks, httpMetrics, metrics.Handler(reg))
 	if err != nil {
 		slog.Error("Failed to create server", "error", err)
 		os.Exit(1)

go.mod

@@ -10,6 +10,7 @@ require (
 	github.com/jackc/tern/v2 v2.3.5
 	github.com/joho/godotenv v1.5.1
 	github.com/labstack/echo/v4 v4.15.0
+	github.com/prometheus/client_golang v1.23.2
 	github.com/pscheid92/uuid v0.1.1
 	github.com/redis/go-redis/v9 v9.18.0
 	github.com/stretchr/testify v1.11.1
@@ -86,7 +87,6 @@ require (
 	github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // 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.5 // indirect
 	github.com/prometheus/procfs v0.19.2 // indirect

go.sum

@@ -117,6 +117,8 @@ github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
+github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
+github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
 github.com/labstack/echo/v4 v4.15.0 h1:hoRTKWcnR5STXZFe9BmYun9AMTNeSbjHi2vtDuADJ24=
 github.com/labstack/echo/v4 v4.15.0/go.mod h1:xmw1clThob0BSVRX1CRQkGQ/vjwcpOMjQZSZa9fKA/c=
 github.com/labstack/gommon v0.4.2 h1:F8qTUNXgG1+6WQmqoUWnz8WiEU60mXVVw0P4ht1WRA0=

internal/adapter/httpserver/routes.go

@@ -9,6 +9,9 @@ import (
 )
 
 func (s *Server) registerRoutes() {
+	if s.httpMetrics != nil {
+		s.echo.Use(s.httpMetrics.Middleware())
+	}
 	s.echo.Use(s.setupRequestLoggerMiddleware())
 	s.echo.Use(middleware.Recover())
 	s.echo.Use(ErrorHandlingMiddleware())
@@ -35,6 +38,10 @@ func (s *Server) registerRoutes() {
 
 	s.echo.GET("/", s.handleLanding)
 
+	if s.metricsHandler != nil {
+		s.echo.GET("/metrics", echo.WrapHandler(s.metricsHandler))
+	}
+
 	s.registerHealthRoutes()
 	s.registerAuthRoutes(csrfMiddleware, authRL)
 	s.registerDashboardRoutes(csrfMiddleware, dashboardRL)

internal/adapter/httpserver/server.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/gorilla/sessions"
 	"github.com/labstack/echo/v4"
+	"github.com/pscheid92/chatpulse/internal/adapter/metrics"
 	"github.com/pscheid92/chatpulse/internal/adapter/twitch"
 	"github.com/pscheid92/chatpulse/internal/app"
 	"github.com/pscheid92/chatpulse/internal/domain"
@@ -42,13 +43,15 @@ type Server struct {
 
 	templates *template.Template
 
-	oauthClient  twitchOAuthClient
-	sessionStore *sessions.CookieStore
-	healthChecks []HealthCheck
-	startTime    time.Time
+	oauthClient    twitchOAuthClient
+	sessionStore   *sessions.CookieStore
+	healthChecks   []HealthCheck
+	startTime      time.Time
+	httpMetrics    *metrics.HTTPMetrics
+	metricsHandler http.Handler
 }
 
-func NewServer(cfg *config.Config, app appService, presence twitch.ViewerPresence, websocketHandler http.Handler, webhookHandler http.Handler, twitchService domain.EventSubService, healthChecks []HealthCheck) (*Server, error) {
+func NewServer(cfg *config.Config, app appService, presence twitch.ViewerPresence, websocketHandler http.Handler, webhookHandler http.Handler, twitchService domain.EventSubService, healthChecks []HealthCheck, httpMetrics *metrics.HTTPMetrics, metricsHandler http.Handler) (*Server, error) {
 	templates, err := template.ParseFS(web.TemplateFiles, "templates/*.html")
 	if err != nil {
 		return nil, fmt.Errorf("failed to parse templates: %w", err)
@@ -73,6 +76,8 @@ func NewServer(cfg *config.Config, app appService, presence twitch.ViewerPresenc
 		templates:        templates,
 		healthChecks:     healthChecks,
 		startTime:        time.Now(),
+		httpMetrics:      httpMetrics,
+		metricsHandler:   metricsHandler,
 	}
 
 	srv.registerRoutes()

internal/adapter/metrics/cache.go

@@ -0,0 +1,37 @@
+package metrics
+
+import "github.com/prometheus/client_golang/prometheus"
+
+// CacheMetrics holds Prometheus metrics for config cache performance.
+type CacheMetrics struct {
+	Hits          *prometheus.CounterVec
+	Misses        *prometheus.CounterVec
+	Invalidations prometheus.Counter
+}
+
+// NewCacheMetrics creates and registers cache metrics on the given registry.
+func NewCacheMetrics(reg prometheus.Registerer) *CacheMetrics {
+	m := &CacheMetrics{
+		Hits: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: "config_cache",
+			Name:      "hits_total",
+			Help:      "Total number of config cache hits, by layer.",
+		}, []string{"layer"}),
+		Misses: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: "config_cache",
+			Name:      "misses_total",
+			Help:      "Total number of config cache misses, by layer.",
+		}, []string{"layer"}),
+		Invalidations: prometheus.NewCounter(prometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: "config_cache",
+			Name:      "invalidations_total",
+			Help:      "Total number of config cache invalidations.",
+		}),
+	}
+
+	reg.MustRegister(m.Hits, m.Misses, m.Invalidations)
+	return m
+}

internal/adapter/metrics/http.go

@@ -0,0 +1,70 @@
+package metrics
+
+import (
+	"strconv"
+	"strings"
+
+	"github.com/labstack/echo/v4"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+// HTTPMetrics holds Prometheus metrics for HTTP request tracking.
+type HTTPMetrics struct {
+	RequestDuration *prometheus.HistogramVec
+	RequestsTotal   *prometheus.CounterVec
+	InFlightGauge   prometheus.Gauge
+}
+
+// NewHTTPMetrics creates and registers HTTP metrics on the given registry.
+func NewHTTPMetrics(reg prometheus.Registerer) *HTTPMetrics {
+	m := &HTTPMetrics{
+		RequestDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{
+			Namespace: namespace,
+			Subsystem: "http",
+			Name:      "request_duration_seconds",
+			Help:      "Duration of HTTP requests in seconds.",
+			Buckets:   prometheus.DefBuckets,
+		}, []string{"method", "route", "status_code"}),
+		RequestsTotal: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: "http",
+			Name:      "requests_total",
+			Help:      "Total number of HTTP requests.",
+		}, []string{"method", "route", "status_code"}),
+		InFlightGauge: prometheus.NewGauge(prometheus.GaugeOpts{
+			Namespace: namespace,
+			Subsystem: "http",
+			Name:      "in_flight_requests",
+			Help:      "Number of HTTP requests currently being processed.",
+		}),
+	}
+
+	reg.MustRegister(m.RequestDuration, m.RequestsTotal, m.InFlightGauge)
+	return m
+}
+
+// Middleware returns an Echo middleware that records HTTP metrics.
+// It skips /metrics and /health/* endpoints.
+func (m *HTTPMetrics) Middleware() echo.MiddlewareFunc {
+	return func(next echo.HandlerFunc) echo.HandlerFunc {
+		return func(c echo.Context) error {
+			path := c.Path()
+			if path == "/metrics" || strings.HasPrefix(path, "/health/") {
+				return next(c)
+			}
+
+			m.InFlightGauge.Inc()
+			defer m.InFlightGauge.Dec()
+
+			timer := prometheus.NewTimer(prometheus.ObserverFunc(func(v float64) {
+				status := strconv.Itoa(c.Response().Status)
+				m.RequestDuration.WithLabelValues(c.Request().Method, path, status).Observe(v)
+				m.RequestsTotal.WithLabelValues(c.Request().Method, path, status).Inc()
+			}))
+
+			err := next(c)
+			timer.ObserveDuration()
+			return err
+		}
+	}
+}

internal/adapter/metrics/metrics.go

@@ -0,0 +1,24 @@
+package metrics
+
+import (
+	"net/http"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/collectors"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+)
+
+const namespace = "chatpulse"
+
+// NewRegistry creates a Prometheus registry with Go runtime and process collectors.
+func NewRegistry() *prometheus.Registry {
+	reg := prometheus.NewRegistry()
+	reg.MustRegister(collectors.NewGoCollector())
+	reg.MustRegister(collectors.NewProcessCollector(collectors.ProcessCollectorOpts{}))
+	return reg
+}
+
+// Handler returns an http.Handler that serves Prometheus metrics.
+func Handler(reg *prometheus.Registry) http.Handler {
+	return promhttp.HandlerFor(reg, promhttp.HandlerOpts{})
+}

internal/adapter/metrics/vote.go

@@ -0,0 +1,35 @@
+package metrics
+
+import "github.com/prometheus/client_golang/prometheus"
+
+// VoteMetrics holds Prometheus metrics for the vote processing pipeline.
+type VoteMetrics struct {
+	VotesProcessed     *prometheus.CounterVec
+	ProcessingDuration prometheus.Histogram
+	VotesByTarget      *prometheus.CounterVec
+}
+
+// NewVoteMetrics creates and registers vote pipeline metrics on the given registry.
+func NewVoteMetrics(reg prometheus.Registerer) *VoteMetrics {
+	m := &VoteMetrics{
+		VotesProcessed: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Namespace: namespace,
+			Name:      "votes_processed_total",
+			Help:      "Total number of votes processed, by result.",
+		}, []string{"result"}),
+		ProcessingDuration: prometheus.NewHistogram(prometheus.HistogramOpts{
+			Namespace: namespace,
+			Name:      "votes_processing_duration_seconds",
+			Help:      "Duration of vote processing in seconds.",
+			Buckets:   []float64{0.0005, 0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0},
+		}),
+		VotesByTarget: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Namespace: namespace,
+			Name:      "votes_by_target_total",
+			Help:      "Total number of applied votes, by target direction.",
+		}, []string{"target"}),
+	}
+
+	reg.MustRegister(m.VotesProcessed, m.ProcessingDuration, m.VotesByTarget)
+	return m
+}

internal/adapter/metrics/websocket.go

@@ -0,0 +1,30 @@
+package metrics
+
+import "github.com/prometheus/client_golang/prometheus"
+
+// WebSocketMetrics holds Prometheus metrics for WebSocket connections.
+type WebSocketMetrics struct {
+	ActiveConnections prometheus.Gauge
+	MessagesPublished prometheus.Counter
+}
+
+// NewWebSocketMetrics creates and registers WebSocket metrics on the given registry.
+func NewWebSocketMetrics(reg prometheus.Registerer) *WebSocketMetrics {
+	m := &WebSocketMetrics{
+		ActiveConnections: prometheus.NewGauge(prometheus.GaugeOpts{
+			Namespace: namespace,
+			Subsystem: "websocket",
+			Name:      "active_connections",
+			Help:      "Number of active WebSocket connections.",
+		}),
+		MessagesPublished: prometheus.NewCounter(prometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: "websocket",
+			Name:      "messages_published_total",
+			Help:      "Total number of WebSocket messages published.",
+		}),
+	}
+
+	reg.MustRegister(m.ActiveConnections, m.MessagesPublished)
+	return m
+}