Early cordon

Makefile

@@ -14,6 +14,7 @@ all: manager
 # Run tests
 ENVTEST_ASSETS_DIR=$(shell pwd)/testbin
 test: manifests generate fmt vet envtest
+	go test ./controllers/... ./api/...
 	go tool cover -html=./coverage.txt -o cover.html
 
 # Build manager binary

controllers/cloud.go

@@ -27,9 +27,10 @@ import (
 )
 
 var (
-	instanceStateTagKey = "upgrademgr.keikoproj.io/state"
-	inProgressTagValue  = "in-progress"
-	failedDrainTagValue = "failed-drain"
+	instanceStateTagKey   = "upgrademgr.keikoproj.io/state"
+	inProgressTagValue    = "in-progress"
+	failedDrainTagValue   = "failed-drain"
+	earlyCordonedTagValue = "early-cordoned"
 )
 
 type DiscoveredState struct {

controllers/helpers_test.go

@@ -426,3 +426,7 @@ func (mockAutoscalingGroup MockAutoscalingGroup) EnterStandby(_ *autoscaling.Ent
 	output := &autoscaling.EnterStandbyOutput{}
 	return output, nil
 }
+
+func (m *MockEC2) DeleteTags(input *ec2.DeleteTagsInput) (*ec2.DeleteTagsOutput, error) {
+	return &ec2.DeleteTagsOutput{}, nil
+}

controllers/metrics_test.go

@@ -11,8 +11,8 @@ import (
 func TestNodeTurnsOntoStep(t *testing.T) {
 	g := gomega.NewGomegaWithT(t)
 
-	reconsiler := createRollingUpgradeReconciler(t)
-	r := createRollingUpgradeContext(reconsiler)
+	reconciler := createRollingUpgradeReconciler(t)
+	r := createRollingUpgradeContext(reconciler)
 
 	//A map to retain the steps for multiple nodes
 	nodeSteps := make(map[string][]v1alpha1.NodeStepDuration)

controllers/providers/aws/ec2.go

@@ -62,6 +62,31 @@ func (a *AmazonClientSet) DescribeTaggedInstanceIDs(tagKey, tagValue string) ([]
 	return instances, err
 }
 
+func (a *AmazonClientSet) DescribeInstancesWithoutTagValue(tagKey string, tagValue string) ([]string, error) {
+	instances := []string{}
+	input := &ec2.DescribeInstancesInput{}
+	tagAndValueIsPresent := false
+
+	err := a.Ec2Client.DescribeInstancesPages(input, func(page *ec2.DescribeInstancesOutput, lastPage bool) bool {
+		for _, res := range page.Reservations {
+			for _, instance := range res.Instances {
+				for _, t := range instance.Tags {
+					if *t.Key == tagKey && *t.Value == tagValue {
+						tagAndValueIsPresent = true
+						break
+					}
+				}
+				if !tagAndValueIsPresent {
+					instances = append(instances, aws.StringValue(instance.InstanceId))
+				}
+				tagAndValueIsPresent = false
+			}
+		}
+		return page.NextToken != nil
+	})
+	return instances, err
+}
+
 func (a *AmazonClientSet) TagEC2instances(instanceIDs []string, tagKey, tagValue string) error {
 	input := &ec2.CreateTagsInput{
 		Resources: aws.StringSlice(instanceIDs),
@@ -75,3 +100,17 @@ func (a *AmazonClientSet) TagEC2instances(instanceIDs []string, tagKey, tagValue
 	_, err := a.Ec2Client.CreateTags(input)
 	return err
 }
+
+func (a *AmazonClientSet) UntagEC2instances(instanceIDs []string, tagKey, tagValue string) error {
+	input := &ec2.DeleteTagsInput{
+		Resources: aws.StringSlice(instanceIDs),
+		Tags: []*ec2.Tag{
+			{
+				Key:   aws.String(tagKey),
+				Value: aws.String(tagValue),
+			},
+		},
+	}
+	_, err := a.Ec2Client.DeleteTags(input)
+	return err
+}

controllers/providers/kubernetes/nodes.go

@@ -50,6 +50,7 @@ func (k *KubernetesClientSet) DrainNode(node *corev1.Node, PostDrainDelaySeconds
 	}
 
 	helper := &drain.Helper{
+		Ctx:                 context.Background(),
 		Client:              client,
 		Force:               true,
 		GracePeriodSeconds:  -1,
@@ -75,3 +76,33 @@ func (k *KubernetesClientSet) DrainNode(node *corev1.Node, PostDrainDelaySeconds
 	}
 	return nil
 }
+
+// CordonUncordonNode cordons a node.
+func (k *KubernetesClientSet) CordonUncordonNode(node *corev1.Node, client kubernetes.Interface, cordonNode bool) error {
+	if client == nil {
+		return fmt.Errorf("K8sClient not set")
+	}
+
+	if node == nil {
+		return fmt.Errorf("node not set")
+	}
+
+	helper := &drain.Helper{
+		Ctx:                 context.Background(),
+		Client:              client,
+		Force:               true,
+		GracePeriodSeconds:  -1,
+		IgnoreAllDaemonSets: true,
+		Out:                 os.Stdout,
+		ErrOut:              os.Stdout,
+		DeleteEmptyDirData:  true,
+	}
+
+	if err := drain.RunCordonOrUncordon(helper, node, cordonNode); err != nil {
+		if apierrors.IsNotFound(err) {
+			return err
+		}
+		return fmt.Errorf("error cordoning node: %v", err)
+	}
+	return nil
+}

controllers/rollingupgrade_controller.go

@@ -58,6 +58,7 @@
 	IgnoreDrainFailures bool
 	ReplacementNodesMap *sync.Map
 	MaxReplacementNodes int
+	EarlyCordonNodes    bool
 }
 
 // RollingUpgradeAuthenticator has the clients for providers
@@ -212,13 +213,20 @@
 		IgnoreDrainFailures: r.IgnoreDrainFailures,
 		ReplacementNodesMap: r.ReplacementNodesMap,
 		MaxReplacementNodes: r.MaxReplacementNodes,
+		EarlyCordonNodes:    r.EarlyCordonNodes,
 	}
 
 	// process node rotation
 	if err := rollupCtx.RotateNodes(); err != nil {
 		rollingUpgrade.SetCurrentStatus(v1alpha1.StatusError)
 		rollingUpgrade.SetLabel(v1alpha1.LabelKeyRollingUpgradeCurrentStatus, v1alpha1.StatusError)
 		common.SetMetricRollupFailed(rollingUpgrade.Name)
+
+		// try to uncordon all the cordoned nodes.
+		if _, err2 := rollupCtx.CordonUncordonAllNodes(false); err2 != nil {
+			r.Error(err2, "failed to uncordon the nodes.", "name", rollingUpgrade.NamespacedName())
+		}
+
 		return ctrl.Result{}, err
 	}
 

controllers/upgrade.go

@@ -65,6 +65,7 @@
 	ReplacementNodesMap *sync.Map
 	MaxReplacementNodes int
 	AllowReplacements   bool
+	EarlyCordonNodes    bool
 }
 
 func (r *RollingUpgradeContext) RotateNodes() error {
@@ -147,6 +148,13 @@
 		inProcessingNodes = make(map[string]*v1alpha1.NodeInProcessing)
 	}
 
+	//Early-Cordon - Cordon all the nodes to avoid any further scheduling of new pods.
+	if r.EarlyCordonNodes {
+		if ok, err := r.CordonUncordonAllNodes(true); !ok {
+			return ok, err
+		}
+	}
+
 	switch mode {
 	case v1alpha1.UpdateStrategyModeEager:
 		for _, target := range batch {
@@ -750,3 +758,58 @@
 	}
 	return false, batchSize
 }
+
+func (r *RollingUpgradeContext) CordonUncordonAllNodes(cordonNode bool) (bool, error) {
+	scalingGroup := awsprovider.SelectScalingGroup(r.RollingUpgrade.ScalingGroupName(), r.Cloud.ScalingGroups)
+	var instanceIDs []string
+	var err error
+
+	if cordonNode {
+		instanceIDs, err = r.Cloud.AmazonClientSet.DescribeInstancesWithoutTagValue(instanceStateTagKey, earlyCordonedTagValue)
+		if err != nil {
+			r.Error(err, "failed to describe instances for early-cordoning", "name", r.RollingUpgrade.NamespacedName())
+			return false, errors.Wrap(err, "failed to describe instances for early-cordoning")
+		}
+	} else {
+		instanceIDs, err = r.Auth.DescribeTaggedInstanceIDs(instanceStateTagKey, earlyCordonedTagValue)
+		if err != nil {
+			r.Error(err, "failed to discover ec2 instances with early-cordoned tag", "name", r.RollingUpgrade.NamespacedName())
+		}
+
+		r.Info("removing early-cordoning tag while uncordoning instances", "name", r.RollingUpgrade.NamespacedName())
+		if err := r.Auth.UntagEC2instances(instanceIDs, instanceStateTagKey, earlyCordonedTagValue); err != nil {
+			r.Error(err, "failed to delete early-cordoned tag for instances", "name", r.RollingUpgrade.NamespacedName())
+		}
+		// add unit test as well.
+
+	}
+
+	for _, instanceID := range instanceIDs {
+		if instance := awsprovider.SelectScalingGroupInstance(instanceID, scalingGroup); !reflect.DeepEqual(instance, &autoscaling.Instance{}) {
+			//Don't consider if the instance is in terminating state.
+			if !common.ContainsEqualFold(awsprovider.TerminatingInstanceStates, aws.StringValue(instance.LifecycleState)) {
+				node := kubeprovider.SelectNodeByInstanceID(*instance.InstanceId, r.Cloud.ClusterNodes)
+				if node == nil {
+					r.Info("node object not found in clusterNodes, unable to early-cordon node", "instanceID", instance.InstanceId, "name", r.RollingUpgrade.NamespacedName())
+					continue
+				}
+				//Early cordon only the dirfted instances and not the instances that have same scaling-config as the scaling-group
+				if !r.IsInstanceDrifted(instance) {
+					break
+				}
+				r.Info("early cordoning node", "instanceID", instance.InstanceId, "name", r.RollingUpgrade.NamespacedName())
+				if err := r.Auth.CordonUncordonNode(node, r.Auth.Kubernetes, cordonNode); err != nil {
+					r.Error(err, "failed to early cordon the nodes", "instanceID", instance.InstanceId, "name", r.RollingUpgrade.NamespacedName())
+					return false, err
+				}
+				// Set instance-state to early-cordoned tag
+				r.Info("tagging instances with cordoned=true", "instanceID", instance.InstanceId, "name", r.RollingUpgrade.NamespacedName())
+				if err := r.Auth.TagEC2instances([]string{*instance.InstanceId}, instanceStateTagKey, earlyCordonedTagValue); err != nil {
+					r.Error(err, "failed to tag instances with cordoned=true", "instanceID", instance.InstanceId, "name", r.RollingUpgrade.NamespacedName())
+					return true, err
+				}
+			}
+		}
+	}
+	return true, nil
+}

controllers/upgrade_test.go

@@ -1,6 +1,7 @@
 package controllers
 
 import (
+	"context"
 	"os"
 	"testing"
 
@@ -105,6 +106,7 @@ func TestRunCordonOrUncordon(t *testing.T) {
 	for _, test := range tests {
 		rollupCtx := createRollingUpgradeContext(test.Reconciler)
 		helper := &drain.Helper{
+			Ctx:                 context.Background(),
 			Client:              rollupCtx.Auth.Kubernetes,
 			Force:               true,
 			GracePeriodSeconds:  -1,
@@ -299,6 +301,7 @@ func TestRotateNodes(t *testing.T) {
 		rollupCtx := test.RollingUpgradeContext
 		rollupCtx.Cloud.ScalingGroups = test.AsgClient.autoScalingGroups
 		rollupCtx.Auth.AmazonClientSet.AsgClient = test.AsgClient
+		rollupCtx.EarlyCordonNodes = true
 
 		err := rollupCtx.RotateNodes()
 		if err != nil {
@@ -507,6 +510,7 @@ func TestIgnoreDrainFailuresAndDrainTimeout(t *testing.T) {
 		rollupCtx.Cloud.ScalingGroups = test.AsgClient.autoScalingGroups
 		rollupCtx.Cloud.ClusterNodes = test.ClusterNodes
 		rollupCtx.Auth.AmazonClientSet.AsgClient = test.AsgClient
+		rollupCtx.EarlyCordonNodes = true
 
 		err := rollupCtx.RotateNodes()
 		if err != nil {
@@ -545,6 +549,7 @@ func TestClusterBallooning(t *testing.T) {
 				reconciler := createRollingUpgradeReconciler(t)
 				reconciler.MaxReplacementNodes = 500
 				reconciler.ReplacementNodesMap.Store("ReplacementNodes", 500)
+				reconciler.EarlyCordonNodes = true
 				return reconciler
 			}(),
 			createRollingUpgrade(),
@@ -617,3 +622,61 @@ func TestClusterBallooning(t *testing.T) {
 
 	}
 }
+
+func TestCordoningAndUncordoningOfNodes(t *testing.T) {
+	var tests = []struct {
+		TestDescription            string
+		Reconciler                 *RollingUpgradeReconciler
+		Node                       *corev1.Node
+		CordonNodeFlag             bool
+		ExpectedUnschdeulableValue bool
+		ExpectedError              bool
+	}{
+		{
+			"Test if all the nodes are cordoned.",
+			createRollingUpgradeReconciler(t),
+			createNode("mock-node-1"),
+			true,
+			true,
+			false,
+		},
+		{
+			"Test if all the nodes are uncordoned",
+			createRollingUpgradeReconciler(t),
+			createNode("mock-node-1"),
+			false,
+			false,
+			false,
+		},
+		{
+			"Try to cordon an unknown node.",
+			createRollingUpgradeReconciler(t),
+			createNode("mock-node-4"),
+			true,
+			true,
+			true,
+		},
+	}
+	for _, test := range tests {
+		rollupCtx := createRollingUpgradeContext(test.Reconciler)
+
+		if err := rollupCtx.Auth.CordonUncordonNode(test.Node, rollupCtx.Auth.Kubernetes, test.CordonNodeFlag); err != nil && test.ExpectedError {
+			continue
+		}
+
+		// By default, nodes are uncordoned. Therefore, before testing uncordoning the node, first cordon it.
+		if !test.CordonNodeFlag {
+			if err := rollupCtx.Auth.CordonUncordonNode(test.Node, rollupCtx.Auth.Kubernetes, true); err != nil {
+				t.Errorf("Test Description: %s \n error: %v", test.TestDescription, err)
+			}
+		}
+
+		if err := rollupCtx.Auth.CordonUncordonNode(test.Node, rollupCtx.Auth.Kubernetes, test.CordonNodeFlag); err != nil {
+			t.Errorf("Test Description: %s \n error: %v", test.TestDescription, err)
+		}
+
+		if test.ExpectedUnschdeulableValue != test.Node.Spec.Unschedulable {
+			t.Errorf("Test Description: %s \n expectedValue: %v, actualValue: %v", test.TestDescription, test.ExpectedUnschdeulableValue, test.Node.Spec.Unschedulable)
+		}
+	}
+}