From 2d876329d7c16c6cb3293082f566761992b2ece4 Mon Sep 17 00:00:00 2001 From: Ed Snible Date: Fri, 31 Jul 2026 10:00:42 -0400 Subject: [PATCH 1/3] Refactor CLI common functions Signed-off-by: Ed Snible --- authbridge/authlib/runtime/runtime.go | 160 ++++++++++++++++++++++++ authbridge/cmd/authbridge-cpex/main.go | 120 ++---------------- authbridge/cmd/authbridge-envoy/main.go | 77 +----------- authbridge/cmd/authbridge-proxy/main.go | 127 ++----------------- 4 files changed, 182 insertions(+), 302 deletions(-) create mode 100644 authbridge/authlib/runtime/runtime.go diff --git a/authbridge/authlib/runtime/runtime.go b/authbridge/authlib/runtime/runtime.go new file mode 100644 index 00000000..ce61358b --- /dev/null +++ b/authbridge/authlib/runtime/runtime.go @@ -0,0 +1,160 @@ +// Package runtime holds process-level helpers shared by the authbridge +// binaries (authbridge-proxy, authbridge-cpex, authbridge-envoy). Each binary +// has its own main() orchestration and listener wiring; only the byte-identical +// plumbing — logging setup, the SIGUSR1 log-level toggle, the health and stats +// servers, and the HTTP-listener helpers — lives here so a fix has to be made +// once rather than three times. +// +// This package is intentionally listener-agnostic: it imports no gRPC and no +// cgo, so authlib does not gain the envoy (go-control-plane / grpc) or cpex +// (libcpex_ffi) dependency by hosting it. +package runtime + +import ( + "log" + "log/slog" + "net" + "net/http" + "os" + "os/signal" + "strings" + "syscall" + "time" + + "github.com/rossoctl/cortex/authbridge/authlib/config" + "github.com/rossoctl/cortex/authbridge/authlib/listener/reverseproxy" + "github.com/rossoctl/cortex/authbridge/authlib/observe" + "github.com/rossoctl/cortex/authbridge/authlib/pipeline" +) + +// logLevel is the process-wide slog level, mutated live by StartSignalToggle. +var logLevel = new(slog.LevelVar) + +// LogLevel returns the current process-wide slog level, for binaries that log +// their configured level at startup. +func LogLevel() slog.Level { return logLevel.Level() } + +// InitLogging sets the process log level from the LOG_LEVEL env var (debug / +// warn / error, default info) and installs a slog text handler on stderr. +func InitLogging() { + switch strings.ToLower(os.Getenv("LOG_LEVEL")) { + case "debug": + logLevel.Set(slog.LevelDebug) + case "warn": + logLevel.Set(slog.LevelWarn) + case "error": + logLevel.Set(slog.LevelError) + default: + logLevel.Set(slog.LevelInfo) + } + slog.SetDefault(slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: logLevel}))) +} + +// StartSignalToggle installs a SIGUSR1 handler that toggles the process log +// level between DEBUG and INFO, so operators can flip verbose logging on a +// running pod without a restart. +func StartSignalToggle() { + sigCh := make(chan os.Signal, 1) + signal.Notify(sigCh, syscall.SIGUSR1) + go func() { + for range sigCh { + if logLevel.Level() == slog.LevelDebug { + logLevel.Set(slog.LevelInfo) + slog.Info("log level toggled to INFO (send SIGUSR1 to switch back to DEBUG)") + } else { + logLevel.Set(slog.LevelDebug) + slog.Info("log level toggled to DEBUG (send SIGUSR1 to switch back to INFO)") + } + } + }() +} + +// StartHealthServer serves liveness (/healthz) and readiness (/readyz) on :9091 +// in a goroutine. Readiness reports 503 while any inbound or outbound plugin is +// still waiting on a dependency (e.g. a credential file that hasn't landed yet). +func StartHealthServer(inboundH, outboundH *pipeline.Holder) { + go func() { + mux := http.NewServeMux() + mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + }) + mux.HandleFunc("/readyz", func(w http.ResponseWriter, r *http.Request) { + if name := inboundH.NotReadyPlugin(); name != "" { + http.Error(w, "inbound plugin not ready: "+name, http.StatusServiceUnavailable) + return + } + if name := outboundH.NotReadyPlugin(); name != "" { + http.Error(w, "outbound plugin not ready: "+name, http.StatusServiceUnavailable) + return + } + w.WriteHeader(http.StatusOK) + }) + slog.Info("health server listening", "addr", ":9091") + if err := http.ListenAndServe(":9091", mux); err != nil { + slog.Warn("health server failed", "error", err) + } + }() +} + +// StartStatServer starts the stats/config-inspection server (default :9093) in +// a goroutine and returns it for graceful shutdown. +func StartStatServer(cfg *config.Config, cfgProvider observe.ConfigProvider, statsProvider observe.StatsProvider, reloadStatus http.Handler) *observe.StatServer { + srv := observe.NewStatServer(cfg.Stats.StatsAddress, cfgProvider, statsProvider, + observe.WithReloadStatus(reloadStatus)) + go func() { + slog.Info("stat server listening", "addr", cfg.Stats.StatsAddress) + if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { + log.Fatalf("stat server: %v", err) + } + }() + return srv +} + +// StartHTTPServer binds addr, serves handler in a goroutine, and returns the +// server for graceful shutdown. It logs the concrete bound address (resolving +// an ephemeral ":0" to the OS-assigned port). Bind failures are fatal. +func StartHTTPServer(name string, handler http.Handler, addr string) *http.Server { + srv := &http.Server{ + Addr: addr, + Handler: handler, + ReadHeaderTimeout: 10 * time.Second, + } + listener, err := net.Listen("tcp", addr) + if err != nil { + log.Fatalf("%s listen: %v", name, err) + } + go func() { + slog.Info("HTTP server listening", "name", name, "addr", listener.Addr().String()) + if err := srv.Serve(listener); err != nil && err != http.ErrServerClosed { + log.Fatalf("%s serve: %v", name, err) + } + }() + return srv +} + +// StartReverseProxyServer mirrors StartHTTPServer but uses the +// reverseproxy.Server's Listen() method so the byte-peek TLS-sniffing +// listener is wired in when mTLS is enabled. With mTLS off, Listen +// returns a plain net.Listen and behavior matches StartHTTPServer. +// +// Logged "mtls" attribute makes the listener mode visible at startup; +// operators expecting a separate :8443 port for TLS get a clear hint +// that this is the same :8080 with byte-peek detection. +func StartReverseProxyServer(name string, rp *reverseproxy.Server, addr string) *http.Server { + srv := &http.Server{ + Addr: addr, + Handler: rp.Handler(), + ReadHeaderTimeout: 10 * time.Second, + } + listener, err := rp.Listen(addr) + if err != nil { + log.Fatalf("%s listen: %v", name, err) + } + go func() { + slog.Info("Reverse server listening", "name", name, "addr", listener.Addr().String(), "mtls", rp.MTLSEnabled()) + if err := srv.Serve(listener); err != nil && err != http.ErrServerClosed { + log.Fatalf("%s serve: %v", name, err) + } + }() + return srv +} diff --git a/authbridge/cmd/authbridge-cpex/main.go b/authbridge/cmd/authbridge-cpex/main.go index db6a8cd1..228fda5b 100644 --- a/authbridge/cmd/authbridge-cpex/main.go +++ b/authbridge/cmd/authbridge-cpex/main.go @@ -24,20 +24,18 @@ import ( "fmt" "log" "log/slog" - "net" "net/http" "os" "os/signal" - "strings" "syscall" "time" "github.com/rossoctl/cortex/authbridge/authlib/auth" "github.com/rossoctl/cortex/authbridge/authlib/config" - "github.com/rossoctl/cortex/authbridge/authlib/observe" "github.com/rossoctl/cortex/authbridge/authlib/pipeline" "github.com/rossoctl/cortex/authbridge/authlib/plugins" "github.com/rossoctl/cortex/authbridge/authlib/reloader" + "github.com/rossoctl/cortex/authbridge/authlib/runtime" "github.com/rossoctl/cortex/authbridge/authlib/session" "github.com/rossoctl/cortex/authbridge/authlib/sessionapi" "github.com/rossoctl/cortex/authbridge/authlib/shared" @@ -62,44 +60,12 @@ import ( _ "github.com/rossoctl/cortex/authbridge/authlib/plugins/tokenexchange" ) -var logLevel = new(slog.LevelVar) - -func initLogging() { - switch strings.ToLower(os.Getenv("LOG_LEVEL")) { - case "debug": - logLevel.Set(slog.LevelDebug) - case "warn": - logLevel.Set(slog.LevelWarn) - case "error": - logLevel.Set(slog.LevelError) - default: - logLevel.Set(slog.LevelInfo) - } - slog.SetDefault(slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: logLevel}))) -} - -func startSignalToggle() { - sigCh := make(chan os.Signal, 1) - signal.Notify(sigCh, syscall.SIGUSR1) - go func() { - for range sigCh { - if logLevel.Level() == slog.LevelDebug { - logLevel.Set(slog.LevelInfo) - slog.Info("log level toggled to INFO (send SIGUSR1 to switch back to DEBUG)") - } else { - logLevel.Set(slog.LevelDebug) - slog.Info("log level toggled to DEBUG (send SIGUSR1 to switch back to INFO)") - } - } - }() -} - func main() { configPath := flag.String("config", "", "path to config YAML file") flag.Parse() - initLogging() - startSignalToggle() + runtime.InitLogging() + runtime.StartSignalToggle() if *configPath == "" { log.Fatal("--config is required and must point to a YAML file") @@ -236,8 +202,8 @@ func main() { defer sharedStore.Close() rpSrv.Shared = sharedStore fpSrv.Shared = sharedStore - httpServers = append(httpServers, startReverseProxyServer("reverse-proxy", rpSrv, cfg.Listener.ReverseProxyAddr)) - httpServers = append(httpServers, startHTTPServer("forward-proxy", fpSrv.Handler(), cfg.Listener.ForwardProxyAddr)) + httpServers = append(httpServers, runtime.StartReverseProxyServer("reverse-proxy", rpSrv, cfg.Listener.ReverseProxyAddr)) + httpServers = append(httpServers, runtime.StartHTTPServer("forward-proxy", fpSrv.Handler(), cfg.Listener.ForwardProxyAddr)) _ = mtlsMetrics statsProvider := func() *auth.Stats { @@ -245,7 +211,7 @@ func main() { sources = append(sources, plugins.CollectStats(outboundH.Load())...) return auth.MergeStats(sources...) } - statSrv := startStatServer(cfg, rld.ConfigProvider(), statsProvider, rld.Handler()) + statSrv := runtime.StartStatServer(cfg, rld.ConfigProvider(), statsProvider, rld.Handler()) plugins.WarmCatalog() @@ -266,29 +232,9 @@ func main() { }() } - slog.Info("authbridge-cpex starting", "mode", cfg.Mode, "logLevel", logLevel.Level().String()) + slog.Info("authbridge-cpex starting", "mode", cfg.Mode, "logLevel", runtime.LogLevel().String()) - go func() { - mux := http.NewServeMux() - mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { - w.WriteHeader(http.StatusOK) - }) - mux.HandleFunc("/readyz", func(w http.ResponseWriter, r *http.Request) { - if name := inboundH.NotReadyPlugin(); name != "" { - http.Error(w, "inbound plugin not ready: "+name, http.StatusServiceUnavailable) - return - } - if name := outboundH.NotReadyPlugin(); name != "" { - http.Error(w, "outbound plugin not ready: "+name, http.StatusServiceUnavailable) - return - } - w.WriteHeader(http.StatusOK) - }) - slog.Info("health server listening", "addr", ":9091") - if err := http.ListenAndServe(":9091", mux); err != nil { - slog.Warn("health server failed", "error", err) - } - }() + runtime.StartHealthServer(inboundH, outboundH) sigCh := make(chan os.Signal, 1) signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT) @@ -313,53 +259,3 @@ func main() { sessions.Close() } } - -func startHTTPServer(name string, handler http.Handler, addr string) *http.Server { - srv := &http.Server{ - Addr: addr, - Handler: handler, - ReadHeaderTimeout: 10 * time.Second, - } - listener, err := net.Listen("tcp", addr) - if err != nil { - log.Fatalf("%s listen: %v", name, err) - } - go func() { - slog.Info("HTTP server listening", "name", name, "addr", listener.Addr().String()) - if err := srv.Serve(listener); err != nil && err != http.ErrServerClosed { - log.Fatalf("%s serve: %v", name, err) - } - }() - return srv -} - -func startReverseProxyServer(name string, rp *reverseproxy.Server, addr string) *http.Server { - srv := &http.Server{ - Addr: addr, - Handler: rp.Handler(), - ReadHeaderTimeout: 10 * time.Second, - } - listener, err := rp.Listen(addr) - if err != nil { - log.Fatalf("%s listen: %v", name, err) - } - go func() { - slog.Info("Reverse server listening", "name", name, "addr", listener.Addr().String(), "mtls", rp.MTLSEnabled()) - if err := srv.Serve(listener); err != nil && err != http.ErrServerClosed { - log.Fatalf("%s serve: %v", name, err) - } - }() - return srv -} - -func startStatServer(cfg *config.Config, cfgProvider observe.ConfigProvider, statsProvider observe.StatsProvider, reloadStatus http.Handler) *observe.StatServer { - srv := observe.NewStatServer(cfg.Stats.StatsAddress, cfgProvider, statsProvider, - observe.WithReloadStatus(reloadStatus)) - go func() { - slog.Info("stat server listening", "addr", cfg.Stats.StatsAddress) - if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { - log.Fatalf("stat server: %v", err) - } - }() - return srv -} diff --git a/authbridge/cmd/authbridge-envoy/main.go b/authbridge/cmd/authbridge-envoy/main.go index 183c9e8e..be9b0605 100644 --- a/authbridge/cmd/authbridge-envoy/main.go +++ b/authbridge/cmd/authbridge-envoy/main.go @@ -21,7 +21,6 @@ import ( "net/http" "os" "os/signal" - "strings" "syscall" "time" @@ -33,10 +32,10 @@ import ( "github.com/rossoctl/cortex/authbridge/authlib/auth" "github.com/rossoctl/cortex/authbridge/authlib/config" - "github.com/rossoctl/cortex/authbridge/authlib/observe" "github.com/rossoctl/cortex/authbridge/authlib/pipeline" "github.com/rossoctl/cortex/authbridge/authlib/plugins" "github.com/rossoctl/cortex/authbridge/authlib/reloader" + "github.com/rossoctl/cortex/authbridge/authlib/runtime" "github.com/rossoctl/cortex/authbridge/authlib/session" "github.com/rossoctl/cortex/authbridge/authlib/sessionapi" "github.com/rossoctl/cortex/authbridge/authlib/shared" @@ -59,44 +58,12 @@ import ( _ "github.com/rossoctl/cortex/authbridge/authlib/plugins/tokenexchange" ) -var logLevel = new(slog.LevelVar) - -func initLogging() { - switch strings.ToLower(os.Getenv("LOG_LEVEL")) { - case "debug": - logLevel.Set(slog.LevelDebug) - case "warn": - logLevel.Set(slog.LevelWarn) - case "error": - logLevel.Set(slog.LevelError) - default: - logLevel.Set(slog.LevelInfo) - } - slog.SetDefault(slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: logLevel}))) -} - -func startSignalToggle() { - sigCh := make(chan os.Signal, 1) - signal.Notify(sigCh, syscall.SIGUSR1) - go func() { - for range sigCh { - if logLevel.Level() == slog.LevelDebug { - logLevel.Set(slog.LevelInfo) - slog.Info("log level toggled to INFO (send SIGUSR1 to switch back to DEBUG)") - } else { - logLevel.Set(slog.LevelDebug) - slog.Info("log level toggled to DEBUG (send SIGUSR1 to switch back to INFO)") - } - } - }() -} - func main() { configPath := flag.String("config", "", "path to config YAML file") flag.Parse() - initLogging() - startSignalToggle() + runtime.InitLogging() + runtime.StartSignalToggle() if *configPath == "" { log.Fatal("--config is required and must point to a YAML file") @@ -235,7 +202,7 @@ func main() { sources = append(sources, plugins.CollectStats(outboundH.Load())...) return auth.MergeStats(sources...) } - statSrv := startStatServer(cfg, rld.ConfigProvider(), statsProvider, rld.Handler()) + statSrv := runtime.StartStatServer(cfg, rld.ConfigProvider(), statsProvider, rld.Handler()) // Warm the plugin catalog at boot so any factory that violates the // constructor contract surfaces here rather than on the first @@ -259,29 +226,9 @@ func main() { }() } - slog.Info("authbridge-envoy starting", "mode", cfg.Mode, "logLevel", logLevel.Level().String()) + slog.Info("authbridge-envoy starting", "mode", cfg.Mode, "logLevel", runtime.LogLevel().String()) - go func() { - mux := http.NewServeMux() - mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { - w.WriteHeader(http.StatusOK) - }) - mux.HandleFunc("/readyz", func(w http.ResponseWriter, r *http.Request) { - if name := inboundH.NotReadyPlugin(); name != "" { - http.Error(w, "inbound plugin not ready: "+name, http.StatusServiceUnavailable) - return - } - if name := outboundH.NotReadyPlugin(); name != "" { - http.Error(w, "outbound plugin not ready: "+name, http.StatusServiceUnavailable) - return - } - w.WriteHeader(http.StatusOK) - }) - slog.Info("health server listening", "addr", ":9091") - if err := http.ListenAndServe(":9091", mux); err != nil { - slog.Warn("health server failed", "error", err) - } - }() + runtime.StartHealthServer(inboundH, outboundH) sigCh := make(chan os.Signal, 1) signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT) @@ -336,18 +283,6 @@ func startGRPCExtProc(inbound, outbound *pipeline.Holder, sessions *session.Stor return srv } -func startStatServer(cfg *config.Config, cfgProvider observe.ConfigProvider, statsProvider observe.StatsProvider, reloadStatus http.Handler) *observe.StatServer { - srv := observe.NewStatServer(cfg.Stats.StatsAddress, cfgProvider, statsProvider, - observe.WithReloadStatus(reloadStatus)) - go func() { - slog.Info("stat server listening", "addr", cfg.Stats.StatsAddress) - if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { - log.Fatalf("stat server: %v", err) - } - }() - return srv -} - func registerHealth(srv *grpc.Server) { healthSrv := health.NewServer() healthpb.RegisterHealthServer(srv, healthSrv) diff --git a/authbridge/cmd/authbridge-proxy/main.go b/authbridge/cmd/authbridge-proxy/main.go index 2da2c840..f30dd5a4 100644 --- a/authbridge/cmd/authbridge-proxy/main.go +++ b/authbridge/cmd/authbridge-proxy/main.go @@ -27,16 +27,15 @@ import ( "os" "os/signal" "path/filepath" - "strings" "syscall" "time" "github.com/rossoctl/cortex/authbridge/authlib/auth" "github.com/rossoctl/cortex/authbridge/authlib/config" - "github.com/rossoctl/cortex/authbridge/authlib/observe" "github.com/rossoctl/cortex/authbridge/authlib/pipeline" "github.com/rossoctl/cortex/authbridge/authlib/plugins" "github.com/rossoctl/cortex/authbridge/authlib/reloader" + "github.com/rossoctl/cortex/authbridge/authlib/runtime" "github.com/rossoctl/cortex/authbridge/authlib/session" "github.com/rossoctl/cortex/authbridge/authlib/sessionapi" "github.com/rossoctl/cortex/authbridge/authlib/shared" @@ -56,8 +55,6 @@ import ( // authbridge-lite image excludes all but jwt-validation + token-exchange. ) -var logLevel = new(slog.LevelVar) - // version is the authbridge-proxy build version, overridden at release time // via -ldflags "-X main.version=". Defaults to "dev" for local builds. var version = "dev" @@ -70,36 +67,6 @@ var version = "dev" // gate is the only way to keep the demo to the listeners it actually uses. var demoMode bool -func initLogging() { - switch strings.ToLower(os.Getenv("LOG_LEVEL")) { - case "debug": - logLevel.Set(slog.LevelDebug) - case "warn": - logLevel.Set(slog.LevelWarn) - case "error": - logLevel.Set(slog.LevelError) - default: - logLevel.Set(slog.LevelInfo) - } - slog.SetDefault(slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: logLevel}))) -} - -func startSignalToggle() { - sigCh := make(chan os.Signal, 1) - signal.Notify(sigCh, syscall.SIGUSR1) - go func() { - for range sigCh { - if logLevel.Level() == slog.LevelDebug { - logLevel.Set(slog.LevelInfo) - slog.Info("log level toggled to INFO (send SIGUSR1 to switch back to DEBUG)") - } else { - logLevel.Set(slog.LevelDebug) - slog.Info("log level toggled to DEBUG (send SIGUSR1 to switch back to INFO)") - } - } - }() -} - // spiffeProviderNeeded reports whether any configured feature actually consumes // the SPIFFE Provider: top-level mTLS (needs the X509Source on both listeners) // or a plugin whose identity is spiffe-based (needs the JWT-SVID source — today @@ -166,8 +133,8 @@ func main() { return } - initLogging() - startSignalToggle() + runtime.InitLogging() + runtime.StartSignalToggle() if *demo { demoMode = true @@ -422,7 +389,7 @@ func main() { log.Fatalf("creating reverse proxy: %v", rerr) } rpSrv.Shared = sharedStore - httpServers = append(httpServers, startReverseProxyServer("reverse-proxy", rpSrv, cfg.Listener.ReverseProxyAddr)) + httpServers = append(httpServers, runtime.StartReverseProxyServer("reverse-proxy", rpSrv, cfg.Listener.ReverseProxyAddr)) } // The transparent (enforce-redirect) listener rides with the forward proxy; @@ -444,7 +411,7 @@ func main() { fpSrv.SkipHosts = skipHosts fpSrv.TLSBridge = bridge fpSrv.Shared = sharedStore - httpServers = append(httpServers, startHTTPServer("forward-proxy", fpSrv.Handler(), cfg.Listener.ForwardProxyAddr)) + httpServers = append(httpServers, runtime.StartHTTPServer("forward-proxy", fpSrv.Handler(), cfg.Listener.ForwardProxyAddr)) // Outbound transparent listener (enforce-redirect mode). It shares the // forward proxy's outbound pipeline via HandleTransparentConn, so explicit @@ -463,7 +430,7 @@ func main() { sources = append(sources, plugins.CollectStats(outboundH.Load())...) return auth.MergeStats(sources...) } - statSrv := startStatServer(cfg, rld.ConfigProvider(), statsProvider, rld.Handler()) + statSrv := runtime.StartStatServer(cfg, rld.ConfigProvider(), statsProvider, rld.Handler()) // Warm the plugin catalog at boot so any factory that violates the // constructor contract surfaces here rather than on the first @@ -487,29 +454,9 @@ func main() { }() } - slog.Info("authbridge-proxy starting", "version", version, "mode", cfg.Mode, "logLevel", logLevel.Level().String()) + slog.Info("authbridge-proxy starting", "version", version, "mode", cfg.Mode, "logLevel", runtime.LogLevel().String()) - go func() { - mux := http.NewServeMux() - mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { - w.WriteHeader(http.StatusOK) - }) - mux.HandleFunc("/readyz", func(w http.ResponseWriter, r *http.Request) { - if name := inboundH.NotReadyPlugin(); name != "" { - http.Error(w, "inbound plugin not ready: "+name, http.StatusServiceUnavailable) - return - } - if name := outboundH.NotReadyPlugin(); name != "" { - http.Error(w, "outbound plugin not ready: "+name, http.StatusServiceUnavailable) - return - } - w.WriteHeader(http.StatusOK) - }) - slog.Info("health server listening", "addr", ":9091") - if err := http.ListenAndServe(":9091", mux); err != nil { - slog.Warn("health server failed", "error", err) - } - }() + runtime.StartHealthServer(inboundH, outboundH) sigCh := make(chan os.Signal, 1) signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT) @@ -538,52 +485,6 @@ func main() { } } -func startHTTPServer(name string, handler http.Handler, addr string) *http.Server { - srv := &http.Server{ - Addr: addr, - Handler: handler, - ReadHeaderTimeout: 10 * time.Second, - } - listener, err := net.Listen("tcp", addr) - if err != nil { - log.Fatalf("%s listen: %v", name, err) - } - go func() { - slog.Info("HTTP server listening", "name", name, "addr", listener.Addr().String()) - if err := srv.Serve(listener); err != nil && err != http.ErrServerClosed { - log.Fatalf("%s serve: %v", name, err) - } - }() - return srv -} - -// startReverseProxyServer mirrors startHTTPServer but uses the -// reverseproxy.Server's Listen() method so the byte-peek TLS-sniffing -// listener is wired in when mTLS is enabled. With mTLS off, Listen -// returns a plain net.Listen and behavior matches startHTTPServer. -// -// Logged "mtls" attribute makes the listener mode visible at startup; -// operators expecting a separate :8443 port for TLS get a clear hint -// that this is the same :8080 with byte-peek detection. -func startReverseProxyServer(name string, rp *reverseproxy.Server, addr string) *http.Server { - srv := &http.Server{ - Addr: addr, - Handler: rp.Handler(), - ReadHeaderTimeout: 10 * time.Second, - } - listener, err := rp.Listen(addr) - if err != nil { - log.Fatalf("%s listen: %v", name, err) - } - go func() { - slog.Info("Reverse server listening", "name", name, "addr", listener.Addr().String(), "mtls", rp.MTLSEnabled()) - if err := srv.Serve(listener); err != nil && err != http.ErrServerClosed { - log.Fatalf("%s serve: %v", name, err) - } - }() - return srv -} - // startTransparentProxy binds the outbound transparent listener and serves it // in a goroutine, dispatching each REDIRECTed connection through the forward // proxy's outbound pipeline. Returns the listener (for shutdown), or nil when @@ -611,15 +512,3 @@ func startTransparentProxy(fp *forwardproxy.Server, addr string) *net.TCPListene }() return ln } - -func startStatServer(cfg *config.Config, cfgProvider observe.ConfigProvider, statsProvider observe.StatsProvider, reloadStatus http.Handler) *observe.StatServer { - srv := observe.NewStatServer(cfg.Stats.StatsAddress, cfgProvider, statsProvider, - observe.WithReloadStatus(reloadStatus)) - go func() { - slog.Info("stat server listening", "addr", cfg.Stats.StatsAddress) - if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { - log.Fatalf("stat server: %v", err) - } - }() - return srv -} From 2d8a5ea0cc4adf99626fa21726923b92f0ccf1c0 Mon Sep 17 00:00:00 2001 From: Ed Snible Date: Fri, 31 Jul 2026 13:18:05 -0400 Subject: [PATCH 2/3] Address review comments Signed-off-by: Ed Snible --- authbridge/authlib/observe/statserver.go | 7 ++ .../runtime.go => runtimeutil/runtimeutil.go} | 67 +++++++++-------- .../authlib/runtimeutil/runtimeutil_test.go | 72 +++++++++++++++++++ authbridge/cmd/authbridge-cpex/main.go | 27 ++++--- authbridge/cmd/authbridge-envoy/main.go | 15 ++-- authbridge/cmd/authbridge-proxy/main.go | 27 ++++--- 6 files changed, 162 insertions(+), 53 deletions(-) rename authbridge/authlib/{runtime/runtime.go => runtimeutil/runtimeutil.go} (72%) create mode 100644 authbridge/authlib/runtimeutil/runtimeutil_test.go diff --git a/authbridge/authlib/observe/statserver.go b/authbridge/authlib/observe/statserver.go index 7750fa40..ae2c423b 100644 --- a/authbridge/authlib/observe/statserver.go +++ b/authbridge/authlib/observe/statserver.go @@ -5,6 +5,7 @@ import ( "encoding/json" "fmt" "log/slog" + "net" "net/http" "time" @@ -126,6 +127,12 @@ func (s *StatServer) ListenAndServe() error { return s.server.ListenAndServe() } +// Serve serves on an already-bound listener, letting the caller bind first (and +// handle a bind error synchronously) before the server starts accepting. +func (s *StatServer) Serve(l net.Listener) error { + return s.server.Serve(l) +} + func (s *StatServer) Shutdown(ctx context.Context) error { return s.server.Shutdown(ctx) } diff --git a/authbridge/authlib/runtime/runtime.go b/authbridge/authlib/runtimeutil/runtimeutil.go similarity index 72% rename from authbridge/authlib/runtime/runtime.go rename to authbridge/authlib/runtimeutil/runtimeutil.go index ce61358b..212dc547 100644 --- a/authbridge/authlib/runtime/runtime.go +++ b/authbridge/authlib/runtimeutil/runtimeutil.go @@ -1,17 +1,12 @@ -// Package runtime holds process-level helpers shared by the authbridge +// Package runtimeutil holds process-level helpers shared by the authbridge // binaries (authbridge-proxy, authbridge-cpex, authbridge-envoy). Each binary // has its own main() orchestration and listener wiring; only the byte-identical // plumbing — logging setup, the SIGUSR1 log-level toggle, the health and stats // servers, and the HTTP-listener helpers — lives here so a fix has to be made // once rather than three times. -// -// This package is intentionally listener-agnostic: it imports no gRPC and no -// cgo, so authlib does not gain the envoy (go-control-plane / grpc) or cpex -// (libcpex_ffi) dependency by hosting it. -package runtime +package runtimeutil import ( - "log" "log/slog" "net" "net/http" @@ -35,8 +30,10 @@ var logLevel = new(slog.LevelVar) func LogLevel() slog.Level { return logLevel.Level() } // InitLogging sets the process log level from the LOG_LEVEL env var (debug / -// warn / error, default info) and installs a slog text handler on stderr. -func InitLogging() { +// warn / error, default info) and installs a slog text handler on stderr. The +// binaryName is attached to every record as the "binary" attribute so logs from +// co-located sidecars stay distinguishable. +func InitLogging(binaryName string) { switch strings.ToLower(os.Getenv("LOG_LEVEL")) { case "debug": logLevel.Set(slog.LevelDebug) @@ -47,7 +44,8 @@ func InitLogging() { default: logLevel.Set(slog.LevelInfo) } - slog.SetDefault(slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: logLevel}))) + h := slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: logLevel}) + slog.SetDefault(slog.New(h).With("binary", binaryName)) } // StartSignalToggle installs a SIGUSR1 handler that toggles the process log @@ -69,10 +67,10 @@ func StartSignalToggle() { }() } -// StartHealthServer serves liveness (/healthz) and readiness (/readyz) on :9091 +// StartHealthServer serves liveness (/healthz) and readiness (/readyz) on addr // in a goroutine. Readiness reports 503 while any inbound or outbound plugin is // still waiting on a dependency (e.g. a credential file that hasn't landed yet). -func StartHealthServer(inboundH, outboundH *pipeline.Holder) { +func StartHealthServer(inboundH, outboundH *pipeline.Holder, addr string) { go func() { mux := http.NewServeMux() mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { @@ -89,31 +87,38 @@ func StartHealthServer(inboundH, outboundH *pipeline.Holder) { } w.WriteHeader(http.StatusOK) }) - slog.Info("health server listening", "addr", ":9091") - if err := http.ListenAndServe(":9091", mux); err != nil { + slog.Info("health server listening", "addr", addr) + if err := http.ListenAndServe(addr, mux); err != nil { slog.Warn("health server failed", "error", err) } }() } -// StartStatServer starts the stats/config-inspection server (default :9093) in -// a goroutine and returns it for graceful shutdown. -func StartStatServer(cfg *config.Config, cfgProvider observe.ConfigProvider, statsProvider observe.StatsProvider, reloadStatus http.Handler) *observe.StatServer { - srv := observe.NewStatServer(cfg.Stats.StatsAddress, cfgProvider, statsProvider, +// StartStatServer binds addr for the stats/config-inspection server, serves it +// in a goroutine, and returns the server for graceful shutdown. A bind failure +// is returned so the caller can decide how to handle it (the mains log.Fatalf); +// a serve-time failure after bind is logged. +func StartStatServer(cfg *config.Config, cfgProvider observe.ConfigProvider, statsProvider observe.StatsProvider, reloadStatus http.Handler, addr string) (*observe.StatServer, error) { + srv := observe.NewStatServer(addr, cfgProvider, statsProvider, observe.WithReloadStatus(reloadStatus)) + listener, err := net.Listen("tcp", addr) + if err != nil { + return nil, err + } go func() { - slog.Info("stat server listening", "addr", cfg.Stats.StatsAddress) - if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { - log.Fatalf("stat server: %v", err) + slog.Info("stat server listening", "addr", listener.Addr().String()) + if err := srv.Serve(listener); err != nil && err != http.ErrServerClosed { + slog.Error("stat server failed", "error", err) } }() - return srv + return srv, nil } // StartHTTPServer binds addr, serves handler in a goroutine, and returns the // server for graceful shutdown. It logs the concrete bound address (resolving -// an ephemeral ":0" to the OS-assigned port). Bind failures are fatal. -func StartHTTPServer(name string, handler http.Handler, addr string) *http.Server { +// an ephemeral ":0" to the OS-assigned port). A bind failure is returned so the +// caller can decide how to handle it; a serve-time failure after bind is logged. +func StartHTTPServer(name string, handler http.Handler, addr string) (*http.Server, error) { srv := &http.Server{ Addr: addr, Handler: handler, @@ -121,15 +126,15 @@ func StartHTTPServer(name string, handler http.Handler, addr string) *http.Serve } listener, err := net.Listen("tcp", addr) if err != nil { - log.Fatalf("%s listen: %v", name, err) + return nil, err } go func() { slog.Info("HTTP server listening", "name", name, "addr", listener.Addr().String()) if err := srv.Serve(listener); err != nil && err != http.ErrServerClosed { - log.Fatalf("%s serve: %v", name, err) + slog.Error("HTTP server failed", "name", name, "error", err) } }() - return srv + return srv, nil } // StartReverseProxyServer mirrors StartHTTPServer but uses the @@ -140,7 +145,7 @@ func StartHTTPServer(name string, handler http.Handler, addr string) *http.Serve // Logged "mtls" attribute makes the listener mode visible at startup; // operators expecting a separate :8443 port for TLS get a clear hint // that this is the same :8080 with byte-peek detection. -func StartReverseProxyServer(name string, rp *reverseproxy.Server, addr string) *http.Server { +func StartReverseProxyServer(name string, rp *reverseproxy.Server, addr string) (*http.Server, error) { srv := &http.Server{ Addr: addr, Handler: rp.Handler(), @@ -148,13 +153,13 @@ func StartReverseProxyServer(name string, rp *reverseproxy.Server, addr string) } listener, err := rp.Listen(addr) if err != nil { - log.Fatalf("%s listen: %v", name, err) + return nil, err } go func() { slog.Info("Reverse server listening", "name", name, "addr", listener.Addr().String(), "mtls", rp.MTLSEnabled()) if err := srv.Serve(listener); err != nil && err != http.ErrServerClosed { - log.Fatalf("%s serve: %v", name, err) + slog.Error("Reverse server failed", "name", name, "error", err) } }() - return srv + return srv, nil } diff --git a/authbridge/authlib/runtimeutil/runtimeutil_test.go b/authbridge/authlib/runtimeutil/runtimeutil_test.go new file mode 100644 index 00000000..206275a4 --- /dev/null +++ b/authbridge/authlib/runtimeutil/runtimeutil_test.go @@ -0,0 +1,72 @@ +package runtimeutil + +import ( + "log/slog" + "syscall" + "testing" + "time" +) + +// TestInitLogging verifies InitLogging maps LOG_LEVEL to the process level and +// defaults to INFO for unset/unknown values. +func TestInitLogging(t *testing.T) { + cases := []struct { + env string + want slog.Level + }{ + {"debug", slog.LevelDebug}, + {"DEBUG", slog.LevelDebug}, // case-insensitive + {"warn", slog.LevelWarn}, + {"error", slog.LevelError}, + {"info", slog.LevelInfo}, + {"", slog.LevelInfo}, // unset -> default + {"nonsense", slog.LevelInfo}, // unknown -> default + } + for _, tc := range cases { + t.Run(tc.env, func(t *testing.T) { + t.Setenv("LOG_LEVEL", tc.env) + InitLogging("test-binary") + if got := LogLevel(); got != tc.want { + t.Fatalf("LOG_LEVEL=%q: LogLevel()=%v, want %v", tc.env, got, tc.want) + } + }) + } +} + +// TestStartSignalToggle verifies a SIGUSR1 flips the level between DEBUG and +// INFO on each delivery. +func TestStartSignalToggle(t *testing.T) { + t.Setenv("LOG_LEVEL", "info") + InitLogging("test-binary") + if LogLevel() != slog.LevelInfo { + t.Fatalf("precondition: want INFO, got %v", LogLevel()) + } + + StartSignalToggle() + + // INFO -> DEBUG + if err := syscall.Kill(syscall.Getpid(), syscall.SIGUSR1); err != nil { + t.Fatalf("kill SIGUSR1: %v", err) + } + waitForLevel(t, slog.LevelDebug) + + // DEBUG -> INFO + if err := syscall.Kill(syscall.Getpid(), syscall.SIGUSR1); err != nil { + t.Fatalf("kill SIGUSR1: %v", err) + } + waitForLevel(t, slog.LevelInfo) +} + +// waitForLevel polls until the process log level reaches want, since the signal +// is handled asynchronously by the goroutine StartSignalToggle launched. +func waitForLevel(t *testing.T, want slog.Level) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if LogLevel() == want { + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatalf("timed out waiting for level %v (still %v)", want, LogLevel()) +} diff --git a/authbridge/cmd/authbridge-cpex/main.go b/authbridge/cmd/authbridge-cpex/main.go index 228fda5b..10a2b7f8 100644 --- a/authbridge/cmd/authbridge-cpex/main.go +++ b/authbridge/cmd/authbridge-cpex/main.go @@ -35,7 +35,7 @@ import ( "github.com/rossoctl/cortex/authbridge/authlib/pipeline" "github.com/rossoctl/cortex/authbridge/authlib/plugins" "github.com/rossoctl/cortex/authbridge/authlib/reloader" - "github.com/rossoctl/cortex/authbridge/authlib/runtime" + "github.com/rossoctl/cortex/authbridge/authlib/runtimeutil" "github.com/rossoctl/cortex/authbridge/authlib/session" "github.com/rossoctl/cortex/authbridge/authlib/sessionapi" "github.com/rossoctl/cortex/authbridge/authlib/shared" @@ -64,8 +64,8 @@ func main() { configPath := flag.String("config", "", "path to config YAML file") flag.Parse() - runtime.InitLogging() - runtime.StartSignalToggle() + runtimeutil.InitLogging("authbridge-cpex") + runtimeutil.StartSignalToggle() if *configPath == "" { log.Fatal("--config is required and must point to a YAML file") @@ -202,8 +202,16 @@ func main() { defer sharedStore.Close() rpSrv.Shared = sharedStore fpSrv.Shared = sharedStore - httpServers = append(httpServers, runtime.StartReverseProxyServer("reverse-proxy", rpSrv, cfg.Listener.ReverseProxyAddr)) - httpServers = append(httpServers, runtime.StartHTTPServer("forward-proxy", fpSrv.Handler(), cfg.Listener.ForwardProxyAddr)) + rpHTTP, err := runtimeutil.StartReverseProxyServer("reverse-proxy", rpSrv, cfg.Listener.ReverseProxyAddr) + if err != nil { + log.Fatalf("reverse-proxy listen: %v", err) + } + httpServers = append(httpServers, rpHTTP) + fpHTTP, err := runtimeutil.StartHTTPServer("forward-proxy", fpSrv.Handler(), cfg.Listener.ForwardProxyAddr) + if err != nil { + log.Fatalf("forward-proxy listen: %v", err) + } + httpServers = append(httpServers, fpHTTP) _ = mtlsMetrics statsProvider := func() *auth.Stats { @@ -211,7 +219,10 @@ func main() { sources = append(sources, plugins.CollectStats(outboundH.Load())...) return auth.MergeStats(sources...) } - statSrv := runtime.StartStatServer(cfg, rld.ConfigProvider(), statsProvider, rld.Handler()) + statSrv, statErr := runtimeutil.StartStatServer(cfg, rld.ConfigProvider(), statsProvider, rld.Handler(), cfg.Stats.StatsAddress) + if statErr != nil { + log.Fatalf("stat server listen: %v", statErr) + } plugins.WarmCatalog() @@ -232,9 +243,9 @@ func main() { }() } - slog.Info("authbridge-cpex starting", "mode", cfg.Mode, "logLevel", runtime.LogLevel().String()) + slog.Info("authbridge-cpex starting", "mode", cfg.Mode, "logLevel", runtimeutil.LogLevel().String()) - runtime.StartHealthServer(inboundH, outboundH) + runtimeutil.StartHealthServer(inboundH, outboundH, ":9091") sigCh := make(chan os.Signal, 1) signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT) diff --git a/authbridge/cmd/authbridge-envoy/main.go b/authbridge/cmd/authbridge-envoy/main.go index be9b0605..b1a4161f 100644 --- a/authbridge/cmd/authbridge-envoy/main.go +++ b/authbridge/cmd/authbridge-envoy/main.go @@ -35,7 +35,7 @@ import ( "github.com/rossoctl/cortex/authbridge/authlib/pipeline" "github.com/rossoctl/cortex/authbridge/authlib/plugins" "github.com/rossoctl/cortex/authbridge/authlib/reloader" - "github.com/rossoctl/cortex/authbridge/authlib/runtime" + "github.com/rossoctl/cortex/authbridge/authlib/runtimeutil" "github.com/rossoctl/cortex/authbridge/authlib/session" "github.com/rossoctl/cortex/authbridge/authlib/sessionapi" "github.com/rossoctl/cortex/authbridge/authlib/shared" @@ -62,8 +62,8 @@ func main() { configPath := flag.String("config", "", "path to config YAML file") flag.Parse() - runtime.InitLogging() - runtime.StartSignalToggle() + runtimeutil.InitLogging("authbridge-envoy") + runtimeutil.StartSignalToggle() if *configPath == "" { log.Fatal("--config is required and must point to a YAML file") @@ -202,7 +202,10 @@ func main() { sources = append(sources, plugins.CollectStats(outboundH.Load())...) return auth.MergeStats(sources...) } - statSrv := runtime.StartStatServer(cfg, rld.ConfigProvider(), statsProvider, rld.Handler()) + statSrv, statErr := runtimeutil.StartStatServer(cfg, rld.ConfigProvider(), statsProvider, rld.Handler(), cfg.Stats.StatsAddress) + if statErr != nil { + log.Fatalf("stat server listen: %v", statErr) + } // Warm the plugin catalog at boot so any factory that violates the // constructor contract surfaces here rather than on the first @@ -226,9 +229,9 @@ func main() { }() } - slog.Info("authbridge-envoy starting", "mode", cfg.Mode, "logLevel", runtime.LogLevel().String()) + slog.Info("authbridge-envoy starting", "mode", cfg.Mode, "logLevel", runtimeutil.LogLevel().String()) - runtime.StartHealthServer(inboundH, outboundH) + runtimeutil.StartHealthServer(inboundH, outboundH, ":9091") sigCh := make(chan os.Signal, 1) signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT) diff --git a/authbridge/cmd/authbridge-proxy/main.go b/authbridge/cmd/authbridge-proxy/main.go index f30dd5a4..bbc9b3e2 100644 --- a/authbridge/cmd/authbridge-proxy/main.go +++ b/authbridge/cmd/authbridge-proxy/main.go @@ -35,7 +35,7 @@ import ( "github.com/rossoctl/cortex/authbridge/authlib/pipeline" "github.com/rossoctl/cortex/authbridge/authlib/plugins" "github.com/rossoctl/cortex/authbridge/authlib/reloader" - "github.com/rossoctl/cortex/authbridge/authlib/runtime" + "github.com/rossoctl/cortex/authbridge/authlib/runtimeutil" "github.com/rossoctl/cortex/authbridge/authlib/session" "github.com/rossoctl/cortex/authbridge/authlib/sessionapi" "github.com/rossoctl/cortex/authbridge/authlib/shared" @@ -133,8 +133,8 @@ func main() { return } - runtime.InitLogging() - runtime.StartSignalToggle() + runtimeutil.InitLogging("authbridge-proxy") + runtimeutil.StartSignalToggle() if *demo { demoMode = true @@ -389,7 +389,11 @@ func main() { log.Fatalf("creating reverse proxy: %v", rerr) } rpSrv.Shared = sharedStore - httpServers = append(httpServers, runtime.StartReverseProxyServer("reverse-proxy", rpSrv, cfg.Listener.ReverseProxyAddr)) + rpHTTP, rerr := runtimeutil.StartReverseProxyServer("reverse-proxy", rpSrv, cfg.Listener.ReverseProxyAddr) + if rerr != nil { + log.Fatalf("reverse-proxy listen: %v", rerr) + } + httpServers = append(httpServers, rpHTTP) } // The transparent (enforce-redirect) listener rides with the forward proxy; @@ -411,7 +415,11 @@ func main() { fpSrv.SkipHosts = skipHosts fpSrv.TLSBridge = bridge fpSrv.Shared = sharedStore - httpServers = append(httpServers, runtime.StartHTTPServer("forward-proxy", fpSrv.Handler(), cfg.Listener.ForwardProxyAddr)) + fpHTTP, herr := runtimeutil.StartHTTPServer("forward-proxy", fpSrv.Handler(), cfg.Listener.ForwardProxyAddr) + if herr != nil { + log.Fatalf("forward-proxy listen: %v", herr) + } + httpServers = append(httpServers, fpHTTP) // Outbound transparent listener (enforce-redirect mode). It shares the // forward proxy's outbound pipeline via HandleTransparentConn, so explicit @@ -430,7 +438,10 @@ func main() { sources = append(sources, plugins.CollectStats(outboundH.Load())...) return auth.MergeStats(sources...) } - statSrv := runtime.StartStatServer(cfg, rld.ConfigProvider(), statsProvider, rld.Handler()) + statSrv, statErr := runtimeutil.StartStatServer(cfg, rld.ConfigProvider(), statsProvider, rld.Handler(), cfg.Stats.StatsAddress) + if statErr != nil { + log.Fatalf("stat server listen: %v", statErr) + } // Warm the plugin catalog at boot so any factory that violates the // constructor contract surfaces here rather than on the first @@ -454,9 +465,9 @@ func main() { }() } - slog.Info("authbridge-proxy starting", "version", version, "mode", cfg.Mode, "logLevel", runtime.LogLevel().String()) + slog.Info("authbridge-proxy starting", "version", version, "mode", cfg.Mode, "logLevel", runtimeutil.LogLevel().String()) - runtime.StartHealthServer(inboundH, outboundH) + runtimeutil.StartHealthServer(inboundH, outboundH, ":9091") sigCh := make(chan os.Signal, 1) signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT) From 2ce54074df552adfbc2ff64e8155a9f104c32cc6 Mon Sep 17 00:00:00 2001 From: Ed Snible Date: Fri, 31 Jul 2026 13:30:10 -0400 Subject: [PATCH 3/3] Minor refactor inspired by review comments Signed-off-by: Ed Snible --- authbridge/authlib/runtimeutil/runtimeutil.go | 57 ++++++++++++------- authbridge/cmd/authbridge-cpex/main.go | 6 +- authbridge/cmd/authbridge-envoy/main.go | 6 +- authbridge/cmd/authbridge-proxy/main.go | 6 +- 4 files changed, 50 insertions(+), 25 deletions(-) diff --git a/authbridge/authlib/runtimeutil/runtimeutil.go b/authbridge/authlib/runtimeutil/runtimeutil.go index 212dc547..fe2d827f 100644 --- a/authbridge/authlib/runtimeutil/runtimeutil.go +++ b/authbridge/authlib/runtimeutil/runtimeutil.go @@ -67,31 +67,44 @@ func StartSignalToggle() { }() } -// StartHealthServer serves liveness (/healthz) and readiness (/readyz) on addr -// in a goroutine. Readiness reports 503 while any inbound or outbound plugin is -// still waiting on a dependency (e.g. a credential file that hasn't landed yet). -func StartHealthServer(inboundH, outboundH *pipeline.Holder, addr string) { +// StartHealthServer binds addr, serves liveness (/healthz) and readiness +// (/readyz) in a goroutine, and returns the server for graceful shutdown. +// Readiness reports 503 while any inbound or outbound plugin is still waiting on +// a dependency (e.g. a credential file that hasn't landed yet). A bind failure +// is returned so the caller can decide how to handle it; a serve-time failure +// after bind is logged. +func StartHealthServer(inboundH, outboundH *pipeline.Holder, addr string) (*http.Server, error) { + mux := http.NewServeMux() + mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + }) + mux.HandleFunc("/readyz", func(w http.ResponseWriter, r *http.Request) { + if name := inboundH.NotReadyPlugin(); name != "" { + http.Error(w, "inbound plugin not ready: "+name, http.StatusServiceUnavailable) + return + } + if name := outboundH.NotReadyPlugin(); name != "" { + http.Error(w, "outbound plugin not ready: "+name, http.StatusServiceUnavailable) + return + } + w.WriteHeader(http.StatusOK) + }) + srv := &http.Server{ + Addr: addr, + Handler: mux, + ReadHeaderTimeout: 10 * time.Second, + } + listener, err := net.Listen("tcp", addr) + if err != nil { + return nil, err + } go func() { - mux := http.NewServeMux() - mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { - w.WriteHeader(http.StatusOK) - }) - mux.HandleFunc("/readyz", func(w http.ResponseWriter, r *http.Request) { - if name := inboundH.NotReadyPlugin(); name != "" { - http.Error(w, "inbound plugin not ready: "+name, http.StatusServiceUnavailable) - return - } - if name := outboundH.NotReadyPlugin(); name != "" { - http.Error(w, "outbound plugin not ready: "+name, http.StatusServiceUnavailable) - return - } - w.WriteHeader(http.StatusOK) - }) - slog.Info("health server listening", "addr", addr) - if err := http.ListenAndServe(addr, mux); err != nil { - slog.Warn("health server failed", "error", err) + slog.Info("health server listening", "addr", listener.Addr().String()) + if err := srv.Serve(listener); err != nil && err != http.ErrServerClosed { + slog.Error("health server failed", "error", err) } }() + return srv, nil } // StartStatServer binds addr for the stats/config-inspection server, serves it diff --git a/authbridge/cmd/authbridge-cpex/main.go b/authbridge/cmd/authbridge-cpex/main.go index 10a2b7f8..050889df 100644 --- a/authbridge/cmd/authbridge-cpex/main.go +++ b/authbridge/cmd/authbridge-cpex/main.go @@ -245,7 +245,10 @@ func main() { slog.Info("authbridge-cpex starting", "mode", cfg.Mode, "logLevel", runtimeutil.LogLevel().String()) - runtimeutil.StartHealthServer(inboundH, outboundH, ":9091") + healthSrv, healthErr := runtimeutil.StartHealthServer(inboundH, outboundH, ":9091") + if healthErr != nil { + log.Fatalf("health server listen: %v", healthErr) + } sigCh := make(chan os.Signal, 1) signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT) @@ -259,6 +262,7 @@ func main() { srv.Shutdown(shutdownCtx) } statSrv.Shutdown(shutdownCtx) + healthSrv.Shutdown(shutdownCtx) if sessionAPISrv != nil { sessionAPISrv.Shutdown(shutdownCtx) } diff --git a/authbridge/cmd/authbridge-envoy/main.go b/authbridge/cmd/authbridge-envoy/main.go index b1a4161f..1774694a 100644 --- a/authbridge/cmd/authbridge-envoy/main.go +++ b/authbridge/cmd/authbridge-envoy/main.go @@ -231,7 +231,10 @@ func main() { slog.Info("authbridge-envoy starting", "mode", cfg.Mode, "logLevel", runtimeutil.LogLevel().String()) - runtimeutil.StartHealthServer(inboundH, outboundH, ":9091") + healthSrv, healthErr := runtimeutil.StartHealthServer(inboundH, outboundH, ":9091") + if healthErr != nil { + log.Fatalf("health server listen: %v", healthErr) + } sigCh := make(chan os.Signal, 1) signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT) @@ -249,6 +252,7 @@ func main() { srv.GracefulStop() } statSrv.Shutdown(shutdownCtx) + healthSrv.Shutdown(shutdownCtx) if sessionAPISrv != nil { sessionAPISrv.Shutdown(shutdownCtx) } diff --git a/authbridge/cmd/authbridge-proxy/main.go b/authbridge/cmd/authbridge-proxy/main.go index bbc9b3e2..131ca85b 100644 --- a/authbridge/cmd/authbridge-proxy/main.go +++ b/authbridge/cmd/authbridge-proxy/main.go @@ -467,7 +467,10 @@ func main() { slog.Info("authbridge-proxy starting", "version", version, "mode", cfg.Mode, "logLevel", runtimeutil.LogLevel().String()) - runtimeutil.StartHealthServer(inboundH, outboundH, ":9091") + healthSrv, healthErr := runtimeutil.StartHealthServer(inboundH, outboundH, ":9091") + if healthErr != nil { + log.Fatalf("health server listen: %v", healthErr) + } sigCh := make(chan os.Signal, 1) signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT) @@ -484,6 +487,7 @@ func main() { _ = transparentLn.Close() } statSrv.Shutdown(shutdownCtx) + healthSrv.Shutdown(shutdownCtx) if sessionAPISrv != nil { sessionAPISrv.Shutdown(shutdownCtx) }