Merge pull request #387 from rohanpm/retry-rehttp

Implement retries at http RoundTripper level [RHELDST-22079]

README.md

@@ -230,6 +230,12 @@ gwpollinterval: 5000
 # When adding items onto an exodus-gw publish, what is the maximum number of
 # items we'll include in a single HTTP request.
 gwbatchsize: 10000
+
+# How many times to retry failing HTTP requests.
+gwmaxattempts: 3
+
+# Maximum duration (in milliseconds) between retries of HTTP requests.
+gwmaxbackoff: 20000
 ```
 
 In order to publish to exodus CDN it is necessary to configure all of the

go.mod

@@ -16,6 +16,7 @@ require (
 )
 
 require (
+	github.com/PuerkitoBio/rehttp v1.3.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/fatih/color v1.16.0 // indirect
 	github.com/go-playground/locales v0.14.1 // indirect

go.sum

@@ -1,3 +1,5 @@
+github.com/PuerkitoBio/rehttp v1.3.0 h1:w54Pb72MQn2eJrSdPsvGqXlAfiK1+NMTGDrOJJ4YvSU=
+github.com/PuerkitoBio/rehttp v1.3.0/go.mod h1:LUwKPoDbDIA2RL5wYZCNsQ90cx4OJ4AWBmq6KzWZL1s=
 github.com/adrg/xdg v0.4.0 h1:RzRqFcjH4nE5C6oTAxhBtoE2IRyjBSa62SCbyPidvls=
 github.com/adrg/xdg v0.4.0/go.mod h1:N6ag73EX4wyxeaoeHctc1mas01KZgsj5tYiAIwqJE/E=
 github.com/alecthomas/assert/v2 v2.1.0 h1:tbredtNcQnoSd3QBhQWI7QZ3XHOVkw1Moklp2ojoH/0=
@@ -12,12 +14,15 @@ github.com/aphistic/sweet v0.2.0/go.mod h1:fWDlIh/isSE9n6EPsRmC0det+whmX6dJid3st
 github.com/aws/aws-sdk-go v1.20.6/go.mod h1:KmX6BPdI08NWTb3/sm4ZGu5ShLoqVDhKgpiN924inxo=
 github.com/aws/aws-sdk-go v1.48.3 h1:btYjT+opVFxUbRz+qSCjJe07cdX82BHmMX/FXYmoL7g=
 github.com/aws/aws-sdk-go v1.48.3/go.mod h1:LF8svs817+Nz+DmiMQKTO3ubZ/6IaTpq3TjupRn3Eqk=
+github.com/aybabtme/iocontrol v0.0.0-20150809002002-ad15bcfc95a0/go.mod h1:6L7zgvqo0idzI7IO8de6ZC051AfXb5ipkIJ7bIA2tGA=
 github.com/aybabtme/rgbterm v0.0.0-20170906152045-cc83f3b3ce59/go.mod h1:q/89r3U2H7sSsE2t6Kca0lfwTK8JdoNGS/yzM/4iH5I=
+github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
 github.com/coreos/go-systemd/v22 v22.5.0 h1:RrqgGjYQKalulkV8NGVIfkXQf6YYmOyiJKk8iXXhfZs=
 github.com/coreos/go-systemd/v22 v22.5.0/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
 github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
 github.com/fatih/color v1.16.0 h1:zmkK9Ngbjj+K0yRhTVONQh1p/HknKYSlNT+vZCzyokM=
 github.com/fatih/color v1.16.0/go.mod h1:fL2Sau1YI5c0pdGEVCbKQbLXB6edEj1ZgiY4NijnWvE=
@@ -105,6 +110,7 @@ golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73r
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
+golang.org/x/net v0.0.0-20210510120150-4163338589ed/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.18.0 h1:mIYleuAkSbHh0tCv7RvjL3F6ZVbLjq4+R7zbOn3Kokg=
 golang.org/x/net v0.18.0/go.mod h1:/czyP5RqHAH4odGYxBJ1qz0+CE5WZ+2j1YgoEo8F2jQ=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -116,6 +122,7 @@ golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5h
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20211025201205-69cdffdb9359/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
@@ -126,6 +133,7 @@ golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9sn
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
 golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=

internal/conf/conf.go

@@ -43,6 +43,12 @@ type Config interface {
 	// Commit mode for publishes.
 	GwCommit() string
 
+	// Maximum attempts for any HTTP request to exodus-gw.
+	GwMaxAttempts() int
+
+	// Maximum backoff between retried HTTP requests, in milliseconds.
+	GwMaxBackoff() int
+
 	// Execution mode for rsync.
 	RsyncMode() string
 

internal/conf/conf_test.go

@@ -41,6 +41,8 @@ environments:
   gwkey: override-key
   gwpollinterval: 123
   gwcommit: cba
+  gwmaxattempts: 50
+  gwmaxbackoff: 60
   rsyncmode: mixed
   strip: dest:/foo/bar
   uploadthreads: 6
@@ -100,6 +102,8 @@ environments:
 	assertEqual("global gwenv", cfg.GwEnv(), "global-env")
 	assertEqual("global gwpollinterval", cfg.GwPollInterval(), 5000)
 	assertEqual("global gwcommit", cfg.GwCommit(), "abc")
+	assertEqual("global gwmaxattempts", cfg.GwMaxAttempts(), 3)
+	assertEqual("global gwmaxbackoff", cfg.GwMaxBackoff(), 20000)
 	assertEqual("global rsyncmode", cfg.RsyncMode(), "exodus")
 	assertEqual("global strip", cfg.Strip(), "dest:/foo")
 	assertEqual("global uploadthreads", cfg.UploadThreads(), 4)
@@ -109,6 +113,8 @@ environments:
 	assertEqual("env gwkey", env.GwKey(), "override-key")
 	assertEqual("env gwpollinterval", env.GwPollInterval(), 123)
 	assertEqual("env gwcommit", env.GwCommit(), "cba")
+	assertEqual("env gwmaxattempts", env.GwMaxAttempts(), 50)
+	assertEqual("env gwmaxbackoff", env.GwMaxBackoff(), 60)
 	assertEqual("env rsyncmode", env.RsyncMode(), "mixed")
 	assertEqual("env strip", env.Strip(), "dest:/foo/bar")
 	assertEqual("env uploadthreads", env.UploadThreads(), 6)

internal/conf/structs.go

@@ -15,6 +15,8 @@ type sharedConfig struct {
 	GwPollIntervalRaw int    `yaml:"gwpollinterval"`
 	GwBatchSizeRaw    int    `yaml:"gwbatchsize"`
 	GwCommitRaw       string `yaml:"gwcommit"`
+	GwMaxAttemptsRaw  int    `yaml:"gwmaxattempts"`
+	GwMaxBackoffRaw   int    `yaml:"gwmaxbackoff"`
 	RsyncModeRaw      string `yaml:"rsyncmode"`
 	LogLevelRaw       string `yaml:"loglevel"`
 	LoggerRaw         string `yaml:"logger"`
@@ -78,6 +80,14 @@ func (g *globalConfig) GwCommit() string {
 	return g.GwCommitRaw
 }
 
+func (g *globalConfig) GwMaxAttempts() int {
+	return nonEmptyInt(g.GwMaxAttemptsRaw, 3)
+}
+
+func (g *globalConfig) GwMaxBackoff() int {
+	return nonEmptyInt(g.GwMaxBackoffRaw, 20000)
+}
+
 func (g *globalConfig) UploadThreads() int {
 	return nonEmptyInt(g.UploadThreadsRaw, 4)
 }
@@ -148,6 +158,14 @@ func (e *environment) GwCommit() string {
 	return nonEmptyString(e.GwCommitRaw, e.parent.GwCommit())
 }
 
+func (e *environment) GwMaxAttempts() int {
+	return nonEmptyInt(e.GwMaxAttemptsRaw, e.parent.GwMaxAttempts())
+}
+
+func (e *environment) GwMaxBackoff() int {
+	return nonEmptyInt(e.GwMaxBackoffRaw, e.parent.GwMaxBackoff())
+}
+
 func (e *environment) RsyncMode() string {
 	return nonEmptyString(e.RsyncModeRaw, e.parent.RsyncMode())
 }

internal/diag/diag.go

@@ -64,6 +64,8 @@ func logConfig(ctx context.Context, cfg conf.Config) {
 		"gwenv", cfg.GwEnv(),
 		"gwpollinterval", cfg.GwPollInterval(),
 		"gwbatchsize", cfg.GwBatchSize(),
+		"gwmaxattempts", cfg.GwMaxAttempts(),
+		"gwmaxbackoff", cfg.GwMaxBackoff(),
 	).Warn("exodus-gw")
 
 	logger.F(

internal/diag/diag_test.go

@@ -24,6 +24,8 @@ func mockConfig(ctrl *gomock.Controller) conf.Config {
 	e.GwEnv().Return("test-env").AnyTimes()
 	e.GwPollInterval().Return(123).AnyTimes()
 	e.GwBatchSize().Return(234).AnyTimes()
+	e.GwMaxAttempts().Return(345).AnyTimes()
+	e.GwMaxBackoff().Return(456).AnyTimes()
 	e.RsyncMode().Return("mixed").AnyTimes()
 	e.LogLevel().Return("debug").AnyTimes()
 	e.Logger().Return("syslog").AnyTimes()

internal/gw/client.go

@@ -11,7 +11,9 @@ import (
 	"os"
 	"runtime"
 	"sync"
+	"time"
 
+	"github.com/PuerkitoBio/rehttp"
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/awserr"
 	"github.com/aws/aws-sdk-go/aws/credentials"
@@ -341,6 +343,64 @@ func (c *client) EnsureUploaded(
 	return <-out
 }
 
+func retryWithLogging(logger *log.Logger, fn rehttp.RetryFn) rehttp.RetryFn {
+	// Wraps a rehttp.RetryFn to add warnings on retries.
+	return func(attempt rehttp.Attempt) bool {
+		willRetry := fn(attempt)
+		status := "<none>"
+		if attempt.Response != nil {
+			// status is the HTTP response status and that
+			// sometimes won't be present.
+			status = attempt.Response.Status
+		}
+
+		entry := logger.F(
+			"url", attempt.Request.URL,
+			"index", attempt.Index,
+			"method", attempt.Request.Method,
+			"status", status,
+			"error", attempt.Error,
+		)
+
+		if willRetry {
+			entry.Warn("Retrying failed request")
+		} else {
+			// This is Debug because we get here even for successful
+			// requests, so it's not normally worth logging.
+			// But if we don't log at all, it's hard to find out which
+			// errors are not getting retried in order to tune it.
+			entry.Debug("Not retrying request")
+		}
+
+		return willRetry
+	}
+}
+
+func retryTransport(ctx context.Context, cfg conf.Config, rt http.RoundTripper) http.RoundTripper {
+	// Wrap a roundtripper with retries.
+	logger := log.FromContext(ctx)
+
+	retryFn := rehttp.RetryAll(
+		rehttp.RetryMaxRetries(cfg.GwMaxAttempts()),
+		rehttp.RetryAny(
+			rehttp.RetryStatuses(500, 502, 503, 504),
+			rehttp.RetryTimeoutErr(),
+			rehttp.RetryIsErr(func(err error) bool {
+				return err == io.EOF
+			}),
+		),
+	)
+	retryFn = retryWithLogging(logger, retryFn)
+
+	return rehttp.NewTransport(rt,
+		retryFn,
+		rehttp.ExpJitterDelay(
+			time.Duration(2)*time.Second,
+			time.Duration(cfg.GwMaxBackoff())*time.Millisecond,
+		),
+	)
+}
+
 func (impl) NewClient(ctx context.Context, cfg conf.Config) (Client, error) {
 	cert, err := tls.LoadX509KeyPair(cfg.GwCert(), cfg.GwKey())
 	if err != nil {
@@ -354,7 +414,15 @@ func (impl) NewClient(ctx context.Context, cfg conf.Config) (Client, error) {
 			Certificates: []tls.Certificate{cert},
 		},
 	}
-	out.httpClient = &http.Client{Transport: &transport}
+
+	// This client is passed into AWS SDK and it should not add any
+	// retry logic because the AWS SDK already does that:
+	s3HttpClient := &http.Client{Transport: &transport}
+
+	// This client is used outside of the AWS SDK (i.e. for requests
+	// to "publish" API) and it should wrap the transport to enable
+	// retries for certain types of error.
+	out.httpClient = &http.Client{Transport: retryTransport(ctx, cfg, &transport)}
 
 	awsLogLevel := aws.LogOff
 	if cfg.Verbosity() > 2 || cfg.LogLevel() == "trace" {
@@ -368,7 +436,7 @@ func (impl) NewClient(ctx context.Context, cfg conf.Config) (Client, error) {
 			S3ForcePathStyle: aws.Bool(true),
 			Region:           aws.String("us-east-1"),
 			Credentials:      credentials.AnonymousCredentials,
-			HTTPClient:       out.httpClient,
+			HTTPClient:       s3HttpClient,
 			Logger:           log.FromContext(ctx),
 			LogLevel:         aws.LogLevel(awsLogLevel),
 		},

internal/gw/client_publish_errors_test.go

@@ -119,6 +119,28 @@ func TestClientPublishErrors(t *testing.T) {
 		}
 	})
 
+	t.Run("can recover by retrying", func(t *testing.T) {
+		// Force an EOF, then a 500 error, and finally a successful
+		// response. The caller should only see the success.
+		gw.nextHTTPError = io.EOF
+		gw.nextHTTPResponse = &http.Response{
+			Status:     "500 Internal Server Error",
+			StatusCode: 500,
+			Body:       io.NopCloser(strings.NewReader("some error")),
+		}
+
+		gw.publishes["some-id"] = &fakePublish{id: "some-id"}
+
+		p, err := clientIface.GetPublish(ctx, "some-id")
+		if err != nil {
+			t.Errorf("failed to get publish, err = %v", err)
+		}
+		id := p.ID()
+		if id != "some-id" {
+			t.Errorf("got unexpected id %s", id)
+		}
+	})
+
 	t.Run("missing link for commit", func(t *testing.T) {
 		// Create a publish object directly without filling in any Links.
 		publish := publish{client: clientIface.(*client)}