add unit tests

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

Author: jnyi

pkg/compact/compact.go

@@ -842,7 +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)
+	HandleError(ctx context.Context, logger log.Logger, group *Group, toCompactBlocks []*metadata.Meta, err error) int // how many error cases it handles
 }
 
 type DefaultCompactionLifecycleCallback struct {
@@ -874,7 +874,8 @@ func (c DefaultCompactionLifecycleCallback) GetBlockPopulator(_ context.Context,
 	return tsdb.DefaultBlockPopulator{}, nil
 }
 
-func (c DefaultCompactionLifecycleCallback) HandleError(_ context.Context, _ log.Logger, _ *Group, _ []*metadata.Meta, err error) {
+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.
@@ -1257,8 +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 {
-		compactionLifecycleCallback.HandleError(ctx, cg.logger, cg, toCompact, err)
-		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

@@ -23,7 +23,7 @@ const (
 	overlappingReason = "blocks-overlapping"
 
 	symbolTableSizeExceedsError = "symbol table size exceeds"
-	symbolTableSizeLimit        = 1024 * 1024
+	symbolTableSizeLimit        = 128 * 1024
 )
 
 type OverlappingCompactionLifecycleCallback struct {
@@ -130,7 +130,11 @@ func (o OverlappingCompactionLifecycleCallback) GetBlockPopulator(_ context.Cont
 	return tsdb.DefaultBlockPopulator{}, nil
 }
 
-func (o OverlappingCompactionLifecycleCallback) HandleError(ctx context.Context, logger log.Logger, g *Group, toCompact []*metadata.Meta, compactErr error) {
+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 {
@@ -142,10 +146,12 @@ func (o OverlappingCompactionLifecycleCallback) HandleError(ctx context.Context,
 				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"
@@ -150,19 +151,48 @@ func TestHandleError(t *testing.T) {
 	logger := log.NewNopLogger()
 	callback := NewOverlappingCompactionLifecycleCallback(reg, logger, true)
 	for _, tcase := range []struct {
-		testName string
-		input    []*metadata.Meta
-		err      error
+		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}
-			callback.HandleError(ctx, logger, group, tcase.input, tcase.err)
+			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)
+				}
+			}
 		})
 	}
 }