traefik/vendor/github.com/vulcand/oxy/memmetrics/counter.go

164 lines
3.8 KiB
Go
Raw Normal View History

2017-02-07 21:33:23 +00:00
package memmetrics
import (
"fmt"
"time"
"github.com/mailgun/timetools"
)
type rcOptSetter func(*RollingCounter) error
// CounterClock defines a counter clock
2017-02-07 21:33:23 +00:00
func CounterClock(c timetools.TimeProvider) rcOptSetter {
return func(r *RollingCounter) error {
r.clock = c
return nil
}
}
// RollingCounter Calculates in memory failure rate of an endpoint using rolling window of a predefined size
2017-02-07 21:33:23 +00:00
type RollingCounter struct {
clock timetools.TimeProvider
resolution time.Duration
values []int
countedBuckets int // how many samples in different buckets have we collected so far
lastBucket int // last recorded bucket
lastUpdated time.Time
}
// NewCounter creates a counter with fixed amount of buckets that are rotated every resolution period.
// E.g. 10 buckets with 1 second means that every new second the bucket is refreshed, so it maintains 10 second rolling window.
// By default creates a bucket with 10 buckets and 1 second resolution
func NewCounter(buckets int, resolution time.Duration, options ...rcOptSetter) (*RollingCounter, error) {
if buckets <= 0 {
return nil, fmt.Errorf("Buckets should be >= 0")
}
if resolution < time.Second {
return nil, fmt.Errorf("Resolution should be larger than a second")
}
rc := &RollingCounter{
lastBucket: -1,
resolution: resolution,
values: make([]int, buckets),
}
for _, o := range options {
if err := o(rc); err != nil {
return nil, err
}
}
if rc.clock == nil {
rc.clock = &timetools.RealTime{}
}
return rc, nil
}
// Append append a counter
2017-02-07 21:33:23 +00:00
func (c *RollingCounter) Append(o *RollingCounter) error {
c.Inc(int(o.Count()))
return nil
}
// Clone clone a counter
2017-02-07 21:33:23 +00:00
func (c *RollingCounter) Clone() *RollingCounter {
c.cleanup()
other := &RollingCounter{
resolution: c.resolution,
values: make([]int, len(c.values)),
clock: c.clock,
lastBucket: c.lastBucket,
lastUpdated: c.lastUpdated,
}
2017-11-22 17:20:03 +00:00
copy(other.values, c.values)
2017-02-07 21:33:23 +00:00
return other
}
// Reset reset a counter
2017-02-07 21:33:23 +00:00
func (c *RollingCounter) Reset() {
c.lastBucket = -1
c.countedBuckets = 0
c.lastUpdated = time.Time{}
for i := range c.values {
c.values[i] = 0
}
}
// CountedBuckets gets counted buckets
2017-02-07 21:33:23 +00:00
func (c *RollingCounter) CountedBuckets() int {
return c.countedBuckets
}
// Count counts
2017-02-07 21:33:23 +00:00
func (c *RollingCounter) Count() int64 {
c.cleanup()
return c.sum()
}
// Resolution gets resolution
2017-02-07 21:33:23 +00:00
func (c *RollingCounter) Resolution() time.Duration {
return c.resolution
}
// Buckets gets buckets
2017-02-07 21:33:23 +00:00
func (c *RollingCounter) Buckets() int {
return len(c.values)
}
// WindowSize gets windows size
2017-02-07 21:33:23 +00:00
func (c *RollingCounter) WindowSize() time.Duration {
return time.Duration(len(c.values)) * c.resolution
}
// Inc increment counter
2017-02-07 21:33:23 +00:00
func (c *RollingCounter) Inc(v int) {
c.cleanup()
c.incBucketValue(v)
}
func (c *RollingCounter) incBucketValue(v int) {
now := c.clock.UtcNow()
bucket := c.getBucket(now)
c.values[bucket] += v
c.lastUpdated = now
// Update usage stats if we haven't collected enough data
if c.countedBuckets < len(c.values) {
// Only update if we have advanced to the next bucket and not incremented the value
// in the current bucket.
if c.lastBucket != bucket {
c.lastBucket = bucket
c.countedBuckets++
}
}
}
// Returns the number in the moving window bucket that this slot occupies
func (c *RollingCounter) getBucket(t time.Time) int {
return int(t.Truncate(c.resolution).Unix() % int64(len(c.values)))
}
// Reset buckets that were not updated
func (c *RollingCounter) cleanup() {
now := c.clock.UtcNow()
for i := 0; i < len(c.values); i++ {
now = now.Add(time.Duration(-1*i) * c.resolution)
if now.Truncate(c.resolution).After(c.lastUpdated.Truncate(c.resolution)) {
c.values[c.getBucket(now)] = 0
} else {
break
}
}
}
func (c *RollingCounter) sum() int64 {
out := int64(0)
for _, v := range c.values {
out += int64(v)
}
return out
}