Provide flag to compress the tsdb WAL

Signed-off-by: Chris Marchbanks <csmarchbanks@gmail.com>
This commit is contained in:
Chris Marchbanks 2019-07-03 07:23:13 -06:00
parent 475ca2ecd0
commit 06f1ba73eb
No known key found for this signature in database
GPG key ID: B7FD940BC86A8E7A
3 changed files with 374 additions and 335 deletions

View file

@ -207,6 +207,9 @@ func main() {
a.Flag("storage.tsdb.allow-overlapping-blocks", "[EXPERIMENTAL] Allow overlapping blocks, which in turn enables vertical compaction and vertical query merge."). a.Flag("storage.tsdb.allow-overlapping-blocks", "[EXPERIMENTAL] Allow overlapping blocks, which in turn enables vertical compaction and vertical query merge.").
Default("false").BoolVar(&cfg.tsdb.AllowOverlappingBlocks) Default("false").BoolVar(&cfg.tsdb.AllowOverlappingBlocks)
a.Flag("storage.tsdb.wal-compression", "Compress the tsdb WAL.").
Default("false").BoolVar(&cfg.tsdb.WALCompression)
a.Flag("storage.remote.flush-deadline", "How long to wait flushing sample on shutdown or config reload."). a.Flag("storage.remote.flush-deadline", "How long to wait flushing sample on shutdown or config reload.").
Default("1m").PlaceHolder("<duration>").SetValue(&cfg.RemoteFlushDeadline) Default("1m").PlaceHolder("<duration>").SetValue(&cfg.RemoteFlushDeadline)
@ -671,6 +674,7 @@ func main() {
"RetentionDuration", cfg.tsdb.RetentionDuration, "RetentionDuration", cfg.tsdb.RetentionDuration,
"WALSegmentSize", cfg.tsdb.WALSegmentSize, "WALSegmentSize", cfg.tsdb.WALSegmentSize,
"AllowOverlappingBlocks", cfg.tsdb.AllowOverlappingBlocks, "AllowOverlappingBlocks", cfg.tsdb.AllowOverlappingBlocks,
"WALCompression", cfg.tsdb.WALCompression,
) )
startTimeMargin := int64(2 * time.Duration(cfg.tsdb.MinBlockDuration).Seconds() * 1000) startTimeMargin := int64(2 * time.Duration(cfg.tsdb.MinBlockDuration).Seconds() * 1000)

View file

@ -92,6 +92,8 @@ func TestTailSamples(t *testing.T) {
pageSize := 32 * 1024 pageSize := 32 * 1024
const seriesCount = 10 const seriesCount = 10
const samplesCount = 250 const samplesCount = 250
for _, compress := range []bool{false, true} {
t.Run(fmt.Sprintf("compress=%t", compress), func(t *testing.T) {
now := time.Now() now := time.Now()
dir, err := ioutil.TempDir("", "readCheckpoint") dir, err := ioutil.TempDir("", "readCheckpoint")
@ -103,7 +105,7 @@ func TestTailSamples(t *testing.T) {
testutil.Ok(t, err) testutil.Ok(t, err)
enc := tsdb.RecordEncoder{} enc := tsdb.RecordEncoder{}
w, err := wal.NewSize(nil, nil, wdir, 128*pageSize, false) w, err := wal.NewSize(nil, nil, wdir, 128*pageSize, compress)
testutil.Ok(t, err) testutil.Ok(t, err)
// Write to the initial segment then checkpoint. // Write to the initial segment then checkpoint.
@ -157,6 +159,8 @@ func TestTailSamples(t *testing.T) {
}) })
testutil.Equals(t, expectedSeries, wt.checkNumLabels()) testutil.Equals(t, expectedSeries, wt.checkNumLabels())
testutil.Equals(t, expectedSamples, wt.samplesAppended) testutil.Equals(t, expectedSamples, wt.samplesAppended)
})
}
} }
func TestReadToEndNoCheckpoint(t *testing.T) { func TestReadToEndNoCheckpoint(t *testing.T) {
@ -164,6 +168,8 @@ func TestReadToEndNoCheckpoint(t *testing.T) {
const seriesCount = 10 const seriesCount = 10
const samplesCount = 250 const samplesCount = 250
for _, compress := range []bool{false, true} {
t.Run(fmt.Sprintf("compress=%t", compress), func(t *testing.T) {
dir, err := ioutil.TempDir("", "readToEnd_noCheckpoint") dir, err := ioutil.TempDir("", "readToEnd_noCheckpoint")
testutil.Ok(t, err) testutil.Ok(t, err)
defer os.RemoveAll(dir) defer os.RemoveAll(dir)
@ -171,7 +177,7 @@ func TestReadToEndNoCheckpoint(t *testing.T) {
err = os.Mkdir(wdir, 0777) err = os.Mkdir(wdir, 0777)
testutil.Ok(t, err) testutil.Ok(t, err)
w, err := wal.NewSize(nil, nil, wdir, 128*pageSize, false) w, err := wal.NewSize(nil, nil, wdir, 128*pageSize, compress)
testutil.Ok(t, err) testutil.Ok(t, err)
var recs [][]byte var recs [][]byte
@ -219,6 +225,8 @@ func TestReadToEndNoCheckpoint(t *testing.T) {
}) })
watcher.Stop() watcher.Stop()
testutil.Equals(t, expected, wt.checkNumLabels()) testutil.Equals(t, expected, wt.checkNumLabels())
})
}
} }
func TestReadToEndWithCheckpoint(t *testing.T) { func TestReadToEndWithCheckpoint(t *testing.T) {
@ -228,6 +236,8 @@ func TestReadToEndWithCheckpoint(t *testing.T) {
const seriesCount = 10 const seriesCount = 10
const samplesCount = 250 const samplesCount = 250
for _, compress := range []bool{false, true} {
t.Run(fmt.Sprintf("compress=%t", compress), func(t *testing.T) {
dir, err := ioutil.TempDir("", "readToEnd_withCheckpoint") dir, err := ioutil.TempDir("", "readToEnd_withCheckpoint")
testutil.Ok(t, err) testutil.Ok(t, err)
defer os.RemoveAll(dir) defer os.RemoveAll(dir)
@ -237,7 +247,7 @@ func TestReadToEndWithCheckpoint(t *testing.T) {
testutil.Ok(t, err) testutil.Ok(t, err)
enc := tsdb.RecordEncoder{} enc := tsdb.RecordEncoder{}
w, err := wal.NewSize(nil, nil, wdir, segmentSize, false) w, err := wal.NewSize(nil, nil, wdir, segmentSize, compress)
testutil.Ok(t, err) testutil.Ok(t, err)
// Write to the initial segment then checkpoint. // Write to the initial segment then checkpoint.
@ -301,6 +311,8 @@ func TestReadToEndWithCheckpoint(t *testing.T) {
}) })
watcher.Stop() watcher.Stop()
testutil.Equals(t, expected, wt.checkNumLabels()) testutil.Equals(t, expected, wt.checkNumLabels())
})
}
} }
func TestReadCheckpoint(t *testing.T) { func TestReadCheckpoint(t *testing.T) {
@ -308,6 +320,8 @@ func TestReadCheckpoint(t *testing.T) {
const seriesCount = 10 const seriesCount = 10
const samplesCount = 250 const samplesCount = 250
for _, compress := range []bool{false, true} {
t.Run(fmt.Sprintf("compress=%t", compress), func(t *testing.T) {
dir, err := ioutil.TempDir("", "readCheckpoint") dir, err := ioutil.TempDir("", "readCheckpoint")
testutil.Ok(t, err) testutil.Ok(t, err)
defer os.RemoveAll(dir) defer os.RemoveAll(dir)
@ -319,7 +333,7 @@ func TestReadCheckpoint(t *testing.T) {
os.Create(wal.SegmentName(wdir, 30)) os.Create(wal.SegmentName(wdir, 30))
enc := tsdb.RecordEncoder{} enc := tsdb.RecordEncoder{}
w, err := wal.NewSize(nil, nil, wdir, 128*pageSize, false) w, err := wal.NewSize(nil, nil, wdir, 128*pageSize, compress)
testutil.Ok(t, err) testutil.Ok(t, err)
// Write to the initial segment then checkpoint. // Write to the initial segment then checkpoint.
@ -363,6 +377,8 @@ func TestReadCheckpoint(t *testing.T) {
}) })
watcher.Stop() watcher.Stop()
testutil.Equals(t, expectedSeries, wt.checkNumLabels()) testutil.Equals(t, expectedSeries, wt.checkNumLabels())
})
}
} }
func TestReadCheckpointMultipleSegments(t *testing.T) { func TestReadCheckpointMultipleSegments(t *testing.T) {
@ -372,6 +388,8 @@ func TestReadCheckpointMultipleSegments(t *testing.T) {
const seriesCount = 20 const seriesCount = 20
const samplesCount = 300 const samplesCount = 300
for _, compress := range []bool{false, true} {
t.Run(fmt.Sprintf("compress=%t", compress), func(t *testing.T) {
dir, err := ioutil.TempDir("", "readCheckpoint") dir, err := ioutil.TempDir("", "readCheckpoint")
testutil.Ok(t, err) testutil.Ok(t, err)
defer os.RemoveAll(dir) defer os.RemoveAll(dir)
@ -381,7 +399,7 @@ func TestReadCheckpointMultipleSegments(t *testing.T) {
testutil.Ok(t, err) testutil.Ok(t, err)
enc := tsdb.RecordEncoder{} enc := tsdb.RecordEncoder{}
w, err := wal.NewSize(nil, nil, wdir, pageSize, false) w, err := wal.NewSize(nil, nil, wdir, pageSize, compress)
testutil.Ok(t, err) testutil.Ok(t, err)
// Write a bunch of data. // Write a bunch of data.
@ -431,6 +449,8 @@ func TestReadCheckpointMultipleSegments(t *testing.T) {
err = watcher.readCheckpoint(lastCheckpoint) err = watcher.readCheckpoint(lastCheckpoint)
testutil.Ok(t, err) testutil.Ok(t, err)
})
}
} }
func TestCheckpointSeriesReset(t *testing.T) { func TestCheckpointSeriesReset(t *testing.T) {
@ -439,7 +459,16 @@ func TestCheckpointSeriesReset(t *testing.T) {
// in order to get enough segments for us to checkpoint. // in order to get enough segments for us to checkpoint.
const seriesCount = 20 const seriesCount = 20
const samplesCount = 350 const samplesCount = 350
testCases := []struct {
compress bool
segments int
}{
{compress: false, segments: 14},
{compress: true, segments: 13},
}
for _, tc := range testCases {
t.Run(fmt.Sprintf("compress=%t", tc.compress), func(t *testing.T) {
dir, err := ioutil.TempDir("", "seriesReset") dir, err := ioutil.TempDir("", "seriesReset")
testutil.Ok(t, err) testutil.Ok(t, err)
defer os.RemoveAll(dir) defer os.RemoveAll(dir)
@ -449,7 +478,7 @@ func TestCheckpointSeriesReset(t *testing.T) {
testutil.Ok(t, err) testutil.Ok(t, err)
enc := tsdb.RecordEncoder{} enc := tsdb.RecordEncoder{}
w, err := wal.NewSize(nil, nil, wdir, segmentSize, false) w, err := wal.NewSize(nil, nil, wdir, segmentSize, tc.compress)
testutil.Ok(t, err) testutil.Ok(t, err)
// Write to the initial segment, then checkpoint later. // Write to the initial segment, then checkpoint later.
@ -505,5 +534,7 @@ func TestCheckpointSeriesReset(t *testing.T) {
// If you modify the checkpoint and truncate segment #'s run the test to see how // If you modify the checkpoint and truncate segment #'s run the test to see how
// many series records you end up with and change the last Equals check accordingly // many series records you end up with and change the last Equals check accordingly
// or modify the Equals to Assert(len(wt.seriesLabels) < seriesCount*10) // or modify the Equals to Assert(len(wt.seriesLabels) < seriesCount*10)
testutil.Equals(t, 14, wt.checkNumLabels()) testutil.Equals(t, tc.segments, wt.checkNumLabels())
})
}
} }

View file

@ -130,6 +130,9 @@ type Options struct {
// When true it disables the overlapping blocks check. // When true it disables the overlapping blocks check.
// This in-turn enables vertical compaction and vertical query merge. // This in-turn enables vertical compaction and vertical query merge.
AllowOverlappingBlocks bool AllowOverlappingBlocks bool
// When true records in the WAL will be compressed.
WALCompression bool
} }
var ( var (
@ -195,6 +198,7 @@ func Open(path string, l log.Logger, r prometheus.Registerer, opts *Options) (*t
BlockRanges: rngs, BlockRanges: rngs,
NoLockfile: opts.NoLockfile, NoLockfile: opts.NoLockfile,
AllowOverlappingBlocks: opts.AllowOverlappingBlocks, AllowOverlappingBlocks: opts.AllowOverlappingBlocks,
WALCompression: opts.WALCompression,
}) })
if err != nil { if err != nil {
return nil, err return nil, err