add test on BucketStores

alanprot · alanprot · commit 7ee5d1a70249 · 2023-07-03T09:33:22.000-07:00
Signed-off-by: Alan Protasio &lt;alanprot@gmail.com&gt;
diff --git a/pkg/compactor/blocks_cleaner.go b/pkg/compactor/blocks_cleaner.go
@@ -325,7 +325,7 @@ func (c *BlocksCleaner) cleanUser(ctx context.Context, userID string, firstRun b
 	idx, err := bucketindex.ReadIndex(ctx, c.bucketClient, userID, c.cfgProvider, c.logger)
 	if errors.Is(err, bucketindex.ErrIndexCorrupted) {
 		level.Warn(userLogger).Log("msg", "found a corrupted bucket index, recreating it")
-	} else if errors.Is(err, bucketindex.ErrCustomerManagedKeyError) {
+	} else if errors.Is(err, bucket.ErrCustomerManagedKeyError) {
 		// Give up cleaning if we get access denied
 		level.Warn(userLogger).Log("msg", err.Error())
 		return nil
diff --git a/pkg/querier/blocks_finder_bucket_index.go b/pkg/querier/blocks_finder_bucket_index.go
@@ -65,7 +65,7 @@ func (f *BucketIndexBlocksFinder) GetBlocks(ctx context.Context, userID string,
 		return nil, nil, nil
 	}
 
-	if errors.Is(err, bucketindex.ErrCustomerManagedKeyError) {
+	if errors.Is(err, bucket.ErrCustomerManagedKeyError) {
 		return nil, nil, validation.AccessDeniedError(err.Error())
 	}
 
diff --git a/pkg/storage/bucket/client.go b/pkg/storage/bucket/client.go
@@ -40,6 +40,8 @@ var (
 	SupportedBackends = []string{S3, GCS, Azure, Swift, Filesystem}
 
 	ErrUnsupportedStorageBackend = errors.New("unsupported storage backend")
+
+	ErrCustomerManagedKeyError = errors.New("access denied: customer key")
 )
 
 // Config holds configuration for accessing long-term storage.
diff --git a/pkg/storage/tsdb/bucketindex/loader.go b/pkg/storage/tsdb/bucketindex/loader.go
@@ -115,7 +115,7 @@ func (l *Loader) GetIndex(ctx context.Context, userID string) (*Index, error) {
 
 		if errors.Is(err, ErrIndexNotFound) {
 			level.Warn(l.logger).Log("msg", "bucket index not found", "user", userID)
-		} else if errors.Is(err, ErrCustomerManagedKeyError) {
+		} else if errors.Is(err, bucket.ErrCustomerManagedKeyError) {
 			level.Warn(l.logger).Log("msg", "key access denied when reading bucket index", "user", userID)
 		} else {
 			// We don't track ErrIndexNotFound as failure because it's a legit case (eg. a tenant just
@@ -198,7 +198,7 @@ func (l *Loader) updateCachedIndex(ctx context.Context, userID string) {
 	l.loadAttempts.Inc()
 	startTime := time.Now()
 	idx, err := ReadIndex(readCtx, l.bkt, userID, l.cfgProvider, l.logger)
-	if err != nil && !errors.Is(err, ErrIndexNotFound) && !errors.Is(err, ErrCustomerManagedKeyError) {
+	if err != nil && !errors.Is(err, ErrIndexNotFound) && !errors.Is(err, bucket.ErrCustomerManagedKeyError) {
 		l.loadFailures.Inc()
 		level.Warn(l.logger).Log("msg", "unable to update bucket index", "user", userID, "err", err)
 		return
diff --git a/pkg/storage/tsdb/bucketindex/loader_test.go b/pkg/storage/tsdb/bucketindex/loader_test.go
@@ -8,6 +8,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/cortexproject/cortex/pkg/storage/bucket"
 	"github.com/go-kit/log"
 	"github.com/oklog/ulid"
 	"github.com/pkg/errors"
@@ -601,7 +602,7 @@ func TestLoader_ShouldUpdateIndexInBackgroundOnPreviousKeyAcessDenied(t *testing
 	})
 
 	_, err := loader.GetIndex(ctx, user)
-	require.True(t, errors.Is(err, ErrCustomerManagedKeyError))
+	require.True(t, errors.Is(err, bucket.ErrCustomerManagedKeyError))
 
 	// Check cached
 	require.NoError(t, loader.checkCachedIndexes(ctx))
@@ -667,7 +668,7 @@ func TestLoader_GetIndex_ShouldCacheKeyDeniedErrors(t *testing.T) {
 	// Request the index multiple times.
 	for i := 0; i < 10; i++ {
 		_, err := loader.GetIndex(ctx, "user-1")
-		require.True(t, errors.Is(err, ErrCustomerManagedKeyError))
+		require.True(t, errors.Is(err, bucket.ErrCustomerManagedKeyError))
 	}
 
 	// Ensure metrics have been updated accordingly.
diff --git a/pkg/storage/tsdb/bucketindex/storage.go b/pkg/storage/tsdb/bucketindex/storage.go
@@ -17,9 +17,8 @@ import (
 )
 
 var (
-	ErrIndexNotFound           = errors.New("bucket index not found")
-	ErrIndexCorrupted          = errors.New("bucket index corrupted")
-	ErrCustomerManagedKeyError = errors.New("access denied: customer key")
+	ErrIndexNotFound  = errors.New("bucket index not found")
+	ErrIndexCorrupted = errors.New("bucket index corrupted")
 )
 
 // ReadIndex reads, parses and returns a bucket index from the bucket.
@@ -34,7 +33,7 @@ func ReadIndex(ctx context.Context, bkt objstore.Bucket, userID string, cfgProvi
 		}
 
 		if userBkt.IsCustomerManagedKeyError(err) {
-			return nil, ErrCustomerManagedKeyError
+			return nil, bucket.ErrCustomerManagedKeyError
 		}
 
 		return nil, errors.Wrap(err, "read bucket index")
diff --git a/pkg/storage/tsdb/bucketindex/storage_test.go b/pkg/storage/tsdb/bucketindex/storage_test.go
@@ -6,6 +6,7 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/cortexproject/cortex/pkg/storage/bucket"
 	"github.com/go-kit/log"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -45,7 +46,7 @@ func TestReadIndex_ShouldReturnErrorIfKeyAccessDeniedErr(t *testing.T) {
 		},
 	}
 	idx, err := ReadIndex(context.Background(), bkt, "user-1", nil, log.NewNopLogger())
-	require.Equal(t, ErrCustomerManagedKeyError, err)
+	require.Equal(t, bucket.ErrCustomerManagedKeyError, err)
 	require.Nil(t, idx)
 }
 
diff --git a/pkg/storage/tsdb/testutil/objstore.go b/pkg/storage/tsdb/testutil/objstore.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"io"
 	"os"
+	"strings"
 	"testing"
 
 	"github.com/pkg/errors"
@@ -46,6 +47,11 @@ func (m *MockBucketFailure) Delete(ctx context.Context, name string) error {
 }
 
 func (m *MockBucketFailure) Get(ctx context.Context, name string) (io.ReadCloser, error) {
+	for prefix, err := range m.GetFailures {
+		if strings.HasPrefix(name, prefix) {
+			return nil, err
+		}
+	}
 	if e, ok := m.GetFailures[name]; ok {
 		return nil, e
 	}
@@ -58,7 +64,7 @@ func (m *MockBucketFailure) WithExpectedErrs(expectedFunc objstore.IsOpFailureEx
 		return &MockBucketFailure{Bucket: ibkt.WithExpectedErrs(expectedFunc), DeleteFailures: m.DeleteFailures, GetFailures: m.GetFailures}
 	}
 
-	return m.WithExpectedErrs(expectedFunc)
+	return m
 }
 
 func (m *MockBucketFailure) ReaderWithExpectedErrs(expectedFunc objstore.IsOpFailureExpectedFunc) objstore.BucketReader {
diff --git a/pkg/storegateway/bucket_index_metadata_fetcher.go b/pkg/storegateway/bucket_index_metadata_fetcher.go
@@ -68,7 +68,7 @@ func (f *BucketIndexMetadataFetcher) Fetch(ctx context.Context) (metas map[ulid.
 	start := time.Now()
 	defer func() {
 		f.metrics.SyncDuration.Observe(time.Since(start).Seconds())
-		if err != nil {
+		if err != nil && !errors.Is(err, bucket.ErrCustomerManagedKeyError) {
 			f.metrics.SyncFailures.Inc()
 		}
 	}()
@@ -95,14 +95,14 @@ func (f *BucketIndexMetadataFetcher) Fetch(ctx context.Context) (metas map[ulid.
 		return nil, nil, nil
 	}
 
-	if errors.Is(err, bucketindex.ErrCustomerManagedKeyError) {
+	if errors.Is(err, bucket.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, bucketindex.ErrCustomerManagedKeyError
+		return nil, nil, bucket.ErrCustomerManagedKeyError
 	}
 
 	if err != nil {
diff --git a/pkg/storegateway/bucket_index_metadata_fetcher_test.go b/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.ErrCustomerManagedKeyError)
+	bkt.MockGet(userID+"/bucket-index.json.gz", "c", bucket.ErrCustomerManagedKeyError)
 
 	fetcher := NewBucketIndexMetadataFetcher(userID, bkt, NewNoShardingStrategy(), nil, log.NewNopLogger(), reg, nil)
 	metas, _, err := fetcher.Fetch(ctx)
-	require.ErrorIs(t, bucketindex.ErrCustomerManagedKeyError, err)
+	require.ErrorIs(t, bucket.ErrCustomerManagedKeyError, err)
 	assert.Empty(t, metas)
 
 	assert.NoError(t, testutil.GatherAndCompare(reg, bytes.NewBufferString(`
diff --git a/pkg/storegateway/bucket_stores.go b/pkg/storegateway/bucket_stores.go
@@ -32,8 +32,6 @@ import (
 	"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"
@@ -234,7 +232,7 @@ func (u *BucketStores) syncUsersBlocks(ctx context.Context, f func(context.Conte
 
 			for job := range jobs {
 				if err := f(ctx, job.store.BucketStore); err != nil {
-					if errors.Is(err, bucketindex.ErrCustomerManagedKeyError) {
+					if errors.Is(err, bucket.ErrCustomerManagedKeyError) {
 						job.store.err = err
 					} else {
 						errsMx.Lock()
@@ -300,7 +298,7 @@ func (u *BucketStores) Series(req *storepb.SeriesRequest, srv storepb.Store_Seri
 		return nil
 	}
 
-	if store.err != nil && errors.Is(store.err, bucketindex.ErrCustomerManagedKeyError) {
+	if store.err != nil && errors.Is(store.err, bucket.ErrCustomerManagedKeyError) {
 		return httpgrpc.Errorf(int(codes.ResourceExhausted), "store error: %s", store.err)
 	}
 
@@ -309,7 +307,7 @@ func (u *BucketStores) Series(req *storepb.SeriesRequest, srv storepb.Store_Seri
 		ctx:                spanCtx,
 	})
 
-	if err != nil && errors.Is(err, bucketindex.ErrCustomerManagedKeyError) {
+	if err != nil && errors.Is(err, bucket.ErrCustomerManagedKeyError) {
 		return httpgrpc.Errorf(int(codes.ResourceExhausted), "store error: %s", err)
 	}
 
diff --git a/pkg/storegateway/bucket_stores_test.go b/pkg/storegateway/bucket_stores_test.go
@@ -33,13 +33,60 @@ import (
 	"go.uber.org/atomic"
 	"google.golang.org/grpc/metadata"
 
+	cortex_testutil "github.com/cortexproject/cortex/pkg/storage/tsdb/testutil"
+
 	"github.com/cortexproject/cortex/pkg/storage/bucket"
 	"github.com/cortexproject/cortex/pkg/storage/bucket/filesystem"
 	cortex_tsdb "github.com/cortexproject/cortex/pkg/storage/tsdb"
 	"github.com/cortexproject/cortex/pkg/util"
 	"github.com/cortexproject/cortex/pkg/util/flagext"
 )
 
+func TestBucketStores_CustomerKeyError(t *testing.T) {
+	userToMetric := map[string]string{
+		"user-1": "series_1",
+		"user-2": "series_2",
+	}
+
+	ctx := context.Background()
+	cfg := prepareStorageConfig(t)
+	cfg.BucketStore.BucketIndex.Enabled = true
+
+	storageDir := t.TempDir()
+
+	for userID, metricName := range userToMetric {
+		generateStorageBlock(t, storageDir, userID, metricName, 10, 100, 15)
+	}
+
+	b, err := filesystem.NewBucketClient(filesystem.Config{Directory: storageDir})
+
+	mBucket := &cortex_testutil.MockBucketFailure{
+		Bucket: b,
+		GetFailures: map[string]error{
+			"user-1": cortex_testutil.ErrKeyAccessDeniedError,
+		},
+	}
+	require.NoError(t, err)
+
+	reg := prometheus.NewPedanticRegistry()
+	stores, err := NewBucketStores(cfg, NewNoShardingStrategy(), mBucket, defaultLimitsOverrides(t), mockLoggingLevel(), log.NewNopLogger(), reg)
+	require.NoError(t, err)
+
+	// Should set the error on user-1
+	require.NoError(t, stores.InitialSync(ctx))
+	require.ErrorIs(t, stores.stores["user-1"].err, bucket.ErrCustomerManagedKeyError)
+	require.ErrorIs(t, stores.stores["user-2"].err, nil)
+	require.NoError(t, stores.SyncBlocks(context.Background()))
+	require.ErrorIs(t, stores.stores["user-1"].err, bucket.ErrCustomerManagedKeyError)
+	require.ErrorIs(t, stores.stores["user-2"].err, nil)
+
+	// Cleaning the error
+	mBucket.GetFailures = map[string]error{}
+	require.NoError(t, stores.SyncBlocks(context.Background()))
+	require.ErrorIs(t, stores.stores["user-1"].err, nil)
+	require.ErrorIs(t, stores.stores["user-2"].err, nil)
+}
+
 func TestBucketStores_InitialSync(t *testing.T) {
 	t.Parallel()
 	userToMetric := map[string]string{

Original file line number	Diff line number	Diff line change
`@@ -65,7 +65,7 @@ func (f *BucketIndexBlocksFinder) GetBlocks(ctx context.Context, userID string,`
`65`	`65`	`return nil, nil, nil`
`66`	`66`	`}`
`67`	`67`
`68`		`- if errors.Is(err, bucketindex.ErrCustomerManagedKeyError) {`
	`68`	`+ if errors.Is(err, bucket.ErrCustomerManagedKeyError) {`
`69`	`69`	`return nil, nil, validation.AccessDeniedError(err.Error())`
`70`	`70`	`}`
`71`	`71`
Original file line number	Diff line number	Diff line change
`@@ -40,6 +40,8 @@ var (`
`40`	`40`	`SupportedBackends = []string{S3, GCS, Azure, Swift, Filesystem}`
`41`	`41`
`42`	`42`	`ErrUnsupportedStorageBackend = errors.New("unsupported storage backend")`
	`43`	`+`
	`44`	`+ ErrCustomerManagedKeyError = errors.New("access denied: customer key")`
`43`	`45`	`)`
`44`	`46`
`45`	`47`	`// Config holds configuration for accessing long-term storage.`
Original file line number	Diff line number	Diff line change
`@@ -17,9 +17,8 @@ import (`
`17`	`17`	`)`
`18`	`18`
`19`	`19`	`var (`
`20`		`- ErrIndexNotFound = errors.New("bucket index not found")`
`21`		`- ErrIndexCorrupted = errors.New("bucket index corrupted")`
`22`		`- ErrCustomerManagedKeyError = errors.New("access denied: customer key")`
	`20`	`+ ErrIndexNotFound = errors.New("bucket index not found")`
	`21`	`+ ErrIndexCorrupted = errors.New("bucket index corrupted")`
`23`	`22`	`)`
`24`	`23`
`25`	`24`	`// ReadIndex reads, parses and returns a bucket index from the bucket.`
`@@ -34,7 +33,7 @@ func ReadIndex(ctx context.Context, bkt objstore.Bucket, userID string, cfgProvi`
`34`	`33`	`}`
`35`	`34`
`36`	`35`	`if userBkt.IsCustomerManagedKeyError(err) {`
`37`		`- return nil, ErrCustomerManagedKeyError`
	`36`	`+ return nil, bucket.ErrCustomerManagedKeyError`
`38`	`37`	`}`
`39`	`38`
`40`	`39`	`return nil, errors.Wrap(err, "read bucket index")`
Original file line number	Diff line number	Diff line change
`@@ -6,6 +6,7 @@ import (`
`6`	`6`	`"strings"`
`7`	`7`	`"testing"`
`8`	`8`
	`9`	`+ "github.com/cortexproject/cortex/pkg/storage/bucket"`
`9`	`10`	`"github.com/go-kit/log"`
`10`	`11`	`"github.com/stretchr/testify/assert"`
`11`	`12`	`"github.com/stretchr/testify/require"`
`@@ -45,7 +46,7 @@ func TestReadIndex_ShouldReturnErrorIfKeyAccessDeniedErr(t *testing.T) {`
`45`	`46`	`},`
`46`	`47`	`}`
`47`	`48`	`idx, err := ReadIndex(context.Background(), bkt, "user-1", nil, log.NewNopLogger())`
`48`		`- require.Equal(t, ErrCustomerManagedKeyError, err)`
	`49`	`+ require.Equal(t, bucket.ErrCustomerManagedKeyError, err)`
`49`	`50`	`require.Nil(t, idx)`
`50`	`51`	`}`
`51`	`52`
Original file line number	Diff line number	Diff line change
`@@ -4,6 +4,7 @@ import (`
`4`	`4`	`"context"`
`5`	`5`	`"io"`
`6`	`6`	`"os"`
	`7`	`+ "strings"`
`7`	`8`	`"testing"`
`8`	`9`
`9`	`10`	`"github.com/pkg/errors"`
`@@ -46,6 +47,11 @@ func (m *MockBucketFailure) Delete(ctx context.Context, name string) error {`
`46`	`47`	`}`
`47`	`48`
`48`	`49`	`func (m *MockBucketFailure) Get(ctx context.Context, name string) (io.ReadCloser, error) {`
	`50`	`+ for prefix, err := range m.GetFailures {`
	`51`	`+ if strings.HasPrefix(name, prefix) {`
	`52`	`+ return nil, err`
	`53`	`+ }`
	`54`	`+ }`
`49`	`55`	`if e, ok := m.GetFailures[name]; ok {`
`50`	`56`	`return nil, e`
`51`	`57`	`}`
`@@ -58,7 +64,7 @@ func (m *MockBucketFailure) WithExpectedErrs(expectedFunc objstore.IsOpFailureEx`
`58`	`64`	`return &MockBucketFailure{Bucket: ibkt.WithExpectedErrs(expectedFunc), DeleteFailures: m.DeleteFailures, GetFailures: m.GetFailures}`
`59`	`65`	`}`
`60`	`66`
`61`		`- return m.WithExpectedErrs(expectedFunc)`
	`67`	`+ return m`
`62`	`68`	`}`
`63`	`69`
`64`	`70`	`func (m *MockBucketFailure) ReaderWithExpectedErrs(expectedFunc objstore.IsOpFailureExpectedFunc) objstore.BucketReader {`