mudhorn/prometheus
1
0
Fork 0
mirror of https://github.com/prometheus/prometheus.git synced 2024-12-28 15:09:39 -08:00
Code Issues Actions 4 Packages Projects Releases Wiki Activity

Merge pull request #161 from prometheus/julius-integrate-storage

Browse source 
Integrate memory and disk layers in view rendering.
This commit is contained in:
juliusv 2013-04-19 07:12:11 -07:00
parent 30c7acfaa4 99dcbe0f94
commit 6a5d2f68fc
12 changed files with 153 additions and 110 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
model/metric.go
View file

@ -102,6 +102,18 @@ func (v Values) Swap(i, j int) {
v[i], v[j] = v[j], v[i] v[i], v[j] = v[j], v[i]
} }
// FirstTimeAfter indicates whether the first sample of a set is after a given
// timestamp.
func (v Values) FirstTimeAfter(t time.Time) bool {
return v[0].Timestamp.After(t)
}
// LastTimeBefore indicates whether the last sample of a set is before a given
// timestamp.
func (v Values) LastTimeBefore(t time.Time) bool {
return v[len(v)-1].Timestamp.Before(t)
}
// InsideInterval indicates whether a given range of sorted values could contain // InsideInterval indicates whether a given range of sorted values could contain
// a value for a given time. // a value for a given time.
func (v Values) InsideInterval(t time.Time) (s bool) { func (v Values) InsideInterval(t time.Time) (s bool) {

2
rules/ast/ast.go
View file

@ -289,7 +289,7 @@ func EvalVectorRange(node VectorNode, start time.Time, end time.Time, interval t
if sampleSets[groupingKey] == nil { if sampleSets[groupingKey] == nil {
sampleSets[groupingKey] = &model.SampleSet{ sampleSets[groupingKey] = &model.SampleSet{
Metric: sample.Metric, Metric: sample.Metric,
Values: []model.SamplePair{samplePair}, Values: model.Values{samplePair},
} }
} else { } else {
sampleSets[groupingKey].Values = append(sampleSets[groupingKey].Values, samplePair) sampleSets[groupingKey].Values = append(sampleSets[groupingKey].Values, samplePair)

2
rules/ast/persistence_adapter.go
View file

@ -57,7 +57,7 @@ func interpolateSamples(first, second *model.SamplePair, timestamp time.Time) *m
// surrounding a given target time. If samples are found both before and after // surrounding a given target time. If samples are found both before and after
// the target time, the sample value is interpolated between these. Otherwise, // the target time, the sample value is interpolated between these. Otherwise,
// the single closest sample is returned verbatim. // the single closest sample is returned verbatim.
func (v *viewAdapter) chooseClosestSample(samples []model.SamplePair, timestamp time.Time) (sample *model.SamplePair) { func (v *viewAdapter) chooseClosestSample(samples model.Values, timestamp time.Time) (sample *model.SamplePair) {
var closestBefore *model.SamplePair var closestBefore *model.SamplePair
var closestAfter *model.SamplePair var closestAfter *model.SamplePair
for _, candidate := range samples { for _, candidate := range samples {

2
rules/testdata.go
View file

@ -25,7 +25,7 @@ var testStartTime = time.Time{}
func getTestValueStream(startVal model.SampleValue, func getTestValueStream(startVal model.SampleValue,
endVal model.SampleValue, endVal model.SampleValue,
stepVal model.SampleValue) (resultValues []model.SamplePair) { stepVal model.SampleValue) (resultValues model.Values) {
currentTime := testStartTime currentTime := testStartTime
for currentVal := startVal; currentVal <= endVal; currentVal += stepVal { for currentVal := startVal; currentVal <= endVal; currentVal += stepVal {
sample := model.SamplePair{ sample := model.SamplePair{

12
storage/metric/interface.go
View file

@ -47,9 +47,9 @@ type MetricPersistence interface {
// Get the metric associated with the provided fingerprint. // Get the metric associated with the provided fingerprint.
GetMetricForFingerprint(model.Fingerprint) (*model.Metric, error) GetMetricForFingerprint(model.Fingerprint) (*model.Metric, error)
GetValueAtTime(model.Fingerprint, time.Time) []model.SamplePair GetValueAtTime(model.Fingerprint, time.Time) model.Values
GetBoundaryValues(model.Fingerprint, model.Interval) (first []model.SamplePair, second []model.SamplePair) GetBoundaryValues(model.Fingerprint, model.Interval) (first model.Values, second model.Values)
GetRangeValues(model.Fingerprint, model.Interval) []model.SamplePair GetRangeValues(model.Fingerprint, model.Interval) model.Values
ForEachSample(IteratorsForFingerprintBuilder) (err error) ForEachSample(IteratorsForFingerprintBuilder) (err error)
@ -64,9 +64,9 @@ type MetricPersistence interface {
// View provides view of the values in the datastore subject to the request of a // View provides view of the values in the datastore subject to the request of a
// preloading operation. // preloading operation.
type View interface { type View interface {
GetValueAtTime(model.Fingerprint, time.Time) []model.SamplePair GetValueAtTime(model.Fingerprint, time.Time) model.Values
GetBoundaryValues(model.Fingerprint, model.Interval) (first []model.SamplePair, second []model.SamplePair) GetBoundaryValues(model.Fingerprint, model.Interval) (first model.Values, second model.Values)
GetRangeValues(model.Fingerprint, model.Interval) []model.SamplePair GetRangeValues(model.Fingerprint, model.Interval) model.Values
// Destroy this view. // Destroy this view.
Close() Close()

6
storage/metric/leveldb.go
View file

@ -862,15 +862,15 @@ func (l *LevelDBMetricPersistence) GetMetricForFingerprint(f model.Fingerprint)
return return
} }
func (l LevelDBMetricPersistence) GetValueAtTime(f model.Fingerprint, t time.Time) (samples []model.SamplePair) { func (l LevelDBMetricPersistence) GetValueAtTime(f model.Fingerprint, t time.Time) (samples model.Values) {
panic("Not implemented") panic("Not implemented")
} }
func (l LevelDBMetricPersistence) GetBoundaryValues(f model.Fingerprint, i model.Interval) (first []model.SamplePair, second []model.SamplePair) { func (l LevelDBMetricPersistence) GetBoundaryValues(f model.Fingerprint, i model.Interval) (first model.Values, second model.Values) {
panic("Not implemented") panic("Not implemented")
} }
func (l *LevelDBMetricPersistence) GetRangeValues(f model.Fingerprint, i model.Interval) (samples []model.SamplePair) { func (l *LevelDBMetricPersistence) GetRangeValues(f model.Fingerprint, i model.Interval) (samples model.Values) {
panic("Not implemented") panic("Not implemented")
} }

6
storage/metric/memory.go
View file

@ -210,7 +210,7 @@ func (s memorySeriesStorage) GetMetricForFingerprint(f model.Fingerprint) (metri
return return
} }
func (s memorySeriesStorage) GetValueAtTime(f model.Fingerprint, t time.Time) (samples []model.SamplePair) { func (s memorySeriesStorage) GetValueAtTime(f model.Fingerprint, t time.Time) (samples model.Values) {
series, ok := s.fingerprintToSeries[f] series, ok := s.fingerprintToSeries[f]
if !ok { if !ok {
return return
@ -251,13 +251,13 @@ func (s memorySeriesStorage) GetValueAtTime(f model.Fingerprint, t time.Time) (s
return return
} }
func (s memorySeriesStorage) GetBoundaryValues(f model.Fingerprint, i model.Interval) (first []model.SamplePair, second []model.SamplePair) { func (s memorySeriesStorage) GetBoundaryValues(f model.Fingerprint, i model.Interval) (first model.Values, second model.Values) {
first = s.GetValueAtTime(f, i.OldestInclusive) first = s.GetValueAtTime(f, i.OldestInclusive)
second = s.GetValueAtTime(f, i.NewestInclusive) second = s.GetValueAtTime(f, i.NewestInclusive)
return return
} }
func (s memorySeriesStorage) GetRangeValues(f model.Fingerprint, i model.Interval) (samples []model.SamplePair) { func (s memorySeriesStorage) GetRangeValues(f model.Fingerprint, i model.Interval) (samples model.Values) {
series, ok := s.fingerprintToSeries[f] series, ok := s.fingerprintToSeries[f]
if !ok { if !ok {
return return

10
storage/metric/operation.go
View file

@ -26,7 +26,7 @@ type op interface {
// The time at which this operation starts. // The time at which this operation starts.
StartsAt() time.Time StartsAt() time.Time
// Extract samples from stream of values and advance operation time. // Extract samples from stream of values and advance operation time.
ExtractSamples(in []model.SamplePair) (out []model.SamplePair) ExtractSamples(in model.Values) (out model.Values)
// Get current operation time or nil if no subsequent work associated with // Get current operation time or nil if no subsequent work associated with
// this operator remains. // this operator remains.
CurrentTime() *time.Time CurrentTime() *time.Time
@ -73,7 +73,7 @@ func (g getValuesAtTimeOp) StartsAt() time.Time {
return g.time return g.time
} }
func (g *getValuesAtTimeOp) ExtractSamples(in []model.SamplePair) (out []model.SamplePair) { func (g *getValuesAtTimeOp) ExtractSamples(in model.Values) (out model.Values) {
if len(in) == 0 { if len(in) == 0 {
return return
} }
@ -100,7 +100,7 @@ func (g getValuesAtTimeOp) GreedierThan(op op) (superior bool) {
// are adjacent to it. // are adjacent to it.
// //
// An assumption of this is that the provided samples are already sorted! // An assumption of this is that the provided samples are already sorted!
func extractValuesAroundTime(t time.Time, in []model.SamplePair) (out []model.SamplePair) { func extractValuesAroundTime(t time.Time, in model.Values) (out model.Values) {
i := sort.Search(len(in), func(i int) bool { i := sort.Search(len(in), func(i int) bool {
return !in[i].Timestamp.Before(t) return !in[i].Timestamp.Before(t)
}) })
@ -151,7 +151,7 @@ func (g getValuesAtIntervalOp) Through() time.Time {
return g.through return g.through
} }
func (g *getValuesAtIntervalOp) ExtractSamples(in []model.SamplePair) (out []model.SamplePair) { func (g *getValuesAtIntervalOp) ExtractSamples(in model.Values) (out model.Values) {
if len(in) == 0 { if len(in) == 0 {
return return
} }
@ -206,7 +206,7 @@ func (g getValuesAlongRangeOp) Through() time.Time {
return g.through return g.through
} }
func (g *getValuesAlongRangeOp) ExtractSamples(in []model.SamplePair) (out []model.SamplePair) { func (g *getValuesAlongRangeOp) ExtractSamples(in model.Values) (out model.Values) {
if len(in) == 0 { if len(in) == 0 {
return return
} }

92
storage/metric/operation_test.go
View file

@ -1177,8 +1177,8 @@ func BenchmarkOptimize(b *testing.B) {
func TestGetValuesAtTimeOp(t *testing.T) { func TestGetValuesAtTimeOp(t *testing.T) {
var scenarios = []struct { var scenarios = []struct {
op getValuesAtTimeOp op getValuesAtTimeOp
in []model.SamplePair in model.Values
out []model.SamplePair out model.Values
}{ }{
// No values. // No values.
{ {
@ -1191,13 +1191,13 @@ func TestGetValuesAtTimeOp(t *testing.T) {
op: getValuesAtTimeOp{ op: getValuesAtTimeOp{
time: testInstant, time: testInstant,
}, },
in: []model.SamplePair{ in: model.Values{
{ {
Timestamp: testInstant.Add(1 * time.Minute), Timestamp: testInstant.Add(1 * time.Minute),
Value: 1, Value: 1,
}, },
}, },
out: []model.SamplePair{ out: model.Values{
{ {
Timestamp: testInstant.Add(1 * time.Minute), Timestamp: testInstant.Add(1 * time.Minute),
Value: 1, Value: 1,
@ -1209,13 +1209,13 @@ func TestGetValuesAtTimeOp(t *testing.T) {
op: getValuesAtTimeOp{ op: getValuesAtTimeOp{
time: testInstant.Add(1 * time.Minute), time: testInstant.Add(1 * time.Minute),
}, },
in: []model.SamplePair{ in: model.Values{
{ {
Timestamp: testInstant.Add(1 * time.Minute), Timestamp: testInstant.Add(1 * time.Minute),
Value: 1, Value: 1,
}, },
}, },
out: []model.SamplePair{ out: model.Values{
{ {
Timestamp: testInstant.Add(1 * time.Minute), Timestamp: testInstant.Add(1 * time.Minute),
Value: 1, Value: 1,
@ -1227,13 +1227,13 @@ func TestGetValuesAtTimeOp(t *testing.T) {
op: getValuesAtTimeOp{ op: getValuesAtTimeOp{
time: testInstant.Add(2 * time.Minute), time: testInstant.Add(2 * time.Minute),
}, },
in: []model.SamplePair{ in: model.Values{
{ {
Timestamp: testInstant.Add(1 * time.Minute), Timestamp: testInstant.Add(1 * time.Minute),
Value: 1, Value: 1,
}, },
}, },
out: []model.SamplePair{ out: model.Values{
{ {
Timestamp: testInstant.Add(1 * time.Minute), Timestamp: testInstant.Add(1 * time.Minute),
Value: 1, Value: 1,
@ -1245,7 +1245,7 @@ func TestGetValuesAtTimeOp(t *testing.T) {
op: getValuesAtTimeOp{ op: getValuesAtTimeOp{
time: testInstant, time: testInstant,
}, },
in: []model.SamplePair{ in: model.Values{
{ {
Timestamp: testInstant.Add(1 * time.Minute), Timestamp: testInstant.Add(1 * time.Minute),
Value: 1, Value: 1,
@ -1255,7 +1255,7 @@ func TestGetValuesAtTimeOp(t *testing.T) {
Value: 1, Value: 1,
}, },
}, },
out: []model.SamplePair{ out: model.Values{
{ {
Timestamp: testInstant.Add(1 * time.Minute), Timestamp: testInstant.Add(1 * time.Minute),
Value: 1, Value: 1,
@ -1267,7 +1267,7 @@ func TestGetValuesAtTimeOp(t *testing.T) {
op: getValuesAtTimeOp{ op: getValuesAtTimeOp{
time: testInstant.Add(1 * time.Minute), time: testInstant.Add(1 * time.Minute),
}, },
in: []model.SamplePair{ in: model.Values{
{ {
Timestamp: testInstant.Add(1 * time.Minute), Timestamp: testInstant.Add(1 * time.Minute),
Value: 1, Value: 1,
@ -1277,7 +1277,7 @@ func TestGetValuesAtTimeOp(t *testing.T) {
Value: 1, Value: 1,
}, },
}, },
out: []model.SamplePair{ out: model.Values{
{ {
Timestamp: testInstant.Add(1 * time.Minute), Timestamp: testInstant.Add(1 * time.Minute),
Value: 1, Value: 1,
@ -1289,7 +1289,7 @@ func TestGetValuesAtTimeOp(t *testing.T) {
op: getValuesAtTimeOp{ op: getValuesAtTimeOp{
time: testInstant.Add(90 * time.Second), time: testInstant.Add(90 * time.Second),
}, },
in: []model.SamplePair{ in: model.Values{
{ {
Timestamp: testInstant.Add(1 * time.Minute), Timestamp: testInstant.Add(1 * time.Minute),
Value: 1, Value: 1,
@ -1299,7 +1299,7 @@ func TestGetValuesAtTimeOp(t *testing.T) {
Value: 1, Value: 1,
}, },
}, },
out: []model.SamplePair{ out: model.Values{
{ {
Timestamp: testInstant.Add(1 * time.Minute), Timestamp: testInstant.Add(1 * time.Minute),
Value: 1, Value: 1,
@ -1315,7 +1315,7 @@ func TestGetValuesAtTimeOp(t *testing.T) {
op: getValuesAtTimeOp{ op: getValuesAtTimeOp{
time: testInstant.Add(2 * time.Minute), time: testInstant.Add(2 * time.Minute),
}, },
in: []model.SamplePair{ in: model.Values{
{ {
Timestamp: testInstant.Add(1 * time.Minute), Timestamp: testInstant.Add(1 * time.Minute),
Value: 1, Value: 1,
@ -1325,7 +1325,7 @@ func TestGetValuesAtTimeOp(t *testing.T) {
Value: 1, Value: 1,
}, },
}, },
out: []model.SamplePair{ out: model.Values{
{ {
Timestamp: testInstant.Add(1 * time.Minute), Timestamp: testInstant.Add(1 * time.Minute),
Value: 1, Value: 1,
@ -1341,7 +1341,7 @@ func TestGetValuesAtTimeOp(t *testing.T) {
op: getValuesAtTimeOp{ op: getValuesAtTimeOp{
time: testInstant.Add(3 * time.Minute), time: testInstant.Add(3 * time.Minute),
}, },
in: []model.SamplePair{ in: model.Values{
{ {
Timestamp: testInstant.Add(1 * time.Minute), Timestamp: testInstant.Add(1 * time.Minute),
Value: 1, Value: 1,
@ -1351,7 +1351,7 @@ func TestGetValuesAtTimeOp(t *testing.T) {
Value: 1, Value: 1,
}, },
}, },
out: []model.SamplePair{ out: model.Values{
{ {
Timestamp: testInstant.Add(2 * time.Minute), Timestamp: testInstant.Add(2 * time.Minute),
Value: 1, Value: 1,
@ -1376,8 +1376,8 @@ func TestGetValuesAtTimeOp(t *testing.T) {
func TestGetValuesAtIntervalOp(t *testing.T) { func TestGetValuesAtIntervalOp(t *testing.T) {
var scenarios = []struct { var scenarios = []struct {
op getValuesAtIntervalOp op getValuesAtIntervalOp
in []model.SamplePair in model.Values
out []model.SamplePair out model.Values
}{ }{
// No values. // No values.
{ {
@ -1394,7 +1394,7 @@ func TestGetValuesAtIntervalOp(t *testing.T) {
through: testInstant.Add(1 * time.Minute), through: testInstant.Add(1 * time.Minute),
interval: 30 * time.Second, interval: 30 * time.Second,
}, },
in: []model.SamplePair{ in: model.Values{
{ {
Timestamp: testInstant.Add(2 * time.Minute), Timestamp: testInstant.Add(2 * time.Minute),
Value: 1, Value: 1,
@ -1404,7 +1404,7 @@ func TestGetValuesAtIntervalOp(t *testing.T) {
Value: 1, Value: 1,
}, },
}, },
out: []model.SamplePair{ out: model.Values{
{ {
Timestamp: testInstant.Add(2 * time.Minute), Timestamp: testInstant.Add(2 * time.Minute),
Value: 1, Value: 1,
@ -1418,7 +1418,7 @@ func TestGetValuesAtIntervalOp(t *testing.T) {
through: testInstant.Add(2 * time.Minute), through: testInstant.Add(2 * time.Minute),
interval: 30 * time.Second, interval: 30 * time.Second,
}, },
in: []model.SamplePair{ in: model.Values{
{ {
Timestamp: testInstant.Add(1 * time.Minute), Timestamp: testInstant.Add(1 * time.Minute),
Value: 1, Value: 1,
@ -1428,7 +1428,7 @@ func TestGetValuesAtIntervalOp(t *testing.T) {
Value: 1, Value: 1,
}, },
}, },
out: []model.SamplePair{ out: model.Values{
{ {
Timestamp: testInstant.Add(1 * time.Minute), Timestamp: testInstant.Add(1 * time.Minute),
Value: 1, Value: 1,
@ -1446,7 +1446,7 @@ func TestGetValuesAtIntervalOp(t *testing.T) {
through: testInstant.Add(2 * time.Minute), through: testInstant.Add(2 * time.Minute),
interval: 30 * time.Second, interval: 30 * time.Second,
}, },
in: []model.SamplePair{ in: model.Values{
{ {
Timestamp: testInstant, Timestamp: testInstant,
Value: 1, Value: 1,
@ -1460,7 +1460,7 @@ func TestGetValuesAtIntervalOp(t *testing.T) {
Value: 1, Value: 1,
}, },
}, },
out: []model.SamplePair{ out: model.Values{
{ {
Timestamp: testInstant.Add(1 * time.Minute), Timestamp: testInstant.Add(1 * time.Minute),
Value: 1, Value: 1,
@ -1478,7 +1478,7 @@ func TestGetValuesAtIntervalOp(t *testing.T) {
through: testInstant.Add(3 * time.Minute), through: testInstant.Add(3 * time.Minute),
interval: 30 * time.Second, interval: 30 * time.Second,
}, },
in: []model.SamplePair{ in: model.Values{
{ {
Timestamp: testInstant.Add(1 * time.Minute), Timestamp: testInstant.Add(1 * time.Minute),
Value: 1, Value: 1,
@ -1488,7 +1488,7 @@ func TestGetValuesAtIntervalOp(t *testing.T) {
Value: 1, Value: 1,
}, },
}, },
out: []model.SamplePair{ out: model.Values{
{ {
Timestamp: testInstant.Add(1 * time.Minute), Timestamp: testInstant.Add(1 * time.Minute),
Value: 1, Value: 1,
@ -1506,7 +1506,7 @@ func TestGetValuesAtIntervalOp(t *testing.T) {
through: testInstant.Add(4 * time.Minute), through: testInstant.Add(4 * time.Minute),
interval: 30 * time.Second, interval: 30 * time.Second,
}, },
in: []model.SamplePair{ in: model.Values{
{ {
Timestamp: testInstant, Timestamp: testInstant,
Value: 1, Value: 1,
@ -1524,7 +1524,7 @@ func TestGetValuesAtIntervalOp(t *testing.T) {
Value: 1, Value: 1,
}, },
}, },
out: []model.SamplePair{ out: model.Values{
{ {
Timestamp: testInstant.Add(2 * time.Minute), Timestamp: testInstant.Add(2 * time.Minute),
Value: 1, Value: 1,
@ -1542,7 +1542,7 @@ func TestGetValuesAtIntervalOp(t *testing.T) {
through: testInstant.Add(3 * time.Minute), through: testInstant.Add(3 * time.Minute),
interval: 30 * time.Second, interval: 30 * time.Second,
}, },
in: []model.SamplePair{ in: model.Values{
{ {
Timestamp: testInstant, Timestamp: testInstant,
Value: 1, Value: 1,
@ -1552,7 +1552,7 @@ func TestGetValuesAtIntervalOp(t *testing.T) {
Value: 1, Value: 1,
}, },
}, },
out: []model.SamplePair{ out: model.Values{
{ {
Timestamp: testInstant.Add(1 * time.Minute), Timestamp: testInstant.Add(1 * time.Minute),
Value: 1, Value: 1,
@ -1577,8 +1577,8 @@ func TestGetValuesAtIntervalOp(t *testing.T) {
func TestGetValuesAlongRangeOp(t *testing.T) { func TestGetValuesAlongRangeOp(t *testing.T) {
var scenarios = []struct { var scenarios = []struct {
op getValuesAlongRangeOp op getValuesAlongRangeOp
in []model.SamplePair in model.Values
out []model.SamplePair out model.Values
}{ }{
// No values. // No values.
{ {
@ -1593,7 +1593,7 @@ func TestGetValuesAlongRangeOp(t *testing.T) {
from: testInstant, from: testInstant,
through: testInstant.Add(1 * time.Minute), through: testInstant.Add(1 * time.Minute),
}, },
in: []model.SamplePair{ in: model.Values{
{ {
Timestamp: testInstant.Add(2 * time.Minute), Timestamp: testInstant.Add(2 * time.Minute),
Value: 1, Value: 1,
@ -1603,7 +1603,7 @@ func TestGetValuesAlongRangeOp(t *testing.T) {
Value: 1, Value: 1,
}, },
}, },
out: []model.SamplePair{}, out: model.Values{},
}, },
// Operator range starts before first value, ends within available values. // Operator range starts before first value, ends within available values.
{ {
@ -1611,7 +1611,7 @@ func TestGetValuesAlongRangeOp(t *testing.T) {
from: testInstant, from: testInstant,
through: testInstant.Add(2 * time.Minute), through: testInstant.Add(2 * time.Minute),
}, },
in: []model.SamplePair{ in: model.Values{
{ {
Timestamp: testInstant.Add(1 * time.Minute), Timestamp: testInstant.Add(1 * time.Minute),
Value: 1, Value: 1,
@ -1621,7 +1621,7 @@ func TestGetValuesAlongRangeOp(t *testing.T) {
Value: 1, Value: 1,
}, },
}, },
out: []model.SamplePair{ out: model.Values{
{ {
Timestamp: testInstant.Add(1 * time.Minute), Timestamp: testInstant.Add(1 * time.Minute),
Value: 1, Value: 1,
@ -1634,7 +1634,7 @@ func TestGetValuesAlongRangeOp(t *testing.T) {
from: testInstant.Add(1 * time.Minute), from: testInstant.Add(1 * time.Minute),
through: testInstant.Add(2 * time.Minute), through: testInstant.Add(2 * time.Minute),
}, },
in: []model.SamplePair{ in: model.Values{
{ {
Timestamp: testInstant, Timestamp: testInstant,
Value: 1, Value: 1,
@ -1648,7 +1648,7 @@ func TestGetValuesAlongRangeOp(t *testing.T) {
Value: 1, Value: 1,
}, },
}, },
out: []model.SamplePair{ out: model.Values{
{ {
Timestamp: testInstant.Add(1 * time.Minute), Timestamp: testInstant.Add(1 * time.Minute),
Value: 1, Value: 1,
@ -1661,7 +1661,7 @@ func TestGetValuesAlongRangeOp(t *testing.T) {
from: testInstant, from: testInstant,
through: testInstant.Add(3 * time.Minute), through: testInstant.Add(3 * time.Minute),
}, },
in: []model.SamplePair{ in: model.Values{
{ {
Timestamp: testInstant.Add(1 * time.Minute), Timestamp: testInstant.Add(1 * time.Minute),
Value: 1, Value: 1,
@ -1671,7 +1671,7 @@ func TestGetValuesAlongRangeOp(t *testing.T) {
Value: 1, Value: 1,
}, },
}, },
out: []model.SamplePair{ out: model.Values{
{ {
Timestamp: testInstant.Add(1 * time.Minute), Timestamp: testInstant.Add(1 * time.Minute),
Value: 1, Value: 1,
@ -1688,7 +1688,7 @@ func TestGetValuesAlongRangeOp(t *testing.T) {
from: testInstant.Add(2 * time.Minute), from: testInstant.Add(2 * time.Minute),
through: testInstant.Add(4 * time.Minute), through: testInstant.Add(4 * time.Minute),
}, },
in: []model.SamplePair{ in: model.Values{
{ {
Timestamp: testInstant, Timestamp: testInstant,
Value: 1, Value: 1,
@ -1706,7 +1706,7 @@ func TestGetValuesAlongRangeOp(t *testing.T) {
Value: 1, Value: 1,
}, },
}, },
out: []model.SamplePair{ out: model.Values{
{ {
Timestamp: testInstant.Add(2 * time.Minute), Timestamp: testInstant.Add(2 * time.Minute),
Value: 1, Value: 1,
@ -1723,7 +1723,7 @@ func TestGetValuesAlongRangeOp(t *testing.T) {
from: testInstant.Add(2 * time.Minute), from: testInstant.Add(2 * time.Minute),
through: testInstant.Add(3 * time.Minute), through: testInstant.Add(3 * time.Minute),
}, },
in: []model.SamplePair{ in: model.Values{
{ {
Timestamp: testInstant, Timestamp: testInstant,
Value: 1, Value: 1,
@ -1733,7 +1733,7 @@ func TestGetValuesAlongRangeOp(t *testing.T) {
Value: 1, Value: 1,
}, },
}, },
out: []model.SamplePair{}, out: model.Values{},
}, },
} }
for i, scenario := range scenarios { for i, scenario := range scenarios {

4
storage/metric/stochastic_test.go
View file

@ -188,7 +188,7 @@ func AppendSampleAsPureSingleEntityAppendTests(p MetricPersistence, t test.Teste
} }
} }
func levelDBGetRangeValues(l *LevelDBMetricPersistence, fp model.Fingerprint, i model.Interval) (samples []model.SamplePair, err error) { func levelDBGetRangeValues(l *LevelDBMetricPersistence, fp model.Fingerprint, i model.Interval) (samples model.Values, err error) {
begin := time.Now() begin := time.Now()
defer func() { defer func() {
@ -465,7 +465,7 @@ func StochasticTests(persistenceMaker func() (MetricPersistence, test.Closer), t
NewestInclusive: time.Unix(end, 0), NewestInclusive: time.Unix(end, 0),
} }
samples := []model.SamplePair{} samples := model.Values{}
fp := model.NewFingerprintFromMetric(metric) fp := model.NewFingerprintFromMetric(metric)
switch persistence := p.(type) { switch persistence := p.(type) {
case *LevelDBMetricPersistence: case *LevelDBMetricPersistence:

49
storage/metric/tiered.go
View file

@ -370,19 +370,14 @@ func (t *tieredStorage) renderView(viewJob viewJob) {
if err != nil { if err != nil {
panic(err) panic(err)
} }
if t.diskFrontier == nil {
// Storage still empty, return an empty view.
viewJob.output <- view
return
}
for _, scanJob := range scans { for _, scanJob := range scans {
seriesFrontier, err := newSeriesFrontier(scanJob.fingerprint, *t.diskFrontier, iterator) var seriesFrontier *seriesFrontier = nil
if t.diskFrontier != nil {
seriesFrontier, err = newSeriesFrontier(scanJob.fingerprint, *t.diskFrontier, iterator)
if err != nil { if err != nil {
panic(err) panic(err)
} }
if seriesFrontier == nil {
continue
} }
standingOps := scanJob.operations standingOps := scanJob.operations
@ -393,25 +388,47 @@ func (t *tieredStorage) renderView(viewJob viewJob) {
} }
// Load data value chunk(s) around the first standing op's current time. // Load data value chunk(s) around the first standing op's current time.
highWatermark := *standingOps[0].CurrentTime() targetTime := *standingOps[0].CurrentTime()
chunk := model.Values{}
memValues := t.memoryArena.GetValueAtTime(scanJob.fingerprint, targetTime)
// If we aimed before the oldest value in memory, load more data from disk.
if (len(memValues) == 0 || memValues.FirstTimeAfter(targetTime)) && seriesFrontier != nil {
// XXX: For earnest performance gains analagous to the benchmarking we // XXX: For earnest performance gains analagous to the benchmarking we
// performed, chunk should only be reloaded if it no longer contains // performed, chunk should only be reloaded if it no longer contains
// the values we're looking for. // the values we're looking for.
// //
// To better understand this, look at https://github.com/prometheus/prometheus/blob/benchmark/leveldb/iterator-seek-characteristics/leveldb.go#L239 and note the behavior around retrievedValue. // To better understand this, look at https://github.com/prometheus/prometheus/blob/benchmark/leveldb/iterator-seek-characteristics/leveldb.go#L239 and note the behavior around retrievedValue.
chunk := t.loadChunkAroundTime(iterator, seriesFrontier, scanJob.fingerprint, highWatermark) diskValues := t.loadChunkAroundTime(iterator, seriesFrontier, scanJob.fingerprint, targetTime)
// If we aimed past the newest value on disk, combine it with the next value from memory.
if len(memValues) > 0 && diskValues.LastTimeBefore(targetTime) {
latestDiskValue := diskValues[len(diskValues)-1 : len(diskValues)]
chunk = append(latestDiskValue, memValues...)
} else {
chunk = diskValues
}
} else {
chunk = memValues
}
// There's no data at all for this fingerprint, so stop processing ops for it.
if len(chunk) == 0 {
break
}
lastChunkTime := chunk[len(chunk)-1].Timestamp lastChunkTime := chunk[len(chunk)-1].Timestamp
if lastChunkTime.After(highWatermark) { if lastChunkTime.After(targetTime) {
highWatermark = lastChunkTime targetTime = lastChunkTime
} }
// For each op, extract all needed data from the current chunk. // For each op, extract all needed data from the current chunk.
out := []model.SamplePair{} out := model.Values{}
for _, op := range standingOps { for _, op := range standingOps {
if op.CurrentTime().After(highWatermark) { if op.CurrentTime().After(targetTime) {
break break
} }
for op.CurrentTime() != nil && !op.CurrentTime().After(highWatermark) { for op.CurrentTime() != nil && !op.CurrentTime().After(targetTime) {
out = op.ExtractSamples(chunk) out = op.ExtractSamples(chunk)
} }
} }
@ -450,7 +467,7 @@ func (t *tieredStorage) renderView(viewJob viewJob) {
return return
} }
func (t *tieredStorage) loadChunkAroundTime(iterator leveldb.Iterator, frontier *seriesFrontier, fingerprint model.Fingerprint, ts time.Time) (chunk []model.SamplePair) { func (t *tieredStorage) loadChunkAroundTime(iterator leveldb.Iterator, frontier *seriesFrontier, fingerprint model.Fingerprint, ts time.Time) (chunk model.Values) {
var ( var (
targetKey = &dto.SampleKey{ targetKey = &dto.SampleKey{
Fingerprint: fingerprint.ToDTO(), Fingerprint: fingerprint.ToDTO(),

48
storage/metric/tiered_test.go
View file

@ -69,7 +69,7 @@ func buildSamples(from, to time.Time, interval time.Duration, m model.Metric) (v
return return
} }
func testMakeView(t test.Tester) { func testMakeView(t test.Tester, flushToDisk bool) {
type in struct { type in struct {
atTime []getValuesAtTimeOp atTime []getValuesAtTimeOp
atInterval []getValuesAtIntervalOp atInterval []getValuesAtIntervalOp
@ -77,9 +77,9 @@ func testMakeView(t test.Tester) {
} }
type out struct { type out struct {
atTime [][]model.SamplePair atTime []model.Values
atInterval [][]model.SamplePair atInterval []model.Values
alongRange [][]model.SamplePair alongRange []model.Values
} }
var ( var (
instant = time.Date(1984, 3, 30, 0, 0, 0, 0, time.Local) instant = time.Date(1984, 3, 30, 0, 0, 0, 0, time.Local)
@ -100,7 +100,7 @@ func testMakeView(t test.Tester) {
}, },
}, },
out: out{ out: out{
atTime: [][]model.SamplePair{{}}, atTime: []model.Values{{}},
}, },
}, },
// Single sample, query asks for exact sample time. // Single sample, query asks for exact sample time.
@ -120,7 +120,7 @@ func testMakeView(t test.Tester) {
}, },
}, },
out: out{ out: out{
atTime: [][]model.SamplePair{ atTime: []model.Values{
{ {
{ {
Timestamp: instant, Timestamp: instant,
@ -152,7 +152,7 @@ func testMakeView(t test.Tester) {
}, },
}, },
out: out{ out: out{
atTime: [][]model.SamplePair{ atTime: []model.Values{
{ {
{ {
Timestamp: instant.Add(time.Second), Timestamp: instant.Add(time.Second),
@ -179,7 +179,7 @@ func testMakeView(t test.Tester) {
}, },
}, },
out: out{ out: out{
atTime: [][]model.SamplePair{ atTime: []model.Values{
{ {
{ {
Timestamp: instant, Timestamp: instant,
@ -211,7 +211,7 @@ func testMakeView(t test.Tester) {
}, },
}, },
out: out{ out: out{
atTime: [][]model.SamplePair{ atTime: []model.Values{
{ {
{ {
Timestamp: instant, Timestamp: instant,
@ -248,7 +248,7 @@ func testMakeView(t test.Tester) {
}, },
}, },
out: out{ out: out{
atTime: [][]model.SamplePair{ atTime: []model.Values{
{ {
{ {
Timestamp: instant.Add(time.Second), Timestamp: instant.Add(time.Second),
@ -285,7 +285,7 @@ func testMakeView(t test.Tester) {
}, },
}, },
out: out{ out: out{
atTime: [][]model.SamplePair{ atTime: []model.Values{
{ {
{ {
Timestamp: instant, Timestamp: instant,
@ -326,7 +326,7 @@ func testMakeView(t test.Tester) {
}, },
}, },
out: out{ out: out{
atTime: [][]model.SamplePair{ atTime: []model.Values{
{ {
{ {
Timestamp: instant.Add(time.Second * 2), Timestamp: instant.Add(time.Second * 2),
@ -351,7 +351,7 @@ func testMakeView(t test.Tester) {
}, },
}, },
out: out{ out: out{
atTime: [][]model.SamplePair{ atTime: []model.Values{
{ {
{ {
Timestamp: instant.Add(time.Second * time.Duration(*leveldbChunkSize/2)), Timestamp: instant.Add(time.Second * time.Duration(*leveldbChunkSize/2)),
@ -378,7 +378,11 @@ func testMakeView(t test.Tester) {
} }
} }
if flushToDisk {
tiered.Flush() tiered.Flush()
} else {
tiered.(*tieredStorage).writeMemory()
}
requestBuilder := NewViewRequestBuilder() requestBuilder := NewViewRequestBuilder()
@ -421,13 +425,23 @@ func testMakeView(t test.Tester) {
} }
} }
func TestMakeView(t *testing.T) { func TestMakeViewFlush(t *testing.T) {
testMakeView(t) testMakeView(t, true)
} }
func BenchmarkMakeView(b *testing.B) { func BenchmarkMakeViewFlush(b *testing.B) {
for i := 0; i < b.N; i++ { for i := 0; i < b.N; i++ {
testMakeView(b) testMakeView(b, true)
}
}
func TestMakeViewNoFlush(t *testing.T) {
testMakeView(t, false)
}
func BenchmarkMakeViewNoFlush(b *testing.B) {
for i := 0; i < b.N; i++ {
testMakeView(b, false)
} }
} }