Add warmup counter to EWMA rate

pkg/util/limiter/utilization.go

@@ -74,8 +74,6 @@ type UtilizationBasedLimiter struct {
 	cpuLimit float64
 	// Last CPU utilization time counter.
 	lastCPUTime float64
-	// The time of the first CPU update.
-	firstCPUUpdate time.Time
 	// The time of the last CPU update.
 	lastCPUUpdate  time.Time
 	cpuMovingAvg   *math.EwmaRate
@@ -187,14 +185,10 @@ func (l *UtilizationBasedLimiter) compute(nowFn func() time.Time) (currCPUUtil f
 	}
 
 	// The CPU utilization moving average requires a warmup period before getting
-	// stable results. In this implementation we use a warmup period equal to the
-	// sliding window. During the warmup, the reported CPU utilization will be 0.
-	if l.firstCPUUpdate.IsZero() {
-		l.firstCPUUpdate = now
-	} else if now.Sub(l.firstCPUUpdate) >= resourceUtilizationSlidingWindow {
-		currCPUUtil = l.cpuMovingAvg.Rate() / 100
-		l.currCPUUtil.Store(currCPUUtil)
-	}
+	// stable results. The EWMA rate assumes a warmup period of N ticks (currently N=60).
+	// During the warmup, the reported CPU utilization will be 0.
+	currCPUUtil = l.cpuMovingAvg.Rate() / 100
+	l.currCPUUtil.Store(currCPUUtil)
 
 	var reason string
 	if l.memoryLimit > 0 && currHeapSize >= l.memoryLimit {

pkg/util/math/rate.go

@@ -12,29 +12,50 @@ import (
 	"go.uber.org/atomic"
 )
 
+const (
+	defaultWarmupSamples uint8 = 60
+)
+
 // EwmaRate tracks an exponentially weighted moving average of a per-second rate.
 type EwmaRate struct {
 	newEvents atomic.Int64
 
 	alpha    float64
 	interval time.Duration
 
-	mutex    sync.RWMutex
-	lastRate float64
-	init     bool
+	mutex         sync.RWMutex
+	lastRate      float64
+	init          bool
+	count         uint8
+	warmupSamples uint8
 }
 
 func NewEWMARate(alpha float64, interval time.Duration) *EwmaRate {
 	return &EwmaRate{
-		alpha:    alpha,
-		interval: interval,
+		alpha:         alpha,
+		interval:      interval,
+		warmupSamples: defaultWarmupSamples,
+	}
+}
+
+func NewEWMARateWithWarmup(alpha float64, interval time.Duration, warmupSamples uint8) *EwmaRate {
+	return &EwmaRate{
+		alpha:         alpha,
+		interval:      interval,
+		warmupSamples: warmupSamples,
 	}
 }
 
 // Rate returns the per-second rate.
 func (r *EwmaRate) Rate() float64 {
 	r.mutex.RLock()
 	defer r.mutex.RUnlock()
+
+	// until the first `warmupSamples` have been seen, the moving average is "not ready" to be queried
+	if r.count < r.warmupSamples {
+		return 0.0
+	}
+
 	return r.lastRate
 }
 
@@ -46,6 +67,10 @@ func (r *EwmaRate) Tick() {
 	r.mutex.Lock()
 	defer r.mutex.Unlock()
 
+	if r.count < r.warmupSamples {
+		r.count++
+	}
+
 	if r.init {
 		r.lastRate += r.alpha * (instantRate - r.lastRate)
 	} else {

pkg/util/math/rate_test.go

@@ -12,11 +12,14 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestRate(t *testing.T) {
-	ticks := []struct {
-		events int
-		want   float64
-	}{
+type tickTest struct {
+	events int
+	want   float64
+}
+
+func TestRate_WithDefaultWarmupSamples(t *testing.T) {
+	ticks := populateWarmupSamples(defaultWarmupSamples)
+	ticks = append(ticks, []tickTest{
 		{60, 1},
 		{30, 0.9},
 		{0, 0.72},
@@ -28,7 +31,7 @@ func TestRate(t *testing.T) {
 		{0, 0.25427968},
 		{0, 0.203423744},
 		{0, 0.1627389952},
-	}
+	}...)
 	r := NewEWMARate(0.2, time.Minute)
 
 	for _, tick := range ticks {
@@ -50,3 +53,49 @@ func TestRate(t *testing.T) {
 		require.InDelta(t, tick.want, r.Rate(), 0.0000000001, "unexpected rate")
 	}
 }
+
+func TestRate_WithSpecifiedWarmupSamples(t *testing.T) {
+	const warmupSamples uint8 = 10
+	ticks := populateWarmupSamples(warmupSamples)
+	ticks = append(ticks, []tickTest{
+		{60, 1},
+		{30, 0.9},
+		{0, 0.72},
+		{60, 0.776},
+		{0, 0.6208},
+		{0, 0.49664},
+		{0, 0.397312},
+		{0, 0.3178496},
+		{0, 0.25427968},
+		{0, 0.203423744},
+		{0, 0.1627389952},
+	}...)
+	r := NewEWMARateWithWarmup(0.2, time.Minute, warmupSamples)
+
+	for _, tick := range ticks {
+		for e := 0; e < tick.events; e++ {
+			r.Inc()
+		}
+		r.Tick()
+		// We cannot do double comparison, because double operations on different
+		// platforms may actually produce results that differ slightly.
+		// There are multiple issues about this in Go's github, eg: 18354 or 20319.
+		require.InDelta(t, tick.want, r.Rate(), 0.0000000001, "unexpected rate")
+	}
+
+	r = NewEWMARateWithWarmup(0.2, time.Minute, warmupSamples)
+
+	for _, tick := range ticks {
+		r.Add(int64(tick.events))
+		r.Tick()
+		require.InDelta(t, tick.want, r.Rate(), 0.0000000001, "unexpected rate")
+	}
+}
+
+func populateWarmupSamples(warmupSamples uint8) []tickTest {
+	samples := make([]tickTest, 0, warmupSamples-1)
+	for range warmupSamples - 1 {
+		samples = append(samples, tickTest{60, 0})
+	}
+	return samples
+}