homelab/traefik
1
0
Fork 0
Code Activity

Introduce Lingering Timeout

Browse source 
Co-authored-by: Baptiste Mayelle <baptiste.mayelle@traefik.io>
Co-authored-by: Kevin Pollet <pollet.kevin@gmail.com>
This commit is contained in:
Romain 2024-04-08 17:16:04 +02:00 • committed by GitHub
parent e5062cef42
commit cef842245c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
14 changed files with 418 additions and 86 deletions
Download patch file Download diff file Expand all files Collapse all files

12
pkg/server/router/tcp/router.go
View file

@ -9,7 +9,6 @@ import (
"net"
"net/http"
"slices"
"time"
"github.com/go-acme/lego/v4/challenge/tlsalpn01"
"github.com/traefik/traefik/v2/pkg/log"
@ -117,17 +116,6 @@ func (r *Router) ServeTCP(conn tcp.WriteCloser) {
return
}
// Remove read/write deadline and delegate this to underlying tcp server (for now only handled by HTTP Server)
err = conn.SetReadDeadline(time.Time{})
if err != nil {
log.WithoutContext().Errorf("Error while setting read deadline: %v", err)
}
err = conn.SetWriteDeadline(time.Time{})
if err != nil {
log.WithoutContext().Errorf("Error while setting write deadline: %v", err)
}
connData, err := tcpmuxer.NewConnData(hello.serverName, conn, hello.protos)
if err != nil {
log.WithoutContext().Errorf("Error while reading TCP connection data: %v", err)

82
pkg/server/server_entrypoint_tcp.go
View file

@ -244,24 +244,15 @@ func (e *TCPEntryPoint) Start(ctx context.Context) {
panic(err)
}
if e.transportConfiguration != nil &&
e.transportConfiguration.RespondingTimeouts != nil &&
e.transportConfiguration.RespondingTimeouts.TCP != nil &&
e.transportConfiguration.RespondingTimeouts.TCP.LingeringTimeout > 0 {
lingeringTimeout := time.Duration(e.transportConfiguration.RespondingTimeouts.TCP.LingeringTimeout)
writeCloser = newLingeringConnection(writeCloser, lingeringTimeout)
}
safe.Go(func() {
// Enforce read/write deadlines at the connection level,
// because when we're peeking the first byte to determine whether we are doing TLS,
// the deadlines at the server level are not taken into account.
if e.transportConfiguration.RespondingTimeouts.ReadTimeout > 0 {
err := writeCloser.SetReadDeadline(time.Now().Add(time.Duration(e.transportConfiguration.RespondingTimeouts.ReadTimeout)))
if err != nil {
logger.Errorf("Error while setting read deadline: %v", err)
}
}
if e.transportConfiguration.RespondingTimeouts.WriteTimeout > 0 {
err = writeCloser.SetWriteDeadline(time.Now().Add(time.Duration(e.transportConfiguration.RespondingTimeouts.WriteTimeout)))
if err != nil {
logger.Errorf("Error while setting write deadline: %v", err)
}
}
e.switcher.ServeTCP(newTrackedConnection(writeCloser, e.tracker))
})
}
@ -391,6 +382,55 @@ func writeCloser(conn net.Conn) (tcp.WriteCloser, error) {
}
}
// lingeringConn represents a writeCloser with lingeringTimeout handling.
type lingeringConn struct {
tcp.WriteCloser
lingeringTimeout time.Duration
rdlMu sync.RWMutex
// readDeadline is the current readDeadline set by an upper caller.
// In case of HTTP, the HTTP go server manipulates deadlines on the connection.
readDeadline time.Time
}
// newLingeringConnection returns the given writeCloser augmented with lingeringTimeout handling.
func newLingeringConnection(conn tcp.WriteCloser, timeout time.Duration) tcp.WriteCloser {
return &lingeringConn{
WriteCloser: conn,
lingeringTimeout: timeout,
}
}
// Read reads data from the connection and postpones the connection readDeadline according to the lingeringTimeout config.
// It also ensures that the upper level set readDeadline is enforced.
func (l *lingeringConn) Read(b []byte) (int, error) {
if l.lingeringTimeout > 0 {
deadline := time.Now().Add(l.lingeringTimeout)
l.rdlMu.RLock()
if !l.readDeadline.IsZero() && deadline.After(l.readDeadline) {
deadline = l.readDeadline
}
l.rdlMu.RUnlock()
if err := l.WriteCloser.SetReadDeadline(deadline); err != nil {
return 0, err
}
}
return l.WriteCloser.Read(b)
}
// SetReadDeadline sets and save the read deadline.
func (l *lingeringConn) SetReadDeadline(t time.Time) error {
l.rdlMu.Lock()
l.readDeadline = t
l.rdlMu.Unlock()
return l.WriteCloser.SetReadDeadline(t)
}
// tcpKeepAliveListener sets TCP keep-alive timeouts on accepted
// connections.
type tcpKeepAliveListener struct {
@ -419,7 +459,7 @@ func (ln tcpKeepAliveListener) Accept() (net.Conn, error) {
}
func buildProxyProtocolListener(ctx context.Context, entryPoint *static.EntryPoint, listener net.Listener) (net.Listener, error) {
timeout := entryPoint.Transport.RespondingTimeouts.ReadTimeout
timeout := *entryPoint.Transport.RespondingTimeouts.HTTP.ReadTimeout
// proxyproto use 200ms if ReadHeaderTimeout is set to 0 and not no timeout
if timeout == 0 {
timeout = -1
@ -588,9 +628,9 @@ func createHTTPServer(ctx context.Context, ln net.Listener, configuration *stati
serverHTTP := &http.Server{
Handler: handler,
ErrorLog: httpServerLogger,
ReadTimeout: time.Duration(configuration.Transport.RespondingTimeouts.ReadTimeout),
WriteTimeout: time.Duration(configuration.Transport.RespondingTimeouts.WriteTimeout),
IdleTimeout: time.Duration(configuration.Transport.RespondingTimeouts.IdleTimeout),
ReadTimeout: time.Duration(*configuration.Transport.RespondingTimeouts.HTTP.ReadTimeout),
WriteTimeout: time.Duration(*configuration.Transport.RespondingTimeouts.HTTP.WriteTimeout),
IdleTimeout: time.Duration(*configuration.Transport.RespondingTimeouts.HTTP.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 {

88
pkg/server/server_entrypoint_tcp_test.go
View file

@ -69,8 +69,10 @@ func testShutdown(t *testing.T, router *tcprouter.Router) {
epConfig.LifeCycle.RequestAcceptGraceTimeout = 0
epConfig.LifeCycle.GraceTimeOut = ptypes.Duration(5 * time.Second)
epConfig.RespondingTimeouts.ReadTimeout = ptypes.Duration(5 * time.Second)
epConfig.RespondingTimeouts.WriteTimeout = ptypes.Duration(5 * time.Second)
readTimeout := ptypes.Duration(5 * time.Second)
epConfig.RespondingTimeouts.HTTP.ReadTimeout = &readTimeout
writeTimeout := ptypes.Duration(5 * time.Second)
epConfig.RespondingTimeouts.HTTP.WriteTimeout = &writeTimeout
entryPoint, err := NewTCPEntryPoint(context.Background(), &static.EntryPoint{
// We explicitly use an IPV4 address because on Alpine, with an IPV6 address
@ -157,7 +159,8 @@ func startEntrypoint(entryPoint *TCPEntryPoint, router *tcprouter.Router) (net.C
func TestReadTimeoutWithoutFirstByte(t *testing.T) {
epConfig := &static.EntryPointsTransport{}
epConfig.SetDefaults()
epConfig.RespondingTimeouts.ReadTimeout = ptypes.Duration(2 * time.Second)
readTimeout := ptypes.Duration(2 * time.Second)
epConfig.RespondingTimeouts.HTTP.ReadTimeout = &readTimeout
entryPoint, err := NewTCPEntryPoint(context.Background(), &static.EntryPoint{
Address: ":0",
@ -194,7 +197,84 @@ func TestReadTimeoutWithoutFirstByte(t *testing.T) {
func TestReadTimeoutWithFirstByte(t *testing.T) {
epConfig := &static.EntryPointsTransport{}
epConfig.SetDefaults()
epConfig.RespondingTimeouts.ReadTimeout = ptypes.Duration(2 * time.Second)
readTimeout := ptypes.Duration(2 * time.Second)
epConfig.RespondingTimeouts.HTTP.ReadTimeout = &readTimeout
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)
_, err = conn.Write([]byte("GET /some HTTP/1.1\r\n"))
require.NoError(t, err)
errChan := make(chan error)
go func() {
b := make([]byte, 2048)
_, err := conn.Read(b)
errChan <- err
}()
select {
case err := <-errChan:
require.Equal(t, io.EOF, err)
case <-time.Tick(5 * time.Second):
t.Error("Timeout while read")
}
}
func TestLingeringTimeoutWithoutFirstByte(t *testing.T) {
epConfig := &static.EntryPointsTransport{}
epConfig.SetDefaults()
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)
errChan := make(chan error)
go func() {
b := make([]byte, 2048)
_, err := conn.Read(b)
errChan <- err
}()
select {
case err := <-errChan:
require.Equal(t, io.EOF, err)
case <-time.Tick(5 * time.Second):
t.Error("Timeout while read")
}
}
func TestLingeringTimeoutWithFirstByte(t *testing.T) {
epConfig := &static.EntryPointsTransport{}
epConfig.SetDefaults()
epConfig.RespondingTimeouts.TCP.LingeringTimeout = ptypes.Duration(time.Second)
entryPoint, err := NewTCPEntryPoint(context.Background(), &static.EntryPoint{
Address: ":0",