diff --git a/pkg/compact/overlapping.go b/pkg/compact/overlapping.go index 1f211ce5539..81853f5c0c3 100644 --- a/pkg/compact/overlapping.go +++ b/pkg/compact/overlapping.go @@ -2,6 +2,10 @@ package compact import ( "context" + "fmt" + "os" + "path/filepath" + "github.com/go-kit/log" "github.com/go-kit/log/level" "github.com/oklog/ulid" @@ -11,8 +15,6 @@ import ( "github.com/thanos-io/objstore" "github.com/thanos-io/thanos/pkg/block" "github.com/thanos-io/thanos/pkg/block/metadata" - "os" - "path/filepath" ) type OverlappingCompactionLifecycleCallback struct { @@ -33,7 +35,7 @@ func (c *OverlappingCompactionLifecycleCallback) PreCompactionCallback(ctx conte previous := 0 for i, m := range toCompact { kept := toCompact[previous] - if previous == 0 || m.Thanos.Source == metadata.ReceiveSource || kept.MaxTime <= m.MinTime { + if i == 0 || m.Thanos.Source == metadata.ReceiveSource || kept.MaxTime <= m.MinTime { // no overlapping with previous blocks, skip it previous = i continue @@ -49,6 +51,15 @@ func (c *OverlappingCompactionLifecycleCallback) PreCompactionCallback(ctx conte return retry(err) } toCompact[i] = nil + } else if m.MinTime == kept.MinTime { + level.Warn(logger).Log("msg", "found overlapping block in plan, only keep the later one", + "toKeep", m.String(), "toDelete", kept.String()) + cg.overlappingBlocks.Inc() + if err := DeleteBlockNow(ctx, logger, cg.bkt, kept, c.metaDir); err != nil { + return retry(err) + } + toCompact[previous] = nil + previous = i } else { err := errors.Errorf("found partially overlapping block: %s -- %s", kept.String(), m.String()) if cg.enableVerticalCompaction { @@ -81,7 +92,9 @@ func FilterNilBlocks(blocks []*metadata.Meta) (res []*metadata.Meta) { func DeleteBlockNow(ctx context.Context, logger log.Logger, bkt objstore.Bucket, m *metadata.Meta, dir string) error { level.Warn(logger).Log("msg", "delete polluted block immediately", "block", m.String(), - "level", m.Compaction.Level, "source", m.Thanos.Source, "labels", m.Thanos.GetLabels()) + "level", m.Compaction.Level, "parents", fmt.Sprintf("%v", m.Compaction.Parents), + "resolution", m.Thanos.Downsample.Resolution, "source", m.Thanos.Source, "labels", m.Thanos.GetLabels(), + "series", m.Stats.NumSeries, "samples", m.Stats.NumSamples, "chunks", m.Stats.NumChunks) if err := block.Delete(ctx, logger, bkt, m.ULID); err != nil { return errors.Wrapf(err, "delete overlapping block %s", m.String()) } diff --git a/pkg/compact/overlapping_test.go b/pkg/compact/overlapping_test.go index 78f248156c7..59c57331c77 100644 --- a/pkg/compact/overlapping_test.go +++ b/pkg/compact/overlapping_test.go @@ -1,13 +1,174 @@ package compact import ( + "context" + "testing" + "time" + "github.com/efficientgo/core/testutil" + "github.com/go-kit/log" + "github.com/pkg/errors" + "github.com/thanos-io/objstore" "github.com/thanos-io/thanos/pkg/block/metadata" - "testing" + "github.com/thanos-io/thanos/pkg/compact/downsample" ) func TestFilterNilCompact(t *testing.T) { blocks := []*metadata.Meta{nil, nil} filtered := FilterNilBlocks(blocks) testutil.Equals(t, 0, len(filtered)) + + meta := []*metadata.Meta{ + createBlockMeta(6, 1, int64(time.Now().Add(-6*30*24*time.Hour).Unix()*1000), map[string]string{"a": "1"}, downsample.ResLevel0, []uint64{}), + nil, + createBlockMeta(7, 1, int64(time.Now().Add(-4*30*24*time.Hour).Unix()*1000), map[string]string{"b": "2"}, downsample.ResLevel1, []uint64{}), + createBlockMeta(8, 1, int64(time.Now().Add(-7*30*24*time.Hour).Unix()*1000), map[string]string{"a": "1", "b": "2"}, downsample.ResLevel2, []uint64{}), + nil, + } + testutil.Equals(t, 3, len(FilterNilBlocks(meta))) +} + +func TestPreCompactionCallback(t *testing.T) { + logger := log.NewNopLogger() + bkt := objstore.NewInMemBucket() + group := &Group{ + logger: log.NewNopLogger(), + bkt: bkt, + } + labels := map[string]string{"a": "1"} + callback := NewOverlappingCompactionLifecycleCallback() + for _, tcase := range []struct { + testName string + input []*metadata.Meta + enableVerticalCompaction bool + expectedSize int + expectedBlocks []*metadata.Meta + err error + }{ + { + testName: "empty blocks", + }, + { + testName: "no overlapping blocks", + input: []*metadata.Meta{ + createBlockMeta(6, 1, 3, labels, downsample.ResLevel0, []uint64{}), + createBlockMeta(7, 3, 5, labels, downsample.ResLevel0, []uint64{}), + createBlockMeta(8, 5, 10, labels, downsample.ResLevel0, []uint64{}), + }, + expectedSize: 3, + }, + { + testName: "duplicated blocks", + input: []*metadata.Meta{ + createBlockMeta(6, 1, 7, labels, downsample.ResLevel0, []uint64{}), + createBlockMeta(7, 1, 7, labels, downsample.ResLevel0, []uint64{}), + createBlockMeta(8, 1, 7, labels, downsample.ResLevel0, []uint64{}), + }, + expectedSize: 1, + expectedBlocks: []*metadata.Meta{ + createBlockMeta(6, 1, 7, labels, downsample.ResLevel0, []uint64{}), + }, + }, + { + testName: "receive blocks", + input: []*metadata.Meta{ + createReceiveBlockMeta(6, 1, 7, labels, downsample.ResLevel0, []uint64{}), + createReceiveBlockMeta(7, 1, 7, labels, downsample.ResLevel0, []uint64{}), + createReceiveBlockMeta(8, 1, 7, labels, downsample.ResLevel0, []uint64{}), + }, + expectedSize: 3, + }, + { + testName: "receive + compactor blocks", + input: []*metadata.Meta{ + createReceiveBlockMeta(6, 1, 7, labels, downsample.ResLevel0, []uint64{}), + createBlockMeta(7, 1, 7, labels, downsample.ResLevel0, []uint64{}), + createReceiveBlockMeta(8, 2, 7, labels, downsample.ResLevel0, []uint64{}), + }, + expectedSize: 2, + expectedBlocks: []*metadata.Meta{ + createReceiveBlockMeta(6, 1, 7, labels, downsample.ResLevel0, []uint64{}), + createReceiveBlockMeta(8, 2, 7, labels, downsample.ResLevel0, []uint64{}), + }, + }, + { + testName: "full overlapping blocks", + input: []*metadata.Meta{ + createBlockMeta(6, 1, 10, labels, downsample.ResLevel0, []uint64{}), + createBlockMeta(7, 3, 6, labels, downsample.ResLevel0, []uint64{}), + createBlockMeta(8, 5, 8, labels, downsample.ResLevel0, []uint64{}), + }, + expectedSize: 1, + expectedBlocks: []*metadata.Meta{ + createBlockMeta(6, 1, 10, labels, downsample.ResLevel0, []uint64{}), + }, + }, + { + testName: "part overlapping blocks", + input: []*metadata.Meta{ + createBlockMeta(1, 1, 2, labels, downsample.ResLevel0, []uint64{}), + createBlockMeta(2, 1, 6, labels, downsample.ResLevel0, []uint64{}), + createBlockMeta(3, 6, 8, labels, downsample.ResLevel0, []uint64{}), + }, + expectedSize: 2, + expectedBlocks: []*metadata.Meta{ + createBlockMeta(2, 1, 6, labels, downsample.ResLevel0, []uint64{}), + createBlockMeta(3, 6, 8, labels, downsample.ResLevel0, []uint64{}), + }, + }, + { + testName: "out of order blocks", + input: []*metadata.Meta{ + createBlockMeta(6, 2, 3, labels, downsample.ResLevel0, []uint64{}), + createBlockMeta(7, 0, 5, labels, downsample.ResLevel0, []uint64{}), + createBlockMeta(8, 5, 8, labels, downsample.ResLevel0, []uint64{}), + }, + err: halt(errors.Errorf("expect halt error")), + }, + { + testName: "partially overlapping blocks with vertical compaction off", + input: []*metadata.Meta{ + createBlockMeta(6, 2, 4, labels, downsample.ResLevel0, []uint64{}), + createBlockMeta(7, 3, 5, labels, downsample.ResLevel0, []uint64{}), + createBlockMeta(8, 5, 8, labels, downsample.ResLevel0, []uint64{}), + }, + err: halt(errors.Errorf("expect halt error")), + }, + { + testName: "partially overlapping blocks with vertical compaction on", + input: []*metadata.Meta{ + createBlockMeta(6, 2, 4, labels, downsample.ResLevel0, []uint64{}), + createBlockMeta(7, 3, 6, labels, downsample.ResLevel0, []uint64{}), + createBlockMeta(8, 5, 8, labels, downsample.ResLevel0, []uint64{}), + }, + enableVerticalCompaction: true, + expectedSize: 3, + }, + } { + if ok := t.Run(tcase.testName, func(t *testing.T) { + group.enableVerticalCompaction = tcase.enableVerticalCompaction + err := callback.PreCompactionCallback(context.Background(), logger, group, tcase.input) + if tcase.err != nil { + testutil.NotOk(t, err) + if IsHaltError(tcase.err) { + testutil.Assert(t, IsHaltError(err), "expected halt error") + } else if IsRetryError(tcase.err) { + testutil.Assert(t, IsRetryError(err), "expected retry error") + } + return + } + testutil.Equals(t, tcase.expectedSize, len(FilterNilBlocks(tcase.input))) + if tcase.expectedSize != len(tcase.input) { + testutil.Equals(t, tcase.expectedBlocks, FilterNilBlocks(tcase.input)) + } + }); !ok { + return + } + } +} + +func createReceiveBlockMeta(id uint64, minTime, maxTime int64, labels map[string]string, resolution int64, sources []uint64) *metadata.Meta { + m := createBlockMeta(id, minTime, maxTime, labels, resolution, sources) + m.Thanos.Source = metadata.ReceiveSource + return m }