ignore permission denied error in parquet converter

Signed-off-by: yeya24 <[email protected]>

Author: yeya24

pkg/parquetconverter/converter.go

@@ -27,6 +27,8 @@ import (
 	"github.com/thanos-io/thanos/pkg/block/metadata"
 	"github.com/thanos-io/thanos/pkg/extprom"
 	"github.com/thanos-io/thanos/pkg/logutil"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 
 	"github.com/cortexproject/cortex/pkg/ring"
 	"github.com/cortexproject/cortex/pkg/storage/bucket"
@@ -36,6 +38,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/storage/tsdb/users"
 	"github.com/cortexproject/cortex/pkg/tenant"
 	"github.com/cortexproject/cortex/pkg/util"
+	cortex_errors "github.com/cortexproject/cortex/pkg/util/errors"
 	util_log "github.com/cortexproject/cortex/pkg/util/log"
 	"github.com/cortexproject/cortex/pkg/util/services"
 	"github.com/cortexproject/cortex/pkg/util/validation"
@@ -247,6 +250,10 @@ func (c *Converter) running(ctx context.Context) error {
 					if ctx.Err() != nil {
 						return ctx.Err()
 					}
+					if c.isCausedByPermissionDenied(err) {
+						level.Warn(userLogger).Log("msg", "skipping convert user due to PermissionDenied", "err", err)
+						continue
+					}
 					level.Error(userLogger).Log("msg", "failed to convert user", "err", err)
 				}
 			}
@@ -474,14 +481,33 @@ func (c *Converter) convertUser(ctx context.Context, logger log.Logger, ring rin
 }
 
 func (c *Converter) checkConvertError(userID string, err error) (terminate bool) {
-	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) || c.isCausedByPermissionDenied(err) {
 		terminate = true
 	} else {
 		c.metrics.convertBlockFailures.WithLabelValues(userID).Inc()
 	}
 	return
 }
 
+func (c *Converter) isCausedByPermissionDenied(err error) bool {
+	cause := errors.Cause(err)
+	res := cortex_errors.ErrorIs(cause, func(err error) bool {
+		return c.isPermissionDeniedErr(err)
+	})
+	return res
+}
+
+func (c *Converter) isPermissionDeniedErr(err error) bool {
+	if c.bkt.IsAccessDeniedErr(err) {
+		return true
+	}
+	s, ok := status.FromError(err)
+	if !ok {
+		return false
+	}
+	return s.Code() == codes.PermissionDenied
+}
+
 func (c *Converter) ownUser(r ring.ReadRing, userId string) (bool, error) {
 	if userId == tenant.GlobalMarkersDir {
 		// __markers__ is reserved for global markers and no tenant should be allowed to have that name.

pkg/parquetconverter/converter_test.go

@@ -2,10 +2,12 @@ package parquetconverter
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"math/rand"
 	"path"
+	"strings"
 	"testing"
 	"time"
 
@@ -21,6 +23,8 @@ import (
 	"github.com/thanos-io/objstore/providers/filesystem"
 	"github.com/thanos-io/thanos/pkg/block"
 	"github.com/thanos-io/thanos/pkg/block/metadata"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 
 	"github.com/cortexproject/cortex/integration/e2e"
 	"github.com/cortexproject/cortex/pkg/ring"
@@ -31,6 +35,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/storage/tsdb/bucketindex"
 	"github.com/cortexproject/cortex/pkg/storage/tsdb/users"
 	"github.com/cortexproject/cortex/pkg/util/concurrency"
+	cortex_errors "github.com/cortexproject/cortex/pkg/util/errors"
 	"github.com/cortexproject/cortex/pkg/util/flagext"
 	"github.com/cortexproject/cortex/pkg/util/services"
 	"github.com/cortexproject/cortex/pkg/util/test"
@@ -269,7 +274,8 @@ func TestConverter_BlockConversionFailure(t *testing.T) {
 
 	// Create a mock bucket that wraps the filesystem bucket but fails uploads
 	mockBucket := &mockBucket{
-		Bucket: fsBucket,
+		Bucket:        fsBucket,
+		uploadFailure: fmt.Errorf("mock upload failure"),
 	}
 
 	converter := newConverter(cfg, objstore.WithNoopInstr(mockBucket), storageCfg, []int64{3600000, 7200000}, nil, reg, overrides, nil)
@@ -284,13 +290,83 @@ func TestConverter_BlockConversionFailure(t *testing.T) {
 	assert.Equal(t, 1.0, testutil.ToFloat64(converter.metrics.convertBlockFailures.WithLabelValues(userID)))
 }
 
+func TestConverter_ShouldNotFailOnAccessDenyError(t *testing.T) {
+	// Create a new registry for testing
+	reg := prometheus.NewRegistry()
+
+	// Create a new converter with test configuration
+	cfg := Config{
+		MetaSyncConcurrency: 1,
+		DataDir:             t.TempDir(),
+	}
+	logger := log.NewNopLogger()
+	storageCfg := cortex_tsdb.BlocksStorageConfig{}
+	flagext.DefaultValues(&storageCfg)
+	limits := &validation.Limits{}
+	flagext.DefaultValues(limits)
+	overrides := validation.NewOverrides(*limits, nil)
+	limits.ParquetConverterEnabled = true
+
+	// Create a filesystem bucket for initial block upload
+	fsBucket, err := filesystem.NewBucket(t.TempDir())
+	require.NoError(t, err)
+
+	// Create test labels
+	lbls := labels.Labels{labels.Label{
+		Name:  "__name__",
+		Value: "test",
+	}}
+
+	// Create a real TSDB block
+	dir := t.TempDir()
+	rnd := rand.New(rand.NewSource(time.Now().Unix()))
+	blockID, err := e2e.CreateBlock(context.Background(), rnd, dir, []labels.Labels{lbls}, 2, 0, 2*time.Hour.Milliseconds(), time.Minute.Milliseconds(), 10)
+	require.NoError(t, err)
+	bdir := path.Join(dir, blockID.String())
+
+	userID := "test-user"
+
+	// Upload the block to filesystem bucket
+	err = block.Upload(context.Background(), logger, bucket.NewPrefixedBucketClient(fsBucket, userID), bdir, metadata.NoneFunc)
+	require.NoError(t, err)
+
+	// Create a mock bucket that wraps the filesystem bucket but fails with permission denied error.
+	mockBucket := &mockBucket{
+		Bucket:     fsBucket,
+		getFailure: cortex_errors.WithCause(errors.New("dummy error"), status.Error(codes.PermissionDenied, "dummy")),
+	}
+
+	converter := newConverter(cfg, objstore.WithNoopInstr(mockBucket), storageCfg, []int64{3600000, 7200000}, nil, reg, overrides, nil)
+	converter.ringLifecycler = &ring.Lifecycler{
+		Addr: "1.2.3.4",
+	}
+
+	err = converter.convertUser(context.Background(), logger, &RingMock{ReadRing: &ring.Ring{}}, userID)
+	require.Error(t, err)
+
+	// Verify the failure metric was not incremented
+	assert.Equal(t, 0.0, testutil.ToFloat64(converter.metrics.convertBlockFailures.WithLabelValues(userID)))
+}
+
 // mockBucket implements objstore.Bucket for testing
 type mockBucket struct {
 	objstore.Bucket
+	uploadFailure error
+	getFailure    error
 }
 
 func (m *mockBucket) Upload(ctx context.Context, name string, r io.Reader) error {
-	return fmt.Errorf("mock upload failure")
+	if m.uploadFailure != nil {
+		return m.uploadFailure
+	}
+	return m.Bucket.Upload(ctx, name, r)
+}
+
+func (m *mockBucket) Get(ctx context.Context, name string) (io.ReadCloser, error) {
+	if m.getFailure != nil && strings.Contains(name, "index") {
+		return nil, m.getFailure
+	}
+	return m.Bucket.Get(ctx, name)
 }
 
 type RingMock struct {