feature: httpclient supporting retry net.Client.Retry()

feature: retry() filter

Signed-off-by: Sandor Szücs <[email protected]>

filters/builtin/builtin.go

@@ -15,6 +15,7 @@ import (
 	"github.com/zalando/skipper/filters/fadein"
 	"github.com/zalando/skipper/filters/flowid"
 	logfilter "github.com/zalando/skipper/filters/log"
+	"github.com/zalando/skipper/filters/retry"
 	"github.com/zalando/skipper/filters/rfc"
 	"github.com/zalando/skipper/filters/scheduler"
 	"github.com/zalando/skipper/filters/sed"
@@ -229,6 +230,7 @@ func Filters() []filters.Spec {
 		fadein.NewEndpointCreated(),
 		consistenthash.NewConsistentHashKey(),
 		consistenthash.NewConsistentHashBalanceFactor(),
+		retry.NewRetry(),
 		tls.New(),
 	}
 }

filters/filters.go

@@ -333,6 +333,7 @@ const (
 	FifoWithBodyName                           = "fifoWithBody"
 	LifoName                                   = "lifo"
 	LifoGroupName                              = "lifoGroup"
+	RetryName                                  = "retry"
 	RfcPathName                                = "rfcPath"
 	RfcHostName                                = "rfcHost"
 	BearerInjectorName                         = "bearerinjector"

filters/retry/retry.go

@@ -0,0 +1,18 @@
+package retry
+
+import (
+	"github.com/zalando/skipper/filters"
+)
+
+type retry struct{}
+
+// NewRetry creates a filter specification for the retry() filter
+func NewRetry() filters.Spec { return retry{} }
+
+func (retry) Name() string                                       { return filters.RetryName }
+func (retry) CreateFilter([]interface{}) (filters.Filter, error) { return retry{}, nil }
+func (retry) Response(filters.FilterContext)                     {}
+
+func (retry) Request(ctx filters.FilterContext) {
+	ctx.StateBag()[filters.RetryName] = struct{}{}
+}

filters/retry/retry_test.go

@@ -0,0 +1,95 @@
+package retry
+
+import (
+	"bytes"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/AlexanderYastrebov/noleak"
+	"github.com/zalando/skipper/eskip"
+	"github.com/zalando/skipper/filters"
+	"github.com/zalando/skipper/proxy/proxytest"
+)
+
+func TestRetry(t *testing.T) {
+	for _, tt := range []struct {
+		name   string
+		method string
+		body   string
+	}{
+		{
+			name:   "test GET",
+			method: "GET",
+		},
+		{
+			name:   "test POST",
+			method: "POST",
+			body:   "hello POST",
+		},
+		{
+			name:   "test PATCH",
+			method: "PATCH",
+			body:   "hello PATCH",
+		},
+		{
+			name:   "test PUT",
+			method: "PUT",
+			body:   "hello PUT",
+		}} {
+		t.Run(tt.name, func(t *testing.T) {
+			i := 0
+			backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				if i == 0 {
+					i++
+					w.WriteHeader(http.StatusBadGateway)
+					return
+				}
+
+				got, err := io.ReadAll(r.Body)
+				if err != nil {
+					t.Fatalf("got no data")
+				}
+				s := string(got)
+				if tt.body != s {
+					t.Fatalf("Failed to get the right data want: %q, got: %q", tt.body, s)
+				}
+
+				w.WriteHeader(http.StatusOK)
+			}))
+			defer backend.Close()
+
+			noleak.Check(t)
+
+			fr := make(filters.Registry)
+			retry := NewRetry()
+			fr.Register(retry)
+			r := &eskip.Route{
+				Filters: []*eskip.Filter{
+					{Name: retry.Name()},
+				},
+				Backend: backend.URL,
+			}
+
+			proxy := proxytest.New(fr, r)
+			defer proxy.Close()
+
+			buf := bytes.NewBufferString(tt.body)
+			req, err := http.NewRequest(tt.method, proxy.URL, buf)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			rsp, err := http.DefaultClient.Do(req)
+			if err != nil {
+				t.Fatalf("Failed to execute retry: %v", err)
+			}
+
+			if rsp.StatusCode != http.StatusOK {
+				t.Fatalf("unexpected status code: %s", rsp.Status)
+			}
+			rsp.Body.Close()
+		})
+	}
+}

io/copy_stream.go

@@ -0,0 +1,49 @@
+package io
+
+import (
+	"io"
+)
+
+type ReadWriterLen interface {
+	io.ReadWriter
+	Len() int
+}
+
+type CopyBodyStream struct {
+	left  int
+	buf   ReadWriterLen
+	input io.ReadCloser
+}
+
+func NewCopyBodyStream(left int, buf ReadWriterLen, rc io.ReadCloser) *CopyBodyStream {
+	return &CopyBodyStream{
+		left:  left,
+		buf:   buf,
+		input: rc,
+	}
+}
+
+func (cb *CopyBodyStream) Len() int {
+	return cb.buf.Len()
+}
+
+func (cb *CopyBodyStream) Read(p []byte) (n int, err error) {
+	n, err = cb.input.Read(p)
+	if cb.left > 0 && n > 0 {
+		m := min(n, cb.left)
+		written, err := cb.buf.Write(p[:m])
+		if err != nil {
+			return 0, err
+		}
+		cb.left -= written
+	}
+	return n, err
+}
+
+func (cb *CopyBodyStream) Close() error {
+	return cb.input.Close()
+}
+
+func (cb *CopyBodyStream) GetBody() io.ReadCloser {
+	return io.NopCloser(cb.buf)
+}

io/copy_stream_test.go

@@ -0,0 +1,50 @@
+package io
+
+import (
+	"bytes"
+	"io"
+	"testing"
+)
+
+func TestCopyBodyStream(t *testing.T) {
+	s := "content"
+	bbuf := io.NopCloser(bytes.NewBufferString(s))
+	cbs := NewCopyBodyStream(len(s), &bytes.Buffer{}, bbuf)
+
+	buf := make([]byte, len(s))
+	_, err := cbs.Read(buf)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if cbs.Len() != len(buf) {
+		t.Fatalf("Failed to have the same buf buffer size want: %d, got: %d", cbs.Len(), len(buf))
+	}
+
+	got, err := io.ReadAll(cbs.GetBody())
+	if err != nil {
+		t.Fatalf("Failed to read: %v", err)
+	}
+	if gotStr := string(got); s != gotStr {
+		t.Fatalf("Failed to get the right content: %s != %s", s, gotStr)
+	}
+
+	if err = cbs.Close(); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestCopyBodyStreamFailedRead(t *testing.T) {
+	s := "content"
+	bbuf := io.NopCloser(bytes.NewBufferString(s))
+
+	failingBuf := &failingWriter{buf: &bytes.Buffer{}}
+
+	cbs := NewCopyBodyStream(len(s), failingBuf, bbuf)
+
+	buf := make([]byte, len(s))
+	_, err := cbs.Read(buf)
+	if err == nil {
+		t.Fatal("Want to have failing buffer write inside Read()")
+	}
+}

io/failing_write_buffer_test.go

@@ -0,0 +1,22 @@
+package io
+
+import (
+	"bytes"
+	"fmt"
+)
+
+type failingWriter struct {
+	buf *bytes.Buffer
+}
+
+func (*failingWriter) Write([]byte) (int, error) {
+	return 0, fmt.Errorf("failed to write")
+}
+
+func (fw *failingWriter) Read(p []byte) (int, error) {
+	return fw.buf.Read(p)
+}
+
+func (fw *failingWriter) Len() int {
+	return fw.buf.Len()
+}

net/httpclient.go

@@ -1,7 +1,9 @@
 package net
 
 import (
+	"bytes"
 	"crypto/tls"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -14,6 +16,7 @@ import (
 
 	"github.com/opentracing/opentracing-go"
 	"github.com/opentracing/opentracing-go/ext"
+	skpio "github.com/zalando/skipper/io"
 	"github.com/zalando/skipper/logging"
 	"github.com/zalando/skipper/secrets"
 )
@@ -23,15 +26,18 @@ const (
 	defaultRefreshInterval = 5 * time.Minute
 )
 
+var errRequestNotFound = errors.New("request not found")
+
 // Client adds additional features like Bearer token injection, and
 // opentracing to the wrapped http.Client with the same interface as
 // http.Client from the stdlib.
 type Client struct {
-	once   sync.Once
-	client http.Client
-	tr     *Transport
-	log    logging.Logger
-	sr     secrets.SecretsReader
+	once         sync.Once
+	client       http.Client
+	tr           *Transport
+	log          logging.Logger
+	sr           secrets.SecretsReader
+	retryBuffers *sync.Map
 }
 
 // NewClient creates a wrapped http.Client and uses Transport to
@@ -67,9 +73,10 @@ func NewClient(o Options) *Client {
 			Transport:     tr,
 			CheckRedirect: o.CheckRedirect,
 		},
-		tr:  tr,
-		log: o.Log,
-		sr:  sr,
+		tr:           tr,
+		log:          o.Log,
+		sr:           sr,
+		retryBuffers: &sync.Map{},
 	}
 
 	return c
@@ -125,9 +132,33 @@ func (c *Client) Do(req *http.Request) (*http.Response, error) {
 			req.Header.Set("Authorization", "Bearer "+string(b))
 		}
 	}
+	if req.Body != nil && req.Body != http.NoBody && req.ContentLength > 0 {
+		retryBuffer := skpio.NewCopyBodyStream(int(req.ContentLength), &bytes.Buffer{}, req.Body)
+		c.retryBuffers.Store(req, retryBuffer)
+		req.Body = retryBuffer
+	}
 	return c.client.Do(req)
 }
 
+func (c *Client) Retry(req *http.Request) (*http.Response, error) {
+	if req.Body == nil || req.Body == http.NoBody {
+		return c.Do(req)
+	}
+
+	buf, ok := c.retryBuffers.LoadAndDelete(req)
+	if !ok {
+		return nil, fmt.Errorf("no retry possible, %w: %s %s", errRequestNotFound, req.Method, req.URL)
+	}
+
+	retryBuffer, ok := buf.(*skpio.CopyBodyStream)
+	if !ok {
+		return nil, fmt.Errorf("no retry possible, no retry buffer for request: %s %s", req.Method, req.URL)
+	}
+	req.Body = retryBuffer.GetBody()
+
+	return c.Do(req)
+}
+
 // CloseIdleConnections delegates the call to the underlying
 // http.Client.
 func (c *Client) CloseIdleConnections() {