92 lines
2.5 KiB
Go
92 lines
2.5 KiB
Go
package metrics
|
|
|
|
import (
|
|
"math"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
)
|
|
|
|
// EWMASnapshot is a read-only copy of an EWMA.
|
|
type EWMASnapshot float64
|
|
|
|
// Rate returns the rate of events per second at the time the snapshot was
|
|
// taken.
|
|
func (a EWMASnapshot) Rate() float64 { return float64(a) }
|
|
|
|
// NewEWMA constructs a new EWMA with the given alpha.
|
|
func NewEWMA(alpha float64) *EWMA {
|
|
return &EWMA{alpha: alpha}
|
|
}
|
|
|
|
// NewEWMA1 constructs a new EWMA for a one-minute moving average.
|
|
func NewEWMA1() *EWMA {
|
|
return NewEWMA(1 - math.Exp(-5.0/60.0/1))
|
|
}
|
|
|
|
// NewEWMA5 constructs a new EWMA for a five-minute moving average.
|
|
func NewEWMA5() *EWMA {
|
|
return NewEWMA(1 - math.Exp(-5.0/60.0/5))
|
|
}
|
|
|
|
// NewEWMA15 constructs a new EWMA for a fifteen-minute moving average.
|
|
func NewEWMA15() *EWMA {
|
|
return NewEWMA(1 - math.Exp(-5.0/60.0/15))
|
|
}
|
|
|
|
// EWMA continuously calculate an exponentially-weighted moving average
|
|
// based on an outside source of clock ticks.
|
|
type EWMA struct {
|
|
uncounted atomic.Int64
|
|
alpha float64
|
|
rate atomic.Uint64
|
|
init atomic.Bool
|
|
mutex sync.Mutex
|
|
}
|
|
|
|
// Snapshot returns a read-only copy of the EWMA.
|
|
func (a *EWMA) Snapshot() EWMASnapshot {
|
|
r := math.Float64frombits(a.rate.Load()) * float64(time.Second)
|
|
return EWMASnapshot(r)
|
|
}
|
|
|
|
// tick ticks the clock to update the moving average. It assumes it is called
|
|
// every five seconds.
|
|
func (a *EWMA) tick() {
|
|
// Optimization to avoid mutex locking in the hot-path.
|
|
if a.init.Load() {
|
|
a.updateRate(a.fetchInstantRate())
|
|
return
|
|
}
|
|
// Slow-path: this is only needed on the first tick() and preserves transactional updating
|
|
// of init and rate in the else block. The first conditional is needed below because
|
|
// a different thread could have set a.init = 1 between the time of the first atomic load and when
|
|
// the lock was acquired.
|
|
a.mutex.Lock()
|
|
if a.init.Load() {
|
|
// The fetchInstantRate() uses atomic loading, which is unnecessary in this critical section
|
|
// but again, this section is only invoked on the first successful tick() operation.
|
|
a.updateRate(a.fetchInstantRate())
|
|
} else {
|
|
a.init.Store(true)
|
|
a.rate.Store(math.Float64bits(a.fetchInstantRate()))
|
|
}
|
|
a.mutex.Unlock()
|
|
}
|
|
|
|
func (a *EWMA) fetchInstantRate() float64 {
|
|
count := a.uncounted.Swap(0)
|
|
return float64(count) / float64(5*time.Second)
|
|
}
|
|
|
|
func (a *EWMA) updateRate(instantRate float64) {
|
|
currentRate := math.Float64frombits(a.rate.Load())
|
|
currentRate += a.alpha * (instantRate - currentRate)
|
|
a.rate.Store(math.Float64bits(currentRate))
|
|
}
|
|
|
|
// Update adds n uncounted events.
|
|
func (a *EWMA) Update(n int64) {
|
|
a.uncounted.Add(n)
|
|
}
|