+++ /dev/null
-// Copyright 2015 The Go Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style
-// license that can be found in the LICENSE file.
-
-// Package timeseries implements a time series structure for stats collection.
-package timeseries // import "golang.org/x/net/internal/timeseries"
-
-import (
- "fmt"
- "log"
- "time"
-)
-
-const (
- timeSeriesNumBuckets = 64
- minuteHourSeriesNumBuckets = 60
-)
-
-var timeSeriesResolutions = []time.Duration{
- 1 * time.Second,
- 10 * time.Second,
- 1 * time.Minute,
- 10 * time.Minute,
- 1 * time.Hour,
- 6 * time.Hour,
- 24 * time.Hour, // 1 day
- 7 * 24 * time.Hour, // 1 week
- 4 * 7 * 24 * time.Hour, // 4 weeks
- 16 * 7 * 24 * time.Hour, // 16 weeks
-}
-
-var minuteHourSeriesResolutions = []time.Duration{
- 1 * time.Second,
- 1 * time.Minute,
-}
-
-// An Observable is a kind of data that can be aggregated in a time series.
-type Observable interface {
- Multiply(ratio float64) // Multiplies the data in self by a given ratio
- Add(other Observable) // Adds the data from a different observation to self
- Clear() // Clears the observation so it can be reused.
- CopyFrom(other Observable) // Copies the contents of a given observation to self
-}
-
-// Float attaches the methods of Observable to a float64.
-type Float float64
-
-// NewFloat returns a Float.
-func NewFloat() Observable {
- f := Float(0)
- return &f
-}
-
-// String returns the float as a string.
-func (f *Float) String() string { return fmt.Sprintf("%g", f.Value()) }
-
-// Value returns the float's value.
-func (f *Float) Value() float64 { return float64(*f) }
-
-func (f *Float) Multiply(ratio float64) { *f *= Float(ratio) }
-
-func (f *Float) Add(other Observable) {
- o := other.(*Float)
- *f += *o
-}
-
-func (f *Float) Clear() { *f = 0 }
-
-func (f *Float) CopyFrom(other Observable) {
- o := other.(*Float)
- *f = *o
-}
-
-// A Clock tells the current time.
-type Clock interface {
- Time() time.Time
-}
-
-type defaultClock int
-
-var defaultClockInstance defaultClock
-
-func (defaultClock) Time() time.Time { return time.Now() }
-
-// Information kept per level. Each level consists of a circular list of
-// observations. The start of the level may be derived from end and the
-// len(buckets) * sizeInMillis.
-type tsLevel struct {
- oldest int // index to oldest bucketed Observable
- newest int // index to newest bucketed Observable
- end time.Time // end timestamp for this level
- size time.Duration // duration of the bucketed Observable
- buckets []Observable // collections of observations
- provider func() Observable // used for creating new Observable
-}
-
-func (l *tsLevel) Clear() {
- l.oldest = 0
- l.newest = len(l.buckets) - 1
- l.end = time.Time{}
- for i := range l.buckets {
- if l.buckets[i] != nil {
- l.buckets[i].Clear()
- l.buckets[i] = nil
- }
- }
-}
-
-func (l *tsLevel) InitLevel(size time.Duration, numBuckets int, f func() Observable) {
- l.size = size
- l.provider = f
- l.buckets = make([]Observable, numBuckets)
-}
-
-// Keeps a sequence of levels. Each level is responsible for storing data at
-// a given resolution. For example, the first level stores data at a one
-// minute resolution while the second level stores data at a one hour
-// resolution.
-
-// Each level is represented by a sequence of buckets. Each bucket spans an
-// interval equal to the resolution of the level. New observations are added
-// to the last bucket.
-type timeSeries struct {
- provider func() Observable // make more Observable
- numBuckets int // number of buckets in each level
- levels []*tsLevel // levels of bucketed Observable
- lastAdd time.Time // time of last Observable tracked
- total Observable // convenient aggregation of all Observable
- clock Clock // Clock for getting current time
- pending Observable // observations not yet bucketed
- pendingTime time.Time // what time are we keeping in pending
- dirty bool // if there are pending observations
-}
-
-// init initializes a level according to the supplied criteria.
-func (ts *timeSeries) init(resolutions []time.Duration, f func() Observable, numBuckets int, clock Clock) {
- ts.provider = f
- ts.numBuckets = numBuckets
- ts.clock = clock
- ts.levels = make([]*tsLevel, len(resolutions))
-
- for i := range resolutions {
- if i > 0 && resolutions[i-1] >= resolutions[i] {
- log.Print("timeseries: resolutions must be monotonically increasing")
- break
- }
- newLevel := new(tsLevel)
- newLevel.InitLevel(resolutions[i], ts.numBuckets, ts.provider)
- ts.levels[i] = newLevel
- }
-
- ts.Clear()
-}
-
-// Clear removes all observations from the time series.
-func (ts *timeSeries) Clear() {
- ts.lastAdd = time.Time{}
- ts.total = ts.resetObservation(ts.total)
- ts.pending = ts.resetObservation(ts.pending)
- ts.pendingTime = time.Time{}
- ts.dirty = false
-
- for i := range ts.levels {
- ts.levels[i].Clear()
- }
-}
-
-// Add records an observation at the current time.
-func (ts *timeSeries) Add(observation Observable) {
- ts.AddWithTime(observation, ts.clock.Time())
-}
-
-// AddWithTime records an observation at the specified time.
-func (ts *timeSeries) AddWithTime(observation Observable, t time.Time) {
-
- smallBucketDuration := ts.levels[0].size
-
- if t.After(ts.lastAdd) {
- ts.lastAdd = t
- }
-
- if t.After(ts.pendingTime) {
- ts.advance(t)
- ts.mergePendingUpdates()
- ts.pendingTime = ts.levels[0].end
- ts.pending.CopyFrom(observation)
- ts.dirty = true
- } else if t.After(ts.pendingTime.Add(-1 * smallBucketDuration)) {
- // The observation is close enough to go into the pending bucket.
- // This compensates for clock skewing and small scheduling delays
- // by letting the update stay in the fast path.
- ts.pending.Add(observation)
- ts.dirty = true
- } else {
- ts.mergeValue(observation, t)
- }
-}
-
-// mergeValue inserts the observation at the specified time in the past into all levels.
-func (ts *timeSeries) mergeValue(observation Observable, t time.Time) {
- for _, level := range ts.levels {
- index := (ts.numBuckets - 1) - int(level.end.Sub(t)/level.size)
- if 0 <= index && index < ts.numBuckets {
- bucketNumber := (level.oldest + index) % ts.numBuckets
- if level.buckets[bucketNumber] == nil {
- level.buckets[bucketNumber] = level.provider()
- }
- level.buckets[bucketNumber].Add(observation)
- }
- }
- ts.total.Add(observation)
-}
-
-// mergePendingUpdates applies the pending updates into all levels.
-func (ts *timeSeries) mergePendingUpdates() {
- if ts.dirty {
- ts.mergeValue(ts.pending, ts.pendingTime)
- ts.pending = ts.resetObservation(ts.pending)
- ts.dirty = false
- }
-}
-
-// advance cycles the buckets at each level until the latest bucket in
-// each level can hold the time specified.
-func (ts *timeSeries) advance(t time.Time) {
- if !t.After(ts.levels[0].end) {
- return
- }
- for i := 0; i < len(ts.levels); i++ {
- level := ts.levels[i]
- if !level.end.Before(t) {
- break
- }
-
- // If the time is sufficiently far, just clear the level and advance
- // directly.
- if !t.Before(level.end.Add(level.size * time.Duration(ts.numBuckets))) {
- for _, b := range level.buckets {
- ts.resetObservation(b)
- }
- level.end = time.Unix(0, (t.UnixNano()/level.size.Nanoseconds())*level.size.Nanoseconds())
- }
-
- for t.After(level.end) {
- level.end = level.end.Add(level.size)
- level.newest = level.oldest
- level.oldest = (level.oldest + 1) % ts.numBuckets
- ts.resetObservation(level.buckets[level.newest])
- }
-
- t = level.end
- }
-}
-
-// Latest returns the sum of the num latest buckets from the level.
-func (ts *timeSeries) Latest(level, num int) Observable {
- now := ts.clock.Time()
- if ts.levels[0].end.Before(now) {
- ts.advance(now)
- }
-
- ts.mergePendingUpdates()
-
- result := ts.provider()
- l := ts.levels[level]
- index := l.newest
-
- for i := 0; i < num; i++ {
- if l.buckets[index] != nil {
- result.Add(l.buckets[index])
- }
- if index == 0 {
- index = ts.numBuckets
- }
- index--
- }
-
- return result
-}
-
-// LatestBuckets returns a copy of the num latest buckets from level.
-func (ts *timeSeries) LatestBuckets(level, num int) []Observable {
- if level < 0 || level > len(ts.levels) {
- log.Print("timeseries: bad level argument: ", level)
- return nil
- }
- if num < 0 || num >= ts.numBuckets {
- log.Print("timeseries: bad num argument: ", num)
- return nil
- }
-
- results := make([]Observable, num)
- now := ts.clock.Time()
- if ts.levels[0].end.Before(now) {
- ts.advance(now)
- }
-
- ts.mergePendingUpdates()
-
- l := ts.levels[level]
- index := l.newest
-
- for i := 0; i < num; i++ {
- result := ts.provider()
- results[i] = result
- if l.buckets[index] != nil {
- result.CopyFrom(l.buckets[index])
- }
-
- if index == 0 {
- index = ts.numBuckets
- }
- index -= 1
- }
- return results
-}
-
-// ScaleBy updates observations by scaling by factor.
-func (ts *timeSeries) ScaleBy(factor float64) {
- for _, l := range ts.levels {
- for i := 0; i < ts.numBuckets; i++ {
- l.buckets[i].Multiply(factor)
- }
- }
-
- ts.total.Multiply(factor)
- ts.pending.Multiply(factor)
-}
-
-// Range returns the sum of observations added over the specified time range.
-// If start or finish times don't fall on bucket boundaries of the same
-// level, then return values are approximate answers.
-func (ts *timeSeries) Range(start, finish time.Time) Observable {
- return ts.ComputeRange(start, finish, 1)[0]
-}
-
-// Recent returns the sum of observations from the last delta.
-func (ts *timeSeries) Recent(delta time.Duration) Observable {
- now := ts.clock.Time()
- return ts.Range(now.Add(-delta), now)
-}
-
-// Total returns the total of all observations.
-func (ts *timeSeries) Total() Observable {
- ts.mergePendingUpdates()
- return ts.total
-}
-
-// ComputeRange computes a specified number of values into a slice using
-// the observations recorded over the specified time period. The return
-// values are approximate if the start or finish times don't fall on the
-// bucket boundaries at the same level or if the number of buckets spanning
-// the range is not an integral multiple of num.
-func (ts *timeSeries) ComputeRange(start, finish time.Time, num int) []Observable {
- if start.After(finish) {
- log.Printf("timeseries: start > finish, %v>%v", start, finish)
- return nil
- }
-
- if num < 0 {
- log.Printf("timeseries: num < 0, %v", num)
- return nil
- }
-
- results := make([]Observable, num)
-
- for _, l := range ts.levels {
- if !start.Before(l.end.Add(-l.size * time.Duration(ts.numBuckets))) {
- ts.extract(l, start, finish, num, results)
- return results
- }
- }
-
- // Failed to find a level that covers the desired range. So just
- // extract from the last level, even if it doesn't cover the entire
- // desired range.
- ts.extract(ts.levels[len(ts.levels)-1], start, finish, num, results)
-
- return results
-}
-
-// RecentList returns the specified number of values in slice over the most
-// recent time period of the specified range.
-func (ts *timeSeries) RecentList(delta time.Duration, num int) []Observable {
- if delta < 0 {
- return nil
- }
- now := ts.clock.Time()
- return ts.ComputeRange(now.Add(-delta), now, num)
-}
-
-// extract returns a slice of specified number of observations from a given
-// level over a given range.
-func (ts *timeSeries) extract(l *tsLevel, start, finish time.Time, num int, results []Observable) {
- ts.mergePendingUpdates()
-
- srcInterval := l.size
- dstInterval := finish.Sub(start) / time.Duration(num)
- dstStart := start
- srcStart := l.end.Add(-srcInterval * time.Duration(ts.numBuckets))
-
- srcIndex := 0
-
- // Where should scanning start?
- if dstStart.After(srcStart) {
- advance := dstStart.Sub(srcStart) / srcInterval
- srcIndex += int(advance)
- srcStart = srcStart.Add(advance * srcInterval)
- }
-
- // The i'th value is computed as show below.
- // interval = (finish/start)/num
- // i'th value = sum of observation in range
- // [ start + i * interval,
- // start + (i + 1) * interval )
- for i := 0; i < num; i++ {
- results[i] = ts.resetObservation(results[i])
- dstEnd := dstStart.Add(dstInterval)
- for srcIndex < ts.numBuckets && srcStart.Before(dstEnd) {
- srcEnd := srcStart.Add(srcInterval)
- if srcEnd.After(ts.lastAdd) {
- srcEnd = ts.lastAdd
- }
-
- if !srcEnd.Before(dstStart) {
- srcValue := l.buckets[(srcIndex+l.oldest)%ts.numBuckets]
- if !srcStart.Before(dstStart) && !srcEnd.After(dstEnd) {
- // dst completely contains src.
- if srcValue != nil {
- results[i].Add(srcValue)
- }
- } else {
- // dst partially overlaps src.
- overlapStart := maxTime(srcStart, dstStart)
- overlapEnd := minTime(srcEnd, dstEnd)
- base := srcEnd.Sub(srcStart)
- fraction := overlapEnd.Sub(overlapStart).Seconds() / base.Seconds()
-
- used := ts.provider()
- if srcValue != nil {
- used.CopyFrom(srcValue)
- }
- used.Multiply(fraction)
- results[i].Add(used)
- }
-
- if srcEnd.After(dstEnd) {
- break
- }
- }
- srcIndex++
- srcStart = srcStart.Add(srcInterval)
- }
- dstStart = dstStart.Add(dstInterval)
- }
-}
-
-// resetObservation clears the content so the struct may be reused.
-func (ts *timeSeries) resetObservation(observation Observable) Observable {
- if observation == nil {
- observation = ts.provider()
- } else {
- observation.Clear()
- }
- return observation
-}
-
-// TimeSeries tracks data at granularities from 1 second to 16 weeks.
-type TimeSeries struct {
- timeSeries
-}
-
-// NewTimeSeries creates a new TimeSeries using the function provided for creating new Observable.
-func NewTimeSeries(f func() Observable) *TimeSeries {
- return NewTimeSeriesWithClock(f, defaultClockInstance)
-}
-
-// NewTimeSeriesWithClock creates a new TimeSeries using the function provided for creating new Observable and the clock for
-// assigning timestamps.
-func NewTimeSeriesWithClock(f func() Observable, clock Clock) *TimeSeries {
- ts := new(TimeSeries)
- ts.timeSeries.init(timeSeriesResolutions, f, timeSeriesNumBuckets, clock)
- return ts
-}
-
-// MinuteHourSeries tracks data at granularities of 1 minute and 1 hour.
-type MinuteHourSeries struct {
- timeSeries
-}
-
-// NewMinuteHourSeries creates a new MinuteHourSeries using the function provided for creating new Observable.
-func NewMinuteHourSeries(f func() Observable) *MinuteHourSeries {
- return NewMinuteHourSeriesWithClock(f, defaultClockInstance)
-}
-
-// NewMinuteHourSeriesWithClock creates a new MinuteHourSeries using the function provided for creating new Observable and the clock for
-// assigning timestamps.
-func NewMinuteHourSeriesWithClock(f func() Observable, clock Clock) *MinuteHourSeries {
- ts := new(MinuteHourSeries)
- ts.timeSeries.init(minuteHourSeriesResolutions, f,
- minuteHourSeriesNumBuckets, clock)
- return ts
-}
-
-func (ts *MinuteHourSeries) Minute() Observable {
- return ts.timeSeries.Latest(0, 60)
-}
-
-func (ts *MinuteHourSeries) Hour() Observable {
- return ts.timeSeries.Latest(1, 60)
-}
-
-func minTime(a, b time.Time) time.Time {
- if a.Before(b) {
- return a
- }
- return b
-}
-
-func maxTime(a, b time.Time) time.Time {
- if a.After(b) {
- return a
- }
- return b
-}
OSDN Git Service
