Skip to content
This repository was archived by the owner on Aug 13, 2019. It is now read-only.

Commit e679638

Browse files
cstyancstyan
committed
Move WAL Watcher from Prometheus to TSDB WAL package.
Signed-off-by: Callum Styan <[email protected]>
1 parent 882162d commit e679638

22 files changed

+1976
-794
lines changed

‎block.go‎

Lines changed: 13 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@ import (
3232
"github.com/prometheus/tsdb/fileutil"
3333
"github.com/prometheus/tsdb/index"
3434
"github.com/prometheus/tsdb/labels"
35+
"github.com/prometheus/tsdb/record"
3536
)
3637

3738
// IndexWriter serializes the index for a block of series data.
@@ -136,7 +137,7 @@ type BlockReader interface {
136137
Chunks() (ChunkReader, error)
137138

138139
// Tombstones returns a TombstoneReader over the block's deleted data.
139-
Tombstones() (TombstoneReader, error)
140+
Tombstones() (record.TombstoneReader, error)
140141

141142
// MinTime returns the min time of the block.
142143
MinTime() int64
@@ -282,7 +283,7 @@ type Block struct {
282283

283284
chunkr ChunkReader
284285
indexr IndexReader
285-
tombstones TombstoneReader
286+
tombstones record.TombstoneReader
286287

287288
logger log.Logger
288289
}
@@ -319,7 +320,7 @@ func OpenBlock(logger log.Logger, dir string, pool chunkenc.Pool) (pb *Block, er
319320
}
320321
closers = append(closers, ir)
321322

322-
tr, tsr, err := readTombstones(dir)
323+
tr, tsr, err := record.ReadTombstones(dir)
323324
if err != nil {
324325
return nil, err
325326
}
@@ -423,7 +424,7 @@ func (pb *Block) Chunks() (ChunkReader, error) {
423424
}
424425

425426
// Tombstones returns a new TombstoneReader against the block data.
426-
func (pb *Block) Tombstones() (TombstoneReader, error) {
427+
func (pb *Block) Tombstones() (record.TombstoneReader, error) {
427428
if err := pb.startRead(); err != nil {
428429
return nil, err
429430
}
@@ -487,7 +488,7 @@ func (r blockIndexReader) Close() error {
487488
}
488489

489490
type blockTombstoneReader struct {
490-
TombstoneReader
491+
record.TombstoneReader
491492
b *Block
492493
}
493494

@@ -523,7 +524,7 @@ func (pb *Block) Delete(mint, maxt int64, ms ...labels.Matcher) error {
523524
ir := pb.indexr
524525

525526
// Choose only valid postings which have chunks in the time-range.
526-
stones := newMemTombstones()
527+
stones := record.NewMemTombstones()
527528

528529
var lset labels.Labels
529530
var chks []chunks.Meta
@@ -539,7 +540,7 @@ Outer:
539540
if chk.OverlapsClosedInterval(mint, maxt) {
540541
// Delete only until the current values and not beyond.
541542
tmin, tmax := clampInterval(mint, maxt, chks[0].MinTime, chks[len(chks)-1].MaxTime)
542-
stones.addInterval(p.At(), Interval{tmin, tmax})
543+
stones.AddInterval(p.At(), record.Interval{tmin, tmax})
543544
continue Outer
544545
}
545546
}
@@ -549,9 +550,9 @@ Outer:
549550
return p.Err()
550551
}
551552

552-
err = pb.tombstones.Iter(func(id uint64, ivs Intervals) error {
553+
err = pb.tombstones.Iter(func(id uint64, ivs record.Intervals) error {
553554
for _, iv := range ivs {
554-
stones.addInterval(id, iv)
555+
stones.AddInterval(id, iv)
555556
}
556557
return nil
557558
})
@@ -561,7 +562,7 @@ Outer:
561562
pb.tombstones = stones
562563
pb.meta.Stats.NumTombstones = pb.tombstones.Total()
563564

564-
if err := writeTombstoneFile(pb.logger, pb.dir, pb.tombstones); err != nil {
565+
if err := record.WriteTombstoneFile(pb.logger, pb.dir, pb.tombstones); err != nil {
565566
return err
566567
}
567568
return writeMetaFile(pb.logger, pb.dir, &pb.meta)
@@ -572,7 +573,7 @@ Outer:
572573
func (pb *Block) CleanTombstones(dest string, c Compactor) (*ulid.ULID, error) {
573574
numStones := 0
574575

575-
if err := pb.tombstones.Iter(func(id uint64, ivs Intervals) error {
576+
if err := pb.tombstones.Iter(func(id uint64, ivs record.Intervals) error {
576577
numStones += len(ivs)
577578
return nil
578579
}); err != nil {
@@ -607,7 +608,7 @@ func (pb *Block) Snapshot(dir string) error {
607608
for _, fname := range []string{
608609
metaFilename,
609610
indexFilename,
610-
tombstoneFilename,
611+
record.TombstoneFilename,
611612
} {
612613
if err := os.Link(filepath.Join(pb.dir, fname), filepath.Join(blockDir, fname)); err != nil {
613614
return errors.Wrapf(err, "create snapshot %s", fname)

‎cmd/tsdb/main.go‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@ import (
3636
"github.com/prometheus/tsdb"
3737
"github.com/prometheus/tsdb/chunks"
3838
"github.com/prometheus/tsdb/labels"
39+
"github.com/prometheus/tsdb/record"
3940
"gopkg.in/alecthomas/kingpin.v2"
4041
)
4142

@@ -265,7 +266,7 @@ func (b *writeBenchmark) ingestScrapesShard(lbls []labels.Labels, scrapeCount in
265266
s.ref = &ref
266267
} else if err := app.AddFast(*s.ref, ts, float64(s.value)); err != nil {
267268

268-
if errors.Cause(err) != tsdb.ErrNotFound {
269+
if errors.Cause(err) != record.ErrNotFound {
269270
panic(err)
270271
}
271272

‎compact.go‎

Lines changed: 10 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,7 @@ import (
3535
"github.com/prometheus/tsdb/fileutil"
3636
"github.com/prometheus/tsdb/index"
3737
"github.com/prometheus/tsdb/labels"
38+
"github.com/prometheus/tsdb/record"
3839
)
3940

4041
// ExponentialBlockRanges returns the time ranges based on the stepSize.
@@ -605,7 +606,7 @@ func (c *LeveledCompactor) write(dest string, meta *BlockMeta, blocks ...BlockRe
605606
}
606607

607608
// Create an empty tombstones file.
608-
if err := writeTombstoneFile(c.logger, tmp, newMemTombstones()); err != nil {
609+
if err := record.WriteTombstoneFile(c.logger, tmp, record.NewMemTombstones()); err != nil {
609610
return errors.Wrap(err, "write new tombstones file")
610611
}
611612

@@ -848,15 +849,15 @@ type compactionSeriesSet struct {
848849
p index.Postings
849850
index IndexReader
850851
chunks ChunkReader
851-
tombstones TombstoneReader
852+
tombstones record.TombstoneReader
852853

853854
l labels.Labels
854855
c []chunks.Meta
855-
intervals Intervals
856+
intervals record.Intervals
856857
err error
857858
}
858859

859-
func newCompactionSeriesSet(i IndexReader, c ChunkReader, t TombstoneReader, p index.Postings) *compactionSeriesSet {
860+
func newCompactionSeriesSet(i IndexReader, c ChunkReader, t record.TombstoneReader, p index.Postings) *compactionSeriesSet {
860861
return &compactionSeriesSet{
861862
index: i,
862863
chunks: c,
@@ -886,7 +887,7 @@ func (c *compactionSeriesSet) Next() bool {
886887
if len(c.intervals) > 0 {
887888
chks := make([]chunks.Meta, 0, len(c.c))
888889
for _, chk := range c.c {
889-
if !(Interval{chk.MinTime, chk.MaxTime}.isSubrange(c.intervals)) {
890+
if !(record.Interval{chk.MinTime, chk.MaxTime}.IsSubrange(c.intervals)) {
890891
chks = append(chks, chk)
891892
}
892893
}
@@ -914,7 +915,7 @@ func (c *compactionSeriesSet) Err() error {
914915
return c.p.Err()
915916
}
916917

917-
func (c *compactionSeriesSet) At() (labels.Labels, []chunks.Meta, Intervals) {
918+
func (c *compactionSeriesSet) At() (labels.Labels, []chunks.Meta, record.Intervals) {
918919
return c.l, c.c, c.intervals
919920
}
920921

@@ -924,7 +925,7 @@ type compactionMerger struct {
924925
aok, bok bool
925926
l labels.Labels
926927
c []chunks.Meta
927-
intervals Intervals
928+
intervals record.Intervals
928929
}
929930

930931
func newCompactionMerger(a, b ChunkSeriesSet) (*compactionMerger, error) {
@@ -980,7 +981,7 @@ func (c *compactionMerger) Next() bool {
980981
_, cb, rb := c.b.At()
981982

982983
for _, r := range rb {
983-
ra = ra.add(r)
984+
ra = ra.Add(r)
984985
}
985986

986987
c.l = append(c.l[:0], l...)
@@ -1001,6 +1002,6 @@ func (c *compactionMerger) Err() error {
10011002
return c.b.Err()
10021003
}
10031004

1004-
func (c *compactionMerger) At() (labels.Labels, []chunks.Meta, Intervals) {
1005+
func (c *compactionMerger) At() (labels.Labels, []chunks.Meta, record.Intervals) {
10051006
return c.l, c.c, c.intervals
10061007
}

‎compact_test.go‎

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@ import (
3030
"github.com/prometheus/tsdb/chunks"
3131
"github.com/prometheus/tsdb/fileutil"
3232
"github.com/prometheus/tsdb/labels"
33+
"github.com/prometheus/tsdb/record"
3334
"github.com/prometheus/tsdb/testutil"
3435
)
3536

@@ -455,11 +456,13 @@ func metaRange(name string, mint, maxt int64, stats *BlockStats) dirMeta {
455456

456457
type erringBReader struct{}
457458

458-
func (erringBReader) Index() (IndexReader, error) { return nil, errors.New("index") }
459-
func (erringBReader) Chunks() (ChunkReader, error) { return nil, errors.New("chunks") }
460-
func (erringBReader) Tombstones() (TombstoneReader, error) { return nil, errors.New("tombstones") }
461-
func (erringBReader) MinTime() int64 { return 0 }
462-
func (erringBReader) MaxTime() int64 { return 0 }
459+
func (erringBReader) Index() (IndexReader, error) { return nil, errors.New("index") }
460+
func (erringBReader) Chunks() (ChunkReader, error) { return nil, errors.New("chunks") }
461+
func (erringBReader) Tombstones() (record.TombstoneReader, error) {
462+
return nil, errors.New("tombstones")
463+
}
464+
func (erringBReader) MinTime() int64 { return 0 }
465+
func (erringBReader) MaxTime() int64 { return 0 }
463466

464467
type nopChunkWriter struct{}
465468

‎db_test.go‎

Lines changed: 24 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@ import (
3333
"github.com/prometheus/tsdb/chunks"
3434
"github.com/prometheus/tsdb/index"
3535
"github.com/prometheus/tsdb/labels"
36+
"github.com/prometheus/tsdb/record"
3637
"github.com/prometheus/tsdb/testutil"
3738
"github.com/prometheus/tsdb/tsdbutil"
3839
"github.com/prometheus/tsdb/wal"
@@ -199,7 +200,7 @@ func TestDBAppenderAddRef(t *testing.T) {
199200
testutil.Ok(t, err)
200201

201202
err = app2.AddFast(9999999, 1, 1)
202-
testutil.Equals(t, ErrNotFound, errors.Cause(err))
203+
testutil.Equals(t, record.ErrNotFound, errors.Cause(err))
203204

204205
testutil.Ok(t, app2.Commit())
205206

@@ -246,27 +247,27 @@ func TestDeleteSimple(t *testing.T) {
246247
numSamples := int64(10)
247248

248249
cases := []struct {
249-
intervals Intervals
250+
intervals record.Intervals
250251
remaint []int64
251252
}{
252253
{
253-
intervals: Intervals{{0, 3}},
254+
intervals: record.Intervals{{0, 3}},
254255
remaint: []int64{4, 5, 6, 7, 8, 9},
255256
},
256257
{
257-
intervals: Intervals{{1, 3}},
258+
intervals: record.Intervals{{1, 3}},
258259
remaint: []int64{0, 4, 5, 6, 7, 8, 9},
259260
},
260261
{
261-
intervals: Intervals{{1, 3}, {4, 7}},
262+
intervals: record.Intervals{{1, 3}, {4, 7}},
262263
remaint: []int64{0, 8, 9},
263264
},
264265
{
265-
intervals: Intervals{{1, 3}, {4, 700}},
266+
intervals: record.Intervals{{1, 3}, {4, 700}},
266267
remaint: []int64{0},
267268
},
268269
{ // This case is to ensure that labels and symbols are deleted.
269-
intervals: Intervals{{0, 9}},
270+
intervals: record.Intervals{{0, 9}},
270271
remaint: []int64{},
271272
},
272273
}
@@ -362,7 +363,7 @@ func TestAmendDatapointCausesError(t *testing.T) {
362363

363364
app = db.Appender()
364365
_, err = app.Add(labels.Labels{}, 0, 1)
365-
testutil.Equals(t, ErrAmendSample, err)
366+
testutil.Equals(t, record.ErrAmendSample, err)
366367
testutil.Ok(t, app.Rollback())
367368
}
368369

@@ -396,7 +397,7 @@ func TestNonDuplicateNaNDatapointsCausesAmendError(t *testing.T) {
396397

397398
app = db.Appender()
398399
_, err = app.Add(labels.Labels{}, 0, math.Float64frombits(0x7ff0000000000002))
399-
testutil.Equals(t, ErrAmendSample, err)
400+
testutil.Equals(t, record.ErrAmendSample, err)
400401
}
401402

402403
func TestSkippingInvalidValuesInSameTxn(t *testing.T) {
@@ -508,11 +509,11 @@ func TestDB_SnapshotWithDelete(t *testing.T) {
508509

509510
testutil.Ok(t, app.Commit())
510511
cases := []struct {
511-
intervals Intervals
512+
intervals record.Intervals
512513
remaint []int64
513514
}{
514515
{
515-
intervals: Intervals{{1, 3}, {4, 7}},
516+
intervals: record.Intervals{{1, 3}, {4, 7}},
516517
remaint: []int64{0, 8, 9},
517518
},
518519
}
@@ -835,11 +836,11 @@ func TestTombstoneClean(t *testing.T) {
835836

836837
testutil.Ok(t, app.Commit())
837838
cases := []struct {
838-
intervals Intervals
839+
intervals record.Intervals
839840
remaint []int64
840841
}{
841842
{
842-
intervals: Intervals{{1, 3}, {4, 7}},
843+
intervals: record.Intervals{{1, 3}, {4, 7}},
843844
remaint: []int64{0, 8, 9},
844845
},
845846
}
@@ -911,7 +912,7 @@ func TestTombstoneClean(t *testing.T) {
911912
}
912913

913914
for _, b := range db.Blocks() {
914-
testutil.Equals(t, newMemTombstones(), b.tombstones)
915+
testutil.Equals(t, record.NewMemTombstones(), b.tombstones)
915916
}
916917
}
917918
}
@@ -937,8 +938,8 @@ func TestTombstoneCleanFail(t *testing.T) {
937938
block, err := OpenBlock(nil, blockDir, nil)
938939
testutil.Ok(t, err)
939940
// Add some some fake tombstones to trigger the compaction.
940-
tomb := newMemTombstones()
941-
tomb.addInterval(0, Interval{0, 1})
941+
tomb := record.NewMemTombstones()
942+
tomb.AddInterval(0, record.Interval{0, 1})
942943
block.tombstones = tomb
943944

944945
db.blocks = append(db.blocks, block)
@@ -1091,7 +1092,7 @@ func dbDiskSize(dir string) int64 {
10911092
// Include only index,tombstone and chunks.
10921093
if filepath.Dir(path) == chunkDir(filepath.Dir(filepath.Dir(path))) ||
10931094
info.Name() == indexFilename ||
1094-
info.Name() == tombstoneFilename {
1095+
info.Name() == record.TombstoneFilename {
10951096
statSize += info.Size()
10961097
}
10971098
return nil
@@ -1407,13 +1408,13 @@ func TestInitializeHeadTimestamp(t *testing.T) {
14071408
w, err := wal.New(nil, nil, path.Join(dir, "wal"))
14081409
testutil.Ok(t, err)
14091410

1410-
var enc RecordEncoder
1411+
var enc record.RecordEncoder
14111412
err = w.Log(
1412-
enc.Series([]RefSeries{
1413+
enc.Series([]record.RefSeries{
14131414
{Ref: 123, Labels: labels.FromStrings("a", "1")},
14141415
{Ref: 124, Labels: labels.FromStrings("a", "2")},
14151416
}, nil),
1416-
enc.Samples([]RefSample{
1417+
enc.Samples([]record.RefSample{
14171418
{Ref: 123, T: 5000, V: 1},
14181419
{Ref: 124, T: 15000, V: 1},
14191420
}, nil),
@@ -1457,13 +1458,13 @@ func TestInitializeHeadTimestamp(t *testing.T) {
14571458
w, err := wal.New(nil, nil, path.Join(dir, "wal"))
14581459
testutil.Ok(t, err)
14591460

1460-
var enc RecordEncoder
1461+
var enc record.RecordEncoder
14611462
err = w.Log(
1462-
enc.Series([]RefSeries{
1463+
enc.Series([]record.RefSeries{
14631464
{Ref: 123, Labels: labels.FromStrings("a", "1")},
14641465
{Ref: 124, Labels: labels.FromStrings("a", "2")},
14651466
}, nil),
1466-
enc.Samples([]RefSample{
1467+
enc.Samples([]record.RefSample{
14671468
{Ref: 123, T: 5000, V: 1},
14681469
{Ref: 124, T: 15000, V: 1},
14691470
}, nil),

0 commit comments

Comments
 (0)