Add async node group creation

Author: pmendelski

cluster-autoscaler/cloudprovider/test/test_cloud_provider.go

@@ -216,14 +216,15 @@ 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, id string) (cloudprovider.NodeGroup, error) {
+	taints []apiv1.Taint, extraResources map[string]resource.Quantity, upcoming bool, 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,
@@ -254,12 +255,34 @@ func (tcp *TestCloudProvider) BuildNodeGroup(id string, min, max, size int, auto
 	}
 }
 
+// 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,
+	}
+}
+
 // 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)
 	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)
+	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) {

cluster-autoscaler/clusterstate/clusterstate.go

@@ -684,6 +684,11 @@ func (csr *ClusterStateRegistry) updateIncorrectNodeGroupSizes(currentTime time.
 			klog.Warningf("Acceptable range for node group %s not found", nodeGroup.Id())
 			continue
 		}
+		if nodeGroup.IsUpcoming() {
+			// 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
+		}
 		readiness, found := csr.perNodeGroupReadiness[nodeGroup.Id()]
 		if !found {
 			// if MinNodes == 0 node group has been scaled to 0 and everything's fine
@@ -981,6 +986,13 @@ func (csr *ClusterStateRegistry) GetUpcomingNodes() (upcomingCounts map[string]i
 	registeredNodeNames = map[string][]string{}
 	for _, nodeGroup := range csr.cloudProvider.NodeGroups() {
 		id := nodeGroup.Id()
+		if nodeGroup.IsUpcoming() {
+			size, err := nodeGroup.TargetSize()
+			if size >= 0 || err != nil {
+				upcomingCounts[id] = size
+			}
+			continue
+		}
 		readiness := csr.perNodeGroupReadiness[id]
 		ar := csr.acceptableRanges[id]
 		// newNodes is the number of nodes that

cluster-autoscaler/clusterstate/clusterstate_test.go

@@ -1380,3 +1380,27 @@ func TestTruncateIfExceedMaxSize(t *testing.T) {
 		})
 	}
 }
+
+func TestUpcomingNodesFromUpcomingNodeGroups(t *testing.T) {
+	now := time.Now()
+
+	ng := BuildTestNode("ng", 1000, 1000)
+
+	provider := testprovider.NewTestCloudProvider(nil, nil)
+	provider.AddUpcomingNodeGroup("ng", 1, 10, 2)
+
+	assert.NotNil(t, provider)
+	fakeClient := &fake.Clientset{}
+	fakeLogRecorder, _ := utils.NewStatusMapRecorder(fakeClient, "kube-system", kube_record.NewFakeRecorder(5), false, "some-map")
+	clusterstate := NewClusterStateRegistry(provider, ClusterStateRegistryConfig{
+		MaxTotalUnreadyPercentage: 10,
+		OkTotalUnreadyCount:       1,
+	}, fakeLogRecorder, newBackoff(), nodegroupconfig.NewDefaultNodeGroupConfigProcessor(config.NodeGroupAutoscalingOptions{MaxNodeProvisionTime: 15 * time.Minute}))
+	err := clusterstate.UpdateNodes([]*apiv1.Node{ng}, nil, now)
+	assert.NoError(t, err)
+	assert.Equal(t, 0, len(clusterstate.GetClusterReadiness().Unready))
+	assert.Equal(t, 0, len(clusterstate.GetClusterReadiness().NotStarted))
+	upcoming, upcomingRegistered := clusterstate.GetUpcomingNodes()
+	assert.Equal(t, 2, upcoming["ng"])
+	assert.Empty(t, upcomingRegistered["ng"])
+}

cluster-autoscaler/config/autoscaling_options.go

@@ -295,6 +295,8 @@ type AutoscalingOptions struct {
 	BypassedSchedulers map[string]bool
 	// ProvisioningRequestEnabled tells if CA processes ProvisioningRequest.
 	ProvisioningRequestEnabled bool
+	// AsyncNodeGroupsEnabled tells if CA creates/deletes node groups asynchronously.
+	AsyncNodeGroupsEnabled bool
 }
 
 // KubeClientOptions specify options for kube client

cluster-autoscaler/core/scaleup/orchestrator/async_initializer.go

@@ -0,0 +1,117 @@
+/*
+Copyright 2016 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package orchestrator
+
+import (
+	"time"
+
+	appsv1 "k8s.io/api/apps/v1"
+	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
+	"k8s.io/autoscaler/cluster-autoscaler/context"
+	"k8s.io/autoscaler/cluster-autoscaler/core/utils"
+	"k8s.io/autoscaler/cluster-autoscaler/processors/nodegroups"
+	"k8s.io/autoscaler/cluster-autoscaler/processors/nodegroupset"
+	"k8s.io/autoscaler/cluster-autoscaler/processors/status"
+	"k8s.io/autoscaler/cluster-autoscaler/utils/errors"
+	"k8s.io/autoscaler/cluster-autoscaler/utils/taints"
+	"k8s.io/klog/v2"
+	"k8s.io/kubernetes/pkg/scheduler/framework"
+)
+
+type asyncNodeGroupInitializer struct {
+	nodeGroup              cloudprovider.NodeGroup
+	nodeInfo               *framework.NodeInfo
+	scaleUpExecutor        *scaleUpExecutor
+	taintConfig            taints.TaintConfig
+	daemonSets             []*appsv1.DaemonSet
+	scaleUpStatusProcessor status.ScaleUpStatusProcessor
+	context                *context.AutoscalingContext
+	atomicScaleUp          bool
+}
+
+func newAsyncNodeGroupInitializer(
+	nodeGroup cloudprovider.NodeGroup,
+	nodeInfo *framework.NodeInfo,
+	scaleUpExecutor *scaleUpExecutor,
+	taintConfig taints.TaintConfig,
+	daemonSets []*appsv1.DaemonSet,
+	scaleUpStatusProcessor status.ScaleUpStatusProcessor,
+	context *context.AutoscalingContext,
+	atomicScaleUp bool,
+) *asyncNodeGroupInitializer {
+	return &asyncNodeGroupInitializer{
+		nodeGroup,
+		nodeInfo,
+		scaleUpExecutor,
+		taintConfig,
+		daemonSets,
+		scaleUpStatusProcessor,
+		context,
+		atomicScaleUp,
+	}
+}
+
+func (s *asyncNodeGroupInitializer) InitializeNodeGroup(result nodegroups.AsyncNodeGroupCreationResult) {
+	if result.Error != nil {
+		klog.Errorf("Async node group creation failed. Async scale-up is cancelled. %v", result.Error)
+		scaleUpStatus, _ := status.UpdateScaleUpError(&status.ScaleUpStatus{}, errors.ToAutoscalerError(errors.InternalError, result.Error))
+		s.scaleUpStatusProcessor.Process(s.context, scaleUpStatus)
+		return
+	}
+	mainCreatedNodeGroup := result.CreationResult.MainCreatedNodeGroup
+	// If possible replace candidate node-info with node info based on crated node group. The latter
+	// one should be more in line with nodes which will be created by node group.
+	nodeInfo, aErr := utils.GetNodeInfoFromTemplate(mainCreatedNodeGroup, s.daemonSets, s.taintConfig)
+	if aErr != nil {
+		klog.Warningf("Cannot build node info for newly created main node group %s. Using fallback. Error: %v", mainCreatedNodeGroup.Id(), aErr)
+		nodeInfo = s.nodeInfo
+	}
+
+	nodeInfos := make(map[string]*framework.NodeInfo)
+	var scaleUpInfos []nodegroupset.ScaleUpInfo
+	for _, nodeGroup := range result.CreationResult.AllCreatedNodeGroups() {
+		if targetSize := result.TargetSizes[nodeGroup.Id()]; targetSize > 0 {
+			nodeInfos[nodeGroup.Id()] = nodeInfo
+			scaleUpInfo := nodegroupset.ScaleUpInfo{
+				Group:       nodeGroup,
+				CurrentSize: 0,
+				NewSize:     targetSize,
+				MaxSize:     nodeGroup.MaxSize(),
+			}
+			scaleUpInfos = append(scaleUpInfos, scaleUpInfo)
+		}
+	}
+	klog.Infof("Starting initial scale-up for async created node groups. Scale ups: %v", scaleUpInfos)
+	err, failedNodeGroups := s.scaleUpExecutor.ExecuteScaleUps(scaleUpInfos, nodeInfos, time.Now(), s.atomicScaleUp)
+	if err != nil {
+		var failedNodeGroupIds []string
+		for _, failedNodeGroup := range failedNodeGroups {
+			failedNodeGroupIds = append(failedNodeGroupIds, failedNodeGroup.Id())
+		}
+		klog.Errorf("Async scale-up for asynchronously created node group failed: %v (node groups: %v)", err, failedNodeGroupIds)
+		return
+	}
+	klog.Infof("Initial scale-up succeeded. Scale ups: %v", scaleUpInfos)
+}
+
+func nodeGroupIds(nodeGroups []cloudprovider.NodeGroup) []string {
+	var result []string
+	for _, ng := range nodeGroups {
+		result = append(result, ng.Id())
+	}
+	return result
+}

cluster-autoscaler/core/scaleup/orchestrator/executor.go

@@ -172,6 +172,11 @@ func (e *scaleUpExecutor) executeScaleUp(
 	if increase < 0 {
 		return errors.NewAutoscalerError(errors.InternalError, fmt.Sprintf("increase in number of nodes cannot be negative, got: %v", increase))
 	}
+	if info.Group.IsUpcoming() {
+		// Don't emit scale up event for upcoming node group as it will be generated after
+		// the node group is created, during initial scale up.
+		return nil
+	}
 	e.scaleStateNotifier.RegisterScaleUp(info.Group, increase, time.Now())
 	metrics.RegisterScaleUp(increase, gpuResourceName, gpuType)
 	e.autoscalingContext.LogRecorder.Eventf(apiv1.EventTypeNormal, "ScaledUpGroup",

cluster-autoscaler/core/scaleup/orchestrator/orchestrator.go

@@ -214,7 +214,7 @@ func (o *ScaleUpOrchestrator) ScaleUp(
 
 	// If necessary, create the node group. This is no longer simulation, an empty node group will be created by cloud provider if supported.
 	createNodeGroupResults := make([]nodegroups.CreateNodeGroupResult, 0)
-	if !bestOption.NodeGroup.Exist() {
+	if !bestOption.NodeGroup.Exist() && !bestOption.NodeGroup.IsUpcoming() {
 		if allOrNothing && bestOption.NodeGroup.MaxSize() < newNodes {
 			klog.V(1).Infof("Can only create a new node group with max %d nodes, need %d nodes", bestOption.NodeGroup.MaxSize(), newNodes)
 			// Can't execute a scale-up that will accommodate all pods, so nothing is considered schedulable.
@@ -223,7 +223,7 @@ func (o *ScaleUpOrchestrator) ScaleUp(
 			return buildNoOptionsAvailableStatus(markedEquivalenceGroups, skippedNodeGroups, nodeGroups), nil
 		}
 		var scaleUpStatus *status.ScaleUpStatus
-		createNodeGroupResults, scaleUpStatus, aErr = o.CreateNodeGroup(bestOption, nodeInfos, schedulablePodGroups, podEquivalenceGroups, daemonSets)
+		createNodeGroupResults, scaleUpStatus, aErr = o.CreateNodeGroup(bestOption, nodeInfos, schedulablePodGroups, podEquivalenceGroups, daemonSets, allOrNothing)
 		if aErr != nil {
 			return scaleUpStatus, aErr
 		}
@@ -503,11 +503,19 @@ func (o *ScaleUpOrchestrator) CreateNodeGroup(
 	schedulablePodGroups map[string][]estimator.PodEquivalenceGroup,
 	podEquivalenceGroups []*equivalence.PodGroup,
 	daemonSets []*appsv1.DaemonSet,
+	allOrNothing bool,
 ) ([]nodegroups.CreateNodeGroupResult, *status.ScaleUpStatus, errors.AutoscalerError) {
 	createNodeGroupResults := make([]nodegroups.CreateNodeGroupResult, 0)
 
 	oldId := initialOption.NodeGroup.Id()
-	createNodeGroupResult, aErr := o.processors.NodeGroupManager.CreateNodeGroup(o.autoscalingContext, initialOption.NodeGroup)
+	var createNodeGroupResult nodegroups.CreateNodeGroupResult
+	var aErr errors.AutoscalerError
+	if o.autoscalingContext.AsyncNodeGroupsEnabled {
+		initializer := newAsyncNodeGroupInitializer(initialOption.NodeGroup, nodeInfos[oldId], o.scaleUpExecutor, o.taintConfig, daemonSets, o.processors.ScaleUpStatusProcessor, o.autoscalingContext, allOrNothing)
+		createNodeGroupResult, aErr = o.processors.NodeGroupManager.CreateNodeGroupAsync(o.autoscalingContext, initialOption.NodeGroup, initializer)
+	} else {
+		createNodeGroupResult, aErr = o.processors.NodeGroupManager.CreateNodeGroup(o.autoscalingContext, initialOption.NodeGroup)
+	}
 	if aErr != nil {
 		status, err := status.UpdateScaleUpError(
 			&status.ScaleUpStatus{FailedCreationNodeGroups: []cloudprovider.NodeGroup{initialOption.NodeGroup}, PodsTriggeredScaleUp: initialOption.Pods},