Add ResourceMonitor module in Cortex, and add ResourceBasedLimiter in Ingesters and StoreGateways

pkg/configs/instance_limits.go

@@ -20,10 +20,18 @@ func (cfg *InstanceLimits) RegisterFlagsWithPrefix(f *flag.FlagSet, prefix strin
 }
 
 func (cfg *InstanceLimits) Validate(monitoredResources flagext.StringSliceCSV) error {
+	if cfg.CPUUtilization > 1 || cfg.CPUUtilization < 0 {
+		return errors.New("cpu_utilization must be between 0 and 1")
+	}
+
 	if cfg.CPUUtilization > 0 && !strings.Contains(monitoredResources.String(), string(resource.CPU)) {
 		return errors.New("monitored_resources config must include \"cpu\" as well")
 	}
 
+	if cfg.HeapUtilization > 1 || cfg.HeapUtilization < 0 {
+		return errors.New("heap_utilization must be between 0 and 1")
+	}
+
 	if cfg.HeapUtilization > 0 && !strings.Contains(monitoredResources.String(), string(resource.Heap)) {
 		return errors.New("monitored_resources config must include \"heap\" as well")
 	}

pkg/configs/instance_limits_test.go

@@ -0,0 +1,64 @@
+package configs
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func Test_Validate(t *testing.T) {
+	for name, tc := range map[string]struct {
+		instanceLimits     InstanceLimits
+		monitoredResources []string
+		err                error
+	}{
+		"correct config should pass validation": {
+			instanceLimits: InstanceLimits{
+				CPUUtilization:  0.5,
+				HeapUtilization: 0.5,
+			},
+			monitoredResources: []string{"cpu", "heap"},
+			err:                nil,
+		},
+		"utilization config less than 0 should fail validation": {
+			instanceLimits: InstanceLimits{
+				CPUUtilization:  -0.5,
+				HeapUtilization: 0.5,
+			},
+			monitoredResources: []string{"cpu", "heap"},
+			err:                errors.New("cpu_utilization must be between 0 and 1"),
+		},
+		"utilization config greater than 1 should fail validation": {
+			instanceLimits: InstanceLimits{
+				CPUUtilization:  0.5,
+				HeapUtilization: 1.5,
+			},
+			monitoredResources: []string{"cpu", "heap"},
+			err:                errors.New("heap_utilization must be between 0 and 1"),
+		},
+		"missing cpu in monitored_resources config should fail validation": {
+			instanceLimits: InstanceLimits{
+				CPUUtilization: 0.5,
+			},
+			monitoredResources: []string{"heap"},
+			err:                errors.New("monitored_resources config must include \"cpu\" as well"),
+		},
+		"missing heap in monitored_resources config should fail validation": {
+			instanceLimits: InstanceLimits{
+				HeapUtilization: 0.5,
+			},
+			monitoredResources: []string{"cpu"},
+			err:                errors.New("monitored_resources config must include \"heap\" as well"),
+		},
+	} {
+		t.Run(name, func(t *testing.T) {
+			err := tc.instanceLimits.Validate(tc.monitoredResources)
+			if tc.err != nil {
+				require.Errorf(t, err, tc.err.Error())
+			} else {
+				require.NoError(t, err)
+			}
+		})
+	}
+}

pkg/ingester/ingester.go

@@ -790,7 +790,7 @@ func New(cfg Config, limits *validation.Overrides, registerer prometheus.Registe
 		if cfg.DefaultLimits.HeapUtilization > 0 {
 			resourceLimits[resource.Heap] = cfg.DefaultLimits.HeapUtilization
 		}
-		i.resourceBasedLimiter, err = limiter.NewResourceBasedLimiter(resourceMonitor, resourceLimits, registerer)
+		i.resourceBasedLimiter, err = limiter.NewResourceBasedLimiter(resourceMonitor, resourceLimits, registerer, "ingester")
 		if err != nil {
 			return nil, errors.Wrap(err, "error creating resource based limiter")
 		}

pkg/ingester/ingester_test.go

@@ -3078,7 +3078,7 @@ func Test_Ingester_Query_ResourceThresholdBreached(t *testing.T) {
 		resource.CPU:  0.5,
 		resource.Heap: 0.5,
 	}
-	i.resourceBasedLimiter, err = limiter.NewResourceBasedLimiter(&mockResourceMonitor{cpu: 0.4, heap: 0.6}, limits, nil)
+	i.resourceBasedLimiter, err = limiter.NewResourceBasedLimiter(&mockResourceMonitor{cpu: 0.4, heap: 0.6}, limits, nil, "ingester")
 	require.NoError(t, err)
 
 	// Wait until it's ACTIVE

pkg/storegateway/gateway.go

@@ -250,7 +250,7 @@ func newStoreGateway(gatewayCfg Config, storageCfg cortex_tsdb.BlocksStorageConf
 		if gatewayCfg.InstanceLimits.HeapUtilization > 0 {
 			resourceLimits[resource.Heap] = gatewayCfg.InstanceLimits.HeapUtilization
 		}
-		g.resourceBasedLimiter, err = util_limiter.NewResourceBasedLimiter(resourceMonitor, resourceLimits, reg)
+		g.resourceBasedLimiter, err = util_limiter.NewResourceBasedLimiter(resourceMonitor, resourceLimits, reg, "store-gateway")
 		if err != nil {
 			return nil, errors.Wrap(err, "error creating resource based limiter")
 		}

pkg/storegateway/gateway_test.go

@@ -1220,7 +1220,7 @@ func TestStoreGateway_SeriesThrottledByResourceMonitor(t *testing.T) {
 		resource.CPU:  0.5,
 		resource.Heap: 0.5,
 	}
-	g.resourceBasedLimiter, err = util_limiter.NewResourceBasedLimiter(&mockResourceMonitor{cpu: 0.4, heap: 0.6}, limits, nil)
+	g.resourceBasedLimiter, err = util_limiter.NewResourceBasedLimiter(&mockResourceMonitor{cpu: 0.4, heap: 0.6}, limits, nil, "store-gateway")
 	require.NoError(t, err)
 
 	srv := newBucketStoreSeriesServer(setUserIDToGRPCContext(ctx, userID))

pkg/util/limiter/resource_based_limiter.go

@@ -23,13 +23,14 @@ type ResourceBasedLimiter struct {
 	limitBreachedCount *prometheus.CounterVec
 }
 
-func NewResourceBasedLimiter(resourceMonitor resource.IMonitor, limits map[resource.Type]float64, registerer prometheus.Registerer) (*ResourceBasedLimiter, error) {
+func NewResourceBasedLimiter(resourceMonitor resource.IMonitor, limits map[resource.Type]float64, registerer prometheus.Registerer, component string) (*ResourceBasedLimiter, error) {
 	for resType, limit := range limits {
 		switch resType {
 		case resource.CPU, resource.Heap:
 			promauto.With(registerer).NewGauge(prometheus.GaugeOpts{
 				Name:        "cortex_resource_based_limiter_limit",
-				ConstLabels: map[string]string{"resource": string(resType)},
+				Help:        "Limit set for the resource utilization.",
+				ConstLabels: map[string]string{"component": component},
 			}).Set(limit)
 		default:
 			return nil, fmt.Errorf("unsupported resource type: [%s]", resType)
@@ -41,8 +42,9 @@ func NewResourceBasedLimiter(resourceMonitor resource.IMonitor, limits map[resou
 		limits:          limits,
 		limitBreachedCount: promauto.With(registerer).NewCounterVec(
 			prometheus.CounterOpts{
-				Name: "cortex_resource_based_limiter_limit_breached",
-				Help: "The total number of times resource based limiter was throttled.",
+				Name:        "cortex_resource_based_limiter_limit_breached",
+				Help:        "The total number of times resource based limiter was throttled.",
+				ConstLabels: map[string]string{"component": component},
 			},
 			[]string{"resource"},
 		),