Add shard tombstone handling in compaction worker

Author: kolesnikovae

.mockery.yaml

@@ -37,12 +37,12 @@ packages:
   github.com/grafana/pyroscope/pkg/experiment/metastore/discovery:
     interfaces:
       Discovery:
-  github.com/grafana/pyroscope/pkg/experiment/metastore/dlq:
-    interfaces:
-      LocalServer:
   github.com/grafana/pyroscope/pkg/experiment/metastore/index:
     interfaces:
       Store:
+  github.com/grafana/pyroscope/pkg/experiment/metastore/index/dlq:
+    interfaces:
+      Metastore:
   github.com/grafana/pyroscope/pkg/experiment/distributor/placement:
     interfaces:
       Placement:

pkg/experiment/block/object.go

@@ -8,6 +8,7 @@ import (
 	"strings"
 
 	"github.com/grafana/dskit/multierror"
+	"github.com/oklog/ulid"
 	"golang.org/x/sync/errgroup"
 
 	metastorev1 "github.com/grafana/pyroscope/api/gen/proto/go/metastore/v1"
@@ -75,6 +76,23 @@ func ObjectPath(md *metastorev1.BlockMeta) string {
 	return BuildObjectPath(metadata.Tenant(md), md.Shard, md.CompactionLevel, md.Id)
 }
 
+func BuildObjectDir(tenant string, shard uint32) string {
+	topLevel := DirNameBlock
+	tenantDirName := tenant
+	if tenant == "" {
+		topLevel = DirNameSegment
+		tenantDirName = DirNameAnonTenant
+	}
+	var b strings.Builder
+	b.WriteString(topLevel)
+	b.WriteByte('/')
+	b.WriteString(strconv.Itoa(int(shard)))
+	b.WriteByte('/')
+	b.WriteString(tenantDirName)
+	b.WriteByte('/')
+	return b.String()
+}
+
 func BuildObjectPath(tenant string, shard uint32, level uint32, block string) string {
 	topLevel := DirNameBlock
 	tenantDirName := tenant
@@ -113,6 +131,18 @@ func MetadataDLQObjectPath(md *metastorev1.BlockMeta) string {
 	return b.String()
 }
 
+func ParseBlockIDFromPath(path string) (ulid.ULID, error) {
+	tokens := strings.Split(path, "/")
+	if len(tokens) < 2 {
+		return ulid.ULID{}, fmt.Errorf("invalid path format: %s", path)
+	}
+	blockID, err := ulid.Parse(tokens[len(tokens)-2])
+	if err != nil {
+		return ulid.ULID{}, fmt.Errorf("expected ULID: %s: %w", path, err)
+	}
+	return blockID, nil
+}
+
 // Open opens the object, loading the data into memory if it's small enough.
 //
 // Open may be called multiple times concurrently, but the

pkg/experiment/compactor/compaction_worker_metrics.go renamed to pkg/experiment/compactor/metrics.go

@@ -13,6 +13,7 @@ type compactionWorkerMetrics struct {
 	jobsCompleted    *prometheus.CounterVec
 	jobDuration      *prometheus.HistogramVec
 	timeToCompaction *prometheus.HistogramVec
+	blocksDeleted    *prometheus.CounterVec
 }
 
 func newMetrics(r prometheus.Registerer) *compactionWorkerMetrics {
@@ -31,7 +32,7 @@ func newMetrics(r prometheus.Registerer) *compactionWorkerMetrics {
 			Name: "job_duration_seconds",
 			Help: "Duration of compaction job runs",
 
-			Buckets:                         prometheus.ExponentialBuckets(1, 300, 16),
+			Buckets:                         prometheus.ExponentialBuckets(1, 2, 30),
 			NativeHistogramBucketFactor:     1.1,
 			NativeHistogramMaxBucketNumber:  16,
 			NativeHistogramMinResetDuration: time.Hour,
@@ -41,18 +42,24 @@ func newMetrics(r prometheus.Registerer) *compactionWorkerMetrics {
 			Name: "time_to_compaction_seconds",
 			Help: "The time elapsed since the oldest compacted block was created.",
 
-			Buckets:                         prometheus.ExponentialBuckets(1, 3600, 16),
+			Buckets:                         prometheus.ExponentialBuckets(1, 2, 30),
 			NativeHistogramBucketFactor:     1.1,
 			NativeHistogramMaxBucketNumber:  16,
 			NativeHistogramMinResetDuration: time.Hour,
 		}, []string{"tenant", "level"}),
+
+		blocksDeleted: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Name: "blocks_deleted_total",
+			Help: "Total number of block deletion attempts.",
+		}, []string{"status"}),
 	}
 
 	util.Register(r,
 		m.jobsInProgress,
 		m.jobsCompleted,
 		m.jobDuration,
 		m.timeToCompaction,
+		m.blocksDeleted,
 	)
 
 	return m