mirror of
https://github.com/prometheus/prometheus.git
synced 2024-11-10 15:44:05 -08:00
0e202dacb4
This will fix issue #1035 and will also help to make issue #1264 less bad. The fundamental problem in the current code: In the preload phase, we quite accurately determine which chunks will be used for the query being executed. However, in the subsequent step of creating series iterators, the created iterators are referencing _all_ in-memory chunks in their series, even the un-pinned ones. In iterator creation, we copy a pointer to each in-memory chunk of a series into the iterator. While this creates a certain amount of allocation churn, the worst thing about it is that copying the chunk pointer out of the chunkDesc requires a mutex acquisition. (Remember that the iterator will also reference un-pinned chunks, so we need to acquire the mutex to protect against concurrent eviction.) The worst case happens if a series doesn't even contain any relevant samples for the query time range. We notice that during preloading but then we will still create a series iterator for it. But even for series that do contain relevant samples, the overhead is quite bad for instant queries that retrieve a single sample from each series, but still go through all the effort of series iterator creation. All of that is particularly bad if a series has many in-memory chunks. This commit addresses the problem from two sides: First, it merges preloading and iterator creation into one step, i.e. the preload call returns an iterator for exactly the preloaded chunks. Second, the required mutex acquisition in chunkDesc has been greatly reduced. That was enabled by a side effect of the first step, which is that the iterator is only referencing pinned chunks, so there is no risk of concurrent eviction anymore, and chunks can be accessed without mutex acquisition. To simplify the code changes for the above, the long-planned change of ValueAtTime to ValueAtOrBefore time was performed at the same time. (It should have been done first, but it kind of accidentally happened while I was in the middle of writing the series iterator changes. Sorry for that.) So far, we actively filtered the up to two values that were returned by ValueAtTime, i.e. we invested work to retrieve up to two values, and then we invested more work to throw one of them away. The SeriesIterator.BoundaryValues method can be removed once #1401 is fixed. But I really didn't want to load even more changes into this PR. Benchmarks: The BenchmarkFuzz.* benchmarks run 83% faster (i.e. about six times faster) and allocate 95% fewer bytes. The reason for that is that the benchmark reads one sample after another from the time series and creates a new series iterator for each sample read. To find out how much these improvements matter in practice, I have mirrored a beefy Prometheus server at SoundCloud that suffers from both issues #1035 and #1264. To reach steady state that would be comparable, the server needs to run for 15d. So far, it has run for 1d. The test server currently has only half as many memory time series and 60% of the memory chunks the main server has. The 90th percentile rule evaluation cycle time is ~11s on the main server and only ~3s on the test server. However, these numbers might get much closer over time. In addition to performance improvements, this commit removes about 150 LOC.
1598 lines
43 KiB
Go
1598 lines
43 KiB
Go
// Copyright 2014 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 local
|
|
|
|
import (
|
|
"fmt"
|
|
"hash/fnv"
|
|
"math/rand"
|
|
"os"
|
|
"reflect"
|
|
"testing"
|
|
"testing/quick"
|
|
"time"
|
|
|
|
"github.com/prometheus/common/log"
|
|
"github.com/prometheus/common/model"
|
|
|
|
"github.com/prometheus/prometheus/storage/metric"
|
|
"github.com/prometheus/prometheus/util/testutil"
|
|
)
|
|
|
|
func TestMatches(t *testing.T) {
|
|
storage, closer := NewTestStorage(t, 1)
|
|
defer closer.Close()
|
|
|
|
samples := make([]*model.Sample, 100)
|
|
fingerprints := make(model.Fingerprints, 100)
|
|
|
|
for i := range samples {
|
|
metric := model.Metric{
|
|
model.MetricNameLabel: model.LabelValue(fmt.Sprintf("test_metric_%d", i)),
|
|
"label1": model.LabelValue(fmt.Sprintf("test_%d", i/10)),
|
|
"label2": model.LabelValue(fmt.Sprintf("test_%d", (i+5)/10)),
|
|
"all": "const",
|
|
}
|
|
samples[i] = &model.Sample{
|
|
Metric: metric,
|
|
Timestamp: model.Time(i),
|
|
Value: model.SampleValue(i),
|
|
}
|
|
fingerprints[i] = metric.FastFingerprint()
|
|
}
|
|
for _, s := range samples {
|
|
storage.Append(s)
|
|
}
|
|
storage.WaitForIndexing()
|
|
|
|
newMatcher := func(matchType metric.MatchType, name model.LabelName, value model.LabelValue) *metric.LabelMatcher {
|
|
lm, err := metric.NewLabelMatcher(matchType, name, value)
|
|
if err != nil {
|
|
t.Fatalf("error creating label matcher: %s", err)
|
|
}
|
|
return lm
|
|
}
|
|
|
|
var matcherTests = []struct {
|
|
matchers metric.LabelMatchers
|
|
expected model.Fingerprints
|
|
}{
|
|
{
|
|
matchers: metric.LabelMatchers{newMatcher(metric.Equal, "label1", "x")},
|
|
expected: model.Fingerprints{},
|
|
},
|
|
{
|
|
matchers: metric.LabelMatchers{newMatcher(metric.Equal, "label1", "test_0")},
|
|
expected: fingerprints[:10],
|
|
},
|
|
{
|
|
matchers: metric.LabelMatchers{
|
|
newMatcher(metric.Equal, "label1", "test_0"),
|
|
newMatcher(metric.Equal, "label2", "test_1"),
|
|
},
|
|
expected: fingerprints[5:10],
|
|
},
|
|
{
|
|
matchers: metric.LabelMatchers{
|
|
newMatcher(metric.Equal, "all", "const"),
|
|
newMatcher(metric.NotEqual, "label1", "x"),
|
|
},
|
|
expected: fingerprints,
|
|
},
|
|
{
|
|
matchers: metric.LabelMatchers{
|
|
newMatcher(metric.Equal, "all", "const"),
|
|
newMatcher(metric.NotEqual, "label1", "test_0"),
|
|
},
|
|
expected: fingerprints[10:],
|
|
},
|
|
{
|
|
matchers: metric.LabelMatchers{
|
|
newMatcher(metric.Equal, "all", "const"),
|
|
newMatcher(metric.NotEqual, "label1", "test_0"),
|
|
newMatcher(metric.NotEqual, "label1", "test_1"),
|
|
newMatcher(metric.NotEqual, "label1", "test_2"),
|
|
},
|
|
expected: fingerprints[30:],
|
|
},
|
|
{
|
|
matchers: metric.LabelMatchers{
|
|
newMatcher(metric.Equal, "label1", ""),
|
|
},
|
|
expected: fingerprints[:0],
|
|
},
|
|
{
|
|
matchers: metric.LabelMatchers{
|
|
newMatcher(metric.NotEqual, "label1", "test_0"),
|
|
newMatcher(metric.Equal, "label1", ""),
|
|
},
|
|
expected: fingerprints[:0],
|
|
},
|
|
{
|
|
matchers: metric.LabelMatchers{
|
|
newMatcher(metric.NotEqual, "label1", "test_0"),
|
|
newMatcher(metric.Equal, "label2", ""),
|
|
},
|
|
expected: fingerprints[:0],
|
|
},
|
|
{
|
|
matchers: metric.LabelMatchers{
|
|
newMatcher(metric.Equal, "all", "const"),
|
|
newMatcher(metric.NotEqual, "label1", "test_0"),
|
|
newMatcher(metric.Equal, "not_existant", ""),
|
|
},
|
|
expected: fingerprints[10:],
|
|
},
|
|
{
|
|
matchers: metric.LabelMatchers{
|
|
newMatcher(metric.RegexMatch, "label1", `test_[3-5]`),
|
|
},
|
|
expected: fingerprints[30:60],
|
|
},
|
|
{
|
|
matchers: metric.LabelMatchers{
|
|
newMatcher(metric.Equal, "all", "const"),
|
|
newMatcher(metric.RegexNoMatch, "label1", `test_[3-5]`),
|
|
},
|
|
expected: append(append(model.Fingerprints{}, fingerprints[:30]...), fingerprints[60:]...),
|
|
},
|
|
{
|
|
matchers: metric.LabelMatchers{
|
|
newMatcher(metric.RegexMatch, "label1", `test_[3-5]`),
|
|
newMatcher(metric.RegexMatch, "label2", `test_[4-6]`),
|
|
},
|
|
expected: fingerprints[35:60],
|
|
},
|
|
{
|
|
matchers: metric.LabelMatchers{
|
|
newMatcher(metric.RegexMatch, "label1", `test_[3-5]`),
|
|
newMatcher(metric.NotEqual, "label2", `test_4`),
|
|
},
|
|
expected: append(append(model.Fingerprints{}, fingerprints[30:35]...), fingerprints[45:60]...),
|
|
},
|
|
{
|
|
matchers: metric.LabelMatchers{
|
|
newMatcher(metric.Equal, "label1", `nonexistent`),
|
|
newMatcher(metric.RegexMatch, "label2", `test`),
|
|
},
|
|
expected: model.Fingerprints{},
|
|
},
|
|
{
|
|
matchers: metric.LabelMatchers{
|
|
newMatcher(metric.Equal, "label1", `test_0`),
|
|
newMatcher(metric.RegexMatch, "label2", `nonexistent`),
|
|
},
|
|
expected: model.Fingerprints{},
|
|
},
|
|
}
|
|
|
|
for _, mt := range matcherTests {
|
|
res := storage.MetricsForLabelMatchers(mt.matchers...)
|
|
if len(mt.expected) != len(res) {
|
|
t.Fatalf("expected %d matches for %q, found %d", len(mt.expected), mt.matchers, len(res))
|
|
}
|
|
for fp1 := range res {
|
|
found := false
|
|
for _, fp2 := range mt.expected {
|
|
if fp1 == fp2 {
|
|
found = true
|
|
break
|
|
}
|
|
}
|
|
if !found {
|
|
t.Errorf("expected fingerprint %s for %q not in result", fp1, mt.matchers)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestFingerprintsForLabels(t *testing.T) {
|
|
storage, closer := NewTestStorage(t, 1)
|
|
defer closer.Close()
|
|
|
|
samples := make([]*model.Sample, 100)
|
|
fingerprints := make(model.Fingerprints, 100)
|
|
|
|
for i := range samples {
|
|
metric := model.Metric{
|
|
model.MetricNameLabel: model.LabelValue(fmt.Sprintf("test_metric_%d", i)),
|
|
"label1": model.LabelValue(fmt.Sprintf("test_%d", i/10)),
|
|
"label2": model.LabelValue(fmt.Sprintf("test_%d", (i+5)/10)),
|
|
}
|
|
samples[i] = &model.Sample{
|
|
Metric: metric,
|
|
Timestamp: model.Time(i),
|
|
Value: model.SampleValue(i),
|
|
}
|
|
fingerprints[i] = metric.FastFingerprint()
|
|
}
|
|
for _, s := range samples {
|
|
storage.Append(s)
|
|
}
|
|
storage.WaitForIndexing()
|
|
|
|
var matcherTests = []struct {
|
|
pairs []model.LabelPair
|
|
expected model.Fingerprints
|
|
}{
|
|
{
|
|
pairs: []model.LabelPair{{"label1", "x"}},
|
|
expected: fingerprints[:0],
|
|
},
|
|
{
|
|
pairs: []model.LabelPair{{"label1", "test_0"}},
|
|
expected: fingerprints[:10],
|
|
},
|
|
{
|
|
pairs: []model.LabelPair{
|
|
{"label1", "test_0"},
|
|
{"label1", "test_1"},
|
|
},
|
|
expected: fingerprints[:0],
|
|
},
|
|
{
|
|
pairs: []model.LabelPair{
|
|
{"label1", "test_0"},
|
|
{"label2", "test_1"},
|
|
},
|
|
expected: fingerprints[5:10],
|
|
},
|
|
{
|
|
pairs: []model.LabelPair{
|
|
{"label1", "test_1"},
|
|
{"label2", "test_2"},
|
|
},
|
|
expected: fingerprints[15:20],
|
|
},
|
|
}
|
|
|
|
for _, mt := range matcherTests {
|
|
resfps := storage.fingerprintsForLabelPairs(mt.pairs...)
|
|
if len(mt.expected) != len(resfps) {
|
|
t.Fatalf("expected %d matches for %q, found %d", len(mt.expected), mt.pairs, len(resfps))
|
|
}
|
|
for fp1 := range resfps {
|
|
found := false
|
|
for _, fp2 := range mt.expected {
|
|
if fp1 == fp2 {
|
|
found = true
|
|
break
|
|
}
|
|
}
|
|
if !found {
|
|
t.Errorf("expected fingerprint %s for %q not in result", fp1, mt.pairs)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
var benchLabelMatchingRes map[model.Fingerprint]metric.Metric
|
|
|
|
func BenchmarkLabelMatching(b *testing.B) {
|
|
s, closer := NewTestStorage(b, 1)
|
|
defer closer.Close()
|
|
|
|
h := fnv.New64a()
|
|
lbl := func(x int) model.LabelValue {
|
|
h.Reset()
|
|
h.Write([]byte(fmt.Sprintf("%d", x)))
|
|
return model.LabelValue(fmt.Sprintf("%d", h.Sum64()))
|
|
}
|
|
|
|
M := 32
|
|
met := model.Metric{}
|
|
for i := 0; i < M; i++ {
|
|
met["label_a"] = lbl(i)
|
|
for j := 0; j < M; j++ {
|
|
met["label_b"] = lbl(j)
|
|
for k := 0; k < M; k++ {
|
|
met["label_c"] = lbl(k)
|
|
for l := 0; l < M; l++ {
|
|
met["label_d"] = lbl(l)
|
|
s.Append(&model.Sample{
|
|
Metric: met.Clone(),
|
|
Timestamp: 0,
|
|
Value: 1,
|
|
})
|
|
}
|
|
}
|
|
}
|
|
}
|
|
s.WaitForIndexing()
|
|
|
|
newMatcher := func(matchType metric.MatchType, name model.LabelName, value model.LabelValue) *metric.LabelMatcher {
|
|
lm, err := metric.NewLabelMatcher(matchType, name, value)
|
|
if err != nil {
|
|
b.Fatalf("error creating label matcher: %s", err)
|
|
}
|
|
return lm
|
|
}
|
|
|
|
var matcherTests = []metric.LabelMatchers{
|
|
{
|
|
newMatcher(metric.Equal, "label_a", lbl(1)),
|
|
},
|
|
{
|
|
newMatcher(metric.Equal, "label_a", lbl(3)),
|
|
newMatcher(metric.Equal, "label_c", lbl(3)),
|
|
},
|
|
{
|
|
newMatcher(metric.Equal, "label_a", lbl(3)),
|
|
newMatcher(metric.Equal, "label_c", lbl(3)),
|
|
newMatcher(metric.NotEqual, "label_d", lbl(3)),
|
|
},
|
|
{
|
|
newMatcher(metric.Equal, "label_a", lbl(3)),
|
|
newMatcher(metric.Equal, "label_b", lbl(3)),
|
|
newMatcher(metric.Equal, "label_c", lbl(3)),
|
|
newMatcher(metric.NotEqual, "label_d", lbl(3)),
|
|
},
|
|
{
|
|
newMatcher(metric.RegexMatch, "label_a", ".+"),
|
|
},
|
|
{
|
|
newMatcher(metric.Equal, "label_a", lbl(3)),
|
|
newMatcher(metric.RegexMatch, "label_a", ".+"),
|
|
},
|
|
{
|
|
newMatcher(metric.Equal, "label_a", lbl(1)),
|
|
newMatcher(metric.RegexMatch, "label_c", "("+lbl(3)+"|"+lbl(10)+")"),
|
|
},
|
|
{
|
|
newMatcher(metric.Equal, "label_a", lbl(3)),
|
|
newMatcher(metric.Equal, "label_a", lbl(4)),
|
|
newMatcher(metric.RegexMatch, "label_c", "("+lbl(3)+"|"+lbl(10)+")"),
|
|
},
|
|
}
|
|
|
|
b.ReportAllocs()
|
|
b.ResetTimer()
|
|
|
|
for i := 0; i < b.N; i++ {
|
|
benchLabelMatchingRes = map[model.Fingerprint]metric.Metric{}
|
|
for _, mt := range matcherTests {
|
|
benchLabelMatchingRes = s.MetricsForLabelMatchers(mt...)
|
|
}
|
|
}
|
|
// Stop timer to not count the storage closing.
|
|
b.StopTimer()
|
|
}
|
|
|
|
func TestRetentionCutoff(t *testing.T) {
|
|
now := model.Now()
|
|
insertStart := now.Add(-2 * time.Hour)
|
|
|
|
s, closer := NewTestStorage(t, 1)
|
|
defer closer.Close()
|
|
|
|
// Stop maintenance loop to prevent actual purging.
|
|
close(s.loopStopping)
|
|
<-s.loopStopped
|
|
<-s.logThrottlingStopped
|
|
// Recreate channel to avoid panic when we really shut down.
|
|
s.loopStopping = make(chan struct{})
|
|
|
|
s.dropAfter = 1 * time.Hour
|
|
|
|
for i := 0; i < 120; i++ {
|
|
smpl := &model.Sample{
|
|
Metric: model.Metric{"job": "test"},
|
|
Timestamp: insertStart.Add(time.Duration(i) * time.Minute), // 1 minute intervals.
|
|
Value: 1,
|
|
}
|
|
s.Append(smpl)
|
|
}
|
|
s.WaitForIndexing()
|
|
|
|
var fp model.Fingerprint
|
|
for f := range s.fingerprintsForLabelPairs(model.LabelPair{Name: "job", Value: "test"}) {
|
|
fp = f
|
|
break
|
|
}
|
|
|
|
pl := s.NewPreloader()
|
|
defer pl.Close()
|
|
|
|
// Preload everything.
|
|
it, err := pl.PreloadRange(fp, insertStart, now)
|
|
if err != nil {
|
|
t.Fatalf("Error preloading outdated chunks: %s", err)
|
|
}
|
|
|
|
val := it.ValueAtOrBeforeTime(now.Add(-61 * time.Minute))
|
|
if val.Timestamp != model.Earliest {
|
|
t.Errorf("unexpected result for timestamp before retention period")
|
|
}
|
|
|
|
vals := it.RangeValues(metric.Interval{OldestInclusive: insertStart, NewestInclusive: now})
|
|
// We get 59 values here because the model.Now() is slightly later
|
|
// than our now.
|
|
if len(vals) != 59 {
|
|
t.Errorf("expected 59 values but got %d", len(vals))
|
|
}
|
|
if expt := now.Add(-1 * time.Hour).Add(time.Minute); vals[0].Timestamp != expt {
|
|
t.Errorf("unexpected timestamp for first sample: %v, expected %v", vals[0].Timestamp.Time(), expt.Time())
|
|
}
|
|
|
|
vals = it.BoundaryValues(metric.Interval{OldestInclusive: insertStart, NewestInclusive: now})
|
|
if len(vals) != 2 {
|
|
t.Errorf("expected 2 values but got %d", len(vals))
|
|
}
|
|
if expt := now.Add(-1 * time.Hour).Add(time.Minute); vals[0].Timestamp != expt {
|
|
t.Errorf("unexpected timestamp for first sample: %v, expected %v", vals[0].Timestamp.Time(), expt.Time())
|
|
}
|
|
}
|
|
|
|
func TestDropMetrics(t *testing.T) {
|
|
now := model.Now()
|
|
insertStart := now.Add(-2 * time.Hour)
|
|
|
|
s, closer := NewTestStorage(t, 1)
|
|
defer closer.Close()
|
|
|
|
chunkFileExists := func(fp model.Fingerprint) (bool, error) {
|
|
f, err := s.persistence.openChunkFileForReading(fp)
|
|
if err == nil {
|
|
f.Close()
|
|
return true, nil
|
|
}
|
|
if os.IsNotExist(err) {
|
|
return false, nil
|
|
}
|
|
return false, err
|
|
}
|
|
|
|
m1 := model.Metric{model.MetricNameLabel: "test", "n1": "v1"}
|
|
m2 := model.Metric{model.MetricNameLabel: "test", "n1": "v2"}
|
|
m3 := model.Metric{model.MetricNameLabel: "test", "n1": "v3"}
|
|
|
|
N := 120000
|
|
|
|
for j, m := range []model.Metric{m1, m2, m3} {
|
|
for i := 0; i < N; i++ {
|
|
smpl := &model.Sample{
|
|
Metric: m,
|
|
Timestamp: insertStart.Add(time.Duration(i) * time.Millisecond), // 1 millisecond intervals.
|
|
Value: model.SampleValue(j),
|
|
}
|
|
s.Append(smpl)
|
|
}
|
|
}
|
|
s.WaitForIndexing()
|
|
|
|
// Archive m3, but first maintain it so that at least something is written to disk.
|
|
fpToBeArchived := m3.FastFingerprint()
|
|
s.maintainMemorySeries(fpToBeArchived, 0)
|
|
s.fpLocker.Lock(fpToBeArchived)
|
|
s.fpToSeries.del(fpToBeArchived)
|
|
if err := s.persistence.archiveMetric(
|
|
fpToBeArchived, m3, 0, insertStart.Add(time.Duration(N-1)*time.Millisecond),
|
|
); err != nil {
|
|
t.Error(err)
|
|
}
|
|
s.fpLocker.Unlock(fpToBeArchived)
|
|
|
|
fps := s.fingerprintsForLabelPairs(model.LabelPair{Name: model.MetricNameLabel, Value: "test"})
|
|
if len(fps) != 3 {
|
|
t.Errorf("unexpected number of fingerprints: %d", len(fps))
|
|
}
|
|
|
|
fpList := model.Fingerprints{m1.FastFingerprint(), m2.FastFingerprint(), fpToBeArchived}
|
|
|
|
s.DropMetricsForFingerprints(fpList[0])
|
|
s.WaitForIndexing()
|
|
|
|
fps2 := s.fingerprintsForLabelPairs(model.LabelPair{
|
|
Name: model.MetricNameLabel, Value: "test",
|
|
})
|
|
if len(fps2) != 2 {
|
|
t.Errorf("unexpected number of fingerprints: %d", len(fps2))
|
|
}
|
|
|
|
_, it, err := s.preloadChunksForRange(fpList[0], model.Earliest, model.Latest)
|
|
if err != nil {
|
|
t.Fatalf("Error preloading everything: %s", err)
|
|
}
|
|
if vals := it.RangeValues(metric.Interval{OldestInclusive: insertStart, NewestInclusive: now}); len(vals) != 0 {
|
|
t.Errorf("unexpected number of samples: %d", len(vals))
|
|
}
|
|
|
|
_, it, err = s.preloadChunksForRange(fpList[1], model.Earliest, model.Latest)
|
|
if err != nil {
|
|
t.Fatalf("Error preloading everything: %s", err)
|
|
}
|
|
if vals := it.RangeValues(metric.Interval{OldestInclusive: insertStart, NewestInclusive: now}); len(vals) != N {
|
|
t.Errorf("unexpected number of samples: %d", len(vals))
|
|
}
|
|
exists, err := chunkFileExists(fpList[2])
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if !exists {
|
|
t.Errorf("chunk file does not exist for fp=%v", fpList[2])
|
|
}
|
|
|
|
s.DropMetricsForFingerprints(fpList...)
|
|
s.WaitForIndexing()
|
|
|
|
fps3 := s.fingerprintsForLabelPairs(model.LabelPair{
|
|
Name: model.MetricNameLabel, Value: "test",
|
|
})
|
|
if len(fps3) != 0 {
|
|
t.Errorf("unexpected number of fingerprints: %d", len(fps3))
|
|
}
|
|
|
|
_, it, err = s.preloadChunksForRange(fpList[0], model.Earliest, model.Latest)
|
|
if err != nil {
|
|
t.Fatalf("Error preloading everything: %s", err)
|
|
}
|
|
if vals := it.RangeValues(metric.Interval{OldestInclusive: insertStart, NewestInclusive: now}); len(vals) != 0 {
|
|
t.Errorf("unexpected number of samples: %d", len(vals))
|
|
}
|
|
|
|
_, it, err = s.preloadChunksForRange(fpList[1], model.Earliest, model.Latest)
|
|
if err != nil {
|
|
t.Fatalf("Error preloading everything: %s", err)
|
|
}
|
|
if vals := it.RangeValues(metric.Interval{OldestInclusive: insertStart, NewestInclusive: now}); len(vals) != 0 {
|
|
t.Errorf("unexpected number of samples: %d", len(vals))
|
|
}
|
|
exists, err = chunkFileExists(fpList[2])
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if exists {
|
|
t.Errorf("chunk file still exists for fp=%v", fpList[2])
|
|
}
|
|
}
|
|
|
|
// TestLoop is just a smoke test for the loop method, if we can switch it on and
|
|
// off without disaster.
|
|
func TestLoop(t *testing.T) {
|
|
if testing.Short() {
|
|
t.Skip("Skipping test in short mode.")
|
|
}
|
|
samples := make(model.Samples, 1000)
|
|
for i := range samples {
|
|
samples[i] = &model.Sample{
|
|
Timestamp: model.Time(2 * i),
|
|
Value: model.SampleValue(float64(i) * 0.2),
|
|
}
|
|
}
|
|
directory := testutil.NewTemporaryDirectory("test_storage", t)
|
|
defer directory.Close()
|
|
o := &MemorySeriesStorageOptions{
|
|
MemoryChunks: 50,
|
|
MaxChunksToPersist: 1000000,
|
|
PersistenceRetentionPeriod: 24 * 7 * time.Hour,
|
|
PersistenceStoragePath: directory.Path(),
|
|
CheckpointInterval: 250 * time.Millisecond,
|
|
SyncStrategy: Adaptive,
|
|
MinShrinkRatio: 0.1,
|
|
}
|
|
storage := NewMemorySeriesStorage(o)
|
|
if err := storage.Start(); err != nil {
|
|
t.Errorf("Error starting storage: %s", err)
|
|
}
|
|
for _, s := range samples {
|
|
storage.Append(s)
|
|
}
|
|
storage.WaitForIndexing()
|
|
series, _ := storage.(*memorySeriesStorage).fpToSeries.get(model.Metric{}.FastFingerprint())
|
|
cdsBefore := len(series.chunkDescs)
|
|
time.Sleep(fpMaxWaitDuration + time.Second) // TODO(beorn7): Ugh, need to wait for maintenance to kick in.
|
|
cdsAfter := len(series.chunkDescs)
|
|
storage.Stop()
|
|
if cdsBefore <= cdsAfter {
|
|
t.Errorf(
|
|
"Number of chunk descriptors should have gone down by now. Got before %d, after %d.",
|
|
cdsBefore, cdsAfter,
|
|
)
|
|
}
|
|
}
|
|
|
|
func testChunk(t *testing.T, encoding chunkEncoding) {
|
|
samples := make(model.Samples, 500000)
|
|
for i := range samples {
|
|
samples[i] = &model.Sample{
|
|
Timestamp: model.Time(i),
|
|
Value: model.SampleValue(float64(i) * 0.2),
|
|
}
|
|
}
|
|
s, closer := NewTestStorage(t, encoding)
|
|
defer closer.Close()
|
|
|
|
for _, sample := range samples {
|
|
s.Append(sample)
|
|
}
|
|
s.WaitForIndexing()
|
|
|
|
for m := range s.fpToSeries.iter() {
|
|
s.fpLocker.Lock(m.fp)
|
|
|
|
var values []model.SamplePair
|
|
for _, cd := range m.series.chunkDescs {
|
|
if cd.isEvicted() {
|
|
continue
|
|
}
|
|
for sample := range cd.c.newIterator().values() {
|
|
values = append(values, *sample)
|
|
}
|
|
}
|
|
|
|
for i, v := range values {
|
|
if samples[i].Timestamp != v.Timestamp {
|
|
t.Errorf("%d. Got %v; want %v", i, v.Timestamp, samples[i].Timestamp)
|
|
}
|
|
if samples[i].Value != v.Value {
|
|
t.Errorf("%d. Got %v; want %v", i, v.Value, samples[i].Value)
|
|
}
|
|
}
|
|
s.fpLocker.Unlock(m.fp)
|
|
}
|
|
log.Info("test done, closing")
|
|
}
|
|
|
|
func TestChunkType0(t *testing.T) {
|
|
testChunk(t, 0)
|
|
}
|
|
|
|
func TestChunkType1(t *testing.T) {
|
|
testChunk(t, 1)
|
|
}
|
|
|
|
func testValueAtOrBeforeTime(t *testing.T, encoding chunkEncoding) {
|
|
samples := make(model.Samples, 10000)
|
|
for i := range samples {
|
|
samples[i] = &model.Sample{
|
|
Timestamp: model.Time(2 * i),
|
|
Value: model.SampleValue(float64(i) * 0.2),
|
|
}
|
|
}
|
|
s, closer := NewTestStorage(t, encoding)
|
|
defer closer.Close()
|
|
|
|
for _, sample := range samples {
|
|
s.Append(sample)
|
|
}
|
|
s.WaitForIndexing()
|
|
|
|
fp := model.Metric{}.FastFingerprint()
|
|
|
|
_, it, err := s.preloadChunksForRange(fp, model.Earliest, model.Latest)
|
|
if err != nil {
|
|
t.Fatalf("Error preloading everything: %s", err)
|
|
}
|
|
|
|
// #1 Exactly on a sample.
|
|
for i, expected := range samples {
|
|
actual := it.ValueAtOrBeforeTime(expected.Timestamp)
|
|
|
|
if expected.Timestamp != actual.Timestamp {
|
|
t.Errorf("1.%d. Got %v; want %v", i, actual.Timestamp, expected.Timestamp)
|
|
}
|
|
if expected.Value != actual.Value {
|
|
t.Errorf("1.%d. Got %v; want %v", i, actual.Value, expected.Value)
|
|
}
|
|
}
|
|
|
|
// #2 Between samples.
|
|
for i, expected := range samples {
|
|
if i == len(samples)-1 {
|
|
continue
|
|
}
|
|
actual := it.ValueAtOrBeforeTime(expected.Timestamp + 1)
|
|
|
|
if expected.Timestamp != actual.Timestamp {
|
|
t.Errorf("2.%d. Got %v; want %v", i, actual.Timestamp, expected.Timestamp)
|
|
}
|
|
if expected.Value != actual.Value {
|
|
t.Errorf("2.%d. Got %v; want %v", i, actual.Value, expected.Value)
|
|
}
|
|
}
|
|
|
|
// #3 Corner cases: Just before the first sample, just after the last.
|
|
expected := &model.Sample{Timestamp: model.Earliest}
|
|
actual := it.ValueAtOrBeforeTime(samples[0].Timestamp - 1)
|
|
if expected.Timestamp != actual.Timestamp {
|
|
t.Errorf("3.1. Got %v; want %v", actual.Timestamp, expected.Timestamp)
|
|
}
|
|
if expected.Value != actual.Value {
|
|
t.Errorf("3.1. Got %v; want %v", actual.Value, expected.Value)
|
|
}
|
|
expected = samples[len(samples)-1]
|
|
actual = it.ValueAtOrBeforeTime(expected.Timestamp + 1)
|
|
if expected.Timestamp != actual.Timestamp {
|
|
t.Errorf("3.2. Got %v; want %v", actual.Timestamp, expected.Timestamp)
|
|
}
|
|
if expected.Value != actual.Value {
|
|
t.Errorf("3.2. Got %v; want %v", actual.Value, expected.Value)
|
|
}
|
|
}
|
|
|
|
func TestValueAtTimeChunkType0(t *testing.T) {
|
|
testValueAtOrBeforeTime(t, 0)
|
|
}
|
|
|
|
func TestValueAtTimeChunkType1(t *testing.T) {
|
|
testValueAtOrBeforeTime(t, 1)
|
|
}
|
|
|
|
func benchmarkValueAtOrBeforeTime(b *testing.B, encoding chunkEncoding) {
|
|
samples := make(model.Samples, 10000)
|
|
for i := range samples {
|
|
samples[i] = &model.Sample{
|
|
Timestamp: model.Time(2 * i),
|
|
Value: model.SampleValue(float64(i) * 0.2),
|
|
}
|
|
}
|
|
s, closer := NewTestStorage(b, encoding)
|
|
defer closer.Close()
|
|
|
|
for _, sample := range samples {
|
|
s.Append(sample)
|
|
}
|
|
s.WaitForIndexing()
|
|
|
|
fp := model.Metric{}.FastFingerprint()
|
|
|
|
_, it, err := s.preloadChunksForRange(fp, model.Earliest, model.Latest)
|
|
if err != nil {
|
|
b.Fatalf("Error preloading everything: %s", err)
|
|
}
|
|
|
|
b.ResetTimer()
|
|
|
|
for i := 0; i < b.N; i++ {
|
|
// #1 Exactly on a sample.
|
|
for i, expected := range samples {
|
|
actual := it.ValueAtOrBeforeTime(expected.Timestamp)
|
|
|
|
if expected.Timestamp != actual.Timestamp {
|
|
b.Errorf("1.%d. Got %v; want %v", i, actual.Timestamp, expected.Timestamp)
|
|
}
|
|
if expected.Value != actual.Value {
|
|
b.Errorf("1.%d. Got %v; want %v", i, actual.Value, expected.Value)
|
|
}
|
|
}
|
|
|
|
// #2 Between samples.
|
|
for i, expected := range samples {
|
|
if i == len(samples)-1 {
|
|
continue
|
|
}
|
|
actual := it.ValueAtOrBeforeTime(expected.Timestamp + 1)
|
|
|
|
if expected.Timestamp != actual.Timestamp {
|
|
b.Errorf("2.%d. Got %v; want %v", i, actual.Timestamp, expected.Timestamp)
|
|
}
|
|
if expected.Value != actual.Value {
|
|
b.Errorf("2.%d. Got %v; want %v", i, actual.Value, expected.Value)
|
|
}
|
|
}
|
|
|
|
// #3 Corner cases: Just before the first sample, just after the last.
|
|
expected := &model.Sample{Timestamp: model.Earliest}
|
|
actual := it.ValueAtOrBeforeTime(samples[0].Timestamp - 1)
|
|
if expected.Timestamp != actual.Timestamp {
|
|
b.Errorf("3.1. Got %v; want %v", actual.Timestamp, expected.Timestamp)
|
|
}
|
|
if expected.Value != actual.Value {
|
|
b.Errorf("3.1. Got %v; want %v", actual.Value, expected.Value)
|
|
}
|
|
expected = samples[len(samples)-1]
|
|
actual = it.ValueAtOrBeforeTime(expected.Timestamp + 1)
|
|
if expected.Timestamp != actual.Timestamp {
|
|
b.Errorf("3.2. Got %v; want %v", actual.Timestamp, expected.Timestamp)
|
|
}
|
|
if expected.Value != actual.Value {
|
|
b.Errorf("3.2. Got %v; want %v", actual.Value, expected.Value)
|
|
}
|
|
}
|
|
}
|
|
|
|
func BenchmarkValueAtOrBeforeTimeChunkType0(b *testing.B) {
|
|
benchmarkValueAtOrBeforeTime(b, 0)
|
|
}
|
|
|
|
func BenchmarkValueAtTimeChunkType1(b *testing.B) {
|
|
benchmarkValueAtOrBeforeTime(b, 1)
|
|
}
|
|
|
|
func testRangeValues(t *testing.T, encoding chunkEncoding) {
|
|
samples := make(model.Samples, 10000)
|
|
for i := range samples {
|
|
samples[i] = &model.Sample{
|
|
Timestamp: model.Time(2 * i),
|
|
Value: model.SampleValue(float64(i) * 0.2),
|
|
}
|
|
}
|
|
s, closer := NewTestStorage(t, encoding)
|
|
defer closer.Close()
|
|
|
|
for _, sample := range samples {
|
|
s.Append(sample)
|
|
}
|
|
s.WaitForIndexing()
|
|
|
|
fp := model.Metric{}.FastFingerprint()
|
|
|
|
_, it, err := s.preloadChunksForRange(fp, model.Earliest, model.Latest)
|
|
if err != nil {
|
|
t.Fatalf("Error preloading everything: %s", err)
|
|
}
|
|
|
|
// #1 Zero length interval at sample.
|
|
for i, expected := range samples {
|
|
actual := it.RangeValues(metric.Interval{
|
|
OldestInclusive: expected.Timestamp,
|
|
NewestInclusive: expected.Timestamp,
|
|
})
|
|
|
|
if len(actual) != 1 {
|
|
t.Fatalf("1.%d. Expected exactly one result, got %d.", i, len(actual))
|
|
}
|
|
if expected.Timestamp != actual[0].Timestamp {
|
|
t.Errorf("1.%d. Got %v; want %v.", i, actual[0].Timestamp, expected.Timestamp)
|
|
}
|
|
if expected.Value != actual[0].Value {
|
|
t.Errorf("1.%d. Got %v; want %v.", i, actual[0].Value, expected.Value)
|
|
}
|
|
}
|
|
|
|
// #2 Zero length interval off sample.
|
|
for i, expected := range samples {
|
|
actual := it.RangeValues(metric.Interval{
|
|
OldestInclusive: expected.Timestamp + 1,
|
|
NewestInclusive: expected.Timestamp + 1,
|
|
})
|
|
|
|
if len(actual) != 0 {
|
|
t.Fatalf("2.%d. Expected no result, got %d.", i, len(actual))
|
|
}
|
|
}
|
|
|
|
// #3 2sec interval around sample.
|
|
for i, expected := range samples {
|
|
actual := it.RangeValues(metric.Interval{
|
|
OldestInclusive: expected.Timestamp - 1,
|
|
NewestInclusive: expected.Timestamp + 1,
|
|
})
|
|
|
|
if len(actual) != 1 {
|
|
t.Fatalf("3.%d. Expected exactly one result, got %d.", i, len(actual))
|
|
}
|
|
if expected.Timestamp != actual[0].Timestamp {
|
|
t.Errorf("3.%d. Got %v; want %v.", i, actual[0].Timestamp, expected.Timestamp)
|
|
}
|
|
if expected.Value != actual[0].Value {
|
|
t.Errorf("3.%d. Got %v; want %v.", i, actual[0].Value, expected.Value)
|
|
}
|
|
}
|
|
|
|
// #4 2sec interval sample to sample.
|
|
for i, expected1 := range samples {
|
|
if i == len(samples)-1 {
|
|
continue
|
|
}
|
|
expected2 := samples[i+1]
|
|
actual := it.RangeValues(metric.Interval{
|
|
OldestInclusive: expected1.Timestamp,
|
|
NewestInclusive: expected1.Timestamp + 2,
|
|
})
|
|
|
|
if len(actual) != 2 {
|
|
t.Fatalf("4.%d. Expected exactly 2 results, got %d.", i, len(actual))
|
|
}
|
|
if expected1.Timestamp != actual[0].Timestamp {
|
|
t.Errorf("4.%d. Got %v for 1st result; want %v.", i, actual[0].Timestamp, expected1.Timestamp)
|
|
}
|
|
if expected1.Value != actual[0].Value {
|
|
t.Errorf("4.%d. Got %v for 1st result; want %v.", i, actual[0].Value, expected1.Value)
|
|
}
|
|
if expected2.Timestamp != actual[1].Timestamp {
|
|
t.Errorf("4.%d. Got %v for 2nd result; want %v.", i, actual[1].Timestamp, expected2.Timestamp)
|
|
}
|
|
if expected2.Value != actual[1].Value {
|
|
t.Errorf("4.%d. Got %v for 2nd result; want %v.", i, actual[1].Value, expected2.Value)
|
|
}
|
|
}
|
|
|
|
// #5 corner cases: Interval ends at first sample, interval starts
|
|
// at last sample, interval entirely before/after samples.
|
|
expected := samples[0]
|
|
actual := it.RangeValues(metric.Interval{
|
|
OldestInclusive: expected.Timestamp - 2,
|
|
NewestInclusive: expected.Timestamp,
|
|
})
|
|
if len(actual) != 1 {
|
|
t.Fatalf("5.1. Expected exactly one result, got %d.", len(actual))
|
|
}
|
|
if expected.Timestamp != actual[0].Timestamp {
|
|
t.Errorf("5.1. Got %v; want %v.", actual[0].Timestamp, expected.Timestamp)
|
|
}
|
|
if expected.Value != actual[0].Value {
|
|
t.Errorf("5.1. Got %v; want %v.", actual[0].Value, expected.Value)
|
|
}
|
|
expected = samples[len(samples)-1]
|
|
actual = it.RangeValues(metric.Interval{
|
|
OldestInclusive: expected.Timestamp,
|
|
NewestInclusive: expected.Timestamp + 2,
|
|
})
|
|
if len(actual) != 1 {
|
|
t.Fatalf("5.2. Expected exactly one result, got %d.", len(actual))
|
|
}
|
|
if expected.Timestamp != actual[0].Timestamp {
|
|
t.Errorf("5.2. Got %v; want %v.", actual[0].Timestamp, expected.Timestamp)
|
|
}
|
|
if expected.Value != actual[0].Value {
|
|
t.Errorf("5.2. Got %v; want %v.", actual[0].Value, expected.Value)
|
|
}
|
|
firstSample := samples[0]
|
|
actual = it.RangeValues(metric.Interval{
|
|
OldestInclusive: firstSample.Timestamp - 4,
|
|
NewestInclusive: firstSample.Timestamp - 2,
|
|
})
|
|
if len(actual) != 0 {
|
|
t.Fatalf("5.3. Expected no results, got %d.", len(actual))
|
|
}
|
|
lastSample := samples[len(samples)-1]
|
|
actual = it.RangeValues(metric.Interval{
|
|
OldestInclusive: lastSample.Timestamp + 2,
|
|
NewestInclusive: lastSample.Timestamp + 4,
|
|
})
|
|
if len(actual) != 0 {
|
|
t.Fatalf("5.3. Expected no results, got %d.", len(actual))
|
|
}
|
|
}
|
|
|
|
func TestRangeValuesChunkType0(t *testing.T) {
|
|
testRangeValues(t, 0)
|
|
}
|
|
|
|
func TestRangeValuesChunkType1(t *testing.T) {
|
|
testRangeValues(t, 1)
|
|
}
|
|
|
|
func benchmarkRangeValues(b *testing.B, encoding chunkEncoding) {
|
|
samples := make(model.Samples, 10000)
|
|
for i := range samples {
|
|
samples[i] = &model.Sample{
|
|
Timestamp: model.Time(2 * i),
|
|
Value: model.SampleValue(float64(i) * 0.2),
|
|
}
|
|
}
|
|
s, closer := NewTestStorage(b, encoding)
|
|
defer closer.Close()
|
|
|
|
for _, sample := range samples {
|
|
s.Append(sample)
|
|
}
|
|
s.WaitForIndexing()
|
|
|
|
fp := model.Metric{}.FastFingerprint()
|
|
|
|
_, it, err := s.preloadChunksForRange(fp, model.Earliest, model.Latest)
|
|
if err != nil {
|
|
b.Fatalf("Error preloading everything: %s", err)
|
|
}
|
|
|
|
b.ResetTimer()
|
|
|
|
for i := 0; i < b.N; i++ {
|
|
for _, sample := range samples {
|
|
actual := it.RangeValues(metric.Interval{
|
|
OldestInclusive: sample.Timestamp - 20,
|
|
NewestInclusive: sample.Timestamp + 20,
|
|
})
|
|
|
|
if len(actual) < 10 {
|
|
b.Fatalf("not enough samples found")
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func BenchmarkRangeValuesChunkType0(b *testing.B) {
|
|
benchmarkRangeValues(b, 0)
|
|
}
|
|
|
|
func BenchmarkRangeValuesChunkType1(b *testing.B) {
|
|
benchmarkRangeValues(b, 1)
|
|
}
|
|
|
|
func testEvictAndPurgeSeries(t *testing.T, encoding chunkEncoding) {
|
|
samples := make(model.Samples, 10000)
|
|
for i := range samples {
|
|
samples[i] = &model.Sample{
|
|
Timestamp: model.Time(2 * i),
|
|
Value: model.SampleValue(float64(i * i)),
|
|
}
|
|
}
|
|
s, closer := NewTestStorage(t, encoding)
|
|
defer closer.Close()
|
|
|
|
for _, sample := range samples {
|
|
s.Append(sample)
|
|
}
|
|
s.WaitForIndexing()
|
|
|
|
fp := model.Metric{}.FastFingerprint()
|
|
|
|
// Drop ~half of the chunks.
|
|
s.maintainMemorySeries(fp, 10000)
|
|
_, it, err := s.preloadChunksForRange(fp, model.Earliest, model.Latest)
|
|
if err != nil {
|
|
t.Fatalf("Error preloading everything: %s", err)
|
|
}
|
|
actual := it.BoundaryValues(metric.Interval{
|
|
OldestInclusive: 0,
|
|
NewestInclusive: 100000,
|
|
})
|
|
if len(actual) != 2 {
|
|
t.Fatal("expected two results after purging half of series")
|
|
}
|
|
if actual[0].Timestamp < 6000 || actual[0].Timestamp > 10000 {
|
|
t.Errorf("1st timestamp out of expected range: %v", actual[0].Timestamp)
|
|
}
|
|
want := model.Time(19998)
|
|
if actual[1].Timestamp != want {
|
|
t.Errorf("2nd timestamp: want %v, got %v", want, actual[1].Timestamp)
|
|
}
|
|
|
|
// Drop everything.
|
|
s.maintainMemorySeries(fp, 100000)
|
|
_, it, err = s.preloadChunksForRange(fp, model.Earliest, model.Latest)
|
|
if err != nil {
|
|
t.Fatalf("Error preloading everything: %s", err)
|
|
}
|
|
actual = it.BoundaryValues(metric.Interval{
|
|
OldestInclusive: 0,
|
|
NewestInclusive: 100000,
|
|
})
|
|
if len(actual) != 0 {
|
|
t.Fatal("expected zero results after purging the whole series")
|
|
}
|
|
|
|
// Recreate series.
|
|
for _, sample := range samples {
|
|
s.Append(sample)
|
|
}
|
|
s.WaitForIndexing()
|
|
|
|
series, ok := s.fpToSeries.get(fp)
|
|
if !ok {
|
|
t.Fatal("could not find series")
|
|
}
|
|
|
|
// Persist head chunk so we can safely archive.
|
|
series.headChunkClosed = true
|
|
s.maintainMemorySeries(fp, model.Earliest)
|
|
|
|
// Archive metrics.
|
|
s.fpToSeries.del(fp)
|
|
if err := s.persistence.archiveMetric(
|
|
fp, series.metric, series.firstTime(), series.head().lastTime(),
|
|
); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
archived, _, _, err := s.persistence.hasArchivedMetric(fp)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if !archived {
|
|
t.Fatal("not archived")
|
|
}
|
|
|
|
// Drop ~half of the chunks of an archived series.
|
|
s.maintainArchivedSeries(fp, 10000)
|
|
archived, _, _, err = s.persistence.hasArchivedMetric(fp)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if !archived {
|
|
t.Fatal("archived series purged although only half of the chunks dropped")
|
|
}
|
|
|
|
// Drop everything.
|
|
s.maintainArchivedSeries(fp, 100000)
|
|
archived, _, _, err = s.persistence.hasArchivedMetric(fp)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if archived {
|
|
t.Fatal("archived series not dropped")
|
|
}
|
|
|
|
// Recreate series.
|
|
for _, sample := range samples {
|
|
s.Append(sample)
|
|
}
|
|
s.WaitForIndexing()
|
|
|
|
series, ok = s.fpToSeries.get(fp)
|
|
if !ok {
|
|
t.Fatal("could not find series")
|
|
}
|
|
|
|
// Persist head chunk so we can safely archive.
|
|
series.headChunkClosed = true
|
|
s.maintainMemorySeries(fp, model.Earliest)
|
|
|
|
// Archive metrics.
|
|
s.fpToSeries.del(fp)
|
|
if err := s.persistence.archiveMetric(
|
|
fp, series.metric, series.firstTime(), series.head().lastTime(),
|
|
); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
archived, _, _, err = s.persistence.hasArchivedMetric(fp)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if !archived {
|
|
t.Fatal("not archived")
|
|
}
|
|
|
|
// Unarchive metrics.
|
|
s.getOrCreateSeries(fp, model.Metric{})
|
|
|
|
series, ok = s.fpToSeries.get(fp)
|
|
if !ok {
|
|
t.Fatal("could not find series")
|
|
}
|
|
archived, _, _, err = s.persistence.hasArchivedMetric(fp)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if archived {
|
|
t.Fatal("archived")
|
|
}
|
|
|
|
// This will archive again, but must not drop it completely, despite the
|
|
// memorySeries being empty.
|
|
s.maintainMemorySeries(fp, 10000)
|
|
archived, _, _, err = s.persistence.hasArchivedMetric(fp)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if !archived {
|
|
t.Fatal("series purged completely")
|
|
}
|
|
}
|
|
|
|
func TestEvictAndPurgeSeriesChunkType0(t *testing.T) {
|
|
testEvictAndPurgeSeries(t, 0)
|
|
}
|
|
|
|
func TestEvictAndPurgeSeriesChunkType1(t *testing.T) {
|
|
testEvictAndPurgeSeries(t, 1)
|
|
}
|
|
|
|
func testEvictAndLoadChunkDescs(t *testing.T, encoding chunkEncoding) {
|
|
samples := make(model.Samples, 10000)
|
|
for i := range samples {
|
|
samples[i] = &model.Sample{
|
|
Timestamp: model.Time(2 * i),
|
|
Value: model.SampleValue(float64(i * i)),
|
|
}
|
|
}
|
|
// Give last sample a timestamp of now so that the head chunk will not
|
|
// be closed (which would then archive the time series later as
|
|
// everything will get evicted).
|
|
samples[len(samples)-1] = &model.Sample{
|
|
Timestamp: model.Now(),
|
|
Value: model.SampleValue(3.14),
|
|
}
|
|
|
|
s, closer := NewTestStorage(t, encoding)
|
|
defer closer.Close()
|
|
|
|
// Adjust memory chunks to lower value to see evictions.
|
|
s.maxMemoryChunks = 1
|
|
|
|
for _, sample := range samples {
|
|
s.Append(sample)
|
|
}
|
|
s.WaitForIndexing()
|
|
|
|
fp := model.Metric{}.FastFingerprint()
|
|
|
|
series, ok := s.fpToSeries.get(fp)
|
|
if !ok {
|
|
t.Fatal("could not find series")
|
|
}
|
|
|
|
oldLen := len(series.chunkDescs)
|
|
// Maintain series without any dropped chunks.
|
|
s.maintainMemorySeries(fp, 0)
|
|
// Give the evict goroutine an opportunity to run.
|
|
time.Sleep(50 * time.Millisecond)
|
|
// Maintain series again to trigger chunkDesc eviction
|
|
s.maintainMemorySeries(fp, 0)
|
|
|
|
if oldLen <= len(series.chunkDescs) {
|
|
t.Errorf("Expected number of chunkDescs to decrease, old number %d, current number %d.", oldLen, len(series.chunkDescs))
|
|
}
|
|
|
|
// Load everything back.
|
|
p := s.NewPreloader()
|
|
p.PreloadRange(fp, 0, 100000)
|
|
|
|
if oldLen != len(series.chunkDescs) {
|
|
t.Errorf("Expected number of chunkDescs to have reached old value again, old number %d, current number %d.", oldLen, len(series.chunkDescs))
|
|
}
|
|
|
|
p.Close()
|
|
|
|
// Now maintain series with drops to make sure nothing crazy happens.
|
|
s.maintainMemorySeries(fp, 100000)
|
|
|
|
if len(series.chunkDescs) != 1 {
|
|
t.Errorf("Expected exactly one chunkDesc left, got %d.", len(series.chunkDescs))
|
|
}
|
|
}
|
|
|
|
func TestEvictAndLoadChunkDescsType0(t *testing.T) {
|
|
testEvictAndLoadChunkDescs(t, 0)
|
|
}
|
|
|
|
func TestEvictAndLoadChunkDescsType1(t *testing.T) {
|
|
testEvictAndLoadChunkDescs(t, 1)
|
|
}
|
|
|
|
func benchmarkAppend(b *testing.B, encoding chunkEncoding) {
|
|
samples := make(model.Samples, b.N)
|
|
for i := range samples {
|
|
samples[i] = &model.Sample{
|
|
Metric: model.Metric{
|
|
model.MetricNameLabel: model.LabelValue(fmt.Sprintf("test_metric_%d", i%10)),
|
|
"label1": model.LabelValue(fmt.Sprintf("test_metric_%d", i%10)),
|
|
"label2": model.LabelValue(fmt.Sprintf("test_metric_%d", i%10)),
|
|
},
|
|
Timestamp: model.Time(i),
|
|
Value: model.SampleValue(i),
|
|
}
|
|
}
|
|
b.ResetTimer()
|
|
s, closer := NewTestStorage(b, encoding)
|
|
defer closer.Close()
|
|
|
|
for _, sample := range samples {
|
|
s.Append(sample)
|
|
}
|
|
}
|
|
|
|
func BenchmarkAppendType0(b *testing.B) {
|
|
benchmarkAppend(b, 0)
|
|
}
|
|
|
|
func BenchmarkAppendType1(b *testing.B) {
|
|
benchmarkAppend(b, 1)
|
|
}
|
|
|
|
// Append a large number of random samples and then check if we can get them out
|
|
// of the storage alright.
|
|
func testFuzz(t *testing.T, encoding chunkEncoding) {
|
|
if testing.Short() {
|
|
t.Skip("Skipping test in short mode.")
|
|
}
|
|
|
|
check := func(seed int64) bool {
|
|
rand.Seed(seed)
|
|
s, c := NewTestStorage(t, encoding)
|
|
defer c.Close()
|
|
|
|
samples := createRandomSamples("test_fuzz", 10000)
|
|
for _, sample := range samples {
|
|
s.Append(sample)
|
|
}
|
|
return verifyStorage(t, s, samples, 24*7*time.Hour)
|
|
}
|
|
|
|
if err := quick.Check(check, nil); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
func TestFuzzChunkType0(t *testing.T) {
|
|
testFuzz(t, 0)
|
|
}
|
|
|
|
func TestFuzzChunkType1(t *testing.T) {
|
|
testFuzz(t, 1)
|
|
}
|
|
|
|
// benchmarkFuzz is the benchmark version of testFuzz. The storage options are
|
|
// set such that evictions, checkpoints, and purging will happen concurrently,
|
|
// too. This benchmark will have a very long runtime (up to minutes). You can
|
|
// use it as an actual benchmark. Run it like this:
|
|
//
|
|
// go test -cpu 1,2,4,8 -run=NONE -bench BenchmarkFuzzChunkType -benchmem
|
|
//
|
|
// You can also use it as a test for races. In that case, run it like this (will
|
|
// make things even slower):
|
|
//
|
|
// go test -race -cpu 8 -short -bench BenchmarkFuzzChunkType
|
|
func benchmarkFuzz(b *testing.B, encoding chunkEncoding) {
|
|
DefaultChunkEncoding = encoding
|
|
const samplesPerRun = 100000
|
|
rand.Seed(42)
|
|
directory := testutil.NewTemporaryDirectory("test_storage", b)
|
|
defer directory.Close()
|
|
o := &MemorySeriesStorageOptions{
|
|
MemoryChunks: 100,
|
|
MaxChunksToPersist: 1000000,
|
|
PersistenceRetentionPeriod: time.Hour,
|
|
PersistenceStoragePath: directory.Path(),
|
|
CheckpointInterval: time.Second,
|
|
SyncStrategy: Adaptive,
|
|
MinShrinkRatio: 0.1,
|
|
}
|
|
s := NewMemorySeriesStorage(o)
|
|
if err := s.Start(); err != nil {
|
|
b.Fatalf("Error starting storage: %s", err)
|
|
}
|
|
s.Start()
|
|
defer s.Stop()
|
|
|
|
samples := createRandomSamples("benchmark_fuzz", samplesPerRun*b.N)
|
|
|
|
b.ResetTimer()
|
|
|
|
for i := 0; i < b.N; i++ {
|
|
start := samplesPerRun * i
|
|
end := samplesPerRun * (i + 1)
|
|
middle := (start + end) / 2
|
|
for _, sample := range samples[start:middle] {
|
|
s.Append(sample)
|
|
}
|
|
verifyStorage(b, s.(*memorySeriesStorage), samples[:middle], o.PersistenceRetentionPeriod)
|
|
for _, sample := range samples[middle:end] {
|
|
s.Append(sample)
|
|
}
|
|
verifyStorage(b, s.(*memorySeriesStorage), samples[:end], o.PersistenceRetentionPeriod)
|
|
}
|
|
}
|
|
|
|
func BenchmarkFuzzChunkType0(b *testing.B) {
|
|
benchmarkFuzz(b, 0)
|
|
}
|
|
|
|
func BenchmarkFuzzChunkType1(b *testing.B) {
|
|
benchmarkFuzz(b, 1)
|
|
}
|
|
|
|
func createRandomSamples(metricName string, minLen int) model.Samples {
|
|
type valueCreator func() model.SampleValue
|
|
type deltaApplier func(model.SampleValue) model.SampleValue
|
|
|
|
var (
|
|
maxMetrics = 5
|
|
maxStreakLength = 500
|
|
maxTimeDelta = 10000
|
|
maxTimeDeltaFactor = 10
|
|
timestamp = model.Now() - model.Time(maxTimeDelta*maxTimeDeltaFactor*minLen/4) // So that some timestamps are in the future.
|
|
generators = []struct {
|
|
createValue valueCreator
|
|
applyDelta []deltaApplier
|
|
}{
|
|
{ // "Boolean".
|
|
createValue: func() model.SampleValue {
|
|
return model.SampleValue(rand.Intn(2))
|
|
},
|
|
applyDelta: []deltaApplier{
|
|
func(_ model.SampleValue) model.SampleValue {
|
|
return model.SampleValue(rand.Intn(2))
|
|
},
|
|
},
|
|
},
|
|
{ // Integer with int deltas of various byte length.
|
|
createValue: func() model.SampleValue {
|
|
return model.SampleValue(rand.Int63() - 1<<62)
|
|
},
|
|
applyDelta: []deltaApplier{
|
|
func(v model.SampleValue) model.SampleValue {
|
|
return model.SampleValue(rand.Intn(1<<8) - 1<<7 + int(v))
|
|
},
|
|
func(v model.SampleValue) model.SampleValue {
|
|
return model.SampleValue(rand.Intn(1<<16) - 1<<15 + int(v))
|
|
},
|
|
func(v model.SampleValue) model.SampleValue {
|
|
return model.SampleValue(rand.Int63n(1<<32) - 1<<31 + int64(v))
|
|
},
|
|
},
|
|
},
|
|
{ // Float with float32 and float64 deltas.
|
|
createValue: func() model.SampleValue {
|
|
return model.SampleValue(rand.NormFloat64())
|
|
},
|
|
applyDelta: []deltaApplier{
|
|
func(v model.SampleValue) model.SampleValue {
|
|
return v + model.SampleValue(float32(rand.NormFloat64()))
|
|
},
|
|
func(v model.SampleValue) model.SampleValue {
|
|
return v + model.SampleValue(rand.NormFloat64())
|
|
},
|
|
},
|
|
},
|
|
}
|
|
)
|
|
|
|
// Prefill result with two samples with colliding metrics (to test fingerprint mapping).
|
|
result := model.Samples{
|
|
&model.Sample{
|
|
Metric: model.Metric{
|
|
"instance": "ip-10-33-84-73.l05.ams5.s-cloud.net:24483",
|
|
"status": "503",
|
|
},
|
|
Value: 42,
|
|
Timestamp: timestamp,
|
|
},
|
|
&model.Sample{
|
|
Metric: model.Metric{
|
|
"instance": "ip-10-33-84-73.l05.ams5.s-cloud.net:24480",
|
|
"status": "500",
|
|
},
|
|
Value: 2010,
|
|
Timestamp: timestamp + 1,
|
|
},
|
|
}
|
|
|
|
metrics := []model.Metric{}
|
|
for n := rand.Intn(maxMetrics); n >= 0; n-- {
|
|
metrics = append(metrics, model.Metric{
|
|
model.MetricNameLabel: model.LabelValue(metricName),
|
|
model.LabelName(fmt.Sprintf("labelname_%d", n+1)): model.LabelValue(fmt.Sprintf("labelvalue_%d", rand.Int())),
|
|
})
|
|
}
|
|
|
|
for len(result) < minLen {
|
|
// Pick a metric for this cycle.
|
|
metric := metrics[rand.Intn(len(metrics))]
|
|
timeDelta := rand.Intn(maxTimeDelta) + 1
|
|
generator := generators[rand.Intn(len(generators))]
|
|
createValue := generator.createValue
|
|
applyDelta := generator.applyDelta[rand.Intn(len(generator.applyDelta))]
|
|
incTimestamp := func() { timestamp += model.Time(timeDelta * (rand.Intn(maxTimeDeltaFactor) + 1)) }
|
|
switch rand.Intn(4) {
|
|
case 0: // A single sample.
|
|
result = append(result, &model.Sample{
|
|
Metric: metric,
|
|
Value: createValue(),
|
|
Timestamp: timestamp,
|
|
})
|
|
incTimestamp()
|
|
case 1: // A streak of random sample values.
|
|
for n := rand.Intn(maxStreakLength); n >= 0; n-- {
|
|
result = append(result, &model.Sample{
|
|
Metric: metric,
|
|
Value: createValue(),
|
|
Timestamp: timestamp,
|
|
})
|
|
incTimestamp()
|
|
}
|
|
case 2: // A streak of sample values with incremental changes.
|
|
value := createValue()
|
|
for n := rand.Intn(maxStreakLength); n >= 0; n-- {
|
|
result = append(result, &model.Sample{
|
|
Metric: metric,
|
|
Value: value,
|
|
Timestamp: timestamp,
|
|
})
|
|
incTimestamp()
|
|
value = applyDelta(value)
|
|
}
|
|
case 3: // A streak of constant sample values.
|
|
value := createValue()
|
|
for n := rand.Intn(maxStreakLength); n >= 0; n-- {
|
|
result = append(result, &model.Sample{
|
|
Metric: metric,
|
|
Value: value,
|
|
Timestamp: timestamp,
|
|
})
|
|
incTimestamp()
|
|
}
|
|
}
|
|
}
|
|
|
|
return result
|
|
}
|
|
|
|
func verifyStorage(t testing.TB, s *memorySeriesStorage, samples model.Samples, maxAge time.Duration) bool {
|
|
s.WaitForIndexing()
|
|
result := true
|
|
for _, i := range rand.Perm(len(samples)) {
|
|
sample := samples[i]
|
|
if sample.Timestamp.Before(model.TimeFromUnixNano(time.Now().Add(-maxAge).UnixNano())) {
|
|
continue
|
|
// TODO: Once we have a guaranteed cutoff at the
|
|
// retention period, we can verify here that no results
|
|
// are returned.
|
|
}
|
|
fp, err := s.mapper.mapFP(sample.Metric.FastFingerprint(), sample.Metric)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
p := s.NewPreloader()
|
|
it, err := p.PreloadRange(fp, sample.Timestamp, sample.Timestamp)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
found := it.ValueAtOrBeforeTime(sample.Timestamp)
|
|
if found.Timestamp == model.Earliest {
|
|
t.Errorf("Sample %#v: Expected sample not found.", sample)
|
|
result = false
|
|
p.Close()
|
|
continue
|
|
}
|
|
if sample.Value != found.Value || sample.Timestamp != found.Timestamp {
|
|
t.Errorf(
|
|
"Value (or timestamp) mismatch, want %f (at time %v), got %f (at time %v).",
|
|
sample.Value, sample.Timestamp, found.Value, found.Timestamp,
|
|
)
|
|
result = false
|
|
}
|
|
p.Close()
|
|
}
|
|
return result
|
|
}
|
|
|
|
func TestAppendOutOfOrder(t *testing.T) {
|
|
s, closer := NewTestStorage(t, 1)
|
|
defer closer.Close()
|
|
|
|
m := model.Metric{
|
|
model.MetricNameLabel: "out_of_order",
|
|
}
|
|
|
|
for i, t := range []int{0, 2, 2, 1} {
|
|
s.Append(&model.Sample{
|
|
Metric: m,
|
|
Timestamp: model.Time(t),
|
|
Value: model.SampleValue(i),
|
|
})
|
|
}
|
|
|
|
fp, err := s.mapper.mapFP(m.FastFingerprint(), m)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
pl := s.NewPreloader()
|
|
defer pl.Close()
|
|
|
|
it, err := pl.PreloadRange(fp, 0, 2)
|
|
if err != nil {
|
|
t.Fatalf("Error preloading chunks: %s", err)
|
|
}
|
|
|
|
want := []model.SamplePair{
|
|
{
|
|
Timestamp: 0,
|
|
Value: 0,
|
|
},
|
|
{
|
|
Timestamp: 2,
|
|
Value: 1,
|
|
},
|
|
}
|
|
got := it.RangeValues(metric.Interval{OldestInclusive: 0, NewestInclusive: 2})
|
|
if !reflect.DeepEqual(want, got) {
|
|
t.Fatalf("want %v, got %v", want, got)
|
|
}
|
|
}
|