Generate "parallelizable" plans for compaction

pkg/compactor/compactor.go

@@ -62,7 +62,7 @@ var (
 			return nil, nil, err
 		}
 
-		planner := compact.NewTSDBBasedPlanner(logger, cfg.BlockRanges.ToMilliseconds())
+		planner, _ := NewPlanner(logger, cfg.BlockRanges.ToMilliseconds(), "default")
 		return compactor, planner, nil
 	}
 )

pkg/compactor/planner.go

@@ -0,0 +1,58 @@
+package compactor
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/go-kit/kit/log"
+	"github.com/go-kit/kit/log/level"
+	"github.com/thanos-io/thanos/pkg/block/metadata"
+	"github.com/thanos-io/thanos/pkg/compact"
+)
+
+var (
+	errPlannerNotImplemented = errors.New("planner type not implemented")
+)
+
+type Planner struct {
+	blocksPlanner compact.Planner
+	logger        log.Logger
+}
+
+func NewPlanner(logger log.Logger, ranges []int64, plannerType string) (*Planner, error) {
+	if logger == nil {
+		logger = log.NewNopLogger()
+	}
+
+	var blocksPlanner compact.Planner
+	switch plannerType {
+	case "parallel":
+		blocksPlanner = NewParallelPlanner(logger, ranges)
+	case "default":
+		blocksPlanner = compact.NewTSDBBasedPlanner(logger, ranges)
+	default:
+		return nil, errPlannerNotImplemented
+	}
+
+	p := &Planner{
+		logger:        logger,
+		blocksPlanner: blocksPlanner,
+	}
+
+	return p, nil
+}
+
+func (p *Planner) Plan(ctx context.Context, metasByMinTime []*metadata.Meta) ([]*metadata.Meta, error) {
+	return p.blocksPlanner.Plan(ctx, metasByMinTime)
+}
+
+func (p *Planner) PrintPlan(ctx context.Context, metasByMinTime []*metadata.Meta) {
+	toCompact, err := p.Plan(ctx, metasByMinTime)
+
+	if err != nil {
+		return
+	}
+
+	level.Info(p.logger).Log("msg", "Compaction plan generated: ", "plan", fmt.Sprintf("%v", toCompact))
+}

pkg/compactor/planner_test.go

@@ -0,0 +1,58 @@
+package compactor
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/thanos-io/thanos/pkg/block/metadata"
+)
+
+type PlannerMock struct {
+	mock.Mock
+}
+
+func (p *PlannerMock) Plan(ctx context.Context, metasByMinTime []*metadata.Meta) ([]*metadata.Meta, error) {
+	args := p.Called(ctx, metasByMinTime)
+	return args.Get(0).([]*metadata.Meta), args.Error(1)
+}
+
+func TestPlanner_ShouldReturnParallelPlanner(t *testing.T) {
+	var ranges []int64
+	p, _ := NewPlanner(nil, ranges, "parallel")
+
+	parallelPlanner := &ParallelPlanner{}
+
+	assert.IsType(t, parallelPlanner, p.blocksPlanner)
+}
+
+func TestPlanner_ShouldReturnDefaultPlanner(t *testing.T) {
+	var ranges []int64
+	p, _ := NewPlanner(nil, ranges, "default")
+
+	parallelPlanner := &ParallelPlanner{}
+
+	assert.IsType(t, parallelPlanner, p.blocksPlanner)
+}
+
+func TestPlanner_ShouldErrorOnNonExistentPlanner(t *testing.T) {
+	var ranges []int64
+	_, err := NewPlanner(nil, ranges, "non-existent")
+
+	assert.ErrorIs(t, err, errPlannerNotImplemented)
+}
+
+func TestPlanner_PlanShouldCallBlocksPlannerPlan(t *testing.T) {
+	blockPlannerMock := &PlannerMock{}
+	blockPlannerMock.On("Plan", mock.Anything, mock.Anything).Return([]*metadata.Meta{}, nil)
+
+	p := &Planner{
+		blocksPlanner: blockPlannerMock,
+	}
+
+	blockMetas := []*metadata.Meta{}
+	p.Plan(context.Background(), blockMetas)
+
+	blockPlannerMock.AssertCalled(t, "Plan", context.Background(), blockMetas)
+}