Sidecar Flag: Random delay before upload blocks on sidecar

cmd/thanos/config.go

@@ -164,6 +164,7 @@ type shipperConfig struct {
 	allowOutOfOrderUpload bool
 	hashFunc              string
 	metaFileName          string
+	uploadJitter          time.Duration
 }
 
 func (sc *shipperConfig) registerFlag(cmd extkingpin.FlagClause) *shipperConfig {

cmd/thanos/sidecar.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"fmt"
 	"math"
+	"math/rand"
 	"net/http"
 	"net/url"
 	"sync"
@@ -105,6 +106,18 @@ func registerSidecar(app *extkingpin.App) {
 	})
 }
 
+// DurationWithJitter returns random duration from "input - input*variancePerc" to "input + input*variancePerc" interval.
+func DurationWithJitter(input time.Duration, variancePerc float64) time.Duration {
+	if input == 0 {
+		return 0
+	}
+
+	variance := int64(float64(input) * variancePerc)
+	jitter := rand.Int63n(variance*2) - variance
+
+	return input + time.Duration(jitter)
+}
+
 func runSidecar(
 	g *run.Group,
 	logger log.Logger,
@@ -362,7 +375,7 @@ func runSidecar(
 			s := shipper.New(logger, reg, conf.tsdb.path, bkt, m.Labels, metadata.SidecarSource,
 				uploadCompactedFunc, conf.shipper.allowOutOfOrderUpload, metadata.HashFunc(conf.shipper.hashFunc), conf.shipper.metaFileName)
 
-			return runutil.Repeat(30*time.Second, ctx.Done(), func() error {
+			return runutil.RepeatWithJitter(30*time.Second, ctx, 0.05, func() error {
 				if uploaded, err := s.Sync(ctx); err != nil {
 					level.Warn(logger).Log("err", err, "uploaded", uploaded)
 				}
@@ -518,4 +531,6 @@ func (sc *sidecarConfig) registerFlag(cmd extkingpin.FlagClause) {
 	sc.storeRateLimits.RegisterFlags(cmd)
 	cmd.Flag("min-time", "Start of time range limit to serve. Thanos sidecar will serve only metrics, which happened later than this value. Option can be a constant time in RFC3339 format or time duration relative to current time, such as -1d or 2h45m. Valid duration units are ms, s, m, h, d, w, y.").
 		Default("0000-01-01T00:00:00Z").SetValue(&sc.limitMinTime)
+	conf := &sidecarConfig{}
+	cmd.Flag("upload-jitter", "Maximum random delay before uploading blocks.").Default("0s").DurationVar(&conf.shipper.uploadJitter)
 }

docs/components/sidecar.md

@@ -228,6 +228,7 @@ Flags:
                                  configuration. See format details:
                                  https://thanos.io/tip/thanos/tracing.md/#configuration
       --tsdb.path="./data"       Data directory of TSDB.
+      --upload-jitter=0s         Maximum random delay before uploading blocks.
       --version                  Show application version.
 
 ```

pkg/runutil/runutil.go

@@ -12,7 +12,10 @@
 //	err := runutil.Repeat(10*time.Second, stopc, func() error {
 //		// ...
 //	})
-//
+//err := runutil.RepeatWithJitter(30*time.Second, ctx, 0.05, func() error {
+// Your code here
+//})
+
 // Retry starts executing closure function f until no error is returned from f:
 //
 //	err := runutil.Retry(10*time.Second, stopc, func() error {
@@ -50,8 +53,10 @@
 package runutil
 
 import (
+	"context"
 	"fmt"
 	"io"
+	"math/rand"
 	"os"
 	"path/filepath"
 	"strings"
@@ -99,6 +104,32 @@ func Repeat(interval time.Duration, stopc <-chan struct{}, f func() error) error
 	}
 }
 
+// RepeatWithJitter executes f with a random jitter added to the interval between each execution.
+// It continues until ctx is done or f returns an error.
+// The jitter factor should be between 0 and 1, where 0 means no jitter and 1 means the interval can vary from 0 to 2 times the original interval.
+func RepeatWithJitter(interval time.Duration, ctx context.Context, jitterFactor float64, f func() error) error {
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		default:
+			if err := f(); err != nil {
+				return err
+			}
+
+			jitter := time.Duration(float64(interval) * jitterFactor)
+
+			jitteredInterval := interval + time.Duration(rand.Float64()*float64(jitter))
+
+			select {
+			case <-ctx.Done():
+				return nil
+			case <-time.After(jitteredInterval):
+			}
+		}
+	}
+}
+
 // Retry executes f every interval seconds until timeout or no error is returned from f.
 func Retry(interval time.Duration, stopc <-chan struct{}, f func() error) error {
 	return RetryWithLog(log.NewNopLogger(), interval, stopc, f)