Merge branch v2.11 into v3.3
This commit is contained in:
commit
32ea014d07
13 changed files with 351 additions and 215 deletions
|
@ -1,10 +1,9 @@
|
|||
package api
|
||||
|
||||
import (
|
||||
"cmp"
|
||||
"net/url"
|
||||
"sort"
|
||||
|
||||
"golang.org/x/exp/constraints"
|
||||
)
|
||||
|
||||
const (
|
||||
|
@ -357,7 +356,7 @@ func sortByName[T orderedWithName](direction string, results []T) {
|
|||
})
|
||||
}
|
||||
|
||||
func sortByFunc[T orderedWithName, U constraints.Ordered](direction string, results []T, fn func(int) U) {
|
||||
func sortByFunc[T orderedWithName, U cmp.Ordered](direction string, results []T, fn func(int) U) {
|
||||
// Ascending
|
||||
if direction == ascendantSorting {
|
||||
sort.Slice(results, func(i, j int) bool {
|
||||
|
|
|
@ -36,6 +36,48 @@ type Listener interface {
|
|||
// each of them about a retry attempt.
|
||||
type Listeners []Listener
|
||||
|
||||
// Retried exists to implement the Listener interface. It calls Retried on each of its slice entries.
|
||||
func (l Listeners) Retried(req *http.Request, attempt int) {
|
||||
for _, listener := range l {
|
||||
listener.Retried(req, attempt)
|
||||
}
|
||||
}
|
||||
|
||||
type shouldRetryContextKey struct{}
|
||||
|
||||
// ShouldRetry is a function allowing to enable/disable the retry middleware mechanism.
|
||||
type ShouldRetry func(shouldRetry bool)
|
||||
|
||||
// ContextShouldRetry returns the ShouldRetry function if it has been set by the Retry middleware in the chain.
|
||||
func ContextShouldRetry(ctx context.Context) ShouldRetry {
|
||||
f, _ := ctx.Value(shouldRetryContextKey{}).(ShouldRetry)
|
||||
return f
|
||||
}
|
||||
|
||||
// WrapHandler wraps a given http.Handler to inject the httptrace.ClientTrace in the request context when it is needed
|
||||
// by the retry middleware.
|
||||
func WrapHandler(next http.Handler) http.Handler {
|
||||
return http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
|
||||
if shouldRetry := ContextShouldRetry(req.Context()); shouldRetry != nil {
|
||||
shouldRetry(true)
|
||||
|
||||
trace := &httptrace.ClientTrace{
|
||||
WroteHeaders: func() {
|
||||
shouldRetry(false)
|
||||
},
|
||||
WroteRequest: func(httptrace.WroteRequestInfo) {
|
||||
shouldRetry(false)
|
||||
},
|
||||
}
|
||||
newCtx := httptrace.WithClientTrace(req.Context(), trace)
|
||||
next.ServeHTTP(rw, req.WithContext(newCtx))
|
||||
return
|
||||
}
|
||||
|
||||
next.ServeHTTP(rw, req)
|
||||
})
|
||||
}
|
||||
|
||||
// retry is a middleware that retries requests.
|
||||
type retry struct {
|
||||
attempts int
|
||||
|
@ -101,19 +143,13 @@ func (r *retry) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
|
|||
req = req.WithContext(tracingCtx)
|
||||
}
|
||||
|
||||
shouldRetry := attempts < r.attempts
|
||||
retryResponseWriter := newResponseWriter(rw, shouldRetry)
|
||||
remainAttempts := attempts < r.attempts
|
||||
retryResponseWriter := newResponseWriter(rw)
|
||||
|
||||
// Disable retries when the backend already received request data
|
||||
clientTrace := &httptrace.ClientTrace{
|
||||
WroteHeaders: func() {
|
||||
retryResponseWriter.DisableRetries()
|
||||
},
|
||||
WroteRequest: func(httptrace.WroteRequestInfo) {
|
||||
retryResponseWriter.DisableRetries()
|
||||
},
|
||||
var shouldRetry ShouldRetry = func(shouldRetry bool) {
|
||||
retryResponseWriter.SetShouldRetry(remainAttempts && shouldRetry)
|
||||
}
|
||||
newCtx := httptrace.WithClientTrace(req.Context(), clientTrace)
|
||||
newCtx := context.WithValue(req.Context(), shouldRetryContextKey{}, shouldRetry)
|
||||
|
||||
r.next.ServeHTTP(retryResponseWriter, req.Clone(newCtx))
|
||||
|
||||
|
@ -164,18 +200,10 @@ func (r *retry) newBackOff() backoff.BackOff {
|
|||
return b
|
||||
}
|
||||
|
||||
// Retried exists to implement the Listener interface. It calls Retried on each of its slice entries.
|
||||
func (l Listeners) Retried(req *http.Request, attempt int) {
|
||||
for _, listener := range l {
|
||||
listener.Retried(req, attempt)
|
||||
}
|
||||
}
|
||||
|
||||
func newResponseWriter(rw http.ResponseWriter, shouldRetry bool) *responseWriter {
|
||||
func newResponseWriter(rw http.ResponseWriter) *responseWriter {
|
||||
return &responseWriter{
|
||||
responseWriter: rw,
|
||||
headers: make(http.Header),
|
||||
shouldRetry: shouldRetry,
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -190,8 +218,8 @@ func (r *responseWriter) ShouldRetry() bool {
|
|||
return r.shouldRetry
|
||||
}
|
||||
|
||||
func (r *responseWriter) DisableRetries() {
|
||||
r.shouldRetry = false
|
||||
func (r *responseWriter) SetShouldRetry(shouldRetry bool) {
|
||||
r.shouldRetry = shouldRetry
|
||||
}
|
||||
|
||||
func (r *responseWriter) Header() http.Header {
|
||||
|
@ -205,20 +233,14 @@ func (r *responseWriter) Write(buf []byte) (int, error) {
|
|||
if r.ShouldRetry() {
|
||||
return len(buf), nil
|
||||
}
|
||||
if !r.written {
|
||||
r.WriteHeader(http.StatusOK)
|
||||
}
|
||||
return r.responseWriter.Write(buf)
|
||||
}
|
||||
|
||||
func (r *responseWriter) WriteHeader(code int) {
|
||||
if r.ShouldRetry() && code == http.StatusServiceUnavailable {
|
||||
// We get a 503 HTTP Status Code when there is no backend server in the pool
|
||||
// to which the request could be sent. Also, note that r.ShouldRetry()
|
||||
// will never return true in case there was a connection established to
|
||||
// the backend server and so we can be sure that the 503 was produced
|
||||
// inside Traefik already and we don't have to retry in this cases.
|
||||
r.DisableRetries()
|
||||
}
|
||||
|
||||
if r.ShouldRetry() || r.written {
|
||||
if r.shouldRetry || r.written {
|
||||
return
|
||||
}
|
||||
|
||||
|
|
|
@ -105,12 +105,21 @@ func TestRetry(t *testing.T) {
|
|||
t.Parallel()
|
||||
|
||||
retryAttempts := 0
|
||||
next := http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {
|
||||
next := http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
|
||||
// This signals that a connection will be established with the backend
|
||||
// to enable the Retry middleware mechanism.
|
||||
shouldRetry := ContextShouldRetry(req.Context())
|
||||
if shouldRetry != nil {
|
||||
shouldRetry(true)
|
||||
}
|
||||
|
||||
retryAttempts++
|
||||
|
||||
if retryAttempts > test.amountFaultyEndpoints {
|
||||
// calls WroteHeaders on httptrace.
|
||||
_ = r.Write(io.Discard)
|
||||
// This signals that request headers have been sent to the backend.
|
||||
if shouldRetry != nil {
|
||||
shouldRetry(false)
|
||||
}
|
||||
|
||||
rw.WriteHeader(http.StatusOK)
|
||||
return
|
||||
|
@ -152,27 +161,16 @@ func TestRetryEmptyServerList(t *testing.T) {
|
|||
assert.Equal(t, 0, retryListener.timesCalled)
|
||||
}
|
||||
|
||||
func TestRetryListeners(t *testing.T) {
|
||||
req := httptest.NewRequest(http.MethodGet, "/", nil)
|
||||
retryListeners := Listeners{&countingRetryListener{}, &countingRetryListener{}}
|
||||
|
||||
retryListeners.Retried(req, 1)
|
||||
retryListeners.Retried(req, 1)
|
||||
|
||||
for _, retryListener := range retryListeners {
|
||||
listener := retryListener.(*countingRetryListener)
|
||||
if listener.timesCalled != 2 {
|
||||
t.Errorf("retry listener was called %d time(s), want %d time(s)", listener.timesCalled, 2)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestMultipleRetriesShouldNotLooseHeaders(t *testing.T) {
|
||||
attempt := 0
|
||||
expectedHeaderName := "X-Foo-Test-2"
|
||||
expectedHeaderValue := "bar"
|
||||
|
||||
next := http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
|
||||
shouldRetry := ContextShouldRetry(req.Context())
|
||||
if shouldRetry != nil {
|
||||
shouldRetry(true)
|
||||
}
|
||||
|
||||
headerName := fmt.Sprintf("X-Foo-Test-%d", attempt)
|
||||
rw.Header().Add(headerName, expectedHeaderValue)
|
||||
if attempt < 2 {
|
||||
|
@ -181,43 +179,54 @@ func TestMultipleRetriesShouldNotLooseHeaders(t *testing.T) {
|
|||
}
|
||||
|
||||
// Request has been successfully written to backend
|
||||
trace := httptrace.ContextClientTrace(req.Context())
|
||||
trace.WroteHeaders()
|
||||
shouldRetry(false)
|
||||
|
||||
// And we decide to answer to client
|
||||
// And we decide to answer to client.
|
||||
rw.WriteHeader(http.StatusNoContent)
|
||||
})
|
||||
|
||||
retry, err := New(context.Background(), next, dynamic.Retry{Attempts: 3}, &countingRetryListener{}, "traefikTest")
|
||||
require.NoError(t, err)
|
||||
|
||||
responseRecorder := httptest.NewRecorder()
|
||||
retry.ServeHTTP(responseRecorder, testhelpers.MustNewRequest(http.MethodGet, "http://test", http.NoBody))
|
||||
res := httptest.NewRecorder()
|
||||
retry.ServeHTTP(res, testhelpers.MustNewRequest(http.MethodGet, "http://test", http.NoBody))
|
||||
|
||||
headerValue := responseRecorder.Header().Get(expectedHeaderName)
|
||||
|
||||
// Validate if we have the correct header
|
||||
if headerValue != expectedHeaderValue {
|
||||
t.Errorf("Expected to have %s for header %s, got %s", expectedHeaderValue, expectedHeaderName, headerValue)
|
||||
}
|
||||
// The third header attempt is kept.
|
||||
headerValue := res.Header().Get("X-Foo-Test-2")
|
||||
assert.Equal(t, expectedHeaderValue, headerValue)
|
||||
|
||||
// Validate that we don't have headers from previous attempts
|
||||
for i := range attempt {
|
||||
headerName := fmt.Sprintf("X-Foo-Test-%d", i)
|
||||
headerValue = responseRecorder.Header().Get("headerName")
|
||||
headerValue = res.Header().Get(headerName)
|
||||
if headerValue != "" {
|
||||
t.Errorf("Expected no value for header %s, got %s", headerName, headerValue)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// countingRetryListener is a Listener implementation to count the times the Retried fn is called.
|
||||
type countingRetryListener struct {
|
||||
timesCalled int
|
||||
}
|
||||
func TestRetryShouldNotLooseHeadersOnWrite(t *testing.T) {
|
||||
next := http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
|
||||
rw.Header().Add("X-Foo-Test", "bar")
|
||||
|
||||
func (l *countingRetryListener) Retried(req *http.Request, attempt int) {
|
||||
l.timesCalled++
|
||||
// Request has been successfully written to backend.
|
||||
shouldRetry := ContextShouldRetry(req.Context())
|
||||
if shouldRetry != nil {
|
||||
shouldRetry(false)
|
||||
}
|
||||
// And we decide to answer to client without calling WriteHeader.
|
||||
_, err := rw.Write([]byte("bar"))
|
||||
require.NoError(t, err)
|
||||
})
|
||||
|
||||
retry, err := New(context.Background(), next, dynamic.Retry{Attempts: 3}, &countingRetryListener{}, "traefikTest")
|
||||
require.NoError(t, err)
|
||||
|
||||
res := httptest.NewRecorder()
|
||||
retry.ServeHTTP(res, testhelpers.MustNewRequest(http.MethodGet, "http://test", http.NoBody))
|
||||
|
||||
headerValue := res.Header().Get("X-Foo-Test")
|
||||
assert.Equal(t, "bar", headerValue)
|
||||
}
|
||||
|
||||
func TestRetryWithFlush(t *testing.T) {
|
||||
|
@ -275,12 +284,24 @@ func TestRetryWebsocket(t *testing.T) {
|
|||
t.Parallel()
|
||||
|
||||
retryAttempts := 0
|
||||
next := http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {
|
||||
next := http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
|
||||
// This signals that a connection will be established with the backend
|
||||
// to enable the Retry middleware mechanism.
|
||||
shouldRetry := ContextShouldRetry(req.Context())
|
||||
if shouldRetry != nil {
|
||||
shouldRetry(true)
|
||||
}
|
||||
|
||||
retryAttempts++
|
||||
|
||||
if retryAttempts > test.amountFaultyEndpoints {
|
||||
// This signals that request headers have been sent to the backend.
|
||||
if shouldRetry != nil {
|
||||
shouldRetry(false)
|
||||
}
|
||||
|
||||
upgrader := websocket.Upgrader{}
|
||||
_, err := upgrader.Upgrade(rw, r, nil)
|
||||
_, err := upgrader.Upgrade(rw, req, nil)
|
||||
if err != nil {
|
||||
http.Error(rw, err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
|
@ -387,3 +408,12 @@ func Test1xxResponses(t *testing.T) {
|
|||
|
||||
assert.Equal(t, 0, retryListener.timesCalled)
|
||||
}
|
||||
|
||||
// countingRetryListener is a Listener implementation to count the times the Retried fn is called.
|
||||
type countingRetryListener struct {
|
||||
timesCalled int
|
||||
}
|
||||
|
||||
func (l *countingRetryListener) Retried(req *http.Request, attempt int) {
|
||||
l.timesCalled++
|
||||
}
|
||||
|
|
|
@ -22,6 +22,7 @@ import (
|
|||
"github.com/traefik/traefik/v3/pkg/middlewares/capture"
|
||||
metricsMiddle "github.com/traefik/traefik/v3/pkg/middlewares/metrics"
|
||||
"github.com/traefik/traefik/v3/pkg/middlewares/observability"
|
||||
"github.com/traefik/traefik/v3/pkg/middlewares/retry"
|
||||
"github.com/traefik/traefik/v3/pkg/proxy/httputil"
|
||||
"github.com/traefik/traefik/v3/pkg/safe"
|
||||
"github.com/traefik/traefik/v3/pkg/server/cookie"
|
||||
|
@ -349,6 +350,10 @@ func (m *Manager) getLoadBalancerServiceHandler(ctx context.Context, serviceName
|
|||
if err != nil {
|
||||
return nil, fmt.Errorf("error building proxy for server URL %s: %w", server.URL, err)
|
||||
}
|
||||
// The retry wrapping must be done just before the proxy handler,
|
||||
// to make sure that the retry will not be triggered/disabled by
|
||||
// middlewares in the chain.
|
||||
proxy = retry.WrapHandler(proxy)
|
||||
|
||||
// Prevents from enabling observability for internal resources.
|
||||
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue