[ES-1393569] azure compactor crash loop due to symbol table size limits (#146)

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/block/fetcher.go

@@ -345,7 +345,7 @@ func (f *ConcurrentLister) GetActiveAndPartialBlockIDs(ctx context.Context, ch c
 		case metaChan <- id:
 		}
 		return nil
-	}); err != nil {
+	}, objstore.WithUpdatedAt()); err != nil {
 		return nil, err
 	}
 	close(metaChan)

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) int // how many error cases it handles
 }
 
 type DefaultCompactionLifecycleCallback struct {
@@ -873,6 +874,10 @@ func (c DefaultCompactionLifecycleCallback) GetBlockPopulator(_ context.Context,
 	return tsdb.DefaultBlockPopulator{}, nil
 }
 
+func (c DefaultCompactionLifecycleCallback) HandleError(_ context.Context, _ log.Logger, _ *Group, _ []*metadata.Meta, err error) int {
+	return 0
+}
+
 // 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,7 +1258,8 @@ func (cg *Group) compact(ctx context.Context, dir string, planner Planner, comp
 		compIDs, e = comp.CompactWithBlockPopulator(dir, toCompactDirs, nil, populateBlockFunc)
 		return e
 	}); err != nil {
-		return false, nil, halt(errors.Wrapf(err, "compact blocks %v", toCompactDirs))
+		handledErrs := compactionLifecycleCallback.HandleError(ctx, cg.logger, cg, toCompact, err)
+		return false, nil, halt(errors.Wrapf(err, "compact blocks %v, handled %d errors", toCompactDirs, handledErrs))
 	}
 	if len(compIDs) == 0 {
 		// No compacted blocks means all compacted blocks are of no sample.

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        = 128 * 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,29 @@ 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) int {
+	handledErrs := 0
+	if compactErr == nil {
+		return handledErrs
+	}
+	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
+			}
+			handledErrs++
+			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)
+			}
+		}
+	}
+	return handledErrs
+}

pkg/compact/overlapping_test.go

@@ -13,6 +13,7 @@ import (
 	"github.com/go-kit/log"
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/stretchr/testify/require"
 	"github.com/thanos-io/objstore"
 	"github.com/thanos-io/thanos/pkg/block/metadata"
 	"github.com/thanos-io/thanos/pkg/compact/downsample"
@@ -21,7 +22,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 +146,57 @@ 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
+		handledErrs int
+		errBlockIdx int
+	}{
+		{
+			testName: "empty error",
+			input:    []*metadata.Meta{},
+		},
+		{
+			testName: "non empty error but not handled",
+			input: []*metadata.Meta{
+				createCustomBlockMeta(1, 1, 2, metadata.ReceiveSource, 1),
+				createCustomBlockMeta(2, 1, 6, metadata.CompactorSource, 1),
+			},
+			err:         errors.New("some error"),
+			handledErrs: 0,
+		},
+		{
+			testName: "non empty error symbol table size exceeds",
+			input: []*metadata.Meta{
+				createCustomBlockMeta(1, 1, 2, metadata.ReceiveSource, 1),
+				createCustomBlockMeta(2, 1, 6, metadata.CompactorSource, 1),
+				createCustomBlockMeta(3, 1, 6, metadata.ReceiveSource, 1024*1024),
+			},
+			err:         errors.New(symbolTableSizeExceedsError + " 1024*1024"),
+			handledErrs: 1,
+			errBlockIdx: 2,
+		},
+	} {
+		t.Run(tcase.testName, func(t *testing.T) {
+			ctx := context.Background()
+			bkt := objstore.NewInMemBucket()
+			group := &Group{logger: log.NewNopLogger(), bkt: bkt}
+			require.Equal(t, tcase.handledErrs, callback.HandleError(ctx, logger, group, tcase.input, tcase.err))
+			if tcase.handledErrs > 0 {
+				for i := 0; i < len(tcase.input); i++ {
+					ok, _ := bkt.Exists(ctx, path.Join(tcase.input[i].ULID.String(), metadata.NoCompactMarkFilename))
+					require.Equal(t, i == tcase.errBlockIdx, ok)
+				}
+			}
+		})
+	}
+}
+
 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{})