diff --git a/cmd/server/main.go b/cmd/server/main.go index 837ce6b..4c1731d 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -12,7 +12,6 @@ import ( "syscall" "time" - "github.com/gorilla/mux" grpcapi "github.com/igodwin/notifier/api/grpc" pb "github.com/igodwin/notifier/api/grpc/pb" "github.com/igodwin/notifier/api/rest" @@ -20,10 +19,14 @@ import ( "github.com/igodwin/notifier/internal/config" "github.com/igodwin/notifier/internal/domain" "github.com/igodwin/notifier/internal/logging" + "github.com/igodwin/notifier/internal/metrics" "github.com/igodwin/notifier/internal/notifier" "github.com/igodwin/notifier/internal/queue" "github.com/igodwin/notifier/internal/service" "google.golang.org/grpc" + "google.golang.org/grpc/credentials" + "google.golang.org/grpc/health" + healthgrpc "google.golang.org/grpc/health/grpc_health_v1" "google.golang.org/grpc/reflection" ) @@ -54,10 +57,10 @@ func main() { } // Create logger from config - logger, err := logging.NewFromConfig(cfg.Logging.Level, cfg.Logging.OutputPath) + logger, err := logging.NewFromOptions(cfg.Logging.Level, cfg.Logging.Format, cfg.Logging.OutputPath) if err != nil { // Fallback to stdout if log file can't be opened - logger, _ = logging.NewFromConfig(cfg.Logging.Level, "stdout") + logger, _ = logging.NewFromOptions(cfg.Logging.Level, cfg.Logging.Format, "stdout") logger.Warnf("Failed to open log file, using stdout: %v", err) } @@ -91,13 +94,13 @@ func main() { var authStore *auth.APIKeyStore var hybridKeyStore *auth.HybridKeyStore var authz *auth.NotifierAuthz + var dbStore *auth.KeyStoreDB if cfg.Auth.Enabled { authStore = auth.NewAPIKeyStore() authz = auth.NewNotifierAuthz() logger.Info("API authentication enabled") // Create database backend if configured - var dbStore *auth.KeyStoreDB if cfg.Auth.Database.URL != "" { dbStore, err = auth.NewKeyStoreDB(cfg.Auth.Database.URL, logger) if err != nil { @@ -197,6 +200,7 @@ func main() { // Create notification service (pass config as account resolver and authz for RBAC) svc := service.NewNotificationService(factory, q, cfg.Queue.WorkerCount, cfg, authz, logger) + svc.WithRetryBackoff(cfg.Queue.RetryBackoff) // Configure notification retention if enabled if err := svc.WithRetentionConfig(cfg.Retention); err != nil { @@ -214,7 +218,24 @@ func main() { } logger.Infof("Started %d worker(s)", cfg.Queue.WorkerCount) - // Wait group for both servers + // Readiness checks shared by the REST /readyz route and the dedicated + // health listener: the queue must be open and, when configured, the + // auth database reachable. + readiness := map[string]rest.ReadinessCheck{ + "queue": q.HealthCheck, + } + if dbStore != nil { + readiness["database"] = dbStore.Ping + } + + // Metrics collector + endpoint + var collector *metrics.Collector + if cfg.Metrics.Enabled { + collector = metrics.NewCollector(svc, q, logger) + go collector.Run(ctx, 10*time.Second) + } + + // Wait group for all servers var wg sync.WaitGroup // Start gRPC server if enabled @@ -228,7 +249,22 @@ func main() { var restServer *http.Server if cfg.Server.Mode == "both" || cfg.Server.Mode == "rest" { wg.Add(1) - restServer = startRESTServer(ctx, &wg, cfg, svc, logger, authStore, hybridKeyStore) + restServer = startRESTServer(ctx, &wg, cfg, svc, logger, authStore, hybridKeyStore, readiness, collector) + } + + // Start metrics server if enabled + var metricsServer *http.Server + if collector != nil { + wg.Add(1) + metricsServer = startMetricsServer(&wg, cfg, collector, logger) + } + + // Start dedicated health listener if enabled (needed for probes when + // running in grpc-only mode; harmless duplication otherwise) + var healthServer *http.Server + if cfg.HealthCheck.Enabled { + wg.Add(1) + healthServer = startHealthServer(&wg, cfg, readiness, logger) } // Wait for interrupt signal @@ -242,10 +278,12 @@ func main() { shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 30*time.Second) defer shutdownCancel() - // Stop REST server - if restServer != nil { - if err := restServer.Shutdown(shutdownCtx); err != nil { - logger.Errorf("Error during REST server shutdown: %v", err) + // Stop HTTP servers + for _, server := range []*http.Server{restServer, metricsServer, healthServer} { + if server != nil { + if err := server.Shutdown(shutdownCtx); err != nil { + logger.Errorf("Error during HTTP server shutdown (%s): %v", server.Addr, err) + } } } @@ -337,6 +375,18 @@ func startGRPCServer(ctx context.Context, wg *sync.WaitGroup, cfg *config.Config // Create gRPC server options var serverOpts []grpc.ServerOption + // TLS credentials if configured + if cfg.Server.TLS.Enabled { + creds, err := credentials.NewServerTLSFromFile(cfg.Server.TLS.CertFile, cfg.Server.TLS.KeyFile) + if err != nil { + logger.Fatalf("Failed to load gRPC TLS credentials: %v", err) + } + serverOpts = append(serverOpts, grpc.Creds(creds)) + } + + // Bound message sizes to match the REST body limit + serverOpts = append(serverOpts, grpc.MaxRecvMsgSize(1<<20)) + // Add authentication interceptors if enabled if authStore != nil { authMiddleware := auth.NewGRPCAuthMiddleware(authStore, logger) @@ -352,6 +402,13 @@ func startGRPCServer(ctx context.Context, wg *sync.WaitGroup, cfg *config.Config grpcHandler := grpcapi.NewNotifierHandler(svc, logger) pb.RegisterNotifierServiceServer(grpcServer, grpcHandler) + // Standard gRPC health service (grpc.health.v1) for Kubernetes gRPC + // probes and load balancers. + healthSvc := health.NewServer() + healthSvc.SetServingStatus("", healthgrpc.HealthCheckResponse_SERVING) + healthSvc.SetServingStatus(pb.NotifierService_ServiceDesc.ServiceName, healthgrpc.HealthCheckResponse_SERVING) + healthgrpc.RegisterHealthServer(grpcServer, healthSvc) + // Enable reflection for tools like grpcurl reflection.Register(grpcServer) @@ -368,16 +425,33 @@ func startGRPCServer(ctx context.Context, wg *sync.WaitGroup, cfg *config.Config return grpcServer } -func startRESTServer(ctx context.Context, wg *sync.WaitGroup, cfg *config.Config, svc domain.NotificationService, logger *logging.Logger, authStore *auth.APIKeyStore, hybridKeyStore *auth.HybridKeyStore) *http.Server { - var router *mux.Router - if authStore != nil && hybridKeyStore != nil { - router = rest.NewRouterWithAuthAndKeyStore(svc, logger, authStore, hybridKeyStore) - } else if authStore != nil { - router = rest.NewRouterWithAuth(svc, logger, authStore) - } else { - router = rest.NewRouter(svc, logger) +func startRESTServer(ctx context.Context, wg *sync.WaitGroup, cfg *config.Config, svc domain.NotificationService, logger *logging.Logger, authStore *auth.APIKeyStore, hybridKeyStore *auth.HybridKeyStore, readiness map[string]rest.ReadinessCheck, collector *metrics.Collector) *http.Server { + opts := rest.RouterOptions{ + Service: svc, + Logger: logger, + AuthStore: authStore, + KeyStore: hybridKeyStore, + Readiness: readiness, } + // Wire CORS from config (validated at load time; empty origins = disabled) + if len(cfg.CORS.AllowedOrigins) > 0 { + opts.CORS = &rest.CORSConfig{ + AllowedOrigins: cfg.CORS.AllowedOrigins, + AllowedMethods: cfg.CORS.AllowedMethods, + AllowedHeaders: cfg.CORS.AllowedHeaders, + AllowCredentials: cfg.CORS.AllowCredentials, + MaxAge: cfg.CORS.MaxAge, + } + logger.Infof("CORS enabled for %d origin(s)", len(cfg.CORS.AllowedOrigins)) + } + + if collector != nil { + opts.Instrument = collector.InstrumentHTTP + } + + router := rest.NewRouterWithOptions(opts) + addr := fmt.Sprintf("%s:%d", cfg.Server.Host, cfg.Server.RESTPort) server := &http.Server{ Addr: addr, @@ -389,8 +463,15 @@ func startRESTServer(ctx context.Context, wg *sync.WaitGroup, cfg *config.Config go func() { defer wg.Done() - logger.Infof("REST server listening on %s", addr) - if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed { + var err error + if cfg.Server.TLS.Enabled { + logger.Infof("REST server listening on %s (TLS)", addr) + err = server.ListenAndServeTLS(cfg.Server.TLS.CertFile, cfg.Server.TLS.KeyFile) + } else { + logger.Infof("REST server listening on %s", addr) + err = server.ListenAndServe() + } + if err != nil && err != http.ErrServerClosed { logger.Fatalf("Failed to start REST server: %v", err) } }() @@ -398,6 +479,60 @@ func startRESTServer(ctx context.Context, wg *sync.WaitGroup, cfg *config.Config return server } +// startMetricsServer serves Prometheus metrics on the configured port. +func startMetricsServer(wg *sync.WaitGroup, cfg *config.Config, collector *metrics.Collector, logger *logging.Logger) *http.Server { + path := cfg.Metrics.Path + if path == "" { + path = "/metrics" + } + mux := http.NewServeMux() + mux.Handle(path, collector.Handler()) + + addr := fmt.Sprintf("%s:%d", cfg.Server.Host, cfg.Metrics.Port) + server := &http.Server{ + Addr: addr, + Handler: mux, + ReadTimeout: 10 * time.Second, + WriteTimeout: 10 * time.Second, + } + + go func() { + defer wg.Done() + logger.Infof("Metrics server listening on %s%s", addr, path) + if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed { + logger.Errorf("Metrics server stopped: %v", err) + } + }() + + return server +} + +// startHealthServer serves liveness (/health) and readiness (/readyz) on a +// dedicated port so probes work even in grpc-only mode. +func startHealthServer(wg *sync.WaitGroup, cfg *config.Config, readiness map[string]rest.ReadinessCheck, logger *logging.Logger) *http.Server { + mux := http.NewServeMux() + mux.Handle("/health", rest.LivenessHandler()) + mux.Handle("/readyz", rest.ReadinessHandler(readiness)) + + addr := fmt.Sprintf("%s:%d", cfg.Server.Host, cfg.HealthCheck.Port) + server := &http.Server{ + Addr: addr, + Handler: mux, + ReadTimeout: 10 * time.Second, + WriteTimeout: 10 * time.Second, + } + + go func() { + defer wg.Done() + logger.Infof("Health server listening on %s", addr) + if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed { + logger.Errorf("Health server stopped: %v", err) + } + }() + + return server +} + func registerAuthorizationRules(cfg *config.Config, authz *auth.NotifierAuthz, logger *logging.Logger) { // Register SMTP authorization rules for accountName, smtpConfig := range cfg.Notifiers.SMTP { diff --git a/internal/logging/logger.go b/internal/logging/logger.go index e013262..a07aee4 100644 --- a/internal/logging/logger.go +++ b/internal/logging/logger.go @@ -3,15 +3,17 @@ package logging import ( "fmt" "io" - "log" + "log/slog" "os" "time" ) -// Logger provides structured logging with ISO 8601 timestamps +// Logger provides structured logging backed by log/slog, with UTC RFC3339 +// timestamps and a level-gated API compatible with the previous +// *log.Logger-based implementation. type Logger struct { - *log.Logger - level LogLevel + slogger *slog.Logger + level LogLevel } // LogLevel represents the logging level @@ -24,20 +26,73 @@ const ( ErrorLevel ) -// New creates a new logger with ISO 8601 timestamp format +// toSlogLevel maps our LogLevel to the equivalent slog.Level. +func (l LogLevel) toSlogLevel() slog.Level { + switch l { + case DebugLevel: + return slog.LevelDebug + case InfoLevel: + return slog.LevelInfo + case WarnLevel: + return slog.LevelWarn + case ErrorLevel: + return slog.LevelError + default: + return slog.LevelInfo + } +} + +// replaceAttr normalizes the slog time attribute to a UTC RFC3339 timestamp +// so log output has a stable, predictable format regardless of handler. +func replaceAttr(_ []string, a slog.Attr) slog.Attr { + if a.Key == slog.TimeKey { + if t, ok := a.Value.Any().(time.Time); ok { + a.Value = slog.StringValue(t.UTC().Format(time.RFC3339)) + } + } + return a +} + +// newHandler builds a slog.Handler for the given format ("json" or "text", +// with "json" as the default), level, and output writer. +func newHandler(format string, level LogLevel, output io.Writer) slog.Handler { + opts := &slog.HandlerOptions{ + Level: level.toSlogLevel(), + ReplaceAttr: replaceAttr, + } + + switch format { + case "text": + return slog.NewTextHandler(output, opts) + default: + return slog.NewJSONHandler(output, opts) + } +} + +// New creates a new logger using the text format with the given level and +// output writer. func New(level LogLevel, output io.Writer) *Logger { if output == nil { output = os.Stdout } return &Logger{ - Logger: log.New(output, "", 0), // No flags, we'll format ourselves - level: level, + slogger: slog.New(newHandler("text", level, output)), + level: level, } } -// NewFromConfig creates a logger from configuration +// NewFromConfig creates a logger from configuration using the json format, +// matching the documented default for logging.format. func NewFromConfig(levelStr string, outputPath string) (*Logger, error) { + return NewFromOptions(levelStr, "json", outputPath) +} + +// NewFromOptions creates a logger from configuration with an explicit +// format ("json", "text", or "" which defaults to "json"). outputPath may be +// "stdout", "stderr", "" (defaults to stdout), or a file path, which is +// opened for append (creating it if necessary). +func NewFromOptions(levelStr string, format string, outputPath string) (*Logger, error) { level := parseLevel(levelStr) var output io.Writer @@ -54,85 +109,66 @@ func NewFromConfig(levelStr string, outputPath string) (*Logger, error) { output = file } - return New(level, output), nil + return &Logger{ + slogger: slog.New(newHandler(format, level, output)), + level: level, + }, nil } -// formatMessage formats a log message with ISO 8601 timestamp -func (l *Logger) formatMessage(level string, msg string) string { - timestamp := time.Now().UTC().Format(time.RFC3339) - return timestamp + " [" + level + "] " + msg +// Slog exposes the underlying *slog.Logger for structured call sites. +func (l *Logger) Slog() *slog.Logger { + return l.slogger } // Debug logs a debug message func (l *Logger) Debug(msg string) { - if l.level <= DebugLevel { - l.Logger.Println(l.formatMessage("DEBUG", msg)) - } + l.slogger.Debug(msg) } // Debugf logs a formatted debug message func (l *Logger) Debugf(format string, args ...interface{}) { - if l.level <= DebugLevel { - msg := sprintf(format, args...) - l.Logger.Println(l.formatMessage("DEBUG", msg)) - } + l.slogger.Debug(fmt.Sprintf(format, args...)) } // Info logs an info message func (l *Logger) Info(msg string) { - if l.level <= InfoLevel { - l.Logger.Println(l.formatMessage("INFO", msg)) - } + l.slogger.Info(msg) } // Infof logs a formatted info message func (l *Logger) Infof(format string, args ...interface{}) { - if l.level <= InfoLevel { - msg := sprintf(format, args...) - l.Logger.Println(l.formatMessage("INFO", msg)) - } + l.slogger.Info(fmt.Sprintf(format, args...)) } // Warn logs a warning message func (l *Logger) Warn(msg string) { - if l.level <= WarnLevel { - l.Logger.Println(l.formatMessage("WARN", msg)) - } + l.slogger.Warn(msg) } // Warnf logs a formatted warning message func (l *Logger) Warnf(format string, args ...interface{}) { - if l.level <= WarnLevel { - msg := sprintf(format, args...) - l.Logger.Println(l.formatMessage("WARN", msg)) - } + l.slogger.Warn(fmt.Sprintf(format, args...)) } // Error logs an error message func (l *Logger) Error(msg string) { - if l.level <= ErrorLevel { - l.Logger.Println(l.formatMessage("ERROR", msg)) - } + l.slogger.Error(msg) } // Errorf logs a formatted error message func (l *Logger) Errorf(format string, args ...interface{}) { - if l.level <= ErrorLevel { - msg := sprintf(format, args...) - l.Logger.Println(l.formatMessage("ERROR", msg)) - } + l.slogger.Error(fmt.Sprintf(format, args...)) } -// Fatal logs a fatal message and exits +// Fatal logs a fatal message at error level and exits func (l *Logger) Fatal(msg string) { - l.Logger.Println(l.formatMessage("FATAL", msg)) + l.slogger.Error(msg) os.Exit(1) } -// Fatalf logs a formatted fatal message and exits +// Fatalf logs a formatted fatal message at error level and exits func (l *Logger) Fatalf(format string, args ...interface{}) { - msg := sprintf(format, args...) - l.Logger.Println(l.formatMessage("FATAL", msg)) + l.slogger.Error(fmt.Sprintf(format, args...)) os.Exit(1) } @@ -151,8 +187,3 @@ func parseLevel(levelStr string) LogLevel { return InfoLevel } } - -// sprintf is a helper using fmt -func sprintf(format string, args ...interface{}) string { - return fmt.Sprintf(format, args...) -} diff --git a/internal/logging/logger_test.go b/internal/logging/logger_test.go new file mode 100644 index 0000000..52b0193 --- /dev/null +++ b/internal/logging/logger_test.go @@ -0,0 +1,221 @@ +package logging + +import ( + "encoding/json" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +func TestNew_TextFormat(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "text.log") + file, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644) + if err != nil { + t.Fatalf("unexpected error opening file: %v", err) + } + defer file.Close() + + logger := New(InfoLevel, file) + logger.Info("hello world") + + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("failed to read log file: %v", err) + } + out := string(data) + + if !strings.Contains(out, "hello world") { + t.Fatalf("expected output to contain message, got: %q", out) + } + if !strings.Contains(out, "level=INFO") { + t.Fatalf("expected text output to contain level=INFO, got: %q", out) + } + if json.Valid(data) { + t.Fatalf("expected non-JSON text output, got JSON: %q", out) + } +} + +func TestNewFromConfig_JSONOutput(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "json.log") + + logger, err := NewFromConfig("info", path) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + logger.Info("structured message") + + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("failed to read log file: %v", err) + } + + var payload map[string]interface{} + if err := json.Unmarshal(data, &payload); err != nil { + t.Fatalf("expected valid JSON output, got error %v; output: %q", err, string(data)) + } + + if payload["msg"] != "structured message" { + t.Fatalf("expected msg field to equal message, got: %v", payload["msg"]) + } + if payload["level"] != "INFO" { + t.Fatalf("expected level field INFO, got: %v", payload["level"]) + } + timeVal, ok := payload["time"].(string) + if !ok { + t.Fatalf("expected time field to be a string, got: %v", payload["time"]) + } + if _, err := time.Parse(time.RFC3339, timeVal); err != nil { + t.Fatalf("expected time field to be RFC3339, got %q: %v", timeVal, err) + } +} + +func TestLevelFiltering_DebugSuppressedAtInfo(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "level.log") + file, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644) + if err != nil { + t.Fatalf("unexpected error opening file: %v", err) + } + defer file.Close() + + logger := New(InfoLevel, file) + logger.Debug("should not appear") + logger.Info("should appear") + + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("failed to read log file: %v", err) + } + out := string(data) + + if strings.Contains(out, "should not appear") { + t.Fatalf("expected debug message to be suppressed, got: %q", out) + } + if !strings.Contains(out, "should appear") { + t.Fatalf("expected info message to be present, got: %q", out) + } +} + +func TestNewFromOptions_FileOutput(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "out.log") + + logger, err := NewFromOptions("info", "json", path) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + logger.Info("file message") + + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("failed to read log file: %v", err) + } + + var payload map[string]interface{} + if err := json.Unmarshal(data, &payload); err != nil { + t.Fatalf("expected valid JSON in file, got error %v; contents: %q", err, string(data)) + } + if payload["msg"] != "file message" { + t.Fatalf("expected msg field, got: %v", payload["msg"]) + } +} + +func TestNewFromOptions_FormatSelection(t *testing.T) { + dir := t.TempDir() + + jsonPath := filepath.Join(dir, "json.log") + jsonLogger, err := NewFromOptions("info", "json", jsonPath) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + jsonLogger.Info("json message") + + jsonData, err := os.ReadFile(jsonPath) + if err != nil { + t.Fatalf("failed to read json log file: %v", err) + } + if !json.Valid(jsonData) { + t.Fatalf("expected valid JSON, got: %q", string(jsonData)) + } + + textPath := filepath.Join(dir, "text.log") + textLogger, err := NewFromOptions("info", "text", textPath) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + textLogger.Info("text message") + + textData, err := os.ReadFile(textPath) + if err != nil { + t.Fatalf("failed to read text log file: %v", err) + } + if json.Valid(textData) { + t.Fatalf("expected non-JSON text output, got: %q", string(textData)) + } + if !strings.Contains(string(textData), `msg="text message"`) { + t.Fatalf("expected text output to contain msg attribute, got: %q", string(textData)) + } + + // Default format ("" -> json) via NewFromOptions. + defaultPath := filepath.Join(dir, "default.log") + defaultLogger, err := NewFromOptions("info", "", defaultPath) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + defaultLogger.Info("default message") + + defaultData, err := os.ReadFile(defaultPath) + if err != nil { + t.Fatalf("failed to read default log file: %v", err) + } + if !json.Valid(defaultData) { + t.Fatalf("expected empty format to default to JSON, got: %q", string(defaultData)) + } + + // NewFromConfig should also default to JSON. + configPath := filepath.Join(dir, "config.log") + configLogger, err := NewFromConfig("info", configPath) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + configLogger.Info("config message") + + configData, err := os.ReadFile(configPath) + if err != nil { + t.Fatalf("failed to read config log file: %v", err) + } + if !json.Valid(configData) { + t.Fatalf("expected NewFromConfig to default to JSON format, got: %q", string(configData)) + } +} + +func TestSlogAccessor(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "slog.log") + file, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644) + if err != nil { + t.Fatalf("unexpected error opening file: %v", err) + } + defer file.Close() + + logger := New(InfoLevel, file) + if logger.Slog() == nil { + t.Fatal("expected Slog() to return a non-nil *slog.Logger") + } + + logger.Slog().Info("via slog accessor") + + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("failed to read log file: %v", err) + } + if !strings.Contains(string(data), "via slog accessor") { + t.Fatalf("expected message logged via Slog() accessor, got: %q", string(data)) + } +} diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go new file mode 100644 index 0000000..a1d18d4 --- /dev/null +++ b/internal/metrics/metrics.go @@ -0,0 +1,159 @@ +// Package metrics exposes Prometheus metrics for the notifier service. +// +// Notification totals are sampled from the service's own stats rather than +// instrumented inline, so the service layer stays metrics-agnostic; the +// sampling interval bounds staleness at a few seconds, which is fine for +// counters scraped every 15-60s. +package metrics + +import ( + "context" + "net/http" + "strconv" + "time" + + "github.com/igodwin/notifier/internal/domain" + "github.com/igodwin/notifier/internal/logging" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promhttp" +) + +// Collector owns the notifier metrics and the sampling loop. +type Collector struct { + registry *prometheus.Registry + + notificationsByStatus *prometheus.GaugeVec + notificationsByType *prometheus.GaugeVec + queueDepth prometheus.Gauge + + httpRequests *prometheus.CounterVec + httpDuration *prometheus.HistogramVec + + service domain.NotificationService + queue domain.Queue + logger *logging.Logger +} + +// NewCollector creates and registers the notifier metrics. +func NewCollector(service domain.NotificationService, queue domain.Queue, logger *logging.Logger) *Collector { + registry := prometheus.NewRegistry() + + c := &Collector{ + registry: registry, + service: service, + queue: queue, + logger: logger, + notificationsByStatus: prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Name: "notifier_notifications", + Help: "Number of tracked notifications by status.", + }, []string{"status"}), + notificationsByType: prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Name: "notifier_notifications_by_type", + Help: "Number of tracked notifications by notification type.", + }, []string{"type"}), + queueDepth: prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "notifier_queue_depth", + Help: "Number of messages currently waiting in the queue.", + }), + httpRequests: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "notifier_http_requests_total", + Help: "REST API requests by method, path pattern, and status code.", + }, []string{"method", "path", "code"}), + httpDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "notifier_http_request_duration_seconds", + Help: "REST API request duration.", + Buckets: prometheus.DefBuckets, + }, []string{"method", "path"}), + } + + registry.MustRegister( + c.notificationsByStatus, + c.notificationsByType, + c.queueDepth, + c.httpRequests, + c.httpDuration, + prometheus.NewGoCollector(), + prometheus.NewProcessCollector(prometheus.ProcessCollectorOpts{}), + ) + + return c +} + +// Handler returns the /metrics HTTP handler. +func (c *Collector) Handler() http.Handler { + return promhttp.HandlerFor(c.registry, promhttp.HandlerOpts{}) +} + +// Run samples service stats until ctx is cancelled. +func (c *Collector) Run(ctx context.Context, interval time.Duration) { + if interval <= 0 { + interval = 10 * time.Second + } + ticker := time.NewTicker(interval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + c.sample(ctx) + } + } +} + +func (c *Collector) sample(ctx context.Context) { + sampleCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + + if stats, err := c.service.GetStats(sampleCtx); err == nil { + c.notificationsByStatus.Reset() + for status, count := range stats.ByStatus { + c.notificationsByStatus.WithLabelValues(status).Set(float64(count)) + } + c.notificationsByType.Reset() + for typ, count := range stats.ByType { + c.notificationsByType.WithLabelValues(typ).Set(float64(count)) + } + } else if c.logger != nil { + c.logger.Debugf("metrics: failed to sample service stats: %v", err) + } + + if size, err := c.queue.Size(sampleCtx); err == nil { + c.queueDepth.Set(float64(size)) + } +} + +// InstrumentHTTP wraps an HTTP handler with request count and duration +// metrics. The path label uses the route pattern when available (via +// gorilla/mux CurrentRoute) to keep cardinality bounded. +func (c *Collector) InstrumentHTTP(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + start := time.Now() + sw := &statusWriter{ResponseWriter: w, status: http.StatusOK} + next.ServeHTTP(sw, r) + + path := routePattern(r) + c.httpRequests.WithLabelValues(r.Method, path, strconv.Itoa(sw.status)).Inc() + c.httpDuration.WithLabelValues(r.Method, path).Observe(time.Since(start).Seconds()) + }) +} + +type statusWriter struct { + http.ResponseWriter + status int +} + +func (w *statusWriter) WriteHeader(status int) { + w.status = status + w.ResponseWriter.WriteHeader(status) +} + +// routePattern extracts the mux route template (e.g. /api/v1/notifications/{id}) +// so metrics don't explode into one series per notification ID. +func routePattern(r *http.Request) string { + if route := currentRoute(r); route != "" { + return route + } + return "unmatched" +} diff --git a/internal/metrics/mux.go b/internal/metrics/mux.go new file mode 100644 index 0000000..05c7ce9 --- /dev/null +++ b/internal/metrics/mux.go @@ -0,0 +1,20 @@ +package metrics + +import ( + "net/http" + + "github.com/gorilla/mux" +) + +// currentRoute returns the gorilla/mux path template for the request, if the +// request was matched by a mux router. +func currentRoute(r *http.Request) string { + route := mux.CurrentRoute(r) + if route == nil { + return "" + } + if tmpl, err := route.GetPathTemplate(); err == nil { + return tmpl + } + return "" +}