PreFilteringScaleDownNodeProcessor should exclude upcoming nodes from destination candidates

cluster-autoscaler/core/static_autoscaler.go

@@ -49,6 +49,7 @@ import (
 	"k8s.io/autoscaler/cluster-autoscaler/simulator/framework"
 	"k8s.io/autoscaler/cluster-autoscaler/simulator/options"
 	"k8s.io/autoscaler/cluster-autoscaler/simulator/predicatechecker"
+	ca_utils "k8s.io/autoscaler/cluster-autoscaler/utils"
 	"k8s.io/autoscaler/cluster-autoscaler/utils/backoff"
 	caerrors "k8s.io/autoscaler/cluster-autoscaler/utils/errors"
 	kube_util "k8s.io/autoscaler/cluster-autoscaler/utils/kubernetes"
@@ -69,9 +70,6 @@ const (
 	// a bit more latency to wait for more pods and make a more informed scale-up decision.
 	unschedulablePodWithGpuTimeBuffer = 30 * time.Second
 
-	// NodeUpcomingAnnotation is an annotation CA adds to nodes which are upcoming.
-	NodeUpcomingAnnotation = "cluster-autoscaler.k8s.io/upcoming-node"
-
 	// podScaleUpDelayAnnotationKey is an annotation how long pod can wait to be scaled up.
 	podScaleUpDelayAnnotationKey = "cluster-autoscaler.kubernetes.io/pod-scale-up-delay"
 )
@@ -1027,7 +1025,7 @@ func getUpcomingNodeInfos(upcomingCounts map[string]int, nodeInfos map[string]*f
 		if nodeTemplate.Node().Annotations == nil {
 			nodeTemplate.Node().Annotations = make(map[string]string)
 		}
-		nodeTemplate.Node().Annotations[NodeUpcomingAnnotation] = "true"
+		nodeTemplate.Node().Annotations[ca_utils.NodeUpcomingAnnotation] = "true"
 
 		var nodes []*framework.NodeInfo
 		for i := 0; i < numberOfNodes; i++ {

cluster-autoscaler/processors/nodes/pre_filtering_processor.go

@@ -37,7 +37,15 @@ type PreFilteringScaleDownNodeProcessor struct {
 // that would become unscheduled after a scale down.
 func (n *PreFilteringScaleDownNodeProcessor) GetPodDestinationCandidates(ctx *context.AutoscalingContext,
 	nodes []*apiv1.Node) ([]*apiv1.Node, errors.AutoscalerError) {
-	return nodes, nil
+	result := make([]*apiv1.Node, 0, len(nodes))
+	for _, node := range nodes {
+		if node.Annotations == nil {
+			result = append(result, node)
+		} else if val, ok := node.Annotations[utils.NodeUpcomingAnnotation]; !ok || val != "true" {
+			result = append(result, node)
+		}
+	}
+	return result, nil
 }
 
 // GetScaleDownCandidates returns nodes that potentially could be scaled down and

cluster-autoscaler/processors/nodes/pre_filtering_processor_test.go

@@ -24,20 +24,23 @@ import (
 
 	testprovider "k8s.io/autoscaler/cluster-autoscaler/cloudprovider/test"
 	"k8s.io/autoscaler/cluster-autoscaler/context"
+	"k8s.io/autoscaler/cluster-autoscaler/utils"
 	. "k8s.io/autoscaler/cluster-autoscaler/utils/test"
 )
 
 func TestPreFilteringScaleDownNodeProcessor_GetPodDestinationCandidates(t *testing.T) {
 	n1 := BuildTestNode("n1", 100, 1000)
 	n2 := BuildTestNode("n2", 100, 1000)
+	n3 := BuildTestNode("n3", 100, 1000)
+	n3.Annotations = map[string]string{utils.NodeUpcomingAnnotation: "true"}
 	ctx := &context.AutoscalingContext{}
 	defaultProcessor := NewPreFilteringScaleDownNodeProcessor()
 	expectedNodes := []*apiv1.Node{n1, n2}
-	nodes := []*apiv1.Node{n1, n2}
+	nodes := []*apiv1.Node{n1, n2, n3}
 	nodes, err := defaultProcessor.GetPodDestinationCandidates(ctx, nodes)
 
 	assert.NoError(t, err)
-	assert.Equal(t, nodes, expectedNodes)
+	assert.Equal(t, expectedNodes, nodes)
 }
 
 func TestPreFilteringScaleDownNodeProcessor_GetScaleDownCandidateNodes(t *testing.T) {

cluster-autoscaler/utils/utils.go

@@ -24,6 +24,11 @@ import (
 	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
 )
 
+const (
+	// NodeUpcomingAnnotation is an annotation CA adds to nodes which are upcoming.
+	NodeUpcomingAnnotation = "cluster-autoscaler.k8s.io/upcoming-node"
+)
+
 // GetNodeGroupSizeMap return a map of node group id and its target size
 func GetNodeGroupSizeMap(cloudProvider cloudprovider.CloudProvider) map[string]int {
 	nodeGroupSize := make(map[string]int)