Don't use exponential backoff in http_forwarder for non-retryable errors when fail-fast-on-non-retryable-errors set to true (#773)

* Add logic to provide a contextual error when invoking the returned `post` function returned from `constructPost` in `handler_http_forwarder_v2.go` which indicates whether the request is retryable or not. Does not attempt exponential backoff if not retryable.

* add guards for errors.As() calls to ensure expected behavior, fallback to old error format if not ok. Use RetryableErrorCodes instead of using inverse logic for `NonRetryableErrorCodes`

* Add Test to confirm backoff is not triggered on bad certificate

* restructure code so we don't call b.NextBackOff unless needed

* use a guard clause to exit on non-retryable errors instead of nexted conditionals

* always return `RequestError` from postInstanceFn

* add specific atomic counter for fast failure

* make new failfast behaviour opt-in via configuration

Author: akavatl

README.md

@@ -146,6 +146,11 @@ following configuration options:
   in both `custom-header` and `dynamic-header`, the vaule set by `custom-header` takes precedence. Not required, default
   is empty. Example: `--dynamic-headers='["region", "service"]'`.
   This is an experimental feature and it may be removed or changed in future versions.
+- `fail-fast-on-non-retryable-errors`: boolean flag that controls whether to immediately drop messages that encounter
+  non-retryable HTTP errors (e.g., 4xx status codes indicating client errors). When set to `true`, non-retryable errors
+  immediately drop the message without retry attempts. When set to `false`, even non-retryable errors will attempt retries
+  via exponential backoff up to the `max-request-elapsed-time` limit. Note: Some errors (e.g., TLS certificate validation
+  failures) are always treated as non-retryable. Defaults to `false`.
 
 The following settings from the previous section are also supported:
 - `expiry-*`

cmd/gostatsd/main.go

@@ -174,12 +174,13 @@ func constructServer(v *viper.Viper) (*statsd.Server, error) {
 			fmt.Sprintf("version:%s", Version),
 			fmt.Sprintf("commit:%s", GitCommit),
 		},
-		DisabledSubTypes:          gostatsd.DisabledSubMetrics(v),
-		BadLineRateLimitPerSecond: rate.Limit(v.GetFloat64(gostatsd.ParamBadLinesPerMinute) / 60.0),
-		HistogramLimit:            v.GetUint32(gostatsd.ParamTimerHistogramLimit),
-		DisableInternalEvents:     v.GetBool(gostatsd.ParamDisableInternalEvents),
-		Viper:                     v,
-		TransportPool:             pool,
+		DisabledSubTypes:             gostatsd.DisabledSubMetrics(v),
+		BadLineRateLimitPerSecond:    rate.Limit(v.GetFloat64(gostatsd.ParamBadLinesPerMinute) / 60.0),
+		HistogramLimit:               v.GetUint32(gostatsd.ParamTimerHistogramLimit),
+		DisableInternalEvents:        v.GetBool(gostatsd.ParamDisableInternalEvents),
+		Viper:                        v,
+		TransportPool:                pool,
+		FastFailOnNonRetryableErrors: v.GetBool(gostatsd.ParamFastFailOnNonRetryableErrors),
 	}, nil
 }
 

cmd/lambda-extension/main.go

@@ -95,9 +95,10 @@ func NewServer(v *viper.Viper, logger logrus.FieldLogger) *statsd.Server {
 			fmt.Sprintf("version:%s", Version),
 			fmt.Sprintf("commit:%s", GitCommit),
 		},
-		BadLineRateLimitPerSecond: rate.Limit(v.GetFloat64(gostatsd.ParamBadLinesPerMinute) / 60.0),
-		Viper:                     v,
-		TransportPool:             transport.NewTransportPool(logger, v),
+		BadLineRateLimitPerSecond:    rate.Limit(v.GetFloat64(gostatsd.ParamBadLinesPerMinute) / 60.0),
+		Viper:                        v,
+		TransportPool:                transport.NewTransportPool(logger, v),
+		FastFailOnNonRetryableErrors: v.GetBool(gostatsd.ParamFastFailOnNonRetryableErrors),
 	}
 
 	if v.GetBool(gostatsd.ParamLambdaExtensionManualFlush) {

defaults_and_params.go

@@ -97,6 +97,8 @@ const (
 	DefaultLogRawMetric = false
 	// DefaultDisableInternalEvents is the default value for disabling internal events being sent
 	DefaultDisableInternalEvents = false
+	// DefaultFastFailOnNonRetryableErrors is the default value for whether to forego exponential backoff on non-retryable errors
+	DefaultFastFailOnNonRetryableErrors = false
 )
 
 const (
@@ -184,6 +186,8 @@ const (
 	ParamLambdaExtensionManualFlush = "lambda-extension-manual-flush"
 	// ParamLambdaExtensionTelemetryAddress enables the manual flushing of metrics in forwarder mode, the flush interval is ignored
 	ParamLambdaExtensionTelemetryAddress = "lambda-extension-telemetry-address"
+	// ParamFastFailOnNonRetryableErrors disables exponential backoff on http client errors and non retryable http status codes
+	ParamFastFailOnNonRetryableErrors = "fast-fail-on-non-retryable-errors"
 )
 
 // AddFlags adds flags to the specified FlagSet.
@@ -227,6 +231,8 @@ func AddFlags(fs *pflag.FlagSet) {
 	fs.Uint32(ParamTimerHistogramLimit, DefaultTimerHistogramLimit, "upper limit of timer histogram buckets (MaxUint32 by default)")
 	fs.Bool(ParamLogRawMetric, DefaultLogRawMetric, "Print metrics received from network to stdout in JSON format")
 	fs.Bool(ParamDisableInternalEvents, DefaultDisableInternalEvents, "Disables sending internal events from gostatsd")
+	fs.Bool(ParamFastFailOnNonRetryableErrors, DefaultFastFailOnNonRetryableErrors, "Disables exponential backoff and retry on non-retryable http statuses and client errors if true.")
+
 }
 
 func minInt(a, b int) int {

pkg/statsd/handler_http_forwarder_v2.go

@@ -3,11 +3,14 @@ package statsd
 import (
 	"bytes"
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"io/ioutil"
 	"maps"
 	"net/http"
+	"net/url"
+	"slices"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -31,27 +34,43 @@ import (
 )
 
 const (
-	defaultConsolidatorFlushInterval = 1 * time.Second
-	defaultCompress                  = true
-	defaultCompressionType           = "zlib"
-	defaultCompressionLevel          = 9
-	defaultApiEndpoint               = ""
-	defaultMaxRequestElapsedTime     = 30 * time.Second
-	defaultMaxRequests               = 1000
-	defaultConcurrentMerge           = 1
-	defaultTransport                 = "default"
+	defaultConsolidatorFlushInterval   = 1 * time.Second
+	defaultCompress                    = true
+	defaultCompressionType             = "zlib"
+	defaultCompressionLevel            = 9
+	defaultApiEndpoint                 = ""
+	defaultMaxRequestElapsedTime       = 30 * time.Second
+	defaultMaxRequests                 = 1000
+	defaultConcurrentMerge             = 1
+	defaultTransport                   = "default"
+	defaultFastFailOnNonRetryableError = false
 )
 
+var (
+	RetryableErrorCodes = []int{408, 429, 500, 502, 503, 504}
+)
+
+type RequestError struct {
+	StatusCode int
+	Retryable  bool
+	Err        error
+}
+
+func (r *RequestError) Error() string {
+	return fmt.Sprintf("status %d: retryable %t: err %v", r.StatusCode, r.Retryable, r.Err)
+}
+
 // HttpForwarderHandlerV2 is a PipelineHandler which sends metrics to another gostatsd instance
 type HttpForwarderHandlerV2 struct {
-	postId           uint64       // atomic - used for an id in logs
-	messagesInvalid  uint64       // atomic - messages which failed to be created
-	messagesCreated  uint64       // atomic - messages which were created
-	messagesSent     uint64       // atomic - messages successfully sent
-	messagesRetried  uint64       // atomic - retries (first send is not a retry, final failure is not a retry)
-	messagesDropped  uint64       // atomic - final failure
-	postLatencyTotal atomic.Int64 // total of the time taken to send messages in a flush interval
-	postLatencyMax   atomic.Int64 // maximum time taken to send a message in a flush interval
+	postId                  uint64       // atomic - used for an id in logs
+	messagesInvalid         uint64       // atomic - messages which failed to be created
+	messagesCreated         uint64       // atomic - messages which were created
+	messagesSent            uint64       // atomic - messages successfully sent
+	messagesRetried         uint64       // atomic - retries (first send is not a retry, final failure is not a retry)
+	messagesDroppedFastFail uint64       // atomic - messages which failed due to non-retryable errors
+	messagesDropped         uint64       // atomic - final failure
+	postLatencyTotal        atomic.Int64 // total of the time taken to send messages in a flush interval
+	postLatencyMax          atomic.Int64 // maximum time taken to send a message in a flush interval
 
 	lastSuccessfulSend atomic.Int64
 
@@ -72,7 +91,8 @@ type HttpForwarderHandlerV2 struct {
 	consolidator        *gostatsd.MetricConsolidator
 	consolidatedMetrics chan []*gostatsd.MetricMap
 
-	flushCoordinator flush.Coordinator
+	flushCoordinator             flush.Coordinator
+	fastFailOnNonRetryableErrors bool
 }
 
 var (
@@ -84,16 +104,17 @@ var (
 // as part of the configuration passed through.
 func newHTTPForwarderHandlerViperConfig(overrides *viper.Viper) *viper.Viper {
 	values := map[string]any{
-		"transport":                defaultTransport,
-		"compress":                 defaultCompress,
-		"compression-type":         defaultCompressionType,
-		"compression-level":        defaultCompressionLevel,
-		"api-endpoint":             defaultApiEndpoint,
-		"max-requests":             defaultMaxRequests,
-		"max-request-elapsed-time": defaultMaxRequestElapsedTime,
-		"consolidator-slots":       gostatsd.DefaultMaxParsers,
-		"flush-interval":           defaultConsolidatorFlushInterval,
-		"concurrent-merge":         defaultConcurrentMerge,
+		"transport":                         defaultTransport,
+		"compress":                          defaultCompress,
+		"compression-type":                  defaultCompressionType,
+		"compression-level":                 defaultCompressionLevel,
+		"api-endpoint":                      defaultApiEndpoint,
+		"max-requests":                      defaultMaxRequests,
+		"max-request-elapsed-time":          defaultMaxRequestElapsedTime,
+		"consolidator-slots":                gostatsd.DefaultMaxParsers,
+		"flush-interval":                    defaultConsolidatorFlushInterval,
+		"concurrent-merge":                  defaultConcurrentMerge,
+		"fast-fail-on-non-retryable-errors": defaultFastFailOnNonRetryableError,
 	}
 	maps.Copy(values, util.GetSubViper(overrides, "http-transport").AllSettings())
 
@@ -123,6 +144,7 @@ func NewHttpForwarderHandlerV2FromViper(logger logrus.FieldLogger, v *viper.Vipe
 		values.GetStringSlice("dynamic-headers"),
 		pool,
 		fc,
+		values.GetBool("fast-fail-on-non-retryable-errors"),
 	)
 }
 
@@ -143,6 +165,7 @@ func NewHttpForwarderHandlerV2(
 	dynHeaderNames []string,
 	pool *transport.TransportPool,
 	fc flush.Coordinator,
+	failFastOnNonRetryableErrors bool,
 ) (*HttpForwarderHandlerV2, error) {
 	if apiEndpoint == "" {
 		return nil, fmt.Errorf("api-endpoint is required")
@@ -178,14 +201,15 @@ func NewHttpForwarderHandlerV2(
 	}
 
 	logger.WithFields(logrus.Fields{
-		"api-endpoint":             apiEndpoint,
-		"compress":                 compress,
-		"compression-type":         compressionTypeStr,
-		"compression-level":        compressionLevel,
-		"max-request-elapsed-time": maxRequestElapsedTime,
-		"max-requests":             maxRequests,
-		"consolidator-slots":       consolidatorSlots,
-		"flush-interval":           flushInterval,
+		"api-endpoint":                      apiEndpoint,
+		"compress":                          compress,
+		"compression-type":                  compressionTypeStr,
+		"compression-level":                 compressionLevel,
+		"max-request-elapsed-time":          maxRequestElapsedTime,
+		"max-requests":                      maxRequests,
+		"consolidator-slots":                consolidatorSlots,
+		"flush-interval":                    flushInterval,
+		"fast-fail-on-non-retryable-errors": failFastOnNonRetryableErrors,
 	}).Info("created HttpForwarderHandler")
 
 	// Default set of headers used for the forwarder
@@ -232,21 +256,22 @@ func NewHttpForwarderHandlerV2(
 	}
 
 	return &HttpForwarderHandlerV2{
-		logger:                logger.WithField("component", "http-forwarder-handler-v2"),
-		apiEndpoint:           apiEndpoint,
-		maxRequestElapsedTime: maxRequestElapsedTime,
-		metricsSem:            metricsSem,
-		metricsMergingSem:     metricsMergingSem,
-		compress:              compress,
-		compressionType:       compressionType,
-		compressionLevel:      compressionLevel,
-		consolidator:          consolidator,
-		consolidatedMetrics:   ch,
-		client:                httpClient.Client,
-		headers:               headers,
-		dynHeaderNames:        dynHeaderNamesWithColon,
-		done:                  make(chan struct{}),
-		flushCoordinator:      fc,
+		logger:                       logger.WithField("component", "http-forwarder-handler-v2"),
+		apiEndpoint:                  apiEndpoint,
+		maxRequestElapsedTime:        maxRequestElapsedTime,
+		metricsSem:                   metricsSem,
+		metricsMergingSem:            metricsMergingSem,
+		compress:                     compress,
+		compressionType:              compressionType,
+		compressionLevel:             compressionLevel,
+		consolidator:                 consolidator,
+		consolidatedMetrics:          ch,
+		client:                       httpClient.Client,
+		headers:                      headers,
+		dynHeaderNames:               dynHeaderNamesWithColon,
+		done:                         make(chan struct{}),
+		flushCoordinator:             fc,
+		fastFailOnNonRetryableErrors: failFastOnNonRetryableErrors,
 	}, nil
 }
 
@@ -460,7 +485,7 @@ func (hfh *HttpForwarderHandlerV2) post(ctx context.Context, message proto.Messa
 		"type": endpointType,
 	})
 
-	post, err := hfh.constructPost(ctx, logger, hfh.apiEndpoint+endpoint, message, dynHeaderTags)
+	postInstanceFn, err := hfh.constructPost(ctx, logger, hfh.apiEndpoint+endpoint, message, dynHeaderTags)
 	if err != nil {
 		atomic.AddUint64(&hfh.messagesInvalid, 1)
 		logger.WithError(err).Error("failed to create request")
@@ -474,7 +499,7 @@ func (hfh *HttpForwarderHandlerV2) post(ctx context.Context, message proto.Messa
 
 	for {
 		startTime := clock.Now(ctx)
-		if err = post(); err == nil {
+		if err = postInstanceFn(); err == nil {
 			atomic.AddUint64(&hfh.messagesSent, 1)
 			hfh.lastSuccessfulSend.Store(clock.Now(ctx).UnixNano())
 
@@ -491,6 +516,15 @@ func (hfh *HttpForwarderHandlerV2) post(ctx context.Context, message proto.Messa
 			return
 		}
 
+		// All errors coming back from postInstanceFn() should be a RequestError
+		var reqErr *RequestError
+		ok := errors.As(err, &reqErr)
+		if ok && !reqErr.Retryable && hfh.fastFailOnNonRetryableErrors {
+			atomic.AddUint64(&hfh.messagesDroppedFastFail, 1)
+			logger.WithError(err).Info("failed to send due to non-retryable error giving up")
+			return
+		}
+
 		next := b.NextBackOff()
 		if next == backoff.Stop {
 			atomic.AddUint64(&hfh.messagesDropped, 1)
@@ -593,7 +627,21 @@ func (hfh *HttpForwarderHandlerV2) constructPost(ctx context.Context, logger log
 		req.Header.Set("Content-Encoding", encoding)
 		resp, err := hfh.client.Do(req)
 		if err != nil {
-			return fmt.Errorf("error POSTing: %v", err)
+			// Any error returned from client.Do returns *url.Error
+			// Documentation Ref: https://pkg.go.dev/net/http#Client.Do
+			var urlErr *url.Error
+			ok := errors.As(err, &urlErr)
+			if ok {
+				return &RequestError{
+					Retryable: urlErr.Temporary(),
+					Err:       err,
+				}
+			}
+
+			return &RequestError{
+				Retryable: false,
+				Err:       err,
+			}
 		}
 		defer func() {
 			_, _ = io.Copy(ioutil.Discard, resp.Body)
@@ -605,7 +653,11 @@ func (hfh *HttpForwarderHandlerV2) constructPost(ctx context.Context, logger log
 				"status": resp.StatusCode,
 				"body":   string(bodyStart),
 			}).Info("failed request")
-			return fmt.Errorf("received bad status code %d", resp.StatusCode)
+			return &RequestError{
+				StatusCode: resp.StatusCode,
+				Retryable:  slices.Contains(RetryableErrorCodes, resp.StatusCode),
+				Err:        err,
+			}
 		}
 		return nil
 	}, nil