diff --git a/server.go b/server.go index 9b3920c8..e94779bf 100644 --- a/server.go +++ b/server.go @@ -98,7 +98,7 @@ func NewServer(opts ...Option) *Server { inst.ctx, inst.cancel = context.WithCancel(inst.ctx) inst.g, inst.gCtx = errgroup.WithContext(inst.ctx) - inst.gracefulCtx, inst.gracefulCancel = signal.NotifyContext(inst.ctx, inst.shutdownSignals...) + inst.gracefulCtx, inst.gracefulCancel = signal.NotifyContext(inst.gCtx, inst.shutdownSignals...) // gracefully shutdown inst.g.Go(func() error { @@ -407,6 +407,7 @@ func (s *Server) startService(services ...Service) { log.WithError(s.l, err).Debug("server has closed") } else if err != nil { log.WithError(s.l, err).Error("failed to start service") + return err } diff --git a/server_port_test.go b/server_port_test.go new file mode 100644 index 00000000..c16142ee --- /dev/null +++ b/server_port_test.go @@ -0,0 +1,114 @@ +package keel_test + +import ( + "net" + "net/http" + "testing" + "time" + + "github.com/foomo/keel" + "github.com/foomo/keel/service" + "github.com/stretchr/testify/require" + "go.uber.org/zap/zaptest" +) + +// TestServerPortInUse covers a pod restarting while its port is still held, e.g. +// after an OOMKill. The server must stop instead of staying alive with a service +// that never got a listener, so the container is restarted rather than reporting +// healthy while serving nothing. +func TestServerPortInUse(t *testing.T) { + t.Parallel() + + l := zaptest.NewLogger(t) + + var lc net.ListenConfig + + occupied, err := lc.Listen(t.Context(), "tcp", "localhost:0") + require.NoError(t, err) + + t.Cleanup(func() { _ = occupied.Close() }) + + svr := keel.NewServer( + keel.WithContext(t.Context()), + keel.WithLogger(l), + keel.WithGracefulPeriod(3*time.Second), + ) + + blocked := service.NewHTTP(l, "blocked", occupied.Addr().String(), http.NewServeMux()) + svr.AddService(blocked) + + done := make(chan struct{}) + + go func() { + defer close(done) + + svr.Run() + }() + + select { + case <-done: + case <-time.After(15 * time.Second): + t.Fatal("server kept running although a service could not bind its port") + } + + require.Error(t, blocked.Healthz(), "service must not report healthy without a listener") + require.Error(t, svr.Healthz(), "server must not report healthy after it stopped") +} + +// TestServerPortInUseStopsOtherServices asserts the whole server goes down, not +// just the service that failed, so a partially serving pod cannot pass its probes. +func TestServerPortInUseStopsOtherServices(t *testing.T) { + t.Parallel() + + l := zaptest.NewLogger(t) + + var lc net.ListenConfig + + occupied, err := lc.Listen(t.Context(), "tcp", "localhost:0") + require.NoError(t, err) + + t.Cleanup(func() { _ = occupied.Close() }) + + free, err := lc.Listen(t.Context(), "tcp", "localhost:0") + require.NoError(t, err) + + freeAddr := free.Addr().String() + require.NoError(t, free.Close()) + + svr := keel.NewServer( + keel.WithContext(t.Context()), + keel.WithLogger(l), + keel.WithGracefulPeriod(3*time.Second), + ) + + healthy := service.NewHTTP(l, "healthy", freeAddr, http.NewServeMux()) + blocked := service.NewHTTP(l, "blocked", occupied.Addr().String(), http.NewServeMux()) + svr.AddServices(healthy, blocked) + + done := make(chan struct{}) + + go func() { + defer close(done) + + svr.Run() + }() + + select { + case <-done: + case <-time.After(15 * time.Second): + t.Fatal("server kept running although a service could not bind its port") + } + + dialer := net.Dialer{Timeout: time.Second} + + require.Eventually(t, func() bool { + conn, err := dialer.DialContext(t.Context(), "tcp", freeAddr) + if err != nil { + return true + } + + _ = conn.Close() + + return false + }, 5*time.Second, 50*time.Millisecond, "the remaining service must not keep serving") +} diff --git a/service/http.go b/service/http.go index 060e1b4c..2937e984 100644 --- a/service/http.go +++ b/service/http.go @@ -91,9 +91,19 @@ func (s *HTTP) Start(ctx context.Context) error { s.server.RegisterOnShutdown(func() { s.running.Store(false) }) + + // Bind before reporting healthy: an occupied port must fail the service + // rather than leave it marked running while nothing is listening. + var lc net.ListenConfig + + ln, err := lc.Listen(ctx, "tcp", s.server.Addr) + if err != nil { + return errors.Wrapf(err, "failed to listen to addr: %s", s.server.Addr) + } + s.running.Store(true) - if err := s.server.ListenAndServe(); errors.Is(err, http.ErrServerClosed) { + if err := s.server.Serve(ln); errors.Is(err, http.ErrServerClosed) { return nil } else if err != nil { return errors.Wrap(err, "failed to start service") diff --git a/service/http_test.go b/service/http_test.go index e20339fe..44fd37a7 100644 --- a/service/http_test.go +++ b/service/http_test.go @@ -1,14 +1,67 @@ package service_test import ( + "net" "net/http" + "testing" "time" "github.com/foomo/keel" "github.com/foomo/keel/service" + "github.com/stretchr/testify/require" "go.uber.org/zap" + "go.uber.org/zap/zaptest" ) +// TestHTTPStartPortInUse covers the case of a restarted pod whose predecessor +// still holds the port: Start must fail and the service must never report +// healthy, or the pod would come up without ever accepting a request. +func TestHTTPStartPortInUse(t *testing.T) { + t.Parallel() + + l := zaptest.NewLogger(t) + + var lc net.ListenConfig + + occupied, err := lc.Listen(t.Context(), "tcp", "localhost:0") + require.NoError(t, err) + + t.Cleanup(func() { _ = occupied.Close() }) + + s := service.NewHTTP(l, "blocked", occupied.Addr().String(), http.NewServeMux()) + + require.Error(t, s.Healthz(), "service must not be healthy before it started") + require.Error(t, s.Start(t.Context()), "start must fail while the port is occupied") + require.ErrorIs(t, s.Healthz(), service.ErrServiceNotRunning, + "service must not report healthy when it never got a listener") +} + +// TestHTTPStartFreePort guards the fix against regressing the happy path. +func TestHTTPStartFreePort(t *testing.T) { + t.Parallel() + + l := zaptest.NewLogger(t) + + var lc net.ListenConfig + + free, err := lc.Listen(t.Context(), "tcp", "localhost:0") + require.NoError(t, err) + + addr := free.Addr().String() + require.NoError(t, free.Close()) + + s := service.NewHTTP(l, "free", addr, http.NewServeMux()) + + done := make(chan error, 1) + go func() { done <- s.Start(t.Context()) }() + + require.Eventually(t, func() bool { return s.Healthz() == nil }, 5*time.Second, 10*time.Millisecond, + "service must report healthy once it is listening") + + require.NoError(t, s.Close(t.Context())) + require.NoError(t, <-done) +} + func _ExampleNewHTTP() { svr := keel.NewServer( keel.WithLogger(zap.NewExample()),