Merge pull request #88 from hyperledger/retry-ext

Add extra HTTP client / retry options
hyperledger · Jul 17, 2023 · cea2846 · cea2846
2 parents da6668c + 223178f
commit cea2846
Show file tree

Hide file tree

Showing 4 changed files with 144 additions and 20 deletions.
diff --git a/pkg/ffresty/config.go b/pkg/ffresty/config.go
@@ -31,6 +31,7 @@ const (
 	defaultRequestTimeout                = "30s"
 	defaultHTTPIdleTimeout               = "475ms" // Node.js default keepAliveTimeout is 5 seconds, so we have to set a base below this
 	defaultHTTPMaxIdleConns              = 100     // match Go's default
+	defaultHTTPMaxConnsPerHost           = 0
 	defaultHTTPConnectionTimeout         = "30s"
 	defaultHTTPTLSHandshakeTimeout       = "10s" // match Go's default
 	defaultHTTPExpectContinueTimeout     = "1s"  // match Go's default
@@ -62,6 +63,8 @@ const (
 	HTTPIdleTimeout = "idleTimeout"
 	// HTTPMaxIdleConns the max number of idle connections to hold pooled
 	HTTPMaxIdleConns = "maxIdleConns"
+	// HTTPMaxConnsPerHost the max number of concurrent connections
+	HTTPMaxConnsPerHost = "maxConnsPerHost"
 	// HTTPConnectionTimeout the connection timeout for new connections
 	HTTPConnectionTimeout = "connectionTimeout"
 	// HTTPTLSHandshakeTimeout the TLS handshake connection timeout
@@ -88,6 +91,7 @@ func InitConfig(conf config.Section) {
 	conf.AddKnownKey(HTTPConfigRequestTimeout, defaultRequestTimeout)
 	conf.AddKnownKey(HTTPIdleTimeout, defaultHTTPIdleTimeout)
 	conf.AddKnownKey(HTTPMaxIdleConns, defaultHTTPMaxIdleConns)
+	conf.AddKnownKey(HTTPMaxConnsPerHost, defaultHTTPMaxConnsPerHost)
 	conf.AddKnownKey(HTTPConnectionTimeout, defaultHTTPConnectionTimeout)
 	conf.AddKnownKey(HTTPTLSHandshakeTimeout, defaultHTTPTLSHandshakeTimeout)
 	conf.AddKnownKey(HTTPExpectContinueTimeout, defaultHTTPExpectContinueTimeout)

diff --git a/pkg/ffresty/ffresty.go b/pkg/ffresty/ffresty.go
@@ -45,25 +45,28 @@ type retryCtx struct {
 }
 
 type Config struct {
-	URL                           string             `json:"httpURL,omitempty"`
-	ProxyURL                      string             `json:"proxyURL,omitempty"`
-	HTTPRequestTimeout            time.Duration      `json:"requestTimeout,omitempty"`
-	HTTPIdleConnTimeout           time.Duration      `json:"idleTimeout,omitempty"`
-	HTTPMaxIdleTimeout            time.Duration      `json:"maxIdleTimeout,omitempty"`
-	HTTPConnectionTimeout         time.Duration      `json:"connectionTimeout,omitempty"`
-	HTTPExpectContinueTimeout     time.Duration      `json:"expectContinueTimeout,omitempty"`
-	AuthUsername                  string             `json:"authUsername,omitempty"`
-	AuthPassword                  string             `json:"authPassword,omitempty"`
-	Retry                         bool               `json:"retry,omitempty"`
-	RetryCount                    int                `json:"retryCount,omitempty"`
-	RetryInitialDelay             time.Duration      `json:"retryInitialDelay,omitempty"`
-	RetryMaximumDelay             time.Duration      `json:"retryMaximumDelay,omitempty"`
-	HTTPMaxIdleConns              int                `json:"maxIdleConns,omitempty"`
-	HTTPPassthroughHeadersEnabled bool               `json:"httpPassthroughHeadersEnabled,omitempty"`
-	HTTPHeaders                   fftypes.JSONObject `json:"headers,omitempty"`
-	TLSClientConfig               *tls.Config        `json:"tlsClientConfig,omitempty"`
-	HTTPTLSHandshakeTimeout       time.Duration      `json:"tlsHandshakeTimeout,omitempty"`
-	HTTPCustomClient              interface{}        `json:"httpCustomClient,omitempty"`
+	URL                           string                                    `json:"httpURL,omitempty"`
+	ProxyURL                      string                                    `json:"proxyURL,omitempty"`
+	HTTPRequestTimeout            time.Duration                             `json:"requestTimeout,omitempty"`
+	HTTPIdleConnTimeout           time.Duration                             `json:"idleTimeout,omitempty"`
+	HTTPMaxIdleTimeout            time.Duration                             `json:"maxIdleTimeout,omitempty"`
+	HTTPConnectionTimeout         time.Duration                             `json:"connectionTimeout,omitempty"`
+	HTTPExpectContinueTimeout     time.Duration                             `json:"expectContinueTimeout,omitempty"`
+	AuthUsername                  string                                    `json:"authUsername,omitempty"`
+	AuthPassword                  string                                    `json:"authPassword,omitempty"`
+	Retry                         bool                                      `json:"retry,omitempty"`
+	RetryCount                    int                                       `json:"retryCount,omitempty"`
+	RetryInitialDelay             time.Duration                             `json:"retryInitialDelay,omitempty"`
+	RetryMaximumDelay             time.Duration                             `json:"retryMaximumDelay,omitempty"`
+	HTTPMaxIdleConns              int                                       `json:"maxIdleConns,omitempty"`
+	HTTPMaxConnsPerHost           int                                       `json:"maxConnsPerHost,omitempty"`
+	HTTPPassthroughHeadersEnabled bool                                      `json:"httpPassthroughHeadersEnabled,omitempty"`
+	HTTPHeaders                   fftypes.JSONObject                        `json:"headers,omitempty"`
+	TLSClientConfig               *tls.Config                               `json:"tlsClientConfig,omitempty"`
+	HTTPTLSHandshakeTimeout       time.Duration                             `json:"tlsHandshakeTimeout,omitempty"`
+	HTTPCustomClient              interface{}                               `json:"httpCustomClient,omitempty"`
+	OnCheckRetry                  func(res *resty.Response, err error) bool `json:"-"` // response could be nil on err
+	OnBeforeRequest               func(req *resty.Request) error            `json:"-"` // called before each request, even retry
 }
 
 // OnAfterResponse when using SetDoNotParseResponse(true) for streaming binary replies,
@@ -121,6 +124,7 @@ func NewWithConfig(ctx context.Context, ffrestyConfig Config) (client *resty.Cli
 			}).DialContext,
 			ForceAttemptHTTP2:     true,
 			MaxIdleConns:          ffrestyConfig.HTTPMaxIdleConns,
+			MaxConnsPerHost:       ffrestyConfig.HTTPMaxConnsPerHost,
 			IdleConnTimeout:       ffrestyConfig.HTTPIdleConnTimeout,
 			TLSHandshakeTimeout:   ffrestyConfig.HTTPTLSHandshakeTimeout,
 			ExpectContinueTimeout: ffrestyConfig.HTTPExpectContinueTimeout,
@@ -181,6 +185,12 @@ func NewWithConfig(ctx context.Context, ffrestyConfig Config) (client *resty.Cli
 			req.Header.Set(ffapi.FFRequestIDHeader, ffRequestID.(string))
 		}
 
+		if ffrestyConfig.OnBeforeRequest != nil {
+			if err := ffrestyConfig.OnBeforeRequest(req); err != nil {
+				return err
+			}
+		}
+
 		log.L(rCtx).Debugf("==> %s %s%s", req.Method, url, req.URL)
 		return nil
 	})
@@ -212,8 +222,12 @@ func NewWithConfig(ctx context.Context, ffrestyConfig Config) (client *resty.Cli
 				}
 				rCtx := r.Request.Context()
 				rc := rCtx.Value(retryCtxKey{}).(*retryCtx)
-				log.L(rCtx).Infof("retry %d/%d (min=%dms/max=%dms) status=%d", rc.attempts, retryCount, minTimeout.Milliseconds(), maxTimeout.Milliseconds(), r.StatusCode())
+				if ffrestyConfig.OnCheckRetry != nil && !ffrestyConfig.OnCheckRetry(r, err) {
+					log.L(rCtx).Debugf("retry cancelled after %d attempts", rc.attempts)
+					return false
+				}
 				rc.attempts++
+				log.L(rCtx).Infof("retry %d/%d (min=%dms/max=%dms) status=%d", rc.attempts, retryCount, minTimeout.Milliseconds(), maxTimeout.Milliseconds(), r.StatusCode())
 				return true
 			})
 	}

diff --git a/pkg/ffresty/ffresty_test.go b/pkg/ffresty/ffresty_test.go
@@ -35,6 +35,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/go-resty/resty/v2"
 	"github.com/hyperledger/firefly-common/pkg/config"
 	"github.com/hyperledger/firefly-common/pkg/ffapi"
 	"github.com/hyperledger/firefly-common/pkg/fftls"
@@ -177,6 +178,110 @@ func TestErrResponse(t *testing.T) {
 	assert.Error(t, err)
 }
 
+func TestErrResponseBlockRetry(t *testing.T) {
+
+	ctx := context.Background()
+
+	resetConf()
+	utConf.Set(HTTPConfigURL, "http://localhost:12345")
+	utConf.Set(HTTPConfigRetryEnabled, true)
+	utConf.Set(HTTPConfigRetryInitDelay, "1ms")
+	utConf.Set(HTTPConfigRetryCount, 10000)
+
+	ffrestyConfig, err := GenerateConfig(ctx, utConf)
+	assert.NoError(t, err)
+	called := make(chan bool, 1)
+	ffrestyConfig.OnCheckRetry = func(res *resty.Response, err error) bool {
+		assert.NotNil(t, res) // We expect a response object, even though it was an error
+		assert.NotNil(t, err)
+		called <- true
+		return false
+	}
+
+	c := NewWithConfig(ctx, *ffrestyConfig)
+	httpmock.ActivateNonDefault(c.GetClient())
+	defer httpmock.DeactivateAndReset()
+
+	resText := strings.Builder{}
+	for i := 0; i < 512; i++ {
+		resText.WriteByte(byte('a' + (i % 26)))
+	}
+	httpmock.RegisterResponder("POST", "http://localhost:12345/test",
+		httpmock.NewErrorResponder(fmt.Errorf("pop")))
+
+	resp, err := c.R().SetBody("stuff").Post("/test")
+	err = WrapRestErr(ctx, resp, err, i18n.MsgConfigFailed)
+	assert.Error(t, err)
+
+	<-called
+}
+
+func TestRetryReGenHeaderOnEachRequest(t *testing.T) {
+
+	ctx := context.Background()
+
+	resetConf()
+	utConf.Set(HTTPConfigURL, "http://localhost:12345")
+	utConf.Set(HTTPConfigRetryEnabled, true)
+	utConf.Set(HTTPConfigRetryInitDelay, "1ms")
+	utConf.Set(HTTPConfigRetryCount, 1)
+
+	ffrestyConfig, err := GenerateConfig(ctx, utConf)
+	assert.NoError(t, err)
+	reqCount := 0
+	ffrestyConfig.OnBeforeRequest = func(req *resty.Request) error {
+		reqCount++
+		req.Header.Set("utheader", fmt.Sprintf("request_%d", reqCount))
+		return nil
+	}
+
+	c := NewWithConfig(ctx, *ffrestyConfig)
+	httpmock.ActivateNonDefault(c.GetClient())
+	defer httpmock.DeactivateAndReset()
+
+	resText := strings.Builder{}
+	for i := 0; i < 512; i++ {
+		resText.WriteByte(byte('a' + (i % 26)))
+	}
+	utHeaders := make(chan string, 2)
+	httpmock.RegisterResponder("POST", "http://localhost:12345/test", func(req *http.Request) (*http.Response, error) {
+		utHeaders <- req.Header.Get("utheader")
+		return &http.Response{StatusCode: 500, Body: http.NoBody}, nil
+	})
+
+	resp, err := c.R().SetBody("stuff").Post("/test")
+	err = WrapRestErr(ctx, resp, err, i18n.MsgConfigFailed)
+	assert.Error(t, err)
+
+	assert.Equal(t, "request_1", <-utHeaders)
+	assert.Equal(t, "request_2", <-utHeaders)
+}
+
+func TestRetryReGenHeaderOnEachRequestFail(t *testing.T) {
+
+	ctx := context.Background()
+
+	resetConf()
+	utConf.Set(HTTPConfigURL, "http://localhost:12345")
+	utConf.Set(HTTPConfigRetryEnabled, true)
+	utConf.Set(HTTPConfigRetryInitDelay, "1ms")
+	utConf.Set(HTTPConfigRetryCount, 1)
+
+	ffrestyConfig, err := GenerateConfig(ctx, utConf)
+	assert.NoError(t, err)
+	ffrestyConfig.OnBeforeRequest = func(req *resty.Request) error {
+		return fmt.Errorf("pop")
+	}
+
+	c := NewWithConfig(ctx, *ffrestyConfig)
+	httpmock.ActivateNonDefault(c.GetClient())
+	defer httpmock.DeactivateAndReset()
+
+	resp, err := c.R().SetBody("stuff").Post("/test")
+	err = WrapRestErr(ctx, resp, err, i18n.MsgConfigFailed)
+	assert.Regexp(t, "pop", err)
+}
+
 func TestOnAfterResponseNil(t *testing.T) {
 	OnAfterResponse(nil, nil)
 }

diff --git a/pkg/i18n/en_base_config_descriptions.go b/pkg/i18n/en_base_config_descriptions.go
@@ -79,6 +79,7 @@ var (
 	ConfigGlobalHeaders                   = ffc("config.global.headers", "Adds custom headers to HTTP requests", MapStringStringType)
 	ConfigGlobalIdleTimeout               = ffc("config.global.idleTimeout", "The max duration to hold a HTTP keepalive connection between calls", TimeDurationType)
 	ConfigGlobalMaxIdleConns              = ffc("config.global.maxIdleConns", "The max number of idle connections to hold pooled", IntType)
+	ConfigGlobalMaxConnsPerHost           = ffc("config.global.maxConnsPerHost", "The max number of connections, per unique hostname. Zero means no limit", IntType)
 	ConfigGlobalMethod                    = ffc("config.global.method", "The HTTP method to use when making requests to the Address Resolver", StringType)
 	ConfigGlobalAuthType                  = ffc("config.global.auth.type", "The auth plugin to use for server side authentication of requests", StringType)
 	ConfigGlobalPassthroughHeadersEnabled = ffc("config.global.passthroughHeadersEnabled", "Enable passing through the set of allowed HTTP request headers", BooleanType)