switch to use global marker path for parquet converter mark

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

Author: yeya24

pkg/storage/parquet/converter_marker.go

@@ -0,0 +1,7 @@
+package parquet
+
+const ConverterMakerFileName = "parquet-converter-mark.json"
+
+type ParquetMeta struct {
+	Version int `json:"version"`
+}

pkg/storage/tsdb/bucketindex/index.go

@@ -12,6 +12,7 @@ import (
 	"github.com/thanos-io/thanos/pkg/block"
 	"github.com/thanos-io/thanos/pkg/block/metadata"
 
+	"github.com/cortexproject/cortex/pkg/storage/parquet"
 	cortex_tsdb "github.com/cortexproject/cortex/pkg/storage/tsdb"
 	"github.com/cortexproject/cortex/pkg/util"
 )
@@ -94,7 +95,7 @@ type Block struct {
 	UploadedAt int64 `json:"uploaded_at"`
 
 	// Parquet metadata if exists. If doesn't exist it will be nil.
-	Parquet *cortex_tsdb.ParquetMeta `json:"parquet,omitempty"`
+	Parquet *parquet.ParquetMeta `json:"parquet,omitempty"`
 }
 
 // Within returns whether the block contains samples within the provided range.

pkg/storage/tsdb/bucketindex/markers.go

@@ -15,6 +15,7 @@ import (
 	"github.com/thanos-io/thanos/pkg/block/metadata"
 
 	"github.com/cortexproject/cortex/pkg/storage/bucket"
+	"github.com/cortexproject/cortex/pkg/storage/parquet"
 )
 
 const (
@@ -76,6 +77,24 @@ func IsBlockNoCompactMarkFilename(name string) (ulid.ULID, bool) {
 	return id, err == nil
 }
 
+// IsBlockParquetConverterMarkFilename returns whether the input filename matches the expected pattern
+// of block parquet conveter markers stored in the markers location.
+func IsBlockParquetConverterMarkFilename(name string) (ulid.ULID, bool) {
+	parts := strings.SplitN(name, "-", 2)
+	if len(parts) != 2 {
+		return ulid.ULID{}, false
+	}
+
+	// Ensure the 2nd part matches the parquet conveter mark filename.
+	if parts[1] != parquet.ConverterMakerFileName {
+		return ulid.ULID{}, false
+	}
+
+	// Ensure the 1st part is a valid block ID.
+	id, err := ulid.Parse(filepath.Base(parts[0]))
+	return id, err == nil
+}
+
 // MigrateBlockDeletionMarksToGlobalLocation list all tenant's blocks and, for each of them, look for
 // a deletion mark in the block location. Found deletion marks are copied to the global markers location.
 // The migration continues on error and returns once all blocks have been checked.

pkg/storage/tsdb/bucketindex/updater.go

@@ -15,6 +15,7 @@ import (
 	"github.com/thanos-io/thanos/pkg/block"
 	"github.com/thanos-io/thanos/pkg/block/metadata"
 
+	"github.com/cortexproject/cortex/pkg/storage/parquet"
 	"github.com/cortexproject/cortex/pkg/storage/tsdb"
 
 	"github.com/cortexproject/cortex/pkg/storage/bucket"
@@ -65,12 +66,12 @@ func (w *Updater) UpdateIndex(ctx context.Context, old *Index) (*Index, map[ulid
 		oldBlockDeletionMarks = old.BlockDeletionMarks
 	}
 
-	blockDeletionMarks, deletedBlocks, totalBlocksBlocksMarkedForNoCompaction, err := w.updateBlockMarks(ctx, oldBlockDeletionMarks)
+	blockDeletionMarks, deletedBlocks, totalBlocksBlocksMarkedForNoCompaction, discoveredParquetBlocks, err := w.updateBlockMarks(ctx, oldBlockDeletionMarks)
 	if err != nil {
 		return nil, nil, 0, err
 	}
 
-	blocks, partials, err := w.updateBlocks(ctx, oldBlocks, deletedBlocks)
+	blocks, partials, err := w.updateBlocks(ctx, oldBlocks, deletedBlocks, discoveredParquetBlocks)
 	if err != nil {
 		return nil, nil, 0, err
 	}
@@ -83,7 +84,7 @@ func (w *Updater) UpdateIndex(ctx context.Context, old *Index) (*Index, map[ulid
 	}, partials, totalBlocksBlocksMarkedForNoCompaction, nil
 }
 
-func (w *Updater) updateBlocks(ctx context.Context, old []*Block, deletedBlocks map[ulid.ULID]struct{}) (blocks []*Block, partials map[ulid.ULID]error, _ error) {
+func (w *Updater) updateBlocks(ctx context.Context, old []*Block, deletedBlocks map[ulid.ULID]struct{}, discoveredParquetBlocks map[ulid.ULID]struct{}) (blocks []*Block, partials map[ulid.ULID]error, _ error) {
 	discovered := map[ulid.ULID]struct{}{}
 	partials = map[ulid.ULID]error{}
 
@@ -107,12 +108,14 @@ func (w *Updater) updateBlocks(ctx context.Context, old []*Block, deletedBlocks
 				level.Warn(w.logger).Log("msg", "skipped block with missing global deletion marker", "block", b.ID.String())
 				continue
 			}
-			// Check if parquet mark has been uploaded for the old block.
-			// TODO: this should be optimized to have all parquet marker in a global path.
-			// We assume parquet marker cannot be removed from a block if it exists before.
-			if w.parquetEnabled && b.Parquet == nil {
-				if err := w.updateParquetBlockIndexEntry(ctx, b.ID, b); err != nil {
-					return nil, nil, err
+			// Check if parquet mark has been uploaded or deleted for the old block.
+			if w.parquetEnabled {
+				if _, ok := discoveredParquetBlocks[b.ID]; ok {
+					if err := w.updateParquetBlockIndexEntry(ctx, b.ID, b); err != nil {
+						return nil, nil, err
+					}
+				} else if b.Parquet != nil {
+					b.Parquet = nil
 				}
 			}
 			blocks = append(blocks, b)
@@ -126,7 +129,11 @@ func (w *Updater) updateBlocks(ctx context.Context, old []*Block, deletedBlocks
 		b, err := w.updateBlockIndexEntry(ctx, id)
 		if err == nil {
 			if w.parquetEnabled {
-				err = w.updateParquetBlockIndexEntry(ctx, id, b)
+				if _, ok := discoveredParquetBlocks[b.ID]; ok {
+					if err := w.updateParquetBlockIndexEntry(ctx, b.ID, b); err != nil {
+						return nil, nil, err
+					}
+				}
 			}
 			if err == nil {
 				blocks = append(blocks, b)
@@ -203,7 +210,7 @@ func (w *Updater) updateBlockIndexEntry(ctx context.Context, id ulid.ULID) (*Blo
 }
 
 func (w *Updater) updateParquetBlockIndexEntry(ctx context.Context, id ulid.ULID, block *Block) error {
-	parquetMarkFile := path.Join(id.String(), tsdb.ParquetConverterMakerFileName)
+	parquetMarkFile := path.Join(id.String(), parquet.ConverterMakerFileName)
 
 	// Get the block's parquet marker file.
 	r, err := w.bkt.ReaderWithExpectedErrs(tsdb.IsOneOfTheExpectedErrors(w.bkt.IsObjNotFoundErr, w.bkt.IsAccessDeniedErr)).Get(ctx, parquetMarkFile)
@@ -223,18 +230,19 @@ func (w *Updater) updateParquetBlockIndexEntry(ctx context.Context, id ulid.ULID
 		return errors.Wrapf(err, "read parquet converter marker file: %v", parquetMarkFile)
 	}
 
-	m := tsdb.ParquetMeta{}
+	m := parquet.ParquetMeta{}
 	if err := json.Unmarshal(markContent, &m); err != nil {
 		return errors.Wrapf(ErrBlockParquetMarkCorrupted, "unmarshal parquet converter marker file %s: %v", parquetMarkFile, err)
 	}
 	block.Parquet = &m
 	return nil
 }
 
-func (w *Updater) updateBlockMarks(ctx context.Context, old []*BlockDeletionMark) ([]*BlockDeletionMark, map[ulid.ULID]struct{}, int64, error) {
+func (w *Updater) updateBlockMarks(ctx context.Context, old []*BlockDeletionMark) ([]*BlockDeletionMark, map[ulid.ULID]struct{}, int64, map[ulid.ULID]struct{}, error) {
 	out := make([]*BlockDeletionMark, 0, len(old))
 	deletedBlocks := map[ulid.ULID]struct{}{}
 	discovered := map[ulid.ULID]struct{}{}
+	discoveredParquetBlocks := map[ulid.ULID]struct{}{}
 	totalBlocksBlocksMarkedForNoCompaction := int64(0)
 
 	// Find all markers in the storage.
@@ -247,10 +255,16 @@ func (w *Updater) updateBlockMarks(ctx context.Context, old []*BlockDeletionMark
 			totalBlocksBlocksMarkedForNoCompaction++
 		}
 
+		if w.parquetEnabled {
+			if blockID, ok := IsBlockParquetConverterMarkFilename(path.Base(name)); ok {
+				discoveredParquetBlocks[blockID] = struct{}{}
+			}
+		}
+
 		return nil
 	})
 	if err != nil {
-		return nil, nil, totalBlocksBlocksMarkedForNoCompaction, errors.Wrap(err, "list block deletion marks")
+		return nil, nil, totalBlocksBlocksMarkedForNoCompaction, discoveredParquetBlocks, errors.Wrap(err, "list block deletion marks")
 	}
 
 	// Since deletion marks are immutable, all markers already existing in the index can just be copied.
@@ -276,13 +290,13 @@ func (w *Updater) updateBlockMarks(ctx context.Context, old []*BlockDeletionMark
 			continue
 		}
 		if err != nil {
-			return nil, nil, totalBlocksBlocksMarkedForNoCompaction, err
+			return nil, nil, totalBlocksBlocksMarkedForNoCompaction, discoveredParquetBlocks, err
 		}
 
 		out = append(out, m)
 	}
 
-	return out, deletedBlocks, totalBlocksBlocksMarkedForNoCompaction, nil
+	return out, deletedBlocks, totalBlocksBlocksMarkedForNoCompaction, discoveredParquetBlocks, nil
 }
 
 func (w *Updater) updateBlockDeletionMarkIndexEntry(ctx context.Context, id ulid.ULID) (*BlockDeletionMark, error) {

pkg/storage/tsdb/bucketindex/updater_test.go

@@ -22,7 +22,7 @@ import (
 	"github.com/thanos-io/thanos/pkg/block/metadata"
 
 	"github.com/cortexproject/cortex/pkg/storage/bucket"
-	cortex_tsdb "github.com/cortexproject/cortex/pkg/storage/tsdb"
+	"github.com/cortexproject/cortex/pkg/storage/parquet"
 	"github.com/cortexproject/cortex/pkg/storage/tsdb/testutil"
 )
 
@@ -324,7 +324,7 @@ func TestUpdater_UpdateIndex_WithParquet(t *testing.T) {
 	require.NoError(t, err)
 	assertBucketIndexEqualWithParquet(t, returnedIdx, bkt, userID,
 		[]tsdb.BlockMeta{block1, block2},
-		[]*metadata.DeletionMark{block2Mark}, map[string]*cortex_tsdb.ParquetMeta{
+		[]*metadata.DeletionMark{block2Mark}, map[string]*parquet.ParquetMeta{
 			block1.ULID.String(): block1ParquetMark,
 		})
 
@@ -338,7 +338,7 @@ func TestUpdater_UpdateIndex_WithParquet(t *testing.T) {
 	assertBucketIndexEqualWithParquet(t, returnedIdx, bkt, userID,
 		[]tsdb.BlockMeta{block1, block2, block3, block4},
 		[]*metadata.DeletionMark{block2Mark, block4Mark},
-		map[string]*cortex_tsdb.ParquetMeta{
+		map[string]*parquet.ParquetMeta{
 			block1.ULID.String(): block1ParquetMark,
 		})
 
@@ -349,7 +349,7 @@ func TestUpdater_UpdateIndex_WithParquet(t *testing.T) {
 	require.NoError(t, err)
 	assertBucketIndexEqualWithParquet(t, returnedIdx, bkt, userID,
 		[]tsdb.BlockMeta{block1, block3, block4},
-		[]*metadata.DeletionMark{block4Mark}, map[string]*cortex_tsdb.ParquetMeta{
+		[]*metadata.DeletionMark{block4Mark}, map[string]*parquet.ParquetMeta{
 			block1.ULID.String(): block1ParquetMark,
 		})
 
@@ -359,7 +359,7 @@ func TestUpdater_UpdateIndex_WithParquet(t *testing.T) {
 	require.NoError(t, err)
 	assertBucketIndexEqualWithParquet(t, returnedIdx, bkt, userID,
 		[]tsdb.BlockMeta{block1, block3, block4},
-		[]*metadata.DeletionMark{block4Mark}, map[string]*cortex_tsdb.ParquetMeta{
+		[]*metadata.DeletionMark{block4Mark}, map[string]*parquet.ParquetMeta{
 			block1.ULID.String(): block1ParquetMark,
 			block3.ULID.String(): block3ParquetMark,
 		})
@@ -375,22 +375,22 @@ func TestUpdater_UpdateParquetBlockIndexEntry(t *testing.T) {
 		setupBucket       func(t *testing.T, bkt objstore.InstrumentedBucket, blockID ulid.ULID) objstore.InstrumentedBucket
 		expectedError     error
 		expectParquet     bool
-		expectParquetMeta *cortex_tsdb.ParquetMeta
+		expectParquetMeta *parquet.ParquetMeta
 	}{
 		{
 			name: "should successfully read parquet marker",
 			setupBucket: func(t *testing.T, bkt objstore.InstrumentedBucket, blockID ulid.ULID) objstore.InstrumentedBucket {
-				parquetMark := cortex_tsdb.ParquetMeta{
+				parquetMark := parquet.ParquetMeta{
 					Version: 1,
 				}
 				data, err := json.Marshal(parquetMark)
 				require.NoError(t, err)
-				require.NoError(t, bkt.Upload(ctx, path.Join(userID, blockID.String(), cortex_tsdb.ParquetConverterMakerFileName), bytes.NewReader(data)))
+				require.NoError(t, bkt.Upload(ctx, path.Join(userID, blockID.String(), parquet.ConverterMakerFileName), bytes.NewReader(data)))
 				return bkt
 			},
 			expectedError:     nil,
 			expectParquet:     true,
-			expectParquetMeta: &cortex_tsdb.ParquetMeta{Version: 1},
+			expectParquetMeta: &parquet.ParquetMeta{Version: 1},
 		},
 		{
 			name: "should handle missing parquet marker",
@@ -407,7 +407,7 @@ func TestUpdater_UpdateParquetBlockIndexEntry(t *testing.T) {
 				return &testutil.MockBucketFailure{
 					Bucket: bkt,
 					GetFailures: map[string]error{
-						path.Join(userID, blockID.String(), cortex_tsdb.ParquetConverterMakerFileName): testutil.ErrKeyAccessDeniedError,
+						path.Join(userID, blockID.String(), parquet.ConverterMakerFileName): testutil.ErrKeyAccessDeniedError,
 					},
 				}
 			},
@@ -417,7 +417,7 @@ func TestUpdater_UpdateParquetBlockIndexEntry(t *testing.T) {
 		{
 			name: "should handle corrupted parquet marker",
 			setupBucket: func(t *testing.T, bkt objstore.InstrumentedBucket, blockID ulid.ULID) objstore.InstrumentedBucket {
-				require.NoError(t, bkt.Upload(ctx, path.Join(userID, blockID.String(), cortex_tsdb.ParquetConverterMakerFileName), bytes.NewReader([]byte("invalid json"))))
+				require.NoError(t, bkt.Upload(ctx, path.Join(userID, blockID.String(), parquet.ConverterMakerFileName), bytes.NewReader([]byte("invalid json"))))
 				return bkt
 			},
 			expectedError: ErrBlockParquetMarkCorrupted,
@@ -494,7 +494,7 @@ func assertBucketIndexEqual(t testing.TB, idx *Index, bkt objstore.Bucket, userI
 	assert.ElementsMatch(t, expectedMarkEntries, idx.BlockDeletionMarks)
 }
 
-func assertBucketIndexEqualWithParquet(t testing.TB, idx *Index, bkt objstore.Bucket, userID string, expectedBlocks []tsdb.BlockMeta, expectedDeletionMarks []*metadata.DeletionMark, parquetBlocks map[string]*cortex_tsdb.ParquetMeta) {
+func assertBucketIndexEqualWithParquet(t testing.TB, idx *Index, bkt objstore.Bucket, userID string, expectedBlocks []tsdb.BlockMeta, expectedDeletionMarks []*metadata.DeletionMark, parquetBlocks map[string]*parquet.ParquetMeta) {
 	assert.Equal(t, IndexVersion1, idx.Version)
 	assert.InDelta(t, time.Now().Unix(), idx.UpdatedAt, 2)
 

pkg/storage/tsdb/parquet.go

@@ -15,7 +15,7 @@ import (
 	"github.com/thanos-io/objstore"
 	"github.com/thanos-io/thanos/pkg/block/metadata"
 
-	cortex_tsdb "github.com/cortexproject/cortex/pkg/storage/tsdb"
+	"github.com/cortexproject/cortex/pkg/storage/parquet"
 )
 
 func MockStorageBlock(t testing.TB, bucket objstore.Bucket, userID string, minT, maxT int64) tsdb.BlockMeta {
@@ -89,8 +89,8 @@ func MockStorageNonCompactionMark(t testing.TB, bucket objstore.Bucket, userID s
 	return &mark
 }
 
-func MockStorageParquetMark(t testing.TB, bucket objstore.Bucket, userID string, meta tsdb.BlockMeta) *cortex_tsdb.ParquetMeta {
-	mark := cortex_tsdb.ParquetMeta{
+func MockStorageParquetMark(t testing.TB, bucket objstore.Bucket, userID string, meta tsdb.BlockMeta) *parquet.ParquetMeta {
+	mark := parquet.ParquetMeta{
 		Version: 1,
 	}
 
@@ -100,7 +100,7 @@ func MockStorageParquetMark(t testing.TB, bucket objstore.Bucket, userID string,
 	}
 
 	markContentReader := strings.NewReader(string(markContent))
-	markPath := fmt.Sprintf("%s/%s/%s", userID, meta.ULID.String(), cortex_tsdb.ParquetConverterMakerFileName)
+	markPath := fmt.Sprintf("%s/%s/%s", userID, meta.ULID.String(), parquet.ConverterMakerFileName)
 	require.NoError(t, bucket.Upload(context.Background(), markPath, markContentReader))
 
 	return &mark