wrap bucket with the global markers bucket

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

Author: alanprot

pkg/parquetconverter/converter.go

@@ -109,6 +109,8 @@ func NewConverter(cfg Config, storageCfg cortex_tsdb.BlocksStorageConfig, blockR
 
 func (c *Converter) starting(ctx context.Context) error {
 	bkt, err := bucket.NewClient(ctx, c.storageCfg.Bucket, nil, "parquet-converter", c.logger, c.reg)
+	bkt = bucketindex.BucketWithGlobalMarkers(bkt)
+	
 	if err != nil {
 		return err
 	}

pkg/storage/parquet/converter_marker.go

@@ -16,18 +16,16 @@ import (
 )
 
 const (
-	Name           = "parquet-converter-mark.json"
-	CurrentVersion = 1
+	ConverterMarkerFileName = "parquet-converter-mark.json"
+	CurrentVersion          = 1
 )
 
 type ConverterMark struct {
 	Version int `json:"version"`
 }
 
-func (m *ConverterMark) markerFilename() string { return Name }
-
 func ReadConverterMark(ctx context.Context, id ulid.ULID, userBkt objstore.InstrumentedBucket, logger log.Logger) (*ConverterMark, error) {
-	markerPath := path.Join(id.String(), Name)
+	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) {
@@ -40,7 +38,7 @@ func ReadConverterMark(ctx context.Context, id ulid.ULID, userBkt objstore.Instr
 
 	metaContent, err := io.ReadAll(reader)
 	if err != nil {
-		return nil, errors.Wrapf(err, "read file: %s", Name)
+		return nil, errors.Wrapf(err, "read file: %s", ConverterMarkerFileName)
 	}
 
 	marker := ConverterMark{}
@@ -52,7 +50,7 @@ func WriteConverterMark(ctx context.Context, id ulid.ULID, userBkt objstore.Buck
 	marker := ConverterMark{
 		Version: CurrentVersion,
 	}
-	markerPath := path.Join(id.String(), Name)
+	markerPath := path.Join(id.String(), ConverterMarkerFileName)
 	b, err := json.Marshal(marker)
 	if err != nil {
 		return err

pkg/storage/tsdb/bucketindex/markers.go

@@ -3,6 +3,7 @@ package bucketindex
 import (
 	"context"
 	"fmt"
+	"github.com/cortexproject/cortex/pkg/storage/parquet"
 	"path"
 	"path/filepath"
 	"strings"
@@ -23,8 +24,9 @@ const (
 
 var (
 	MarkersMap = map[string]func(ulid.ULID) string{
-		metadata.DeletionMarkFilename:  BlockDeletionMarkFilepath,
-		metadata.NoCompactMarkFilename: NoCompactMarkFilenameMarkFilepath,
+		metadata.DeletionMarkFilename:   BlockDeletionMarkFilepath,
+		metadata.NoCompactMarkFilename:  NoCompactMarkFilenameMarkFilepath,
+		parquet.ConverterMarkerFileName: ConverterMarkFilePath,
 	}
 )
 
@@ -40,6 +42,10 @@ func NoCompactMarkFilenameMarkFilepath(blockID ulid.ULID) string {
 	return fmt.Sprintf("%s/%s-%s", MarkersPathname, blockID.String(), metadata.NoCompactMarkFilename)
 }
 
+func ConverterMarkFilePath(blockID ulid.ULID) string {
+	return fmt.Sprintf("%s/%s-%s", MarkersPathname, blockID.String(), parquet.ConverterMarkerFileName)
+}
+
 // IsBlockDeletionMarkFilename returns whether the input filename matches the expected pattern
 // of block deletion markers stored in the markers location.
 func IsBlockDeletionMarkFilename(name string) (ulid.ULID, bool) {