Skip to content

Commit

Permalink
[PLAT-104290] fixed some bugs and added unit tests
Browse files Browse the repository at this point in the history 
Signed-off-by: Yi Jin <yi.jin@databricks.com>
  • Loading branch information
@jnyi
jnyi committed Mar 22, 2024
1 parent 5fc3435 commit e5c7ca0
Show file tree
Hide file tree
Showing 2 changed files with 179 additions and 5 deletions.
21 changes: 17 additions & 4 deletions pkg/compact/overlapping.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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 {
Expand All @@ -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
Expand All @@ -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 {
Expand Down Expand Up @@ -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())
}
Expand Down
163 changes: 162 additions & 1 deletion pkg/compact/overlapping_test.go
View file
Original file line number Diff line number Diff line change
@@ -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 {

Check failure on line 170 in pkg/compact/overlapping_test.go

View workflow job for this annotation

GitHub Actions / Linters (Static Analysis) for Go 

`createReceiveBlockMeta` - `maxTime` always receives `7` (unparam)
m := createBlockMeta(id, minTime, maxTime, labels, resolution, sources)
m.Thanos.Source = metadata.ReceiveSource
return m
}

0 comments on commit e5c7ca0

Please sign in to comment.