Using another global prefix for the parquet markers

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

Author: alanprot

pkg/compactor/blocks_cleaner_test.go

@@ -21,6 +21,7 @@ import (
 	"github.com/thanos-io/thanos/pkg/block/metadata"
 
 	"github.com/cortexproject/cortex/pkg/storage/bucket"
+	"github.com/cortexproject/cortex/pkg/storage/parquet"
 	"github.com/cortexproject/cortex/pkg/storage/tsdb"
 	"github.com/cortexproject/cortex/pkg/storage/tsdb/bucketindex"
 	cortex_testutil "github.com/cortexproject/cortex/pkg/storage/tsdb/testutil"
@@ -161,6 +162,7 @@ func testBlocksCleanerWithOptions(t *testing.T, options testBlocksCleanerOptions
 	require.NoError(t, tsdb.WriteTenantDeletionMark(context.Background(), bucketClient, "user-3", tsdb.NewTenantDeletionMark(time.Now())))
 	block9 := createTSDBBlock(t, bucketClient, "user-3", 10, 30, nil)
 	block10 := createTSDBBlock(t, bucketClient, "user-3", 30, 50, nil)
+	createParquetMarker(t, bucketClient, "user-3", block10)
 
 	// User-4 with no more blocks, but couple of mark and debug files. Should be fully deleted.
 	user4Mark := tsdb.NewTenantDeletionMark(time.Now())
@@ -174,6 +176,10 @@ func testBlocksCleanerWithOptions(t *testing.T, options testBlocksCleanerOptions
 	block12 := createTSDBBlock(t, bucketClient, "user-5", 30, 50, nil)
 	createNoCompactionMark(t, bucketClient, "user-5", block12)
 
+	// Create Parquet marker
+	block13 := createTSDBBlock(t, bucketClient, "user-6", 30, 50, nil)
+	createParquetMarker(t, bucketClient, "user-6", block13)
+
 	// The fixtures have been created. If the bucket client wasn't wrapped to write
 	// deletion marks to the global location too, then this is the right time to do it.
 	if options.markersMigrationEnabled {
@@ -230,7 +236,9 @@ func testBlocksCleanerWithOptions(t *testing.T, options testBlocksCleanerOptions
 		{path: path.Join("user-3", block9.String(), "index"), expectedExists: false},
 		{path: path.Join("user-3", block10.String(), metadata.MetaFilename), expectedExists: false},
 		{path: path.Join("user-3", block10.String(), "index"), expectedExists: false},
+		{path: path.Join("user-3", block10.String(), parquet.ConverterMarkerFileName), expectedExists: false},
 		{path: path.Join("user-4", block.DebugMetas, "meta.json"), expectedExists: options.user4FilesExist},
+		{path: path.Join("user-6", block13.String(), parquet.ConverterMarkerFileName), expectedExists: true},
 	} {
 		exists, err := bucketClient.Exists(ctx, tc.path)
 		require.NoError(t, err)
@@ -298,21 +306,25 @@ func testBlocksCleanerWithOptions(t *testing.T, options testBlocksCleanerOptions
 		cortex_bucket_blocks_count{user="user-1"} 2
 		cortex_bucket_blocks_count{user="user-2"} 1
 		cortex_bucket_blocks_count{user="user-5"} 2
+		cortex_bucket_blocks_count{user="user-6"} 1
 		# HELP cortex_bucket_blocks_marked_for_deletion_count Total number of blocks marked for deletion in the bucket.
 		# TYPE cortex_bucket_blocks_marked_for_deletion_count gauge
 		cortex_bucket_blocks_marked_for_deletion_count{user="user-1"} 1
 		cortex_bucket_blocks_marked_for_deletion_count{user="user-2"} 0
 		cortex_bucket_blocks_marked_for_deletion_count{user="user-5"} 0
+		cortex_bucket_blocks_marked_for_deletion_count{user="user-6"} 0
 		# HELP cortex_bucket_blocks_marked_for_no_compaction_count Total number of blocks marked for no compaction in the bucket.
 		# TYPE cortex_bucket_blocks_marked_for_no_compaction_count gauge
 		cortex_bucket_blocks_marked_for_no_compaction_count{user="user-1"} 0
 		cortex_bucket_blocks_marked_for_no_compaction_count{user="user-2"} 0
 		cortex_bucket_blocks_marked_for_no_compaction_count{user="user-5"} 1
+		cortex_bucket_blocks_marked_for_no_compaction_count{user="user-6"} 0
 		# HELP cortex_bucket_blocks_partials_count Total number of partial blocks.
 		# TYPE cortex_bucket_blocks_partials_count gauge
 		cortex_bucket_blocks_partials_count{user="user-1"} 2
 		cortex_bucket_blocks_partials_count{user="user-2"} 0
 		cortex_bucket_blocks_partials_count{user="user-5"} 0
+		cortex_bucket_blocks_partials_count{user="user-6"} 0
 	`),
 		"cortex_bucket_blocks_count",
 		"cortex_bucket_blocks_marked_for_deletion_count",

pkg/compactor/compactor_test.go

@@ -36,6 +36,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/ring"
 	"github.com/cortexproject/cortex/pkg/ring/kv/consul"
 	"github.com/cortexproject/cortex/pkg/storage/bucket"
+	"github.com/cortexproject/cortex/pkg/storage/parquet"
 	cortex_tsdb "github.com/cortexproject/cortex/pkg/storage/tsdb"
 	"github.com/cortexproject/cortex/pkg/storage/tsdb/bucketindex"
 	cortex_storage_testutil "github.com/cortexproject/cortex/pkg/storage/tsdb/testutil"
@@ -1429,6 +1430,14 @@ func createBlockVisitMarker(t *testing.T, bkt objstore.Bucket, userID string, bl
 	require.NoError(t, bkt.Upload(context.Background(), markPath, strings.NewReader(content)))
 }
 
+func createParquetMarker(t *testing.T, bkt objstore.Bucket, userID string, blockID ulid.ULID) {
+	content := mockParquetMarker()
+	blockPath := path.Join(userID, blockID.String())
+	markPath := path.Join(blockPath, parquet.ConverterMarkerFileName)
+
+	require.NoError(t, bkt.Upload(context.Background(), markPath, strings.NewReader(content)))
+}
+
 func findCompactorByUserID(compactors []*Compactor, logs []*concurrency.SyncBuffer, userID string) (*Compactor, *concurrency.SyncBuffer, error) {
 	var compactor *Compactor
 	var log *concurrency.SyncBuffer
@@ -1727,6 +1736,19 @@ func mockBlockVisitMarker() string {
 	return string(content)
 }
 
+func mockParquetMarker() string {
+	parquetMarker := parquet.ConverterMark{
+		Version: 1,
+	}
+
+	content, err := json.Marshal(parquetMarker)
+	if err != nil {
+		panic("failed to marshal mocked block visit marker")
+	}
+
+	return string(content)
+}
+
 func TestCompactor_DeleteLocalSyncFiles(t *testing.T) {
 	numUsers := 10
 

pkg/parquetconverter/converter_test.go

@@ -96,7 +96,7 @@ func TestConverter(t *testing.T) {
 	for _, block := range blocksConverted {
 		for _, file := range []string{
 			fmt.Sprintf("%s/parquet-converter-mark.json", block.String()),
-			fmt.Sprintf("markers/%s-parquet-converter-mark.json", block.String()),
+			fmt.Sprintf("parquet-markers/%s-parquet-converter-mark.json", block.String()),
 			fmt.Sprintf("%s/0.chunks.parquet", block.String()),
 			fmt.Sprintf("%s/0.labels.parquet", block.String()),
 		} {

pkg/storage/parquet/converter_marker.go

@@ -17,6 +17,7 @@ import (
 )
 
 const (
+	ConverterMarkerPrefix   = "parquet-markers"
 	ConverterMarkerFileName = "parquet-converter-mark.json"
 	CurrentVersion          = 1
 )
@@ -29,7 +30,7 @@ func ReadConverterMark(ctx context.Context, id ulid.ULID, userBkt objstore.Instr
 	markerPath := path.Join(id.String(), ConverterMarkerFileName)
 	reader, err := userBkt.WithExpectedErrs(tsdb.IsOneOfTheExpectedErrors(userBkt.IsAccessDeniedErr, userBkt.IsObjNotFoundErr)).Get(ctx, markerPath)
 	if err != nil {
-		if userBkt.IsObjNotFoundErr(err) {
+		if userBkt.IsObjNotFoundErr(err) || userBkt.IsAccessDeniedErr(err) {
 			return &ConverterMark{}, nil
 		}
 

pkg/storage/tsdb/bucketindex/markers.go

@@ -43,7 +43,7 @@ func NoCompactMarkFilenameMarkFilepath(blockID ulid.ULID) string {
 }
 
 func ConverterMarkFilePath(blockID ulid.ULID) string {
-	return fmt.Sprintf("%s/%s-%s", MarkersPathname, blockID.String(), parquet.ConverterMarkerFileName)
+	return fmt.Sprintf("%s/%s-%s", parquet.ConverterMarkerPrefix, blockID.String(), parquet.ConverterMarkerFileName)
 }
 
 // IsBlockDeletionMarkFilename returns whether the input filename matches the expected pattern