feat: consolidation enhancements (#2488)

Author: tzneal

charts/karpenter/templates/clusterrole.yaml

@@ -23,7 +23,7 @@ rules:
     resources: ["storageclasses", "csinodes"]
     verbs: ["get", "watch", "list"]
   - apiGroups: ["apps"]
-    resources: ["daemonsets", "replicasets", "statefulsets"]
+    resources: ["daemonsets", "deployments", "replicasets", "statefulsets"]
     verbs: ["list", "watch"]
   - apiGroups: ["admissionregistration.k8s.io"]
     resources: ["validatingwebhookconfigurations", "mutatingwebhookconfigurations"]

pkg/cloudprovider/aws/amifamily/ami.go

@@ -90,7 +90,7 @@ func (p *AMIProvider) getDefaultAMIFromSSM(ctx context.Context, _ cloudprovider.
 	}
 	ami := aws.StringValue(output.Parameter.Value)
 	p.ssmCache.SetDefault(ssmQuery, ami)
-	if p.cm.HasChanged("ssmquery", ami+ssmQuery) {
+	if p.cm.HasChanged("ssmquery-"+ssmQuery, ami) {
 		logging.FromContext(ctx).Debugf("Discovered %s for query %q", ami, ssmQuery)
 	}
 	return ami, nil

pkg/cloudprovider/aws/launchtemplate.go

@@ -160,7 +160,9 @@ func (p *LaunchTemplateProvider) ensureLaunchTemplate(ctx context.Context, optio
 	} else if len(output.LaunchTemplates) != 1 {
 		return nil, fmt.Errorf("expected to find one launch template, but found %d", len(output.LaunchTemplates))
 	} else {
-		logging.FromContext(ctx).Debugf("Discovered launch template %s", name)
+		if p.cm.HasChanged("launchtemplate-"+name, name) {
+			logging.FromContext(ctx).Debugf("Discovered launch template %s", name)
+		}
 		launchTemplate = output.LaunchTemplates[0]
 	}
 	p.cache.SetDefault(name, launchTemplate)

pkg/controllers/consolidation/controller.go

@@ -28,7 +28,6 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
-
 	"k8s.io/utils/clock"
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/ptr"
@@ -109,12 +108,13 @@ func (c *Controller) run(ctx context.Context) {
 				continue
 			}
 
-			// don't consolidate as we recently scaled down too soon
+			// don't consolidate as we recently scaled up/down too soon
 			stabilizationTime := c.clock.Now().Add(-c.stabilizationWindow(ctx))
 			// capture the state before we process so if something changes during consolidation we'll re-look
 			// immediately
 			clusterState := c.cluster.ClusterConsolidationState()
-			if c.cluster.LastNodeDeletionTime().Before(stabilizationTime) {
+			if c.cluster.LastNodeDeletionTime().Before(stabilizationTime) &&
+				c.cluster.LastNodeCreationTime().Before(stabilizationTime) {
 				result, err := c.ProcessCluster(ctx)
 				if err != nil {
 					logging.FromContext(ctx).Errorf("consolidating cluster, %s", err)
@@ -480,10 +480,12 @@ func (c *Controller) nodeConsolidationOptionReplaceOrDelete(ctx context.Context,
 		if node.Name == n.Node.Name && n.MarkedForDeletion {
 			candidateNodeIsDeleting = true
 		}
-		if !n.MarkedForDeletion {
-			stateNodes = append(stateNodes, n.DeepCopy())
-		} else {
-			markedForDeletionNodes = append(markedForDeletionNodes, n.DeepCopy())
+		if n.Node.Name != node.Name {
+			if !n.MarkedForDeletion {
+				stateNodes = append(stateNodes, n.DeepCopy())
+			} else {
+				markedForDeletionNodes = append(markedForDeletionNodes, n.DeepCopy())
+			}
 		}
 		return true
 	})
@@ -502,7 +504,6 @@ func (c *Controller) nodeConsolidationOptionReplaceOrDelete(ctx context.Context,
 	pods := append(node.pods, deletingNodePods...)
 	scheduler, err := c.provisioner.NewScheduler(ctx, pods, stateNodes, scheduling.SchedulerOptions{
 		SimulationMode: true,
-		ExcludeNodes:   []string{node.Name},
 	})
 
 	if err != nil {
@@ -576,11 +577,30 @@ func (c *Controller) hasPendingPods(ctx context.Context) bool {
 	return false
 }
 
+func (c *Controller) deploymentsReady(ctx context.Context) bool {
+	var depList appsv1.DeploymentList
+	if err := c.kubeClient.List(ctx, &depList); err != nil {
+		// failed to list, assume there must be one non-ready as it's harmless and just ensures we wait longer
+		return false
+	}
+	for _, ds := range depList.Items {
+		desired := ptr.Int32Value(ds.Spec.Replicas)
+		if ds.Spec.Replicas == nil {
+			// unspecified defaults to 1
+			desired = 1
+		}
+		if ds.Status.ReadyReplicas < desired || ds.Status.UpdatedReplicas < desired {
+			return false
+		}
+	}
+	return true
+}
+
 func (c *Controller) replicaSetsReady(ctx context.Context) bool {
 	var rsList appsv1.ReplicaSetList
 	if err := c.kubeClient.List(ctx, &rsList); err != nil {
 		// failed to list, assume there must be one non-ready as it's harmless and just ensures we wait longer
-		return true
+		return false
 	}
 	for _, rs := range rsList.Items {
 		desired := ptr.Int32Value(rs.Spec.Replicas)
@@ -599,7 +619,7 @@ func (c *Controller) replicationControllersReady(ctx context.Context) bool {
 	var rsList v1.ReplicationControllerList
 	if err := c.kubeClient.List(ctx, &rsList); err != nil {
 		// failed to list, assume there must be one non-ready as it's harmless and just ensures we wait longer
-		return true
+		return false
 	}
 	for _, rs := range rsList.Items {
 		desired := ptr.Int32Value(rs.Spec.Replicas)
@@ -618,15 +638,15 @@ func (c *Controller) statefulSetsReady(ctx context.Context) bool {
 	var sslist appsv1.StatefulSetList
 	if err := c.kubeClient.List(ctx, &sslist); err != nil {
 		// failed to list, assume there must be one non-ready as it's harmless and just ensures we wait longer
-		return true
+		return false
 	}
 	for _, rs := range sslist.Items {
 		desired := ptr.Int32Value(rs.Spec.Replicas)
 		if rs.Spec.Replicas == nil {
 			// unspecified defaults to 1
 			desired = 1
 		}
-		if rs.Status.ReadyReplicas < desired {
+		if rs.Status.ReadyReplicas < desired || rs.Status.UpdatedReplicas < desired {
 			return false
 		}
 	}
@@ -635,7 +655,7 @@ func (c *Controller) statefulSetsReady(ctx context.Context) bool {
 
 func (c *Controller) stabilizationWindow(ctx context.Context) time.Duration {
 	// no pending pods, and all replica sets/replication controllers are reporting ready so quickly consider another consolidation
-	if !c.hasPendingPods(ctx) && c.replicaSetsReady(ctx) &&
+	if !c.hasPendingPods(ctx) && c.deploymentsReady(ctx) && c.replicaSetsReady(ctx) &&
 		c.replicationControllersReady(ctx) && c.statefulSetsReady(ctx) {
 		return 0
 	}

pkg/controllers/provisioning/scheduling/scheduler.go

@@ -22,7 +22,6 @@ import (
 	"github.com/samber/lo"
 	"go.uber.org/multierr"
 	v1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/util/sets"
 	"knative.dev/pkg/logging"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
@@ -38,8 +37,6 @@ import (
 type SchedulerOptions struct {
 	// SimulationMode if true will prevent recording of the pod nomination decisions as events
 	SimulationMode bool
-	// ExcludeNodes are a list of node names that are excluded from existingNodes nodes for scheduling purposes.
-	ExcludeNodes []string
 }
 
 func NewScheduler(ctx context.Context, kubeClient client.Client, nodeTemplates []*scheduling.NodeTemplate,
@@ -82,7 +79,7 @@ func NewScheduler(ctx context.Context, kubeClient client.Client, nodeTemplates [
 		}
 	}
 
-	s.calculateExistingNodes(opts, namedNodeTemplates, stateNodes)
+	s.calculateExistingNodes(namedNodeTemplates, stateNodes)
 	return s
 }
 
@@ -228,14 +225,9 @@ func (s *Scheduler) add(ctx context.Context, pod *v1.Pod) error {
 	return errs
 }
 
-func (s *Scheduler) calculateExistingNodes(opts SchedulerOptions, namedNodeTemplates map[string]*scheduling.NodeTemplate, stateNodes []*state.Node) {
-	// create our in-flight nodes
-	excluded := sets.NewString(opts.ExcludeNodes...)
+func (s *Scheduler) calculateExistingNodes(namedNodeTemplates map[string]*scheduling.NodeTemplate, stateNodes []*state.Node) {
+	// create our existing nodes
 	for _, node := range stateNodes {
-		// skip any nodes that have been excluded
-		if excluded.Has(node.Node.Name) {
-			continue
-		}
 		name, ok := node.Node.Labels[v1alpha5.ProvisionerNameLabelKey]
 		if !ok {
 			// ignoring this node as it wasn't launched by us

pkg/controllers/state/cluster.go

@@ -64,6 +64,7 @@ type Cluster struct {
 	// it previously couldn't occur.
 	consolidationState   int64
 	lastNodeDeletionTime int64
+	lastNodeCreationTime int64
 }
 
 func NewCluster(clk clock.Clock, cfg config.Config, client client.Client, cp cloudprovider.CloudProvider) *Cluster {
@@ -358,6 +359,10 @@ func (c *Cluster) updateNode(ctx context.Context, node *v1.Node) error {
 			atomic.StoreInt64(&c.lastNodeDeletionTime, nodeDeletionTime)
 		}
 	}
+	nodeCreationTime := node.CreationTimestamp.UnixMilli()
+	if nodeCreationTime > atomic.LoadInt64(&c.lastNodeCreationTime) {
+		atomic.StoreInt64(&c.lastNodeCreationTime, nodeCreationTime)
+	}
 	return nil
 }
 
@@ -381,6 +386,11 @@ func (c *Cluster) LastNodeDeletionTime() time.Time {
 	return time.UnixMilli(atomic.LoadInt64(&c.lastNodeDeletionTime))
 }
 
+// LastNodeCreationTime returns the last time that at a node was created.
+func (c *Cluster) LastNodeCreationTime() time.Time {
+	return time.UnixMilli(atomic.LoadInt64(&c.lastNodeCreationTime))
+}
+
 // deletePod is called when the pod has been deleted
 func (c *Cluster) deletePod(podKey types.NamespacedName) {
 	c.antiAffinityPods.Delete(podKey)

website/content/en/preview/faq.md

@@ -190,3 +190,9 @@ error: error validating "provisioner.yaml": error validating data: ValidationErr
 ```
 
 The `startupTaints` parameter was added in v0.10.0.  Helm upgrades do not upgrade the CRD describing the provisioner, so it must be done manually. For specific details, see the [Upgrade Guide]({{< ref "./upgrade-guide/#upgrading-to-v0100" >}})
+
+## Consolidation
+
+### Why do I sometimes see an extra node get launched when updating a deployment that remains empty and is later removed?
+
+Consolidation packs pods tightly onto nodes which can leave little free allocatable CPU/memory on your nodes.  If a deployment uses a deployment strategy with a non-zero `maxSurge`, such as the default 25%, those surge pods may not have anywhere to run. In this case, Karpenter will launch a new node so that the surge pods can run and then remove it soon after if it's not needed.