Suspend and Resume ASG Processes

Author: Skarlso

config/crd/bases/infrastructure.cluster.x-k8s.io_awsmachinepools.yaml

@@ -846,6 +846,13 @@ spec:
                       type: string
                   type: object
                 type: array
+              suspendProcesses:
+                description: SuspendProcesses defines a list of processes to suspend
+                  for the given ASG. This is constantly reconciled. If a process is
+                  removed from this list it will automatically be resumed.
+                items:
+                  type: string
+                type: array
             required:
             - awsLaunchTemplate
             - maxSize

docs/book/src/topics/suspend-asg-processes.md

@@ -0,0 +1,60 @@
+# Suspend ASG Processes
+
+- **Feature status:** Experimental
+- **Feature gate:** MachinePool=true
+
+MachinePool allows users to manage many machines as a single entity. Infrastructure providers implement a separate CRD that handles infrastructure side of the feature.
+
+## Suspend Processes
+
+It's possible to suspend certain processes for ASG. The list of processes can be found [here](https://docs.aws.amazon.com/autoscaling/ec2/APIReference/API_SuspendProcesses.html).
+
+To utilize this feature, simply denote the list of processes that are desired to be suspended.
+
+```yaml
+---
+apiVersion: infrastructure.cluster.x-k8s.io/v1beta2
+kind: AWSMachinePool
+metadata:
+  name: capa-mp-0
+spec:
+  minSize: 1
+  maxSize: 10
+  availabilityZones:
+    - "${AWS_AVAILABILITY_ZONE}"
+  awsLaunchTemplate:
+    instanceType: "${AWS_CONTROL_PLANE_MACHINE_TYPE}"
+    sshKeyName: "${AWS_SSH_KEY_NAME}"  
+  suspendProcesses:
+    - Launch
+    - AlarmNotification
+    - AZRebalance
+---
+```
+
+## Resume Processes
+
+If a service is desired to be resumed, simply remove it from the list of suspended processes. The reconciler will then
+resume any process that is not part of the desired suspended processes list.
+
+```yaml
+---
+apiVersion: infrastructure.cluster.x-k8s.io/v1beta2
+kind: AWSMachinePool
+metadata:
+  name: capa-mp-0
+spec:
+  minSize: 1
+  maxSize: 10
+  availabilityZones:
+    - "${AWS_AVAILABILITY_ZONE}"
+  awsLaunchTemplate:
+    instanceType: "${AWS_CONTROL_PLANE_MACHINE_TYPE}"
+    sshKeyName: "${AWS_SSH_KEY_NAME}"  
+  suspendProcesses:
+    - Launch
+---
+```
+
+_Note_ that now `AlarmNotification` and `AZRebalance` will be resumed, but the reconciler will not try to suspend
+`Launch` again. So it doesn't incur additional expensive, redundant API calls.

exp/api/v1beta1/conversion.go

@@ -21,6 +21,7 @@ import (
 	infrav1beta1 "sigs.k8s.io/cluster-api-provider-aws/api/v1beta1"
 	infrav1 "sigs.k8s.io/cluster-api-provider-aws/api/v1beta2"
 	infrav1exp "sigs.k8s.io/cluster-api-provider-aws/exp/api/v1beta2"
+	utilconversion "sigs.k8s.io/cluster-api/util/conversion"
 	"sigs.k8s.io/controller-runtime/pkg/conversion"
 )
 
@@ -30,7 +31,19 @@ func (src *AWSMachinePool) ConvertTo(dstRaw conversion.Hub) error {
 	if err := Convert_v1beta1_AWSMachinePool_To_v1beta2_AWSMachinePool(src, dst, nil); err != nil {
 		return err
 	}
-	
+
+	// Manually restore data.
+	restored := &infrav1exp.AWSMachinePool{}
+	if ok, err := utilconversion.UnmarshalData(src, restored); err != nil || !ok {
+		return err
+	}
+
+	if restored.Spec.SuspendProcesses != nil {
+		restoredProcesses := make([]string, len(restored.Spec.SuspendProcesses))
+		copy(restoredProcesses, restored.Spec.SuspendProcesses)
+		dst.Spec.SuspendProcesses = restoredProcesses
+	}
+
 	return nil
 }
 
@@ -41,8 +54,8 @@ func (r *AWSMachinePool) ConvertFrom(srcRaw conversion.Hub) error {
 	if err := Convert_v1beta2_AWSMachinePool_To_v1beta1_AWSMachinePool(src, r, nil); err != nil {
 		return err
 	}
-	
-	return nil
+
+	return utilconversion.MarshalData(src, r)
 }
 
 // ConvertTo converts the v1beta1 AWSMachinePoolList receiver to a v1beta2 AWSMachinePoolList.
@@ -75,7 +88,7 @@ func (r *AWSManagedMachinePool) ConvertFrom(srcRaw conversion.Hub) error {
 	if err := Convert_v1beta2_AWSManagedMachinePool_To_v1beta1_AWSManagedMachinePool(src, r, nil); err != nil {
 		return err
 	}
-	
+
 	return nil
 }
 
@@ -147,3 +160,20 @@ func Convert_v1beta1_Instance_To_v1beta2_Instance(in *infrav1beta1.Instance, out
 func Convert_v1beta2_AWSLaunchTemplate_To_v1beta1_AWSLaunchTemplate(in *infrav1exp.AWSLaunchTemplate, out *AWSLaunchTemplate, s apiconversion.Scope) error {
 	return autoConvert_v1beta2_AWSLaunchTemplate_To_v1beta1_AWSLaunchTemplate(in, out, s)
 }
+
+func Convert_v1beta1_AWSMachinePoolSpec_To_v1beta2_AWSMachinePoolSpec(in *AWSMachinePoolSpec, out *infrav1exp.AWSMachinePoolSpec, s apiconversion.Scope) error {
+	return autoConvert_v1beta1_AWSMachinePoolSpec_To_v1beta2_AWSMachinePoolSpec(in, out, s)
+}
+
+func Convert_v1beta2_AWSMachinePoolSpec_To_v1beta1_AWSMachinePoolSpec(in *infrav1exp.AWSMachinePoolSpec, out *AWSMachinePoolSpec, s apiconversion.Scope) error {
+	return autoConvert_v1beta2_AWSMachinePoolSpec_To_v1beta1_AWSMachinePoolSpec(in, out, s)
+}
+
+func Convert_v1beta1_AutoScalingGroup_To_v1beta2_AutoScalingGroup(in *AutoScalingGroup, out *infrav1exp.AutoScalingGroup, s apiconversion.Scope) error {
+	return autoConvert_v1beta1_AutoScalingGroup_To_v1beta2_AutoScalingGroup(in, out, s)
+}
+
+func Convert_v1beta2_AutoScalingGroup_To_v1beta1_AutoScalingGroup(in *infrav1exp.AutoScalingGroup, out *AutoScalingGroup, s apiconversion.Scope) error {
+	// explicitly ignore CurrentlySuspended.
+	return autoConvert_v1beta2_AutoScalingGroup_To_v1beta1_AutoScalingGroup(in, out, s)
+}

exp/api/v1beta1/zz_generated.conversion.go

@@ -83,6 +83,10 @@ type AWSMachinePoolSpec struct {
 	// Enable or disable the capacity rebalance autoscaling group feature
 	// +optional
 	CapacityRebalance bool `json:"capacityRebalance,omitempty"`
+
+	// SuspendProcesses defines a list of processes to suspend for the given ASG. This is constantly reconciled.
+	// If a process is removed from this list it will automatically be resumed.
+	SuspendProcesses []string `json:"suspendProcesses,omitempty"`
 }
 
 // RefreshPreferences defines the specs for instance refreshing.

exp/api/v1beta2/awsmachinepool_types.go

@@ -193,9 +193,10 @@ type AutoScalingGroup struct {
 	DefaultCoolDown   metav1.Duration `json:"defaultCoolDown,omitempty"`
 	CapacityRebalance bool            `json:"capacityRebalance,omitempty"`
 
-	MixedInstancesPolicy *MixedInstancesPolicy `json:"mixedInstancesPolicy,omitempty"`
-	Status               ASGStatus
-	Instances            []infrav1.Instance `json:"instances,omitempty"`
+	MixedInstancesPolicy      *MixedInstancesPolicy `json:"mixedInstancesPolicy,omitempty"`
+	Status                    ASGStatus
+	Instances                 []infrav1.Instance `json:"instances,omitempty"`
+	CurrentlySuspendProcesses []string           `json:"currentlySuspendProcesses,omitempty"`
 }
 
 // ASGStatus is a status string returned by the autoscaling API.

exp/api/v1beta2/types.go

@@ -374,16 +374,64 @@ func (r *AWSMachinePoolReconciler) reconcileDelete(machinePoolScope *scope.Machi
 }
 
 func (r *AWSMachinePoolReconciler) updatePool(machinePoolScope *scope.MachinePoolScope, clusterScope cloud.ClusterScoper, existingASG *expinfrav1.AutoScalingGroup) error {
+	asgSvc := r.getASGService(clusterScope)
 	if asgNeedsUpdates(machinePoolScope, existingASG) {
 		machinePoolScope.Info("updating AutoScalingGroup")
-		asgSvc := r.getASGService(clusterScope)
 
 		if err := asgSvc.UpdateASG(machinePoolScope); err != nil {
 			r.Recorder.Eventf(machinePoolScope.AWSMachinePool, corev1.EventTypeWarning, "FailedUpdate", "Failed to update ASG: %v", err)
 			return errors.Wrap(err, "unable to update ASG")
 		}
 	}
 
+	if !cmp.Equal(existingASG.CurrentlySuspendProcesses, machinePoolScope.AWSMachinePool.Spec.SuspendProcesses) {
+		var (
+			toBeSuspended []string
+			toBeResumed   []string
+
+			currentlySuspended = make(map[string]struct{})
+			desiredSuspended   = make(map[string]struct{})
+		)
+
+		// Convert the items to a map, so it's easy to create an effective diff from these two slices.
+		for _, p := range existingASG.CurrentlySuspendProcesses {
+			currentlySuspended[p] = struct{}{}
+		}
+
+		for _, p := range machinePoolScope.AWSMachinePool.Spec.SuspendProcesses {
+			desiredSuspended[p] = struct{}{}
+		}
+
+		// Anything that remains in the desired items is not currently suspended so must be suspended.
+		// Anything that remains in the currentlySuspended list must be resumed since they were not part of
+		// desiredSuspended.
+		for k := range desiredSuspended {
+			if _, ok := currentlySuspended[k]; ok {
+				delete(desiredSuspended, k)
+			}
+			delete(currentlySuspended, k)
+		}
+
+		// Convert them back into lists so
+		for k := range desiredSuspended {
+			toBeSuspended = append(toBeSuspended, k)
+		}
+
+		for k := range currentlySuspended {
+			toBeResumed = append(toBeResumed, k)
+		}
+
+		if len(toBeSuspended) > 0 {
+			if err := asgSvc.SuspendProcesses(existingASG.Name, toBeSuspended); err != nil {
+				return errors.Wrapf(err, "failed to suspend processes while trying update pool")
+			}
+		}
+		if len(toBeResumed) > 0 {
+			if err := asgSvc.ResumeProcesses(existingASG.Name, toBeResumed); err != nil {
+				return errors.Wrapf(err, "failed to resume processes while trying update pool")
+			}
+		}
+	}
 	return nil
 }
 
@@ -397,7 +445,9 @@ func (r *AWSMachinePoolReconciler) createPool(machinePoolScope *scope.MachinePoo
 	if err != nil {
 		return nil, errors.Wrapf(err, "failed to create AWSMachinePool")
 	}
-
+	if err := asgsvc.SuspendProcesses(asg.Name, machinePoolScope.AWSMachinePool.Spec.SuspendProcesses); err != nil {
+		return nil, errors.Wrapf(err, "failed to suspend processes while trying to create Pool")
+	}
 	return asg, nil
 }