Add upcoming node groups state checker

Author: abdelrahman882

cluster-autoscaler/cloudprovider/test/test_cloud_provider.go

@@ -216,83 +216,66 @@ func (tcp *TestCloudProvider) NewNodeGroup(machineType string, labels map[string
 
 // NewNodeGroupWithId creates a new node group with custom ID suffix.
 func (tcp *TestCloudProvider) NewNodeGroupWithId(machineType string, labels map[string]string, systemLabels map[string]string,
-	taints []apiv1.Taint, extraResources map[string]resource.Quantity, upcoming bool, id string) (cloudprovider.NodeGroup, error) {
+	taints []apiv1.Taint, extraResources map[string]resource.Quantity, id string) (cloudprovider.NodeGroup, error) {
 	return &TestNodeGroup{
 		cloudProvider:   tcp,
 		id:              "autoprovisioned-" + machineType + "-" + id,
 		minSize:         0,
 		maxSize:         1000,
 		targetSize:      0,
 		exist:           false,
-		upcoming:        upcoming,
 		autoprovisioned: true,
 		machineType:     machineType,
 		labels:          labels,
 		taints:          taints,
 	}, nil
 }
 
-// InsertNodeGroup adds already created node group to test cloud provider.
-func (tcp *TestCloudProvider) InsertNodeGroup(nodeGroup cloudprovider.NodeGroup) {
-	tcp.Lock()
-	defer tcp.Unlock()
-
-	tcp.groups[nodeGroup.Id()] = nodeGroup
-}
-
 // BuildNodeGroup returns a test node group.
-func (tcp *TestCloudProvider) BuildNodeGroup(id string, min, max, size int, autoprovisioned bool, machineType string, opts *config.NodeGroupAutoscalingOptions) *TestNodeGroup {
+func (tcp *TestCloudProvider) BuildNodeGroup(id string, min, max, size int, autoprovisioned bool, machineType string, opts *config.NodeGroupAutoscalingOptions, exists bool) *TestNodeGroup {
 	return &TestNodeGroup{
 		cloudProvider:   tcp,
 		id:              id,
 		minSize:         min,
 		maxSize:         max,
 		targetSize:      size,
-		exist:           true,
+		exist:           exists,
 		autoprovisioned: autoprovisioned,
 		machineType:     machineType,
 		opts:            opts,
 	}
 }
 
-// BuildUpcomingNodeGroup returns an upcoming test node group.
-func (tcp *TestCloudProvider) BuildUpcomingNodeGroup(id string, min, max, size int, autoprovisioned bool, machineType string, opts *config.NodeGroupAutoscalingOptions) *TestNodeGroup {
-	return &TestNodeGroup{
-		cloudProvider:   tcp,
-		id:              id,
-		minSize:         min,
-		maxSize:         max,
-		targetSize:      size,
-		exist:           false,
-		upcoming:        true,
-		autoprovisioned: autoprovisioned,
-		machineType:     machineType,
-		opts:            opts,
-	}
+// InsertNodeGroup adds already created node group to test cloud provider.
+func (tcp *TestCloudProvider) InsertNodeGroup(nodeGroup cloudprovider.NodeGroup) {
+	tcp.Lock()
+	defer tcp.Unlock()
+
+	tcp.groups[nodeGroup.Id()] = nodeGroup
 }
 
 // AddNodeGroup adds node group to test cloud provider.
 func (tcp *TestCloudProvider) AddNodeGroup(id string, min int, max int, size int) {
-	nodeGroup := tcp.BuildNodeGroup(id, min, max, size, false, "", nil)
+	nodeGroup := tcp.BuildNodeGroup(id, min, max, size, false, "", nil, true)
 	tcp.InsertNodeGroup(nodeGroup)
 }
 
 // AddUpcomingNodeGroup adds upcoming node group to test cloud provider.
 func (tcp *TestCloudProvider) AddUpcomingNodeGroup(id string, min int, max int, size int) {
-	nodeGroup := tcp.BuildUpcomingNodeGroup(id, min, max, size, false, "", nil)
+	nodeGroup := tcp.BuildNodeGroup(id, min, max, size, false, "", nil, false)
 	tcp.InsertNodeGroup(nodeGroup)
 }
 
 // AddNodeGroupWithCustomOptions adds node group with custom options
 // to test cloud provider.
 func (tcp *TestCloudProvider) AddNodeGroupWithCustomOptions(id string, min int, max int, size int, opts *config.NodeGroupAutoscalingOptions) {
-	nodeGroup := tcp.BuildNodeGroup(id, min, max, size, false, "", opts)
+	nodeGroup := tcp.BuildNodeGroup(id, min, max, size, false, "", opts, true)
 	tcp.InsertNodeGroup(nodeGroup)
 }
 
 // AddAutoprovisionedNodeGroup adds node group to test cloud provider.
 func (tcp *TestCloudProvider) AddAutoprovisionedNodeGroup(id string, min int, max int, size int, machineType string) *TestNodeGroup {
-	nodeGroup := tcp.BuildNodeGroup(id, min, max, size, true, machineType, nil)
+	nodeGroup := tcp.BuildNodeGroup(id, min, max, size, true, machineType, nil, true)
 	tcp.InsertNodeGroup(nodeGroup)
 	return nodeGroup
 }

cluster-autoscaler/clusterstate/clusterstate.go

@@ -29,6 +29,7 @@ import (
 	"k8s.io/autoscaler/cluster-autoscaler/clusterstate/utils"
 	"k8s.io/autoscaler/cluster-autoscaler/metrics"
 	"k8s.io/autoscaler/cluster-autoscaler/processors/nodegroupconfig"
+	"k8s.io/autoscaler/cluster-autoscaler/processors/nodegroups/asyncnodegroups"
 	"k8s.io/autoscaler/cluster-autoscaler/utils/backoff"
 	"k8s.io/autoscaler/cluster-autoscaler/utils/gpu"
 	kube_util "k8s.io/autoscaler/cluster-autoscaler/utils/kubernetes"
@@ -141,6 +142,7 @@ type ClusterStateRegistry struct {
 	cloudProviderNodeInstancesCache    *utils.CloudProviderNodeInstancesCache
 	interrupt                          chan struct{}
 	nodeGroupConfigProcessor           nodegroupconfig.NodeGroupConfigProcessor
+	asyncNodeGroupStateChecker         asyncnodegroups.AsyncNodeGroupStateChecker
 
 	// scaleUpFailures contains information about scale-up failures for each node group. It should be
 	// cleared periodically to avoid unnecessary accumulation.
@@ -155,7 +157,7 @@ type NodeGroupScalingSafety struct {
 }
 
 // NewClusterStateRegistry creates new ClusterStateRegistry.
-func NewClusterStateRegistry(cloudProvider cloudprovider.CloudProvider, config ClusterStateRegistryConfig, logRecorder *utils.LogEventRecorder, backoff backoff.Backoff, nodeGroupConfigProcessor nodegroupconfig.NodeGroupConfigProcessor) *ClusterStateRegistry {
+func NewClusterStateRegistry(cloudProvider cloudprovider.CloudProvider, config ClusterStateRegistryConfig, logRecorder *utils.LogEventRecorder, backoff backoff.Backoff, nodeGroupConfigProcessor nodegroupconfig.NodeGroupConfigProcessor, asyncNodeGroupStateChecker asyncnodegroups.AsyncNodeGroupStateChecker) *ClusterStateRegistry {
 	return &ClusterStateRegistry{
 		scaleUpRequests:                 make(map[string]*ScaleUpRequest),
 		scaleDownRequests:               make([]*ScaleDownRequest, 0),
@@ -175,6 +177,7 @@ func NewClusterStateRegistry(cloudProvider cloudprovider.CloudProvider, config C
 		interrupt:                       make(chan struct{}),
 		scaleUpFailures:                 make(map[string][]ScaleUpFailure),
 		nodeGroupConfigProcessor:        nodeGroupConfigProcessor,
+		asyncNodeGroupStateChecker:      asyncNodeGroupStateChecker,
 	}
 }
 
@@ -684,7 +687,7 @@ func (csr *ClusterStateRegistry) updateIncorrectNodeGroupSizes(currentTime time.
 			klog.Warningf("Acceptable range for node group %s not found", nodeGroup.Id())
 			continue
 		}
-		if nodeGroup.IsUpcoming() {
+		if csr.asyncNodeGroupStateChecker.IsUpcoming(nodeGroup) {
 			// Nodes for upcoming node groups reside in-memory and wait for node group to be fully
 			// created. There is no need to mark their sizes incorrect.
 			continue
@@ -986,7 +989,7 @@ func (csr *ClusterStateRegistry) GetUpcomingNodes() (upcomingCounts map[string]i
 	registeredNodeNames = map[string][]string{}
 	for _, nodeGroup := range csr.cloudProvider.NodeGroups() {
 		id := nodeGroup.Id()
-		if nodeGroup.IsUpcoming() {
+		if csr.asyncNodeGroupStateChecker.IsUpcoming(nodeGroup) {
 			size, err := nodeGroup.TargetSize()
 			if size >= 0 || err != nil {
 				upcomingCounts[id] = size