cherry-pick 7065, 7087 into datadog-master-13.0 (handle failed nodegroups better)

cluster-autoscaler/clusterstate/clusterstate.go

@@ -1171,17 +1171,17 @@ func (csr *ClusterStateRegistry) buildInstanceToErrorCodeMappings(instances []cl
 	return
 }
 
-// GetCreatedNodesWithErrors returns list of nodes being created which reported create error.
-func (csr *ClusterStateRegistry) GetCreatedNodesWithErrors() []*apiv1.Node {
+// GetCreatedNodesWithErrors returns a map from node group id to list of nodes which reported a create error.
+func (csr *ClusterStateRegistry) GetCreatedNodesWithErrors() map[string][]*apiv1.Node {
 	csr.Lock()
 	defer csr.Unlock()
 
-	nodesWithCreateErrors := make([]*apiv1.Node, 0, 0)
-	for _, nodeGroupInstances := range csr.cloudProviderNodeInstances {
+	nodesWithCreateErrors := make(map[string][]*apiv1.Node)
+	for nodeGroupId, nodeGroupInstances := range csr.cloudProviderNodeInstances {
 		_, _, instancesByErrorCode := csr.buildInstanceToErrorCodeMappings(nodeGroupInstances)
 		for _, instances := range instancesByErrorCode {
 			for _, instance := range instances {
-				nodesWithCreateErrors = append(nodesWithCreateErrors, FakeNode(instance, cloudprovider.FakeNodeCreateError))
+				nodesWithCreateErrors[nodeGroupId] = append(nodesWithCreateErrors[nodeGroupId], FakeNode(instance, cloudprovider.FakeNodeCreateError))
 			}
 		}
 	}

cluster-autoscaler/clusterstate/clusterstate_test.go

@@ -135,6 +135,77 @@ func TestEmptyOK(t *testing.T) {
 	assert.True(t, clusterstate.IsNodeGroupScalingUp("ng1"))
 }
 
+// TestRecalculateStateAfterNodeGroupSizeChanged checks that Recalculate updates state correctly after
+// some node group size changed. We verify that acceptable ranges are updated accordingly
+// and that the UpcomingNodes reflect the node group size change (important for recalculating state after
+// deleting scale-up nodes that failed to create).
+func TestRecalculateStateAfterNodeGroupSizeChanged(t *testing.T) {
+	ngName := "ng1"
+	testCases := []struct {
+		name                string
+		acceptableRange     AcceptableRange
+		readiness           Readiness
+		newTarget           int
+		scaleUpRequest      *ScaleUpRequest
+		wantAcceptableRange AcceptableRange
+		wantUpcoming        int
+	}{
+		{
+			name:                "failed scale up by 3 nodes",
+			acceptableRange:     AcceptableRange{MinNodes: 1, CurrentTarget: 4, MaxNodes: 4},
+			readiness:           Readiness{Ready: make([]string, 1)},
+			newTarget:           1,
+			wantAcceptableRange: AcceptableRange{MinNodes: 1, CurrentTarget: 1, MaxNodes: 1},
+			wantUpcoming:        0,
+		}, {
+			name:                "partially failed scale up",
+			acceptableRange:     AcceptableRange{MinNodes: 5, CurrentTarget: 7, MaxNodes: 8},
+			readiness:           Readiness{Ready: make([]string, 5)},
+			newTarget:           6,
+			wantAcceptableRange: AcceptableRange{MinNodes: 5, CurrentTarget: 6, MaxNodes: 6},
+			scaleUpRequest:      &ScaleUpRequest{Increase: 1},
+			wantUpcoming:        1,
+		}, {
+			name:                "scale up ongoing, no change",
+			acceptableRange:     AcceptableRange{MinNodes: 1, CurrentTarget: 4, MaxNodes: 4},
+			readiness:           Readiness{Ready: make([]string, 1)},
+			newTarget:           4,
+			wantAcceptableRange: AcceptableRange{MinNodes: 1, CurrentTarget: 4, MaxNodes: 4},
+			scaleUpRequest:      &ScaleUpRequest{Increase: 3},
+			wantUpcoming:        3,
+		}, {
+			name:                "no scale up, no change",
+			acceptableRange:     AcceptableRange{MinNodes: 4, CurrentTarget: 4, MaxNodes: 4},
+			readiness:           Readiness{Ready: make([]string, 4)},
+			newTarget:           4,
+			wantAcceptableRange: AcceptableRange{MinNodes: 4, CurrentTarget: 4, MaxNodes: 4},
+			wantUpcoming:        0,
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			provider := testprovider.NewTestCloudProvider(nil, nil)
+			provider.AddNodeGroup(ngName, 0, 1000, tc.newTarget)
+
+			fakeLogRecorder, _ := utils.NewStatusMapRecorder(&fake.Clientset{}, "kube-system", kube_record.NewFakeRecorder(5), false, "my-cool-configmap")
+			clusterState := NewClusterStateRegistry(provider, ClusterStateRegistryConfig{}, fakeLogRecorder,
+				newBackoff(), nodegroupconfig.NewDefaultNodeGroupConfigProcessor(config.NodeGroupAutoscalingOptions{}))
+			clusterState.acceptableRanges = map[string]AcceptableRange{ngName: tc.acceptableRange}
+			clusterState.perNodeGroupReadiness = map[string]Readiness{ngName: tc.readiness}
+			if tc.scaleUpRequest != nil {
+				clusterState.scaleUpRequests = map[string]*ScaleUpRequest{ngName: tc.scaleUpRequest}
+			}
+
+			clusterState.Recalculate()
+			assert.Equal(t, tc.wantAcceptableRange, clusterState.acceptableRanges[ngName])
+			upcomingCounts, _ := clusterState.GetUpcomingNodes()
+			if upcoming, found := upcomingCounts[ngName]; found {
+				assert.Equal(t, tc.wantUpcoming, upcoming, "Unexpected upcoming nodes count, want: %d got: %d", tc.wantUpcoming, upcomingCounts[ngName])
+			}
+		})
+	}
+}
+
 func TestOKOneUnreadyNode(t *testing.T) {
 	now := time.Now()
 

cluster-autoscaler/core/static_autoscaler.go

@@ -421,15 +421,7 @@ func (a *StaticAutoscaler) RunOnce(currentTime time.Time) caerrors.AutoscalerErr
 		return nil
 	}
 
-	danglingNodes, err := a.deleteCreatedNodesWithErrors()
-	if err != nil {
-		klog.Warningf("Failed to remove nodes that were created with errors, skipping iteration: %v", err)
-		return nil
-	}
-	if danglingNodes {
-		klog.V(0).Infof("Some nodes that failed to create were removed, skipping iteration")
-		return nil
-	}
+	a.deleteCreatedNodesWithErrors()
 
 	// Check if there has been a constant difference between the number of nodes in k8s and
 	// the number of nodes on the cloud provider side.
@@ -813,30 +805,11 @@ func toNodes(unregisteredNodes []clusterstate.UnregisteredNode) []*apiv1.Node {
 	return nodes
 }
 
-func (a *StaticAutoscaler) deleteCreatedNodesWithErrors() (bool, error) {
+func (a *StaticAutoscaler) deleteCreatedNodesWithErrors() {
 	// We always schedule deleting of incoming errornous nodes
 	// TODO[lukaszos] Consider adding logic to not retry delete every loop iteration
-	nodes := a.clusterStateRegistry.GetCreatedNodesWithErrors()
-
 	nodeGroups := a.nodeGroupsById()
-	nodesToBeDeletedByNodeGroupId := make(map[string][]*apiv1.Node)
-
-	for _, node := range nodes {
-		nodeGroup, err := a.CloudProvider.NodeGroupForNode(node)
-		if err != nil {
-			id := "<nil>"
-			if node != nil {
-				id = node.Spec.ProviderID
-			}
-			klog.Warningf("Cannot determine nodeGroup for node %v; %v", id, err)
-			continue
-		}
-		if nodeGroup == nil || reflect.ValueOf(nodeGroup).IsNil() {
-			a.clusterStateRegistry.RefreshCloudProviderNodeInstancesCache()
-			return false, fmt.Errorf("node %s has no known nodegroup", node.GetName())
-		}
-		nodesToBeDeletedByNodeGroupId[nodeGroup.Id()] = append(nodesToBeDeletedByNodeGroupId[nodeGroup.Id()], node)
-	}
+	nodesToBeDeletedByNodeGroupId := a.clusterStateRegistry.GetCreatedNodesWithErrors()
 
 	deletedAny := false
 
@@ -868,13 +841,16 @@ func (a *StaticAutoscaler) deleteCreatedNodesWithErrors() (bool, error) {
 
 		if err != nil {
 			klog.Warningf("Error while trying to delete nodes from %v: %v", nodeGroupId, err)
+		} else {
+			deletedAny = true
+			a.clusterStateRegistry.InvalidateNodeInstancesCacheEntry(nodeGroup)
 		}
-
-		deletedAny = deletedAny || err == nil
-		a.clusterStateRegistry.InvalidateNodeInstancesCacheEntry(nodeGroup)
 	}
 
-	return deletedAny, nil
+	if deletedAny {
+		klog.V(0).Infof("Some nodes that failed to create were removed, recalculating cluster state.")
+		a.clusterStateRegistry.Recalculate()
+	}
 }
 
 // instancesToNodes returns a list of fake nodes with just names populated,

cluster-autoscaler/core/static_autoscaler_test.go

@@ -1164,9 +1164,11 @@ func TestStaticAutoscalerInstanceCreationErrors(t *testing.T) {
 	clusterState.UpdateNodes([]*apiv1.Node{}, nil, now)
 
 	// delete nodes with create errors
-	removedNodes, err := autoscaler.deleteCreatedNodesWithErrors()
-	assert.True(t, removedNodes)
-	assert.NoError(t, err)
+	autoscaler.deleteCreatedNodesWithErrors()
+
+	// nodes should be deleted
+	expectedDeleteCalls := 1
+	nodeGroupA.AssertNumberOfCalls(t, "DeleteNodes", expectedDeleteCalls)
 
 	// check delete was called on correct nodes
 	nodeGroupA.AssertCalled(t, "DeleteNodes", mock.MatchedBy(
@@ -1190,11 +1192,12 @@ func TestStaticAutoscalerInstanceCreationErrors(t *testing.T) {
 	clusterState.UpdateNodes([]*apiv1.Node{}, nil, now)
 
 	// delete nodes with create errors
-	removedNodes, err = autoscaler.deleteCreatedNodesWithErrors()
-	assert.True(t, removedNodes)
-	assert.NoError(t, err)
+	autoscaler.deleteCreatedNodesWithErrors()
 
 	// nodes should be deleted again
+	expectedDeleteCalls += 1
+	nodeGroupA.AssertNumberOfCalls(t, "DeleteNodes", expectedDeleteCalls)
+
 	nodeGroupA.AssertCalled(t, "DeleteNodes", mock.MatchedBy(
 		func(nodes []*apiv1.Node) bool {
 			if len(nodes) != 4 {
@@ -1255,12 +1258,10 @@ func TestStaticAutoscalerInstanceCreationErrors(t *testing.T) {
 	clusterState.UpdateNodes([]*apiv1.Node{}, nil, now)
 
 	// delete nodes with create errors
-	removedNodes, err = autoscaler.deleteCreatedNodesWithErrors()
-	assert.False(t, removedNodes)
-	assert.NoError(t, err)
+	autoscaler.deleteCreatedNodesWithErrors()
 
-	// we expect no more Delete Nodes
-	nodeGroupA.AssertNumberOfCalls(t, "DeleteNodes", 2)
+	// we expect no more Delete Nodes, don't increase expectedDeleteCalls
+	nodeGroupA.AssertNumberOfCalls(t, "DeleteNodes", expectedDeleteCalls)
 
 	// failed node not included by NodeGroupForNode
 	nodeGroupC := &mockprovider.NodeGroup{}
@@ -1297,10 +1298,8 @@ func TestStaticAutoscalerInstanceCreationErrors(t *testing.T) {
 	// update cluster state
 	clusterState.UpdateNodes([]*apiv1.Node{}, nil, time.Now())
 
-	// return early on failed nodes without matching nodegroups
-	removedNodes, err = autoscaler.deleteCreatedNodesWithErrors()
-	assert.False(t, removedNodes)
-	assert.Error(t, err)
+	// No nodes are deleted when failed nodes don't have matching node groups
+	autoscaler.deleteCreatedNodesWithErrors()
 	nodeGroupC.AssertNumberOfCalls(t, "DeleteNodes", 0)
 
 	nodeGroupAtomic := &mockprovider.NodeGroup{}
@@ -1355,9 +1354,7 @@ func TestStaticAutoscalerInstanceCreationErrors(t *testing.T) {
 	clusterState.UpdateNodes([]*apiv1.Node{}, nil, now)
 
 	// delete nodes with create errors
-	removedNodes, err = autoscaler.deleteCreatedNodesWithErrors()
-	assert.True(t, removedNodes)
-	assert.NoError(t, err)
+	autoscaler.deleteCreatedNodesWithErrors()
 
 	nodeGroupAtomic.AssertCalled(t, "DeleteNodes", mock.MatchedBy(
 		func(nodes []*apiv1.Node) bool {
@@ -1370,6 +1367,56 @@ func TestStaticAutoscalerInstanceCreationErrors(t *testing.T) {
 			}
 			return names["D1"] && names["D2"] && names["D3"]
 		}))
+
+	// Node group with getOptions error gets no deletes.
+	nodeGroupError := &mockprovider.NodeGroup{}
+	nodeGroupError.On("Exist").Return(true)
+	nodeGroupError.On("Autoprovisioned").Return(false)
+	nodeGroupError.On("TargetSize").Return(1, nil)
+	nodeGroupError.On("Id").Return("E")
+	nodeGroupError.On("DeleteNodes", mock.Anything).Return(nil)
+	nodeGroupError.On("GetOptions", options.NodeGroupDefaults).Return(nil, fmt.Errorf("Failed to get options"))
+	nodeGroupError.On("Nodes").Return([]cloudprovider.Instance{
+		{
+			Id: "E1",
+			Status: &cloudprovider.InstanceStatus{
+				State: cloudprovider.InstanceRunning,
+			},
+		},
+		{
+
+			Id: "E2",
+			Status: &cloudprovider.InstanceStatus{
+				State: cloudprovider.InstanceCreating,
+				ErrorInfo: &cloudprovider.InstanceErrorInfo{
+					ErrorClass: cloudprovider.OutOfResourcesErrorClass,
+					ErrorCode:  "QUOTA",
+				},
+			},
+		},
+	}, nil)
+
+	provider = &mockprovider.CloudProvider{}
+	provider.On("NodeGroups").Return([]cloudprovider.NodeGroup{nodeGroupError})
+	provider.On("NodeGroupForNode", mock.Anything).Return(
+		func(node *apiv1.Node) cloudprovider.NodeGroup {
+			if strings.HasPrefix(node.Spec.ProviderID, "E") {
+				return nodeGroupError
+			}
+			return nil
+		}, nil).Times(2)
+
+	clusterState = clusterstate.NewClusterStateRegistry(provider, clusterStateConfig, context.LogRecorder, NewBackoff(), nodeGroupConfigProcessor)
+	clusterState.RefreshCloudProviderNodeInstancesCache()
+	autoscaler.CloudProvider = provider
+	autoscaler.clusterStateRegistry = clusterState
+	// propagate nodes info in cluster state
+	clusterState.UpdateNodes([]*apiv1.Node{}, nil, now)
+
+	// delete nodes with create errors
+	autoscaler.deleteCreatedNodesWithErrors()
+
+	nodeGroupError.AssertNumberOfCalls(t, "DeleteNodes", 0)
 }
 
 type candidateTrackingFakePlanner struct {