[ES-1393569] add compaction errors handler

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

Author: jnyi

cmd/thanos/compact.go

@@ -404,7 +404,7 @@ func runCompact(
 		planner,
 		comp,
 		compact.DefaultBlockDeletableChecker{},
-		compact.NewOverlappingCompactionLifecycleCallback(reg, conf.enableOverlappingRemoval),
+		compact.NewOverlappingCompactionLifecycleCallback(reg, logger, conf.enableOverlappingRemoval),
 		compactDir,
 		insBkt,
 		conf.compactionConcurrency,

pkg/compact/compact.go

@@ -842,6 +842,7 @@ type CompactionLifecycleCallback interface {
 	PreCompactionCallback(ctx context.Context, logger log.Logger, group *Group, toCompactBlocks []*metadata.Meta) error
 	PostCompactionCallback(ctx context.Context, logger log.Logger, group *Group, blockID ulid.ULID) error
 	GetBlockPopulator(ctx context.Context, logger log.Logger, group *Group) (tsdb.BlockPopulator, error)
+	HandleError(ctx context.Context, logger log.Logger, group *Group, toCompactBlocks []*metadata.Meta, err error)
 }
 
 type DefaultCompactionLifecycleCallback struct {
@@ -873,6 +874,9 @@ func (c DefaultCompactionLifecycleCallback) GetBlockPopulator(_ context.Context,
 	return tsdb.DefaultBlockPopulator{}, nil
 }
 
+func (c DefaultCompactionLifecycleCallback) HandleError(_ context.Context, _ log.Logger, _ *Group, _ []*metadata.Meta, err error) {
+}
+
 // Compactor provides compaction against an underlying storage of time series data.
 // It is similar to tsdb.Compactor but only relevant methods are kept. Plan and Write are removed.
 // TODO(bwplotka): Split the Planner from Compactor on upstream as well, so we can import it.
@@ -1253,6 +1257,7 @@ func (cg *Group) compact(ctx context.Context, dir string, planner Planner, comp
 		compIDs, e = comp.CompactWithBlockPopulator(dir, toCompactDirs, nil, populateBlockFunc)
 		return e
 	}); err != nil {
+		compactionLifecycleCallback.HandleError(ctx, cg.logger, cg, toCompact, err)
 		return false, nil, halt(errors.Wrapf(err, "compact blocks %v", toCompactDirs))
 	}
 	if len(compIDs) == 0 {

pkg/compact/overlapping.go

@@ -6,6 +6,7 @@ package compact
 import (
 	"context"
 	"fmt"
+	"strings"
 
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
@@ -18,16 +19,22 @@ import (
 	"github.com/thanos-io/thanos/pkg/block/metadata"
 )
 
-const overlappingReason = "blocks-overlapping"
+const (
+	overlappingReason = "blocks-overlapping"
+
+	symbolTableSizeExceedsError = "symbol table size exceeds"
+	symbolTableSizeLimit        = 1024 * 1024
+)
 
 type OverlappingCompactionLifecycleCallback struct {
 	overlappingBlocks prometheus.Counter
 	noCompaction      prometheus.Counter
 	noDownsampling    prometheus.Counter
 }
 
-func NewOverlappingCompactionLifecycleCallback(reg *prometheus.Registry, enabled bool) CompactionLifecycleCallback {
+func NewOverlappingCompactionLifecycleCallback(reg *prometheus.Registry, logger log.Logger, enabled bool) CompactionLifecycleCallback {
 	if enabled {
+		level.Info(logger).Log("msg", "enabled overlapping blocks compaction lifecycle callback")
 		return OverlappingCompactionLifecycleCallback{
 			overlappingBlocks: promauto.With(reg).NewCounter(prometheus.CounterOpts{
 				Name: "thanos_compact_group_overlapping_blocks_total",
@@ -122,3 +129,23 @@ func (o OverlappingCompactionLifecycleCallback) PostCompactionCallback(_ context
 func (o OverlappingCompactionLifecycleCallback) GetBlockPopulator(_ context.Context, _ log.Logger, _ *Group) (tsdb.BlockPopulator, error) {
 	return tsdb.DefaultBlockPopulator{}, nil
 }
+
+func (o OverlappingCompactionLifecycleCallback) HandleError(ctx context.Context, logger log.Logger, g *Group, toCompact []*metadata.Meta, compactErr error) {
+	level.Error(logger).Log("msg", "failed to compact blocks", "err", compactErr)
+	if strings.Contains(compactErr.Error(), symbolTableSizeExceedsError) {
+		for _, m := range toCompact {
+			if m.Thanos.Source != metadata.ReceiveSource {
+				level.Debug(logger).Log("msg", "bypass blocks that are already compacted", "block", m.String())
+				continue
+			}
+			if m.Stats.NumSeries < symbolTableSizeLimit {
+				level.Warn(logger).Log("msg", "bypass small blocks", "block", m.String(), "series", m.Stats.NumSeries)
+				continue
+			}
+			if err := block.MarkForNoCompact(ctx, logger, g.bkt, m.ULID, symbolTableSizeExceedsError,
+				fmt.Sprintf("failed to compact blocks: %s", m.ULID.String()), o.noCompaction); err != nil {
+				level.Error(logger).Log("msg", "failed to mark block for no compact", "block", m.String(), "err", err)
+			}
+		}
+	}
+}

pkg/compact/overlapping_test.go

@@ -21,7 +21,7 @@ import (
 func TestPreCompactionCallback(t *testing.T) {
 	reg := prometheus.NewRegistry()
 	logger := log.NewNopLogger()
-	callback := NewOverlappingCompactionLifecycleCallback(reg, true)
+	callback := NewOverlappingCompactionLifecycleCallback(reg, logger, true)
 	for _, tcase := range []struct {
 		testName      string
 		input         []*metadata.Meta
@@ -145,6 +145,28 @@ func TestPreCompactionCallback(t *testing.T) {
 	}
 }
 
+func TestHandleError(t *testing.T) {
+	reg := prometheus.NewRegistry()
+	logger := log.NewNopLogger()
+	callback := NewOverlappingCompactionLifecycleCallback(reg, logger, true)
+	for _, tcase := range []struct {
+		testName string
+		input    []*metadata.Meta
+		err      error
+	}{
+		{
+			testName: "empty error",
+		},
+	} {
+		t.Run(tcase.testName, func(t *testing.T) {
+			ctx := context.Background()
+			bkt := objstore.NewInMemBucket()
+			group := &Group{logger: log.NewNopLogger(), bkt: bkt}
+			callback.HandleError(ctx, logger, group, tcase.input, tcase.err)
+		})
+	}
+}
+
 func createCustomBlockMeta(id uint64, minTime, maxTime int64, source metadata.SourceType, numSeries uint64) *metadata.Meta {
 	labels := map[string]string{"a": "1"}
 	m := createBlockMeta(id, minTime, maxTime, labels, downsample.ResLevel0, []uint64{})