rollback InitialSync

Signed-off-by: SungJin1212 <tjdwls1201@gmail.com>

Author: SungJin1212

pkg/storegateway/bucket_store_metrics.go

@@ -7,6 +7,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/util"
 )
 
+// CortexBucketStoreMetrics common metrics in thanos and parquet block stores (in future)
 type CortexBucketStoreMetrics struct {
 	syncTimes         prometheus.Histogram
 	syncLastSuccess   prometheus.Gauge

pkg/storegateway/bucket_stores.go

@@ -103,7 +103,7 @@ var ErrTooManyInflightRequests = status.Error(codes.ResourceExhausted, "too many
 func NewBucketStores(cfg tsdb.BlocksStorageConfig, shardingStrategy ShardingStrategy, bucketClient objstore.InstrumentedBucket, limits *validation.Overrides, logLevel logging.Level, logger log.Logger, reg prometheus.Registerer) (BucketStores, error) {
 	switch cfg.BucketStore.BucketStoreType {
 	case string(tsdb.ParquetBucketStore):
-		return newParquetBucketStores(cfg, shardingStrategy, bucketClient, limits, logger, reg)
+		return newParquetBucketStores(cfg, bucketClient, limits, logger, reg)
 	case string(tsdb.TSDBBucketStore):
 		return newThanosBucketStores(cfg, shardingStrategy, bucketClient, limits, logLevel, logger, reg)
 	default:

pkg/storegateway/parquet_bucket_stores.go

@@ -7,7 +7,6 @@ import (
 	"path/filepath"
 	"sort"
 	"sync"
-	"time"
 
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
@@ -22,7 +21,6 @@ import (
 	"github.com/prometheus/prometheus/model/labels"
 	prom_storage "github.com/prometheus/prometheus/storage"
 	"github.com/prometheus/prometheus/tsdb/chunkenc"
-	tsdb_errors "github.com/prometheus/prometheus/tsdb/errors"
 	"github.com/thanos-io/objstore"
 	storecache "github.com/thanos-io/thanos/pkg/store/cache"
 	"github.com/thanos-io/thanos/pkg/store/storepb"
@@ -36,7 +34,6 @@ import (
 	"github.com/cortexproject/cortex/pkg/storage/tsdb"
 	"github.com/cortexproject/cortex/pkg/storage/tsdb/users"
 	cortex_util "github.com/cortexproject/cortex/pkg/util"
-	"github.com/cortexproject/cortex/pkg/util/backoff"
 	cortex_errors "github.com/cortexproject/cortex/pkg/util/errors"
 	"github.com/cortexproject/cortex/pkg/util/spanlogger"
 	"github.com/cortexproject/cortex/pkg/util/validation"
@@ -64,14 +61,13 @@ type ParquetBucketStores struct {
 
 	cortexBucketStoreMetrics *CortexBucketStoreMetrics
 	userScanner              users.Scanner
-	shardingStrategy         ShardingStrategy
 
 	userTokenBucketsMu sync.RWMutex
 	userTokenBuckets   map[string]*cortex_util.TokenBucket
 }
 
 // newParquetBucketStores creates a new multi-tenant parquet bucket stores
-func newParquetBucketStores(cfg tsdb.BlocksStorageConfig, shardingStrategy ShardingStrategy, bucketClient objstore.InstrumentedBucket, limits *validation.Overrides, logger log.Logger, reg prometheus.Registerer) (*ParquetBucketStores, error) {
+func newParquetBucketStores(cfg tsdb.BlocksStorageConfig, bucketClient objstore.InstrumentedBucket, limits *validation.Overrides, logger log.Logger, reg prometheus.Registerer) (*ParquetBucketStores, error) {
 	// Create caching bucket client for parquet bucket stores
 	cachingBucket, err := createCachingBucketClientForParquet(cfg, bucketClient, "parquet-storegateway", logger, reg)
 	if err != nil {
@@ -88,7 +84,6 @@ func newParquetBucketStores(cfg tsdb.BlocksStorageConfig, shardingStrategy Shard
 		chunksDecoder:            schema.NewPrometheusParquetChunksDecoder(chunkenc.NewPool()),
 		inflightRequests:         cortex_util.NewInflightRequestTracker(),
 		cortexBucketStoreMetrics: NewCortexBucketStoreMetrics(reg),
-		shardingStrategy:         shardingStrategy,
 		userTokenBuckets:         make(map[string]*cortex_util.TokenBucket),
 	}
 	u.userScanner, err = users.NewScanner(cfg.UsersScanner, bucketClient, logger, reg)
@@ -209,147 +204,14 @@ func (u *ParquetBucketStores) LabelValues(ctx context.Context, req *storepb.Labe
 
 // SyncBlocks implements BucketStores
 func (u *ParquetBucketStores) SyncBlocks(ctx context.Context) error {
-	return u.syncUsersBlocksWithRetries(ctx, func(ctx context.Context, p *parquetBucketStore) error {
-		return p.SyncBlocks(ctx)
-	})
+	return nil
 }
 
 // InitialSync implements BucketStores
 func (u *ParquetBucketStores) InitialSync(ctx context.Context) error {
-	level.Info(u.logger).Log("msg", "synchronizing Parquet blocks for all users")
-
-	if err := u.syncUsersBlocksWithRetries(ctx, func(ctx context.Context, p *parquetBucketStore) error {
-		return p.InitialSync(ctx)
-	}); err != nil {
-		level.Warn(u.logger).Log("msg", "failed to synchronize Parquet blocks", "err", err)
-		return err
-	}
-
-	level.Info(u.logger).Log("msg", "successfully synchronized Parquet blocks for all users")
 	return nil
 }
 
-func (u *ParquetBucketStores) syncUsersBlocksWithRetries(ctx context.Context, f func(context.Context, *parquetBucketStore) error) error {
-	retries := backoff.New(ctx, backoff.Config{
-		MinBackoff: 1 * time.Second,
-		MaxBackoff: 10 * time.Second,
-		MaxRetries: 3,
-	})
-
-	var lastErr error
-	for retries.Ongoing() {
-		lastErr = u.syncUsersBlocks(ctx, f)
-		if lastErr == nil {
-			return nil
-		}
-
-		retries.Wait()
-	}
-
-	if lastErr == nil {
-		return retries.Err()
-	}
-
-	return lastErr
-}
-
-func (u *ParquetBucketStores) syncUsersBlocks(ctx context.Context, f func(context.Context, *parquetBucketStore) error) (returnErr error) {
-	defer func(start time.Time) {
-		u.cortexBucketStoreMetrics.syncTimes.Observe(time.Since(start).Seconds())
-		if returnErr == nil {
-			u.cortexBucketStoreMetrics.syncLastSuccess.SetToCurrentTime()
-		}
-	}(time.Now())
-
-	type job struct {
-		userID string
-		store  *parquetBucketStore
-	}
-
-	wg := &sync.WaitGroup{}
-	jobs := make(chan job)
-	errs := tsdb_errors.NewMulti()
-	errsMx := sync.Mutex{}
-
-	// Scan users in the bucket.
-	userIDs, err := u.scanUsers(ctx)
-	if err != nil {
-		return err
-	}
-
-	includeUserIDs := make(map[string]struct{})
-	for _, userID := range u.shardingStrategy.FilterUsers(ctx, userIDs) {
-		includeUserIDs[userID] = struct{}{}
-	}
-
-	u.cortexBucketStoreMetrics.tenantsDiscovered.Set(float64(len(userIDs)))
-	u.cortexBucketStoreMetrics.tenantsSynced.Set(float64(len(includeUserIDs)))
-
-	// Create a pool of workers which will synchronize blocks. The pool size
-	// is limited in order to avoid to concurrently sync a lot of tenants in
-	// a large cluster.
-	for i := 0; i < u.cfg.BucketStore.TenantSyncConcurrency; i++ {
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-
-			for job := range jobs {
-				if err := f(ctx, job.store); err != nil {
-					if errors.Is(err, bucket.ErrCustomerManagedKeyAccessDenied) {
-						u.storesErrorsMu.Lock()
-						u.storesErrors[job.userID] = httpgrpc.Errorf(int(codes.PermissionDenied), "store error: %s", err)
-						u.storesErrorsMu.Unlock()
-					} else {
-						errsMx.Lock()
-						errs.Add(errors.Wrapf(err, "failed to synchronize Parquet blocks for user %s", job.userID))
-						errsMx.Unlock()
-					}
-				} else {
-					u.storesErrorsMu.Lock()
-					delete(u.storesErrors, job.userID)
-					u.storesErrorsMu.Unlock()
-				}
-			}
-		}()
-	}
-
-	// Lazily create a bucket store for each new user found
-	// and submit a sync job for each user.
-	for _, userID := range userIDs {
-		// If we don't have a store for the tenant yet, then we should skip it if it's not
-		// included in the store-gateway shard. If we already have it, we need to sync it
-		// anyway to make sure all its blocks are unloaded and metrics updated correctly
-		// (but bucket API calls are skipped thanks to the objstore client adapter).
-		if _, included := includeUserIDs[userID]; !included && u.getStore(userID) == nil {
-			continue
-		}
-
-		bs, err := u.getOrCreateStore(userID)
-		if err != nil {
-			errsMx.Lock()
-			errs.Add(err)
-			errsMx.Unlock()
-
-			continue
-		}
-
-		select {
-		case jobs <- job{userID: userID, store: bs}:
-			// Nothing to do. Will loop to push more jobs.
-		case <-ctx.Done():
-			return ctx.Err()
-		}
-	}
-
-	// Wait until all workers completed.
-	close(jobs)
-	wg.Wait()
-
-	u.deleteLocalFilesForExcludedTenants(includeUserIDs)
-
-	return errs.Err()
-}
-
 // deleteLocalFilesForExcludedTenants removes local "sync" directories for tenants that are not included in the current
 // shard.
 func (u *ParquetBucketStores) deleteLocalFilesForExcludedTenants(includeUserIDs map[string]struct{}) {

pkg/storegateway/parquet_bucket_stores_test.go

@@ -224,7 +224,7 @@ func TestParquetBucketStoresWithCaching(t *testing.T) {
 	limits := validation.NewOverrides(validation.Limits{}, nil)
 
 	// Create parquet bucket stores with caching
-	parquetStores, err := newParquetBucketStores(storageCfg, NewNoShardingStrategy(log.NewNopLogger(), nil), bucketClient, limits, log.NewNopLogger(), prometheus.NewRegistry())
+	parquetStores, err := newParquetBucketStores(storageCfg, bucketClient, limits, log.NewNopLogger(), prometheus.NewRegistry())
 	require.NoError(t, err)
 	require.NotNil(t, parquetStores)