Repository: traefik/traefik (language: go)

Issue:
Improve open connections management
### Welcome!

- [X] Yes, I've searched similar issues on [GitHub](https://github.com/traefik/traefik/issues) and didn't find any.
- [X] Yes, I've searched similar issues on the [Traefik community forum](https://community.traefik.io) and didn't find any.

### What did you expect to see?

## Description

Currently, in Traefik, HTTP connections stay open until clients decide to close them. This can lead to more open connections and higher memory usage.

Moreover, when you use Traefik with an upstream Load Balancer, this connection management method avoids an efficient traffic distribution when you scale up your Traefik instance. Rather than distributing requests to the newest instances, ensuring a more equitable traffic spread, the Load Balancer will stick to the same instance.

To tackle this, we're thinking of introducing two new options for the entrypoints' transport:

- The first option limits the number of requests each client is allowed to make within one connection.
- The second option limits the time a client is allowed to keep reusing the same connection for its requests.

Of course, you can use both options together and choose one or the other.

In practice, we believe these options will help spread the incoming traffic to new traffic instances as we scale within a more predictable timeframe.

We'd love to hear your thoughts on this proposal and how it might make Traefik even better.

## Timeline

Conversation is time-boxed to one month.

## Additional Notes

### keepalivemaxrequests

```
--entrypoints.<name>.transport.keepalivemaxrequests # KeepAlive max requests.
```

Whenever a client sends a request within an open connection if this request number is greater than the `keepalivemaxrequests` threshold, Traefik will ask for a connection close (header `Connection:close`).

### keepalivemaxtime

```
--entrypoints.<name>.transport.keepalivemaxtime # KeepAlive max time.
```

Whenever a client sends a request within an open connection if the elapsed time between the first request and the current request is greater than the `keepalivemaxtime` threshold, Traefik will ask for a connection close (header Connection:close`).


Discussion / hints:

Fix patch:
diff --git a/docs/content/reference/static-configuration/cli-ref.md b/docs/content/reference/static-configuration/cli-ref.md
index a0415e1b42..03409a1c38 100644
--- a/docs/content/reference/static-configuration/cli-ref.md
+++ b/docs/content/reference/static-configuration/cli-ref.md
@@ -171,6 +171,12 @@ Trust all. (Default: ```false```)
 `--entrypoints.<name>.proxyprotocol.trustedips`:  
 Trust only selected IPs.
 
+`--entrypoints.<name>.transport.keepalivemaxrequests`:  
+Maximum number of requests before closing a keep-alive connection. (Default: ```0```)
+
+`--entrypoints.<name>.transport.keepalivemaxtime`:  
+Maximum duration before closing a keep-alive connection. (Default: ```0```)
+
 `--entrypoints.<name>.transport.lifecycle.gracetimeout`:  
 Duration to give active requests a chance to finish before Traefik stops. (Default: ```10```)
 
diff --git a/docs/content/reference/static-configuration/env-ref.md b/docs/content/reference/static-configuration/env-ref.md
index a7b3ace47c..d3de13a66b 100644
--- a/docs/content/reference/static-configuration/env-ref.md
+++ b/docs/content/reference/static-configuration/env-ref.md
@@ -171,6 +171,12 @@ Trust all. (Default: ```false```)
 `TRAEFIK_ENTRYPOINTS_<NAME>_PROXYPROTOCOL_TRUSTEDIPS`:  
 Trust only selected IPs.
 
+`TRAEFIK_ENTRYPOINTS_<NAME>_TRANSPORT_KEEPALIVEMAXREQUESTS`:  
+Maximum number of requests before closing a keep-alive connection. (Default: ```0```)
+
+`TRAEFIK_ENTRYPOINTS_<NAME>_TRANSPORT_KEEPALIVEMAXTIME`:  
+Maximum duration before closing a keep-alive connection. (Default: ```0```)
+
 `TRAEFIK_ENTRYPOINTS_<NAME>_TRANSPORT_LIFECYCLE_GRACETIMEOUT`:  
 Duration to give active requests a chance to finish before Traefik stops. (Default: ```10```)
 
diff --git a/docs/content/reference/static-configuration/file.toml b/docs/content/reference/static-configuration/file.toml
index 42dda1fbf7..dc38a2cea1 100644
--- a/docs/content/reference/static-configuration/file.toml
+++ b/docs/content/reference/static-configuration/file.toml
@@ -15,6 +15,8 @@
   [entryPoints.EntryPoint0]
     address = "foobar"
     [entryPoints.EntryPoint0.transport]
+      keepAliveMaxRequests = 42
+      keepAliveMaxTime = "42s"
       [entryPoints.EntryPoint0.transport.lifeCycle]
         requestAcceptGraceTimeout = "42s"
         graceTimeOut = "42s"
diff --git a/docs/content/reference/static-configuration/file.yaml b/docs/content/reference/static-configuration/file.yaml
index 1497f57e54..6a0cb264b6 100644
--- a/docs/content/reference/static-configuration/file.yaml
+++ b/docs/content/reference/static-configuration/file.yaml
@@ -15,6 +15,8 @@ entryPoints:
   EntryPoint0:
     address: foobar
     transport:
+      keepAliveMaxRequests: 42
+      keepAliveMaxTime: 42s
       lifeCycle:
         requestAcceptGraceTimeout: 42s
         graceTimeOut: 42s
diff --git a/docs/content/routing/entrypoints.md b/docs/content/routing/entrypoints.md
index 305e69cd13..c77c8e4060 100644
--- a/docs/content/routing/entrypoints.md
+++ b/docs/content/routing/entrypoints.md
@@ -589,17 +589,77 @@ Controls the behavior of Traefik during the shutdown phase.
     --entryPoints.name.transport.lifeCycle.graceTimeOut=42
     ```
 
+#### `keepAliveMaxRequests`
+
+_Optional, Default=0_
+
+The maximum number of requests Traefik can handle before sending a `Connection: Close` header to the client (for HTTP2, Traefik sends a GOAWAY). Zero means no limit.
+
+    ```yaml tab="File (YAML)"
+    ## Static configuration
+    entryPoints:
+      name:
+        address: ":8888"
+        transport:
+          keepAliveMaxRequests: 42
+    ```
+
+    ```toml tab="File (TOML)"
+    ## Static configuration
+    [entryPoints]
+      [entryPoints.name]
+        address = ":8888"
+        [entryPoints.name.transport]
+          keepAliveMaxRequests = 42
+    ```
+
+    ```bash tab="CLI"
+    ## Static configuration
+    --entryPoints.name.address=:8888
+    --entryPoints.name.transport.keepAliveRequests=42
+    ```
+
+#### `keepAliveMaxTime`
+
+_Optional, Default=0s_
+
+The maximum duration Traefik can handle requests before sending a `Connection: Close` header to the client (for HTTP2, Traefik sends a GOAWAY). Zero means no limit.
+
+    ```yaml tab="File (YAML)"
+    ## Static configuration
+    entryPoints:
+      name:
+        address: ":8888"
+        transport:
+          keepAliveMaxTime: 42s
+    ```
+
+    ```toml tab="File (TOML)"
+    ## Static configuration
+    [entryPoints]
+      [entryPoints.name]
+        address = ":8888"
+        [entryPoints.name.transport]
+          keepAliveMaxTime = 42s
+    ```
+
+    ```bash tab="CLI"
+    ## Static configuration
+    --entryPoints.name.address=:8888
+    --entryPoints.name.transport.keepAliveTime=42s
+    ```
+
 ### ProxyProtocol
 
-Traefik supports [ProxyProtocol](https://www.haproxy.org/download/2.0/doc/proxy-protocol.txt) version 1 and 2.
+Traefik supports [PROXY protocol](https://www.haproxy.org/download/2.0/doc/proxy-protocol.txt) version 1 and 2.
 
-If Proxy Protocol header parsing is enabled for the entry point, this entry point can accept connections with or without Proxy Protocol headers.
+If PROXY protocol header parsing is enabled for the entry point, this entry point can accept connections with or without PROXY protocol headers.
 
-If the Proxy Protocol header is passed, then the version is determined automatically.
+If the PROXY protocol header is passed, then the version is determined automatically.
 
 ??? info "`proxyProtocol.trustedIPs`"
 
-    Enabling Proxy Protocol with Trusted IPs.
+    Enabling PROXY protocol with Trusted IPs.
 
     ```yaml tab="File (YAML)"
     ## Static configuration
@@ -662,7 +722,7 @@ If the Proxy Protocol header is passed, then the version is determined automatic
 
 !!! warning "Queuing Traefik behind Another Load Balancer"
 
-    When queuing Traefik behind another load-balancer, make sure to configure Proxy Protocol on both sides.
+    When queuing Traefik behind another load-balancer, make sure to configure PROXY protocol on both sides.
     Not doing so could introduce a security risk in your system (enabling request forgery).
 
 ## HTTP Options
diff --git a/pkg/config/static/entrypoints.go b/pkg/config/static/entrypoints.go
index 68ac9285e5..b2243aa3f8 100644
--- a/pkg/config/static/entrypoints.go
+++ b/pkg/config/static/entrypoints.go
@@ -122,8 +122,10 @@ type EntryPoints map[string]*EntryPoint
 
 // EntryPointsTransport configures communication between clients and Traefik.
 type EntryPointsTransport struct {
-	LifeCycle          *LifeCycle          `description:"Timeouts influencing the server life cycle." json:"lifeCycle,omitempty" toml:"lifeCycle,omitempty" yaml:"lifeCycle,omitempty" export:"true"`
-	RespondingTimeouts *RespondingTimeouts `description:"Timeouts for incoming requests to the Traefik instance." json:"respondingTimeouts,omitempty" toml:"respondingTimeouts,omitempty" yaml:"respondingTimeouts,omitempty" export:"true"`
+	LifeCycle            *LifeCycle          `description:"Timeouts influencing the server life cycle." json:"lifeCycle,omitempty" toml:"lifeCycle,omitempty" yaml:"lifeCycle,omitempty" export:"true"`
+	RespondingTimeouts   *RespondingTimeouts `description:"Timeouts for incoming requests to the Traefik instance." json:"respondingTimeouts,omitempty" toml:"respondingTimeouts,omitempty" yaml:"respondingTimeouts,omitempty" export:"true"`
+	KeepAliveMaxTime     ptypes.Duration     `description:"Maximum duration before closing a keep-alive connection." json:"keepAliveMaxTime,omitempty" toml:"keepAliveMaxTime,omitempty" yaml:"keepAliveMaxTime,omitempty" export:"true"`
+	KeepAliveMaxRequests int                 `description:"Maximum number of requests before closing a keep-alive connection." json:"keepAliveMaxRequests,omitempty" toml:"keepAliveMaxRequests,omitempty" yaml:"keepAliveMaxRequests,omitempty" export:"true"`
 }
 
 // SetDefaults sets the default values.
diff --git a/pkg/server/keep_alive_middleware.go b/pkg/server/keep_alive_middleware.go
new file mode 100644
index 0000000000..73b2159b10
--- /dev/null
+++ b/pkg/server/keep_alive_middleware.go
@@ -0,0 +1,29 @@
+package server
+
+import (
+	"net/http"
+	"time"
+
+	ptypes "github.com/traefik/paerser/types"
+	"github.com/traefik/traefik/v2/pkg/log"
+)
+
+func newKeepAliveMiddleware(next http.Handler, maxRequests int, maxTime ptypes.Duration) http.Handler {
+	return http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
+		state, ok := req.Context().Value(connStateKey).(*connState)
+		if ok {
+			state.HTTPRequestCount++
+			if maxRequests > 0 && state.HTTPRequestCount >= maxRequests {
+				log.WithoutContext().Debug("Close because of too many requests")
+				state.KeepAliveState = "Close because of too many requests"
+				rw.Header().Set("Connection", "close")
+			}
+			if maxTime > 0 && time.Now().After(state.Start.Add(time.Duration(maxTime))) {
+				log.WithoutContext().Debug("Close because of too long connection")
+				state.KeepAliveState = "Close because of too long connection"
+				rw.Header().Set("Connection", "close")
+			}
+		}
+		next.ServeHTTP(rw, req)
+	})
+}
diff --git a/pkg/server/server_entrypoint_tcp.go b/pkg/server/server_entrypoint_tcp.go
index b33c955eef..e3e22998e4 100644
--- a/pkg/server/server_entrypoint_tcp.go
+++ b/pkg/server/server_entrypoint_tcp.go
@@ -3,6 +3,7 @@ package server
 import (
 	"context"
 	"errors"
+	"expvar"
 	"fmt"
 	stdlog "log"
 	"net"
@@ -34,6 +35,25 @@ import (
 
 var httpServerLogger = stdlog.New(log.WithoutContext().WriterLevel(logrus.DebugLevel), "", 0)
 
+type key string
+
+const (
+	connStateKey       key    = "connState"
+	debugConnectionEnv string = "DEBUG_CONNECTION"
+)
+
+var (
+	clientConnectionStates   = map[string]*connState{}
+	clientConnectionStatesMu = sync.RWMutex{}
+)
+
+type connState struct {
+	State            string
+	KeepAliveState   string
+	Start            time.Time
+	HTTPRequestCount int
+}
+
 type httpForwarder struct {
 	net.Listener
 	connChan chan net.Conn
@@ -68,6 +88,12 @@ type TCPEntryPoints map[string]*TCPEntryPoint
 
 // NewTCPEntryPoints creates a new TCPEntryPoints.
 func NewTCPEntryPoints(entryPointsConfig static.EntryPoints, hostResolverConfig *types.HostResolverConfig) (TCPEntryPoints, error) {
+	if os.Getenv(debugConnectionEnv) != "" {
+		expvar.Publish("clientConnectionStates", expvar.Func(func() any {
+			return clientConnectionStates
+		}))
+	}
+
 	serverEntryPointsTCP := make(TCPEntryPoints)
 	for entryPointName, config := range entryPointsConfig {
 		protocol, err := config.GetProtocol()
@@ -548,6 +574,11 @@ func createHTTPServer(ctx context.Context, ln net.Listener, configuration *stati
 		})
 	}
 
+	debugConnection := os.Getenv(debugConnectionEnv) != ""
+	if debugConnection || (configuration.Transport != nil && (configuration.Transport.KeepAliveMaxTime > 0 || configuration.Transport.KeepAliveMaxRequests > 0)) {
+		handler = newKeepAliveMiddleware(handler, configuration.Transport.KeepAliveMaxRequests, configuration.Transport.KeepAliveMaxTime)
+	}
+
 	serverHTTP := &http.Server{
 		Handler:      handler,
 		ErrorLog:     httpServerLogger,
@@ -555,6 +586,27 @@ func createHTTPServer(ctx context.Context, ln net.Listener, configuration *stati
 		WriteTimeout: time.Duration(configuration.Transport.RespondingTimeouts.WriteTimeout),
 		IdleTimeout:  time.Duration(configuration.Transport.RespondingTimeouts.IdleTimeout),
 	}
+	if debugConnection || (configuration.Transport != nil && (configuration.Transport.KeepAliveMaxTime > 0 || configuration.Transport.KeepAliveMaxRequests > 0)) {
+		serverHTTP.ConnContext = func(ctx context.Context, c net.Conn) context.Context {
+			cState := &connState{Start: time.Now()}
+			if debugConnection {
+				clientConnectionStatesMu.Lock()
+				clientConnectionStates[getConnKey(c)] = cState
+				clientConnectionStatesMu.Unlock()
+			}
+			return context.WithValue(ctx, connStateKey, cState)
+		}
+
+		if debugConnection {
+			serverHTTP.ConnState = func(c net.Conn, state http.ConnState) {
+				clientConnectionStatesMu.Lock()
+				if clientConnectionStates[getConnKey(c)] != nil {
+					clientConnectionStates[getConnKey(c)].State = state.String()
+				}
+				clientConnectionStatesMu.Unlock()
+			}
+		}
+	}
 
 	// ConfigureServer configures HTTP/2 with the MaxConcurrentStreams option for the given server.
 	// Also keeping behavior the same as
@@ -584,6 +636,10 @@ func createHTTPServer(ctx context.Context, ln net.Listener, configuration *stati
 	}, nil
 }
 
+func getConnKey(conn net.Conn) string {
+	return fmt.Sprintf("%s => %s", conn.RemoteAddr(), conn.LocalAddr())
+}
+
 func newTrackedConnection(conn tcp.WriteCloser, tracker *connectionTracker) *trackedConnection {
 	tracker.AddConnection(conn)
 	return &trackedConnection{


Test patch:
diff --git a/pkg/server/server_entrypoint_tcp_test.go b/pkg/server/server_entrypoint_tcp_test.go
index 342d7d9faf..9d66c44e07 100644
--- a/pkg/server/server_entrypoint_tcp_test.go
+++ b/pkg/server/server_entrypoint_tcp_test.go
@@ -230,3 +230,91 @@ func TestReadTimeoutWithFirstByte(t *testing.T) {
 		t.Error("Timeout while read")
 	}
 }
+
+func TestKeepAliveMaxRequests(t *testing.T) {
+	epConfig := &static.EntryPointsTransport{}
+	epConfig.SetDefaults()
+	epConfig.KeepAliveMaxRequests = 3
+
+	entryPoint, err := NewTCPEntryPoint(context.Background(), &static.EntryPoint{
+		Address:          ":0",
+		Transport:        epConfig,
+		ForwardedHeaders: &static.ForwardedHeaders{},
+		HTTP2:            &static.HTTP2Config{},
+	}, nil)
+	require.NoError(t, err)
+
+	router := &tcprouter.Router{}
+	router.SetHTTPHandler(http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
+		rw.WriteHeader(http.StatusOK)
+	}))
+
+	conn, err := startEntrypoint(entryPoint, router)
+	require.NoError(t, err)
+
+	http.DefaultClient.Transport = &http.Transport{
+		DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
+			return conn, nil
+		},
+	}
+
+	resp, err := http.Get("http://" + entryPoint.listener.Addr().String())
+	require.NoError(t, err)
+	require.False(t, resp.Close)
+	err = resp.Body.Close()
+	require.NoError(t, err)
+
+	resp, err = http.Get("http://" + entryPoint.listener.Addr().String())
+	require.NoError(t, err)
+	require.False(t, resp.Close)
+	err = resp.Body.Close()
+	require.NoError(t, err)
+
+	resp, err = http.Get("http://" + entryPoint.listener.Addr().String())
+	require.NoError(t, err)
+	require.True(t, resp.Close)
+	err = resp.Body.Close()
+	require.NoError(t, err)
+}
+
+func TestKeepAliveMaxTime(t *testing.T) {
+	epConfig := &static.EntryPointsTransport{}
+	epConfig.SetDefaults()
+	epConfig.KeepAliveMaxTime = ptypes.Duration(time.Millisecond)
+
+	entryPoint, err := NewTCPEntryPoint(context.Background(), &static.EntryPoint{
+		Address:          ":0",
+		Transport:        epConfig,
+		ForwardedHeaders: &static.ForwardedHeaders{},
+		HTTP2:            &static.HTTP2Config{},
+	}, nil)
+	require.NoError(t, err)
+
+	router := &tcprouter.Router{}
+	router.SetHTTPHandler(http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
+		rw.WriteHeader(http.StatusOK)
+	}))
+
+	conn, err := startEntrypoint(entryPoint, router)
+	require.NoError(t, err)
+
+	http.DefaultClient.Transport = &http.Transport{
+		DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
+			return conn, nil
+		},
+	}
+
+	resp, err := http.Get("http://" + entryPoint.listener.Addr().String())
+	require.NoError(t, err)
+	require.False(t, resp.Close)
+	err = resp.Body.Close()
+	require.NoError(t, err)
+
+	time.Sleep(time.Millisecond)
+
+	resp, err = http.Get("http://" + entryPoint.listener.Addr().String())
+	require.NoError(t, err)
+	require.True(t, resp.Close)
+	err = resp.Body.Close()
+	require.NoError(t, err)
+}