Controller v2: Implementation of Instance termination

api/v1alpha1/rollingupgrade_types.go

@@ -17,7 +17,6 @@ limitations under the License.
 package v1alpha1
 
 import (
-
 	"fmt"
 	"strconv"
 	"strings"
@@ -238,7 +237,7 @@ func (r *RollingUpgrade) Validate() (bool, error) {
 	// validating the Mode value
 	if strategy.Mode == "" {
 		r.Spec.Strategy.Mode = UpdateStrategyModeLazy
-	} else if !common.ContainsEqualFold(AllowedStrategyMode, strategy.Mode) {
+	} else if !common.ContainsEqualFold(AllowedStrategyMode, string(strategy.Mode)) {
 		err := fmt.Errorf("%s: Invalid value for startegy Mode - %d", r.Name, strategy.MaxUnavailable.IntVal)
 		return false, err
 	}

controllers/providers/aws/autoscaling.go

@@ -17,9 +17,18 @@ limitations under the License.
 package aws
 
 import (
+	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/autoscaling"
 )
 
+var (
+	TerminatingInstanceStates = []string{
+		autoscaling.LifecycleStateTerminating,
+		autoscaling.LifecycleStateTerminatingWait,
+		autoscaling.LifecycleStateTerminatingProceed,
+	}
+)
+
 func (a *AmazonClientSet) DescribeScalingGroups() ([]*autoscaling.Group, error) {
 	scalingGroups := []*autoscaling.Group{}
 	err := a.AsgClient.DescribeAutoScalingGroupsPages(&autoscaling.DescribeAutoScalingGroupsInput{}, func(page *autoscaling.DescribeAutoScalingGroupsOutput, lastPage bool) bool {
@@ -31,3 +40,16 @@ func (a *AmazonClientSet) DescribeScalingGroups() ([]*autoscaling.Group, error)
 	}
 	return scalingGroups, nil
 }
+
+func (a *AmazonClientSet) TerminateInstance(instance *autoscaling.Instance) error {
+	instanceID := aws.StringValue(instance.InstanceId)
+	input := &autoscaling.TerminateInstanceInAutoScalingGroupInput{
+		InstanceId:                     aws.String(instanceID),
+		ShouldDecrementDesiredCapacity: aws.Bool(false),
+	}
+
+	if _, err := a.AsgClient.TerminateInstanceInAutoScalingGroup(input); err != nil {
+		return err
+	}
+	return nil
+}

controllers/providers/aws/ec2.go

@@ -57,3 +57,17 @@ func (a *AmazonClientSet) DescribeTaggedInstanceIDs(tagKey, tagValue string) ([]
 	})
 	return instances, err
 }
+
+func (a *AmazonClientSet) TagEC2instance(instanceID, tagKey, tagValue string) error {
+	input := &ec2.CreateTagsInput{
+		Resources: aws.StringSlice([]string{instanceID}),
+		Tags: []*ec2.Tag{
+			{
+				Key:   aws.String(tagKey),
+				Value: aws.String(tagValue),
+			},
+		},
+	}
+	_, err := a.Ec2Client.CreateTags(input)
+	return err
+}

controllers/providers/aws/utils.go

@@ -96,7 +96,6 @@ func GetScalingAZs(instances []*autoscaling.Instance) []string {
 // 	return &autoscaling.Instance{}
 // }
 
-
 // func ListScalingInstanceIDs(group *autoscaling.Group) []string {
 // 	instanceIDs := make([]string, 0)
 // 	for _, instance := range group.Instances {

controllers/rollingupgrade_controller.go

@@ -123,11 +123,11 @@ func (r *RollingUpgradeReconciler) Reconcile(ctx context.Context, req ctrl.Reque
 	}
 
 	r.Info("admitted new rollingupgrade", "name", req.NamespacedName, "scalingGroup", scalingGroupName)
-	r.AdmissionMap.Store(req.NamespacedName, scalingGroupName)
+	r.AdmissionMap.Store(rollingUpgrade.NamespacedName(), scalingGroupName)
 	rollingUpgrade.SetCurrentStatus(v1alpha1.StatusInit)
 
-	discoveredState := NewDiscoveredState(r.Auth, r.Logger)
-	if err := discoveredState.Discover(); err != nil {
+	r.Cloud = NewDiscoveredState(r.Auth, r.Logger)
+	if err := r.Cloud.Discover(); err != nil {
 		rollingUpgrade.SetCurrentStatus(v1alpha1.StatusError)
 		return ctrl.Result{}, err
 	}

controllers/upgrade.go

@@ -23,6 +23,7 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/autoscaling"
 	"github.com/keikoproj/upgrade-manager/api/v1alpha1"
+	"github.com/keikoproj/upgrade-manager/controllers/common"
 	awsprovider "github.com/keikoproj/upgrade-manager/controllers/providers/aws"
 	kubeprovider "github.com/keikoproj/upgrade-manager/controllers/providers/kubernetes"
 	"k8s.io/apimachinery/pkg/util/intstr"
@@ -63,8 +64,17 @@ func (r *RollingUpgradeReconciler) RotateNodes(rollingUpgrade *v1alpha1.RollingU
 	)
 
 	rollingUpgrade.SetTotalNodes(len(scalingGroup.Instances))
+
+	// check if all instances are rotated.
+	if r.IsScalingGroupDrifted(rollingUpgrade) {
+		rollingUpgrade.SetCurrentStatus(v1alpha1.StatusComplete)
+		return nil
+	}
+
 	rotationTargets := r.SelectTargets(rollingUpgrade, scalingGroup)
-	r.ReplaceNodeBatch(rollingUpgrade, rotationTargets)
+	if ok, err := r.ReplaceNodeBatch(rollingUpgrade, rotationTargets); !ok {
+		return err
+	}
 
 	return nil
 }
@@ -79,6 +89,7 @@ func (r *RollingUpgradeReconciler) ReplaceNodeBatch(rollingUpgrade *v1alpha1.Rol
 		for _, target := range batch {
 			_ = target
 			// Add in-progress tag
+			r.Auth.TagEC2instance(aws.StringValue(target.InstanceId), instanceStateTagKey, inProgressTagValue)
 
 			// Standby
 
@@ -97,6 +108,10 @@ func (r *RollingUpgradeReconciler) ReplaceNodeBatch(rollingUpgrade *v1alpha1.Rol
 			// Is drained?
 
 			// Terminate - set lastTerminateTime
+			if err := r.Auth.TerminateInstance(target); err != nil {
+				r.Info("failed to terminate instance", "name", rollingUpgrade.NamespacedName(), "instance", aws.StringValue(target.InstanceId))
+				return true, nil
+			}
 		}
 	case v1alpha1.UpdateStrategyModeLazy:
 		for _, target := range batch {
@@ -187,6 +202,10 @@ func (r *RollingUpgradeReconciler) IsInstanceDrifted(rollingUpgrade *v1alpha1.Ro
 		instanceID       = aws.StringValue(instance.InstanceId)
 	)
 
+	// if an instance is in terminating state, ignore.
+	if common.ContainsEqualFold(awsprovider.TerminatingInstanceStates, aws.StringValue(instance.LifecycleState)) {
+		return false
+	}
 	// check if there is atleast one node that meets the force-referesh criteria
 	if rollingUpgrade.IsForceRefresh() {
 		var (
@@ -257,3 +276,14 @@ func (r *RollingUpgradeReconciler) IsInstanceDrifted(rollingUpgrade *v1alpha1.Ro
 	r.Info("node refresh not required", "name", rollingUpgrade.NamespacedName(), "instance", instanceID)
 	return false
 }
+
+func (r *RollingUpgradeReconciler) IsScalingGroupDrifted(rollingUpgrade *v1alpha1.RollingUpgrade) bool {
+	r.Info("checking if rolling upgrade is completed", "name", rollingUpgrade.NamespacedName())
+	scalingGroup := awsprovider.SelectScalingGroup(rollingUpgrade.ScalingGroupName(), r.Cloud.ScalingGroups)
+	for _, instance := range scalingGroup.Instances {
+		if r.IsInstanceDrifted(rollingUpgrade, instance) {
+			return false
+		}
+	}
+	return true
+}