mirror of https://github.com/prometheus/prometheus
You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
1571 lines
45 KiB
1571 lines
45 KiB
// Copyright 2017 The Prometheus Authors
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
// you may not use this file except in compliance with the License.
|
|
// You may obtain a copy of the License at
|
|
//
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
//
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
// See the License for the specific language governing permissions and
|
|
// limitations under the License.
|
|
|
|
package tsdb
|
|
|
|
import (
|
|
"fmt"
|
|
"io/ioutil"
|
|
"math"
|
|
"math/rand"
|
|
"os"
|
|
"path"
|
|
"path/filepath"
|
|
"sort"
|
|
"strconv"
|
|
"sync"
|
|
"testing"
|
|
|
|
"github.com/pkg/errors"
|
|
prom_testutil "github.com/prometheus/client_golang/prometheus/testutil"
|
|
"github.com/prometheus/prometheus/pkg/labels"
|
|
"github.com/prometheus/prometheus/storage"
|
|
"github.com/prometheus/prometheus/tsdb/chunkenc"
|
|
"github.com/prometheus/prometheus/tsdb/chunks"
|
|
"github.com/prometheus/prometheus/tsdb/index"
|
|
"github.com/prometheus/prometheus/tsdb/record"
|
|
"github.com/prometheus/prometheus/tsdb/tombstones"
|
|
"github.com/prometheus/prometheus/tsdb/tsdbutil"
|
|
"github.com/prometheus/prometheus/tsdb/wal"
|
|
"github.com/prometheus/prometheus/util/testutil"
|
|
)
|
|
|
|
func BenchmarkCreateSeries(b *testing.B) {
|
|
series := genSeries(b.N, 10, 0, 0)
|
|
|
|
h, err := NewHead(nil, nil, nil, 10000, DefaultStripeSize)
|
|
testutil.Ok(b, err)
|
|
defer h.Close()
|
|
|
|
b.ReportAllocs()
|
|
b.ResetTimer()
|
|
|
|
for _, s := range series {
|
|
h.getOrCreate(s.Labels().Hash(), s.Labels())
|
|
}
|
|
}
|
|
|
|
func populateTestWAL(t testing.TB, w *wal.WAL, recs []interface{}) {
|
|
var enc record.Encoder
|
|
for _, r := range recs {
|
|
switch v := r.(type) {
|
|
case []record.RefSeries:
|
|
testutil.Ok(t, w.Log(enc.Series(v, nil)))
|
|
case []record.RefSample:
|
|
testutil.Ok(t, w.Log(enc.Samples(v, nil)))
|
|
case []tombstones.Stone:
|
|
testutil.Ok(t, w.Log(enc.Tombstones(v, nil)))
|
|
}
|
|
}
|
|
}
|
|
|
|
func readTestWAL(t testing.TB, dir string) (recs []interface{}) {
|
|
sr, err := wal.NewSegmentsReader(dir)
|
|
testutil.Ok(t, err)
|
|
defer sr.Close()
|
|
|
|
var dec record.Decoder
|
|
r := wal.NewReader(sr)
|
|
|
|
for r.Next() {
|
|
rec := r.Record()
|
|
|
|
switch dec.Type(rec) {
|
|
case record.Series:
|
|
series, err := dec.Series(rec, nil)
|
|
testutil.Ok(t, err)
|
|
recs = append(recs, series)
|
|
case record.Samples:
|
|
samples, err := dec.Samples(rec, nil)
|
|
testutil.Ok(t, err)
|
|
recs = append(recs, samples)
|
|
case record.Tombstones:
|
|
tstones, err := dec.Tombstones(rec, nil)
|
|
testutil.Ok(t, err)
|
|
recs = append(recs, tstones)
|
|
default:
|
|
t.Fatalf("unknown record type")
|
|
}
|
|
}
|
|
testutil.Ok(t, r.Err())
|
|
return recs
|
|
}
|
|
|
|
func BenchmarkLoadWAL(b *testing.B) {
|
|
cases := []struct {
|
|
// Total series is (batches*seriesPerBatch).
|
|
batches int
|
|
seriesPerBatch int
|
|
samplesPerSeries int
|
|
}{
|
|
{ // Less series and more samples. 2 hour WAL with 1 second scrape interval.
|
|
batches: 10,
|
|
seriesPerBatch: 100,
|
|
samplesPerSeries: 7200,
|
|
},
|
|
{ // More series and less samples.
|
|
batches: 10,
|
|
seriesPerBatch: 10000,
|
|
samplesPerSeries: 50,
|
|
},
|
|
{ // In between.
|
|
batches: 10,
|
|
seriesPerBatch: 1000,
|
|
samplesPerSeries: 480,
|
|
},
|
|
}
|
|
|
|
labelsPerSeries := 5
|
|
for _, c := range cases {
|
|
b.Run(fmt.Sprintf("batches=%d,seriesPerBatch=%d,samplesPerSeries=%d", c.batches, c.seriesPerBatch, c.samplesPerSeries),
|
|
func(b *testing.B) {
|
|
dir, err := ioutil.TempDir("", "test_load_wal")
|
|
testutil.Ok(b, err)
|
|
defer func() {
|
|
testutil.Ok(b, os.RemoveAll(dir))
|
|
}()
|
|
|
|
w, err := wal.New(nil, nil, dir, false)
|
|
testutil.Ok(b, err)
|
|
|
|
// Write series.
|
|
refSeries := make([]record.RefSeries, 0, c.seriesPerBatch)
|
|
for k := 0; k < c.batches; k++ {
|
|
refSeries = refSeries[:0]
|
|
for i := k * c.seriesPerBatch; i < (k+1)*c.seriesPerBatch; i++ {
|
|
lbls := make(map[string]string, labelsPerSeries)
|
|
lbls[defaultLabelName] = strconv.Itoa(i)
|
|
for j := 1; len(lbls) < labelsPerSeries; j++ {
|
|
lbls[defaultLabelName+strconv.Itoa(j)] = defaultLabelValue + strconv.Itoa(j)
|
|
}
|
|
refSeries = append(refSeries, record.RefSeries{Ref: uint64(i) * 100, Labels: labels.FromMap(lbls)})
|
|
}
|
|
populateTestWAL(b, w, []interface{}{refSeries})
|
|
}
|
|
|
|
// Write samples.
|
|
refSamples := make([]record.RefSample, 0, c.seriesPerBatch)
|
|
for i := 0; i < c.samplesPerSeries; i++ {
|
|
for j := 0; j < c.batches; j++ {
|
|
refSamples = refSamples[:0]
|
|
for k := j * c.seriesPerBatch; k < (j+1)*c.seriesPerBatch; k++ {
|
|
refSamples = append(refSamples, record.RefSample{
|
|
Ref: uint64(k) * 100,
|
|
T: int64(i) * 10,
|
|
V: float64(i) * 100,
|
|
})
|
|
}
|
|
populateTestWAL(b, w, []interface{}{refSamples})
|
|
}
|
|
}
|
|
|
|
b.ResetTimer()
|
|
|
|
// Load the WAL.
|
|
for i := 0; i < b.N; i++ {
|
|
h, err := NewHead(nil, nil, w, 10000, DefaultStripeSize)
|
|
testutil.Ok(b, err)
|
|
h.Init(0)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestHead_ReadWAL(t *testing.T) {
|
|
for _, compress := range []bool{false, true} {
|
|
t.Run(fmt.Sprintf("compress=%t", compress), func(t *testing.T) {
|
|
entries := []interface{}{
|
|
[]record.RefSeries{
|
|
{Ref: 10, Labels: labels.FromStrings("a", "1")},
|
|
{Ref: 11, Labels: labels.FromStrings("a", "2")},
|
|
{Ref: 100, Labels: labels.FromStrings("a", "3")},
|
|
},
|
|
[]record.RefSample{
|
|
{Ref: 0, T: 99, V: 1},
|
|
{Ref: 10, T: 100, V: 2},
|
|
{Ref: 100, T: 100, V: 3},
|
|
},
|
|
[]record.RefSeries{
|
|
{Ref: 50, Labels: labels.FromStrings("a", "4")},
|
|
// This series has two refs pointing to it.
|
|
{Ref: 101, Labels: labels.FromStrings("a", "3")},
|
|
},
|
|
[]record.RefSample{
|
|
{Ref: 10, T: 101, V: 5},
|
|
{Ref: 50, T: 101, V: 6},
|
|
{Ref: 101, T: 101, V: 7},
|
|
},
|
|
[]tombstones.Stone{
|
|
{Ref: 0, Intervals: []tombstones.Interval{{Mint: 99, Maxt: 101}}},
|
|
},
|
|
}
|
|
dir, err := ioutil.TempDir("", "test_read_wal")
|
|
testutil.Ok(t, err)
|
|
defer func() {
|
|
testutil.Ok(t, os.RemoveAll(dir))
|
|
}()
|
|
|
|
w, err := wal.New(nil, nil, dir, compress)
|
|
testutil.Ok(t, err)
|
|
defer w.Close()
|
|
populateTestWAL(t, w, entries)
|
|
|
|
head, err := NewHead(nil, nil, w, 10000, DefaultStripeSize)
|
|
testutil.Ok(t, err)
|
|
|
|
testutil.Ok(t, head.Init(math.MinInt64))
|
|
testutil.Equals(t, uint64(101), head.lastSeriesID)
|
|
|
|
s10 := head.series.getByID(10)
|
|
s11 := head.series.getByID(11)
|
|
s50 := head.series.getByID(50)
|
|
s100 := head.series.getByID(100)
|
|
|
|
testutil.Equals(t, labels.FromStrings("a", "1"), s10.lset)
|
|
testutil.Equals(t, (*memSeries)(nil), s11) // Series without samples should be garbage collected at head.Init().
|
|
testutil.Equals(t, labels.FromStrings("a", "4"), s50.lset)
|
|
testutil.Equals(t, labels.FromStrings("a", "3"), s100.lset)
|
|
|
|
expandChunk := func(c chunkenc.Iterator) (x []sample) {
|
|
for c.Next() {
|
|
t, v := c.At()
|
|
x = append(x, sample{t: t, v: v})
|
|
}
|
|
testutil.Ok(t, c.Err())
|
|
return x
|
|
}
|
|
testutil.Equals(t, []sample{{100, 2}, {101, 5}}, expandChunk(s10.iterator(0, nil, nil)))
|
|
testutil.Equals(t, []sample{{101, 6}}, expandChunk(s50.iterator(0, nil, nil)))
|
|
testutil.Equals(t, []sample{{100, 3}, {101, 7}}, expandChunk(s100.iterator(0, nil, nil)))
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestHead_WALMultiRef(t *testing.T) {
|
|
dir, err := ioutil.TempDir("", "test_wal_multi_ref")
|
|
testutil.Ok(t, err)
|
|
defer func() {
|
|
testutil.Ok(t, os.RemoveAll(dir))
|
|
}()
|
|
|
|
w, err := wal.New(nil, nil, dir, false)
|
|
testutil.Ok(t, err)
|
|
|
|
head, err := NewHead(nil, nil, w, 10000, DefaultStripeSize)
|
|
testutil.Ok(t, err)
|
|
|
|
testutil.Ok(t, head.Init(0))
|
|
app := head.Appender()
|
|
ref1, err := app.Add(labels.FromStrings("foo", "bar"), 100, 1)
|
|
testutil.Ok(t, err)
|
|
testutil.Ok(t, app.Commit())
|
|
|
|
testutil.Ok(t, head.Truncate(200))
|
|
|
|
app = head.Appender()
|
|
ref2, err := app.Add(labels.FromStrings("foo", "bar"), 300, 2)
|
|
testutil.Ok(t, err)
|
|
testutil.Ok(t, app.Commit())
|
|
|
|
if ref1 == ref2 {
|
|
t.Fatal("Refs are the same")
|
|
}
|
|
testutil.Ok(t, head.Close())
|
|
|
|
w, err = wal.New(nil, nil, dir, false)
|
|
testutil.Ok(t, err)
|
|
|
|
head, err = NewHead(nil, nil, w, 10000, DefaultStripeSize)
|
|
testutil.Ok(t, err)
|
|
testutil.Ok(t, head.Init(0))
|
|
defer head.Close()
|
|
|
|
q, err := NewBlockQuerier(head, 0, 300)
|
|
testutil.Ok(t, err)
|
|
series := query(t, q, labels.MustNewMatcher(labels.MatchEqual, "foo", "bar"))
|
|
testutil.Equals(t, map[string][]tsdbutil.Sample{`{foo="bar"}`: {sample{100, 1}, sample{300, 2}}}, series)
|
|
}
|
|
|
|
func TestHead_Truncate(t *testing.T) {
|
|
dir, err := ioutil.TempDir("", "test_truncate")
|
|
testutil.Ok(t, err)
|
|
defer func() {
|
|
testutil.Ok(t, os.RemoveAll(dir))
|
|
}()
|
|
|
|
w, err := wal.New(nil, nil, dir, false)
|
|
testutil.Ok(t, err)
|
|
|
|
h, err := NewHead(nil, nil, w, 10000, DefaultStripeSize)
|
|
testutil.Ok(t, err)
|
|
defer h.Close()
|
|
|
|
h.initTime(0)
|
|
|
|
s1, _ := h.getOrCreate(1, labels.FromStrings("a", "1", "b", "1"))
|
|
s2, _ := h.getOrCreate(2, labels.FromStrings("a", "2", "b", "1"))
|
|
s3, _ := h.getOrCreate(3, labels.FromStrings("a", "1", "b", "2"))
|
|
s4, _ := h.getOrCreate(4, labels.FromStrings("a", "2", "b", "2", "c", "1"))
|
|
|
|
s1.chunks = []*memChunk{
|
|
{minTime: 0, maxTime: 999, chunk: chunkenc.NewXORChunk()},
|
|
{minTime: 1000, maxTime: 1999, chunk: chunkenc.NewXORChunk()},
|
|
{minTime: 2000, maxTime: 2999, chunk: chunkenc.NewXORChunk()},
|
|
}
|
|
s2.chunks = []*memChunk{
|
|
{minTime: 1000, maxTime: 1999, chunk: chunkenc.NewXORChunk()},
|
|
{minTime: 2000, maxTime: 2999, chunk: chunkenc.NewXORChunk()},
|
|
{minTime: 3000, maxTime: 3999, chunk: chunkenc.NewXORChunk()},
|
|
}
|
|
s3.chunks = []*memChunk{
|
|
{minTime: 0, maxTime: 999, chunk: chunkenc.NewXORChunk()},
|
|
{minTime: 1000, maxTime: 1999, chunk: chunkenc.NewXORChunk()},
|
|
}
|
|
s4.chunks = []*memChunk{}
|
|
|
|
// Truncation need not be aligned.
|
|
testutil.Ok(t, h.Truncate(1))
|
|
|
|
testutil.Ok(t, h.Truncate(2000))
|
|
|
|
testutil.Equals(t, []*memChunk{
|
|
{minTime: 2000, maxTime: 2999, chunk: chunkenc.NewXORChunk()},
|
|
}, h.series.getByID(s1.ref).chunks)
|
|
|
|
testutil.Equals(t, []*memChunk{
|
|
{minTime: 2000, maxTime: 2999, chunk: chunkenc.NewXORChunk()},
|
|
{minTime: 3000, maxTime: 3999, chunk: chunkenc.NewXORChunk()},
|
|
}, h.series.getByID(s2.ref).chunks)
|
|
|
|
testutil.Assert(t, h.series.getByID(s3.ref) == nil, "")
|
|
testutil.Assert(t, h.series.getByID(s4.ref) == nil, "")
|
|
|
|
postingsA1, _ := index.ExpandPostings(h.postings.Get("a", "1"))
|
|
postingsA2, _ := index.ExpandPostings(h.postings.Get("a", "2"))
|
|
postingsB1, _ := index.ExpandPostings(h.postings.Get("b", "1"))
|
|
postingsB2, _ := index.ExpandPostings(h.postings.Get("b", "2"))
|
|
postingsC1, _ := index.ExpandPostings(h.postings.Get("c", "1"))
|
|
postingsAll, _ := index.ExpandPostings(h.postings.Get("", ""))
|
|
|
|
testutil.Equals(t, []uint64{s1.ref}, postingsA1)
|
|
testutil.Equals(t, []uint64{s2.ref}, postingsA2)
|
|
testutil.Equals(t, []uint64{s1.ref, s2.ref}, postingsB1)
|
|
testutil.Equals(t, []uint64{s1.ref, s2.ref}, postingsAll)
|
|
testutil.Assert(t, postingsB2 == nil, "")
|
|
testutil.Assert(t, postingsC1 == nil, "")
|
|
|
|
testutil.Equals(t, map[string]struct{}{
|
|
"": {}, // from 'all' postings list
|
|
"a": {},
|
|
"b": {},
|
|
"1": {},
|
|
"2": {},
|
|
}, h.symbols)
|
|
|
|
testutil.Equals(t, map[string]stringset{
|
|
"a": {"1": struct{}{}, "2": struct{}{}},
|
|
"b": {"1": struct{}{}},
|
|
"": {"": struct{}{}},
|
|
}, h.values)
|
|
}
|
|
|
|
// Validate various behaviors brought on by firstChunkID accounting for
|
|
// garbage collected chunks.
|
|
func TestMemSeries_truncateChunks(t *testing.T) {
|
|
s := newMemSeries(labels.FromStrings("a", "b"), 1, 2000)
|
|
|
|
for i := 0; i < 4000; i += 5 {
|
|
ok, _ := s.append(int64(i), float64(i), 0)
|
|
testutil.Assert(t, ok == true, "sample append failed")
|
|
}
|
|
|
|
// Check that truncate removes half of the chunks and afterwards
|
|
// that the ID of the last chunk still gives us the same chunk afterwards.
|
|
countBefore := len(s.chunks)
|
|
lastID := s.chunkID(countBefore - 1)
|
|
lastChunk := s.chunk(lastID)
|
|
|
|
testutil.Assert(t, s.chunk(0) != nil, "")
|
|
testutil.Assert(t, lastChunk != nil, "")
|
|
|
|
s.truncateChunksBefore(2000)
|
|
|
|
testutil.Equals(t, int64(2000), s.chunks[0].minTime)
|
|
testutil.Assert(t, s.chunk(0) == nil, "first chunks not gone")
|
|
testutil.Equals(t, countBefore/2, len(s.chunks))
|
|
testutil.Equals(t, lastChunk, s.chunk(lastID))
|
|
|
|
// Validate that the series' sample buffer is applied correctly to the last chunk
|
|
// after truncation.
|
|
it1 := s.iterator(s.chunkID(len(s.chunks)-1), nil, nil)
|
|
_, ok := it1.(*memSafeIterator)
|
|
testutil.Assert(t, ok == true, "")
|
|
|
|
it2 := s.iterator(s.chunkID(len(s.chunks)-2), nil, nil)
|
|
_, ok = it2.(*memSafeIterator)
|
|
testutil.Assert(t, ok == false, "non-last chunk incorrectly wrapped with sample buffer")
|
|
}
|
|
|
|
func TestHeadDeleteSeriesWithoutSamples(t *testing.T) {
|
|
for _, compress := range []bool{false, true} {
|
|
t.Run(fmt.Sprintf("compress=%t", compress), func(t *testing.T) {
|
|
entries := []interface{}{
|
|
[]record.RefSeries{
|
|
{Ref: 10, Labels: labels.FromStrings("a", "1")},
|
|
},
|
|
[]record.RefSample{},
|
|
[]record.RefSeries{
|
|
{Ref: 50, Labels: labels.FromStrings("a", "2")},
|
|
},
|
|
[]record.RefSample{
|
|
{Ref: 50, T: 80, V: 1},
|
|
{Ref: 50, T: 90, V: 1},
|
|
},
|
|
}
|
|
dir, err := ioutil.TempDir("", "test_delete_series")
|
|
testutil.Ok(t, err)
|
|
defer func() {
|
|
testutil.Ok(t, os.RemoveAll(dir))
|
|
}()
|
|
|
|
w, err := wal.New(nil, nil, dir, compress)
|
|
testutil.Ok(t, err)
|
|
defer w.Close()
|
|
populateTestWAL(t, w, entries)
|
|
|
|
head, err := NewHead(nil, nil, w, 1000, DefaultStripeSize)
|
|
testutil.Ok(t, err)
|
|
|
|
testutil.Ok(t, head.Init(math.MinInt64))
|
|
|
|
testutil.Ok(t, head.Delete(0, 100, labels.MustNewMatcher(labels.MatchEqual, "a", "1")))
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestHeadDeleteSimple(t *testing.T) {
|
|
buildSmpls := func(s []int64) []sample {
|
|
ss := make([]sample, 0, len(s))
|
|
for _, t := range s {
|
|
ss = append(ss, sample{t: t, v: float64(t)})
|
|
}
|
|
return ss
|
|
}
|
|
smplsAll := buildSmpls([]int64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9})
|
|
lblDefault := labels.Label{Name: "a", Value: "b"}
|
|
|
|
cases := []struct {
|
|
dranges tombstones.Intervals
|
|
addSamples []sample // Samples to add after delete.
|
|
smplsExp []sample
|
|
}{
|
|
{
|
|
dranges: tombstones.Intervals{{Mint: 0, Maxt: 3}},
|
|
smplsExp: buildSmpls([]int64{4, 5, 6, 7, 8, 9}),
|
|
},
|
|
{
|
|
dranges: tombstones.Intervals{{Mint: 1, Maxt: 3}},
|
|
smplsExp: buildSmpls([]int64{0, 4, 5, 6, 7, 8, 9}),
|
|
},
|
|
{
|
|
dranges: tombstones.Intervals{{Mint: 1, Maxt: 3}, {Mint: 4, Maxt: 7}},
|
|
smplsExp: buildSmpls([]int64{0, 8, 9}),
|
|
},
|
|
{
|
|
dranges: tombstones.Intervals{{Mint: 1, Maxt: 3}, {Mint: 4, Maxt: 700}},
|
|
smplsExp: buildSmpls([]int64{0}),
|
|
},
|
|
{ // This case is to ensure that labels and symbols are deleted.
|
|
dranges: tombstones.Intervals{{Mint: 0, Maxt: 9}},
|
|
smplsExp: buildSmpls([]int64{}),
|
|
},
|
|
{
|
|
dranges: tombstones.Intervals{{Mint: 1, Maxt: 3}},
|
|
addSamples: buildSmpls([]int64{11, 13, 15}),
|
|
smplsExp: buildSmpls([]int64{0, 4, 5, 6, 7, 8, 9, 11, 13, 15}),
|
|
},
|
|
{
|
|
// After delete, the appended samples in the deleted range should be visible
|
|
// as the tombstones are clamped to head min/max time.
|
|
dranges: tombstones.Intervals{{Mint: 7, Maxt: 20}},
|
|
addSamples: buildSmpls([]int64{11, 13, 15}),
|
|
smplsExp: buildSmpls([]int64{0, 1, 2, 3, 4, 5, 6, 11, 13, 15}),
|
|
},
|
|
}
|
|
|
|
for _, compress := range []bool{false, true} {
|
|
t.Run(fmt.Sprintf("compress=%t", compress), func(t *testing.T) {
|
|
for _, c := range cases {
|
|
dir, err := ioutil.TempDir("", "test_wal_reload")
|
|
testutil.Ok(t, err)
|
|
defer func() {
|
|
testutil.Ok(t, os.RemoveAll(dir))
|
|
}()
|
|
|
|
w, err := wal.New(nil, nil, path.Join(dir, "wal"), compress)
|
|
testutil.Ok(t, err)
|
|
defer w.Close()
|
|
|
|
head, err := NewHead(nil, nil, w, 1000, DefaultStripeSize)
|
|
testutil.Ok(t, err)
|
|
defer head.Close()
|
|
|
|
app := head.Appender()
|
|
for _, smpl := range smplsAll {
|
|
_, err = app.Add(labels.Labels{lblDefault}, smpl.t, smpl.v)
|
|
testutil.Ok(t, err)
|
|
|
|
}
|
|
testutil.Ok(t, app.Commit())
|
|
|
|
// Delete the ranges.
|
|
for _, r := range c.dranges {
|
|
testutil.Ok(t, head.Delete(r.Mint, r.Maxt, labels.MustNewMatcher(labels.MatchEqual, lblDefault.Name, lblDefault.Value)))
|
|
}
|
|
|
|
// Add more samples.
|
|
app = head.Appender()
|
|
for _, smpl := range c.addSamples {
|
|
_, err = app.Add(labels.Labels{lblDefault}, smpl.t, smpl.v)
|
|
testutil.Ok(t, err)
|
|
|
|
}
|
|
testutil.Ok(t, app.Commit())
|
|
|
|
// Compare the samples for both heads - before and after the reload.
|
|
reloadedW, err := wal.New(nil, nil, w.Dir(), compress) // Use a new wal to ensure deleted samples are gone even after a reload.
|
|
testutil.Ok(t, err)
|
|
defer reloadedW.Close()
|
|
reloadedHead, err := NewHead(nil, nil, reloadedW, 1000, DefaultStripeSize)
|
|
testutil.Ok(t, err)
|
|
defer reloadedHead.Close()
|
|
testutil.Ok(t, reloadedHead.Init(0))
|
|
|
|
// Compare the query results for both heads - before and after the reload.
|
|
Outer:
|
|
for _, h := range []*Head{head, reloadedHead} {
|
|
q, err := NewBlockQuerier(h, h.MinTime(), h.MaxTime())
|
|
testutil.Ok(t, err)
|
|
actSeriesSet, ws, err := q.Select(false, nil, labels.MustNewMatcher(labels.MatchEqual, lblDefault.Name, lblDefault.Value))
|
|
testutil.Ok(t, err)
|
|
testutil.Equals(t, 0, len(ws))
|
|
expSeriesSet := newMockSeriesSet([]storage.Series{
|
|
newSeries(map[string]string{lblDefault.Name: lblDefault.Value}, func() []tsdbutil.Sample {
|
|
ss := make([]tsdbutil.Sample, 0, len(c.smplsExp))
|
|
for _, s := range c.smplsExp {
|
|
ss = append(ss, s)
|
|
}
|
|
return ss
|
|
}(),
|
|
),
|
|
})
|
|
|
|
for {
|
|
eok, rok := expSeriesSet.Next(), actSeriesSet.Next()
|
|
testutil.Equals(t, eok, rok)
|
|
|
|
if !eok {
|
|
testutil.Ok(t, h.Close())
|
|
continue Outer
|
|
}
|
|
expSeries := expSeriesSet.At()
|
|
actSeries := actSeriesSet.At()
|
|
|
|
testutil.Equals(t, expSeries.Labels(), actSeries.Labels())
|
|
|
|
smplExp, errExp := expandSeriesIterator(expSeries.Iterator())
|
|
smplRes, errRes := expandSeriesIterator(actSeries.Iterator())
|
|
|
|
testutil.Equals(t, errExp, errRes)
|
|
testutil.Equals(t, smplExp, smplRes)
|
|
}
|
|
}
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestDeleteUntilCurMax(t *testing.T) {
|
|
numSamples := int64(10)
|
|
hb, err := NewHead(nil, nil, nil, 1000000, DefaultStripeSize)
|
|
testutil.Ok(t, err)
|
|
defer hb.Close()
|
|
app := hb.Appender()
|
|
smpls := make([]float64, numSamples)
|
|
for i := int64(0); i < numSamples; i++ {
|
|
smpls[i] = rand.Float64()
|
|
_, err := app.Add(labels.Labels{{Name: "a", Value: "b"}}, i, smpls[i])
|
|
testutil.Ok(t, err)
|
|
}
|
|
testutil.Ok(t, app.Commit())
|
|
testutil.Ok(t, hb.Delete(0, 10000, labels.MustNewMatcher(labels.MatchEqual, "a", "b")))
|
|
|
|
// Test the series returns no samples. The series is cleared only after compaction.
|
|
q, err := NewBlockQuerier(hb, 0, 100000)
|
|
testutil.Ok(t, err)
|
|
res, ws, err := q.Select(false, nil, labels.MustNewMatcher(labels.MatchEqual, "a", "b"))
|
|
testutil.Ok(t, err)
|
|
testutil.Equals(t, 0, len(ws))
|
|
testutil.Assert(t, res.Next(), "series is not present")
|
|
s := res.At()
|
|
it := s.Iterator()
|
|
testutil.Assert(t, !it.Next(), "expected no samples")
|
|
|
|
// Add again and test for presence.
|
|
app = hb.Appender()
|
|
_, err = app.Add(labels.Labels{{Name: "a", Value: "b"}}, 11, 1)
|
|
testutil.Ok(t, err)
|
|
testutil.Ok(t, app.Commit())
|
|
q, err = NewBlockQuerier(hb, 0, 100000)
|
|
testutil.Ok(t, err)
|
|
res, ws, err = q.Select(false, nil, labels.MustNewMatcher(labels.MatchEqual, "a", "b"))
|
|
testutil.Ok(t, err)
|
|
testutil.Equals(t, 0, len(ws))
|
|
testutil.Assert(t, res.Next(), "series don't exist")
|
|
exps := res.At()
|
|
it = exps.Iterator()
|
|
resSamples, err := expandSeriesIterator(it)
|
|
testutil.Ok(t, err)
|
|
testutil.Equals(t, []tsdbutil.Sample{sample{11, 1}}, resSamples)
|
|
}
|
|
|
|
func TestDeletedSamplesAndSeriesStillInWALAfterCheckpoint(t *testing.T) {
|
|
dir, err := ioutil.TempDir("", "test_delete_wal")
|
|
testutil.Ok(t, err)
|
|
defer func() {
|
|
testutil.Ok(t, os.RemoveAll(dir))
|
|
}()
|
|
wlog, err := wal.NewSize(nil, nil, dir, 32768, false)
|
|
testutil.Ok(t, err)
|
|
|
|
// Enough samples to cause a checkpoint.
|
|
numSamples := 10000
|
|
hb, err := NewHead(nil, nil, wlog, int64(numSamples)*10, DefaultStripeSize)
|
|
testutil.Ok(t, err)
|
|
defer hb.Close()
|
|
for i := 0; i < numSamples; i++ {
|
|
app := hb.Appender()
|
|
_, err := app.Add(labels.Labels{{Name: "a", Value: "b"}}, int64(i), 0)
|
|
testutil.Ok(t, err)
|
|
testutil.Ok(t, app.Commit())
|
|
}
|
|
testutil.Ok(t, hb.Delete(0, int64(numSamples), labels.MustNewMatcher(labels.MatchEqual, "a", "b")))
|
|
testutil.Ok(t, hb.Truncate(1))
|
|
testutil.Ok(t, hb.Close())
|
|
|
|
// Confirm there's been a checkpoint.
|
|
cdir, _, err := wal.LastCheckpoint(dir)
|
|
testutil.Ok(t, err)
|
|
// Read in checkpoint and WAL.
|
|
recs := readTestWAL(t, cdir)
|
|
recs = append(recs, readTestWAL(t, dir)...)
|
|
|
|
var series, samples, stones int
|
|
for _, rec := range recs {
|
|
switch rec.(type) {
|
|
case []record.RefSeries:
|
|
series++
|
|
case []record.RefSample:
|
|
samples++
|
|
case []tombstones.Stone:
|
|
stones++
|
|
default:
|
|
t.Fatalf("unknown record type")
|
|
}
|
|
}
|
|
testutil.Equals(t, 1, series)
|
|
testutil.Equals(t, 9999, samples)
|
|
testutil.Equals(t, 1, stones)
|
|
|
|
}
|
|
|
|
func TestDelete_e2e(t *testing.T) {
|
|
numDatapoints := 1000
|
|
numRanges := 1000
|
|
timeInterval := int64(2)
|
|
// Create 8 series with 1000 data-points of different ranges, delete and run queries.
|
|
lbls := [][]labels.Label{
|
|
{
|
|
{Name: "a", Value: "b"},
|
|
{Name: "instance", Value: "localhost:9090"},
|
|
{Name: "job", Value: "prometheus"},
|
|
},
|
|
{
|
|
{Name: "a", Value: "b"},
|
|
{Name: "instance", Value: "127.0.0.1:9090"},
|
|
{Name: "job", Value: "prometheus"},
|
|
},
|
|
{
|
|
{Name: "a", Value: "b"},
|
|
{Name: "instance", Value: "127.0.0.1:9090"},
|
|
{Name: "job", Value: "prom-k8s"},
|
|
},
|
|
{
|
|
{Name: "a", Value: "b"},
|
|
{Name: "instance", Value: "localhost:9090"},
|
|
{Name: "job", Value: "prom-k8s"},
|
|
},
|
|
{
|
|
{Name: "a", Value: "c"},
|
|
{Name: "instance", Value: "localhost:9090"},
|
|
{Name: "job", Value: "prometheus"},
|
|
},
|
|
{
|
|
{Name: "a", Value: "c"},
|
|
{Name: "instance", Value: "127.0.0.1:9090"},
|
|
{Name: "job", Value: "prometheus"},
|
|
},
|
|
{
|
|
{Name: "a", Value: "c"},
|
|
{Name: "instance", Value: "127.0.0.1:9090"},
|
|
{Name: "job", Value: "prom-k8s"},
|
|
},
|
|
{
|
|
{Name: "a", Value: "c"},
|
|
{Name: "instance", Value: "localhost:9090"},
|
|
{Name: "job", Value: "prom-k8s"},
|
|
},
|
|
}
|
|
seriesMap := map[string][]tsdbutil.Sample{}
|
|
for _, l := range lbls {
|
|
seriesMap[labels.New(l...).String()] = []tsdbutil.Sample{}
|
|
}
|
|
dir, _ := ioutil.TempDir("", "test")
|
|
defer func() {
|
|
testutil.Ok(t, os.RemoveAll(dir))
|
|
}()
|
|
hb, err := NewHead(nil, nil, nil, 100000, DefaultStripeSize)
|
|
testutil.Ok(t, err)
|
|
defer hb.Close()
|
|
app := hb.Appender()
|
|
for _, l := range lbls {
|
|
ls := labels.New(l...)
|
|
series := []tsdbutil.Sample{}
|
|
ts := rand.Int63n(300)
|
|
for i := 0; i < numDatapoints; i++ {
|
|
v := rand.Float64()
|
|
_, err := app.Add(ls, ts, v)
|
|
testutil.Ok(t, err)
|
|
series = append(series, sample{ts, v})
|
|
ts += rand.Int63n(timeInterval) + 1
|
|
}
|
|
seriesMap[labels.New(l...).String()] = series
|
|
}
|
|
testutil.Ok(t, app.Commit())
|
|
// Delete a time-range from each-selector.
|
|
dels := []struct {
|
|
ms []*labels.Matcher
|
|
drange tombstones.Intervals
|
|
}{
|
|
{
|
|
ms: []*labels.Matcher{labels.MustNewMatcher(labels.MatchEqual, "a", "b")},
|
|
drange: tombstones.Intervals{{Mint: 300, Maxt: 500}, {Mint: 600, Maxt: 670}},
|
|
},
|
|
{
|
|
ms: []*labels.Matcher{
|
|
labels.MustNewMatcher(labels.MatchEqual, "a", "b"),
|
|
labels.MustNewMatcher(labels.MatchEqual, "job", "prom-k8s"),
|
|
},
|
|
drange: tombstones.Intervals{{Mint: 300, Maxt: 500}, {Mint: 100, Maxt: 670}},
|
|
},
|
|
{
|
|
ms: []*labels.Matcher{
|
|
labels.MustNewMatcher(labels.MatchEqual, "a", "c"),
|
|
labels.MustNewMatcher(labels.MatchEqual, "instance", "localhost:9090"),
|
|
labels.MustNewMatcher(labels.MatchEqual, "job", "prometheus"),
|
|
},
|
|
drange: tombstones.Intervals{{Mint: 300, Maxt: 400}, {Mint: 100, Maxt: 6700}},
|
|
},
|
|
// TODO: Add Regexp Matchers.
|
|
}
|
|
for _, del := range dels {
|
|
for _, r := range del.drange {
|
|
testutil.Ok(t, hb.Delete(r.Mint, r.Maxt, del.ms...))
|
|
}
|
|
matched := labels.Slice{}
|
|
for _, ls := range lbls {
|
|
s := labels.Selector(del.ms)
|
|
if s.Matches(ls) {
|
|
matched = append(matched, ls)
|
|
}
|
|
}
|
|
sort.Sort(matched)
|
|
for i := 0; i < numRanges; i++ {
|
|
q, err := NewBlockQuerier(hb, 0, 100000)
|
|
testutil.Ok(t, err)
|
|
defer q.Close()
|
|
ss, ws, err := q.Select(true, nil, del.ms...)
|
|
testutil.Ok(t, err)
|
|
testutil.Equals(t, 0, len(ws))
|
|
// Build the mockSeriesSet.
|
|
matchedSeries := make([]storage.Series, 0, len(matched))
|
|
for _, m := range matched {
|
|
smpls := seriesMap[m.String()]
|
|
smpls = deletedSamples(smpls, del.drange)
|
|
// Only append those series for which samples exist as mockSeriesSet
|
|
// doesn't skip series with no samples.
|
|
// TODO: But sometimes SeriesSet returns an empty SeriesIterator
|
|
if len(smpls) > 0 {
|
|
matchedSeries = append(matchedSeries, newSeries(
|
|
m.Map(),
|
|
smpls,
|
|
))
|
|
}
|
|
}
|
|
expSs := newMockSeriesSet(matchedSeries)
|
|
// Compare both SeriesSets.
|
|
for {
|
|
eok, rok := expSs.Next(), ss.Next()
|
|
// Skip a series if iterator is empty.
|
|
if rok {
|
|
for !ss.At().Iterator().Next() {
|
|
rok = ss.Next()
|
|
if !rok {
|
|
break
|
|
}
|
|
}
|
|
}
|
|
testutil.Equals(t, eok, rok)
|
|
if !eok {
|
|
break
|
|
}
|
|
sexp := expSs.At()
|
|
sres := ss.At()
|
|
testutil.Equals(t, sexp.Labels(), sres.Labels())
|
|
smplExp, errExp := expandSeriesIterator(sexp.Iterator())
|
|
smplRes, errRes := expandSeriesIterator(sres.Iterator())
|
|
testutil.Equals(t, errExp, errRes)
|
|
testutil.Equals(t, smplExp, smplRes)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func boundedSamples(full []tsdbutil.Sample, mint, maxt int64) []tsdbutil.Sample {
|
|
for len(full) > 0 {
|
|
if full[0].T() >= mint {
|
|
break
|
|
}
|
|
full = full[1:]
|
|
}
|
|
for i, s := range full {
|
|
// labels.Labelinate on the first sample larger than maxt.
|
|
if s.T() > maxt {
|
|
return full[:i]
|
|
}
|
|
}
|
|
// maxt is after highest sample.
|
|
return full
|
|
}
|
|
|
|
func deletedSamples(full []tsdbutil.Sample, dranges tombstones.Intervals) []tsdbutil.Sample {
|
|
ds := make([]tsdbutil.Sample, 0, len(full))
|
|
Outer:
|
|
for _, s := range full {
|
|
for _, r := range dranges {
|
|
if r.InBounds(s.T()) {
|
|
continue Outer
|
|
}
|
|
}
|
|
ds = append(ds, s)
|
|
}
|
|
|
|
return ds
|
|
}
|
|
|
|
func TestComputeChunkEndTime(t *testing.T) {
|
|
cases := []struct {
|
|
start, cur, max int64
|
|
res int64
|
|
}{
|
|
{
|
|
start: 0,
|
|
cur: 250,
|
|
max: 1000,
|
|
res: 1000,
|
|
},
|
|
{
|
|
start: 100,
|
|
cur: 200,
|
|
max: 1000,
|
|
res: 550,
|
|
},
|
|
// Case where we fit floored 0 chunks. Must catch division by 0
|
|
// and default to maximum time.
|
|
{
|
|
start: 0,
|
|
cur: 500,
|
|
max: 1000,
|
|
res: 1000,
|
|
},
|
|
// Catch division by zero for cur == start. Strictly not a possible case.
|
|
{
|
|
start: 100,
|
|
cur: 100,
|
|
max: 1000,
|
|
res: 104,
|
|
},
|
|
}
|
|
|
|
for _, c := range cases {
|
|
got := computeChunkEndTime(c.start, c.cur, c.max)
|
|
if got != c.res {
|
|
t.Errorf("expected %d for (start: %d, cur: %d, max: %d), got %d", c.res, c.start, c.cur, c.max, got)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestMemSeries_append(t *testing.T) {
|
|
s := newMemSeries(labels.Labels{}, 1, 500)
|
|
|
|
// Add first two samples at the very end of a chunk range and the next two
|
|
// on and after it.
|
|
// New chunk must correctly be cut at 1000.
|
|
ok, chunkCreated := s.append(998, 1, 0)
|
|
testutil.Assert(t, ok, "append failed")
|
|
testutil.Assert(t, chunkCreated, "first sample created chunk")
|
|
|
|
ok, chunkCreated = s.append(999, 2, 0)
|
|
testutil.Assert(t, ok, "append failed")
|
|
testutil.Assert(t, !chunkCreated, "second sample should use same chunk")
|
|
|
|
ok, chunkCreated = s.append(1000, 3, 0)
|
|
testutil.Assert(t, ok, "append failed")
|
|
testutil.Assert(t, chunkCreated, "expected new chunk on boundary")
|
|
|
|
ok, chunkCreated = s.append(1001, 4, 0)
|
|
testutil.Assert(t, ok, "append failed")
|
|
testutil.Assert(t, !chunkCreated, "second sample should use same chunk")
|
|
|
|
testutil.Assert(t, s.chunks[0].minTime == 998 && s.chunks[0].maxTime == 999, "wrong chunk range")
|
|
testutil.Assert(t, s.chunks[1].minTime == 1000 && s.chunks[1].maxTime == 1001, "wrong chunk range")
|
|
|
|
// Fill the range [1000,2000) with many samples. Intermediate chunks should be cut
|
|
// at approximately 120 samples per chunk.
|
|
for i := 1; i < 1000; i++ {
|
|
ok, _ := s.append(1001+int64(i), float64(i), 0)
|
|
testutil.Assert(t, ok, "append failed")
|
|
}
|
|
|
|
testutil.Assert(t, len(s.chunks) > 7, "expected intermediate chunks")
|
|
|
|
// All chunks but the first and last should now be moderately full.
|
|
for i, c := range s.chunks[1 : len(s.chunks)-1] {
|
|
testutil.Assert(t, c.chunk.NumSamples() > 100, "unexpected small chunk %d of length %d", i, c.chunk.NumSamples())
|
|
}
|
|
}
|
|
|
|
func TestGCChunkAccess(t *testing.T) {
|
|
// Put a chunk, select it. GC it and then access it.
|
|
h, err := NewHead(nil, nil, nil, 1000, DefaultStripeSize)
|
|
testutil.Ok(t, err)
|
|
defer h.Close()
|
|
|
|
h.initTime(0)
|
|
|
|
s, _ := h.getOrCreate(1, labels.FromStrings("a", "1"))
|
|
|
|
// Appending 2 samples for the first chunk.
|
|
ok, chunkCreated := s.append(0, 0, 0)
|
|
testutil.Assert(t, ok, "series append failed")
|
|
testutil.Assert(t, chunkCreated, "chunks was not created")
|
|
ok, chunkCreated = s.append(999, 999, 0)
|
|
testutil.Assert(t, ok, "series append failed")
|
|
testutil.Assert(t, !chunkCreated, "chunks was created")
|
|
|
|
// A new chunks should be created here as it's beyond the chunk range.
|
|
ok, chunkCreated = s.append(1000, 1000, 0)
|
|
testutil.Assert(t, ok, "series append failed")
|
|
testutil.Assert(t, chunkCreated, "chunks was not created")
|
|
ok, chunkCreated = s.append(1999, 1999, 0)
|
|
testutil.Assert(t, ok, "series append failed")
|
|
testutil.Assert(t, !chunkCreated, "chunks was created")
|
|
|
|
idx := h.indexRange(0, 1500)
|
|
var (
|
|
lset labels.Labels
|
|
chunks []chunks.Meta
|
|
)
|
|
testutil.Ok(t, idx.Series(1, &lset, &chunks))
|
|
|
|
testutil.Equals(t, labels.Labels{{
|
|
Name: "a", Value: "1",
|
|
}}, lset)
|
|
testutil.Equals(t, 2, len(chunks))
|
|
|
|
cr := h.chunksRange(0, 1500, nil)
|
|
_, err = cr.Chunk(chunks[0].Ref)
|
|
testutil.Ok(t, err)
|
|
_, err = cr.Chunk(chunks[1].Ref)
|
|
testutil.Ok(t, err)
|
|
|
|
testutil.Ok(t, h.Truncate(1500)) // Remove a chunk.
|
|
|
|
_, err = cr.Chunk(chunks[0].Ref)
|
|
testutil.Equals(t, storage.ErrNotFound, err)
|
|
_, err = cr.Chunk(chunks[1].Ref)
|
|
testutil.Ok(t, err)
|
|
}
|
|
|
|
func TestGCSeriesAccess(t *testing.T) {
|
|
// Put a series, select it. GC it and then access it.
|
|
h, err := NewHead(nil, nil, nil, 1000, DefaultStripeSize)
|
|
testutil.Ok(t, err)
|
|
defer h.Close()
|
|
|
|
h.initTime(0)
|
|
|
|
s, _ := h.getOrCreate(1, labels.FromStrings("a", "1"))
|
|
|
|
// Appending 2 samples for the first chunk.
|
|
ok, chunkCreated := s.append(0, 0, 0)
|
|
testutil.Assert(t, ok, "series append failed")
|
|
testutil.Assert(t, chunkCreated, "chunks was not created")
|
|
ok, chunkCreated = s.append(999, 999, 0)
|
|
testutil.Assert(t, ok, "series append failed")
|
|
testutil.Assert(t, !chunkCreated, "chunks was created")
|
|
|
|
// A new chunks should be created here as it's beyond the chunk range.
|
|
ok, chunkCreated = s.append(1000, 1000, 0)
|
|
testutil.Assert(t, ok, "series append failed")
|
|
testutil.Assert(t, chunkCreated, "chunks was not created")
|
|
ok, chunkCreated = s.append(1999, 1999, 0)
|
|
testutil.Assert(t, ok, "series append failed")
|
|
testutil.Assert(t, !chunkCreated, "chunks was created")
|
|
|
|
idx := h.indexRange(0, 2000)
|
|
var (
|
|
lset labels.Labels
|
|
chunks []chunks.Meta
|
|
)
|
|
testutil.Ok(t, idx.Series(1, &lset, &chunks))
|
|
|
|
testutil.Equals(t, labels.Labels{{
|
|
Name: "a", Value: "1",
|
|
}}, lset)
|
|
testutil.Equals(t, 2, len(chunks))
|
|
|
|
cr := h.chunksRange(0, 2000, nil)
|
|
_, err = cr.Chunk(chunks[0].Ref)
|
|
testutil.Ok(t, err)
|
|
_, err = cr.Chunk(chunks[1].Ref)
|
|
testutil.Ok(t, err)
|
|
|
|
testutil.Ok(t, h.Truncate(2000)) // Remove the series.
|
|
|
|
testutil.Equals(t, (*memSeries)(nil), h.series.getByID(1))
|
|
|
|
_, err = cr.Chunk(chunks[0].Ref)
|
|
testutil.Equals(t, storage.ErrNotFound, err)
|
|
_, err = cr.Chunk(chunks[1].Ref)
|
|
testutil.Equals(t, storage.ErrNotFound, err)
|
|
}
|
|
|
|
func TestUncommittedSamplesNotLostOnTruncate(t *testing.T) {
|
|
h, err := NewHead(nil, nil, nil, 1000, DefaultStripeSize)
|
|
testutil.Ok(t, err)
|
|
defer h.Close()
|
|
|
|
h.initTime(0)
|
|
|
|
app := h.appender()
|
|
lset := labels.FromStrings("a", "1")
|
|
_, err = app.Add(lset, 2100, 1)
|
|
testutil.Ok(t, err)
|
|
|
|
testutil.Ok(t, h.Truncate(2000))
|
|
testutil.Assert(t, nil != h.series.getByHash(lset.Hash(), lset), "series should not have been garbage collected")
|
|
|
|
testutil.Ok(t, app.Commit())
|
|
|
|
q, err := NewBlockQuerier(h, 1500, 2500)
|
|
testutil.Ok(t, err)
|
|
defer q.Close()
|
|
|
|
ss, ws, err := q.Select(false, nil, labels.MustNewMatcher(labels.MatchEqual, "a", "1"))
|
|
testutil.Ok(t, err)
|
|
testutil.Equals(t, 0, len(ws))
|
|
|
|
testutil.Equals(t, true, ss.Next())
|
|
}
|
|
|
|
func TestRemoveSeriesAfterRollbackAndTruncate(t *testing.T) {
|
|
h, err := NewHead(nil, nil, nil, 1000, DefaultStripeSize)
|
|
testutil.Ok(t, err)
|
|
defer h.Close()
|
|
|
|
h.initTime(0)
|
|
|
|
app := h.appender()
|
|
lset := labels.FromStrings("a", "1")
|
|
_, err = app.Add(lset, 2100, 1)
|
|
testutil.Ok(t, err)
|
|
|
|
testutil.Ok(t, h.Truncate(2000))
|
|
testutil.Assert(t, nil != h.series.getByHash(lset.Hash(), lset), "series should not have been garbage collected")
|
|
|
|
testutil.Ok(t, app.Rollback())
|
|
|
|
q, err := NewBlockQuerier(h, 1500, 2500)
|
|
testutil.Ok(t, err)
|
|
defer q.Close()
|
|
|
|
ss, ws, err := q.Select(false, nil, labels.MustNewMatcher(labels.MatchEqual, "a", "1"))
|
|
testutil.Ok(t, err)
|
|
testutil.Equals(t, 0, len(ws))
|
|
|
|
testutil.Equals(t, false, ss.Next())
|
|
|
|
// Truncate again, this time the series should be deleted
|
|
testutil.Ok(t, h.Truncate(2050))
|
|
testutil.Equals(t, (*memSeries)(nil), h.series.getByHash(lset.Hash(), lset))
|
|
}
|
|
|
|
func TestHead_LogRollback(t *testing.T) {
|
|
for _, compress := range []bool{false, true} {
|
|
t.Run(fmt.Sprintf("compress=%t", compress), func(t *testing.T) {
|
|
dir, err := ioutil.TempDir("", "wal_rollback")
|
|
testutil.Ok(t, err)
|
|
defer func() {
|
|
testutil.Ok(t, os.RemoveAll(dir))
|
|
}()
|
|
|
|
w, err := wal.New(nil, nil, dir, compress)
|
|
testutil.Ok(t, err)
|
|
defer w.Close()
|
|
h, err := NewHead(nil, nil, w, 1000, DefaultStripeSize)
|
|
testutil.Ok(t, err)
|
|
|
|
app := h.Appender()
|
|
_, err = app.Add(labels.FromStrings("a", "b"), 1, 2)
|
|
testutil.Ok(t, err)
|
|
|
|
testutil.Ok(t, app.Rollback())
|
|
recs := readTestWAL(t, w.Dir())
|
|
|
|
testutil.Equals(t, 1, len(recs))
|
|
|
|
series, ok := recs[0].([]record.RefSeries)
|
|
testutil.Assert(t, ok, "expected series record but got %+v", recs[0])
|
|
testutil.Equals(t, []record.RefSeries{{Ref: 1, Labels: labels.FromStrings("a", "b")}}, series)
|
|
})
|
|
}
|
|
}
|
|
|
|
// TestWalRepair_DecodingError ensures that a repair is run for an error
|
|
// when decoding a record.
|
|
func TestWalRepair_DecodingError(t *testing.T) {
|
|
var enc record.Encoder
|
|
for name, test := range map[string]struct {
|
|
corrFunc func(rec []byte) []byte // Func that applies the corruption to a record.
|
|
rec []byte
|
|
totalRecs int
|
|
expRecs int
|
|
}{
|
|
"invalid_record": {
|
|
func(rec []byte) []byte {
|
|
// Do not modify the base record because it is Logged multiple times.
|
|
res := make([]byte, len(rec))
|
|
copy(res, rec)
|
|
res[0] = byte(record.Invalid)
|
|
return res
|
|
},
|
|
enc.Series([]record.RefSeries{{Ref: 1, Labels: labels.FromStrings("a", "b")}}, []byte{}),
|
|
9,
|
|
5,
|
|
},
|
|
"decode_series": {
|
|
func(rec []byte) []byte {
|
|
return rec[:3]
|
|
},
|
|
enc.Series([]record.RefSeries{{Ref: 1, Labels: labels.FromStrings("a", "b")}}, []byte{}),
|
|
9,
|
|
5,
|
|
},
|
|
"decode_samples": {
|
|
func(rec []byte) []byte {
|
|
return rec[:3]
|
|
},
|
|
enc.Samples([]record.RefSample{{Ref: 0, T: 99, V: 1}}, []byte{}),
|
|
9,
|
|
5,
|
|
},
|
|
"decode_tombstone": {
|
|
func(rec []byte) []byte {
|
|
return rec[:3]
|
|
},
|
|
enc.Tombstones([]tombstones.Stone{{Ref: 1, Intervals: tombstones.Intervals{}}}, []byte{}),
|
|
9,
|
|
5,
|
|
},
|
|
} {
|
|
for _, compress := range []bool{false, true} {
|
|
t.Run(fmt.Sprintf("%s,compress=%t", name, compress), func(t *testing.T) {
|
|
dir, err := ioutil.TempDir("", "wal_repair")
|
|
testutil.Ok(t, err)
|
|
defer func() {
|
|
testutil.Ok(t, os.RemoveAll(dir))
|
|
}()
|
|
|
|
// Fill the wal and corrupt it.
|
|
{
|
|
w, err := wal.New(nil, nil, filepath.Join(dir, "wal"), compress)
|
|
testutil.Ok(t, err)
|
|
|
|
for i := 1; i <= test.totalRecs; i++ {
|
|
// At this point insert a corrupted record.
|
|
if i-1 == test.expRecs {
|
|
testutil.Ok(t, w.Log(test.corrFunc(test.rec)))
|
|
continue
|
|
}
|
|
testutil.Ok(t, w.Log(test.rec))
|
|
}
|
|
|
|
h, err := NewHead(nil, nil, w, 1, DefaultStripeSize)
|
|
testutil.Ok(t, err)
|
|
testutil.Equals(t, 0.0, prom_testutil.ToFloat64(h.metrics.walCorruptionsTotal))
|
|
initErr := h.Init(math.MinInt64)
|
|
|
|
err = errors.Cause(initErr) // So that we can pick up errors even if wrapped.
|
|
_, corrErr := err.(*wal.CorruptionErr)
|
|
testutil.Assert(t, corrErr, "reading the wal didn't return corruption error")
|
|
testutil.Ok(t, w.Close())
|
|
}
|
|
|
|
// Open the db to trigger a repair.
|
|
{
|
|
db, err := Open(dir, nil, nil, DefaultOptions())
|
|
testutil.Ok(t, err)
|
|
defer func() {
|
|
testutil.Ok(t, db.Close())
|
|
}()
|
|
testutil.Equals(t, 1.0, prom_testutil.ToFloat64(db.head.metrics.walCorruptionsTotal))
|
|
}
|
|
|
|
// Read the wal content after the repair.
|
|
{
|
|
sr, err := wal.NewSegmentsReader(filepath.Join(dir, "wal"))
|
|
testutil.Ok(t, err)
|
|
defer sr.Close()
|
|
r := wal.NewReader(sr)
|
|
|
|
var actRec int
|
|
for r.Next() {
|
|
actRec++
|
|
}
|
|
testutil.Ok(t, r.Err())
|
|
testutil.Equals(t, test.expRecs, actRec, "Wrong number of intact records")
|
|
}
|
|
})
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestNewWalSegmentOnTruncate(t *testing.T) {
|
|
dir, err := ioutil.TempDir("", "test_wal_segments")
|
|
testutil.Ok(t, err)
|
|
defer func() {
|
|
testutil.Ok(t, os.RemoveAll(dir))
|
|
}()
|
|
wlog, err := wal.NewSize(nil, nil, dir, 32768, false)
|
|
testutil.Ok(t, err)
|
|
|
|
h, err := NewHead(nil, nil, wlog, 1000, DefaultStripeSize)
|
|
testutil.Ok(t, err)
|
|
defer h.Close()
|
|
add := func(ts int64) {
|
|
app := h.Appender()
|
|
_, err := app.Add(labels.Labels{{Name: "a", Value: "b"}}, ts, 0)
|
|
testutil.Ok(t, err)
|
|
testutil.Ok(t, app.Commit())
|
|
}
|
|
|
|
add(0)
|
|
_, last, err := wlog.Segments()
|
|
testutil.Ok(t, err)
|
|
testutil.Equals(t, 0, last)
|
|
|
|
add(1)
|
|
testutil.Ok(t, h.Truncate(1))
|
|
_, last, err = wlog.Segments()
|
|
testutil.Ok(t, err)
|
|
testutil.Equals(t, 1, last)
|
|
|
|
add(2)
|
|
testutil.Ok(t, h.Truncate(2))
|
|
_, last, err = wlog.Segments()
|
|
testutil.Ok(t, err)
|
|
testutil.Equals(t, 2, last)
|
|
}
|
|
|
|
func TestAddDuplicateLabelName(t *testing.T) {
|
|
dir, err := ioutil.TempDir("", "test_duplicate_label_name")
|
|
testutil.Ok(t, err)
|
|
defer func() {
|
|
testutil.Ok(t, os.RemoveAll(dir))
|
|
}()
|
|
wlog, err := wal.NewSize(nil, nil, dir, 32768, false)
|
|
testutil.Ok(t, err)
|
|
|
|
h, err := NewHead(nil, nil, wlog, 1000, DefaultStripeSize)
|
|
testutil.Ok(t, err)
|
|
defer h.Close()
|
|
|
|
add := func(labels labels.Labels, labelName string) {
|
|
app := h.Appender()
|
|
_, err = app.Add(labels, 0, 0)
|
|
testutil.NotOk(t, err)
|
|
testutil.Equals(t, fmt.Sprintf(`label name "%s" is not unique: invalid sample`, labelName), err.Error())
|
|
}
|
|
|
|
add(labels.Labels{{Name: "a", Value: "c"}, {Name: "a", Value: "b"}}, "a")
|
|
add(labels.Labels{{Name: "a", Value: "c"}, {Name: "a", Value: "c"}}, "a")
|
|
add(labels.Labels{{Name: "__name__", Value: "up"}, {Name: "job", Value: "prometheus"}, {Name: "le", Value: "500"}, {Name: "le", Value: "400"}, {Name: "unit", Value: "s"}}, "le")
|
|
}
|
|
|
|
func TestMemSeriesIsolation(t *testing.T) {
|
|
// Put a series, select it. GC it and then access it.
|
|
hb, err := NewHead(nil, nil, nil, 1000, DefaultStripeSize)
|
|
testutil.Ok(t, err)
|
|
defer hb.Close()
|
|
|
|
lastValue := func(maxAppendID uint64) int {
|
|
idx, err := hb.Index()
|
|
testutil.Ok(t, err)
|
|
|
|
iso := hb.iso.State()
|
|
iso.maxAppendID = maxAppendID
|
|
|
|
querier := &blockQuerier{
|
|
mint: 0,
|
|
maxt: 10000,
|
|
index: idx,
|
|
chunks: hb.chunksRange(math.MinInt64, math.MaxInt64, iso),
|
|
tombstones: tombstones.NewMemTombstones(),
|
|
}
|
|
|
|
testutil.Ok(t, err)
|
|
defer querier.Close()
|
|
|
|
ss, _, err := querier.Select(false, nil, labels.MustNewMatcher(labels.MatchEqual, "foo", "bar"))
|
|
testutil.Ok(t, err)
|
|
|
|
_, seriesSet, err := expandSeriesSet(ss)
|
|
testutil.Ok(t, err)
|
|
for _, series := range seriesSet {
|
|
return int(series[len(series)-1].v)
|
|
}
|
|
return -1
|
|
}
|
|
|
|
i := 1
|
|
for ; i <= 1000; i++ {
|
|
var app storage.Appender
|
|
// To initialize bounds.
|
|
if hb.MinTime() == math.MaxInt64 {
|
|
app = &initAppender{head: hb}
|
|
} else {
|
|
a := hb.appender()
|
|
a.cleanupAppendIDsBelow = 0
|
|
app = a
|
|
}
|
|
|
|
_, err := app.Add(labels.FromStrings("foo", "bar"), int64(i), float64(i))
|
|
testutil.Ok(t, err)
|
|
testutil.Ok(t, app.Commit())
|
|
}
|
|
|
|
// Test simple cases in different chunks when no appendID cleanup has been performed.
|
|
testutil.Equals(t, 10, lastValue(10))
|
|
testutil.Equals(t, 130, lastValue(130))
|
|
testutil.Equals(t, 160, lastValue(160))
|
|
testutil.Equals(t, 240, lastValue(240))
|
|
testutil.Equals(t, 500, lastValue(500))
|
|
testutil.Equals(t, 750, lastValue(750))
|
|
testutil.Equals(t, 995, lastValue(995))
|
|
testutil.Equals(t, 999, lastValue(999))
|
|
|
|
// Cleanup appendIDs below 500.
|
|
app := hb.appender()
|
|
app.cleanupAppendIDsBelow = 500
|
|
_, err = app.Add(labels.FromStrings("foo", "bar"), int64(i), float64(i))
|
|
testutil.Ok(t, err)
|
|
testutil.Ok(t, app.Commit())
|
|
i++
|
|
|
|
// We should not get queries with a maxAppendID below 500 after the cleanup,
|
|
// but they only take the remaining appendIDs into account.
|
|
testutil.Equals(t, 499, lastValue(10))
|
|
testutil.Equals(t, 499, lastValue(130))
|
|
testutil.Equals(t, 499, lastValue(160))
|
|
testutil.Equals(t, 499, lastValue(240))
|
|
testutil.Equals(t, 500, lastValue(500))
|
|
testutil.Equals(t, 995, lastValue(995))
|
|
testutil.Equals(t, 999, lastValue(999))
|
|
|
|
// Cleanup appendIDs below 1000, which means the sample buffer is
|
|
// the only thing with appendIDs.
|
|
app = hb.appender()
|
|
app.cleanupAppendIDsBelow = 1000
|
|
_, err = app.Add(labels.FromStrings("foo", "bar"), int64(i), float64(i))
|
|
testutil.Ok(t, err)
|
|
testutil.Ok(t, app.Commit())
|
|
testutil.Equals(t, 999, lastValue(998))
|
|
testutil.Equals(t, 999, lastValue(999))
|
|
testutil.Equals(t, 1000, lastValue(1000))
|
|
testutil.Equals(t, 1001, lastValue(1001))
|
|
testutil.Equals(t, 1002, lastValue(1002))
|
|
testutil.Equals(t, 1002, lastValue(1003))
|
|
|
|
i++
|
|
// Cleanup appendIDs below 1001, but with a rollback.
|
|
app = hb.appender()
|
|
app.cleanupAppendIDsBelow = 1001
|
|
_, err = app.Add(labels.FromStrings("foo", "bar"), int64(i), float64(i))
|
|
testutil.Ok(t, err)
|
|
testutil.Ok(t, app.Rollback())
|
|
testutil.Equals(t, 1000, lastValue(999))
|
|
testutil.Equals(t, 1000, lastValue(1000))
|
|
testutil.Equals(t, 1001, lastValue(1001))
|
|
testutil.Equals(t, 1002, lastValue(1002))
|
|
testutil.Equals(t, 1002, lastValue(1003))
|
|
}
|
|
|
|
func TestIsolationRollback(t *testing.T) {
|
|
// Rollback after a failed append and test if the low watermark has progressed anyway.
|
|
hb, err := NewHead(nil, nil, nil, 1000, DefaultStripeSize)
|
|
testutil.Ok(t, err)
|
|
defer hb.Close()
|
|
|
|
app := hb.Appender()
|
|
_, err = app.Add(labels.FromStrings("foo", "bar"), 0, 0)
|
|
testutil.Ok(t, err)
|
|
testutil.Ok(t, app.Commit())
|
|
testutil.Equals(t, uint64(1), hb.iso.lowWatermark())
|
|
|
|
app = hb.Appender()
|
|
_, err = app.Add(labels.FromStrings("foo", "bar"), 1, 1)
|
|
testutil.Ok(t, err)
|
|
_, err = app.Add(labels.FromStrings("foo", "bar", "foo", "baz"), 2, 2)
|
|
testutil.NotOk(t, err)
|
|
testutil.Ok(t, app.Rollback())
|
|
testutil.Equals(t, uint64(2), hb.iso.lowWatermark())
|
|
|
|
app = hb.Appender()
|
|
_, err = app.Add(labels.FromStrings("foo", "bar"), 3, 3)
|
|
testutil.Ok(t, err)
|
|
testutil.Ok(t, app.Commit())
|
|
testutil.Equals(t, uint64(3), hb.iso.lowWatermark(), "Low watermark should proceed to 3 even if append #2 was rolled back.")
|
|
}
|
|
|
|
func TestIsolationLowWatermarkMonotonous(t *testing.T) {
|
|
hb, err := NewHead(nil, nil, nil, 1000, DefaultStripeSize)
|
|
testutil.Ok(t, err)
|
|
defer hb.Close()
|
|
|
|
app1 := hb.Appender()
|
|
_, err = app1.Add(labels.FromStrings("foo", "bar"), 0, 0)
|
|
testutil.Ok(t, err)
|
|
testutil.Ok(t, app1.Commit())
|
|
testutil.Equals(t, uint64(1), hb.iso.lowWatermark(), "Low watermark should by 1 after 1st append.")
|
|
|
|
app1 = hb.Appender()
|
|
_, err = app1.Add(labels.FromStrings("foo", "bar"), 1, 1)
|
|
testutil.Ok(t, err)
|
|
testutil.Equals(t, uint64(2), hb.iso.lowWatermark(), "Low watermark should be two, even if append is not committed yet.")
|
|
|
|
app2 := hb.Appender()
|
|
_, err = app2.Add(labels.FromStrings("foo", "baz"), 1, 1)
|
|
testutil.Ok(t, err)
|
|
testutil.Ok(t, app2.Commit())
|
|
testutil.Equals(t, uint64(2), hb.iso.lowWatermark(), "Low watermark should stay two because app1 is not committed yet.")
|
|
|
|
is := hb.iso.State()
|
|
testutil.Equals(t, uint64(2), hb.iso.lowWatermark(), "After simulated read (iso state retrieved), low watermark should stay at 2.")
|
|
|
|
testutil.Ok(t, app1.Commit())
|
|
testutil.Equals(t, uint64(2), hb.iso.lowWatermark(), "Even after app1 is committed, low watermark should stay at 2 because read is still ongoing.")
|
|
|
|
is.Close()
|
|
testutil.Equals(t, uint64(3), hb.iso.lowWatermark(), "After read has finished (iso state closed), low watermark should jump to three.")
|
|
}
|
|
|
|
func TestIsolationAppendIDZeroIsNoop(t *testing.T) {
|
|
h, err := NewHead(nil, nil, nil, 1000, DefaultStripeSize)
|
|
testutil.Ok(t, err)
|
|
defer h.Close()
|
|
|
|
h.initTime(0)
|
|
|
|
s, _ := h.getOrCreate(1, labels.FromStrings("a", "1"))
|
|
|
|
ok, _ := s.append(0, 0, 0)
|
|
testutil.Assert(t, ok, "Series append failed.")
|
|
testutil.Equals(t, 0, s.txs.txIDCount, "Series should not have an appendID after append with appendID=0.")
|
|
}
|
|
|
|
func TestHeadSeriesChunkRace(t *testing.T) {
|
|
for i := 0; i < 1000; i++ {
|
|
testHeadSeriesChunkRace(t)
|
|
}
|
|
}
|
|
|
|
func TestIsolationWithoutAdd(t *testing.T) {
|
|
hb, err := NewHead(nil, nil, nil, 1000, DefaultStripeSize)
|
|
testutil.Ok(t, err)
|
|
defer hb.Close()
|
|
|
|
app := hb.Appender()
|
|
testutil.Ok(t, app.Commit())
|
|
|
|
app = hb.Appender()
|
|
_, err = app.Add(labels.FromStrings("foo", "baz"), 1, 1)
|
|
testutil.Ok(t, err)
|
|
testutil.Ok(t, app.Commit())
|
|
|
|
testutil.Equals(t, hb.iso.lastAppendID, hb.iso.lowWatermark(), "High watermark should be equal to the low watermark")
|
|
}
|
|
|
|
func testHeadSeriesChunkRace(t *testing.T) {
|
|
h, err := NewHead(nil, nil, nil, 30, DefaultStripeSize)
|
|
testutil.Ok(t, err)
|
|
defer h.Close()
|
|
testutil.Ok(t, h.Init(0))
|
|
app := h.Appender()
|
|
|
|
s2, err := app.Add(labels.FromStrings("foo2", "bar"), 5, 0)
|
|
testutil.Ok(t, err)
|
|
for ts := int64(6); ts < 11; ts++ {
|
|
err = app.AddFast(s2, ts, 0)
|
|
testutil.Ok(t, err)
|
|
}
|
|
testutil.Ok(t, app.Commit())
|
|
|
|
var wg sync.WaitGroup
|
|
matcher := labels.MustNewMatcher(labels.MatchEqual, "", "")
|
|
q, err := NewBlockQuerier(h, 18, 22)
|
|
testutil.Ok(t, err)
|
|
defer q.Close()
|
|
|
|
wg.Add(1)
|
|
go func() {
|
|
h.updateMinMaxTime(20, 25)
|
|
h.gc()
|
|
wg.Done()
|
|
}()
|
|
ss, _, err := q.Select(false, nil, matcher)
|
|
testutil.Ok(t, err)
|
|
testutil.Ok(t, ss.Err())
|
|
wg.Wait()
|
|
}
|