Fix retries on store-gateway initial sync failure

CHANGELOG.md

@@ -43,6 +43,7 @@
 * [ENHANCEMENT] Block Storage Ingester: `/flush` now accepts two new parameters: `tenant` to specify tenant to flush and `wait=true` to make call synchronous. Multiple tenants can be specified by repeating `tenant` parameter. If no `tenant` is specified, all tenants are flushed, as before. #4073
 * [ENHANCEMENT] Alertmanager: validate configured `-alertmanager.web.external-url` and fail if ends with `/`. #4081
 * [ENHANCEMENT] Allow configuration of Cassandra's host selection policy. #4069
+* [ENHANCEMENT] Store-gateway: retry synching blocks if a per-tenant sync fails. #3975 #4088
 * [BUGFIX] Ruler-API: fix bug where `/api/v1/rules/<namespace>/<group_name>` endpoint return `400` instead of `404`. #4013
 * [BUGFIX] Distributor: reverted changes done to rate limiting in #3825. #3948
 * [BUGFIX] Ingester: Fix race condition when opening and closing tsdb concurrently. #3959

pkg/storegateway/bucket_stores.go

@@ -141,7 +141,7 @@ func NewBucketStores(cfg tsdb.BlocksStorageConfig, shardingStrategy ShardingStra
 func (u *BucketStores) InitialSync(ctx context.Context) error {
 	level.Info(u.logger).Log("msg", "synchronizing TSDB blocks for all users")
 
-	if err := u.syncUsersBlocks(ctx, func(ctx context.Context, s *store.BucketStore) error {
+	if err := u.syncUsersBlocksWithRetries(ctx, func(ctx context.Context, s *store.BucketStore) error {
 		return s.InitialSync(ctx)
 	}); err != nil {
 		level.Warn(u.logger).Log("msg", "failed to synchronize TSDB blocks", "err", err)
@@ -161,7 +161,7 @@ func (u *BucketStores) SyncBlocks(ctx context.Context) error {
 
 func (u *BucketStores) syncUsersBlocksWithRetries(ctx context.Context, f func(context.Context, *store.BucketStore) error) error {
 	retries := util.NewBackoff(ctx, util.BackoffConfig{
-		MinBackoff: 100 * time.Millisecond,
+		MinBackoff: 1 * time.Second,
 		MaxBackoff: 10 * time.Second,
 		MaxRetries: 3,
 	})

pkg/storegateway/bucket_stores_test.go

@@ -2,7 +2,9 @@ package storegateway
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"math"
 	"os"
@@ -24,6 +26,7 @@ import (
 	"github.com/stretchr/testify/require"
 	thanos_metadata "github.com/thanos-io/thanos/pkg/block/metadata"
 	"github.com/thanos-io/thanos/pkg/extprom"
+	"github.com/thanos-io/thanos/pkg/objstore"
 	"github.com/thanos-io/thanos/pkg/store"
 	"github.com/thanos-io/thanos/pkg/store/labelpb"
 	"github.com/thanos-io/thanos/pkg/store/storepb"
@@ -118,6 +121,68 @@ func TestBucketStores_InitialSync(t *testing.T) {
 	assert.Greater(t, testutil.ToFloat64(stores.syncLastSuccess), float64(0))
 }
 
+func TestBucketStores_InitialSyncShouldRetryOnFailure(t *testing.T) {
+	ctx := context.Background()
+	cfg, cleanup := prepareStorageConfig(t)
+	defer cleanup()
+
+	storageDir, err := ioutil.TempDir(os.TempDir(), "storage-*")
+	require.NoError(t, err)
+
+	// Generate a block for the user in the storage.
+	generateStorageBlock(t, storageDir, "user-1", "series_1", 10, 100, 15)
+
+	bucket, err := filesystem.NewBucketClient(filesystem.Config{Directory: storageDir})
+	require.NoError(t, err)
+
+	// Wrap the bucket to fail the 1st Get() request.
+	bucket = &failFirstGetBucket{Bucket: bucket}
+
+	reg := prometheus.NewPedanticRegistry()
+	stores, err := NewBucketStores(cfg, NewNoShardingStrategy(), bucket, defaultLimitsOverrides(t), mockLoggingLevel(), log.NewNopLogger(), reg)
+	require.NoError(t, err)
+
+	// Initial sync should succeed even if a transient error occurs.
+	require.NoError(t, stores.InitialSync(ctx))
+
+	// Query series after the initial sync.
+	seriesSet, warnings, err := querySeries(stores, "user-1", "series_1", 20, 40)
+	require.NoError(t, err)
+	assert.Empty(t, warnings)
+	require.Len(t, seriesSet, 1)
+	assert.Equal(t, []labelpb.ZLabel{{Name: labels.MetricName, Value: "series_1"}}, seriesSet[0].Labels)
+
+	assert.NoError(t, testutil.GatherAndCompare(reg, strings.NewReader(`
+			# HELP cortex_blocks_meta_syncs_total Total blocks metadata synchronization attempts
+			# TYPE cortex_blocks_meta_syncs_total counter
+			cortex_blocks_meta_syncs_total 2
+
+			# HELP cortex_blocks_meta_sync_failures_total Total blocks metadata synchronization failures
+			# TYPE cortex_blocks_meta_sync_failures_total counter
+			cortex_blocks_meta_sync_failures_total 1
+
+			# HELP cortex_bucket_store_blocks_loaded Number of currently loaded blocks.
+			# TYPE cortex_bucket_store_blocks_loaded gauge
+			cortex_bucket_store_blocks_loaded 1
+
+			# HELP cortex_bucket_store_block_loads_total Total number of remote block loading attempts.
+			# TYPE cortex_bucket_store_block_loads_total counter
+			cortex_bucket_store_block_loads_total 1
+
+			# HELP cortex_bucket_store_block_load_failures_total Total number of failed remote block loading attempts.
+			# TYPE cortex_bucket_store_block_load_failures_total counter
+			cortex_bucket_store_block_load_failures_total 0
+	`),
+		"cortex_blocks_meta_syncs_total",
+		"cortex_blocks_meta_sync_failures_total",
+		"cortex_bucket_store_block_loads_total",
+		"cortex_bucket_store_block_load_failures_total",
+		"cortex_bucket_store_blocks_loaded",
+	))
+
+	assert.Greater(t, testutil.ToFloat64(stores.syncLastSuccess), float64(0))
+}
+
 func TestBucketStores_SyncBlocks(t *testing.T) {
 	const (
 		userID     = "user-1"
@@ -534,3 +599,18 @@ func (u *userShardingStrategy) FilterBlocks(ctx context.Context, userID string,
 	}
 	return nil
 }
+
+// failFirstGetBucket is an objstore.Bucket wrapper which fails the first Get() request with a mocked error.
+type failFirstGetBucket struct {
+	objstore.Bucket
+
+	firstGet atomic.Bool
+}
+
+func (f *failFirstGetBucket) Get(ctx context.Context, name string) (io.ReadCloser, error) {
+	if f.firstGet.CAS(false, true) {
+		return nil, errors.New("Get() request mocked error")
+	}
+
+	return f.Bucket.Get(ctx, name)
+}