[PLAT-102919] add a flag to control overlapping block removal behavior

Signed-off-by: Yi Jin <[email protected]>

cmd/thanos/compact.go

@@ -369,12 +369,14 @@ func runCompact(
 		compactMetrics.blocksMarked.WithLabelValues(metadata.NoCompactMarkFilename, metadata.IndexSizeExceedingNoCompactReason),
 	)
 	blocksCleaner := compact.NewBlocksCleaner(logger, insBkt, ignoreDeletionMarkFilter, deleteDelay, compactMetrics.blocksCleaned, compactMetrics.blockCleanupFailures)
-	compactor, err := compact.NewBucketCompactor(
+	compactor, err := compact.NewBucketCompactorWithCheckerAndCallback(
 		logger,
 		sy,
 		grouper,
 		planner,
 		comp,
+		compact.DefaultBlockDeletableChecker{},
+		compact.NewOverlappingCompactionLifecycleCallback(reg, conf.enableOverlappingRemoval),
 		compactDir,
 		insBkt,
 		conf.compactionConcurrency,
@@ -710,6 +712,7 @@ type compactConfig struct {
 	maxBlockIndexSize                              units.Base2Bytes
 	hashFunc                                       string
 	enableVerticalCompaction                       bool
+	enableOverlappingRemoval                       bool
 	dedupFunc                                      string
 	skipBlockWithOutOfOrderChunks                  bool
 	progressCalculateInterval                      time.Duration
@@ -786,6 +789,9 @@ func (cc *compactConfig) registerFlag(cmd extkingpin.FlagClause) {
 		"NOTE: This flag is ignored and (enabled) when --deduplication.replica-label flag is set.").
 		Hidden().Default("false").BoolVar(&cc.enableVerticalCompaction)
 
+	cmd.Flag("compact.enable-overlapping-removal", "In house flag to remove overlapping blocks. Turn this on to fix https://github.com/thanos-io/thanos/issues/6775.").
+		Default("false").BoolVar(&cc.enableOverlappingRemoval)
+
 	cmd.Flag("deduplication.func", "Experimental. Deduplication algorithm for merging overlapping blocks. "+
 		"Possible values are: \"\", \"penalty\". If no value is specified, the default compact deduplication merger is used, which performs 1:1 deduplication for samples. "+
 		"When set to penalty, penalty based deduplication algorithm will be used. At least one replica label has to be set via --deduplication.replica-label flag.").

pkg/block/metadata/meta.go

@@ -9,12 +9,13 @@ package metadata
 // this package.
 
 import (
-	"bytes"
 	"encoding/json"
 	"fmt"
 	"io"
 	"os"
 	"path/filepath"
+	"sort"
+	"strings"
 
 	"github.com/go-kit/log"
 	"github.com/oklog/ulid"
@@ -124,11 +125,12 @@ func (m *Thanos) GetTenant() string {
 }
 
 func (m *Thanos) GetLabels() string {
-	b := new(bytes.Buffer)
+	res := make([]string, 0, len(m.Labels))
 	for k, v := range m.Labels {
-		fmt.Fprintf(b, "%s=%s,", k, v)
+		res = append(res, fmt.Sprintf("%s=%s", k, v))
 	}
-	return b.String()
+	sort.Strings(res)
+	return strings.Join(res, ",")
 }
 
 // ConvertExtensions converts extensions with `any` type into specific type `v`

pkg/block/metadata/meta_test.go

@@ -347,7 +347,7 @@ func TestMeta_GetLabels(t *testing.T) {
 			Labels: map[string]string{"a": "b", "c": "d"},
 		},
 	}
-	testutil.Equals(t, "a=b,c=d,", m.Thanos.GetLabels())
+	testutil.Equals(t, "a=b,c=d", m.Thanos.GetLabels())
 }
 
 type TestExtensions struct {

pkg/compact/compact.go

@@ -236,7 +236,6 @@ type DefaultGrouper struct {
 	compactionRunsCompleted       *prometheus.CounterVec
 	compactionFailures            *prometheus.CounterVec
 	verticalCompactions           *prometheus.CounterVec
-	overlappingBlocks             *prometheus.CounterVec
 	garbageCollectedBlocks        prometheus.Counter
 	blocksMarkedForDeletion       prometheus.Counter
 	blocksMarkedForNoCompact      prometheus.Counter
@@ -284,10 +283,6 @@ func NewDefaultGrouper(
 			Name: "thanos_compact_group_vertical_compactions_total",
 			Help: "Total number of group compaction attempts that resulted in a new block based on overlapping blocks.",
 		}, []string{"resolution"}),
-		overlappingBlocks: promauto.With(reg).NewCounterVec(prometheus.CounterOpts{
-			Name: "thanos_compact_group_overlapping_blocks_total",
-			Help: "Total number of blocks that are overlapping and to be deleted.",
-		}, []string{"resolution", "tenant"}),
 		blocksMarkedForNoCompact:      blocksMarkedForNoCompact,
 		garbageCollectedBlocks:        garbageCollectedBlocks,
 		blocksMarkedForDeletion:       blocksMarkedForDeletion,
@@ -308,7 +303,6 @@ func NewDefaultGrouperWithMetrics(
 	compactionRunsCompleted *prometheus.CounterVec,
 	compactionFailures *prometheus.CounterVec,
 	verticalCompactions *prometheus.CounterVec,
-	overrlappingBlocks *prometheus.CounterVec,
 	blocksMarkedForDeletion prometheus.Counter,
 	garbageCollectedBlocks prometheus.Counter,
 	blocksMarkedForNoCompact prometheus.Counter,
@@ -326,7 +320,6 @@ func NewDefaultGrouperWithMetrics(
 		compactionRunsCompleted:       compactionRunsCompleted,
 		compactionFailures:            compactionFailures,
 		verticalCompactions:           verticalCompactions,
-		overlappingBlocks:             overrlappingBlocks,
 		blocksMarkedForNoCompact:      blocksMarkedForNoCompact,
 		garbageCollectedBlocks:        garbageCollectedBlocks,
 		blocksMarkedForDeletion:       blocksMarkedForDeletion,
@@ -359,7 +352,6 @@ func (g *DefaultGrouper) Groups(blocks map[ulid.ULID]*metadata.Meta) (res []*Gro
 				g.compactionRunsCompleted.WithLabelValues(resolutionLabel),
 				g.compactionFailures.WithLabelValues(resolutionLabel),
 				g.verticalCompactions.WithLabelValues(resolutionLabel),
-				g.overlappingBlocks.WithLabelValues(resolutionLabel, m.Thanos.GetLabels()),
 				g.garbageCollectedBlocks,
 				g.blocksMarkedForDeletion,
 				g.blocksMarkedForNoCompact,
@@ -400,7 +392,6 @@ type Group struct {
 	compactionRunsCompleted       prometheus.Counter
 	compactionFailures            prometheus.Counter
 	verticalCompactions           prometheus.Counter
-	overlappingBlocks             prometheus.Counter
 	groupGarbageCollectedBlocks   prometheus.Counter
 	blocksMarkedForDeletion       prometheus.Counter
 	blocksMarkedForNoCompact      prometheus.Counter
@@ -424,7 +415,6 @@ func NewGroup(
 	compactionRunsCompleted prometheus.Counter,
 	compactionFailures prometheus.Counter,
 	verticalCompactions prometheus.Counter,
-	overlappingBlocks prometheus.Counter,
 	groupGarbageCollectedBlocks prometheus.Counter,
 	blocksMarkedForDeletion prometheus.Counter,
 	blocksMarkedForNoCompact prometheus.Counter,
@@ -453,7 +443,6 @@ func NewGroup(
 		compactionRunsCompleted:       compactionRunsCompleted,
 		compactionFailures:            compactionFailures,
 		verticalCompactions:           verticalCompactions,
-		overlappingBlocks:             overlappingBlocks,
 		groupGarbageCollectedBlocks:   groupGarbageCollectedBlocks,
 		blocksMarkedForDeletion:       blocksMarkedForDeletion,
 		blocksMarkedForNoCompact:      blocksMarkedForNoCompact,

pkg/compact/overlapping.go

@@ -11,34 +11,45 @@ import (
 	"github.com/go-kit/log/level"
 	"github.com/oklog/ulid"
 	"github.com/pkg/errors"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/prometheus/prometheus/tsdb"
 	"github.com/thanos-io/objstore"
 	"github.com/thanos-io/thanos/pkg/block"
 	"github.com/thanos-io/thanos/pkg/block/metadata"
 )
 
 type OverlappingCompactionLifecycleCallback struct {
+	overlappingBlocks prometheus.Counter
 }
 
-func NewOverlappingCompactionLifecycleCallback() *OverlappingCompactionLifecycleCallback {
-	return &OverlappingCompactionLifecycleCallback{}
+func NewOverlappingCompactionLifecycleCallback(reg *prometheus.Registry, enabled bool) CompactionLifecycleCallback {
+	if enabled {
+		return OverlappingCompactionLifecycleCallback{
+			overlappingBlocks: promauto.With(reg).NewCounter(prometheus.CounterOpts{
+				Name: "thanos_compact_group_overlapping_blocks_total",
+				Help: "Total number of blocks that are overlapping and to be deleted.",
+			}),
+		}
+	}
+	return DefaultCompactionLifecycleCallback{}
 }
 
 // PreCompactionCallback given the assumption that toCompact is sorted by MinTime in ascending order from Planner
 // (not guaranteed on MaxTime order), we will detect overlapping blocks and delete them while retaining all others.
-func (c *OverlappingCompactionLifecycleCallback) PreCompactionCallback(ctx context.Context, logger log.Logger, cg *Group, toCompact []*metadata.Meta) error {
+func (o OverlappingCompactionLifecycleCallback) PreCompactionCallback(ctx context.Context, logger log.Logger, cg *Group, toCompact []*metadata.Meta) error {
 	if len(toCompact) == 0 {
 		return nil
 	}
 	prev := 0
 	for curr, currB := range toCompact {
 		prevB := toCompact[prev]
 		if curr == 0 || currB.Thanos.Source == metadata.ReceiveSource || prevB.MaxTime <= currB.MinTime {
-			// no  overlapping with previous blocks, skip it
+			// no overlapping with previous blocks, skip it
 			prev = curr
 			continue
 		} else if currB.MinTime < prevB.MinTime {
-			// halt when the assumption is broken, need manual investigation
+			// halt when the assumption is broken, the input toCompact isn't sorted by minTime, need manual investigation
 			return halt(errors.Errorf("later blocks has smaller minTime than previous block: %s -- %s", prevB.String(), currB.String()))
 		} else if prevB.MaxTime < currB.MaxTime && prevB.MinTime != currB.MinTime {
 			err := errors.Errorf("found partially overlapping block: %s -- %s", prevB.String(), currB.String())
@@ -50,7 +61,14 @@ func (c *OverlappingCompactionLifecycleCallback) PreCompactionCallback(ctx conte
 				return halt(err)
 			}
 		} else if prevB.MinTime == currB.MinTime && prevB.MaxTime == currB.MaxTime {
-			continue
+			if prevB.Stats.NumSeries != currB.Stats.NumSeries || prevB.Stats.NumSamples != currB.Stats.NumSamples {
+				level.Warn(logger).Log("msg", "found same time range but different stats, keep both blocks",
+					"prev", prevB.String(), "prevSeries", prevB.Stats.NumSeries, "prevSamples", prevB.Stats.NumSamples,
+					"curr", currB.String(), "currSeries", currB.Stats.NumSeries, "currSamples", currB.Stats.NumSamples,
+				)
+				prev = curr
+				continue
+			}
 		}
 		// prev min <= curr min < prev max
 		toDelete := -1
@@ -64,7 +82,7 @@ func (c *OverlappingCompactionLifecycleCallback) PreCompactionCallback(ctx conte
 			level.Warn(logger).Log("msg", "found overlapping block in plan, keep current block",
 				"toKeep", currB.String(), "toDelete", prevB.String())
 		}
-		cg.overlappingBlocks.Inc()
+		o.overlappingBlocks.Inc()
 		if err := DeleteBlockNow(ctx, logger, cg.bkt, toCompact[toDelete]); err != nil {
 			return retry(err)
 		}
@@ -73,11 +91,11 @@ func (c *OverlappingCompactionLifecycleCallback) PreCompactionCallback(ctx conte
 	return nil
 }
 
-func (c *OverlappingCompactionLifecycleCallback) PostCompactionCallback(_ context.Context, _ log.Logger, _ *Group, _ ulid.ULID) error {
+func (o OverlappingCompactionLifecycleCallback) PostCompactionCallback(_ context.Context, _ log.Logger, _ *Group, _ ulid.ULID) error {
 	return nil
 }
 
-func (c *OverlappingCompactionLifecycleCallback) GetBlockPopulator(_ context.Context, _ log.Logger, _ *Group) (tsdb.BlockPopulator, error) {
+func (o OverlappingCompactionLifecycleCallback) GetBlockPopulator(_ context.Context, _ log.Logger, _ *Group) (tsdb.BlockPopulator, error) {
 	return tsdb.DefaultBlockPopulator{}, nil
 }