feat(observability): slog-backed logging, Prometheus metrics, grpc health

- internal/logging now wraps log/slog; logging.format json/text finally works (json is the documented default). Same exported API. - New internal/metrics: /metrics on the configured metrics port with notification gauges by status/type, queue depth, and HTTP request count/duration labeled by mux route pattern; sampled from service stats so the service layer stays metrics-agnostic. - Standard grpc.health.v1 health service registered (k8s gRPC probes); gRPC MaxRecvMsgSize bounded to match the REST 1 MB body limit. - Dedicated health listener on health_check.port serving /health and /readyz (probes now work in grpc-only mode); metrics, health, and REST servers all shut down gracefully. - main wires retry backoff, CORS, readiness checks, and TLS from config. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-18 09:16:12 -07:00
parent ee82522b7c
commit 72f154ab07
5 changed files with 637 additions and 71 deletions
@@ -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 {