Apply configs to NewMonitor

Author: justinjung04

integration/resource_based_limiter_test.go

@@ -19,7 +19,7 @@ func Test_ResourceBasedLimiter_shouldStartWithoutError(t *testing.T) {
 	defer s.Close()
 
 	flags := mergeFlags(BlocksStorageFlags(), map[string]string{
-		"-monitored.resources": "cpu,heap",
+		"-resource-monitor.resources": "cpu,heap",
 	})
 
 	// Start dependencies.

pkg/configs/resource_monitor.go

@@ -40,8 +40,8 @@ func (cfg *ResourceMonitor) Validate() error {
 		return fmt.Errorf("resource monitor interval must be greater than zero")
 	}
 
-	if cfg.CPURateInterval <= 0 {
-		return fmt.Errorf("resource monitor cpu rate interval must be greater than zero")
+	if cfg.CPURateInterval < cfg.Interval {
+		return fmt.Errorf("resource monitor cpu rate interval cannot be smaller than resource monitor interval")
 	}
 
 	return nil

pkg/cortex/cortex_test.go

@@ -195,11 +195,11 @@ func TestConfigValidation(t *testing.T) {
 				configuration := newDefaultConfig()
 				configuration.ResourceMonitor = configs.ResourceMonitor{
 					Interval:        time.Second,
-					CPURateInterval: -1,
+					CPURateInterval: time.Millisecond,
 				}
 				return configuration
 			},
-			expectedError: fmt.Errorf("resource monitor cpu rate interval must be greater than zero"),
+			expectedError: fmt.Errorf("resource monitor cpu rate interval cannot be smaller than resource monitor interval"),
 		},
 		{
 			name: "should not fail validation for valid resources to monitor",

pkg/cortex/modules.go

@@ -805,7 +805,7 @@ func (t *Cortex) initResourceMonitor() (services.Service, error) {
 	}
 
 	var err error
-	t.ResourceMonitor, err = resource.NewMonitor(containerLimits, prometheus.DefaultRegisterer)
+	t.ResourceMonitor, err = resource.NewMonitor(containerLimits, t.Cfg.ResourceMonitor.Interval, t.Cfg.ResourceMonitor.CPURateInterval, prometheus.DefaultRegisterer)
 	return t.ResourceMonitor, err
 }
 

pkg/util/resource/monitor.go

@@ -15,9 +15,6 @@ import (
 const (
 	CPU  Type = "cpu"
 	Heap Type = "heap"
-
-	monitorInterval = 100 * time.Millisecond
-	dataPointsToAvg = 50
 )
 
 type Type string
@@ -33,31 +30,36 @@ type Monitor struct {
 	scanners       map[Type]scanner
 	containerLimit map[Type]float64
 	utilization    map[Type]float64
+	interval       time.Duration
 
 	// Variables to calculate average CPU utilization
 	index         int
-	cpuRates      [dataPointsToAvg]float64
-	cpuIntervals  [dataPointsToAvg]float64
+	cpuRates      []float64
+	cpuIntervals  []float64
 	totalCPU      float64
 	totalInterval float64
 	lastCPU       float64
 	lastUpdate    time.Time
+	cpuDataPoints int
 
 	lock sync.RWMutex
 }
 
-func NewMonitor(limits map[Type]float64, registerer prometheus.Registerer) (*Monitor, error) {
+func NewMonitor(limits map[Type]float64, interval, cpuAverageInterval time.Duration, registerer prometheus.Registerer) (*Monitor, error) {
 	m := &Monitor{
 		containerLimit: limits,
 		scanners:       make(map[Type]scanner),
 		utilization:    make(map[Type]float64),
-
-		cpuRates:     [dataPointsToAvg]float64{},
-		cpuIntervals: [dataPointsToAvg]float64{},
+		interval:       interval,
 
 		lock: sync.RWMutex{},
 	}
 
+	m.interval = interval
+	m.cpuDataPoints = int(cpuAverageInterval.Nanoseconds() / interval.Nanoseconds())
+	m.cpuRates = make([]float64, m.cpuDataPoints)
+	m.cpuIntervals = make([]float64, m.cpuDataPoints)
+
 	m.Service = services.NewBasicService(nil, m.running, nil)
 
 	for resType, limit := range limits {
@@ -92,7 +94,7 @@ func NewMonitor(limits map[Type]float64, registerer prometheus.Registerer) (*Mon
 }
 
 func (m *Monitor) running(ctx context.Context) error {
-	ticker := time.NewTicker(monitorInterval)
+	ticker := time.NewTicker(m.interval)
 	defer ticker.Stop()
 
 	for {
@@ -141,7 +143,7 @@ func (m *Monitor) storeCPUUtilization(cpuTime float64) {
 
 	m.lastCPU = cpuTime
 	m.lastUpdate = now
-	m.index = (m.index + 1) % dataPointsToAvg
+	m.index = (m.index + 1) % m.cpuDataPoints
 
 	if m.totalInterval > 0 && m.containerLimit[CPU] > 0 {
 		m.utilization[CPU] = m.totalCPU / m.totalInterval / m.containerLimit[CPU]

pkg/util/resource/monitor_test.go

@@ -2,13 +2,14 @@ package resource
 
 import (
 	"testing"
+	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/require"
 )
 
 func Test_Monitor(t *testing.T) {
-	m, err := NewMonitor(map[Type]float64{}, prometheus.DefaultRegisterer)
+	m, err := NewMonitor(map[Type]float64{}, time.Second, time.Minute, prometheus.DefaultRegisterer)
 
 	m.scanners[CPU] = &noopScanner{}
 	m.containerLimit[CPU] = 1