feat(instance): improve instance resource handling and updates (#291)

Issue [#1709](aws-controllers-k8s/community#1709)

Description of changes:
- Add InvalidInstanceID.NotFound error code for 404 exceptions
- Implement instance state handling and requeue logic
- Add support for modifying instance attributes (security groups, API termination, etc.)
- Add helper functions for instance state checks and field updates
- Update E2E tests to verify instance updates and security group modifications
- Set additional fields like SecurityGroupIDs and Monitoring in instance spec
- Improve instance syncing and status conditions

This commit enhances the EC2 instance resource controller by adding better error handling,
state management, and support for modifying various instance attributes. It also includes
improvements to E2E tests and resource syncing.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.

Author: michaelhtm

apis/v1alpha1/generator.yaml

@@ -362,6 +362,10 @@ resources:
     update_operation:
       custom_method_name: customUpdateDHCPOptions
   Instance:
+    exceptions:
+      errors:
+        404:
+          code: InvalidInstanceID.NotFound
     fields:
       HibernationOptions:
         late_initialize: {}

generator.yaml

@@ -362,6 +362,10 @@ resources:
     update_operation:
       custom_method_name: customUpdateDHCPOptions
   Instance:
+    exceptions:
+      errors:
+        404:
+          code: InvalidInstanceID.NotFound
     fields:
       HibernationOptions:
         late_initialize: {}

pkg/resource/instance/hooks.go

@@ -16,15 +16,24 @@ package instance
 import (
 	"context"
 	"errors"
+	"fmt"
+	"time"
 
 	ackcompare "github.com/aws-controllers-k8s/runtime/pkg/compare"
+	ackrequeue "github.com/aws-controllers-k8s/runtime/pkg/requeue"
 	ackrtlog "github.com/aws-controllers-k8s/runtime/pkg/runtime/log"
+	"github.com/aws/aws-sdk-go-v2/aws"
 	svcsdk "github.com/aws/aws-sdk-go-v2/service/ec2"
 	svcsdktypes "github.com/aws/aws-sdk-go-v2/service/ec2/types"
 
+	"github.com/aws-controllers-k8s/ec2-controller/apis/v1alpha1"
 	"github.com/aws-controllers-k8s/ec2-controller/pkg/tags"
 )
 
+const (
+	requeueUntilReadyDuration = 10 * time.Second
+)
+
 // addInstanceIDsToTerminateRequest populates the list of InstanceIDs
 // in the TerminateInstances request with the resource's InstanceID
 // Return error to indicate to callers that the resource is not yet created.
@@ -45,27 +54,123 @@ func (rm *resourceManager) customUpdateInstance(
 ) (updated *resource, err error) {
 	rlog := ackrtlog.FromContext(ctx)
 	exit := rlog.Trace("rm.customUpdateInstance")
-	defer exit(err)
+	defer func() { exit(err) }()
 
 	// Default `updated` to `desired` because it is likely
 	// EC2 `modify` APIs do NOT return output, only errors.
 	// If the `modify` calls (i.e. `sync`) do NOT return
 	// an error, then the update was successful and desired.Spec
 	// (now updated.Spec) reflects the latest resource state.
 	updated = rm.concreteResource(desired.DeepCopy())
+	updated.SetStatus(latest)
 
 	if delta.DifferentAt("Spec.Tags") {
 		if err := tags.Sync(
 			ctx, rm.sdkapi, rm.metrics, *latest.ko.Status.InstanceID,
 			desired.ko.Spec.Tags, latest.ko.Spec.Tags,
 		); err != nil {
-			return nil, err
+			return updated, err
 		}
 	}
 
+	if !delta.DifferentExcept("Spec.Tags") {
+		return updated, nil
+	}
+
+	if !isRunning(updated.ko) {
+		return updated, ackrequeue.NeededAfter(
+			fmt.Errorf("requeuing until state is %s or %s", svcsdktypes.InstanceStateNameRunning, svcsdktypes.InstanceStateNameStopped),
+			requeueUntilReadyDuration,
+		)
+	}
+
+	err = rm.modifyInstanceAttributes(ctx, delta, desired, latest)
+	if err != nil {
+		return updated, err
+	}
+
 	return updated, nil
 }
 
+func (rm *resourceManager) modifyInstanceAttributes(ctx context.Context, delta *ackcompare.Delta, desired, latest *resource) (err error) {
+	rlog := ackrtlog.FromContext(ctx)
+	exit := rlog.Trace("rm.modifyInstanceAttributes")
+	defer func() { exit(err) }()
+	input := &svcsdk.ModifyInstanceAttributeInput{
+		InstanceId: latest.ko.Status.InstanceID,
+	}
+	// we can only update one attribute at a time
+	if delta.DifferentAt("Spec.DisableAPITermination") {
+		input.DisableApiTermination = &svcsdktypes.AttributeBooleanValue{Value: desired.ko.Spec.DisableAPITermination}
+	} else if delta.DifferentAt("Spec.InstanceType") {
+		input.InstanceType = &svcsdktypes.AttributeValue{Value: desired.ko.Spec.InstanceType}
+	} else if delta.DifferentAt("Spec.KernelID") {
+		input.Kernel = &svcsdktypes.AttributeValue{Value: desired.ko.Spec.KernelID}
+	} else if delta.DifferentAt("Spec.RAMDiskID") {
+		input.Ramdisk = &svcsdktypes.AttributeValue{Value: desired.ko.Spec.RAMDiskID}
+	} else if delta.DifferentAt("Spec.InstanceInitiatedShutdownBehavior") {
+		input.InstanceInitiatedShutdownBehavior = &svcsdktypes.AttributeValue{Value: desired.ko.Spec.InstanceInitiatedShutdownBehavior}
+	} else if delta.DifferentAt("Spec.UserData") {
+		input.UserData = &svcsdktypes.BlobAttributeValue{Value: []byte(aws.ToString(desired.ko.Spec.UserData))}
+	} else if delta.DifferentAt("Spec.EBSOptimized") {
+		input.EbsOptimized = &svcsdktypes.AttributeBooleanValue{Value: desired.ko.Spec.EBSOptimized}
+	} else if delta.DifferentAt("Spec.DisableAPIStop") {
+		input.DisableApiStop = &svcsdktypes.AttributeBooleanValue{Value: desired.ko.Spec.DisableAPIStop}
+	} else if delta.DifferentAt("Spec.SecurityGroupIDs") {
+		input.Groups = aws.ToStringSlice(desired.ko.Spec.SecurityGroupIDs)
+	} else {
+		input = nil
+	}
+
+	if input != nil {
+		_, err = rm.sdkapi.ModifyInstanceAttribute(ctx, input)
+		rm.metrics.RecordAPICall("UPDATE", "ModifyInstanceAttribute", err)
+		if err != nil {
+			return err
+		}
+		return fmt.Errorf("requeuing until all fields are updated")
+	}
+	return nil
+}
+
+func isRunning(ko *v1alpha1.Instance) bool {
+	if ko.Status.State == nil || ko.Status.State.Name == nil {
+		return false
+	}
+
+	// NOTE: (michaelhtm) We will count `stopped` as running for now.
+	// TODO: expose annotation to allow users to start/stop instances
+	return *ko.Status.State.Name == string(svcsdktypes.InstanceStateNameRunning) ||
+		*ko.Status.State.Name == string(svcsdktypes.InstanceStateNameStopped)
+}
+
+// needsRestart checks if the Instance is terminated (deleted)
+func needsRestart(ko *v1alpha1.Instance) bool {
+	if ko.Status.State == nil || ko.Status.State.Name == nil {
+		return false
+	}
+
+	return *ko.Status.State.Name == string(svcsdktypes.InstanceStateNameTerminated)
+}
+
+
+func setAdditionalFields(instance svcsdktypes.Instance, ko *v1alpha1.Instance) {
+	ko.Spec.SecurityGroupIDs = []*string{}
+	for _, group := range instance.SecurityGroups {
+		ko.Spec.SecurityGroupIDs = append(ko.Spec.SecurityGroupIDs, group.GroupId)
+	}
+
+	if monitoring := instance.Monitoring; monitoring != nil {
+		switch monitoring.State {
+		case svcsdktypes.MonitoringStateDisabled, svcsdktypes.MonitoringStateDisabling:
+			ko.Spec.Monitoring = &v1alpha1.RunInstancesMonitoringEnabled{Enabled: aws.Bool(false)}
+
+		case svcsdktypes.MonitoringStateEnabled, svcsdktypes.MonitoringStatePending:
+			ko.Spec.Monitoring = &v1alpha1.RunInstancesMonitoringEnabled{Enabled: aws.Bool(true)}
+		}
+	}
+}
+
 var computeTagsDelta = tags.ComputeTagsDelta
 
 // updateTagSpecificationsInCreateRequest adds

pkg/resource/instance/sdk.go

@@ -1,3 +1,5 @@
+	setAdditionalFields(resp.Instances[0], ko)
+	
 	toAdd, toDelete := computeTagsDelta(desired.ko.Spec.Tags, ko.Spec.Tags)
 	if len(toAdd) == 0 && len(toDelete) == 0 {
 		// if desired tags and response tags are equal,

templates/hooks/instance/sdk_create_post_set_output.go.tpl

@@ -1,8 +1,18 @@
+	// Here we want to check if the instance is terminated(deleted)
+	// returning NotFound will trigger a create
+	if needsRestart(ko) {
+		return nil, ackerr.NotFound
+	}
+
+	setAdditionalFields(resp.Reservations[0].Instances[0], ko)
+
+	if !isRunning(ko) {
+		ackcondition.SetSynced(&resource{ko}, corev1.ConditionFalse, nil, aws.String("waiting for resource to be running"))
+	}
 
 	toAdd, toDelete := computeTagsDelta(r.ko.Spec.Tags, ko.Spec.Tags)
 	if len(toAdd) == 0 && len(toDelete) == 0 {
 		// if resource's initial tags and response tags are equal,
 		// then assign resource's tags to maintain tag order
 		ko.Spec.Tags = r.ko.Spec.Tags
 	}
-    

templates/hooks/instance/sdk_read_many_post_set_output.go.tpl

@@ -135,7 +135,7 @@ def instance(ec2_client):
 @service_marker
 @pytest.mark.canary
 class TestInstance:
-    def test_create_delete(self, ec2_client, instance):
+    def test_crud(self, ec2_client, instance):
         (ref, cr) = instance
         resource_id = cr["status"]["instanceID"]
 
@@ -156,6 +156,39 @@ def test_create_delete(self, ec2_client, instance):
                     t['Value'] == INSTANCE_TAG_VAL):
                 tag_present = True
         assert tag_present
+        
+        # Check resource synced successfully
+        assert k8s.wait_on_condition(ref, "ACK.ResourceSynced", "True", wait_periods=5)
+
+        # Ensure instance is running
+        cr = k8s.get_resource(ref)
+        assert 'status' in cr
+        assert 'state' in cr['status']
+        assert 'name' in cr['status']['state']
+        assert cr['status']['state']['name'] == 'running'
+        
+        # Update Instance securityGroupID
+        test_vpc = get_bootstrap_resources().SharedTestVPC
+        updates = {
+            "spec": {
+                "securityGroupIDs": [test_vpc.security_group.group_id]
+            }
+        }
+        k8s.patch_custom_resource(ref, updates)
+        time.sleep(MODIFY_WAIT_AFTER_SECONDS)
+
+        # Check resource synced successfully
+        assert k8s.wait_on_condition(ref, "ACK.ResourceSynced", "True", wait_periods=5)
+
+        # Check Instance updated value
+        instance = get_instance(ec2_client, resource_id)
+        assert instance is not None
+        assert 'SecurityGroups' in instance
+        foundSecurityGroup = False
+        for group in instance['SecurityGroups']:
+            if group['GroupId'] == test_vpc.security_group.group_id:
+                foundSecurityGroup = True
+        assert foundSecurityGroup
 
         # Delete k8s resource
         _, deleted = k8s.delete_custom_resource(ref, 2, 5)