receive: Add retry-after backoff with jitter via header field to active-series-limiting

By default, the active-series-limiting does only return a 429 when the
tenant exceeded the limit. Prometheus does parse the retry-after header
by default¹ and if the header is not set, it will do not set a default
backoff on prometheus side. This results in a instant retry of sending
the data again which may result in a potential increase of requests.

¹https://github.com/prometheus/prometheus/blob/bfbae39931a6ddeb5913a6ea8d48a34bbebc6d29/storage/remote/client.go#L309

Signed-off-by: roth-wine <[email protected]>

Author: roth-wine

CHANGELOG.md

@@ -13,6 +13,7 @@ We use *breaking :warning:* to mark changes that are not backward compatible (re
 ### Fixed
 
 ### Added
+- [#8356](https://github.com/thanos-io/thanos/pull/8356): receive: Add retry-after backoff with jitter via header field to active-series-limiting
 
 - [#8366](https://github.com/thanos-io/thanos/pull/8366) Store: optionally ignore Parquet migrated blocks
 - [#8359](https://github.com/thanos-io/thanos/pull/8359) Tools: add `--shipper.upload-compacted` flag for uploading compacted blocks to bucket upload-blocks

cmd/thanos/receive.go

@@ -293,6 +293,8 @@ func runReceive(
 		ReplicationProtocol:     receive.ReplicationProtocol(conf.replicationProtocol),
 		OtlpEnableTargetInfo:    conf.otlpEnableTargetInfo,
 		OtlpResourceAttributes:  conf.otlpResourceAttributes,
+		RetryAfterBackoff:       time.Duration(*conf.retryAfterBackoff),
+		RetryAfterJitter:        conf.retryAfterJitter,
 	})
 
 	grpcProbe := prober.NewGRPC()
@@ -865,6 +867,8 @@ type receiveConfig struct {
 	compression         string
 	replicationProtocol string
 	grpcServiceConfig   string
+	retryAfterBackoff   *model.Duration
+	retryAfterJitter    float64
 
 	tsdbMinBlockDuration         *model.Duration
 	tsdbMaxBlockDuration         *model.Duration
@@ -1078,6 +1082,10 @@ func (rc *receiveConfig) registerFlag(cmd extkingpin.FlagClause) {
 
 	cmd.Flag("receive.lazy-retrieval-max-buffered-responses", "The lazy retrieval strategy can buffer up to this number of responses. This is to limit the memory usage. This flag takes effect only when the lazy retrieval strategy is enabled.").
 		Default("20").IntVar(&rc.lazyRetrievalMaxBufferedResponses)
+
+	rc.retryAfterBackoff = extkingpin.ModelDuration(cmd.Flag("receive.active-series-limiting.retry-after-backoff", "Backoff for retry-after header, set if active-series-limiting is enabled").Default("5s"))
+
+	cmd.Flag("receive.active-series-limiting.retry-after-jitter", "Jitter for retry-after backoff").Default("0.5").Float64Var(&rc.retryAfterJitter)
 }
 
 // determineMode returns the ReceiverMode that this receiver is configured to run in.

docs/components/receive.md

@@ -372,7 +372,7 @@ Please see the metric `thanos_receive_forward_delay_seconds` to see if you need
 
 The following formula is used for calculating quorum:
 
-```go mdox-exec="sed -n '1046,1056p' pkg/receive/handler.go"
+```go mdox-exec="sed -n '1051,1061p' pkg/receive/handler.go"
 // writeQuorum returns minimum number of replicas that has to confirm write success before claiming replication success.
 func (h *Handler) writeQuorum() int {
 	// NOTE(GiedriusS): this is here because otherwise RF=2 doesn't make sense as all writes
@@ -670,5 +670,10 @@ Flags:
                                  this number of responses. This is to limit the
                                  memory usage. This flag takes effect only when
                                  the lazy retrieval strategy is enabled.
+      --receive.active-series-limiting.retry-after-backoff=5s
+                                 Backoff for retry-after header, set if
+                                 active-series-limiting is enabled
+      --receive.active-series-limiting.retry-after-jitter=0.5
+                                 Jitter for retry-after backoff
 
 ```

pkg/receive/handler.go

@@ -42,6 +42,7 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
+	"github.com/thanos-io/thanos/internal/cortex/util"
 	"github.com/thanos-io/thanos/pkg/api"
 	statusapi "github.com/thanos-io/thanos/pkg/api/status"
 	"github.com/thanos-io/thanos/pkg/logging"
@@ -119,6 +120,8 @@ type Options struct {
 	ReplicationProtocol     ReplicationProtocol
 	OtlpEnableTargetInfo    bool
 	OtlpResourceAttributes  []string
+	RetryAfterBackoff       time.Duration
+	RetryAfterJitter        float64
 }
 
 // Handler serves a Prometheus remote write receiving HTTP endpoint.
@@ -534,6 +537,8 @@ func (h *Handler) receiveHTTP(w http.ResponseWriter, r *http.Request) {
 
 	// Fail request fully if tenant has exceeded set limit.
 	if !under {
+		backoffDuration := util.DurationWithJitter(h.options.RetryAfterBackoff, h.options.RetryAfterJitter)
+		w.Header().Add("Retry-After", time.Now().Add(backoffDuration).Format(http.TimeFormat))
 		http.Error(w, "tenant is above active series limit", http.StatusTooManyRequests)
 		return
 	}