1 // Copyright 2015 The Go Authors. All rights reserved.
2 // Use of this source code is governed by a BSD-style
3 // license that can be found in the LICENSE file.
4 5 // Package timeseries implements a time series structure for stats collection.
6 package timeseries // import "golang.org/x/net/internal/timeseries"
7 8 import (
9 "fmt"
10 "log"
11 "time"
12 )
13 14 const (
15 timeSeriesNumBuckets = 64
16 minuteHourSeriesNumBuckets = 60
17 )
18 19 var timeSeriesResolutions = []time.Duration{
20 1 * time.Second,
21 10 * time.Second,
22 1 * time.Minute,
23 10 * time.Minute,
24 1 * time.Hour,
25 6 * time.Hour,
26 24 * time.Hour, // 1 day
27 7 * 24 * time.Hour, // 1 week
28 4 * 7 * 24 * time.Hour, // 4 weeks
29 16 * 7 * 24 * time.Hour, // 16 weeks
30 }
31 32 var minuteHourSeriesResolutions = []time.Duration{
33 1 * time.Second,
34 1 * time.Minute,
35 }
36 37 // An Observable is a kind of data that can be aggregated in a time series.
38 type Observable interface {
39 Multiply(ratio float64) // Multiplies the data in self by a given ratio
40 Add(other Observable) // Adds the data from a different observation to self
41 Clear() // Clears the observation so it can be reused.
42 CopyFrom(other Observable) // Copies the contents of a given observation to self
43 }
44 45 // Float attaches the methods of Observable to a float64.
46 type Float float64
47 48 // NewFloat returns a Float.
49 func NewFloat() Observable {
50 f := Float(0)
51 return &f
52 }
53 54 // String returns the float as a string.
55 func (f *Float) String() string { return fmt.Sprintf("%g", f.Value()) }
56 57 // Value returns the float's value.
58 func (f *Float) Value() float64 { return float64(*f) }
59 60 func (f *Float) Multiply(ratio float64) { *f *= Float(ratio) }
61 62 func (f *Float) Add(other Observable) {
63 o := other.(*Float)
64 *f += *o
65 }
66 67 func (f *Float) Clear() { *f = 0 }
68 69 func (f *Float) CopyFrom(other Observable) {
70 o := other.(*Float)
71 *f = *o
72 }
73 74 // A Clock tells the current time.
75 type Clock interface {
76 Time() time.Time
77 }
78 79 type defaultClock int
80 81 var defaultClockInstance defaultClock
82 83 func (defaultClock) Time() time.Time { return time.Now() }
84 85 // Information kept per level. Each level consists of a circular list of
86 // observations. The start of the level may be derived from end and the
87 // len(buckets) * sizeInMillis.
88 type tsLevel struct {
89 oldest int // index to oldest bucketed Observable
90 newest int // index to newest bucketed Observable
91 end time.Time // end timestamp for this level
92 size time.Duration // duration of the bucketed Observable
93 buckets []Observable // collections of observations
94 provider func() Observable // used for creating new Observable
95 }
96 97 func (l *tsLevel) Clear() {
98 l.oldest = 0
99 l.newest = len(l.buckets) - 1
100 l.end = time.Time{}
101 for i := range l.buckets {
102 if l.buckets[i] != nil {
103 l.buckets[i].Clear()
104 l.buckets[i] = nil
105 }
106 }
107 }
108 109 func (l *tsLevel) InitLevel(size time.Duration, numBuckets int, f func() Observable) {
110 l.size = size
111 l.provider = f
112 l.buckets = make([]Observable, numBuckets)
113 }
114 115 // Keeps a sequence of levels. Each level is responsible for storing data at
116 // a given resolution. For example, the first level stores data at a one
117 // minute resolution while the second level stores data at a one hour
118 // resolution.
119 120 // Each level is represented by a sequence of buckets. Each bucket spans an
121 // interval equal to the resolution of the level. New observations are added
122 // to the last bucket.
123 type timeSeries struct {
124 provider func() Observable // make more Observable
125 numBuckets int // number of buckets in each level
126 levels []*tsLevel // levels of bucketed Observable
127 lastAdd time.Time // time of last Observable tracked
128 total Observable // convenient aggregation of all Observable
129 clock Clock // Clock for getting current time
130 pending Observable // observations not yet bucketed
131 pendingTime time.Time // what time are we keeping in pending
132 dirty bool // if there are pending observations
133 }
134 135 // init initializes a level according to the supplied criteria.
136 func (ts *timeSeries) init(resolutions []time.Duration, f func() Observable, numBuckets int, clock Clock) {
137 ts.provider = f
138 ts.numBuckets = numBuckets
139 ts.clock = clock
140 ts.levels = make([]*tsLevel, len(resolutions))
141 142 for i := range resolutions {
143 if i > 0 && resolutions[i-1] >= resolutions[i] {
144 log.Print("timeseries: resolutions must be monotonically increasing")
145 break
146 }
147 newLevel := new(tsLevel)
148 newLevel.InitLevel(resolutions[i], ts.numBuckets, ts.provider)
149 ts.levels[i] = newLevel
150 }
151 152 ts.Clear()
153 }
154 155 // Clear removes all observations from the time series.
156 func (ts *timeSeries) Clear() {
157 ts.lastAdd = time.Time{}
158 ts.total = ts.resetObservation(ts.total)
159 ts.pending = ts.resetObservation(ts.pending)
160 ts.pendingTime = time.Time{}
161 ts.dirty = false
162 163 for i := range ts.levels {
164 ts.levels[i].Clear()
165 }
166 }
167 168 // Add records an observation at the current time.
169 func (ts *timeSeries) Add(observation Observable) {
170 ts.AddWithTime(observation, ts.clock.Time())
171 }
172 173 // AddWithTime records an observation at the specified time.
174 func (ts *timeSeries) AddWithTime(observation Observable, t time.Time) {
175 176 smallBucketDuration := ts.levels[0].size
177 178 if t.After(ts.lastAdd) {
179 ts.lastAdd = t
180 }
181 182 if t.After(ts.pendingTime) {
183 ts.advance(t)
184 ts.mergePendingUpdates()
185 ts.pendingTime = ts.levels[0].end
186 ts.pending.CopyFrom(observation)
187 ts.dirty = true
188 } else if t.After(ts.pendingTime.Add(-1 * smallBucketDuration)) {
189 // The observation is close enough to go into the pending bucket.
190 // This compensates for clock skewing and small scheduling delays
191 // by letting the update stay in the fast path.
192 ts.pending.Add(observation)
193 ts.dirty = true
194 } else {
195 ts.mergeValue(observation, t)
196 }
197 }
198 199 // mergeValue inserts the observation at the specified time in the past into all levels.
200 func (ts *timeSeries) mergeValue(observation Observable, t time.Time) {
201 for _, level := range ts.levels {
202 index := (ts.numBuckets - 1) - int(level.end.Sub(t)/level.size)
203 if 0 <= index && index < ts.numBuckets {
204 bucketNumber := (level.oldest + index) % ts.numBuckets
205 if level.buckets[bucketNumber] == nil {
206 level.buckets[bucketNumber] = level.provider()
207 }
208 level.buckets[bucketNumber].Add(observation)
209 }
210 }
211 ts.total.Add(observation)
212 }
213 214 // mergePendingUpdates applies the pending updates into all levels.
215 func (ts *timeSeries) mergePendingUpdates() {
216 if ts.dirty {
217 ts.mergeValue(ts.pending, ts.pendingTime)
218 ts.pending = ts.resetObservation(ts.pending)
219 ts.dirty = false
220 }
221 }
222 223 // advance cycles the buckets at each level until the latest bucket in
224 // each level can hold the time specified.
225 func (ts *timeSeries) advance(t time.Time) {
226 if !t.After(ts.levels[0].end) {
227 return
228 }
229 for i := 0; i < len(ts.levels); i++ {
230 level := ts.levels[i]
231 if !level.end.Before(t) {
232 break
233 }
234 235 // If the time is sufficiently far, just clear the level and advance
236 // directly.
237 if !t.Before(level.end.Add(level.size * time.Duration(ts.numBuckets))) {
238 for _, b := range level.buckets {
239 ts.resetObservation(b)
240 }
241 level.end = time.Unix(0, (t.UnixNano()/level.size.Nanoseconds())*level.size.Nanoseconds())
242 }
243 244 for t.After(level.end) {
245 level.end = level.end.Add(level.size)
246 level.newest = level.oldest
247 level.oldest = (level.oldest + 1) % ts.numBuckets
248 ts.resetObservation(level.buckets[level.newest])
249 }
250 251 t = level.end
252 }
253 }
254 255 // Latest returns the sum of the num latest buckets from the level.
256 func (ts *timeSeries) Latest(level, num int) Observable {
257 now := ts.clock.Time()
258 if ts.levels[0].end.Before(now) {
259 ts.advance(now)
260 }
261 262 ts.mergePendingUpdates()
263 264 result := ts.provider()
265 l := ts.levels[level]
266 index := l.newest
267 268 for i := 0; i < num; i++ {
269 if l.buckets[index] != nil {
270 result.Add(l.buckets[index])
271 }
272 if index == 0 {
273 index = ts.numBuckets
274 }
275 index--
276 }
277 278 return result
279 }
280 281 // LatestBuckets returns a copy of the num latest buckets from level.
282 func (ts *timeSeries) LatestBuckets(level, num int) []Observable {
283 if level < 0 || level > len(ts.levels) {
284 log.Print("timeseries: bad level argument: ", level)
285 return nil
286 }
287 if num < 0 || num >= ts.numBuckets {
288 log.Print("timeseries: bad num argument: ", num)
289 return nil
290 }
291 292 results := make([]Observable, num)
293 now := ts.clock.Time()
294 if ts.levels[0].end.Before(now) {
295 ts.advance(now)
296 }
297 298 ts.mergePendingUpdates()
299 300 l := ts.levels[level]
301 index := l.newest
302 303 for i := 0; i < num; i++ {
304 result := ts.provider()
305 results[i] = result
306 if l.buckets[index] != nil {
307 result.CopyFrom(l.buckets[index])
308 }
309 310 if index == 0 {
311 index = ts.numBuckets
312 }
313 index -= 1
314 }
315 return results
316 }
317 318 // ScaleBy updates observations by scaling by factor.
319 func (ts *timeSeries) ScaleBy(factor float64) {
320 for _, l := range ts.levels {
321 for i := 0; i < ts.numBuckets; i++ {
322 l.buckets[i].Multiply(factor)
323 }
324 }
325 326 ts.total.Multiply(factor)
327 ts.pending.Multiply(factor)
328 }
329 330 // Range returns the sum of observations added over the specified time range.
331 // If start or finish times don't fall on bucket boundaries of the same
332 // level, then return values are approximate answers.
333 func (ts *timeSeries) Range(start, finish time.Time) Observable {
334 return ts.ComputeRange(start, finish, 1)[0]
335 }
336 337 // Recent returns the sum of observations from the last delta.
338 func (ts *timeSeries) Recent(delta time.Duration) Observable {
339 now := ts.clock.Time()
340 return ts.Range(now.Add(-delta), now)
341 }
342 343 // Total returns the total of all observations.
344 func (ts *timeSeries) Total() Observable {
345 ts.mergePendingUpdates()
346 return ts.total
347 }
348 349 // ComputeRange computes a specified number of values into a slice using
350 // the observations recorded over the specified time period. The return
351 // values are approximate if the start or finish times don't fall on the
352 // bucket boundaries at the same level or if the number of buckets spanning
353 // the range is not an integral multiple of num.
354 func (ts *timeSeries) ComputeRange(start, finish time.Time, num int) []Observable {
355 if start.After(finish) {
356 log.Printf("timeseries: start > finish, %v>%v", start, finish)
357 return nil
358 }
359 360 if num < 0 {
361 log.Printf("timeseries: num < 0, %v", num)
362 return nil
363 }
364 365 results := make([]Observable, num)
366 367 for _, l := range ts.levels {
368 if !start.Before(l.end.Add(-l.size * time.Duration(ts.numBuckets))) {
369 ts.extract(l, start, finish, num, results)
370 return results
371 }
372 }
373 374 // Failed to find a level that covers the desired range. So just
375 // extract from the last level, even if it doesn't cover the entire
376 // desired range.
377 ts.extract(ts.levels[len(ts.levels)-1], start, finish, num, results)
378 379 return results
380 }
381 382 // RecentList returns the specified number of values in slice over the most
383 // recent time period of the specified range.
384 func (ts *timeSeries) RecentList(delta time.Duration, num int) []Observable {
385 if delta < 0 {
386 return nil
387 }
388 now := ts.clock.Time()
389 return ts.ComputeRange(now.Add(-delta), now, num)
390 }
391 392 // extract returns a slice of specified number of observations from a given
393 // level over a given range.
394 func (ts *timeSeries) extract(l *tsLevel, start, finish time.Time, num int, results []Observable) {
395 ts.mergePendingUpdates()
396 397 srcInterval := l.size
398 dstInterval := finish.Sub(start) / time.Duration(num)
399 dstStart := start
400 srcStart := l.end.Add(-srcInterval * time.Duration(ts.numBuckets))
401 402 srcIndex := 0
403 404 // Where should scanning start?
405 if dstStart.After(srcStart) {
406 advance := int(dstStart.Sub(srcStart) / srcInterval)
407 srcIndex += advance
408 srcStart = srcStart.Add(time.Duration(advance) * srcInterval)
409 }
410 411 // The i'th value is computed as show below.
412 // interval = (finish/start)/num
413 // i'th value = sum of observation in range
414 // [ start + i * interval,
415 // start + (i + 1) * interval )
416 for i := 0; i < num; i++ {
417 results[i] = ts.resetObservation(results[i])
418 dstEnd := dstStart.Add(dstInterval)
419 for srcIndex < ts.numBuckets && srcStart.Before(dstEnd) {
420 srcEnd := srcStart.Add(srcInterval)
421 if srcEnd.After(ts.lastAdd) {
422 srcEnd = ts.lastAdd
423 }
424 425 if !srcEnd.Before(dstStart) {
426 srcValue := l.buckets[(srcIndex+l.oldest)%ts.numBuckets]
427 if !srcStart.Before(dstStart) && !srcEnd.After(dstEnd) {
428 // dst completely contains src.
429 if srcValue != nil {
430 results[i].Add(srcValue)
431 }
432 } else {
433 // dst partially overlaps src.
434 overlapStart := maxTime(srcStart, dstStart)
435 overlapEnd := minTime(srcEnd, dstEnd)
436 base := srcEnd.Sub(srcStart)
437 fraction := overlapEnd.Sub(overlapStart).Seconds() / base.Seconds()
438 439 used := ts.provider()
440 if srcValue != nil {
441 used.CopyFrom(srcValue)
442 }
443 used.Multiply(fraction)
444 results[i].Add(used)
445 }
446 447 if srcEnd.After(dstEnd) {
448 break
449 }
450 }
451 srcIndex++
452 srcStart = srcStart.Add(srcInterval)
453 }
454 dstStart = dstStart.Add(dstInterval)
455 }
456 }
457 458 // resetObservation clears the content so the struct may be reused.
459 func (ts *timeSeries) resetObservation(observation Observable) Observable {
460 if observation == nil {
461 observation = ts.provider()
462 } else {
463 observation.Clear()
464 }
465 return observation
466 }
467 468 // TimeSeries tracks data at granularities from 1 second to 16 weeks.
469 type TimeSeries struct {
470 timeSeries
471 }
472 473 // NewTimeSeries creates a new TimeSeries using the function provided for creating new Observable.
474 func NewTimeSeries(f func() Observable) *TimeSeries {
475 return NewTimeSeriesWithClock(f, defaultClockInstance)
476 }
477 478 // NewTimeSeriesWithClock creates a new TimeSeries using the function provided for creating new Observable and the clock for
479 // assigning timestamps.
480 func NewTimeSeriesWithClock(f func() Observable, clock Clock) *TimeSeries {
481 ts := new(TimeSeries)
482 ts.timeSeries.init(timeSeriesResolutions, f, timeSeriesNumBuckets, clock)
483 return ts
484 }
485 486 // MinuteHourSeries tracks data at granularities of 1 minute and 1 hour.
487 type MinuteHourSeries struct {
488 timeSeries
489 }
490 491 // NewMinuteHourSeries creates a new MinuteHourSeries using the function provided for creating new Observable.
492 func NewMinuteHourSeries(f func() Observable) *MinuteHourSeries {
493 return NewMinuteHourSeriesWithClock(f, defaultClockInstance)
494 }
495 496 // NewMinuteHourSeriesWithClock creates a new MinuteHourSeries using the function provided for creating new Observable and the clock for
497 // assigning timestamps.
498 func NewMinuteHourSeriesWithClock(f func() Observable, clock Clock) *MinuteHourSeries {
499 ts := new(MinuteHourSeries)
500 ts.timeSeries.init(minuteHourSeriesResolutions, f,
501 minuteHourSeriesNumBuckets, clock)
502 return ts
503 }
504 505 func (ts *MinuteHourSeries) Minute() Observable {
506 return ts.timeSeries.Latest(0, 60)
507 }
508 509 func (ts *MinuteHourSeries) Hour() Observable {
510 return ts.timeSeries.Latest(1, 60)
511 }
512 513 func minTime(a, b time.Time) time.Time {
514 if a.Before(b) {
515 return a
516 }
517 return b
518 }
519 520 func maxTime(a, b time.Time) time.Time {
521 if a.After(b) {
522 return a
523 }
524 return b
525 }
526