-
Notifications
You must be signed in to change notification settings - Fork 451
/
values.go
270 lines (230 loc) · 8 KB
/
values.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
// Copyright (c) 2018 Uber Technologies, Inc.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.
package ts
import (
"time"
"github.com/m3db/m3/src/query/models"
)
// Values holds the values for a timeseries. It provides a minimal interface
// for storing and retrieving values in the series, with Series providing a
// more convenient interface for applications to build on top of. Values
// objects are not specific to a given time, allowing them to be
// pre-allocated, pooled, and re-used across multiple series. There are
// multiple implementations of Values so that we can optimize storage based on
// the density of the series.
type Values interface {
// Len returns the number of values present
Len() int
// ValueAt returns the value at the nth element
ValueAt(n int) float64
// DatapointAt returns the datapoint at the nth element
DatapointAt(n int) Datapoint
// Datapoints returns all the datapoints
Datapoints() []Datapoint
// AlignToBounds returns values aligned to given bounds. To belong to a step,
// values should be <= stepTime and not stale.
AlignToBounds(
bounds models.Bounds,
lookbackDuration time.Duration,
) []Datapoints
// AlignToBoundsNoWriteForward returns values aligned to the start time
// and duration, but does not write points forward after aligning them. This
// differs from AlignToBounds which will write points forwards if no
// additional values are found in the values, adding an empty point instead.
AlignToBoundsNoWriteForward(
bounds models.Bounds,
lookbackDuration time.Duration,
) []Datapoints
}
// A Datapoint is a single data value reported at a given time
type Datapoint struct {
Timestamp time.Time
Value float64
}
// Datapoints is a list of datapoints.
type Datapoints []Datapoint
// Len is the length of the array.
func (d Datapoints) Len() int { return len(d) }
// ValueAt returns the value at the nth element.
func (d Datapoints) ValueAt(n int) float64 { return d[n].Value }
// DatapointAt returns the value at the nth element.
func (d Datapoints) DatapointAt(n int) Datapoint { return d[n] }
// Datapoints returns all the datapoints.
func (d Datapoints) Datapoints() []Datapoint { return d }
// Values returns the values representation.
func (d Datapoints) Values() []float64 {
values := make([]float64, len(d))
for i, dp := range d {
values[i] = dp.Value
}
return values
}
func (d Datapoints) alignToBounds(
bounds models.Bounds,
lookbackDuration time.Duration,
writeForward bool,
) []Datapoints {
numDatapoints := d.Len()
steps := bounds.Steps()
stepValues := make([]Datapoints, steps)
dpIdx := 0
stepSize := bounds.StepSize
t := bounds.Start
for i := 0; i < steps; i++ {
singleStepValues := make(Datapoints, 0)
for dpIdx < numDatapoints && !d[dpIdx].Timestamp.After(t) {
point := d[dpIdx]
dpIdx++
// Skip stale values
if t.Sub(point.Timestamp) > lookbackDuration {
continue
}
singleStepValues = append(singleStepValues, point)
}
// If writeForward is enabled and there is no point found for this
// interval, reuse the last point as long as its not stale
if writeForward {
if len(singleStepValues) == 0 && dpIdx > 0 {
prevPoint := d[dpIdx-1]
if t.Sub(prevPoint.Timestamp) <= lookbackDuration {
singleStepValues = Datapoints{prevPoint}
}
}
}
stepValues[i] = singleStepValues
t = t.Add(stepSize)
}
return stepValues
}
// AlignToBoundsNoWriteForward returns values aligned to the start time
// and duration, but does not write points forward after aligning them. This
// differs from AlignToBounds which will write points forwards if no additional
// values are found in the values, adding an empty point instead.
func (d Datapoints) AlignToBoundsNoWriteForward(
bounds models.Bounds,
lookbackDuration time.Duration,
) []Datapoints {
return d.alignToBounds(bounds, lookbackDuration, false)
}
// AlignToBounds returns values aligned to given bounds. To belong to a step,
// values should be <= stepTime and not stale.
func (d Datapoints) AlignToBounds(
bounds models.Bounds,
lookbackDuration time.Duration,
) []Datapoints {
return d.alignToBounds(bounds, lookbackDuration, true)
}
// MutableValues is the interface for values that can be updated
type MutableValues interface {
Values
// Sets the value at the given entry
SetValueAt(n int, v float64)
}
// FixedResolutionMutableValues are mutable values with fixed resolution between steps
type FixedResolutionMutableValues interface {
MutableValues
Resolution() time.Duration
StepAtTime(t time.Time) int
StartTimeForStep(n int) time.Time
// Time when the series starts
StartTime() time.Time
}
type fixedResolutionValues struct {
resolution time.Duration
numSteps int
values []float64
startTime time.Time
}
func (b *fixedResolutionValues) Len() int { return b.numSteps }
func (b *fixedResolutionValues) ValueAt(point int) float64 { return b.values[point] }
func (b *fixedResolutionValues) DatapointAt(point int) Datapoint {
return Datapoint{
Timestamp: b.StartTimeForStep(point),
Value: b.ValueAt(point),
}
}
func (b *fixedResolutionValues) Datapoints() []Datapoint {
datapoints := make([]Datapoint, 0, len(b.values))
for i := range b.values {
datapoints = append(datapoints, b.DatapointAt(i))
}
return datapoints
}
func (b *fixedResolutionValues) AlignToBounds(
_ models.Bounds,
_ time.Duration,
) []Datapoints {
values := make([]Datapoints, len(b.values))
for i := 0; i < b.Len(); i++ {
values[i] = Datapoints{b.DatapointAt(i)}
}
return values
}
func (b *fixedResolutionValues) AlignToBoundsNoWriteForward(
bb models.Bounds,
d time.Duration,
) []Datapoints {
return b.AlignToBounds(bb, d)
}
// StartTime returns the time the values start
func (b *fixedResolutionValues) StartTime() time.Time {
return b.startTime
}
// Resolution returns resolution per step
func (b *fixedResolutionValues) Resolution() time.Duration {
return b.resolution
}
// StepAtTime returns the step within the block containing the given time
func (b *fixedResolutionValues) StepAtTime(t time.Time) int {
return int(t.Sub(b.StartTime()) / b.Resolution())
}
// StartTimeForStep returns the time at which the given step starts
func (b *fixedResolutionValues) StartTimeForStep(n int) time.Time {
return b.startTime.Add(time.Duration(n) * b.Resolution())
}
// SetValueAt sets the value at the given entry
func (b *fixedResolutionValues) SetValueAt(n int, v float64) {
b.values[n] = v
}
// NewFixedStepValues returns mutable values with fixed resolution
func NewFixedStepValues(
resolution time.Duration,
numSteps int,
initialValue float64,
startTime time.Time,
) FixedResolutionMutableValues {
return newFixedStepValues(resolution, numSteps, initialValue, startTime)
}
func newFixedStepValues(
resolution time.Duration,
numSteps int,
initialValue float64,
startTime time.Time,
) *fixedResolutionValues {
values := make([]float64, numSteps)
// Faster way to initialize an array instead of a loop
Memset(values, initialValue)
return &fixedResolutionValues{
resolution: resolution,
numSteps: numSteps,
startTime: startTime,
values: values,
}
}