"max-replacement-nodes" feature to avoid cluster ballooning during upgrades. (keikoproj#328)

* add feature max-replacement-nodes

Signed-off-by: sbadiger <[email protected]>

Author: shreyas-badiger

controllers/helpers_test.go

@@ -51,8 +51,9 @@ func createRollingUpgradeReconciler(t *testing.T) *RollingUpgradeReconciler {
 		ScriptRunner: ScriptRunner{
 			Logger: logger,
 		},
-		DrainGroupMapper: &sync.Map{},
-		DrainErrorMapper: &sync.Map{},
+		DrainGroupMapper:    &sync.Map{},
+		DrainErrorMapper:    &sync.Map{},
+		ReplacementNodesMap: &sync.Map{},
 	}
 	return reconciler
 
@@ -72,8 +73,10 @@ func createRollingUpgradeContext(r *RollingUpgradeReconciler) *RollingUpgradeCon
 			DrainErrors: drainErrs.(chan error),
 			DrainGroup:  drainGroup.(*sync.WaitGroup),
 		},
-		RollingUpgrade: rollingUpgrade,
-		metricsMutex:   &sync.Mutex{},
+		RollingUpgrade:      rollingUpgrade,
+		metricsMutex:        &sync.Mutex{},
+		ReplacementNodesMap: r.ReplacementNodesMap,
+		MaxReplacementNodes: r.MaxReplacementNodes,
 	}
 
 }

controllers/rollingupgrade_controller.go

@@ -55,6 +55,8 @@ type RollingUpgradeReconciler struct {
 	ReconcileMap        *sync.Map
 	DrainTimeout        int
 	IgnoreDrainFailures bool
+	ReplacementNodesMap *sync.Map
+	MaxReplacementNodes int
 }
 
 // RollingUpgradeAuthenticator has the clients for providers
@@ -172,6 +174,8 @@ func (r *RollingUpgradeReconciler) Reconcile(ctx context.Context, req ctrl.Reque
 
 		DrainTimeout:        r.DrainTimeout,
 		IgnoreDrainFailures: r.IgnoreDrainFailures,
+		ReplacementNodesMap: r.ReplacementNodesMap,
+		MaxReplacementNodes: r.MaxReplacementNodes,
 	}
 
 	// process node rotation
@@ -246,6 +250,14 @@ func (r *RollingUpgradeReconciler) UpdateStatus(rollingUpgrade *v1alpha1.Rolling
 	}
 }
 
+// max number of replacement nodes allowed in a cluster. This will ensure we avoid cluster ballooning.
+func (r *RollingUpgradeReconciler) SetMaxReplacementNodes(n int) {
+	if n >= 1 {
+		r.Info("setting max replacement nodes", "value", n)
+		r.MaxReplacementNodes = n
+	}
+}
+
 // extract node objects from syncMap to a slice
 func (r *RollingUpgradeReconciler) getClusterNodes() []*corev1.Node {
 	var clusterNodes []*corev1.Node

controllers/upgrade.go

@@ -62,6 +62,9 @@ type RollingUpgradeContext struct {
 	metricsMutex        *sync.Mutex
 	DrainTimeout        int
 	IgnoreDrainFailures bool
+	ReplacementNodesMap *sync.Map
+	MaxReplacementNodes int
+	AllowReplacements   bool
 }
 
 func (r *RollingUpgradeContext) RotateNodes() error {
@@ -149,6 +152,18 @@ func (r *RollingUpgradeContext) ReplaceNodeBatch(batch []*autoscaling.Instance)
 		batchInstanceIDs, inServiceInstanceIDs := awsprovider.GetInstanceIDs(batch), awsprovider.GetInServiceInstanceIDs(batch)
 		// Tag and set to StandBy only the InService instances.
 		if len(inServiceInstanceIDs) > 0 {
+
+			// Check if replacement nodes are causing cluster to balloon
+			clusterIsBallooning, allowedBatchSize := r.ClusterBallooning(len(inServiceInstanceIDs))
+			if clusterIsBallooning || allowedBatchSize == 0 {
+				// Allowing more replacement nodes can cause cluster ballooning. Requeue CR.
+				return true, nil
+			}
+			if len(inServiceInstanceIDs) != allowedBatchSize {
+				r.Info("cluster is about to hit max-replacement-nodes capacity, reducing batchSize", "prevBatchSize", len(inServiceInstanceIDs), "currBatchSize", allowedBatchSize, "name", r.RollingUpgrade.NamespacedName())
+				inServiceInstanceIDs = inServiceInstanceIDs[:allowedBatchSize]
+			}
+
 			// Add in-progress tag
 			r.Info("setting instances to in-progress", "batch", batchInstanceIDs, "instances(InService)", inServiceInstanceIDs, "name", r.RollingUpgrade.NamespacedName())
 			if err := r.Auth.TagEC2instances(inServiceInstanceIDs, instanceStateTagKey, inProgressTagValue); err != nil {
@@ -361,6 +376,14 @@ func (r *RollingUpgradeContext) ReplaceNodeBatch(batch []*autoscaling.Instance)
 				return true, nil
 			}
 
+			// Once instances are terminated, decrease them from the count.
+			count, _ := r.ReplacementNodesMap.Load("ReplacementNodes")
+			if count != nil && count.(int) > 0 {
+				r.ReplacementNodesMap.Store("ReplacementNodes", count.(int)-1)
+				r.Info("decrementing replacementNodes count", "ReplacementNodes", count.(int)-1, "name", r.RollingUpgrade.NamespacedName())
+				r.AllowReplacements = false
+			}
+
 			r.RollingUpgrade.SetLastNodeTerminationTime(&metav1.Time{Time: time.Now()})
 
 			// Turns onto NodeRotationTerminate
@@ -662,3 +685,31 @@ func (r *RollingUpgradeContext) SetBatchStandBy(instanceIDs []string) error {
 	}
 	return nil
 }
+
+// Checks for how many replacement nodes exists across all the IGs in the cluster
+func (r *RollingUpgradeContext) ClusterBallooning(batchSize int) (bool, int) {
+	count, _ := r.ReplacementNodesMap.LoadOrStore("ReplacementNodes", 0)
+	newReplacementCount := count.(int) + batchSize
+	partialReplacementCount := r.MaxReplacementNodes - count.(int)
+
+	// By default, no limits on replacement nodes.
+	if r.MaxReplacementNodes == 0 {
+		return false, batchSize
+	}
+
+	// Handle 3 different cases. 1) When entire batch can have replacement nodes. 2) When partial batch can have replacement nodes 3) When there is no availability for replacement nodes and CR has to re-queue
+	if newReplacementCount <= r.MaxReplacementNodes {
+		r.ReplacementNodesMap.Store("ReplacementNodes", newReplacementCount)
+		r.Info("incrementing replacementNodes count", "ReplacementNodes", newReplacementCount, "name", r.RollingUpgrade.NamespacedName())
+		r.AllowReplacements = true
+	} else if partialReplacementCount < batchSize && partialReplacementCount > 0 {
+		r.ReplacementNodesMap.Store("ReplacementNodes", count.(int)+partialReplacementCount)
+		r.Info("incrementing replacementNodes count", "ReplacementNodes", count.(int)+partialReplacementCount, "name", r.RollingUpgrade.NamespacedName())
+		r.AllowReplacements = true
+		batchSize = partialReplacementCount
+	} else if !r.AllowReplacements {
+		r.Info("cluster has hit max-replacement-nodes capacity, requeuing rollingUpgrade CR. ", "replacementNodes", count.(int), "maxReplacementNodes", r.MaxReplacementNodes, "name", r.RollingUpgrade.NamespacedName())
+		return true, 0
+	}
+	return false, batchSize
+}

controllers/upgrade_test.go

@@ -517,3 +517,103 @@ func TestIgnoreDrainFailuresAndDrainTimeout(t *testing.T) {
 		}
 	}
 }
+
+func TestClusterBallooning(t *testing.T) {
+	var tests = []struct {
+		TestDescription     string
+		Reconciler          *RollingUpgradeReconciler
+		RollingUpgrade      *v1alpha1.RollingUpgrade
+		AsgClient           *MockAutoscalingGroup
+		ClusterNodes        []*corev1.Node
+		BatchSize           int
+		IsClusterBallooning bool
+		AllowedBatchSize    int
+	}{
+		{
+			"ClusterBallooning - maxReplacementNodes is not set, expect no clusterBallooning",
+			createRollingUpgradeReconciler(t),
+			createRollingUpgrade(),
+			createASGClient(),
+			createNodeSlice(),
+			3,
+			false,
+			3, // because mock-asg-1 has 3 instances
+		},
+		{
+			"ClusterBallooning - cluster has hit maxReplacementNodes capacity, expect clusterBallooning to be true",
+			func() *RollingUpgradeReconciler {
+				reconciler := createRollingUpgradeReconciler(t)
+				reconciler.MaxReplacementNodes = 500
+				reconciler.ReplacementNodesMap.Store("ReplacementNodes", 500)
+				return reconciler
+			}(),
+			createRollingUpgrade(),
+			createASGClient(),
+			createNodeSlice(),
+			3,
+			true,
+			0,
+		},
+		{
+			"ClusterBallooning - cluster is below maxReplacementNodes capacity, expect no clusterBallooning",
+			func() *RollingUpgradeReconciler {
+				reconciler := createRollingUpgradeReconciler(t)
+				reconciler.MaxReplacementNodes = 500
+				reconciler.ReplacementNodesMap.Store("ReplacementNodes", 100)
+				return reconciler
+			}(),
+			createRollingUpgrade(),
+			createASGClient(),
+			createNodeSlice(),
+			400,
+			false,
+			400,
+		},
+		{
+			"ClusterBallooning - cluster is about to hit maxReplacementNodes capacity, expect reduced batchSize",
+			func() *RollingUpgradeReconciler {
+				reconciler := createRollingUpgradeReconciler(t)
+				reconciler.MaxReplacementNodes = 100
+				reconciler.ReplacementNodesMap.Store("ReplacementNodes", 97)
+				return reconciler
+			}(),
+			createRollingUpgrade(),
+			createASGClient(),
+			createNodeSlice(),
+			100,
+			false,
+			3,
+		},
+		{
+			"ClusterBallooning - cluster is about to hit maxReplacementNodes capacity, expect reduced batchSize",
+			func() *RollingUpgradeReconciler {
+				reconciler := createRollingUpgradeReconciler(t)
+				reconciler.MaxReplacementNodes = 5
+				reconciler.ReplacementNodesMap.Store("ReplacementNodes", 3)
+				return reconciler
+			}(),
+			createRollingUpgrade(),
+			createASGClient(),
+			createNodeSlice(),
+			3,
+			false,
+			2,
+		},
+	}
+	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
+
+		isClusterBallooning, allowedBatchSize := rollupCtx.ClusterBallooning(test.BatchSize)
+		if isClusterBallooning != test.IsClusterBallooning {
+			t.Errorf("Test Description: %s \n isClusterBallooning, expected: %v \n actual: %v", test.TestDescription, test.IsClusterBallooning, isClusterBallooning)
+		}
+		if allowedBatchSize != test.AllowedBatchSize {
+			t.Errorf("Test Description: %s \n allowedBatchSize expected: %v \n actual: %v", test.TestDescription, test.AllowedBatchSize, allowedBatchSize)
+		}
+
+	}
+}

go.mod

@@ -16,5 +16,5 @@ require (
 	k8s.io/client-go v0.20.4
 	k8s.io/kubectl v0.20.4
 	sigs.k8s.io/controller-runtime v0.7.0
-	
+
 )