Merge pull request #236 from yuce/233-request-retrials

Initial retry support

Author: yuce

CHANGELOG.md

@@ -1,5 +1,8 @@
 # Change Log
 
+* **next**
+    * The client retries failed HTTP requests, by default 2 times. Use `OptClientRetries` to set the number of retries.
+
 * **v1.3.0** (2019-04-19)
     * **Compatible with Pilosa 1.2 and 1.3**
     * Deprecated `QueryResponse.Columns` function. Use `QueryResponse.ColumnAttrs` function instead.

client.go

@@ -82,6 +82,10 @@ type Client struct {
 	manualFragmentNode *fragmentNode
 	manualServerURI    *URI
 	tracer             opentracing.Tracer
+	// Number of retries if an HTTP request fails
+	retries           int
+	minRetrySleepTime time.Duration
+	maxRetrySleepTime time.Duration
 
 	importLogEncoder encoder
 	logLock          sync.Mutex
@@ -143,6 +147,9 @@ func newClientWithOptions(options *ClientOptions) *Client {
 	} else {
 		c.tracer = options.tracer
 	}
+	c.retries = *options.retries
+	c.minRetrySleepTime = 1 * time.Second
+	c.maxRetrySleepTime = 2 * time.Minute
 	c.importManager = newRecordImportManager(c)
 	return c
 
@@ -591,7 +598,11 @@ func (c *Client) hasRoaringImportSupport(field *Field) bool {
 	}
 	// Check whether the roaring import endpoint exists
 	path := makeRoaringImportPath(field, 0, url.Values{})
+	// err may contain an HTTP error, but we don't use it.
 	resp, _, _ := c.httpRequest("GET", path, nil, nil, false)
+	if resp == nil {
+		return false
+	}
 	if resp.StatusCode == http.StatusMethodNotAllowed || resp.StatusCode == http.StatusOK {
 		// Roaring import endpoint exists
 		return true
@@ -680,7 +691,7 @@ func (c *Client) fetchCoordinatorNode() (fragmentNode, error) {
 }
 
 func (c *Client) importData(uri *URI, path string, data []byte) error {
-	resp, err := c.doRequest(uri, "POST", path, defaultProtobufHeaders(), bytes.NewReader(data))
+	resp, err := c.doRequest(uri, "POST", path, defaultProtobufHeaders(), data)
 	if err = anyError(resp, err); err != nil {
 		return errors.Wrap(err, "doing import")
 	}
@@ -716,7 +727,7 @@ func (c *Client) importRoaringBitmap(uri *URI, field *Field, shard uint64, views
 
 	c.logImport(field.index.Name(), path, shard, true, data)
 
-	resp, err := c.doRequest(uri, "POST", path, defaultProtobufHeaders(), bytes.NewReader(data))
+	resp, err := c.doRequest(uri, "POST", path, defaultProtobufHeaders(), data)
 	if err = anyError(resp, err); err != nil {
 		return errors.Wrap(err, "doing import")
 	}
@@ -835,7 +846,7 @@ func (c *Client) httpRequest(method string, path string, data []byte, headers ma
 		if err != nil {
 			return nil, nil, err
 		}
-		response, err = c.doRequest(host, method, path, c.augmentHeaders(headers), bytes.NewReader(data))
+		response, err = c.doRequest(host, method, path, c.augmentHeaders(headers), data)
 		if err == nil {
 			break
 		}
@@ -925,12 +936,45 @@ func anyError(resp *http.Response, err error) error {
 }
 
 // doRequest creates and performs an http request.
-func (c *Client) doRequest(host *URI, method, path string, headers map[string]string, reader io.Reader) (*http.Response, error) {
-	req, err := makeRequest(host, method, path, headers, reader)
-	if err != nil {
-		return nil, errors.Wrap(err, "building request")
+func (c *Client) doRequest(host *URI, method, path string, headers map[string]string, data []byte) (*http.Response, error) {
+	var resp *http.Response
+	var err error
+	var req *http.Request
+	var content []byte
+
+	tries := 1 + c.retries
+	sleepTime := c.minRetrySleepTime
+	for tries > 0 {
+		req, err = makeRequest(host, method, path, headers, data)
+		if err != nil {
+			return nil, errors.Wrap(err, "building request")
+		}
+		tries--
+		resp, err = c.client.Do(req)
+		if err == nil {
+			if resp.StatusCode >= 200 && resp.StatusCode < 300 {
+				return resp, nil
+			}
+			// Pilosa nodes sometimes return 400, we retry in that case.
+			// No need to retry in other 4xx cases.
+			if resp.StatusCode > 400 && resp.StatusCode < 500 {
+				return resp, nil
+			}
+			content, err = ioutil.ReadAll(resp.Body)
+			resp.Body.Close()
+			if err != nil {
+				return nil, err
+			}
+			err = errors.New(strings.TrimSpace(string(content)))
+		}
+		c.logger.Printf("request failed with: %s, retrying (%d)", err.Error(), tries)
+		time.Sleep(sleepTime)
+		sleepTime *= 2
+		if sleepTime > c.maxRetrySleepTime {
+			sleepTime = c.maxRetrySleepTime
+		}
 	}
-	return c.client.Do(req)
+	return nil, errors.Wrap(err, "max retries exceeded")
 }
 
 // statusToNodeShardsForIndex finds the hosts which contains shards for the given index
@@ -1058,7 +1102,7 @@ func (c *Client) ExperimentalReplayImport(r io.Reader, concurrency int) error {
 
 				if !log.IsRoaring {
 					for _, node := range nodes {
-						resp, err := c.doRequest(node.URI(), "POST", log.Path, defaultProtobufHeaders(), bytes.NewReader(log.Data))
+						resp, err := c.doRequest(node.URI(), "POST", log.Path, defaultProtobufHeaders(), log.Data)
 						if err = anyError(resp, err); err != nil {
 							return errors.Wrap(err, "doing import")
 						}
@@ -1067,7 +1111,7 @@ func (c *Client) ExperimentalReplayImport(r io.Reader, concurrency int) error {
 				} else {
 					// import-roaring forwards on to all replicas, so we only import to
 					// one node.
-					resp, err := c.doRequest(nodes[0].URI(), "POST", log.Path, defaultProtobufHeaders(), bytes.NewReader(log.Data))
+					resp, err := c.doRequest(nodes[0].URI(), "POST", log.Path, defaultProtobufHeaders(), log.Data)
 					if err = anyError(resp, err); err != nil {
 						return errors.Wrap(err, "doing import")
 					}
@@ -1121,8 +1165,8 @@ func defaultProtobufHeaders() map[string]string {
 	}
 }
 
-func makeRequest(host *URI, method, path string, headers map[string]string, reader io.Reader) (*http.Request, error) {
-	request, err := http.NewRequest(method, host.Normalize()+path, reader)
+func makeRequest(host *URI, method, path string, headers map[string]string, data []byte) (*http.Request, error) {
+	request, err := http.NewRequest(method, host.Normalize()+path, bytes.NewReader(data))
 	if err != nil {
 		return nil, err
 	}
@@ -1326,6 +1370,7 @@ type ClientOptions struct {
 	TLSConfig           *tls.Config
 	manualServerAddress bool
 	tracer              opentracing.Tracer
+	retries             *int
 
 	importLogWriter io.Writer
 }
@@ -1410,6 +1455,17 @@ func OptClientTracer(tracer opentracing.Tracer) ClientOption {
 	}
 }
 
+// OptClientRetries sets the number of retries on HTTP request failures.
+func OptClientRetries(retries int) ClientOption {
+	return func(options *ClientOptions) error {
+		if retries < 0 {
+			return errors.New("retries must be non-negative")
+		}
+		options.retries = &retries
+		return nil
+	}
+}
+
 type versionInfo struct {
 	Version string `json:"version"`
 }
@@ -1434,6 +1490,10 @@ func (co *ClientOptions) withDefaults() (updated *ClientOptions) {
 	if updated.TLSConfig == nil {
 		updated.TLSConfig = &tls.Config{}
 	}
+	if updated.retries == nil {
+		retries := 2
+		updated.retries = &retries
+	}
 	return
 }