Repository: keikoproj/instance-manager (go)

Issue:
CRD Strategy: "Replace" ConcurrencyPolicy
We should add a new option to CRD Strategy concurrencyPolicy called "Repalce", which will delete the custom resource and replace it with a new one when there is a new launch config.

In the case of RollingUpgrade this really helps with modifications that happen during the upgrade, for example if the spot price changes, you don't have to wait for the upgrade to complete, you can just stop the upgrade by deleting the custom resource and submit a new one.

Pull request description:
Add "Replace" ConcurrencyPolicy
Fixes #135 

This adds a new concurrencyPolicy to CRD strategy called "Replace" which will replace the submitted CR with a new one.



Patch:
diff --git a/api/v1alpha1/instancegroup_types.go b/api/v1alpha1/instancegroup_types.go
index a393939..08309be 100644
--- a/api/v1alpha1/instancegroup_types.go
+++ b/api/v1alpha1/instancegroup_types.go
@@ -56,6 +56,10 @@ const (
 	EKSFargateProvisionerName = "eks-fargate"
 
 	NodesReady InstanceGroupConditionType = "NodesReady"
+
+	ForbidConcurrencyPolicy  = "forbid"
+	AllowConcurrencyPolicy   = "allow"
+	ReplaceConcurrencyPolicy = "replace"
 )
 
 var (
@@ -474,12 +478,10 @@ func (c *CRDUpdateStrategy) Validate() error {
 		return errors.New("spec is empty")
 	}
 
-	if strings.ToLower(c.ConcurrencyPolicy) != "forbid" && strings.ToLower(c.ConcurrencyPolicy) != "allow" {
-		c.SetConcurrencyPolicy("forbid")
-	}
+	allowedPolicies := []string{ReplaceConcurrencyPolicy, AllowConcurrencyPolicy, ForbidConcurrencyPolicy}
 
-	if strings.ToLower(c.ConcurrencyPolicy) == "" {
-		c.SetConcurrencyPolicy("forbid")
+	if !common.ContainsEqualFold(allowedPolicies, c.ConcurrencyPolicy) {
+		c.SetConcurrencyPolicy(ForbidConcurrencyPolicy)
 	}
 
 	if c.GetCRDName() == "" {
diff --git a/controllers/providers/kubernetes/crd.go b/controllers/providers/kubernetes/crd.go
index da680f1..6df6a49 100644
--- a/controllers/providers/kubernetes/crd.go
+++ b/controllers/providers/kubernetes/crd.go
@@ -33,7 +33,6 @@ const (
 	OwnershipAnnotationKey   = "app.kubernetes.io/managed-by"
 	ScopeAnnotationKey       = "instancemgr.keikoproj.io/upgrade-scope"
 	OwnershipAnnotationValue = "instance-manager"
-	DefaultConcurrencyPolicy = "forbid"
 )
 
 func ProcessCRDStrategy(kube dynamic.Interface, instanceGroup *v1alpha1.InstanceGroup) (bool, error) {
@@ -65,12 +64,12 @@ func ProcessCRDStrategy(kube dynamic.Interface, instanceGroup *v1alpha1.Instance
 	AddAnnotation(customResource, ScopeAnnotationKey, asgName)
 	GVR := GetGVR(customResource, strategy.GetCRDName())
 
+	launchID := common.GetLastElementBy(lcName, "-")
+	NormalizeName(customResource, launchID)
 	crdFullName := strings.Join([]string{GVR.Resource, GVR.Group}, ".")
 	if !CRDExists(kube, crdFullName) {
 		return false, errors.Errorf("custom resource definition '%v' is missing, could not upgrade", crdFullName)
 	}
-
-	NormalizeName(customResource, common.GetLastElementBy(lcName, "-"))
 	status.SetStrategyResourceName(customResource.GetName())
 
 	activeResources, err := GetActiveResources(kube, instanceGroup, customResource)
@@ -78,12 +77,30 @@ func ProcessCRDStrategy(kube dynamic.Interface, instanceGroup *v1alpha1.Instance
 		return false, errors.Wrap(err, "failed to discover active custom resources")
 	}
 
-	if len(activeResources) > 0 && strings.ToLower(strategy.GetConcurrencyPolicy()) == "forbid" {
+	if len(activeResources) > 0 && strings.EqualFold(strategy.GetConcurrencyPolicy(), v1alpha1.ForbidConcurrencyPolicy) {
 		log.Info("custom resource/s still active, will requeue", "instancegroup", instanceGroup.GetName())
 		instanceGroup.SetState(v1alpha1.ReconcileModifying)
 		return false, nil
 	}
 
+	var isRunning bool
+	if len(activeResources) > 0 && strings.EqualFold(strategy.GetConcurrencyPolicy(), v1alpha1.ReplaceConcurrencyPolicy) {
+		for _, resource := range activeResources {
+			if strings.HasSuffix(resource.GetName(), launchID) {
+				isRunning = true
+				continue
+			}
+			log.Info("active custom resource/s exists, will replace", "instancegroup", instanceGroup.GetName())
+			err = kube.Resource(GVR).Namespace(resource.GetNamespace()).Delete(resource.GetName(), &metav1.DeleteOptions{})
+			if err != nil {
+				return false, errors.Wrap(err, "failed to delete custom resource")
+			}
+		}
+		if isRunning {
+			return false, nil
+		}
+	}
+
 	err = SubmitCustomResource(kube, customResource, strategy.GetCRDName())
 	if err != nil {
 		return false, errors.Wrap(err, "failed to submit custom resource")
diff --git a/controllers/provisioners/eks/helpers.go b/controllers/provisioners/eks/helpers.go
index 34980c8..c67af82 100644
--- a/controllers/provisioners/eks/helpers.go
+++ b/controllers/provisioners/eks/helpers.go
@@ -242,7 +242,7 @@ func (ctx *EksInstanceGroupContext) discoverSpotPrice() error {
 	)
 
 	// Ignore recommendations until instance group is provisioned
-	if !state.IsNodesReady() {
+	if !state.IsProvisioned() {
 		return nil
 	}
 


Test patch:
diff --git a/controllers/provisioners/eks/cloud_test.go b/controllers/provisioners/eks/cloud_test.go
index 635bee4..fe381a1 100644
--- a/controllers/provisioners/eks/cloud_test.go
+++ b/controllers/provisioners/eks/cloud_test.go
@@ -20,12 +20,9 @@ import (
 	"testing"
 	"time"
 
-	corev1 "k8s.io/api/core/v1"
-
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/autoscaling"
 	"github.com/aws/aws-sdk-go/service/iam"
-	"github.com/keikoproj/instance-manager/api/v1alpha1"
 	"github.com/keikoproj/instance-manager/controllers/provisioners"
 	"github.com/onsi/gomega"
 )
@@ -170,10 +167,10 @@ func TestCloudDiscoverySpotPrice(t *testing.T) {
 	ig.SetName(resourceName)
 	ig.SetNamespace(resourceNamespace)
 	configuration.SetClusterName(clusterName)
-
-	asgMock.AutoScalingGroups = []*autoscaling.Group{
+	mockAsg := []*autoscaling.Group{
 		MockScalingGroup(ownedScalingGroupName, ownershipTag, nameTag, namespaceTag),
 	}
+	asgMock.AutoScalingGroups = mockAsg
 
 	asgMock.LaunchConfigurations = []*autoscaling.LaunchConfiguration{
 		{
@@ -191,17 +188,13 @@ func TestCloudDiscoverySpotPrice(t *testing.T) {
 	_, err = k.Kubernetes.CoreV1().Events("").Create(MockSpotEvent("1", ownedScalingGroupName, "0.80", true, time.Now()))
 	g.Expect(err).NotTo(gomega.HaveOccurred())
 
-	// recommendation should not be used if nodes are not ready
+	// recommendation should not be used if nodes are not provisioned yet
+	asgMock.AutoScalingGroups = []*autoscaling.Group{}
 	err = ctx.CloudDiscovery()
 	g.Expect(err).NotTo(gomega.HaveOccurred())
 	g.Expect(configuration.GetSpotPrice()).To(gomega.Equal("0.67"))
 
-	// recommendation should be accepted if nodes are ready
-	status.Conditions = append(status.Conditions, v1alpha1.InstanceGroupCondition{
-		Type:   v1alpha1.NodesReady,
-		Status: corev1.ConditionTrue,
-	})
-
+	asgMock.AutoScalingGroups = mockAsg
 	err = ctx.CloudDiscovery()
 	g.Expect(err).NotTo(gomega.HaveOccurred())
 	g.Expect(configuration.GetSpotPrice()).To(gomega.Equal("0.80"))