Implement scale-down prevention based on resource quotas (#9598)

* Implement scale-down protection based on min limits

* Fix linting

Author: 4llower

cluster-autoscaler/core/scaledown/planner/planner.go

@@ -28,11 +28,11 @@ import (
 	"k8s.io/autoscaler/cluster-autoscaler/core/scaledown/eligibility"
 	"k8s.io/autoscaler/cluster-autoscaler/core/scaledown/nodeevaltracker"
 	"k8s.io/autoscaler/cluster-autoscaler/core/scaledown/pdb"
-	"k8s.io/autoscaler/cluster-autoscaler/core/scaledown/resource"
 	"k8s.io/autoscaler/cluster-autoscaler/core/scaledown/unneeded"
 	"k8s.io/autoscaler/cluster-autoscaler/core/scaledown/unremovable"
 	"k8s.io/autoscaler/cluster-autoscaler/processors"
 	"k8s.io/autoscaler/cluster-autoscaler/processors/nodes"
+	"k8s.io/autoscaler/cluster-autoscaler/resourcequotas"
 	"k8s.io/autoscaler/cluster-autoscaler/simulator"
 	"k8s.io/autoscaler/cluster-autoscaler/simulator/clustersnapshot"
 	"k8s.io/autoscaler/cluster-autoscaler/simulator/drainability/rules"
@@ -73,22 +73,21 @@ type Planner struct {
 	minUpdateInterval     time.Duration
 	eligibilityChecker    eligibilityChecker
 	nodeUtilizationMap    map[string]utilization.Info
-	resourceLimitsFinder  *resource.LimitsFinder
 	cc                    controllerReplicasCalculator
+	quotasTrackerFactory  *resourcequotas.TrackerFactory
 	scaleDownSetProcessor nodes.ScaleDownSetProcessor
 	scaleDownContext      *nodes.ScaleDownContext
 	maxNodeSkipEvalTime   *nodeevaltracker.MaxNodeSkipEvalTime
 }
 
 // New creates a new Planner object.
-func New(autoscalingCtx *ca_context.AutoscalingContext, processors *processors.AutoscalingProcessors, deleteOptions options.NodeDeleteOptions, drainabilityRules rules.Rules) *Planner {
-	resourceLimitsFinder := resource.NewLimitsFinder(processors.CustomResourcesProcessor)
+func New(autoscalingCtx *ca_context.AutoscalingContext, processors *processors.AutoscalingProcessors, deleteOptions options.NodeDeleteOptions, drainabilityRules rules.Rules, quotasTrackerFactory *resourcequotas.TrackerFactory) *Planner {
 	minUpdateInterval := autoscalingCtx.AutoscalingOptions.NodeGroupDefaults.ScaleDownUnneededTime
 	if minUpdateInterval == 0*time.Nanosecond {
 		minUpdateInterval = 1 * time.Nanosecond
 	}
 
-	unneededNodes := unneeded.NewNodes(processors.NodeGroupConfigProcessor, resourceLimitsFinder)
+	unneededNodes := unneeded.NewNodes(processors.NodeGroupConfigProcessor)
 	if autoscalingCtx.AutoscalingOptions.NodeDeletionCandidateTTL != 0 {
 		unneededNodes.LoadFromExistingTaints(autoscalingCtx, time.Now())
 	}
@@ -106,8 +105,8 @@ func New(autoscalingCtx *ca_context.AutoscalingContext, processors *processors.A
 		actuationInjector:     scheduling.NewHintingSimulator(),
 		eligibilityChecker:    eligibility.NewChecker(processors.NodeGroupConfigProcessor),
 		nodeUtilizationMap:    make(map[string]utilization.Info),
-		resourceLimitsFinder:  resourceLimitsFinder,
 		cc:                    newControllerReplicasCalculator(autoscalingCtx.ListerRegistry),
+		quotasTrackerFactory:  quotasTrackerFactory,
 		scaleDownSetProcessor: processors.ScaleDownSetProcessor,
 		scaleDownContext:      nodes.NewDefaultScaleDownContext(),
 		minUpdateInterval:     minUpdateInterval,
@@ -151,18 +150,20 @@ func (p *Planner) CleanUpUnneededNodes() {
 // to the Planner.
 func (p *Planner) NodesToDelete(_ time.Time) (empty, needDrain []*apiv1.Node) {
 	empty, needDrain = []*apiv1.Node{}, []*apiv1.Node{}
+
 	nodes, err := allNodes(p.autoscalingCtx.ClusterSnapshot)
 	if err != nil {
-		klog.Errorf("Nothing will scale down, failed to list nodes from ClusterSnapshot: %v", err)
+		klog.Errorf("Failed to list nodes for final limit check: %v", err)
 		return nil, nil
 	}
-	resourceLimiter, err := p.autoscalingCtx.CloudProvider.GetResourceLimiter()
+
+	tracker, err := p.quotasTrackerFactory.NewMinQuotasTracker(p.autoscalingCtx, nodes)
 	if err != nil {
-		klog.Errorf("Nothing will scale down, failed to create resource limiter: %v", err)
+		klog.Errorf("Failed to create tracker for final limit check: %v", err)
 		return nil, nil
 	}
-	p.scaleDownContext.ResourcesLeft = p.resourceLimitsFinder.LimitsLeft(p.autoscalingCtx, nodes, resourceLimiter, p.latestUpdate).DeepCopy()
-	p.scaleDownContext.ResourcesWithLimits = resourceLimiter.GetResources()
+	p.scaleDownContext.Tracker = tracker
+
 	emptyRemovableNodes, needDrainRemovableNodes, unremovableNodes := p.unneededNodes.RemovableAt(p.autoscalingCtx, *p.scaleDownContext, p.latestUpdate)
 	p.addUnremovableNodes(unremovableNodes)
 
@@ -307,6 +308,7 @@ func (p *Planner) categorizeNodes(podDestinations map[string]bool, scaleDownCand
 			klog.V(4).Infof("%d out of %d nodes skipped in scale down simulation: there are already %d unneeded nodes so no point in looking for more. Total atomic scale down nodes: %d", len(currentlyUnneededNodeNames)-i, len(currentlyUnneededNodeNames), len(removableList), atomicScaleDownNodesCount)
 			break
 		}
+
 		removable, unremovable := p.rs.SimulateNodeRemoval(node, podDestinations, p.latestUpdate, p.autoscalingCtx.RemainingPdbTracker)
 		if removable != nil {
 			_, inParallel, _ := p.autoscalingCtx.RemainingPdbTracker.CanRemovePods(removable.PodsToReschedule)

cluster-autoscaler/core/scaledown/planner/planner_test.go

@@ -18,6 +18,8 @@ package planner
 
 import (
 	"fmt"
+	"k8s.io/autoscaler/cluster-autoscaler/resourcequotas"
+
 	"testing"
 	"time"
 
@@ -505,7 +507,8 @@ func TestUpdateClusterState(t *testing.T) {
 			assert.NoError(t, err)
 			clustersnapshot.InitializeClusterSnapshotOrDie(t, autoscalingCtx.ClusterSnapshot, tc.nodes, tc.pods)
 			deleteOptions := options.NodeDeleteOptions{}
-			p := New(&autoscalingCtx, processors, deleteOptions, nil)
+			factory := resourcequotas.NewTrackerFactory(resourcequotas.TrackerOptions{CustomResourcesProcessor: processors.CustomResourcesProcessor, QuotaProvider: resourcequotas.NewCloudMinProvider(provider)})
+			p := New(&autoscalingCtx, processors, deleteOptions, nil, factory)
 			p.eligibilityChecker = &fakeEligibilityChecker{eligible: asMap(tc.eligible)}
 			if tc.isSimulationTimeout {
 				autoscalingCtx.AutoscalingOptions.ScaleDownSimulationTimeout = 1 * time.Second
@@ -703,7 +706,8 @@ func TestUpdateClusterStatUnneededNodesLimit(t *testing.T) {
 			assert.NoError(t, err)
 			clustersnapshot.InitializeClusterSnapshotOrDie(t, autoscalingCtx.ClusterSnapshot, nodes, nil)
 			deleteOptions := options.NodeDeleteOptions{}
-			p := New(&autoscalingCtx, processors, deleteOptions, nil)
+			factory := resourcequotas.NewTrackerFactory(resourcequotas.TrackerOptions{CustomResourcesProcessor: processors.CustomResourcesProcessor, QuotaProvider: resourcequotas.NewCloudMinProvider(provider)})
+			p := New(&autoscalingCtx, processors, deleteOptions, nil, factory)
 			p.eligibilityChecker = &fakeEligibilityChecker{eligible: asMap(nodeNames(nodes))}
 			p.minUpdateInterval = tc.updateInterval
 			p.unneededNodes.Update(&autoscalingCtx, previouslyUnneeded, time.Now())
@@ -838,7 +842,8 @@ func TestNewPlannerWithExistingDeletionCandidateNodes(t *testing.T) {
 			assert.NoError(t, err)
 
 			deleteOptions := options.NodeDeleteOptions{}
-			p := New(&autoscalingCtx, processors, deleteOptions, nil)
+			factory := resourcequotas.NewTrackerFactory(resourcequotas.TrackerOptions{CustomResourcesProcessor: processors.CustomResourcesProcessor, QuotaProvider: resourcequotas.NewCloudMinProvider(provider)})
+			p := New(&autoscalingCtx, processors, deleteOptions, nil, factory)
 
 			p.unneededNodes.AsList()
 		})
@@ -847,11 +852,13 @@ func TestNewPlannerWithExistingDeletionCandidateNodes(t *testing.T) {
 
 func TestNodesToDelete(t *testing.T) {
 	testCases := []struct {
-		name           string
-		nodes          map[string][]*apiv1.Node
-		removableNodes map[cloudprovider.NodeGroup][]simulator.NodeToBeRemoved
-		wantEmpty      []*apiv1.Node
-		wantDrain      []*apiv1.Node
+		name            string
+		nodes           map[string][]*apiv1.Node
+		removableNodes  map[cloudprovider.NodeGroup][]simulator.NodeToBeRemoved
+		wantEmpty       []*apiv1.Node
+		wantDrain       []*apiv1.Node
+		minLimits       map[string]int64
+		wantUnremovable map[string]simulator.UnremovableReason
 	}{
 		{
 			name:           "empty",
@@ -1016,6 +1023,56 @@ func TestNodesToDelete(t *testing.T) {
 				buildRemovableNode("atomic-partial-ng-partially-registered-1", 0).Node,
 			},
 		},
+		{
+			name: "min limits race condition protection",
+			nodes: map[string][]*apiv1.Node{
+				"ng1": {
+					BuildTestNode("node-3", 1000, 10),
+				},
+			},
+			removableNodes: map[cloudprovider.NodeGroup][]simulator.NodeToBeRemoved{
+				sizedNodeGroup("ng1", 3, false): {
+					buildRemovableNode("node-1", 0),
+					buildRemovableNode("node-2", 1),
+				},
+			},
+			wantEmpty: []*apiv1.Node{
+				buildRemovableNode("node-1", 0).Node,
+			},
+			wantDrain: []*apiv1.Node{},
+			minLimits: map[string]int64{
+				cloudprovider.ResourceNameCores:  2,
+				cloudprovider.ResourceNameMemory: 20,
+				"nodes":                          2,
+			},
+			wantUnremovable: map[string]simulator.UnremovableReason{
+				"node-2": simulator.MinimalResourceLimitExceeded,
+			},
+		},
+		{
+			name: "two nodes removable, quota allows both",
+			nodes: map[string][]*apiv1.Node{
+				"ng1": {
+					BuildTestNode("node-3", 1000, 10),
+				},
+			},
+			removableNodes: map[cloudprovider.NodeGroup][]simulator.NodeToBeRemoved{
+				sizedNodeGroup("ng1", 3, false): {
+					buildRemovableNode("node-1", 0),
+					buildRemovableNode("node-2", 0),
+				},
+			},
+			wantEmpty: []*apiv1.Node{
+				buildRemovableNode("node-1", 0).Node,
+				buildRemovableNode("node-2", 0).Node,
+			},
+			wantDrain: []*apiv1.Node{},
+			minLimits: map[string]int64{
+				cloudprovider.ResourceNameCores:  1,
+				cloudprovider.ResourceNameMemory: 10,
+				"nodes":                          1,
+			},
+		},
 	}
 	for _, tc := range testCases {
 		tc := tc
@@ -1044,20 +1101,41 @@ func TestNodesToDelete(t *testing.T) {
 					ScaleDownUnneededTime: 10 * time.Minute,
 					ScaleDownUnreadyTime:  0 * time.Minute,
 				},
+				ScaleDownSimulationTimeout: 5 * time.Minute,
 			}
 			processors, templateNodeInfoRegistry := processorstest.NewTestProcessors(autoscalingOpts)
 			autoscalingCtx, err := NewScaleTestAutoscalingContext(autoscalingOpts, &fake.Clientset{}, nil, provider, nil, nil, templateNodeInfoRegistry)
 			assert.NoError(t, err)
 			clustersnapshot.InitializeClusterSnapshotOrDie(t, autoscalingCtx.ClusterSnapshot, allNodes, nil)
 			deleteOptions := options.NodeDeleteOptions{}
-			p := New(&autoscalingCtx, processors, deleteOptions, nil)
+			factory := resourcequotas.NewTrackerFactory(resourcequotas.TrackerOptions{CustomResourcesProcessor: processors.CustomResourcesProcessor, QuotaProvider: resourcequotas.NewCloudMinProvider(provider)})
+			p := New(&autoscalingCtx, processors, deleteOptions, nil, factory)
 			p.latestUpdate = time.Now()
+
+			if tc.minLimits != nil {
+				resourceLimiter := cloudprovider.NewResourceLimiter(tc.minLimits, nil)
+				provider.SetResourceLimiter(resourceLimiter)
+			}
+
 			p.scaleDownContext.ActuationStatus = deletiontracker.NewNodeDeletionTracker(0 * time.Second)
 			p.unneededNodes.Update(&autoscalingCtx, allRemovables, time.Now().Add(-1*time.Hour))
 			p.eligibilityChecker = &fakeEligibilityChecker{eligible: asMap(nodeNames(allNodes))}
 			empty, drain := p.NodesToDelete(time.Now())
 			assert.ElementsMatch(t, tc.wantEmpty, empty)
 			assert.ElementsMatch(t, tc.wantDrain, drain)
+
+			if tc.wantUnremovable != nil {
+				unremovable := p.unremovableNodes.AsList()
+				unremovableMap := make(map[string]simulator.UnremovableReason)
+				for _, u := range unremovable {
+					unremovableMap[u.Node.Name] = u.Reason
+				}
+				for nodeName, expectedReason := range tc.wantUnremovable {
+					reason, found := unremovableMap[nodeName]
+					assert.True(t, found, "Node %s not found in unremovable list", nodeName)
+					assert.Equal(t, expectedReason, reason, "Reason mismatch for node %s", nodeName)
+				}
+			}
 		})
 	}
 }

cluster-autoscaler/core/scaledown/resource/limits.go

@@ -146,7 +146,7 @@ func (lf *LimitsFinder) customResourcesTotal(autoscalingCtx *ca_context.Autoscal
 		if !cacheHit {
 			resourceTargets, err = lf.crp.GetNodeResourceTargets(autoscalingCtx, node, nodeGroup)
 			if err != nil {
-				return nil, errors.ToAutoscalerError(errors.CloudProviderError, err).AddPrefix("cannot get custom resource count for node %v when calculating cluster custom resource usage")
+				return nil, errors.ToAutoscalerError(errors.CloudProviderError, err).AddPrefix("cannot get custom resource count for node %v when calculating cluster custom resource usage", node.Name)
 			}
 			if nodeGroup != nil {
 				ngCache[nodeGroup.Id()] = resourceTargets
@@ -185,7 +185,7 @@ func (lf *LimitsFinder) DeltaForNode(autoscalingCtx *ca_context.AutoscalingConte
 	if cloudprovider.ContainsCustomResources(resourcesWithLimits) {
 		resourceTargets, err := lf.crp.GetNodeResourceTargets(autoscalingCtx, node, nodeGroup)
 		if err != nil {
-			return Delta{}, errors.ToAutoscalerError(errors.CloudProviderError, err).AddPrefix("failed to get node %v custom resources: %v", node.Name)
+			return Delta{}, errors.ToAutoscalerError(errors.CloudProviderError, err).AddPrefix("failed to get node %q custom resources: ", node.Name)
 		}
 		for _, resourceTarget := range resourceTargets {
 			resultScaleDownDelta[resourceTarget.ResourceType] = resourceTarget.ResourceCount

cluster-autoscaler/core/scaledown/unneeded/nodes.go

@@ -25,8 +25,7 @@ import (
 	ca_context "k8s.io/autoscaler/cluster-autoscaler/context"
 	"k8s.io/autoscaler/cluster-autoscaler/core/scaledown"
 	"k8s.io/autoscaler/cluster-autoscaler/core/scaledown/eligibility"
-	"k8s.io/autoscaler/cluster-autoscaler/core/scaledown/resource"
-	"k8s.io/autoscaler/cluster-autoscaler/metrics"
+
 	"k8s.io/autoscaler/cluster-autoscaler/processors/nodes"
 	"k8s.io/autoscaler/cluster-autoscaler/simulator"
 	"k8s.io/autoscaler/cluster-autoscaler/utils"
@@ -38,10 +37,10 @@ import (
 
 // Nodes tracks the state of cluster nodes that are not needed.
 type Nodes struct {
-	sdtg         scaleDownTimeGetter
-	limitsFinder *resource.LimitsFinder
-	cachedList   []*scaledown.UnneededNode
-	byName       map[string]*node
+	sdtg scaleDownTimeGetter
+
+	cachedList []*scaledown.UnneededNode
+	byName     map[string]*node
 }
 
 type node struct {
@@ -59,10 +58,9 @@ type scaleDownTimeGetter interface {
 }
 
 // NewNodes returns a new initialized Nodes object.
-func NewNodes(sdtg scaleDownTimeGetter, limitsFinder *resource.LimitsFinder) *Nodes {
+func NewNodes(sdtg scaleDownTimeGetter) *Nodes {
 	return &Nodes{
-		sdtg:         sdtg,
-		limitsFinder: limitsFinder,
+		sdtg: sdtg,
 	}
 }
 
@@ -290,25 +288,14 @@ func (n *Nodes) unremovableReason(autoscalingCtx *ca_context.AutoscalingContext,
 		return reason
 	}
 
-	resourceDelta, err := n.limitsFinder.DeltaForNode(autoscalingCtx, node, nodeGroup, scaleDownContext.ResourcesWithLimits)
+	checkResult, err := scaleDownContext.Tracker.ConsumeQuota(autoscalingCtx, nodeGroup, node, 1)
 	if err != nil {
-		klog.Errorf("Error getting node resources: %v", err)
+		klog.Errorf("Failed to check limits for %s: %v", node.Name, err)
 		return simulator.UnexpectedError
 	}
 
-	checkResult := scaleDownContext.ResourcesLeft.TryDecrementBy(resourceDelta)
 	if checkResult.Exceeded() {
-		klog.V(4).Infof("Skipping %s - minimal limit exceeded for %v", node.Name, checkResult.ExceededResources)
-		for _, resource := range checkResult.ExceededResources {
-			switch resource {
-			case cloudprovider.ResourceNameCores:
-				metrics.RegisterSkippedScaleDownCPU()
-			case cloudprovider.ResourceNameMemory:
-				metrics.RegisterSkippedScaleDownMemory()
-			default:
-				continue
-			}
-		}
+		klog.V(2).Infof("Skipping scale down of %s in this batch - would violate minimum limits", node.Name)
 		return simulator.MinimalResourceLimitExceeded
 	}