Compactor interface

compact.go

@@ -13,7 +13,16 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 )
 
+type Compactor interface {
+	Plan() ([][]string, error)
+
+	Write(dir string, bs ...Block) error
+
+	Compact(dirs ...string) error
+}
+
 type compactor struct {
+	dir     string
 	metrics *compactorMetrics
 	opts    *compactorOptions
 }
@@ -54,8 +63,9 @@ type compactorOptions struct {
 	maxBlockRange uint64
 }
 
-func newCompactor(r prometheus.Registerer, opts *compactorOptions) *compactor {
+func newCompactor(dir string, r prometheus.Registerer, opts *compactorOptions) *compactor {
 	return &compactor{
+		dir:     dir,
 		opts:    opts,
 		metrics: newCompactorMetrics(r),
 	}
@@ -69,61 +79,55 @@ type compactionInfo struct {
 
 const compactionBlocksLen = 3
 
-// pick returns a range [i, j) in the blocks that are suitable to be compacted
+func (c *compactor) Plan() ([][]string, error) {
-// into a single block at position i.
+	dirs, err := blockDirs(c.dir)
-func (c *compactor) pick(bs []compactionInfo) (i, j int, ok bool) {
+	if err != nil {
+		return nil, err
+	}
+
+	var bs []*BlockMeta
+
+	for _, dir := range dirs {
+		meta, err := readMetaFile(dir)
+		if err != nil {
+			return nil, err
+		}
+		if meta.Compaction.Generation > 0 {
+			bs = append(bs, meta)
+		}
+	}
+
 	if len(bs) == 0 {
-		return 0, 0, false
+		return nil, nil
 	}
 
-	// First, we always compact pending in-memory blocks – oldest first.
+	sliceDirs := func(i, j int) [][]string {
-	for i, b := range bs {
+		var res []string
-		if b.generation > 0 {
+		for k := i; k < j; k++ {
-			continue
+			res = append(res, dirs[k])
 		}
-		// Directly compact into 2nd generation with previous generation 1 blocks.
+		return [][]string{res}
-		if i+1 >= compactionBlocksLen {
-			match := true
-			for _, pb := range bs[i-compactionBlocksLen+1 : i] {
-				match = match && pb.generation == 1
-			}
-			if match {
-				return i - compactionBlocksLen + 1, i + 1, true
-			}
-		}
-		// If we have enough generation 0 blocks to directly move to the
-		// 2nd generation, skip generation 1.
-		if len(bs)-i >= compactionBlocksLen {
-			// Guard against the newly compacted block becoming larger than
-			// the previous one.
-			if i == 0 || bs[i-1].generation >= 2 {
-				return i, i + compactionBlocksLen, true
-			}
-		}
-
-		// No optimizations possible, naiively compact the new block.
-		return i, i + 1, true
 	}
 
 	// Then we care about compacting multiple blocks, starting with the oldest.
 	for i := 0; i < len(bs)-compactionBlocksLen+1; i += compactionBlocksLen {
 		if c.match(bs[i : i+3]) {
-			return i, i + compactionBlocksLen, true
+			return sliceDirs(i, i+compactionBlocksLen), nil
 		}
 	}
 
-	return 0, 0, false
+	return nil, nil
 }
 
-func (c *compactor) match(bs []compactionInfo) bool {
+func (c *compactor) match(bs []*BlockMeta) bool {
-	g := bs[0].generation
+	g := bs[0].Compaction.Generation
 
 	for _, b := range bs {
-		if b.generation != g {
+		if b.Compaction.Generation != g {
 			return false
 		}
 	}
-	return uint64(bs[len(bs)-1].maxt-bs[0].mint) <= c.opts.maxBlockRange
+	return uint64(bs[len(bs)-1].MaxTime-bs[0].MinTime) <= c.opts.maxBlockRange
 }
 
 var entropy = rand.New(rand.NewSource(time.Now().UnixNano()))
@@ -136,11 +140,7 @@ func mergeBlockMetas(blocks ...Block) (res BlockMeta) {
 	res.MaxTime = blocks[len(blocks)-1].Meta().MaxTime
 	res.ULID = ulid.MustNew(ulid.Now(), entropy)
 
-	g := m0.Compaction.Generation
+	res.Compaction.Generation = m0.Compaction.Generation + 1
-	if g == 0 && len(blocks) > 1 {
-		g++
-	}
-	res.Compaction.Generation = g + 1
 
 	for _, b := range blocks {
 		res.Stats.NumSamples += b.Meta().Stats.NumSamples
@@ -148,14 +148,26 @@ func mergeBlockMetas(blocks ...Block) (res BlockMeta) {
 	return res
 }
 
-func (c *compactor) compact(dir string, blocks ...Block) (err error) {
+func (c *compactor) Compact(dirs ...string) (err error) {
-	start := time.Now()
+	var blocks []Block
-	defer func() {
+	for _, d := range dirs {
+		b, err := newPersistedBlock(d)
+		if err != nil {
+			return err
+		}
+		blocks = append(blocks, b)
+	}
+
+	return c.Write(dirs[0], blocks...)
+}
+
+func (c *compactor) Write(dir string, blocks ...Block) (err error) {
+	defer func(t time.Time) {
 		if err != nil {
 			c.metrics.failed.Inc()
 		}
-		c.metrics.duration.Observe(time.Since(start).Seconds())
+		c.metrics.duration.Observe(time.Since(t).Seconds())
-	}()
+	}(time.Now())
 
 	if err = os.RemoveAll(dir); err != nil {
 		return err

db.go

@@ -96,9 +96,10 @@ type DB struct {
 	mtx       sync.RWMutex
 	persisted []*persistedBlock
 	heads     []*headBlock
+	seqBlocks map[int]Block
 	headGen   uint8
 
-	compactor *compactor
+	compactor Compactor
 
 	compactc chan struct{}
 	donec    chan struct{}
@@ -171,14 +172,13 @@ func Open(dir string, l log.Logger, r prometheus.Registerer, opts *Options) (db
 		donec:    make(chan struct{}),
 		stopc:    make(chan struct{}),
 	}
-	db.compactor = newCompactor(r, &compactorOptions{
+	db.compactor = newCompactor(dir, r, &compactorOptions{
 		maxBlockRange: opts.MaxBlockDuration,
 	})
 
-	if err := db.initBlocks(); err != nil {
+	if err := db.reloadBlocks(); err != nil {
 		return nil, err
 	}
-
 	go db.run()
 
 	return db, nil
@@ -200,35 +200,8 @@ func (db *DB) run() {
 		case <-db.compactc:
 			db.metrics.compactionsTriggered.Inc()
 
-			var seqs []int
+			if err := db.compact(); err != nil {
-			var infos []compactionInfo
-			for _, b := range db.compactable() {
-				m := b.Meta()
-
-				infos = append(infos, compactionInfo{
-					generation: m.Compaction.Generation,
-					mint:       m.MinTime,
-					maxt:       m.MaxTime,
-					seq:        m.Sequence,
-				})
-				seqs = append(seqs, m.Sequence)
-			}
-
-			i, j, ok := db.compactor.pick(infos)
-			if !ok {
-				continue
-			}
-			db.logger.Log("msg", "compact", "seqs", fmt.Sprintf("%v", seqs[i:j]))
-
-			if err := db.compact(i, j); err != nil {
 				db.logger.Log("msg", "compaction failed", "err", err)
-				continue
-			}
-			db.logger.Log("msg", "compaction completed")
-			// Trigger another compaction in case there's more work to do.
-			select {
-			case db.compactc <- struct{}{}:
-			default:
 			}
 
 		case <-db.stopc:
@@ -237,150 +210,165 @@ func (db *DB) run() {
 	}
 }
 
-func (db *DB) getBlock(i int) Block {
+func (db *DB) compact() error {
-	if i < len(db.persisted) {
+	changes := false
-		return db.persisted[i]
+	// Check whether we have pending head blocks that are ready to be persisted.
-	}
+	// They have the highest priority.
-	return db.heads[i-len(db.persisted)]
+	db.mtx.RLock()
+
+	if len(db.heads) > db.opts.AppendableBlocks {
+		for _, h := range db.heads[:len(db.heads)-db.opts.AppendableBlocks] {
+			// Blocks that won't be appendable when instantiating a new appender
+			// might still have active appenders on them.
+			// Abort at the first one we encounter.
+			if atomic.LoadUint64(&h.activeWriters) > 0 {
+				break
 			}
 
-// removeBlocks removes the blocks in range [i, j) from the list of persisted
+			db.logger.Log("msg", "write head", "seq", h.Meta().Sequence)
-// and head blocks. The blocks are not closed and their files not deleted.
+
-func (db *DB) removeBlocks(i, j int) {
+			select {
-	for k := i; k < j; k++ {
+			case <-db.stopc:
-		if i < len(db.persisted) {
+				return nil
-			db.persisted = append(db.persisted[:i], db.persisted[i+1:]...)
+			default:
-		} else {
-			l := i - len(db.persisted)
-			db.heads = append(db.heads[:l], db.heads[l+1:]...)
 			}
+
+			if err := db.compactor.Write(h.Dir(), h); err != nil {
+				db.mtx.RUnlock()
+				return errors.Wrap(err, "persist head block")
+			}
+			changes = true
 		}
 	}
 
-func (db *DB) blocks() (bs []Block) {
+	db.mtx.RUnlock()
-	for _, b := range db.persisted {
-		bs = append(bs, b)
-	}
-	for _, b := range db.heads {
-		bs = append(bs, b)
-	}
-	return bs
-}
 
-// compact block in range [i, j) into a temporary directory and atomically
+	// Check for compactions of multiple blocks.
-// swap the blocks out on successful completion.
+	for {
-func (db *DB) compact(i, j int) error {
+		plans, err := db.compactor.Plan()
-	if j <= i {
-		return errors.New("invalid compaction block range")
-	}
-	var blocks []Block
-	for k := i; k < j; k++ {
-		blocks = append(blocks, db.getBlock(k))
-	}
-	var (
-		dir    = blocks[0].Dir()
-		tmpdir = dir + ".tmp"
-	)
-
-	if err := db.compactor.compact(tmpdir, blocks...); err != nil {
-		return err
-	}
-
-	pb, err := newPersistedBlock(tmpdir)
 		if err != nil {
-		return err
+			return errors.Wrap(err, "plan compaction")
-	}
-
-	db.mtx.Lock()
-	defer db.mtx.Unlock()
-
-	for _, b := range blocks {
-		if err := b.Close(); err != nil {
-			return errors.Wrapf(err, "close old block %s", b.Dir())
-		}
-	}
-
-	if err := renameFile(tmpdir, dir); err != nil {
-		return errors.Wrap(err, "rename dir")
-	}
-	pb.dir = dir
-
-	db.removeBlocks(i, j)
-	db.persisted = append(db.persisted, pb)
-
-	for _, b := range blocks[1:] {
-		db.logger.Log("msg", "remove old dir", "dir", b.Dir())
-		if err := os.RemoveAll(b.Dir()); err != nil {
-			return errors.Wrap(err, "removing old block")
-		}
-	}
-	if err := db.retentionCutoff(); err != nil {
-		return err
 		}
 
+		select {
+		case <-db.stopc:
 			return nil
+		default:
 		}
+		// We just execute compactions sequentially to not cause too extreme
+		// CPU and memory spikes.
+		// TODO(fabxc): return more descriptive plans in the future that allow
+		// estimation of resource usage and conditional parallelization?
+		for _, p := range plans {
+			db.logger.Log("msg", "compact blocks", "seq", fmt.Sprintf("%v", p))
 
-func (db *DB) retentionCutoff() error {
+			if err := db.compactor.Compact(p...); err != nil {
-	if db.opts.RetentionDuration == 0 {
+				return errors.Wrapf(err, "compact", p)
-		return nil
 			}
-	h := db.heads[len(db.heads)-1]
+			changes = true
-	t := h.meta.MinTime - int64(db.opts.RetentionDuration)
+		}
-
+		// If we didn't compact anything, there's nothing left to do.
-	var (
+		if len(plans) == 0 {
-		blocks = db.blocks()
-		i      int
-		b      Block
-	)
-	for i, b = range blocks {
-		if b.Meta().MinTime >= t {
 			break
 		}
 	}
-	if i <= 1 {
-		return nil
-	}
-	db.logger.Log("msg", "retention cutoff", "idx", i-1)
-	db.removeBlocks(0, i)
 
-	for _, b := range blocks[:i] {
+	if changes {
-		if err := os.RemoveAll(b.Dir()); err != nil {
+		return errors.Wrap(db.reloadBlocks(), "reload blocks")
-			return errors.Wrap(err, "removing old block")
-		}
 	}
 	return nil
 }
 
-func (db *DB) initBlocks() error {
+// func (db *DB) retentionCutoff() error {
-	var (
+// 	if db.opts.RetentionDuration == 0 {
-		persisted []*persistedBlock
+// 		return nil
-		heads     []*headBlock
+// 	}
-	)
+// 	h := db.heads[len(db.heads)-1]
+// 	t := h.meta.MinTime - int64(db.opts.RetentionDuration)
+
+// 	var (
+// 		blocks = db.blocks()
+// 		i      int
+// 		b      Block
+// 	)
+// 	for i, b = range blocks {
+// 		if b.Meta().MinTime >= t {
+// 			break
+// 		}
+// 	}
+// 	if i <= 1 {
+// 		return nil
+// 	}
+// 	db.logger.Log("msg", "retention cutoff", "idx", i-1)
+// 	db.removeBlocks(0, i)
+
+// 	for _, b := range blocks[:i] {
+// 		if err := os.RemoveAll(b.Dir()); err != nil {
+// 			return errors.Wrap(err, "removing old block")
+// 		}
+// 	}
+// 	return nil
+// }
+
+func (db *DB) reloadBlocks() error {
+	db.mtx.Lock()
+	defer db.mtx.Unlock()
 
 	dirs, err := blockDirs(db.dir)
 	if err != nil {
-		return err
+		return errors.Wrap(err, "find blocks")
 	}
+	var (
+		metas     []*BlockMeta
+		persisted []*persistedBlock
+		heads     []*headBlock
+		seqBlocks = make(map[int]Block, len(dirs))
+	)
 
 	for _, dir := range dirs {
-		if fileutil.Exist(filepath.Join(dir, walDirName)) {
+		meta, err := readMetaFile(dir)
-			h, err := openHeadBlock(dir, db.logger)
 		if err != nil {
-				return err
+			return errors.Wrapf(err, "read meta information %s", dir)
 		}
-			h.generation = db.headGen
+		metas = append(metas, meta)
-			db.headGen++
-			heads = append(heads, h)
-			continue
-		}
-		b, err := newPersistedBlock(dir)
-		if err != nil {
-			return err
-		}
-		persisted = append(persisted, b)
 	}
 
+	for i, meta := range metas {
+		b, ok := db.seqBlocks[meta.Sequence]
+		if !ok {
+			return errors.Errorf("missing block for sequence %d", meta.Sequence)
+		}
+
+		if meta.Compaction.Generation == 0 {
+			if meta.ULID != b.Meta().ULID {
+				return errors.Errorf("head block ULID changed unexpectedly")
+			}
+			heads = append(heads, b.(*headBlock))
+		} else {
+			if meta.ULID != b.Meta().ULID {
+				if err := b.Close(); err != nil {
+					return err
+				}
+				b, err = newPersistedBlock(dirs[i])
+				if err != nil {
+					return errors.Wrapf(err, "open persisted block %s", dirs[i])
+				}
+			}
+			persisted = append(persisted, b.(*persistedBlock))
+		}
+
+		seqBlocks[meta.Sequence] = b
+	}
+
+	for seq, b := range db.seqBlocks {
+		if _, ok := seqBlocks[seq]; !ok {
+			if err := b.Close(); err != nil {
+				return err
+			}
+		}
+	}
+
+	db.seqBlocks = seqBlocks
 	db.persisted = persisted
 	db.heads = heads
 
@@ -643,6 +631,7 @@ func (db *DB) cut(mint int64) (*headBlock, error) {
 	}
 
 	db.heads = append(db.heads, newHead)
+	db.seqBlocks[seq] = newHead
 	db.headGen++
 
 	newHead.generation = db.headGen

head.go

@@ -63,7 +63,10 @@ type headBlock struct {
 }
 
 func createHeadBlock(dir string, seq int, l log.Logger, mint, maxt int64) (*headBlock, error) {
-	if err := os.MkdirAll(dir, 0777); err != nil {
+	// Make head block creation appear atomic.
+	tmp := dir + ".tmp"
+
+	if err := os.MkdirAll(tmp, 0777); err != nil {
 		return nil, err
 	}
 	ulid, err := ulid.New(ulid.Now(), entropy)
@@ -71,7 +74,7 @@ func createHeadBlock(dir string, seq int, l log.Logger, mint, maxt int64) (*head
 		return nil, err
 	}
 
-	if err := writeMetaFile(dir, &BlockMeta{
+	if err := writeMetaFile(tmp, &BlockMeta{
 		ULID:     ulid,
 		Sequence: seq,
 		MinTime:  mint,
@@ -79,6 +82,9 @@ func createHeadBlock(dir string, seq int, l log.Logger, mint, maxt int64) (*head
 	}); err != nil {
 		return nil, err
 	}
+	if err := renameFile(tmp, dir); err != nil {
+		return nil, err
+	}
 	return openHeadBlock(dir, l)
 }