Merge pull request #719 from atlassian/htan/retry-test

add tests and refactor retry sending

Author: hstan

pkg/backends/otlp/backend.go

@@ -304,39 +304,44 @@ func (bd *Backend) SendMetricsAsync(ctx context.Context, mm *gostatsd.MetricMap,
 }
 
 func (c *Backend) postMetrics(ctx context.Context, batch group) error {
-	resourceMetrics := batch.values()
+	var (
+		retries int
+		req     *http.Request
+		resp    *http.Response
+		err     error
+	)
 
-	req, err := data.NewMetricsRequest(ctx, c.metricsEndpoint, resourceMetrics)
+	resourceMetrics := batch.values()
+	req, err = data.NewMetricsRequest(ctx, c.metricsEndpoint, resourceMetrics)
 	if err != nil {
 		atomic.AddUint64(&c.batchesDropped, 1)
 		return err
 	}
 
-	c.requestsBufferSem <- struct{}{}
-	resp, err := c.client.Do(req)
-	<-c.requestsBufferSem
-	// OTLP standard specifies 400 will be returned if the request is non-retryable, so we don't retry on 400
-	if err != nil && resp.StatusCode != http.StatusBadRequest {
-		for i := 0; i < c.maxRetries; i++ {
-			atomic.AddUint64(&c.batchesRetried.Cur, 1)
-			if resp != nil {
-				resp.Body.Close()
-			}
-			c.requestsBufferSem <- struct{}{}
-			resp, err = c.client.Do(req)
-			<-c.requestsBufferSem
+	for {
+		var dropped int64
+		c.requestsBufferSem <- struct{}{}
+		resp, err = c.client.Do(req)
+		<-c.requestsBufferSem
+		if err == nil {
+			dropped, err = data.ProcessMetricResponse(resp)
 			if err == nil {
-				break
-			} else {
-				c.logger.WithError(err).WithFields(logrus.Fields{
-					"endpoint": c.metricsEndpoint,
-				}).Error("failed while retrying")
+				return nil
 			}
+			if dropped > 0 {
+				// If partial data points were dropped, it shouldn't retry
+				atomic.AddUint64(&c.seriesDropped, uint64(dropped))
+				return err
+			}
+		}
+
+		if retries >= c.maxRetries {
+			break
 		}
+
+		retries++
+		atomic.AddUint64(&c.batchesRetried.Cur, 1)
 	}
-	defer resp.Body.Close()
-	dropped, err := data.ProcessMetricResponse(resp)
-	atomic.AddUint64(&c.seriesDropped, uint64(dropped))
 
 	return err
 }

pkg/backends/otlp/backend_test.go

@@ -380,6 +380,78 @@ func TestBackendSendAsyncMetrics(t *testing.T) {
 	}
 }
 
+func TestRetrySendMetrics(t *testing.T) {
+	t.Parallel()
+	for _, tc := range []struct {
+		name            string
+		numUntilSuccess int
+		maxRetries      int
+		wantAttempts    int
+		numErrs         int
+	}{
+		{
+			name:            "should retry sending metrics if it fails for the first time",
+			numUntilSuccess: 2,
+			maxRetries:      3,
+			wantAttempts:    2,
+			numErrs:         0,
+		},
+		{
+			name:            "should give up if it still fails when reach the maximum number of retries",
+			numUntilSuccess: 5,
+			maxRetries:      3,
+			wantAttempts:    4,
+			numErrs:         1,
+		},
+		{
+			name:            "should not retry if it succeeds at the first time",
+			numUntilSuccess: 1,
+			maxRetries:      3,
+			wantAttempts:    1,
+			numErrs:         0,
+		},
+		{
+			name:            "should not retry if maxRetries is 0",
+			numUntilSuccess: 5,
+			maxRetries:      0,
+			wantAttempts:    1,
+			numErrs:         1,
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			attempts := 0
+			s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				attempts++
+				if attempts < tc.numUntilSuccess {
+					http.Error(w, "im dead", http.StatusServiceUnavailable)
+					return
+				}
+				w.WriteHeader(http.StatusOK)
+			}))
+			t.Cleanup(s.Close)
+
+			v := viper.New()
+			v.Set("otlp.metrics_endpoint", fmt.Sprintf("%s/%s", s.URL, "v1/metrics"))
+			v.Set("otlp.logs_endpoint", fmt.Sprintf("%s/%s", s.URL, "v1/logs"))
+			v.Set("otlp.max_retries", tc.maxRetries)
+
+			logger := fixtures.NewTestLogger(t)
+
+			b, err := NewClientFromViper(
+				v,
+				logger,
+				transport.NewTransportPool(logger, v),
+			)
+			require.NoError(t, err, "Must not error creating backend")
+
+			b.SendMetricsAsync(context.Background(), gostatsd.NewMetricMap(false), func(errs []error) {
+				assert.Equal(t, tc.numErrs, len(errs))
+				assert.Equal(t, tc.wantAttempts, attempts, "Must retry sending metrics")
+			})
+		})
+	}
+}
+
 func TestSendEvent(t *testing.T) {
 	t.Parallel()
 	for _, tc := range []struct {

pkg/backends/otlp/config.go

@@ -91,7 +91,7 @@ func (c *Config) Validate() (errs error) {
 	if c.MaxRequests <= 0 {
 		errs = multierr.Append(errs, errors.New("max request must be a positive value"))
 	}
-	if c.MaxRetries <= 0 {
+	if c.MaxRetries < 0 {
 		errs = multierr.Append(errs, errors.New("max retries must be a positive value"))
 	}
 	if c.MetricsPerBatch <= 0 {

pkg/backends/otlp/internal/data/metric_response.go

@@ -1,6 +1,7 @@
 package data
 
 import (
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -11,6 +12,10 @@ import (
 )
 
 func ProcessMetricResponse(resp *http.Response) (dropped int64, errs error) {
+	if resp == nil {
+		return 0, errors.New("empty response")
+	}
+
 	buf, err := io.ReadAll(resp.Body)
 	if err != nil {
 		return 0, err