prometheus

1
// Copyright 2021 The Prometheus Authors
2
// Licensed under the Apache License, Version 2.0 (the "License");
3
// you may not use this file except in compliance with the License.
4
// You may obtain a copy of the License at
5
//
6
// http://www.apache.org/licenses/LICENSE-2.0
7
//
8
// Unless required by applicable law or agreed to in writing, software
9
// distributed under the License is distributed on an "AS IS" BASIS,
10
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
11
// See the License for the specific language governing permissions and
12
// limitations under the License.
13

14
package storage
15

16
import (
17
	"math"
18

19
	"github.com/prometheus/prometheus/model/histogram"
20
	"github.com/prometheus/prometheus/tsdb/chunkenc"
21
)
22

23
// MemoizedSeriesIterator wraps an iterator with a buffer to look back the previous element.
24
//
25
// This iterator regards integer histograms as float histograms; calls to Seek() will never return chunkenc.Histogram.
26
// This iterator deliberately does not implement chunkenc.Iterator.
27
type MemoizedSeriesIterator struct {
28
	it    chunkenc.Iterator
29
	delta int64
30

31
	lastTime  int64
32
	valueType chunkenc.ValueType
33

34
	// Keep track of the previously returned value.
35
	prevTime           int64
36
	prevValue          float64
37
	prevFloatHistogram *histogram.FloatHistogram
38
}
39

40
// NewMemoizedEmptyIterator is like NewMemoizedIterator but it's initialised with an empty iterator.
41
func NewMemoizedEmptyIterator(delta int64) *MemoizedSeriesIterator {
42
	return NewMemoizedIterator(chunkenc.NewNopIterator(), delta)
43
}
44

45
// NewMemoizedIterator returns a new iterator that buffers the values within the
46
// time range of the current element and the duration of delta before.
47
func NewMemoizedIterator(it chunkenc.Iterator, delta int64) *MemoizedSeriesIterator {
48
	bit := &MemoizedSeriesIterator{
49
		delta:    delta,
50
		prevTime: math.MinInt64,
51
	}
52
	bit.Reset(it)
53

54
	return bit
55
}
56

57
// Reset the internal state to reuse the wrapper with the provided iterator.
58
func (b *MemoizedSeriesIterator) Reset(it chunkenc.Iterator) {
59
	b.it = it
60
	b.lastTime = math.MinInt64
61
	b.prevTime = math.MinInt64
62
	b.valueType = it.Next()
63
}
64

65
// PeekPrev returns the previous element of the iterator. If there is none buffered,
66
// ok is false.
67
func (b *MemoizedSeriesIterator) PeekPrev() (t int64, v float64, fh *histogram.FloatHistogram, ok bool) {
68
	if b.prevTime == math.MinInt64 {
69
		return 0, 0, nil, false
70
	}
71
	return b.prevTime, b.prevValue, b.prevFloatHistogram, true
72
}
73

74
// Seek advances the iterator to the element at time t or greater.
75
func (b *MemoizedSeriesIterator) Seek(t int64) chunkenc.ValueType {
76
	t0 := t - b.delta
77

78
	if b.valueType != chunkenc.ValNone && t0 > b.lastTime {
79
		// Reset the previously stored element because the seek advanced
80
		// more than the delta.
81
		b.prevTime = math.MinInt64
82

83
		b.valueType = b.it.Seek(t0)
84
		switch b.valueType {
85
		case chunkenc.ValNone:
86
			return chunkenc.ValNone
87
		case chunkenc.ValHistogram:
88
			b.valueType = chunkenc.ValFloatHistogram
89
		}
90
		b.lastTime = b.it.AtT()
91
	}
92
	if b.lastTime >= t {
93
		return b.valueType
94
	}
95
	for b.Next() != chunkenc.ValNone {
96
		if b.lastTime >= t {
97
			return b.valueType
98
		}
99
	}
100

101
	return chunkenc.ValNone
102
}
103

104
// Next advances the iterator to the next element. Note that this does not check whether the element being buffered is
105
// within the time range of the current element and the duration of delta before.
106
func (b *MemoizedSeriesIterator) Next() chunkenc.ValueType {
107
	// Keep track of the previous element.
108
	switch b.valueType {
109
	case chunkenc.ValNone:
110
		return chunkenc.ValNone
111
	case chunkenc.ValFloat:
112
		b.prevTime, b.prevValue = b.it.At()
113
		b.prevFloatHistogram = nil
114
	case chunkenc.ValHistogram, chunkenc.ValFloatHistogram:
115
		b.prevValue = 0
116
		b.prevTime, b.prevFloatHistogram = b.it.AtFloatHistogram(nil)
117
	}
118

119
	b.valueType = b.it.Next()
120
	if b.valueType != chunkenc.ValNone {
121
		b.lastTime = b.it.AtT()
122
	}
123
	if b.valueType == chunkenc.ValHistogram {
124
		b.valueType = chunkenc.ValFloatHistogram
125
	}
126
	return b.valueType
127
}
128

129
// At returns the current float element of the iterator.
130
func (b *MemoizedSeriesIterator) At() (int64, float64) {
131
	return b.it.At()
132
}
133

134
// AtFloatHistogram returns the current float-histogram element of the iterator.
135
func (b *MemoizedSeriesIterator) AtFloatHistogram() (int64, *histogram.FloatHistogram) {
136
	return b.it.AtFloatHistogram(nil)
137
}
138

139
// Err returns the last encountered error.
140
func (b *MemoizedSeriesIterator) Err() error {
141
	return b.it.Err()
142
}
143