Merge branch v2.11 into v3.0
This commit is contained in:
commit
c1d9b9ee1f
33 changed files with 602 additions and 159 deletions
|
@ -9,7 +9,6 @@ import (
|
|||
"net"
|
||||
"net/http"
|
||||
"slices"
|
||||
"time"
|
||||
|
||||
"github.com/go-acme/lego/v4/challenge/tlsalpn01"
|
||||
"github.com/rs/zerolog/log"
|
||||
|
@ -128,17 +127,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.Error().Err(err).Msg("Error while setting read deadline")
|
||||
}
|
||||
|
||||
err = conn.SetWriteDeadline(time.Time{})
|
||||
if err != nil {
|
||||
log.Error().Err(err).Msg("Error while setting write deadline")
|
||||
}
|
||||
|
||||
connData, err := tcpmuxer.NewConnData(hello.serverName, conn, hello.protos)
|
||||
if err != nil {
|
||||
log.Error().Err(err).Msg("Error while reading TCP connection data")
|
||||
|
|
|
@ -249,24 +249,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.Error().Err(err).Msg("Error while setting read deadline")
|
||||
}
|
||||
}
|
||||
|
||||
if e.transportConfiguration.RespondingTimeouts.WriteTimeout > 0 {
|
||||
err = writeCloser.SetWriteDeadline(time.Now().Add(time.Duration(e.transportConfiguration.RespondingTimeouts.WriteTimeout)))
|
||||
if err != nil {
|
||||
logger.Error().Err(err).Msg("Error while setting write deadline")
|
||||
}
|
||||
}
|
||||
|
||||
e.switcher.ServeTCP(newTrackedConnection(writeCloser, e.tracker))
|
||||
})
|
||||
}
|
||||
|
@ -398,6 +389,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 {
|
||||
|
@ -425,7 +465,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
|
||||
|
@ -608,9 +648,9 @@ func createHTTPServer(ctx context.Context, ln net.Listener, configuration *stati
|
|||
serverHTTP := &http.Server{
|
||||
Handler: handler,
|
||||
ErrorLog: stdlog.New(logs.NoLevel(log.Logger, zerolog.DebugLevel), "", 0),
|
||||
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 {
|
||||
|
|
|
@ -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, 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, 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",
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue