Move tombstones to it's own package.

Signed-off-by: Callum Styan <[email protected]>

Author: cstyan

block.go

@@ -32,7 +32,7 @@ import (
 	"github.com/prometheus/tsdb/fileutil"
 	"github.com/prometheus/tsdb/index"
 	"github.com/prometheus/tsdb/labels"
-	"github.com/prometheus/tsdb/record"
+	"github.com/prometheus/tsdb/tombstones"
 )
 
 // IndexWriter serializes the index for a block of series data.
@@ -137,7 +137,7 @@ type BlockReader interface {
 	Chunks() (ChunkReader, error)
 
 	// Tombstones returns a TombstoneReader over the block's deleted data.
-	Tombstones() (record.TombstoneReader, error)
+	Tombstones() (tombstones.TombstoneReader, error)
 
 	// MinTime returns the min time of the block.
 	MinTime() int64
@@ -282,7 +282,7 @@ type Block struct {
 
 	chunkr     ChunkReader
 	indexr     IndexReader
-	tombstones record.TombstoneReader
+	tombstones tombstones.TombstoneReader
 
 	logger log.Logger
 
@@ -324,7 +324,7 @@ func OpenBlock(logger log.Logger, dir string, pool chunkenc.Pool) (pb *Block, er
 	}
 	closers = append(closers, ir)
 
-	tr, sizeTomb, err := record.ReadTombstones(dir)
+	tr, sizeTomb, err := tombstones.ReadTombstones(dir)
 	if err != nil {
 		return nil, err
 	}
@@ -415,7 +415,7 @@ func (pb *Block) Chunks() (ChunkReader, error) {
 }
 
 // Tombstones returns a new TombstoneReader against the block data.
-func (pb *Block) Tombstones() (record.TombstoneReader, error) {
+func (pb *Block) Tombstones() (tombstones.TombstoneReader, error) {
 	if err := pb.startRead(); err != nil {
 		return nil, err
 	}
@@ -484,7 +484,7 @@ func (r blockIndexReader) Close() error {
 }
 
 type blockTombstoneReader struct {
-	record.TombstoneReader
+	tombstones.TombstoneReader
 	b *Block
 }
 
@@ -520,7 +520,7 @@ func (pb *Block) Delete(mint, maxt int64, ms ...labels.Matcher) error {
 	ir := pb.indexr
 
 	// Choose only valid postings which have chunks in the time-range.
-	stones := record.NewMemTombstones()
+	stones := tombstones.NewMemTombstones()
 
 	var lset labels.Labels
 	var chks []chunks.Meta
@@ -536,7 +536,7 @@ Outer:
 			if chk.OverlapsClosedInterval(mint, maxt) {
 				// Delete only until the current values and not beyond.
 				tmin, tmax := clampInterval(mint, maxt, chks[0].MinTime, chks[len(chks)-1].MaxTime)
-				stones.AddInterval(p.At(), record.Interval{tmin, tmax})
+				stones.AddInterval(p.At(), tombstones.Interval{tmin, tmax})
 				continue Outer
 			}
 		}
@@ -546,7 +546,7 @@ Outer:
 		return p.Err()
 	}
 
-	err = pb.tombstones.Iter(func(id uint64, ivs record.Intervals) error {
+	err = pb.tombstones.Iter(func(id uint64, ivs tombstones.Intervals) error {
 		for _, iv := range ivs {
 			stones.AddInterval(id, iv)
 		}
@@ -558,7 +558,7 @@ Outer:
 	pb.tombstones = stones
 	pb.meta.Stats.NumTombstones = pb.tombstones.Total()
 
-	n, err := record.WriteTombstoneFile(pb.logger, pb.dir, pb.tombstones)
+	n, err := tombstones.WriteTombstoneFile(pb.logger, pb.dir, pb.tombstones)
 	if err != nil {
 		return err
 	}
@@ -576,7 +576,7 @@ Outer:
 func (pb *Block) CleanTombstones(dest string, c Compactor) (*ulid.ULID, error) {
 	numStones := 0
 
-	if err := pb.tombstones.Iter(func(id uint64, ivs record.Intervals) error {
+	if err := pb.tombstones.Iter(func(id uint64, ivs tombstones.Intervals) error {
 		numStones += len(ivs)
 		return nil
 	}); err != nil {
@@ -611,7 +611,7 @@ func (pb *Block) Snapshot(dir string) error {
 	for _, fname := range []string{
 		metaFilename,
 		indexFilename,
-		record.TombstoneFilename,
+		tombstones.TombstoneFilename,
 	} {
 		if err := os.Link(filepath.Join(pb.dir, fname), filepath.Join(blockDir, fname)); err != nil {
 			return errors.Wrapf(err, "create snapshot %s", fname)

compact.go

@@ -35,7 +35,7 @@ import (
 	"github.com/prometheus/tsdb/fileutil"
 	"github.com/prometheus/tsdb/index"
 	"github.com/prometheus/tsdb/labels"
-	"github.com/prometheus/tsdb/record"
+	"github.com/prometheus/tsdb/tombstones"
 )
 
 // ExponentialBlockRanges returns the time ranges based on the stepSize.
@@ -608,7 +608,7 @@ func (c *LeveledCompactor) write(dest string, meta *BlockMeta, blocks ...BlockRe
 	}
 
 	// Create an empty tombstones file.
-	if _, err := record.WriteTombstoneFile(c.logger, tmp, record.NewMemTombstones()); err != nil {
+	if _, err := tombstones.WriteTombstoneFile(c.logger, tmp, record.NewMemTombstones()); err != nil {
 		return errors.Wrap(err, "write new tombstones file")
 	}
 
@@ -851,15 +851,15 @@ type compactionSeriesSet struct {
 	p          index.Postings
 	index      IndexReader
 	chunks     ChunkReader
-	tombstones record.TombstoneReader
+	tombstones tombstones.TombstoneReader
 
 	l         labels.Labels
 	c         []chunks.Meta
-	intervals record.Intervals
+	intervals tombstones.Intervals
 	err       error
 }
 
-func newCompactionSeriesSet(i IndexReader, c ChunkReader, t record.TombstoneReader, p index.Postings) *compactionSeriesSet {
+func newCompactionSeriesSet(i IndexReader, c ChunkReader, t tombstones.TombstoneReader, p index.Postings) *compactionSeriesSet {
 	return &compactionSeriesSet{
 		index:      i,
 		chunks:     c,
@@ -889,7 +889,7 @@ func (c *compactionSeriesSet) Next() bool {
 	if len(c.intervals) > 0 {
 		chks := make([]chunks.Meta, 0, len(c.c))
 		for _, chk := range c.c {
-			if !(record.Interval{chk.MinTime, chk.MaxTime}.IsSubrange(c.intervals)) {
+			if !(tombstones.Interval{chk.MinTime, chk.MaxTime}.IsSubrange(c.intervals)) {
 				chks = append(chks, chk)
 			}
 		}
@@ -917,7 +917,7 @@ func (c *compactionSeriesSet) Err() error {
 	return c.p.Err()
 }
 
-func (c *compactionSeriesSet) At() (labels.Labels, []chunks.Meta, record.Intervals) {
+func (c *compactionSeriesSet) At() (labels.Labels, []chunks.Meta, tombstones.Intervals) {
 	return c.l, c.c, c.intervals
 }
 
@@ -927,7 +927,7 @@ type compactionMerger struct {
 	aok, bok  bool
 	l         labels.Labels
 	c         []chunks.Meta
-	intervals record.Intervals
+	intervals tombstones.Intervals
 }
 
 func newCompactionMerger(a, b ChunkSeriesSet) (*compactionMerger, error) {
@@ -1004,6 +1004,6 @@ func (c *compactionMerger) Err() error {
 	return c.b.Err()
 }
 
-func (c *compactionMerger) At() (labels.Labels, []chunks.Meta, record.Intervals) {
+func (c *compactionMerger) At() (labels.Labels, []chunks.Meta, tombstones.Intervals) {
 	return c.l, c.c, c.intervals
 }

compact_test.go

@@ -30,8 +30,8 @@ import (
 	"github.com/prometheus/tsdb/chunks"
 	"github.com/prometheus/tsdb/fileutil"
 	"github.com/prometheus/tsdb/labels"
-	"github.com/prometheus/tsdb/record"
 	"github.com/prometheus/tsdb/testutil"
+	"github.com/prometheus/tsdb/tombstones"
 )
 
 func TestSplitByRange(t *testing.T) {
@@ -458,7 +458,7 @@ type erringBReader struct{}
 
 func (erringBReader) Index() (IndexReader, error)  { return nil, errors.New("index") }
 func (erringBReader) Chunks() (ChunkReader, error) { return nil, errors.New("chunks") }
-func (erringBReader) Tombstones() (record.TombstoneReader, error) {
+func (erringBReader) Tombstones() (tombstones.TombstoneReader, error) {
 	return nil, errors.New("tombstones")
 }
 func (erringBReader) MinTime() int64 { return 0 }

db_test.go

@@ -35,6 +35,7 @@ import (
 	"github.com/prometheus/tsdb/labels"
 	"github.com/prometheus/tsdb/record"
 	"github.com/prometheus/tsdb/testutil"
+	"github.com/prometheus/tsdb/tombstones"
 	"github.com/prometheus/tsdb/tsdbutil"
 	"github.com/prometheus/tsdb/wal"
 )
@@ -244,27 +245,27 @@ func TestDeleteSimple(t *testing.T) {
 	numSamples := int64(10)
 
 	cases := []struct {
-		intervals record.Intervals
+		intervals tombstones.Intervals
 		remaint   []int64
 	}{
 		{
-			intervals: record.Intervals{{0, 3}},
+			intervals: tombstones.Intervals{{0, 3}},
 			remaint:   []int64{4, 5, 6, 7, 8, 9},
 		},
 		{
-			intervals: record.Intervals{{1, 3}},
+			intervals: tombstones.Intervals{{1, 3}},
 			remaint:   []int64{0, 4, 5, 6, 7, 8, 9},
 		},
 		{
-			intervals: record.Intervals{{1, 3}, {4, 7}},
+			intervals: tombstones.Intervals{{1, 3}, {4, 7}},
 			remaint:   []int64{0, 8, 9},
 		},
 		{
-			intervals: record.Intervals{{1, 3}, {4, 700}},
+			intervals: tombstones.Intervals{{1, 3}, {4, 700}},
 			remaint:   []int64{0},
 		},
 		{ // This case is to ensure that labels and symbols are deleted.
-			intervals: record.Intervals{{0, 9}},
+			intervals: tombstones.Intervals{{0, 9}},
 			remaint:   []int64{},
 		},
 	}
@@ -506,11 +507,11 @@ func TestDB_SnapshotWithDelete(t *testing.T) {
 
 	testutil.Ok(t, app.Commit())
 	cases := []struct {
-		intervals record.Intervals
+		intervals tombstones.Intervals
 		remaint   []int64
 	}{
 		{
-			intervals: record.Intervals{{1, 3}, {4, 7}},
+			intervals: tombstones.Intervals{{1, 3}, {4, 7}},
 			remaint:   []int64{0, 8, 9},
 		},
 	}
@@ -833,11 +834,11 @@ func TestTombstoneClean(t *testing.T) {
 
 	testutil.Ok(t, app.Commit())
 	cases := []struct {
-		intervals record.Intervals
+		intervals tombstones.Intervals
 		remaint   []int64
 	}{
 		{
-			intervals: record.Intervals{{1, 3}, {4, 7}},
+			intervals: tombstones.Intervals{{1, 3}, {4, 7}},
 			remaint:   []int64{0, 8, 9},
 		},
 	}
@@ -909,7 +910,7 @@ func TestTombstoneClean(t *testing.T) {
 		}
 
 		for _, b := range db.Blocks() {
-			testutil.Equals(t, record.NewMemTombstones(), b.tombstones)
+			testutil.Equals(t, tombstones.NewMemTombstones(), b.tombstones)
 		}
 	}
 }
@@ -935,8 +936,8 @@ func TestTombstoneCleanFail(t *testing.T) {
 		block, err := OpenBlock(nil, blockDir, nil)
 		testutil.Ok(t, err)
 		// Add some some fake tombstones to trigger the compaction.
-		tomb := record.NewMemTombstones()
-		tomb.AddInterval(0, record.Interval{0, 1})
+		tomb := tombstones.NewMemTombstones()
+		tomb.AddInterval(0, tombstones.Interval{0, 1})
 		block.tombstones = tomb
 
 		db.blocks = append(db.blocks, block)

head.go

@@ -33,6 +33,7 @@ import (
 	"github.com/prometheus/tsdb/index"
 	"github.com/prometheus/tsdb/labels"
 	"github.com/prometheus/tsdb/record"
+	"github.com/prometheus/tsdb/tombstones"
 	"github.com/prometheus/tsdb/wal"
 )
 
@@ -43,7 +44,7 @@ var (
 
 	// emptyTombstoneReader is a no-op Tombstone Reader.
 	// This is used by head to satisfy the Tombstones() function call.
-	emptyTombstoneReader = record.NewMemTombstones()
+	emptyTombstoneReader = tombstones.NewMemTombstones()
 )
 
 // Head handles reads and writes of time series data within a time window.
@@ -346,11 +347,20 @@ func (h *Head) loadWAL(r *wal.Reader, multiRef map[uint64]uint64) (err error) {
 	}
 
 	var (
+<<<<<<< HEAD
 		dec       RecordDecoder
 		series    []RefSeries
 		samples   []RefSample
 		tstones   []Stone
 		allStones = newMemTombstones()
+=======
+		dec       record.RecordDecoder
+		series    []record.RefSeries
+		samples   []record.RefSample
+		tstones   []tombstones.Stone
+		allStones = tombstones.NewMemTombstones()
+		err       error
+>>>>>>> Move tombstones to it's own package.
 	)
 	defer func() {
 		if err := allStones.Close(); err != nil {
@@ -377,7 +387,7 @@ func (h *Head) loadWAL(r *wal.Reader, multiRef map[uint64]uint64) (err error) {
 				if !created {
 					// There's already a different ref for this series.
 					multiRefLock.Lock()
-					multiRef[s.Ref] = series.ref
+					multiRef[s.Ref] = series.Ref
 					multiRefLock.Unlock()
 				}
 
@@ -464,11 +474,15 @@ func (h *Head) loadWAL(r *wal.Reader, multiRef map[uint64]uint64) (err error) {
 	}
 	wg.Wait()
 
+<<<<<<< HEAD
 	if r.Err() != nil {
 		return errors.Wrap(r.Err(), "read records")
 	}
 
 	if err := allStones.Iter(func(ref uint64, dranges Intervals) error {
+=======
+	if err := allStones.Iter(func(ref uint64, dranges tombstones.Intervals) error {
+>>>>>>> Move tombstones to it's own package.
 		return h.chunkRewrite(ref, dranges)
 	}); err != nil {
 		return errors.Wrap(r.Err(), "deleting samples from tombstones")
@@ -676,7 +690,7 @@ func (h *rangeHead) Chunks() (ChunkReader, error) {
 	return h.head.chunksRange(h.mint, h.maxt), nil
 }
 
-func (h *rangeHead) Tombstones() (record.TombstoneReader, error) {
+func (h *rangeHead) Tombstones() (tombstones.TombstoneReader, error) {
 	return emptyTombstoneReader, nil
 }
 
@@ -932,7 +946,7 @@ func (h *Head) Delete(mint, maxt int64, ms ...labels.Matcher) error {
 		return errors.Wrap(err, "select series")
 	}
 
-	var stones []record.Stone
+	var stones []tombstones.Stone
 	dirty := false
 	for p.Next() {
 		series := h.series.getByID(p.At())
@@ -944,9 +958,9 @@ func (h *Head) Delete(mint, maxt int64, ms ...labels.Matcher) error {
 		// Delete only until the current values and not beyond.
 		t0, t1 = clampInterval(mint, maxt, t0, t1)
 		if h.wal != nil {
-			stones = append(stones, record.Stone{p.At(), record.Intervals{{t0, t1}}})
+			stones = append(stones, tombstones.Stone{p.At(), tombstones.Intervals{{t0, t1}}})
 		}
-		if err := h.chunkRewrite(p.At(), record.Intervals{{t0, t1}}); err != nil {
+		if err := h.chunkRewrite(p.At(), tombstones.Intervals{{t0, t1}}); err != nil {
 			return errors.Wrap(err, "delete samples")
 		}
 		dirty = true
@@ -973,7 +987,7 @@ func (h *Head) Delete(mint, maxt int64, ms ...labels.Matcher) error {
 // chunkRewrite re-writes the chunks which overlaps with deleted ranges
 // and removes the samples in the deleted ranges.
 // Chunks is deleted if no samples are left at the end.
-func (h *Head) chunkRewrite(ref uint64, dranges record.Intervals) (err error) {
+func (h *Head) chunkRewrite(ref uint64, dranges tombstones.Intervals) (err error) {
 	if len(dranges) == 0 {
 		return nil
 	}
@@ -1063,7 +1077,7 @@ func (h *Head) gc() {
 }
 
 // Tombstones returns a new reader over the head's tombstones
-func (h *Head) Tombstones() (record.TombstoneReader, error) {
+func (h *Head) Tombstones() (tombstones.TombstoneReader, error) {
 	return emptyTombstoneReader, nil
 }