Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion server.go
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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
}

Expand Down
114 changes: 114 additions & 0 deletions server_port_test.go
Original file line number Diff line number Diff line change
@@ -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")
}
12 changes: 11 additions & 1 deletion service/http.go
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down
53 changes: 53 additions & 0 deletions service/http_test.go
Original file line number Diff line number Diff line change
@@ -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()),
Expand Down
Loading