Repository: influxdata/influxdb-client-go (go)

Issue:
Add Retry Implementation of writer.BucketMetricWriter
The root `influxdb.Client` implements the `writer.BucketMetricWriter` interface (https://github.com/influxdata/influxdb-client-go/blob/develop/writer/writer.go#L11).
A new decorating implementation is required which handles automatically retrying on error conditions.

This was previously implemented in the `influxdb.Client`, however, it was removed in the buffered writer refactor. This should now be reinstated as a decorator for the `writer.BucketMetricWriter` interface.

Pull request description:
writer.RetryWriter implementation
closes #46

This reinstates the ability to automatically retry writes based on known transient error conditions.

See updates to docs.go:
```go
import (
	"time"

	"github.com/influxdata/influxdb-client-go"
	"github.com/influxdata/influxdb-client-go/writer"
)

func main() {
	var (
		cli, _ = influxdb.New("http://localhost:9999", "some-token")
		bucket = "default"
		org    = "influx"
	)

	// construct a writer with 3 maximum attempts per call to Write and linear backoff derived from number of attempts
	// i.e. a first attempt is followed by a 1 second delay before the second attempt
	// a second attempt is followed by a 2 second delay before the third attempt
	var (
		retryOpts = []writer.RetryOption{writer.WithMaxAttempts(3), writer.WithBackoff(writer.LinearBackoff(time.Second))}
		wr        = writer.New(cli, bucket, org, writer.WithRetries(retryOpts...))
	)
}
```

Namely:
- unavailable
- too many requests (rate limiting)

TODO:

- [x] implement basic retry behavior with max attempts
- [x] implement backoff which respects Retry-After response header
- [x] implement backoff which is configured via a backoff function
- [x] configure and wire retry writer into `writer.New()` response type
- [x] more tests around `write.go` handling of error parsing

Patch:
diff --git a/client.go b/client.go
index 33078b1..5ab66d7 100644
--- a/client.go
+++ b/client.go
@@ -4,13 +4,10 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"io"
 	"io/ioutil"
 	"net/http"
 	"net/url"
 	"sync"
-
-	"github.com/influxdata/influxdb-client-go/internal/gzip"
 )
 
 // TODO(docmerlin): change the generator so we don't have to hand edit the generated code
@@ -44,11 +41,13 @@ func New(connection string, token string, options ...Option) (*Client, error) {
 	if connection == "" {
 		connection = `http://127.0.0.1:9999`
 	}
+
 	u, err := url.Parse(connection)
-	u.Path = `/api/v2`
 	if err != nil {
 		return nil, fmt.Errorf("Error: could not parse url: %v", err)
 	}
+	u.Path = `/api/v2`
+
 	c.url = u
 
 	c.userAgent = userAgent()
@@ -105,35 +104,6 @@ func (c *Client) Ping(ctx context.Context) error {
 	return nil
 }
 
-func (c *Client) makeWriteRequest(bucket, org string, body io.Reader) (*http.Request, error) {
-	var err error
-	if c.contentEncoding == "gzip" {
-		body, err = gzip.CompressWithGzip(body, c.compressionLevel)
-		if err != nil {
-			return nil, err
-		}
-	}
-	u, err := makeWriteURL(c.url, bucket, org)
-	if err != nil {
-		return nil, err
-	}
-	req, err := http.NewRequest(http.MethodPost, u, body)
-	if err != nil {
-		return nil, err
-	}
-
-	req.Header.Set("Content-Type", "text/plain; charset=utf-8")
-
-	if c.contentEncoding == "gzip" {
-		req.Header.Set("Content-Encoding", "gzip")
-	}
-
-	req.Header.Set("User-Agent", c.userAgent)
-	req.Header.Set("Authorization", c.authorization)
-
-	return req, nil
-}
-
 // Close closes any idle connections on the Client.
 func (c *Client) Close() error {
 	c.httpClient.CloseIdleConnections()
diff --git a/errors.go b/errors.go
index 96cf9bf..100e986 100644
--- a/errors.go
+++ b/errors.go
@@ -5,22 +5,47 @@ import (
 	"fmt"
 )
 
+// Error code constants copied from influxdb
+const (
+	EInternal            = "internal error"
+	ENotFound            = "not found"
+	EConflict            = "conflict"             // action cannot be performed
+	EInvalid             = "invalid"              // validation failed
+	EUnprocessableEntity = "unprocessable entity" // data type is correct, but out of range
+	EEmptyValue          = "empty value"
+	EUnavailable         = "unavailable"
+	EForbidden           = "forbidden"
+	ETooManyRequests     = "too many requests"
+	EUnauthorized        = "unauthorized"
+	EMethodNotAllowed    = "method not allowed"
+)
+
 // ErrUnimplemented is an error for when pieces of the client's functionality is unimplemented.
 var ErrUnimplemented = errors.New("unimplemented")
 
-type genericRespError struct {
-	Code      string
-	Message   string
-	Line      *int32
-	MaxLength *int32
+// Error is an error returned by a client operation
+// It contains a number of contextual fields which describe the nature
+// and cause of the error
+type Error struct {
+	Code       string
+	Message    string
+	Err        string
+	Op         string
+	Line       *int32
+	MaxLength  *int32
+	RetryAfter *int32
 }
 
-func (g genericRespError) Error() string {
-	errString := g.Code + ": " + g.Message
-	if g.Line != nil {
-		return fmt.Sprintf("%s - line[%d]", errString, g.Line)
-	} else if g.MaxLength != nil {
-		return fmt.Sprintf("%s - maxlen[%d]", errString, g.MaxLength)
+// Error returns the string representation of the Error struct
+func (e *Error) Error() string {
+	errString := fmt.Sprintf("%s (%s): %s", e.Code, e.Op, e.Message)
+	if e.Line != nil {
+		return fmt.Sprintf("%s - line[%d]", errString, e.Line)
 	}
+
+	if e.MaxLength != nil {
+		return fmt.Sprintf("%s - maxlen[%d]", errString, e.MaxLength)
+	}
+
 	return errString
 }
diff --git a/query.go b/query.go
index 7431ca4..b5da5b1 100644
--- a/query.go
+++ b/query.go
@@ -91,9 +91,9 @@ func (c *Client) QueryCSV(ctx context.Context, flux string, org string, extern .
 
 	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
 		r := io.LimitReader(resp.Body, 1<<14) // only support errors that are 16kB long, more than that and something is probably wrong.
-		gerr := &genericRespError{Code: resp.Status}
+		gerr := &Error{Code: resp.Status}
 		if resp.ContentLength != 0 {
-			if err := json.NewDecoder(r).Decode(gerr); err != nil {
+			if err := json.NewDecoder(r).Decode(&gerr); err != nil {
 				gerr.Code = resp.Status
 				message, err := ioutil.ReadAll(r)
 				if err != nil {
diff --git a/write.go b/write.go
index 5bee93c..2d0970a 100644
--- a/write.go
+++ b/write.go
@@ -8,10 +8,13 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
+	"mime"
 	"net/http"
 	"net/url"
 	"path"
+	"strconv"
 
+	"github.com/influxdata/influxdb-client-go/internal/gzip"
 	lp "github.com/influxdata/line-protocol"
 )
 
@@ -53,41 +56,72 @@ func (c *Client) Write(ctx context.Context, bucket, org string, m ...Metric) (n
 		resp.Body.Close()
 	}()
 
-	switch resp.StatusCode {
-	case http.StatusOK, http.StatusNoContent:
-	case http.StatusTooManyRequests:
-		err = &genericRespError{
-			Code:    resp.Status,
-			Message: "too many requests too fast",
-		}
-	case http.StatusServiceUnavailable:
-		err = &genericRespError{
-			Code:    resp.Status,
-			Message: "service temporarily unavaliable",
-		}
-	default:
-		gwerr, err := parseWriteError(resp.Body)
+	eerr, err := parseWriteError(resp)
+	if err != nil {
+		return 0, err
+	}
+
+	if eerr != nil {
+		return 0, eerr
+	}
+
+	return len(m), nil
+}
+
+func parseInt32(v string) (int32, error) {
+	retry, err := strconv.ParseInt(v, 10, 32)
+	if err != nil {
+		return 0, err
+	}
+
+	return int32(retry), nil
+}
+
+func (c *Client) makeWriteRequest(bucket, org string, body io.Reader) (*http.Request, error) {
+	var err error
+	if c.contentEncoding == "gzip" {
+		body, err = gzip.CompressWithGzip(body, c.compressionLevel)
 		if err != nil {
-			return 0, err
+			return nil, err
 		}
+	}
+
+	u, err := makeWriteURL(c.url, bucket, org)
+	if err != nil {
+		return nil, err
+	}
 
-		return 0, gwerr
+	req, err := http.NewRequest(http.MethodPost, u, body)
+	if err != nil {
+		return nil, err
 	}
 
-	return len(m), err
+	req.Header.Set("Content-Type", "text/plain; charset=utf-8")
+
+	if c.contentEncoding == "gzip" {
+		req.Header.Set("Content-Encoding", "gzip")
+	}
+
+	req.Header.Set("User-Agent", c.userAgent)
+	req.Header.Set("Authorization", c.authorization)
+
+	return req, nil
 }
 
 func makeWriteURL(loc *url.URL, bucket, org string) (string, error) {
 	if loc == nil {
 		return "", errors.New("nil url")
 	}
+
 	u, err := url.Parse(loc.String())
 	if err != nil {
 		return "", err
 	}
+
 	params := url.Values{}
 	params.Set("bucket", bucket)
 	params.Set("org", org)
+	u.RawQuery = params.Encode()
 
 	switch loc.Scheme {
 	case "http", "https":
@@ -96,14 +130,48 @@ func makeWriteURL(loc *url.URL, bucket, org string) (string, error) {
 	default:
 		return "", fmt.Errorf("unsupported scheme: %q", u.Scheme)
 	}
-	u.RawQuery = params.Encode()
+
 	return u.String(), nil
 }
 
-func parseWriteError(r io.Reader) (*genericRespError, error) {
-	werr := &genericRespError{}
-	if err := json.NewDecoder(r).Decode(&werr); err != nil {
-		return nil, err
+func parseWriteError(r *http.Response) (err *Error, perr error) {
+	// successful status code range
+	if r.StatusCode >= 200 && r.StatusCode < 300 {
+		return nil, nil
 	}
-	return werr, nil
+
+	err = &Error{}
+	if v := r.Header.Get("Retry-After"); v != "" {
+		if retry, perr := parseInt32(v); perr == nil {
+			err.RetryAfter = &retry
+		}
+	}
+
+	switch r.StatusCode {
+	case http.StatusTooManyRequests:
+		err.Code = ETooManyRequests
+		err.Message = "exceeded rate limit"
+	case http.StatusServiceUnavailable:
+		err.Code = EUnavailable
+		err.Message = "service temporarily unavailable"
+	default:
+		// json encoded error
+		typ, _, _ := mime.ParseMediaType(r.Header.Get("Content-Type"))
+		if typ == "application/json" {
+			perr = json.NewDecoder(r.Body).Decode(err)
+			return
+		}
+
+		// plain error body type
+		var body []byte
+		body, perr = ioutil.ReadAll(r.Body)
+		if perr != nil {
+			return
+		}
+
+		err.Code = r.Status
+		err.Message = string(body)
+	}
+
+	return
 }
diff --git a/writer/doc.go b/writer/doc.go
index 514bc6d..7502c30 100644
--- a/writer/doc.go
+++ b/writer/doc.go
@@ -61,4 +61,32 @@
 //
 // Finally, it wraps the buffered writer in a *PointsWriter which takes care of ensuring Flush is called
 // automatically when it hasn't been called for a configured duration. This final type is safe for concurrent use.
+//
+// Automatic Retries
+//
+// The writer package offers automatic retry capabilities during known transient failures
+// This is when the API being consumed reports "unavailable" or "too many requests" error conditions
+//
+// import (
+// 	"time"
+//
+// 	"github.com/influxdata/influxdb-client-go"
+// 	"github.com/influxdata/influxdb-client-go/writer"
+// )
+//
+// func main() {
+// 	var (
+// 		cli, _ = influxdb.New("http://localhost:9999", "some-token")
+// 		bucket = "default"
+// 		org    = "influx"
+// 	)
+//
+// 	// construct a writer with 3 maximum attempts per call to Write and linear backoff derived from number of attempts
+// 	// i.e. a first attempt is followed by a 1 second delay before the second attempt
+// 	// a second attempt is followed by a 2 second delay before the third attempt
+// 	var (
+// 		retryOpts = []writer.RetryOption{writer.WithMaxAttempts(3), writer.WithBackoff(writer.LinearBackoff(time.Second))}
+// 		wr        = writer.New(cli, bucket, org, writer.WithRetries(retryOpts...))
+// 	)
+// }
 package writer
diff --git a/writer/options.go b/writer/options.go
index f9b317b..21b51f2 100644
--- a/writer/options.go
+++ b/writer/options.go
@@ -6,6 +6,8 @@ import "time"
 type Config struct {
 	size          int
 	flushInterval time.Duration
+	retry         bool
+	retryOptions  []RetryOption
 }
 
 // Option is a functional option for Configuring point writers
@@ -49,3 +51,13 @@ func WithFlushInterval(interval time.Duration) Option {
 		c.flushInterval = interval
 	}
 }
+
+// WithRetries configures automatic retry behavior on specific
+// transient error conditions when attempting to Write metrics
+// to a client
+func WithRetries(options ...RetryOption) Option {
+	return func(c *Config) {
+		c.retry = true
+		c.retryOptions = options
+	}
+}
diff --git a/writer/retry.go b/writer/retry.go
new file mode 100644
index 0000000..646134b
--- /dev/null
+++ b/writer/retry.go
@@ -0,0 +1,108 @@
+package writer
+
+import (
+	"time"
+
+	"github.com/influxdata/influxdb-client-go"
+)
+
+const defaultMaxAttempts = 5
+
+// BackoffFunc is a function which when called with an
+// attempt number returns a duration which should be
+// waited for until a subsequent attempt is made
+type BackoffFunc func(attempt int) time.Duration
+
+// RetryWriter is a metrics writers which decorates other
+// metrics writer implementations and automatically retries
+// attempts to write metrics under certain error conditions
+type RetryWriter struct {
+	MetricsWriter
+
+	sleep   func(time.Duration)
+	backoff BackoffFunc
+
+	maxAttempts int
+}
+
+// NewRetryWriter returns a configured *RetryWriter which decorates
+// the supplied MetricsWriter
+func NewRetryWriter(w MetricsWriter, opts ...RetryOption) *RetryWriter {
+	r := &RetryWriter{
+		MetricsWriter: w,
+		sleep:         time.Sleep,
+		backoff:       func(int) time.Duration { return 0 },
+		maxAttempts:   defaultMaxAttempts,
+	}
+
+	for _, opt := range opts {
+		opt(r)
+	}
+
+	return r
+}
+
+// Write delegates to underlying MetricsWriter and then
+// automatically retries when errors occur
+func (r *RetryWriter) Write(m ...influxdb.Metric) (n int, err error) {
+	for i := 0; i < r.maxAttempts; i++ {
+		n, err = r.MetricsWriter.Write(m...)
+		if err == nil {
+			return
+		}
+
+		ierr, ok := err.(*influxdb.Error)
+		if !ok {
+			return
+		}
+
+		switch ierr.Code {
+		// retriable errors
+		case influxdb.EUnavailable, influxdb.ETooManyRequests:
+			if ierr.RetryAfter != nil {
+				// given retry-after is configured attempt to sleep
+				// for retry-after seconds
+				r.sleep(time.Duration(*ierr.RetryAfter) * time.Second)
+				continue
+			}
+
+			// given a backoff duration > 0
+			if duration := r.backoff(i + 1); duration > 0 {
+				// call sleep with backoff duration
+				r.sleep(duration)
+			}
+		default:
+			return
+		}
+	}
+
+	return
+}
+
+// LinearBackoff returns a BackoffFunc which when called
+// returns attempt * scale.
+// e.g.
+// LinearBackoff(time.Second)(5) returns 5 seconds
+func LinearBackoff(scale time.Duration) BackoffFunc {
+	return func(attempt int) time.Duration {
+		return time.Duration(attempt) * scale
+	}
+}
+
+// RetryOption is a functional option for the RetryWriters type
+type RetryOption func(*RetryWriter)
+
+// WithMaxAttempts sets the maximum number of attempts for a
+// Write operation attempt
+func WithMaxAttempts(maxAttempts int) RetryOption {
+	return func(r *RetryWriter) {
+		r.maxAttempts = maxAttempts
+	}
+}
+
+// WithBackoff sets of the BackoffFunc on the RetryWriter
+func WithBackoff(fn BackoffFunc) RetryOption {
+	return func(r *RetryWriter) {
+		r.backoff = fn
+	}
+}
diff --git a/writer/writer.go b/writer/writer.go
index d4a8129..5a4aa16 100644
--- a/writer/writer.go
+++ b/writer/writer.go
@@ -22,6 +22,12 @@ func New(writer BucketMetricWriter, bkt, org string, opts ...Option) *PointWrite
 		buffered = NewBufferedWriterSize(bucket, config.size)
 	)
 
+	if config.retry {
+		// configure automatic retries for transient errors
+		retry := NewRetryWriter(bucket, config.retryOptions...)
+		buffered = NewBufferedWriterSize(retry, config.size)
+	}
+
 	return NewPointWriter(buffered, config.flushInterval)
 }
 


Test patch:
diff --git a/write_test.go b/write_test.go
new file mode 100644
index 0000000..81df484
--- /dev/null
+++ b/write_test.go
@@ -0,0 +1,131 @@
+package influxdb
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+var five = int32(5)
+
+func Test_Client_Write(t *testing.T) {
+	for _, test := range []struct {
+		name string
+		// api response
+		body       []byte
+		statusCode int
+		headers    http.Header
+		// expectations
+		count int
+		err   error
+	}{
+		{
+			name:       "successful write",
+			statusCode: http.StatusOK,
+			count:      10,
+		},
+		{
+			name:       "rate limited",
+			statusCode: http.StatusTooManyRequests,
+			headers: http.Header{
+				"Retry-After": []string{"5"},
+			},
+			err: &Error{
+				Code:       ETooManyRequests,
+				Message:    "exceeded rate limit",
+				RetryAfter: &five,
+			},
+		},
+		{
+			name:       "unavailable",
+			statusCode: http.StatusServiceUnavailable,
+			headers:    http.Header{},
+			err: &Error{
+				Code:    EUnavailable,
+				Message: "service temporarily unavailable",
+			},
+		},
+		{
+			name:       "json encoded error",
+			statusCode: http.StatusInternalServerError,
+			body:       []byte(`{"code": "internal error", "op": "doing something", "message": "foo"}`),
+			headers: http.Header{
+				"Content-Type": []string{"application/json; charset=utf-8"},
+			},
+			err: &Error{
+				Code:    EInternal,
+				Op:      "doing something",
+				Message: "foo",
+			},
+		},
+		{
+			name:       "plain text encoded error",
+			statusCode: http.StatusBadRequest,
+			body:       []byte(`payload is bad`),
+			err: &Error{
+				Code:    "400 Bad Request",
+				Message: "payload is bad",
+			},
+		},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			var (
+				metrics = createTestRowMetrics(t, 10)
+				server  = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+					// path
+					assert.Equal(t, "/api/v2/write", r.URL.Path)
+					// headers
+					assert.Equal(t, "Token token", r.Header.Get("Authorization"))
+					assert.Equal(t, "text/plain; charset=utf-8", r.Header.Get("Content-Type"))
+					assert.Equal(t, "gzip", r.Header.Get("Content-Encoding"))
+					// params
+					assert.Equal(t, "bucket", r.URL.Query().Get("bucket"))
+					assert.Equal(t, "org", r.URL.Query().Get("org"))
+
+					for k, v := range test.headers {
+						w.Header()[k] = v
+					}
+					w.WriteHeader(test.statusCode)
+					w.Write(test.body)
+				}))
+				client, err = New(server.URL, "token")
+			)
+
+			require.Nil(t, err)
+
+			defer func() {
+				require.Nil(t, client.Close())
+				server.Close()
+			}()
+
+			count, err := client.Write(context.TODO(), "bucket", "org", metrics...)
+			assert.Equal(t, test.err, err)
+			assert.Equal(t, test.count, count)
+		})
+	}
+}
+
+func createTestRowMetrics(t *testing.T, count int) (metrics []Metric) {
+	t.Helper()
+
+	metrics = make([]Metric, 0, count)
+	for i := 0; i < count; i++ {
+		metrics = append(metrics, NewRowMetric(
+			map[string]interface{}{
+				"some_field": "some_value",
+			},
+			"some_measurement",
+			map[string]string{
+				"some_tag": "some_value",
+			},
+			time.Date(2019, time.January, 1, 0, 0, 0, 0, time.UTC),
+		))
+	}
+
+	return
+}
diff --git a/writer/retry_test.go b/writer/retry_test.go
new file mode 100644
index 0000000..c615d28
--- /dev/null
+++ b/writer/retry_test.go
@@ -0,0 +1,215 @@
+package writer
+
+import (
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/influxdata/influxdb-client-go"
+	"github.com/stretchr/testify/assert"
+)
+
+var (
+	errSimple  = errors.New("something went wrong")
+	errTooMany = func(retryAfter *int32) error {
+		return &influxdb.Error{
+			Code:       influxdb.ETooManyRequests,
+			Message:    "too many requests",
+			RetryAfter: retryAfter,
+		}
+	}
+	three int32 = 3
+	four  int32 = 4
+	five  int32 = 5
+)
+
+func Test_RetryWriter_Write(t *testing.T) {
+	for _, test := range []retryWriteCase{
+		{
+			name:    `one non-influxb.Error type error (mac attempts 3)`,
+			options: []RetryOption{WithMaxAttempts(3)},
+			metrics: createTestRowMetrics(t, 3),
+			errors: []error{
+				errSimple,
+			},
+			count: 0,
+			err:   errSimple,
+			writes: [][]influxdb.Metric{
+				// one write attempt, immediate failure
+				createTestRowMetrics(t, 3),
+			},
+		},
+		{
+			name:    `two "too many requests" errors (max attempts 3)`,
+			options: []RetryOption{WithMaxAttempts(3)},
+			metrics: createTestRowMetrics(t, 3),
+			errors: []error{
+				errTooMany(nil),
+				errTooMany(nil),
+			},
+			count: 3,
+			writes: [][]influxdb.Metric{
+				// three writes, third succeeds
+				createTestRowMetrics(t, 3),
+				createTestRowMetrics(t, 3),
+				createTestRowMetrics(t, 3),
+			},
+		},
+		{
+			name:    `one "unavailable" errors (max attempts 3)`,
+			options: []RetryOption{WithMaxAttempts(3)},
+			metrics: createTestRowMetrics(t, 3),
+			errors: []error{
+				&influxdb.Error{Code: influxdb.EUnavailable, Message: "too many requests"},
+			},
+			count: 3,
+			writes: [][]influxdb.Metric{
+				// two writes, second succeeds
+				createTestRowMetrics(t, 3),
+				createTestRowMetrics(t, 3),
+			},
+		},
+		{
+			name:    `one "internal error" error (max attempts 3)`,
+			options: []RetryOption{WithMaxAttempts(3)},
+			metrics: createTestRowMetrics(t, 3),
+			errors: []error{
+				&influxdb.Error{Code: influxdb.EInternal, Message: "something went wrong"},
+			},
+			count: 0,
+			err:   &influxdb.Error{Code: influxdb.EInternal, Message: "something went wrong"},
+			writes: [][]influxdb.Metric{
+				// one attempted write
+				createTestRowMetrics(t, 3),
+			},
+		},
+		{
+			name:    `three "too many requests" errors (max attempts 3)`,
+			options: []RetryOption{WithMaxAttempts(3)},
+			metrics: createTestRowMetrics(t, 3),
+			errors: []error{
+				errTooMany(nil),
+				errTooMany(nil),
+				errTooMany(nil),
+			},
+			count: 0,
+			err:   errTooMany(nil),
+			writes: [][]influxdb.Metric{
+				// three writes all error
+				createTestRowMetrics(t, 3),
+				createTestRowMetrics(t, 3),
+				createTestRowMetrics(t, 3),
+			},
+		},
+		{
+			name:    `three "too many requests" errors (max attempts 3) with retry-after`,
+			options: []RetryOption{WithMaxAttempts(3)},
+			metrics: createTestRowMetrics(t, 3),
+			errors: []error{
+				errTooMany(&three),
+				errTooMany(&three),
+				errTooMany(&three),
+			},
+			count: 0,
+			err:   errTooMany(&three),
+			writes: [][]influxdb.Metric{
+				// three writes all error
+				createTestRowMetrics(t, 3),
+				createTestRowMetrics(t, 3),
+				createTestRowMetrics(t, 3),
+			},
+			sleeps: []time.Duration{
+				3 * time.Second,
+				3 * time.Second,
+				3 * time.Second,
+			},
+		},
+		{
+			name: `three "too many requests" errors (max attempts 3) with backoff`,
+			options: []RetryOption{
+				WithMaxAttempts(3),
+				WithBackoff(LinearBackoff(time.Millisecond)),
+			},
+			metrics: createTestRowMetrics(t, 3),
+			errors: []error{
+				errTooMany(nil),
+				errTooMany(nil),
+				errTooMany(nil),
+			},
+			count: 0,
+			err:   errTooMany(nil),
+			writes: [][]influxdb.Metric{
+				// three writes all error
+				createTestRowMetrics(t, 3),
+				createTestRowMetrics(t, 3),
+				createTestRowMetrics(t, 3),
+			},
+			sleeps: []time.Duration{
+				1 * time.Millisecond,
+				2 * time.Millisecond,
+				3 * time.Millisecond,
+			},
+		},
+		{
+			name: `three "too many requests" errors (max attempts 3) with backoff and retry-after`,
+			options: []RetryOption{
+				WithMaxAttempts(3),
+				WithBackoff(LinearBackoff(time.Millisecond)),
+			},
+			metrics: createTestRowMetrics(t, 3),
+			errors: []error{
+				errTooMany(&three),
+				errTooMany(&four),
+				errTooMany(&five),
+			},
+			count: 0,
+			err:   errTooMany(&five),
+			writes: [][]influxdb.Metric{
+				// three writes all error
+				createTestRowMetrics(t, 3),
+				createTestRowMetrics(t, 3),
+				createTestRowMetrics(t, 3),
+			},
+			sleeps: []time.Duration{
+				3 * time.Second,
+				4 * time.Second,
+				5 * time.Second,
+			},
+		},
+	} {
+		t.Run(test.name, test.Run)
+	}
+}
+
+type retryWriteCase struct {
+	name string
+	// inputs
+	options []RetryOption
+	metrics []influxdb.Metric
+	// errors returned by write
+	errors []error
+	// response expectations
+	count int
+	err   error
+	// attempted writes
+	writes [][]influxdb.Metric
+	sleeps []time.Duration
+}
+
+func (test *retryWriteCase) Run(t *testing.T) {
+	var (
+		writer = newTestWriter(test.errors...)
+		retry  = NewRetryWriter(writer, test.options...)
+		sleeps []time.Duration
+	)
+
+	retry.sleep = func(d time.Duration) {
+		sleeps = append(sleeps, d)
+	}
+
+	count, err := retry.Write(test.metrics...)
+	assert.Equal(t, test.err, err)
+	assert.Equal(t, test.count, count)
+	assert.Equal(t, test.writes, writer.writes)
+	assert.Equal(t, test.sleeps, sleeps)
+}
diff --git a/writer/support_test.go b/writer/support_test.go
index 039d9c5..6d66c23 100644
--- a/writer/support_test.go
+++ b/writer/support_test.go
@@ -31,21 +31,33 @@ type metricsWriter struct {
 	// metrics written
 	when   []time.Time
 	writes [][]influxdb.Metric
+	// error response
+	called int
+	errs   []error
 }
 
-func newTestWriter() *metricsWriter {
-	return &metricsWriter{n: -1}
+func newTestWriter(errs ...error) *metricsWriter {
+	return &metricsWriter{n: -1, errs: errs}
 }
 
-func (w *metricsWriter) Write(m ...influxdb.Metric) (int, error) {
+func (w *metricsWriter) Write(m ...influxdb.Metric) (n int, err error) {
+	defer func() { w.called++ }()
+
 	w.when = append(w.when, time.Now())
 	w.writes = append(w.writes, m)
 
+	n = len(m)
 	if w.n > -1 {
-		return w.n, nil
+		// override length response
+		n = w.n
 	}
 
-	return len(m), nil
+	if w.called < len(w.errs) {
+		n = 0
+		err = w.errs[w.called]
+	}
+
+	return
 }
 
 // permuteCounts returns a set of pseudo-random batch size counts
diff --git a/writer/writer_test.go b/writer/writer_test.go
index fef734b..1c02d3c 100644
--- a/writer/writer_test.go
+++ b/writer/writer_test.go
@@ -13,7 +13,7 @@ func Test_New(t *testing.T) {
 		client  = &influxdb.Client{}
 		bkt     = "default"
 		org     = "influx"
-		options = Options{WithBufferSize(12), WithFlushInterval(5 * time.Minute)}
+		options = Options{WithBufferSize(12), WithFlushInterval(5 * time.Minute), WithRetries()}
 		wr      = New(client, bkt, org, options...)
 	)