fix: Mark nodes that are pending deletion in cluster state for scheduling (#2477)

Author: jonathan-innis

pkg/controllers/consolidation/controller.go

@@ -42,6 +42,7 @@ import (
 	"github.com/aws/karpenter/pkg/controllers/state"
 	"github.com/aws/karpenter/pkg/events"
 	"github.com/aws/karpenter/pkg/metrics"
+	nodeutils "github.com/aws/karpenter/pkg/utils/node"
 	"github.com/aws/karpenter/pkg/utils/pod"
 )
 
@@ -202,6 +203,10 @@ func (c *Controller) candidateNodes(ctx context.Context) ([]candidateNode, error
 			provisioner = provisioners[provName]
 			instanceTypeMap = instanceTypesByProvisioner[provName]
 		}
+		// skip any nodes that are already marked for deletion and being handled
+		if n.MarkedForDeletion {
+			return true
+		}
 		// skip any nodes where we can't determine the provisioner
 		if provisioner == nil || instanceTypeMap == nil {
 			return true
@@ -242,7 +247,7 @@ func (c *Controller) candidateNodes(ctx context.Context) ([]candidateNode, error
 			return true
 		}
 
-		pods, err := c.getNodePods(ctx, n.Node.Name)
+		pods, err := nodeutils.GetNodePods(ctx, c.kubeClient, n.Node)
 		if err != nil {
 			logging.FromContext(ctx).Errorf("Determining node pods, %s", err)
 			return true
@@ -367,21 +372,22 @@ func byNodeDisruptionCost(nodes []candidateNode) func(i int, j int) bool {
 }
 
 // launchReplacementNode launches a replacement node and blocks until it is ready
-func (c *Controller) launchReplacementNode(ctx context.Context, minCost consolidationAction) error {
+func (c *Controller) launchReplacementNode(ctx context.Context, action consolidationAction) error {
 	defer metrics.Measure(consolidationReplacementNodeInitializedHistogram)()
-	if len(minCost.oldNodes) != 1 {
-		return fmt.Errorf("expected a single node to replace, found %d", len(minCost.oldNodes))
+	if len(action.oldNodes) != 1 {
+		return fmt.Errorf("expected a single node to replace, found %d", len(action.oldNodes))
 	}
+	oldNode := action.oldNodes[0]
 
 	// cordon the node before we launch the replacement to prevent new pods from scheduling to the node
-	if err := c.setNodeUnschedulable(ctx, minCost.oldNodes[0].Name, true); err != nil {
-		return fmt.Errorf("cordoning node %s, %w", minCost.oldNodes[0].Name, err)
+	if err := c.setNodeUnschedulable(ctx, action.oldNodes[0].Name, true); err != nil {
+		return fmt.Errorf("cordoning node %s, %w", oldNode.Name, err)
 	}
 
-	nodeNames, err := c.provisioner.LaunchNodes(ctx, provisioning.LaunchOptions{RecordPodNomination: false}, minCost.replacementNode)
+	nodeNames, err := c.provisioner.LaunchNodes(ctx, provisioning.LaunchOptions{RecordPodNomination: false}, action.replacementNode)
 	if err != nil {
 		// uncordon the node as the launch may fail (e.g. ICE or incompatible AMI)
-		err = multierr.Append(err, c.setNodeUnschedulable(ctx, minCost.oldNodes[0].Name, false))
+		err = multierr.Append(err, c.setNodeUnschedulable(ctx, oldNode.Name, false))
 		return err
 	}
 	if len(nodeNames) != 1 {
@@ -391,6 +397,9 @@ func (c *Controller) launchReplacementNode(ctx context.Context, minCost consolid
 
 	consolidationNodesCreatedCounter.Add(1)
 
+	// We have the new node created at the API server so mark the old node for deletion
+	c.cluster.MarkForDeletion(oldNode.Name)
+
 	var k8Node v1.Node
 	// Wait for the node to be ready
 	var once sync.Once
@@ -399,7 +408,7 @@ func (c *Controller) launchReplacementNode(ctx context.Context, minCost consolid
 			return fmt.Errorf("getting node, %w", err)
 		}
 		once.Do(func() {
-			c.recorder.LaunchingNodeForConsolidation(&k8Node, minCost.String())
+			c.recorder.LaunchingNodeForConsolidation(&k8Node, action.String())
 		})
 
 		if _, ok := k8Node.Labels[v1alpha5.LabelNodeInitialized]; !ok {
@@ -410,30 +419,13 @@ func (c *Controller) launchReplacementNode(ctx context.Context, minCost consolid
 		return nil
 	}, waitRetryOptions...); err != nil {
 		// node never become ready, so uncordon the node we were trying to delete and report the error
-		return multierr.Combine(c.setNodeUnschedulable(ctx, minCost.oldNodes[0].Name, false),
+		c.cluster.UnmarkForDeletion(oldNode.Name)
+		return multierr.Combine(c.setNodeUnschedulable(ctx, oldNode.Name, false),
 			fmt.Errorf("timed out checking node readiness, %w", err))
 	}
 	return nil
 }
 
-func (c *Controller) getNodePods(ctx context.Context, nodeName string) ([]*v1.Pod, error) {
-	var podList v1.PodList
-	if err := c.kubeClient.List(ctx, &podList, client.MatchingFields{"spec.nodeName": nodeName}); err != nil {
-		return nil, fmt.Errorf("listing pods, %w", err)
-	}
-	var pods []*v1.Pod
-	for i := range podList.Items {
-		// these pods don't need to be rescheduled
-		if pod.IsOwnedByNode(&podList.Items[i]) ||
-			pod.IsOwnedByDaemonSet(&podList.Items[i]) ||
-			pod.IsTerminal(&podList.Items[i]) {
-			continue
-		}
-		pods = append(pods, &podList.Items[i])
-	}
-	return pods, nil
-}
-
 func (c *Controller) canBeTerminated(node candidateNode, pdbs *PDBLimits) error {
 	if !node.DeletionTimestamp.IsZero() {
 		return fmt.Errorf("already being deleted")
@@ -481,11 +473,34 @@ func (c *Controller) nodeConsolidationOptionReplaceOrDelete(ctx context.Context,
 	defer metrics.Measure(consolidationDurationHistogram.WithLabelValues("Replace/Delete"))()
 
 	var stateNodes []*state.Node
+	var markedForDeletionNodes []*state.Node
+	candidateNodeIsDeleting := false
+
 	c.cluster.ForEachNode(func(n *state.Node) bool {
-		stateNodes = append(stateNodes, n.DeepCopy())
+		if node.Name == n.Node.Name && n.MarkedForDeletion {
+			candidateNodeIsDeleting = true
+		}
+		if !n.MarkedForDeletion {
+			stateNodes = append(stateNodes, n.DeepCopy())
+		} else {
+			markedForDeletionNodes = append(markedForDeletionNodes, n.DeepCopy())
+		}
 		return true
 	})
-	scheduler, err := c.provisioner.NewScheduler(ctx, node.pods, stateNodes, scheduling.SchedulerOptions{
+	// We do one final check to ensure that the node that we are attempting to consolidate isn't
+	// already handled for deletion by some other controller. This could happen if the node was markedForDeletion
+	// between returning the candidateNodes and getting the stateNodes above
+	if candidateNodeIsDeleting {
+		return consolidationAction{result: consolidateResultNoAction}, nil
+	}
+
+	// We get the pods that are on nodes that are deleting
+	deletingNodePods, err := nodeutils.GetNodePods(ctx, c.kubeClient, lo.Map(markedForDeletionNodes, func(n *state.Node, _ int) *v1.Node { return n.Node })...)
+	if err != nil {
+		return consolidationAction{result: consolidateResultUnknown}, fmt.Errorf("failed to get pods from deleting nodes, %w", err)
+	}
+	pods := append(node.pods, deletingNodePods...)
+	scheduler, err := c.provisioner.NewScheduler(ctx, pods, stateNodes, scheduling.SchedulerOptions{
 		SimulationMode: true,
 		ExcludeNodes:   []string{node.Name},
 	})
@@ -494,7 +509,7 @@ func (c *Controller) nodeConsolidationOptionReplaceOrDelete(ctx context.Context,
 		return consolidationAction{result: consolidateResultUnknown}, fmt.Errorf("creating scheduler, %w", err)
 	}
 
-	newNodes, inflightNodes, err := scheduler.Solve(ctx, node.pods)
+	newNodes, inflightNodes, err := scheduler.Solve(ctx, pods)
 	if err != nil {
 		return consolidationAction{result: consolidateResultUnknown}, fmt.Errorf("simulating scheduling, %w", err)
 	}

pkg/controllers/consolidation/suite_test.go

@@ -53,6 +53,7 @@ var ctx context.Context
 var env *test.Environment
 var cluster *state.Cluster
 var controller *consolidation.Controller
+var provisioningController *provisioning.Controller
 var provisioner *provisioning.Provisioner
 var cloudProvider *fake.CloudProvider
 var clientSet *kubernetes.Clientset
@@ -82,6 +83,7 @@ var _ = BeforeSuite(func() {
 		clientSet = kubernetes.NewForConfigOrDie(e.Config)
 		recorder = test.NewEventRecorder()
 		provisioner = provisioning.NewProvisioner(ctx, cfg, env.Client, clientSet.CoreV1(), recorder, cloudProvider, cluster)
+		provisioningController = provisioning.NewController(env.Client, provisioner, recorder)
 	})
 	Expect(env.Start()).To(Succeed(), "Failed to start environment")
 })
@@ -1307,6 +1309,158 @@ var _ = Describe("Empty Nodes", func() {
 	})
 })
 
+var _ = Describe("Parallelization", func() {
+	It("should schedule an additional node when receiving pending pods while consolidating", func() {
+		labels := map[string]string{
+			"app": "test",
+		}
+		// create our RS so we can link a pod to it
+		rs := test.ReplicaSet()
+		ExpectApplied(ctx, env.Client, rs)
+		Expect(env.Client.Get(ctx, client.ObjectKeyFromObject(rs), rs)).To(Succeed())
+
+		pod := test.Pod(test.PodOptions{
+			ObjectMeta: metav1.ObjectMeta{Labels: labels,
+				OwnerReferences: []metav1.OwnerReference{
+					{
+						APIVersion:         "apps/v1",
+						Kind:               "ReplicaSet",
+						Name:               rs.Name,
+						UID:                rs.UID,
+						Controller:         aws.Bool(true),
+						BlockOwnerDeletion: aws.Bool(true),
+					},
+				}}})
+
+		prov := test.Provisioner(test.ProvisionerOptions{Consolidation: &v1alpha5.Consolidation{Enabled: aws.Bool(true)}})
+
+		// Add a finalizer to the node so that it sticks around for the scheduling loop
+		node := test.Node(test.NodeOptions{
+			ObjectMeta: metav1.ObjectMeta{
+				Labels: map[string]string{
+					v1alpha5.ProvisionerNameLabelKey: prov.Name,
+					v1.LabelInstanceTypeStable:       mostExpensiveInstance.Name(),
+					v1alpha5.LabelCapacityType:       mostExpensiveOffering.CapacityType,
+					v1.LabelTopologyZone:             mostExpensiveOffering.Zone,
+				},
+				Finalizers: []string{"karpenter.sh/test-finalizer"},
+			},
+			Allocatable: map[v1.ResourceName]resource.Quantity{v1.ResourceCPU: resource.MustParse("32")}})
+
+		ExpectApplied(ctx, env.Client, rs, pod, node, prov)
+		ExpectMakeNodesReady(ctx, env.Client, node)
+		ExpectReconcileSucceeded(ctx, nodeStateController, client.ObjectKeyFromObject(node))
+		ExpectManualBinding(ctx, env.Client, pod, node)
+		ExpectScheduled(ctx, env.Client, pod)
+		Expect(env.Client.Get(ctx, client.ObjectKeyFromObject(node), node)).To(Succeed())
+
+		fakeClock.Step(10 * time.Minute)
+
+		// Run the processing loop in parallel in the background with environment context
+		go func() {
+			_, err := controller.ProcessCluster(env.Ctx)
+			Expect(err).ToNot(HaveOccurred())
+		}()
+
+		Eventually(func(g Gomega) {
+			// should create a new node as there is a cheaper one that can hold the pod
+			nodes := &v1.NodeList{}
+			g.Expect(env.Client.List(ctx, nodes)).To(Succeed())
+			g.Expect(len(nodes.Items)).To(Equal(2))
+		}).Should(Succeed())
+
+		// Add a new pending pod that should schedule while node is not yet deleted
+		pods := ExpectProvisionedNoBinding(ctx, env.Client, provisioningController, test.UnschedulablePod())
+		nodes := &v1.NodeList{}
+		Expect(env.Client.List(ctx, nodes)).To(Succeed())
+		Expect(len(nodes.Items)).To(Equal(3))
+		Expect(pods[0].Spec.NodeName).NotTo(Equal(node.Name))
+	})
+	It("should not consolidate a node that is launched for pods on a deleting node", func() {
+		labels := map[string]string{
+			"app": "test",
+		}
+		// create our RS so we can link a pod to it
+		rs := test.ReplicaSet()
+		ExpectApplied(ctx, env.Client, rs)
+		Expect(env.Client.Get(ctx, client.ObjectKeyFromObject(rs), rs)).To(Succeed())
+
+		prov := test.Provisioner(test.ProvisionerOptions{Consolidation: &v1alpha5.Consolidation{Enabled: aws.Bool(true)}})
+		podOpts := test.PodOptions{
+			ObjectMeta: metav1.ObjectMeta{
+				Labels: labels,
+				OwnerReferences: []metav1.OwnerReference{
+					{
+						APIVersion:         "apps/v1",
+						Kind:               "ReplicaSet",
+						Name:               rs.Name,
+						UID:                rs.UID,
+						Controller:         aws.Bool(true),
+						BlockOwnerDeletion: aws.Bool(true),
+					},
+				},
+			},
+			ResourceRequirements: v1.ResourceRequirements{
+				Requests: v1.ResourceList{
+					v1.ResourceCPU: resource.MustParse("1"),
+				},
+			},
+		}
+
+		var pods []*v1.Pod
+		for i := 0; i < 5; i++ {
+			pod := test.UnschedulablePod(podOpts)
+			pods = append(pods, pod)
+		}
+		ExpectApplied(ctx, env.Client, rs, prov)
+		ExpectProvisioned(ctx, env.Client, provisioningController, pods...)
+
+		nodeList := &v1.NodeList{}
+		Expect(env.Client.List(ctx, nodeList)).To(Succeed())
+		Expect(len(nodeList.Items)).To(Equal(1))
+
+		// Update cluster state with new node
+		ExpectReconcileSucceeded(ctx, nodeStateController, client.ObjectKeyFromObject(&nodeList.Items[0]))
+
+		// Reset the bindings so we can re-record bindings
+		recorder.ResetBindings()
+
+		// Mark the node for deletion and re-trigger reconciliation
+		oldNodeName := nodeList.Items[0].Name
+		cluster.MarkForDeletion(nodeList.Items[0].Name)
+		ExpectProvisionedNoBinding(ctx, env.Client, provisioningController)
+
+		// Make sure that the cluster state is aware of the current node state
+		Expect(env.Client.List(ctx, nodeList)).To(Succeed())
+		Expect(len(nodeList.Items)).To(Equal(2))
+		newNode, _ := lo.Find(nodeList.Items, func(n v1.Node) bool { return n.Name != oldNodeName })
+
+		for i := range nodeList.Items {
+			node := nodeList.Items[i]
+			ExpectMakeNodesReady(ctx, env.Client, &node)
+			ExpectReconcileSucceeded(ctx, nodeStateController, client.ObjectKeyFromObject(&node))
+		}
+
+		// Wait for the nomination cache to expire
+		time.Sleep(time.Second * 11)
+
+		// Re-create the pods to re-bind them
+		for i := 0; i < 2; i++ {
+			ExpectDeleted(ctx, env.Client, pods[i])
+			pod := test.UnschedulablePod(podOpts)
+			ExpectApplied(ctx, env.Client, pod)
+			ExpectManualBinding(ctx, env.Client, pod, &newNode)
+		}
+
+		// Trigger a reconciliation run which should take into account the deleting node
+		// Consolidation shouldn't trigger additional actions
+		fakeClock.Step(10 * time.Minute)
+		result, err := controller.ProcessCluster(env.Ctx)
+		Expect(err).ToNot(HaveOccurred())
+		Expect(result).To(Equal(consolidation.ProcessResultNothingToDo))
+	})
+})
+
 func leastExpensiveInstanceWithZone(zone string) cloudprovider.InstanceType {
 	for _, elem := range onDemandInstances {
 		if hasZone(elem.Offerings(), zone) {

pkg/controllers/consolidation/types.go

@@ -41,6 +41,7 @@ const (
 	consolidateResultDelete
 	consolidateResultDeleteEmpty
 	consolidateResultReplace
+	consolidateResultNoAction
 )
 
 func (r consolidateResult) String() string {
@@ -55,6 +56,8 @@ func (r consolidateResult) String() string {
 		return "Delete (empty node)"
 	case consolidateResultReplace:
 		return "Replace"
+	case consolidateResultNoAction:
+		return "NoAction"
 	default:
 		return fmt.Sprintf("Unknown (%d)", r)
 	}

pkg/controllers/node/expiration.go

@@ -47,6 +47,9 @@ func (r *Expiration) Reconcile(ctx context.Context, provisioner *v1alpha5.Provis
 	expirationTime := node.CreationTimestamp.Add(expirationTTL)
 	if r.clock.Now().After(expirationTime) {
 		logging.FromContext(ctx).Infof("Triggering termination for expired node after %s (+%s)", expirationTTL, time.Since(expirationTime))
+
+		// The delete operation implicitly marks the node for deletion for handling with scheduling
+		// This also implicitly triggers provisioning of the new node since at least one pod should go pending
 		if err := r.kubeClient.Delete(ctx, node); err != nil {
 			return reconcile.Result{}, fmt.Errorf("deleting node, %w", err)
 		}

pkg/controllers/provisioning/provisioner.go

@@ -25,6 +25,7 @@ import (
 
 	"github.com/imdario/mergo"
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/samber/lo"
 	"go.uber.org/multierr"
 	appsv1 "k8s.io/api/apps/v1"
 	v1 "k8s.io/api/core/v1"
@@ -47,6 +48,7 @@ import (
 	"github.com/aws/karpenter/pkg/metrics"
 	"github.com/aws/karpenter/pkg/scheduling"
 	"github.com/aws/karpenter/pkg/utils/injection"
+	"github.com/aws/karpenter/pkg/utils/node"
 	"github.com/aws/karpenter/pkg/utils/pod"
 	"github.com/aws/karpenter/pkg/utils/resources"
 )
@@ -140,16 +142,33 @@ func (p *Provisioner) Provision(ctx context.Context) error {
 	// scheduling loop when we launch a new node.  When this order is reversed, our node capacity may be reduced by pods
 	// that have bound which we then provision new un-needed capacity for.
 	var stateNodes []*state.Node
+	var markedForDeletionNodes []*state.Node
 	p.cluster.ForEachNode(func(node *state.Node) bool {
-		stateNodes = append(stateNodes, node.DeepCopy())
+		// We don't consider the nodes that are MarkedForDeletion since this capacity shouldn't be considered
+		// as persistent capacity for the cluster (since it will soon be removed). Additionally, we are scheduling for
+		// the pods that are on these nodes so the MarkedForDeletion node capacity can't be considered.
+		if !node.MarkedForDeletion {
+			stateNodes = append(stateNodes, node.DeepCopy())
+		} else {
+			markedForDeletionNodes = append(markedForDeletionNodes, node.DeepCopy())
+		}
 		return true
 	})
 
 	// Get pods, exit if nothing to do
-	pods, err := p.getPods(ctx)
+	pendingPods, err := p.getPendingPods(ctx)
+	if err != nil {
+		return err
+	}
+	// Get pods from nodes that are preparing for deletion
+	// We do this after getting the pending pods so that we undershoot if pods are
+	// actively migrating from a node that is being deleted
+	// NOTE: The assumption is that these nodes are cordoned and no additional pods will schedule to them
+	deletingNodePods, err := node.GetNodePods(ctx, p.kubeClient, lo.Map(markedForDeletionNodes, func(n *state.Node, _ int) *v1.Node { return n.Node })...)
 	if err != nil {
 		return err
 	}
+	pods := append(pendingPods, deletingNodePods...)
 	if len(pods) == 0 {
 		return nil
 	}
@@ -194,7 +213,7 @@ func (p *Provisioner) LaunchNodes(ctx context.Context, opts LaunchOptions, nodes
 	return nodeNames, nil
 }
 
-func (p *Provisioner) getPods(ctx context.Context) ([]*v1.Pod, error) {
+func (p *Provisioner) getPendingPods(ctx context.Context) ([]*v1.Pod, error) {
 	var podList v1.PodList
 	if err := p.kubeClient.List(ctx, &podList, client.MatchingFields{"spec.nodeName": ""}); err != nil {
 		return nil, fmt.Errorf("listing pods, %w", err)

pkg/controllers/provisioning/suite_test.go

@@ -47,6 +47,9 @@ import (
 
 var ctx context.Context
 var fakeClock *clock.FakeClock
+var cluster *state.Cluster
+var nodeController *state.NodeController
+var cloudProvider cloudprovider.CloudProvider
 var controller *provisioning.Controller
 var env *test.Environment
 var recorder *test.EventRecorder
@@ -61,12 +64,12 @@ func TestAPIs(t *testing.T) {
 
 var _ = BeforeSuite(func() {
 	env = test.NewEnvironment(ctx, func(e *test.Environment) {
-		cloudProvider := &fake.CloudProvider{}
-		recorder = test.NewEventRecorder()
+		cloudProvider = &fake.CloudProvider{}
 		cfg = test.NewConfig()
 		recorder = test.NewEventRecorder()
 		fakeClock = clock.NewFakeClock(time.Now())
-		cluster := state.NewCluster(fakeClock, cfg, e.Client, cloudProvider)
+		cluster = state.NewCluster(fakeClock, cfg, e.Client, cloudProvider)
+		nodeController = state.NewNodeController(e.Client, cluster)
 		prov := provisioning.NewProvisioner(ctx, cfg, e.Client, corev1.NewForConfigOrDie(e.Config), recorder, cloudProvider, cluster)
 		controller = provisioning.NewController(e.Client, prov, recorder)
 		instanceTypes, _ := cloudProvider.GetInstanceTypes(context.Background(), nil)
@@ -78,6 +81,11 @@ var _ = BeforeSuite(func() {
 	Expect(env.Start()).To(Succeed(), "Failed to start environment")
 })
 
+var _ = BeforeEach(func() {
+	recorder.Reset()
+	cluster = state.NewCluster(fakeClock, cfg, env.Client, cloudProvider)
+})
+
 var _ = AfterSuite(func() {
 	Expect(env.Stop()).To(Succeed(), "Failed to stop environment")
 })
@@ -182,6 +190,45 @@ var _ = Describe("Provisioning", func() {
 			ExpectScheduled(ctx, env.Client, pod)
 		}
 	})
+	It("should schedule all pods on one node when node is in deleting state", func() {
+		provisioner := test.Provisioner()
+		its, err := cloudProvider.GetInstanceTypes(ctx, provisioner)
+		Expect(err).To(BeNil())
+		node := test.Node(test.NodeOptions{
+			ObjectMeta: metav1.ObjectMeta{
+				Labels: map[string]string{
+					v1alpha5.ProvisionerNameLabelKey: provisioner.Name,
+					v1.LabelInstanceTypeStable:       its[0].Name(),
+				},
+				Finalizers: []string{v1alpha5.TerminationFinalizer},
+			}},
+		)
+		ExpectApplied(ctx, env.Client, node, provisioner)
+		ExpectReconcileSucceeded(ctx, nodeController, client.ObjectKeyFromObject(node))
+
+		// Schedule 3 pods to the node that currently exists
+		for i := 0; i < 3; i++ {
+			pod := test.UnschedulablePod()
+			ExpectApplied(ctx, env.Client, pod)
+			ExpectManualBinding(ctx, env.Client, pod, node)
+		}
+
+		// Node shouldn't fully delete since it has a finalizer
+		Expect(env.Client.Delete(ctx, node)).To(Succeed())
+		ExpectReconcileSucceeded(ctx, nodeController, client.ObjectKeyFromObject(node))
+
+		// Provision without a binding since some pods will already be bound
+		// Should all schedule to the new node, ignoring the old node
+		ExpectProvisionedNoBinding(ctx, env.Client, controller, test.UnschedulablePod(), test.UnschedulablePod())
+		nodes := &v1.NodeList{}
+		Expect(env.Client.List(ctx, nodes)).To(Succeed())
+		Expect(len(nodes.Items)).To(Equal(2))
+
+		// Scheduler should attempt to schedule all the pods to the new node
+		recorder.ForEachBinding(func(p *v1.Pod, n *v1.Node) {
+			Expect(n.Name).ToNot(Equal(node.Name))
+		})
+	})
 	Context("Resource Limits", func() {
 		It("should not schedule when limits are exceeded", func() {
 			ExpectApplied(ctx, env.Client, test.Provisioner(test.ProvisionerOptions{

pkg/controllers/state/cluster.go

@@ -117,6 +117,9 @@ type Node struct {
 	PodTotalRequests v1.ResourceList
 	// PodTotalLimits is the total resource limits scheduled to this node
 	PodTotalLimits v1.ResourceList
+	// MarkedForDeletion marks this node to say that there is some controller that is
+	// planning to delete this node so consider pods that are present on it available for scheduling
+	MarkedForDeletion bool
 }
 
 // ForPodsWithAntiAffinity calls the supplied function once for each pod with required anti affinity terms that is
@@ -177,18 +180,37 @@ func (c *Cluster) NominateNodeForPod(nodeName string) {
 	c.nominatedNodes.SetDefault(nodeName, nil)
 }
 
+// UnmarkForDeletion removes the marking on the node as a node the controller intends to delete
+func (c *Cluster) UnmarkForDeletion(nodeName string) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if _, ok := c.nodes[nodeName]; ok {
+		c.nodes[nodeName].MarkedForDeletion = false
+	}
+}
+
+// MarkForDeletion marks the node as pending deletion in the internal cluster state
+func (c *Cluster) MarkForDeletion(nodeName string) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if _, ok := c.nodes[nodeName]; ok {
+		c.nodes[nodeName].MarkedForDeletion = true
+	}
+}
+
 // newNode always returns a node, even if some portion of the update has failed
 func (c *Cluster) newNode(ctx context.Context, node *v1.Node) (*Node, error) {
 	n := &Node{
-		Node:          node,
-		Capacity:      v1.ResourceList{},
-		Allocatable:   v1.ResourceList{},
-		Available:     v1.ResourceList{},
-		HostPortUsage: scheduling.NewHostPortUsage(),
-		VolumeUsage:   scheduling.NewVolumeLimits(c.kubeClient),
-		VolumeLimits:  scheduling.VolumeCount{},
-		podRequests:   map[types.NamespacedName]v1.ResourceList{},
-		podLimits:     map[types.NamespacedName]v1.ResourceList{},
+		Node:              node,
+		Capacity:          v1.ResourceList{},
+		Allocatable:       v1.ResourceList{},
+		Available:         v1.ResourceList{},
+		HostPortUsage:     scheduling.NewHostPortUsage(),
+		VolumeUsage:       scheduling.NewVolumeLimits(c.kubeClient),
+		VolumeLimits:      scheduling.VolumeCount{},
+		MarkedForDeletion: !node.DeletionTimestamp.IsZero(),
+		podRequests:       map[types.NamespacedName]v1.ResourceList{},
+		podLimits:         map[types.NamespacedName]v1.ResourceList{},
 	}
 	if err := multierr.Combine(
 		c.populateCapacity(ctx, node, n),
@@ -319,10 +341,15 @@ func (c *Cluster) updateNode(ctx context.Context, node *v1.Node) error {
 	oldNode, ok := c.nodes[node.Name]
 	// If the old node existed and its initialization status changed, we want to reconsider consolidation.  This handles
 	// a situation where we re-start with an unready node and it becomes ready later.
-	if ok && oldNode.Node.Labels[v1alpha5.LabelNodeInitialized] != n.Node.Labels[v1alpha5.LabelNodeInitialized] {
-		c.recordConsolidationChange()
+	if ok {
+		if oldNode.Node.Labels[v1alpha5.LabelNodeInitialized] != n.Node.Labels[v1alpha5.LabelNodeInitialized] {
+			c.recordConsolidationChange()
+		}
+		// We mark the node for deletion either:
+		// 1. If the DeletionTimestamp is set (the node is explicitly being deleted)
+		// 2. If the last state of the node has the node MarkedForDeletion
+		n.MarkedForDeletion = n.MarkedForDeletion || oldNode.MarkedForDeletion
 	}
-
 	c.nodes[node.Name] = n
 
 	if node.DeletionTimestamp != nil {

pkg/controllers/state/suite_test.go

@@ -508,6 +508,28 @@ var _ = Describe("Node Resource Level", func() {
 		ExpectNodeResourceRequest(node, v1.ResourceCPU, "2.5")
 		ExpectNodeResourceRequest(node, v1.ResourceMemory, "2Gi")
 	})
+	It("should mark node for deletion when node is deleted", func() {
+		node := test.Node(test.NodeOptions{
+			ObjectMeta: metav1.ObjectMeta{
+				Labels: map[string]string{
+					v1alpha5.ProvisionerNameLabelKey: provisioner.Name,
+					v1.LabelInstanceTypeStable:       cloudProvider.InstanceTypes[0].Name(),
+				},
+				Finalizers: []string{v1alpha5.TerminationFinalizer},
+			},
+			Allocatable: map[v1.ResourceName]resource.Quantity{
+				v1.ResourceCPU: resource.MustParse("4"),
+			}},
+		)
+		ExpectApplied(ctx, env.Client, node)
+
+		ExpectReconcileSucceeded(ctx, nodeController, client.ObjectKeyFromObject(node))
+		Expect(env.Client.Delete(ctx, node)).To(Succeed())
+
+		ExpectReconcileSucceeded(ctx, nodeController, client.ObjectKeyFromObject(node))
+		ExpectNodeExists(ctx, env.Client, node.Name)
+		ExpectNodeDeletionMarked(node)
+	})
 })
 
 var _ = Describe("Pod Anti-Affinity", func() {
@@ -722,3 +744,13 @@ func ExpectNodeDaemonSetRequested(node *v1.Node, resourceName v1.ResourceName, a
 		return false
 	})
 }
+
+func ExpectNodeDeletionMarked(node *v1.Node) {
+	cluster.ForEachNode(func(n *state.Node) bool {
+		if n.Node.Name != node.Name {
+			return true
+		}
+		Expect(n.MarkedForDeletion).To(BeTrue())
+		return false
+	})
+}

pkg/utils/node/node.go

@@ -0,0 +1,46 @@
+/*
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package node
+
+import (
+	"context"
+	"fmt"
+
+	v1 "k8s.io/api/core/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/aws/karpenter/pkg/utils/pod"
+)
+
+// GetNodePods gets the list of schedulable pods from a variadic list of nodes
+func GetNodePods(ctx context.Context, kubeClient client.Client, nodes ...*v1.Node) ([]*v1.Pod, error) {
+	var pods []*v1.Pod
+	for _, node := range nodes {
+		var podList v1.PodList
+		if err := kubeClient.List(ctx, &podList, client.MatchingFields{"spec.nodeName": node.Name}); err != nil {
+			return nil, fmt.Errorf("listing pods, %w", err)
+		}
+		for i := range podList.Items {
+			// these pods don't need to be rescheduled
+			if pod.IsOwnedByNode(&podList.Items[i]) ||
+				pod.IsOwnedByDaemonSet(&podList.Items[i]) ||
+				pod.IsTerminal(&podList.Items[i]) {
+				continue
+			}
+			pods = append(pods, &podList.Items[i])
+		}
+	}
+	return pods, nil
+}

test/pkg/environment/expectations.go

@@ -218,6 +218,13 @@ func (env *Environment) ExpectCreatedNodeCount(comparator string, nodeCount int)
 		fmt.Sprintf("expected %d created nodes, had %d", nodeCount, env.Monitor.CreatedNodes()))
 }
 
+func (env *Environment) EventuallyExpectCreatedNodeCount(comparator string, nodeCount int) {
+	Eventually(func(g Gomega) {
+		g.Expect(env.Monitor.CreatedNodes()).To(BeNumerically(comparator, nodeCount),
+			fmt.Sprintf("expected %d created nodes, had %d", nodeCount, env.Monitor.CreatedNodes()))
+	}).Should(Succeed())
+}
+
 func (env *Environment) GetNode(nodeName string) v1.Node {
 	var node v1.Node
 	Expect(env.Client.Get(env.Context, types.NamespacedName{Name: nodeName}, &node)).To(Succeed())

test/suites/integration/expiration_test.go

@@ -0,0 +1,125 @@
+/*
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package integration_test
+
+import (
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/intstr"
+	"knative.dev/pkg/ptr"
+
+	"github.com/aws/karpenter/pkg/apis/awsnodetemplate/v1alpha1"
+	"github.com/aws/karpenter/pkg/apis/provisioning/v1alpha5"
+	awsv1alpha1 "github.com/aws/karpenter/pkg/cloudprovider/aws/apis/v1alpha1"
+	"github.com/aws/karpenter/pkg/test"
+)
+
+var _ = Describe("Expiration", func() {
+	It("should expire the node after the TTLSecondsUntilExpired is reached", func() {
+		provider := test.AWSNodeTemplate(v1alpha1.AWSNodeTemplateSpec{AWS: awsv1alpha1.AWS{
+			SecurityGroupSelector: map[string]string{"karpenter.sh/discovery": env.ClusterName},
+			SubnetSelector:        map[string]string{"karpenter.sh/discovery": env.ClusterName},
+		}})
+		provisioner := test.Provisioner(test.ProvisionerOptions{
+			ProviderRef:            &v1alpha5.ProviderRef{Name: provider.Name},
+			TTLSecondsUntilExpired: ptr.Int64(30),
+		})
+		var numPods int32 = 3
+
+		dep := test.Deployment(test.DeploymentOptions{
+			Replicas: numPods,
+			PodOptions: test.PodOptions{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{"app": "large-app"},
+				},
+			},
+		})
+
+		env.ExpectCreatedNodeCount("==", 0)
+		env.ExpectCreated(provisioner, provider, dep)
+
+		// We don't care if the pod goes healthy, just if the node is expired
+		env.EventuallyExpectCreatedNodeCount("==", 1)
+		node := env.Monitor.GetCreatedNodes()[0]
+
+		// Eventually expect the node to be gone
+		env.EventuallyExpectNotFound(&node)
+	})
+	It("should replace expired node with a single node and schedule all pods", func() {
+		provider := test.AWSNodeTemplate(v1alpha1.AWSNodeTemplateSpec{AWS: awsv1alpha1.AWS{
+			SecurityGroupSelector: map[string]string{"karpenter.sh/discovery": env.ClusterName},
+			SubnetSelector:        map[string]string{"karpenter.sh/discovery": env.ClusterName},
+		}})
+		provisioner := test.Provisioner(test.ProvisionerOptions{
+			ProviderRef: &v1alpha5.ProviderRef{Name: provider.Name},
+		})
+		var numPods int32 = 5
+
+		// We should setup a PDB that will only allow a minimum of 1 pod to be pending at a time
+		minAvailable := intstr.FromInt(int(numPods) - 1)
+		pdb := test.PodDisruptionBudget(test.PDBOptions{
+			Labels: map[string]string{
+				"app": "large-app",
+			},
+			MinAvailable: &minAvailable,
+		})
+		dep := test.Deployment(test.DeploymentOptions{
+			Replicas: numPods,
+			PodOptions: test.PodOptions{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{"app": "large-app"},
+				},
+			},
+		})
+
+		selector := labels.SelectorFromSet(dep.Spec.Selector.MatchLabels)
+		env.ExpectCreatedNodeCount("==", 0)
+		env.ExpectCreated(provisioner, provider, pdb, dep)
+
+		env.EventuallyExpectHealthyPodCount(selector, int(numPods))
+		env.ExpectCreatedNodeCount("==", 1)
+
+		node := env.Monitor.GetCreatedNodes()[0]
+
+		// Reset the monitor so that we can expect a single node to be spun up after expiration
+		env.Monitor.Reset()
+
+		// Set the TTLSecondsUntilExpired to get the node deleted
+		provisioner.Spec.TTLSecondsUntilExpired = ptr.Int64(60)
+		env.ExpectUpdate(provisioner)
+
+		// Eventually the node deletion timestamp will be set
+		Eventually(func(g Gomega) {
+			n := &v1.Node{}
+			g.Expect(env.Client.Get(env.Context, types.NamespacedName{Name: node.Name}, n)).Should(Succeed())
+			g.Expect(n.DeletionTimestamp.IsZero()).Should(BeFalse())
+		}).Should(Succeed())
+
+		// Remove the TTLSecondsUntilExpired to make sure new node isn't deleted
+		provisioner.Spec.TTLSecondsUntilExpired = nil
+		env.ExpectUpdate(provisioner)
+
+		// After the deletion timestamp is set and all pods are drained
+		// the node should be gone
+		env.EventuallyExpectNotFound(&node)
+
+		env.EventuallyExpectHealthyPodCount(selector, int(numPods))
+		env.ExpectCreatedNodeCount("==", 1)
+	})
+})