Early cordon

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"
+	instanceCordonTagKey = "upgrademgr.keikoproj.io/cordon"
 )
 
 type DiscoveredState struct {

controllers/providers/aws/ec2.go

@@ -62,6 +62,31 @@ func (a *AmazonClientSet) DescribeTaggedInstanceIDs(tagKey, tagValue string) ([]
 	return instances, err
 }
 
+func (a *AmazonClientSet) DescribeInstancesWithoutTag(tagKey string) ([]string, error) {
+	instances := []string{}
+	input := &ec2.DescribeInstancesInput{}
+	tagIsPresent := 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 {
+						tagIsPresent = true
+						break
+					}
+				}
+				if !tagIsPresent {
+					instances = append(instances, aws.StringValue(instance.InstanceId))
+				}
+				tagIsPresent = 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),

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
 }
+
+// CordonNode 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 touncordon 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,51 @@
 	}
 	return false, batchSize
 }
+
+func (r *RollingUpgradeContext) CordonUncordonAllNodes(cordonNode bool) (bool, error) {
+	scalingGroup := awsprovider.SelectScalingGroup(r.RollingUpgrade.ScalingGroupName(), r.Cloud.ScalingGroups)
+	var instances []*autoscaling.Instance
+
+	if cordonNode {
+		instanceIDs, err := r.Cloud.AmazonClientSet.DescribeInstancesWithoutTag(instanceCordonTagKey)
+		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")
+		}
+		for _, instanceID := range instanceIDs {
+			instance := awsprovider.SelectScalingGroupInstance(instanceID, scalingGroup)
+			instances = append(instances, instance)
+		}
+	} else {
+		instances = scalingGroup.Instances
+	}
+
+	for _, instance := range instances {
+		if !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, true); err != nil {
+					r.Error(err, "failed to early cordon the nodes", "instanceID", instance.InstanceId, "name", r.RollingUpgrade.NamespacedName())
+					return false, err
+				}
+			}
+			// Add node-cordoned tag
+			r.Info("tagging instances with cordoned=true", "instanceID", instance.InstanceId, "name", r.RollingUpgrade.NamespacedName())
+			if err := r.Auth.TagEC2instances([]string{*instance.InstanceId}, instanceCordonTagKey, "True"); 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

@@ -299,6 +299,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 +508,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 +547,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 +620,40 @@ func TestClusterBallooning(t *testing.T) {
 
 	}
 }
+
+func TestEarlyCordonFunction(t *testing.T) {
+	var tests = []struct {
+		TestDescription            string
+		Reconciler                 *RollingUpgradeReconciler
+		RollingUpgrade             *v1alpha1.RollingUpgrade
+		AsgClient                  *MockAutoscalingGroup
+		ClusterNodes               []*corev1.Node
+		ExpectedUnschdeulableValue bool
+	}{
+		{
+			"Test if all the nodes are cordoned by default.",
+			createRollingUpgradeReconciler(t),
+			createRollingUpgrade(),
+			createASGClient(),
+			createNodeSlice(),
+			true,
+		},
+	}
+	for _, test := range tests {
+		rollupCtx := createRollingUpgradeContext(test.Reconciler)
+		rollupCtx.RollingUpgrade = test.RollingUpgrade
+		rollupCtx.Cloud.ScalingGroups = test.AsgClient.autoScalingGroups
+		rollupCtx.Cloud.ClusterNodes = test.ClusterNodes
+		rollupCtx.Auth.AmazonClientSet.AsgClient = test.AsgClient
+
+		_, err := rollupCtx.CordonUncordonAllNodes(true)
+		if err != nil {
+			t.Errorf("Test Description: %s \n error: %v", test.TestDescription, err)
+		}
+		for _, node := range rollupCtx.Cloud.ClusterNodes {
+			if test.ExpectedUnschdeulableValue != node.Spec.Unschedulable {
+				t.Errorf("Test Description: %s \n expectedValue: %v, actualValue: %v", test.TestDescription, test.ExpectedUnschdeulableValue, node.Spec.Unschedulable)
+			}
+		}
+	}
+}

main.go

@@ -88,6 +88,7 @@ func main() {
 		drainTimeout         int
 		ignoreDrainFailures  bool
 		maxReplacementNodes  int
+		earlyCordonNodes     bool
 	)
 
 	flag.BoolVar(&debugMode, "debug", false, "enable debug logging")
@@ -102,6 +103,7 @@ func main() {
 	flag.IntVar(&drainTimeout, "drain-timeout", 900, "when the drain command should timeout")
 	flag.BoolVar(&ignoreDrainFailures, "ignore-drain-failures", false, "proceed with instance termination despite drain failures.")
 	flag.IntVar(&maxReplacementNodes, "max-replacement-nodes", 0, "The max number of replacement nodes allowed in a cluster. Avoids cluster-ballooning")
+	flag.BoolVar(&earlyCordonNodes, "early-cordon-nodes", true, "when enabled, will cordon all the nodes in the node-group even before processing the nodes")
 
 	opts := zap.Options{
 		Development: true,
@@ -210,6 +212,7 @@ func main() {
 		DrainTimeout:        drainTimeout,
 		IgnoreDrainFailures: ignoreDrainFailures,
 		ReplacementNodesMap: &sync.Map{},
+		EarlyCordonNodes:    earlyCordonNodes,
 	}
 
 	reconciler.SetMaxParallel(maxParallel)