lint

Signed-off-by: Alan Protasio <[email protected]>

Author: alanprot

pkg/querier/blocks_finder_bucket_index.go

@@ -4,13 +4,14 @@ import (
 	"context"
 	"time"
 
-	"github.com/cortexproject/cortex/pkg/util/validation"
 	"github.com/go-kit/log"
 	"github.com/oklog/ulid"
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/thanos-io/objstore"
 
+	"github.com/cortexproject/cortex/pkg/util/validation"
+
 	"github.com/cortexproject/cortex/pkg/storage/bucket"
 	"github.com/cortexproject/cortex/pkg/storage/tsdb/bucketindex"
 	"github.com/cortexproject/cortex/pkg/util/services"

pkg/querier/blocks_finder_bucket_index_test.go

@@ -7,13 +7,14 @@ import (
 	"testing"
 	"time"
 
-	"github.com/cortexproject/cortex/pkg/util/validation"
 	"github.com/go-kit/log"
 	"github.com/oklog/ulid"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/thanos-io/objstore"
 
+	"github.com/cortexproject/cortex/pkg/util/validation"
+
 	"github.com/cortexproject/cortex/pkg/storage/tsdb/bucketindex"
 	cortex_testutil "github.com/cortexproject/cortex/pkg/storage/tsdb/testutil"
 	"github.com/cortexproject/cortex/pkg/util/services"

pkg/storage/tsdb/bucketindex/loader_test.go

@@ -604,7 +604,7 @@ func TestLoader_ShouldUpdateIndexInBackgroundOnPreviousKeyAcessDenied(t *testing
 	require.True(t, errors.Is(err, ErrCustomerManagedKeyError))
 
 	// Check cached
-	loader.checkCachedIndexes(ctx)
+	require.NoError(t, loader.checkCachedIndexes(ctx))
 
 	loader.bkt = bkt
 
@@ -623,7 +623,7 @@ func TestLoader_ShouldUpdateIndexInBackgroundOnPreviousKeyAcessDenied(t *testing
 	test.Poll(t, 3*time.Second, nil, func() interface{} {
 		_, err := loader.GetIndex(ctx, "user-1")
 		// Check cached
-		loader.checkCachedIndexes(ctx)
+		require.NoError(t, loader.checkCachedIndexes(ctx))
 		return err
 	})
 

pkg/storage/tsdb/bucketindex/storage.go

@@ -6,11 +6,12 @@ import (
 	"context"
 	"encoding/json"
 
-	"github.com/cortexproject/cortex/pkg/storage/tsdb"
 	"github.com/go-kit/log"
 	"github.com/pkg/errors"
 	"github.com/thanos-io/objstore"
 
+	"github.com/cortexproject/cortex/pkg/storage/tsdb"
+
 	"github.com/cortexproject/cortex/pkg/storage/bucket"
 	"github.com/cortexproject/cortex/pkg/util/runutil"
 )

pkg/storage/tsdb/bucketindex/updater.go

@@ -7,7 +7,6 @@ import (
 	"path"
 	"time"
 
-	"github.com/cortexproject/cortex/pkg/storage/tsdb"
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
 	"github.com/oklog/ulid"
@@ -16,17 +15,20 @@ import (
 	"github.com/thanos-io/thanos/pkg/block"
 	"github.com/thanos-io/thanos/pkg/block/metadata"
 
+	"github.com/cortexproject/cortex/pkg/storage/tsdb"
+
 	"github.com/cortexproject/cortex/pkg/storage/bucket"
 	util_log "github.com/cortexproject/cortex/pkg/util/log"
 	"github.com/cortexproject/cortex/pkg/util/runutil"
 )
 
 var (
-	ErrBlockMetaNotFound           = block.ErrorSyncMetaNotFound
-	ErrBlockMetaCorrupted          = block.ErrorSyncMetaCorrupted
-	ErrBlockMetaKeyAccessDeniedErr = errors.New("block meta file key access denied error")
-	ErrBlockDeletionMarkNotFound   = errors.New("block deletion mark not found")
-	ErrBlockDeletionMarkCorrupted  = errors.New("block deletion mark corrupted")
+	ErrBlockMetaNotFound          = block.ErrorSyncMetaNotFound
+	ErrBlockMetaCorrupted         = block.ErrorSyncMetaCorrupted
+	ErrBlockDeletionMarkNotFound  = errors.New("block deletion mark not found")
+	ErrBlockDeletionMarkCorrupted = errors.New("block deletion mark corrupted")
+
+	errBlockMetaKeyAccessDeniedErr = errors.New("block meta file key access denied error")
 )
 
 // Updater is responsible to generate an update in-memory bucket index.
@@ -110,7 +112,7 @@ func (w *Updater) updateBlocks(ctx context.Context, old []*Block) (blocks []*Blo
 			level.Warn(w.logger).Log("msg", "skipped partial block when updating bucket index", "block", id.String())
 			continue
 		}
-		if errors.Is(err, ErrBlockMetaKeyAccessDeniedErr) {
+		if errors.Is(err, errBlockMetaKeyAccessDeniedErr) {
 			partials[id] = err
 			level.Warn(w.logger).Log("msg", "skipped partial block when updating bucket index due key permission", "block", id.String())
 			continue
@@ -135,7 +137,7 @@ func (w *Updater) updateBlockIndexEntry(ctx context.Context, id ulid.ULID) (*Blo
 		return nil, ErrBlockMetaNotFound
 	}
 	if w.bkt.IsCustomerManagedKeyError(err) {
-		return nil, ErrBlockMetaKeyAccessDeniedErr
+		return nil, errBlockMetaKeyAccessDeniedErr
 	}
 	if err != nil {
 		return nil, errors.Wrapf(err, "get block meta file: %v", metaFile)

pkg/storage/tsdb/bucketindex/updater_test.go

@@ -164,7 +164,7 @@ func TestUpdater_UpdateIndex_ShouldNotIncreaseOperationFailureMetricCustomerKey(
 	idx, partials, _, err := w.UpdateIndex(ctx, nil)
 	require.NoError(t, err)
 	assert.Len(t, partials, 1)
-	assert.True(t, errors.Is(partials[block2.ULID], ErrBlockMetaKeyAccessDeniedErr))
+	assert.True(t, errors.Is(partials[block2.ULID], errBlockMetaKeyAccessDeniedErr))
 	assertBucketIndexEqual(t, idx, bkt, userID,
 		[]tsdb.BlockMeta{block1},
 		[]*metadata.DeletionMark{})

pkg/storage/tsdb/testutil/objstore.go

@@ -6,11 +6,12 @@ import (
 	"os"
 	"testing"
 
-	"github.com/cortexproject/cortex/pkg/util"
 	"github.com/pkg/errors"
 	"github.com/stretchr/testify/require"
 	"github.com/thanos-io/objstore"
 
+	"github.com/cortexproject/cortex/pkg/util"
+
 	"github.com/cortexproject/cortex/pkg/storage/bucket/filesystem"
 )
 

pkg/storegateway/bucket_index_metadata_fetcher.go

@@ -95,15 +95,14 @@ func (f *BucketIndexMetadataFetcher) Fetch(ctx context.Context) (metas map[ulid.
 		return nil, nil, nil
 	}
 
-	if errors.Is(err, bucketindex.ErrBlockMetaKeyAccessDeniedErr) {
-		// Do not fail if the permission to the bucket key got revoked. We'll act as if the tenant has no bucket index, but the query
-		// will fail anyway in the querier (the querier fails in the querier if it cannot load the bucket index).
-		// This will cause the store-gateway to unload all blocks
+	if errors.Is(err, bucketindex.ErrCustomerManagedKeyError) {
+		// stop the job and return the error
+		// this error should be used to return Access Denied to the caller
 		level.Error(f.logger).Log("msg", "bucket index key permission revoked", "user", f.userID, "err", err)
 		f.metrics.Synced.WithLabelValues(keyAccessDenied).Set(1)
 		f.metrics.Submit()
 
-		return nil, nil, nil
+		return nil, nil, bucketindex.ErrCustomerManagedKeyError
 	}
 
 	if err != nil {

pkg/storegateway/bucket_index_metadata_fetcher_test.go

@@ -106,11 +106,11 @@ func TestBucketIndexMetadataFetcher_Fetch_KeyPermissionDenied(t *testing.T) {
 	reg := prometheus.NewPedanticRegistry()
 	ctx := context.Background()
 
-	bkt.MockGet(userID+"/bucket-index.json.gz", "c", bucketindex.ErrBlockMetaKeyAccessDeniedErr)
+	bkt.MockGet(userID+"/bucket-index.json.gz", "c", bucketindex.ErrCustomerManagedKeyError)
 
 	fetcher := NewBucketIndexMetadataFetcher(userID, bkt, NewNoShardingStrategy(), nil, log.NewNopLogger(), reg, nil)
 	metas, _, err := fetcher.Fetch(ctx)
-	require.NoError(t, err)
+	require.ErrorIs(t, bucketindex.ErrCustomerManagedKeyError, err)
 	assert.Empty(t, metas)
 
 	assert.NoError(t, testutil.GatherAndCompare(reg, bytes.NewBufferString(`

pkg/storegateway/bucket_stores.go

@@ -29,8 +29,11 @@ import (
 	"github.com/thanos-io/thanos/pkg/store/storepb"
 	"github.com/weaveworks/common/httpgrpc"
 	"github.com/weaveworks/common/logging"
+	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/metadata"
 
+	"github.com/cortexproject/cortex/pkg/storage/tsdb/bucketindex"
+
 	"github.com/cortexproject/cortex/pkg/storage/bucket"
 	"github.com/cortexproject/cortex/pkg/storage/tsdb"
 	"github.com/cortexproject/cortex/pkg/util/backoff"
@@ -64,7 +67,7 @@ type BucketStores struct {
 
 	// Keeps a bucket store for each tenant.
 	storesMu sync.RWMutex
-	stores   map[string]*store.BucketStore
+	stores   map[string]*BucketStoreWithLastError
 
 	// Metrics.
 	syncTimes         prometheus.Histogram
@@ -73,6 +76,11 @@ type BucketStores struct {
 	tenantsSynced     prometheus.Gauge
 }
 
+type BucketStoreWithLastError struct {
+	*store.BucketStore
+	err error
+}
+
 // NewBucketStores makes a new BucketStores.
 func NewBucketStores(cfg tsdb.BlocksStorageConfig, shardingStrategy ShardingStrategy, bucketClient objstore.Bucket, limits *validation.Overrides, logLevel logging.Level, logger log.Logger, reg prometheus.Registerer) (*BucketStores, error) {
 	cachingBucket, err := tsdb.CreateCachingBucket(cfg.BucketStore.ChunksCache, cfg.BucketStore.MetadataCache, bucketClient, logger, reg)
@@ -94,7 +102,7 @@ func NewBucketStores(cfg tsdb.BlocksStorageConfig, shardingStrategy ShardingStra
 		limits:             limits,
 		bucket:             cachingBucket,
 		shardingStrategy:   shardingStrategy,
-		stores:             map[string]*store.BucketStore{},
+		stores:             map[string]*BucketStoreWithLastError{},
 		logLevel:           logLevel,
 		bucketStoreMetrics: NewBucketStoreMetrics(),
 		metaFetcherMetrics: NewMetadataFetcherMetrics(),
@@ -192,7 +200,7 @@ func (u *BucketStores) syncUsersBlocks(ctx context.Context, f func(context.Conte
 
 	type job struct {
 		userID string
-		store  *store.BucketStore
+		store  *BucketStoreWithLastError
 	}
 
 	wg := &sync.WaitGroup{}
@@ -225,10 +233,16 @@ func (u *BucketStores) syncUsersBlocks(ctx context.Context, f func(context.Conte
 			defer wg.Done()
 
 			for job := range jobs {
-				if err := f(ctx, job.store); err != nil {
-					errsMx.Lock()
-					errs.Add(errors.Wrapf(err, "failed to synchronize TSDB blocks for user %s", job.userID))
-					errsMx.Unlock()
+				if err := f(ctx, job.store.BucketStore); err != nil {
+					if errors.Is(err, bucketindex.ErrCustomerManagedKeyError) {
+						job.store.err = err
+					} else {
+						errsMx.Lock()
+						errs.Add(errors.Wrapf(err, "failed to synchronize TSDB blocks for user %s", job.userID))
+						errsMx.Unlock()
+					}
+				} else {
+					job.store.err = nil
 				}
 			}
 		}()
@@ -286,10 +300,20 @@ func (u *BucketStores) Series(req *storepb.SeriesRequest, srv storepb.Store_Seri
 		return nil
 	}
 
-	return store.Series(req, spanSeriesServer{
+	if store.err != nil && errors.Is(store.err, bucketindex.ErrCustomerManagedKeyError) {
+		return httpgrpc.Errorf(int(codes.ResourceExhausted), "store error: %s", store.err)
+	}
+
+	err := store.Series(req, spanSeriesServer{
 		Store_SeriesServer: srv,
 		ctx:                spanCtx,
 	})
+
+	if err != nil && errors.Is(err, bucketindex.ErrCustomerManagedKeyError) {
+		return httpgrpc.Errorf(int(codes.ResourceExhausted), "store error: %s", err)
+	}
+
+	return err
 }
 
 // LabelNames implements the Storegateway proto service.
@@ -344,7 +368,7 @@ func (u *BucketStores) scanUsers(ctx context.Context) ([]string, error) {
 	return users, err
 }
 
-func (u *BucketStores) getStore(userID string) *store.BucketStore {
+func (u *BucketStores) getStore(userID string) *BucketStoreWithLastError {
 	u.storesMu.RLock()
 	defer u.storesMu.RUnlock()
 	return u.stores[userID]
@@ -387,7 +411,7 @@ func (u *BucketStores) closeEmptyBucketStore(userID string) error {
 	return bs.Close()
 }
 
-func isEmptyBucketStore(bs *store.BucketStore) bool {
+func isEmptyBucketStore(bs *BucketStoreWithLastError) bool {
 	min, max := bs.TimeRange()
 	return min == math.MaxInt64 && max == math.MinInt64
 }
@@ -396,7 +420,7 @@ func (u *BucketStores) syncDirForUser(userID string) string {
 	return filepath.Join(u.cfg.BucketStore.SyncDir, userID)
 }
 
-func (u *BucketStores) getOrCreateStore(userID string) (*store.BucketStore, error) {
+func (u *BucketStores) getOrCreateStore(userID string) (*BucketStoreWithLastError, error) {
 	// Check if the store already exists.
 	bs := u.getStore(userID)
 	if bs != nil {
@@ -500,7 +524,7 @@ func (u *BucketStores) getOrCreateStore(userID string) (*store.BucketStore, erro
 		bucketStoreOpts = append(bucketStoreOpts, store.WithDebugLogging())
 	}
 
-	bs, err := store.NewBucketStore(
+	s, err := store.NewBucketStore(
 		userBkt,
 		fetcher,
 		u.syncDirForUser(userID),
@@ -520,6 +544,10 @@ func (u *BucketStores) getOrCreateStore(userID string) (*store.BucketStore, erro
 		return nil, err
 	}
 
+	bs = &BucketStoreWithLastError{
+		BucketStore: s,
+	}
+
 	u.stores[userID] = bs
 	u.metaFetcherMetrics.AddUserRegistry(userID, fetcherReg)
 	u.bucketStoreMetrics.AddUserRegistry(userID, bucketStoreReg)