cubefs

1
package rolling
2

3
import (
4
	"sync"
5
	"time"
6
)
7

8
// Number tracks a numberBucket over a bounded number of
9
// time buckets. Currently the buckets are one second long and only the last 10 seconds are kept.
10
type Number struct {
11
	Buckets map[int64]*numberBucket
12
	Mutex   *sync.RWMutex
13
}
14

15
type numberBucket struct {
16
	Value float64
17
}
18

19
// NewNumber initializes a RollingNumber struct.
20
func NewNumber() *Number {
21
	r := &Number{
22
		Buckets: make(map[int64]*numberBucket),
23
		Mutex:   &sync.RWMutex{},
24
	}
25
	return r
26
}
27

28
func (r *Number) getCurrentBucket() *numberBucket {
29
	now := time.Now().Unix()
30
	var bucket *numberBucket
31
	var ok bool
32

33
	if bucket, ok = r.Buckets[now]; !ok {
34
		bucket = &numberBucket{}
35
		r.Buckets[now] = bucket
36
	}
37

38
	return bucket
39
}
40

41
func (r *Number) removeOldBuckets() {
42
	now := time.Now().Unix() - 10
43

44
	for timestamp := range r.Buckets {
45
		// TODO: configurable rolling window
46
		if timestamp <= now {
47
			delete(r.Buckets, timestamp)
48
		}
49
	}
50
}
51

52
// Increment increments the number in current timeBucket.
53
func (r *Number) Increment(i float64) {
54
	if i == 0 {
55
		return
56
	}
57

58
	r.Mutex.Lock()
59
	defer r.Mutex.Unlock()
60

61
	b := r.getCurrentBucket()
62
	b.Value += i
63
	r.removeOldBuckets()
64
}
65

66
// UpdateMax updates the maximum value in the current bucket.
67
func (r *Number) UpdateMax(n float64) {
68
	r.Mutex.Lock()
69
	defer r.Mutex.Unlock()
70

71
	b := r.getCurrentBucket()
72
	if n > b.Value {
73
		b.Value = n
74
	}
75
	r.removeOldBuckets()
76
}
77

78
// Sum sums the values over the buckets in the last 10 seconds.
79
func (r *Number) Sum(now time.Time) float64 {
80
	sum := float64(0)
81

82
	r.Mutex.RLock()
83
	defer r.Mutex.RUnlock()
84

85
	for timestamp, bucket := range r.Buckets {
86
		// TODO: configurable rolling window
87
		if timestamp >= now.Unix()-10 {
88
			sum += bucket.Value
89
		}
90
	}
91

92
	return sum
93
}
94

95
// Max returns the maximum value seen in the last 10 seconds.
96
func (r *Number) Max(now time.Time) float64 {
97
	var max float64
98

99
	r.Mutex.RLock()
100
	defer r.Mutex.RUnlock()
101

102
	for timestamp, bucket := range r.Buckets {
103
		// TODO: configurable rolling window
104
		if timestamp >= now.Unix()-10 {
105
			if bucket.Value > max {
106
				max = bucket.Value
107
			}
108
		}
109
	}
110

111
	return max
112
}
113

114
func (r *Number) Avg(now time.Time) float64 {
115
	return r.Sum(now) / 10
116
}
117