skipper: add server connection keepalive limits

config/config.go

@@ -243,6 +243,8 @@ type Config struct {
 	ReadHeaderTimeoutServer      time.Duration `yaml:"read-header-timeout-server"`
 	WriteTimeoutServer           time.Duration `yaml:"write-timeout-server"`
 	IdleTimeoutServer            time.Duration `yaml:"idle-timeout-server"`
+	KeepaliveServer              time.Duration `yaml:"keepalive-server"`
+	KeepaliveRequestsServer      int           `yaml:"keepalive-requests-server"`
 	MaxHeaderBytes               int           `yaml:"max-header-bytes"`
 	EnableConnMetricsServer      bool          `yaml:"enable-connection-metrics"`
 	TimeoutBackend               time.Duration `yaml:"timeout-backend"`
@@ -544,6 +546,8 @@ func NewConfig() *Config {
 	flag.DurationVar(&cfg.ReadHeaderTimeoutServer, "read-header-timeout-server", 60*time.Second, "set ReadHeaderTimeout for http server connections")
 	flag.DurationVar(&cfg.WriteTimeoutServer, "write-timeout-server", 60*time.Second, "set WriteTimeout for http server connections")
 	flag.DurationVar(&cfg.IdleTimeoutServer, "idle-timeout-server", 60*time.Second, "set IdleTimeout for http server connections")
+	flag.DurationVar(&cfg.KeepaliveServer, "keepalive-server", 0*time.Second, "sets maximum age for http server connections. The connection is closed after it existed for this duration. Default is 0 for unlimited.")
+	flag.IntVar(&cfg.KeepaliveRequestsServer, "keepalive-requests-server", 0, "sets maximum number of requests for http server connections. The connection is closed after serving this number of requests. Default is 0 for unlimited.")
 	flag.IntVar(&cfg.MaxHeaderBytes, "max-header-bytes", http.DefaultMaxHeaderBytes, "set MaxHeaderBytes for http server connections")
 	flag.BoolVar(&cfg.EnableConnMetricsServer, "enable-connection-metrics", false, "enables connection metrics for http server connections")
 	flag.DurationVar(&cfg.TimeoutBackend, "timeout-backend", 60*time.Second, "sets the TCP client connection timeout for backend connections")
@@ -879,6 +883,8 @@ func (c *Config) ToOptions() skipper.Options {
 		ReadHeaderTimeoutServer:      c.ReadHeaderTimeoutServer,
 		WriteTimeoutServer:           c.WriteTimeoutServer,
 		IdleTimeoutServer:            c.IdleTimeoutServer,
+		KeepaliveServer:              c.KeepaliveServer,
+		KeepaliveRequestsServer:      c.KeepaliveRequestsServer,
 		MaxHeaderBytes:               c.MaxHeaderBytes,
 		EnableConnMetricsServer:      c.EnableConnMetricsServer,
 		TimeoutBackend:               c.TimeoutBackend,

docs/operation/operation.md

@@ -124,6 +124,16 @@ combinations of idle timeouts can lead to a few unexpected HTTP 502.
     -idle-timeout-server duration
         maximum idle connections per backend host (default 1m0s)
 
+This configures maximum number of requests served by server connections:
+
+    -keepalive-requests-server int
+        sets maximum number of requests for http server connections. The connection is closed after serving this number of requests. Default is 0 for unlimited.
+
+This configures maximum age for server connections:
+
+    -keepalive-server duration
+        sets maximum age for http server connections. The connection is closed after it existed for this duration. Default is 0 for unlimited.
+
 This will set MaxHeaderBytes in
 [http.Server](https://golang.org/pkg/net/http/#Server) to limit the
 size of the http header from your clients.

net/connmanager.go

@@ -0,0 +1,95 @@
+package net
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"net/http"
+	"time"
+
+	"github.com/zalando/skipper/metrics"
+)
+
+// ConnManager tracks creation of HTTP server connections and
+// closes connections when their age or number of requests served reaches configured limits.
+// Use [ConnManager.Configure] method to setup ConnManager for an [http.Server].
+type ConnManager struct {
+	// Metrics is an optional metrics registry to count connection events.
+	Metrics metrics.Metrics
+
+	// Keepalive is the duration after which server connection is closed.
+	Keepalive time.Duration
+
+	// KeepaliveRequests is the number of requests after which server connection is closed.
+	KeepaliveRequests int
+
+	handler http.Handler
+}
+
+type connState struct {
+	expiresAt time.Time
+	requests  int
+}
+
+type contextKey struct{}
+
+var connection contextKey
+
+func (cm *ConnManager) Configure(server *http.Server) {
+	cm.handler = server.Handler
+	server.Handler = http.HandlerFunc(cm.serveHTTP)
+
+	if cc := server.ConnContext; cc != nil {
+		server.ConnContext = func(ctx context.Context, c net.Conn) context.Context {
+			ctx = cc(ctx, c)
+			return cm.connContext(ctx, c)
+		}
+	} else {
+		server.ConnContext = cm.connContext
+	}
+
+	if cs := server.ConnState; cs != nil {
+		server.ConnState = func(c net.Conn, state http.ConnState) {
+			cs(c, state)
+			cm.connState(c, state)
+		}
+	} else {
+		server.ConnState = cm.connState
+	}
+}
+
+func (cm *ConnManager) serveHTTP(w http.ResponseWriter, r *http.Request) {
+	state, _ := r.Context().Value(connection).(*connState)
+	state.requests++
+
+	if cm.KeepaliveRequests > 0 && state.requests >= cm.KeepaliveRequests {
+		w.Header().Set("Connection", "close")
+
+		cm.count("lb-conn-closed.keepalive-requests")
+	}
+
+	if cm.Keepalive > 0 && time.Now().After(state.expiresAt) {
+		w.Header().Set("Connection", "close")
+
+		cm.count("lb-conn-closed.keepalive")
+	}
+
+	cm.handler.ServeHTTP(w, r)
+}
+
+func (cm *ConnManager) connContext(ctx context.Context, _ net.Conn) context.Context {
+	state := &connState{
+		expiresAt: time.Now().Add(cm.Keepalive),
+	}
+	return context.WithValue(ctx, connection, state)
+}
+
+func (cm *ConnManager) connState(_ net.Conn, state http.ConnState) {
+	cm.count(fmt.Sprintf("lb-conn-%s", state))
+}
+
+func (cm *ConnManager) count(name string) {
+	if cm.Metrics != nil {
+		cm.Metrics.IncCounter(name)
+	}
+}

net/connmanager_test.go

@@ -0,0 +1,128 @@
+package net_test
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/zalando/skipper/metrics/metricstest"
+	snet "github.com/zalando/skipper/net"
+)
+
+func TestConnManager(t *testing.T) {
+	const (
+		keepaliveRequests = 3
+		keepalive         = 100 * time.Millisecond
+
+		testRequests = keepaliveRequests * 5
+	)
+	t.Run("does not close connection without limits", func(t *testing.T) {
+		ts := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			w.WriteHeader(http.StatusOK)
+		}))
+		m := &metricstest.MockMetrics{}
+		cm := &snet.ConnManager{
+			Metrics: m,
+		}
+		cm.Configure(ts.Config)
+
+		ts.Start()
+		defer ts.Close()
+
+		for i := 0; i < testRequests; i++ {
+			resp, err := ts.Client().Get(ts.URL)
+			require.NoError(t, err)
+			assert.Equal(t, http.StatusOK, resp.StatusCode)
+			assert.False(t, resp.Close)
+		}
+
+		time.Sleep(100 * time.Millisecond) // wait for connection state update
+
+		m.WithCounters(func(counters map[string]int64) {
+			assert.Equal(t, int64(1), counters["lb-conn-new"])
+			assert.Equal(t, int64(testRequests), counters["lb-conn-active"])
+			assert.Equal(t, int64(testRequests), counters["lb-conn-idle"])
+			assert.Equal(t, int64(0), counters["lb-conn-closed"])
+		})
+	})
+	t.Run("closes connection after keepalive requests", func(t *testing.T) {
+		const keepaliveRequests = 3
+
+		ts := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			w.WriteHeader(http.StatusOK)
+		}))
+		m := &metricstest.MockMetrics{}
+		cm := &snet.ConnManager{
+			Metrics:           m,
+			KeepaliveRequests: keepaliveRequests,
+		}
+		cm.Configure(ts.Config)
+
+		ts.Start()
+		defer ts.Close()
+
+		for i := 1; i < testRequests; i++ {
+			resp, err := ts.Client().Get(ts.URL)
+			require.NoError(t, err)
+			assert.Equal(t, http.StatusOK, resp.StatusCode)
+
+			if i%keepaliveRequests == 0 {
+				assert.True(t, resp.Close)
+			} else {
+				assert.False(t, resp.Close)
+			}
+		}
+
+		time.Sleep(100 * time.Millisecond) // wait for connection state update
+
+		m.WithCounters(func(counters map[string]int64) {
+			rounds := int64(testRequests / keepaliveRequests)
+
+			assert.Equal(t, rounds, counters["lb-conn-new"])
+			assert.Equal(t, rounds-1, counters["lb-conn-closed"])
+			assert.Equal(t, rounds-1, counters["lb-conn-closed.keepalive-requests"])
+		})
+	})
+
+	t.Run("closes connection after keepalive timeout", func(t *testing.T) {
+		const keepalive = 100 * time.Millisecond
+
+		ts := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			w.WriteHeader(http.StatusOK)
+		}))
+		m := &metricstest.MockMetrics{}
+		cm := &snet.ConnManager{
+			Metrics:   m,
+			Keepalive: keepalive,
+		}
+		cm.Configure(ts.Config)
+
+		ts.Start()
+		defer ts.Close()
+
+		for i := 0; i < testRequests; i++ {
+			resp, err := ts.Client().Get(ts.URL)
+			require.NoError(t, err)
+			assert.Equal(t, http.StatusOK, resp.StatusCode)
+			assert.False(t, resp.Close)
+		}
+
+		time.Sleep(2 * keepalive)
+
+		resp, err := ts.Client().Get(ts.URL)
+		require.NoError(t, err)
+		assert.Equal(t, http.StatusOK, resp.StatusCode)
+		assert.True(t, resp.Close)
+
+		time.Sleep(100 * time.Millisecond) // wait for connection state update
+
+		m.WithCounters(func(counters map[string]int64) {
+			assert.Equal(t, int64(1), counters["lb-conn-new"])
+			assert.Equal(t, int64(1), counters["lb-conn-closed"])
+			assert.Equal(t, int64(1), counters["lb-conn-closed.keepalive"])
+		})
+	})
+}

skipper.go

@@ -360,6 +360,14 @@ type Options struct {
 	// Defines IdleTimeout for server http connections.
 	IdleTimeoutServer time.Duration
 
+	// KeepaliveServer configures maximum age for server http connections.
+	// The connection is closed after it existed for this duration.
+	KeepaliveServer time.Duration
+
+	// KeepaliveRequestsServer configures maximum number of requests for server http connections.
+	// The connection is closed after serving this number of requests.
+	KeepaliveRequestsServer int
+
 	// Defines MaxHeaderBytes for server http connections.
 	MaxHeaderBytes int
 
@@ -1334,12 +1342,17 @@ func listenAndServeQuit(
 		ErrorLog:          newServerErrorLog(),
 	}
 
+	cm := &skpnet.ConnManager{
+		Keepalive:         o.KeepaliveServer,
+		KeepaliveRequests: o.KeepaliveRequestsServer,
+	}
+
 	if o.EnableConnMetricsServer {
-		srv.ConnState = func(conn net.Conn, state http.ConnState) {
-			mtr.IncCounter(fmt.Sprintf("lb-conn-%s", state))
-		}
+		cm.Metrics = mtr
 	}
 
+	cm.Configure(srv)
+
 	log.Infof("Listen on %v", address)
 
 	l, err := listen(o, address, mtr)