diff --git a/block.go b/block.go index f2fcf97ff..59a9d4dc1 100644 --- a/block.go +++ b/block.go @@ -142,10 +142,9 @@ type Block struct { dir string meta BlockMeta - chunkr ChunkReader - indexr IndexReader - - tombstones tombstoneReader + chunkr ChunkReader + indexr IndexReader + tombstones TombstoneReader } // OpenBlock opens the block in the directory. It can be passed a chunk pool, which is used @@ -293,7 +292,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 := map[uint64]Intervals{} + stones := memTombstones{} var lset labels.Labels var chks []ChunkMeta @@ -325,16 +324,21 @@ Outer: return p.Err() } - // Merge the current and new tombstones. - for k, v := range stones { - pb.tombstones.add(k, v[0]) + err = pb.tombstones.Iter(func(id uint64, ivs Intervals) error { + for _, iv := range ivs { + stones.add(id, iv) + pb.meta.Stats.NumTombstones++ + } + return nil + }) + if err != nil { + return err } + pb.tombstones = stones if err := writeTombstoneFile(pb.dir, pb.tombstones); err != nil { return err } - - pb.meta.Stats.NumTombstones = uint64(len(pb.tombstones)) return writeMetaFile(pb.dir, &pb.meta) } diff --git a/compact.go b/compact.go index 955ba3caf..48d77f72c 100644 --- a/compact.go +++ b/compact.go @@ -418,7 +418,7 @@ func (c *LeveledCompactor) write(dest string, meta *BlockMeta, blocks ...BlockRe } // Create an empty tombstones file. - if err := writeTombstoneFile(tmp, newEmptyTombstoneReader()); err != nil { + if err := writeTombstoneFile(tmp, EmptyTombstoneReader()); err != nil { return errors.Wrap(err, "write new tombstones file") } @@ -631,7 +631,11 @@ func (c *compactionSeriesSet) Next() bool { } var err error - c.intervals = c.tombstones.Get(c.p.At()) + c.intervals, err = c.tombstones.Get(c.p.At()) + if err != nil { + c.err = errors.Wrap(err, "get tombstones") + return false + } if err = c.index.Series(c.p.At(), &c.l, &c.c); err != nil { c.err = errors.Wrapf(err, "get series %d", c.p.At()) diff --git a/head.go b/head.go index 51a7896a2..d8614add7 100644 --- a/head.go +++ b/head.go @@ -66,7 +66,7 @@ type Head struct { postings *memPostings // postings lists for terms - tombstones tombstoneReader + tombstones memTombstones } type headMetrics struct { @@ -186,7 +186,7 @@ func NewHead(r prometheus.Registerer, l log.Logger, wal WAL, chunkRange int64) ( values: map[string]stringset{}, symbols: map[string]struct{}{}, postings: newUnorderedMemPostings(), - tombstones: newEmptyTombstoneReader(), + tombstones: memTombstones{}, } h.metrics = newHeadMetrics(h, r) diff --git a/head_test.go b/head_test.go index 277519ad4..c42c19b5e 100644 --- a/head_test.go +++ b/head_test.go @@ -318,7 +318,7 @@ func TestHeadDeleteSimple(t *testing.T) { Outer: for _, c := range cases { // Reset the tombstones. - head.tombstones = newEmptyTombstoneReader() + head.tombstones = memTombstones{} // Delete the ranges. for _, r := range c.intervals { diff --git a/querier.go b/querier.go index 89a954795..faa29ecda 100644 --- a/querier.go +++ b/querier.go @@ -465,6 +465,7 @@ func (s *baseChunkSeries) Next() bool { var ( lset labels.Labels chunks []ChunkMeta + err error ) Outer: for s.p.Next() { @@ -487,7 +488,11 @@ Outer: s.lset = lset s.chks = chunks - s.intervals = s.tombstones.Get(s.p.At()) + s.intervals, err = s.tombstones.Get(s.p.At()) + if err != nil { + s.err = errors.Wrap(err, "get tombstones") + return false + } if len(s.intervals) > 0 { // Only those chunks that are not entirely deleted. diff --git a/querier_test.go b/querier_test.go index e302e2983..301a2b046 100644 --- a/querier_test.go +++ b/querier_test.go @@ -454,7 +454,7 @@ Outer: querier := &blockQuerier{ index: ir, chunks: cr, - tombstones: newEmptyTombstoneReader(), + tombstones: EmptyTombstoneReader(), mint: c.mint, maxt: c.maxt, @@ -506,7 +506,7 @@ func TestBlockQuerierDelete(t *testing.T) { chunks [][]sample } - tombstones tombstoneReader + tombstones TombstoneReader queries []query }{ data: []struct { @@ -554,13 +554,11 @@ func TestBlockQuerierDelete(t *testing.T) { }, }, }, - tombstones: newTombstoneReader( - map[uint64]Intervals{ - 1: Intervals{{1, 3}}, - 2: Intervals{{1, 3}, {6, 10}}, - 3: Intervals{{6, 10}}, - }, - ), + tombstones: memTombstones{ + 1: Intervals{{1, 3}}, + 2: Intervals{{1, 3}, {6, 10}}, + 3: Intervals{{6, 10}}, + }, queries: []query{ { @@ -736,7 +734,7 @@ func TestBaseChunkSeries(t *testing.T) { bcs := &baseChunkSeries{ p: newListPostings(tc.postings), index: mi, - tombstones: newEmptyTombstoneReader(), + tombstones: EmptyTombstoneReader(), } i := 0 diff --git a/tombstones.go b/tombstones.go index d43cd0bd0..8ca089e61 100644 --- a/tombstones.go +++ b/tombstones.go @@ -35,12 +35,17 @@ const ( // TombstoneReader gives access to tombstone intervals by series reference. type TombstoneReader interface { - Get(ref uint64) Intervals + // Get returns deletion intervals for the series with the given reference. + Get(ref uint64) (Intervals, error) + // Iter calls the given function for each encountered interval. + Iter(func(uint64, Intervals) error) error + + // Close any underlying resources Close() error } -func writeTombstoneFile(dir string, tr tombstoneReader) error { +func writeTombstoneFile(dir string, tr TombstoneReader) error { path := filepath.Join(dir, tombstoneFilename) tmp := path + ".tmp" hash := newCRC32() @@ -67,19 +72,21 @@ func writeTombstoneFile(dir string, tr tombstoneReader) error { mw := io.MultiWriter(f, hash) - for k, v := range tr { - for _, itv := range v { + tr.Iter(func(ref uint64, ivs Intervals) error { + for _, iv := range ivs { buf.reset() - buf.putUvarint64(k) - buf.putVarint64(itv.Mint) - buf.putVarint64(itv.Maxt) + + buf.putUvarint64(ref) + buf.putVarint64(iv.Mint) + buf.putVarint64(iv.Maxt) _, err = mw.Write(buf.get()) if err != nil { return err } } - } + return nil + }) _, err = f.Write(hash.Sum(nil)) if err != nil { @@ -100,7 +107,7 @@ type Stone struct { intervals Intervals } -func readTombstones(dir string) (tombstoneReader, error) { +func readTombstones(dir string) (memTombstones, error) { b, err := ioutil.ReadFile(filepath.Join(dir, tombstoneFilename)) if err != nil { return nil, err @@ -131,7 +138,8 @@ func readTombstones(dir string) (tombstoneReader, error) { return nil, errors.New("checksum did not match") } - stonesMap := newEmptyTombstoneReader() + stonesMap := memTombstones{} + for d.len() > 0 { k := d.uvarint64() mint := d.varint64() @@ -143,28 +151,36 @@ func readTombstones(dir string) (tombstoneReader, error) { stonesMap.add(k, Interval{mint, maxt}) } - return newTombstoneReader(stonesMap), nil + return stonesMap, nil } -type tombstoneReader map[uint64]Intervals +type memTombstones map[uint64]Intervals -func newTombstoneReader(ts map[uint64]Intervals) tombstoneReader { - return tombstoneReader(ts) +var emptyTombstoneReader = memTombstones{} + +// EmptyTombstoneReader returns a TombstoneReader that is always empty. +func EmptyTombstoneReader() TombstoneReader { + return emptyTombstoneReader } -func newEmptyTombstoneReader() tombstoneReader { - return tombstoneReader(make(map[uint64]Intervals)) +func (t memTombstones) Get(ref uint64) (Intervals, error) { + return t[ref], nil } -func (t tombstoneReader) Get(ref uint64) Intervals { - return t[ref] +func (t memTombstones) Iter(f func(uint64, Intervals) error) error { + for ref, ivs := range t { + if err := f(ref, ivs); err != nil { + return err + } + } + return nil } -func (t tombstoneReader) add(ref uint64, itv Interval) { +func (t memTombstones) add(ref uint64, itv Interval) { t[ref] = t[ref].add(itv) } -func (tombstoneReader) Close() error { +func (memTombstones) Close() error { return nil } diff --git a/tombstones_test.go b/tombstones_test.go index 9265b76b3..eb124fc94 100644 --- a/tombstones_test.go +++ b/tombstones_test.go @@ -29,7 +29,7 @@ func TestWriteAndReadbackTombStones(t *testing.T) { ref := uint64(0) - stones := make(map[uint64]Intervals) + stones := memTombstones{} // Generate the tombstones. for i := 0; i < 100; i++ { ref += uint64(rand.Int31n(10)) + 1 @@ -43,13 +43,13 @@ func TestWriteAndReadbackTombStones(t *testing.T) { stones[ref] = dranges } - require.NoError(t, writeTombstoneFile(tmpdir, newTombstoneReader(stones))) + require.NoError(t, writeTombstoneFile(tmpdir, stones)) restr, err := readTombstones(tmpdir) require.NoError(t, err) - exptr := newTombstoneReader(stones) + // Compare the two readers. - require.Equal(t, exptr, restr) + require.Equal(t, stones, restr) } func TestAddingNewIntervals(t *testing.T) {