Report terminating instances as gone in AWS HasInstance

cluster-autoscaler/cloudprovider/aws/auto_scaling_groups.go

@@ -253,6 +253,29 @@ func (m *asgCache) findInstanceLifecycle(ref AwsInstanceRef) (autoscalingtypes.L
 	return "", fmt.Errorf("could not find instance %v", ref)
 }
 
+// isTerminatingLifecycleState returns whether an instance in the given
+// lifecycle state is terminating or already terminated, and therefore no
+// longer counted towards the ASG's desired capacity.
+func isTerminatingLifecycleState(lifecycle autoscalingtypes.LifecycleState) bool {
+	return lifecycle == autoscalingtypes.LifecycleStateTerminated ||
+		lifecycle == autoscalingtypes.LifecycleStateTerminating ||
+		lifecycle == autoscalingtypes.LifecycleStateTerminatingWait ||
+		lifecycle == autoscalingtypes.LifecycleStateTerminatingProceed
+}
+
+// InstanceTerminating returns whether the instance is terminating or already
+// terminated, i.e. it is no longer counted towards the ASG's desired capacity.
+func (m *asgCache) InstanceTerminating(ref AwsInstanceRef) bool {
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+
+	lifecycle, found := m.instanceLifecycle[ref]
+	if !found {
+		return false
+	}
+	return isTerminatingLifecycleState(lifecycle)
+}
+
 func (m *asgCache) SetAsgSize(asg *asg, size int) error {
 	m.mutex.Lock()
 	defer m.mutex.Unlock()
@@ -358,10 +381,7 @@ func (m *asgCache) DeleteInstances(instances []*AwsInstanceRef) error {
 			return err
 		}
 
-		if lifecycle == autoscalingtypes.LifecycleStateTerminated ||
-			lifecycle == autoscalingtypes.LifecycleStateTerminating ||
-			lifecycle == autoscalingtypes.LifecycleStateTerminatingWait ||
-			lifecycle == autoscalingtypes.LifecycleStateTerminatingProceed {
+		if isTerminatingLifecycleState(lifecycle) {
 			klog.V(2).Infof("instance %s is already terminating in state %s, will skip instead", instance.Name, lifecycle)
 			continue
 		}
@@ -381,6 +401,9 @@ func (m *asgCache) DeleteInstances(instances []*AwsInstanceRef) error {
 		// Proactively decrement the size so autoscaler makes better decisions
 		commonAsg.curSize--
 
+		// Mark the instance as terminating in the cache, so that HasInstance
+		// reports it as gone immediately, before the cache is regenerated
+		m.instanceLifecycle[*instance] = autoscalingtypes.LifecycleStateTerminating
 	}
 	return nil
 }
@@ -465,6 +488,14 @@ func (m *asgCache) regenerate() error {
 			newAsgToInstancesCache[asg.AwsRef][i] = ref
 			newInstanceStatusMap[ref] = instance.HealthStatus
 			newInstanceLifecycleMap[ref] = instance.LifecycleState
+			// Termination is one-way, so a stale API response claiming that an
+			// already-terminating instance is healthy must not overwrite the
+			// cached lifecycle state - right after DeleteInstances(), an
+			// eventually consistent response can still report the instance as
+			// InService.
+			if !isTerminatingLifecycleState(instance.LifecycleState) && isTerminatingLifecycleState(m.instanceLifecycle[ref]) {
+				newInstanceLifecycleMap[ref] = m.instanceLifecycle[ref]
+			}
 		}
 	}
 

cluster-autoscaler/cloudprovider/aws/aws_cloud_provider.go

@@ -171,7 +171,16 @@ func (aws *awsCloudProvider) HasInstance(node *apiv1.Node) (bool, error) {
 		return false, err
 	}
 
-	// we don't care about the status
+	// Instances that are being terminated (e.g. after NodeGroup.DeleteNodes())
+	// stay in the ASG API response until termination completes, but they are no
+	// longer counted towards the target size - report them as gone right away
+	// so that ClusterStateRegistry doesn't miscount them (see issue #9877).
+	if aws.awsManager.asgCache.InstanceTerminating(*awsRef) {
+		return false, nil
+	}
+
+	// for instances that are not terminating we don't care about the status
+	// value, only that the instance is present in the cache
 	status, err := aws.awsManager.asgCache.InstanceStatus(*awsRef)
 	if status != nil {
 		return true, nil

cluster-autoscaler/cloudprovider/aws/aws_cloud_provider_test.go

@@ -696,6 +696,16 @@ func TestHasInstance(t *testing.T) {
 					ProviderID: "aws:///us-east-1a/test-instance-id",
 					Name:       "test-instance-id",
 				}: &nodeStatus,
+				{
+					ProviderID: "aws:///us-east-1a/terminating-instance-id",
+					Name:       "terminating-instance-id",
+				}: &nodeStatus,
+			},
+			instanceLifecycle: map[AwsInstanceRef]autoscalingtypes.LifecycleState{
+				{
+					ProviderID: "aws:///us-east-1a/terminating-instance-id",
+					Name:       "terminating-instance-id",
+				}: autoscalingtypes.LifecycleStateTerminating,
 			},
 		},
 		awsService: testAwsService,
@@ -756,6 +766,118 @@ func TestHasInstance(t *testing.T) {
 	present, err = provider.HasInstance(node4)
 	assert.NoError(t, err)
 	assert.False(t, present)
+
+	// Case 5: node with a terminating instance - not present in AWS,
+	// even though the instance still has a status in the cache
+	node5 := &apiv1.Node{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "node-3",
+		},
+		Spec: apiv1.NodeSpec{
+			ProviderID: "aws:///us-east-1a/terminating-instance-id",
+		},
+	}
+	present, err = provider.HasInstance(node5)
+	assert.NoError(t, err)
+	assert.False(t, present)
+}
+
+func TestHasInstanceAfterDeleteNodes(t *testing.T) {
+	a := &autoScalingMock{}
+	m := newTestAwsManagerWithAsgs(t, a, nil, []string{"1:5:test-asg"})
+	provider := testProvider(t, m)
+	asgs := provider.NodeGroups()
+
+	markHealthy := func(out *autoscaling.DescribeAutoScalingGroupsOutput) *autoscaling.DescribeAutoScalingGroupsOutput {
+		for i := range out.AutoScalingGroups[0].Instances {
+			out.AutoScalingGroups[0].Instances[i].HealthStatus = aws.String("Healthy")
+		}
+		return out
+	}
+
+	a.On("TerminateInstanceInAutoScalingGroup",
+		mock.Anything,
+		&autoscaling.TerminateInstanceInAutoScalingGroupInput{
+			InstanceId:                     aws.String("test-instance-id"),
+			ShouldDecrementDesiredCapacity: aws.Bool(true),
+		},
+	).Return(&autoscaling.TerminateInstanceInAutoScalingGroupOutput{
+		Activity: &autoscalingtypes.Activity{Description: aws.String("Deleted instance")},
+	}, nil)
+
+	a.On("DescribeAutoScalingGroups",
+		mock.Anything,
+		&autoscaling.DescribeAutoScalingGroupsInput{
+			AutoScalingGroupNames: []string{"test-asg"},
+			MaxRecords:            aws.Int32(maxRecordsReturnedByAPI),
+		},
+	).Return(markHealthy(testNamedDescribeAutoScalingGroupsOutput("test-asg", 2, "test-instance-id", "second-test-instance-id")), nil).Once()
+
+	err := provider.Refresh()
+	assert.NoError(t, err)
+
+	node := &apiv1.Node{
+		Spec: apiv1.NodeSpec{
+			ProviderID: "aws:///us-east-1a/test-instance-id",
+		},
+	}
+	remainingNode := &apiv1.Node{
+		Spec: apiv1.NodeSpec{
+			ProviderID: "aws:///us-east-1a/second-test-instance-id",
+		},
+	}
+
+	err = asgs[0].DeleteNodes([]*apiv1.Node{node})
+	assert.NoError(t, err)
+
+	// The instance must be reported as gone right after DeleteNodes(),
+	// before the cache is regenerated
+	present, err := provider.HasInstance(node)
+	assert.NoError(t, err)
+	assert.False(t, present)
+
+	// An eventually consistent API response may still report the terminated
+	// instance as InService - it must not flip the instance back to present
+	a.On("DescribeAutoScalingGroups",
+		mock.Anything,
+		&autoscaling.DescribeAutoScalingGroupsInput{
+			AutoScalingGroupNames: []string{"test-asg"},
+			MaxRecords:            aws.Int32(maxRecordsReturnedByAPI),
+		},
+	).Return(markHealthy(testNamedDescribeAutoScalingGroupsOutput("test-asg", 2, "test-instance-id", "second-test-instance-id")), nil).Once()
+
+	err = m.forceRefresh()
+	assert.NoError(t, err)
+
+	present, err = provider.HasInstance(node)
+	assert.NoError(t, err)
+	assert.False(t, present)
+
+	// Until its termination completes, the instance keeps showing up in
+	// the ASG API responses in the Terminating lifecycle state
+	terminatingOutput := markHealthy(testNamedDescribeAutoScalingGroupsOutput("test-asg", 1, "test-instance-id", "second-test-instance-id"))
+	terminatingOutput.AutoScalingGroups[0].Instances[0].LifecycleState = autoscalingtypes.LifecycleStateTerminating
+	a.On("DescribeAutoScalingGroups",
+		mock.Anything,
+		&autoscaling.DescribeAutoScalingGroupsInput{
+			AutoScalingGroupNames: []string{"test-asg"},
+			MaxRecords:            aws.Int32(maxRecordsReturnedByAPI),
+		},
+	).Return(terminatingOutput, nil)
+
+	err = m.forceRefresh()
+	assert.NoError(t, err)
+
+	// The instance must still be reported as gone after the regenerated
+	// cache picked it up from the API in the Terminating state
+	present, err = provider.HasInstance(node)
+	assert.NoError(t, err)
+	assert.False(t, present)
+
+	// The remaining instance is unaffected
+	present, err = provider.HasInstance(remainingNode)
+	assert.NoError(t, err)
+	assert.True(t, present)
 }
 
 func TestDeleteNodesWithPlaceholderAndStaleCache(t *testing.T) {