feat(v2): time-based compaction

Author: kolesnikovae

Diff for: pkg/experiment/metastore/compaction/compactor/compaction_queue.go

@@ -12,6 +12,8 @@ import (
 	"github.com/grafana/pyroscope/pkg/util"
 )
 
+const defaultBlockBatchSize = 20
+
 type compactionKey struct {
 	// Order of the fields is not important.
 	// Can be generalized.
@@ -21,14 +23,14 @@ type compactionKey struct {
 }
 
 type compactionQueue struct {
-	strategy   Strategy
+	config     Config
 	registerer prometheus.Registerer
 	levels     []*blockQueue
 }
 
 // blockQueue stages blocks as they are being added. Once a batch of blocks
-// within the compaction key reaches a certain size, it is pushed to the linked
-// list in the arrival order and to the compaction key queue.
+// within the compaction key reaches a certain size or age, it is pushed to
+// the linked list in the arrival order and to the compaction key queue.
 //
 // This allows to iterate over the blocks in the order of arrival within the
 // compaction dimension, while maintaining an ability to remove blocks from the
@@ -38,7 +40,7 @@ type compactionQueue struct {
 // the queue is through explicit removal. Batch and block iterators provide
 // the read access.
 type blockQueue struct {
-	strategy   Strategy
+	config     Config
 	registerer prometheus.Registerer
 	staged     map[compactionKey]*stagedBlocks
 	// Batches ordered by arrival.
@@ -99,9 +101,9 @@ type batch struct {
 	next, prev *batch
 }
 
-func newCompactionQueue(strategy Strategy, registerer prometheus.Registerer) *compactionQueue {
+func newCompactionQueue(strategy Config, registerer prometheus.Registerer) *compactionQueue {
 	return &compactionQueue{
-		strategy:   strategy,
+		config:     strategy,
 		registerer: registerer,
 	}
 }
@@ -142,15 +144,15 @@ func (q *compactionQueue) blockQueue(l uint32) *blockQueue {
 	}
 	level := q.levels[l]
 	if level == nil {
-		level = newBlockQueue(q.strategy, q.registerer)
+		level = newBlockQueue(q.config, q.registerer)
 		q.levels[l] = level
 	}
 	return level
 }
 
-func newBlockQueue(strategy Strategy, registerer prometheus.Registerer) *blockQueue {
+func newBlockQueue(strategy Config, registerer prometheus.Registerer) *blockQueue {
 	return &blockQueue{
-		strategy:   strategy,
+		config:     strategy,
 		registerer: registerer,
 		staged:     make(map[compactionKey]*stagedBlocks),
 		updates:    new(priorityBlockQueue),
@@ -203,7 +205,7 @@ func (s *stagedBlocks) push(block blockEntry) bool {
 	s.batch.blocks = append(s.batch.blocks, block)
 	s.batch.size++
 	s.stats.blocks.Add(1)
-	if s.queue.strategy.flush(s.batch) && !s.flush() {
+	if s.queue.config.exceedsMaxSize(s.batch) && !s.flush() {
 		// An attempt to flush the same batch twice.
 		// Should not be possible.
 		return false
@@ -317,7 +319,7 @@ func (q *blockQueue) flushOldest(now int64) {
 		return
 	}
 	oldest := (*q.updates)[0]
-	if !q.strategy.flushByAge(oldest.batch, now) {
+	if !q.config.exceedsMaxAge(oldest.batch, now) {
 		return
 	}
 	heap.Pop(q.updates)
@@ -393,6 +395,22 @@ func (it *blockIter) setBatch(b *batch) {
 	it.i = 0
 }
 
+func (it *blockIter) peek() (string, bool) {
+	for it.batch != nil {
+		if it.i >= len(it.batch.blocks) {
+			it.setBatch(it.batch.next)
+			continue
+		}
+		entry := it.batch.blocks[it.i]
+		if _, visited := it.visited[entry.id]; visited {
+			it.i++
+			continue
+		}
+		return entry.id, true
+	}
+	return "", false
+}
+
 func (it *blockIter) next() (string, bool) {
 	for it.batch != nil {
 		if it.i >= len(it.batch.blocks) {

Diff for: pkg/experiment/metastore/compaction/compactor/compaction_queue_bench_test.go

@@ -9,13 +9,7 @@ import (
 )
 
 func BenchmarkCompactionQueue_Push(b *testing.B) {
-	s := Strategy{
-		MaxBlocksPerLevel: []uint{20, 10, 10},
-		MaxBlocksDefault:  defaultBlockBatchSize,
-		MaxBatchAge:       defaultMaxBlockBatchAge,
-	}
-
-	q := newCompactionQueue(s, nil)
+	q := newCompactionQueue(DefaultConfig(), nil)
 	const (
 		tenants = 1
 		levels  = 1

Diff for: pkg/experiment/metastore/compaction/compactor/compaction_queue_test.go

@@ -15,7 +15,7 @@ import (
 func testBlockEntry(id int) blockEntry { return blockEntry{id: strconv.Itoa(id)} }
 
 func TestBlockQueue_Push(t *testing.T) {
-	q := newBlockQueue(Strategy{MaxBlocksDefault: 3}, nil)
+	q := newBlockQueue(Config{Levels: []LevelConfig{{MaxBlocks: 3}}}, nil)
 	key := compactionKey{tenant: "t", shard: 1}
 
 	result := q.stagedBlocks(key).push(testBlockEntry(1))
@@ -46,7 +46,7 @@ func TestBlockQueue_Push(t *testing.T) {
 }
 
 func TestBlockQueue_DuplicateBlock(t *testing.T) {
-	q := newBlockQueue(Strategy{MaxBlocksDefault: 3}, nil)
+	q := newBlockQueue(Config{Levels: []LevelConfig{{MaxBlocks: 3}}}, nil)
 	key := compactionKey{tenant: "t", shard: 1}
 
 	require.True(t, q.stagedBlocks(key).push(testBlockEntry(1)))
@@ -56,7 +56,7 @@ func TestBlockQueue_DuplicateBlock(t *testing.T) {
 }
 
 func TestBlockQueue_Remove(t *testing.T) {
-	q := newBlockQueue(Strategy{MaxBlocksDefault: 3}, nil)
+	q := newBlockQueue(Config{Levels: []LevelConfig{{MaxBlocks: 3}}}, nil)
 	key := compactionKey{tenant: "t", shard: 1}
 	q.stagedBlocks(key).push(testBlockEntry(1))
 	q.stagedBlocks(key).push(testBlockEntry(2))
@@ -73,7 +73,7 @@ func TestBlockQueue_Remove(t *testing.T) {
 }
 
 func TestBlockQueue_RemoveNotFound(t *testing.T) {
-	q := newBlockQueue(Strategy{MaxBlocksDefault: 3}, nil)
+	q := newBlockQueue(Config{Levels: []LevelConfig{{MaxBlocks: 3}}}, nil)
 	key := compactionKey{tenant: "t", shard: 1}
 	remove(q, key, "1")
 	q.stagedBlocks(key).push(testBlockEntry(1))
@@ -85,7 +85,7 @@ func TestBlockQueue_RemoveNotFound(t *testing.T) {
 }
 
 func TestBlockQueue_Linking(t *testing.T) {
-	q := newBlockQueue(Strategy{MaxBlocksDefault: 2}, nil)
+	q := newBlockQueue(Config{Levels: []LevelConfig{{MaxBlocks: 2}}}, nil)
 	key := compactionKey{tenant: "t", shard: 1}
 
 	q.stagedBlocks(key).push(testBlockEntry(1))
@@ -125,7 +125,7 @@ func TestBlockQueue_EmptyQueue(t *testing.T) {
 		numBlocksPerKey = 100
 	)
 
-	q := newBlockQueue(Strategy{MaxBlocksDefault: 3}, nil)
+	q := newBlockQueue(Config{Levels: []LevelConfig{{MaxBlocks: 3}}}, nil)
 	keys := make([]compactionKey, numKeys)
 	for i := 0; i < numKeys; i++ {
 		keys[i] = compactionKey{
@@ -165,10 +165,10 @@ func TestBlockQueue_EmptyQueue(t *testing.T) {
 }
 
 func TestBlockQueue_FlushByAge(t *testing.T) {
-	s := Strategy{
-		MaxBlocksDefault: 5,
-		MaxBatchAge:      1,
-	}
+	s := Config{Levels: []LevelConfig{
+		{MaxBlocks: 3, MaxAge: 1},
+		{MaxBlocks: 5, MaxAge: 1},
+	}}
 
 	c := newCompactionQueue(s, nil)
 	for _, e := range []compaction.BlockEntry{
@@ -200,7 +200,7 @@ func TestBlockQueue_FlushByAge(t *testing.T) {
 }
 
 func TestBlockQueue_BatchIterator(t *testing.T) {
-	q := newBlockQueue(Strategy{MaxBlocksDefault: 3}, nil)
+	q := newBlockQueue(Config{Levels: []LevelConfig{{MaxBlocks: 3}}}, nil)
 	keys := []compactionKey{
 		{tenant: "t-1", shard: 1},
 		{tenant: "t-2", shard: 2},

Diff for: pkg/experiment/metastore/compaction/compactor/compactor.go

@@ -1,7 +1,6 @@
 package compactor
 
 import (
-	"flag"
 	"time"
 
 	"github.com/hashicorp/raft"
@@ -31,15 +30,6 @@ type BlockQueueStore interface {
 	CreateBuckets(*bbolt.Tx) error
 }
 
-type Config struct {
-	Strategy
-}
-
-func (c *Config) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
-	c.Strategy = DefaultStrategy()
-	// TODO
-}
-
 type Compactor struct {
 	config     Config
 	queue      *compactionQueue
@@ -53,7 +43,7 @@ func NewCompactor(
 	tombstones Tombstones,
 	reg prometheus.Registerer,
 ) *Compactor {
-	queue := newCompactionQueue(config.Strategy, reg)
+	queue := newCompactionQueue(config, reg)
 	return &Compactor{
 		config:     config,
 		queue:      queue,
@@ -67,7 +57,7 @@ func NewStore() *store.BlockQueueStore {
 }
 
 func (c *Compactor) Compact(tx *bbolt.Tx, entry compaction.BlockEntry) error {
-	if uint(entry.Level) >= c.config.MaxLevel {
+	if int(entry.Level) >= len(c.config.Levels) {
 		return nil
 	}
 	if err := c.store.StoreEntry(tx, entry); err != nil {
@@ -82,12 +72,14 @@ func (c *Compactor) enqueue(e compaction.BlockEntry) bool {
 }
 
 func (c *Compactor) NewPlan(cmd *raft.Log) compaction.Plan {
+	now := cmd.AppendedAt.UnixNano()
 	before := cmd.AppendedAt.Add(-c.config.CleanupDelay)
 	tombstones := c.tombstones.ListTombstones(before)
 	return &plan{
 		compactor:  c,
 		tombstones: tombstones,
 		blocks:     newBlockIter(),
+		now:        now,
 	}
 }
 

Diff for: pkg/experiment/metastore/compaction/compactor/compactor_config.go

@@ -0,0 +1,75 @@
+package compactor
+
+import (
+	"flag"
+	"time"
+)
+
+type Config struct {
+	Levels []LevelConfig
+
+	CleanupBatchSize   int32
+	CleanupDelay       time.Duration
+	CleanupJobMinLevel int32
+	CleanupJobMaxLevel int32
+}
+
+type LevelConfig struct {
+	MaxBlocks uint
+	MaxAge    int64
+}
+
+func DefaultConfig() Config {
+	return Config{
+		Levels: []LevelConfig{
+			{MaxBlocks: 20, MaxAge: int64(36 * time.Second)},
+			{MaxBlocks: 10, MaxAge: int64(360 * time.Second)},
+			{MaxBlocks: 10, MaxAge: int64(3600 * time.Second)},
+		},
+
+		CleanupBatchSize:   2,
+		CleanupDelay:       15 * time.Minute,
+		CleanupJobMaxLevel: 1,
+		CleanupJobMinLevel: 0,
+	}
+}
+
+func (c *Config) RegisterFlagsWithPrefix(string, *flag.FlagSet) {
+	// NOTE(kolesnikovae): I'm not sure if making this configurable
+	// is a good idea; however, we might want to add a flag to tune
+	// the parameters based on e.g., segment size or max duration.
+	*c = DefaultConfig()
+}
+
+// exceedsSize is called after the block has been added to the batch.
+// If the function returns true, the batch is flushed to the global
+// queue and becomes available for compaction.
+func (c *Config) exceedsMaxSize(b *batch) bool {
+	return uint(b.size) >= c.maxBlocks(b.staged.key.level)
+}
+
+// exceedsAge reports whether the batch update time is older than the
+// maximum age for the level threshold. The function is used in two
+// cases: if the batch is not flushed to the global queue and is the
+// oldest one, or if the batch is flushed (and available to the planner)
+// but the job plan is not complete yet.
+func (c *Config) exceedsMaxAge(b *batch, now int64) bool {
+	if m := c.maxAge(b.staged.key.level); m > 0 && b.staged.updatedAt > 0 {
+		return now-b.staged.updatedAt > m
+	}
+	return false
+}
+
+func (c *Config) maxBlocks(l uint32) uint {
+	if l < uint32(len(c.Levels)) {
+		return c.Levels[l].MaxBlocks
+	}
+	return 0
+}
+
+func (c *Config) maxAge(l uint32) int64 {
+	if l < uint32(len(c.Levels)) {
+		return c.Levels[l].MaxAge
+	}
+	return 0
+}