Partitioning Compactor

docs/blocks-storage/compactor.md

@@ -285,15 +285,33 @@ compactor:
     # CLI flag: -compactor.ring.wait-active-instance-timeout
     [wait_active_instance_timeout: <duration> | default = 10m]
 
-  # How long block visit marker file should be considered as expired and able to
-  # be picked up by compactor again.
-  # CLI flag: -compactor.block-visit-marker-timeout
-  [block_visit_marker_timeout: <duration> | default = 5m]
+  # How long shuffle sharding planner would wait before running planning code.
+  # CLI flag: -compactor.sharding-planner-delay
+  [sharding_planner_delay: <duration> | default = 10s]
 
-  # How frequently block visit marker file should be updated duration
+  # The compaction mode to use. Supported values are: default, partitioning.
+  # CLI flag: -compactor.compaction-mode
+  [compaction_mode: <string> | default = "default"]
+
+  # How long compaction visit marker file should be considered as expired and
+  # able to be picked up by compactor again.
+  # CLI flag: -compactor.compaction-visit-marker-timeout
+  [compaction_visit_marker_timeout: <duration> | default = 1m30s]
+
+  # How frequently compaction visit marker file should be updated duration
   # compaction.
-  # CLI flag: -compactor.block-visit-marker-file-update-interval
-  [block_visit_marker_file_update_interval: <duration> | default = 1m]
+  # CLI flag: -compactor.compaction-visit-marker-file-update-interval
+  [compaction_visit_marker_file_update_interval: <duration> | default = 1m]
+
+  # How long cleaner visit marker file should be considered as expired and able
+  # to be picked up by cleaner again.
+  # CLI flag: -compactor.cleaner-visit-marker-timeout
+  [cleaner_visit_marker_timeout: <duration> | default = 10m]
+
+  # How frequently cleaner visit marker file should be updated when cleaning
+  # user.
+  # CLI flag: -compactor.cleaner-visit-marker-file-update-interval
+  [cleaner_visit_marker_file_update_interval: <duration> | default = 5m]
 
   # When enabled, index verification will ignore out of order label names.
   # CLI flag: -compactor.accept-malformed-index

docs/configuration/config-file-reference.md

@@ -2205,14 +2205,32 @@ sharding_ring:
   # CLI flag: -compactor.ring.wait-active-instance-timeout
   [wait_active_instance_timeout: <duration> | default = 10m]
 
-# How long block visit marker file should be considered as expired and able to
-# be picked up by compactor again.
-# CLI flag: -compactor.block-visit-marker-timeout
-[block_visit_marker_timeout: <duration> | default = 5m]
+# How long shuffle sharding planner would wait before running planning code.
+# CLI flag: -compactor.sharding-planner-delay
+[sharding_planner_delay: <duration> | default = 10s]
 
-# How frequently block visit marker file should be updated duration compaction.
-# CLI flag: -compactor.block-visit-marker-file-update-interval
-[block_visit_marker_file_update_interval: <duration> | default = 1m]
+# The compaction mode to use. Supported values are: default, partitioning.
+# CLI flag: -compactor.compaction-mode
+[compaction_mode: <string> | default = "default"]
+
+# How long compaction visit marker file should be considered as expired and able
+# to be picked up by compactor again.
+# CLI flag: -compactor.compaction-visit-marker-timeout
+[compaction_visit_marker_timeout: <duration> | default = 1m30s]
+
+# How frequently compaction visit marker file should be updated duration
+# compaction.
+# CLI flag: -compactor.compaction-visit-marker-file-update-interval
+[compaction_visit_marker_file_update_interval: <duration> | default = 1m]
+
+# How long cleaner visit marker file should be considered as expired and able to
+# be picked up by cleaner again.
+# CLI flag: -compactor.cleaner-visit-marker-timeout
+[cleaner_visit_marker_timeout: <duration> | default = 10m]
+
+# How frequently cleaner visit marker file should be updated when cleaning user.
+# CLI flag: -compactor.cleaner-visit-marker-file-update-interval
+[cleaner_visit_marker_file_update_interval: <duration> | default = 5m]
 
 # When enabled, index verification will ignore out of order label names.
 # CLI flag: -compactor.accept-malformed-index
@@ -3342,6 +3360,24 @@ query_rejection:
 # CLI flag: -compactor.tenant-shard-size
 [compactor_tenant_shard_size: <int> | default = 0]
 
+# Index size limit in bytes for each compaction partition. 0 means no limit
+# CLI flag: -compactor.partition-index-size-limit-in-bytes
+[compactor_partition_index_size_limit_in_bytes: <int> | default = 0]
+
+# Time series count limit for each compaction partition. 0 means no limit
+# CLI flag: -compactor.partition-series-count-limit
+[compactor_partition_series_count_limit: <int> | default = 0]
+
+# Index size limit in bytes for each level 1 compaction partition. 0 means no
+# limit
+# CLI flag: -compactor.partition-level1-index-size-limit-in-bytes
+[compactor_partition_level1_index_size_limit_in_bytes: <int> | default = 0]
+
+# Time series count limit for each level 1 compaction partition. 0 means no
+# limit
+# CLI flag: -compactor.partition-level1-series-count-limit
+[compactor_partition_level1_series_count_limit: <int> | default = 0]
+
 # S3 server-side encryption type. Required to enable server-side encryption
 # overrides for a specific tenant. If not set, the default S3 client settings
 # are used.

pkg/compactor/background_chunks_series_set.go

@@ -0,0 +1,60 @@
+package compactor
+
+import (
+	"context"
+
+	"github.com/prometheus/prometheus/storage"
+	"github.com/prometheus/prometheus/util/annotations"
+)
+
+type backgrounChunkSeriesSet struct {
+	nextSet chan storage.ChunkSeries
+	actual  storage.ChunkSeries
+	cs      storage.ChunkSeriesSet
+}
+
+func (b *backgrounChunkSeriesSet) Next() bool {
+	s, ok := <-b.nextSet
+	b.actual = s
+	return ok
+}
+
+func (b *backgrounChunkSeriesSet) At() storage.ChunkSeries {
+	return b.actual
+}
+
+func (b *backgrounChunkSeriesSet) Err() error {
+	return b.cs.Err()
+}
+
+func (b *backgrounChunkSeriesSet) Warnings() annotations.Annotations {
+	return b.cs.Warnings()
+}
+
+func (b *backgrounChunkSeriesSet) run(ctx context.Context) {
+	for {
+		if !b.cs.Next() {
+			close(b.nextSet)
+			return
+		}
+
+		select {
+		case b.nextSet <- b.cs.At():
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+func NewBackgroundChunkSeriesSet(ctx context.Context, cs storage.ChunkSeriesSet) storage.ChunkSeriesSet {
+	r := &backgrounChunkSeriesSet{
+		cs:      cs,
+		nextSet: make(chan storage.ChunkSeries, 1000),
+	}
+
+	go func() {
+		r.run(ctx)
+	}()
+
+	return r
+}

pkg/compactor/block_visit_marker.go

@@ -49,8 +49,12 @@ func (b *BlockVisitMarker) isVisitedByCompactor(blockVisitMarkerTimeout time.Dur
 	return b.CompactorID == compactorID && time.Now().Before(time.Unix(b.VisitTime, 0).Add(blockVisitMarkerTimeout))
 }
 
+func GetBlockVisitMarkerFile(blockID string) string {
+	return path.Join(blockID, BlockVisitMarkerFile)
+}
+
 func ReadBlockVisitMarker(ctx context.Context, bkt objstore.InstrumentedBucketReader, logger log.Logger, blockID string, blockVisitMarkerReadFailed prometheus.Counter) (*BlockVisitMarker, error) {
-	visitMarkerFile := path.Join(blockID, BlockVisitMarkerFile)
+	visitMarkerFile := GetBlockVisitMarkerFile(blockID)
 	visitMarkerFileReader, err := bkt.ReaderWithExpectedErrs(bkt.IsObjNotFoundErr).Get(ctx, visitMarkerFile)
 	if err != nil {
 		if bkt.IsObjNotFoundErr(err) {