[lifecycle]: skip initialization timeout for nodes running workload pods

pkg/controllers/nodeclaim/lifecycle/liveness.go

@@ -25,6 +25,7 @@ import (
 	"github.com/samber/lo"
 	"k8s.io/apimachinery/pkg/api/errors"
 
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -38,6 +39,9 @@ import (
 	"sigs.k8s.io/karpenter/pkg/metrics"
 	"sigs.k8s.io/karpenter/pkg/operator/options"
 	"sigs.k8s.io/karpenter/pkg/state/nodepoolhealth"
+	nodeutils "sigs.k8s.io/karpenter/pkg/utils/node"
+	nodeclaimutils "sigs.k8s.io/karpenter/pkg/utils/nodeclaim"
+	podutils "sigs.k8s.io/karpenter/pkg/utils/pod"
 )
 
 type Liveness struct {
@@ -54,6 +58,8 @@ const (
 	registrationTimeoutReason   = "registration_timeout"
 	launchTimeoutReason         = "launch_timeout"
 	initializationTimeoutReason = "initialization_timeout"
+	// initializationTimeoutWorkloadRecheck is how often a timed-out but still-working node is re-examined.
+	initializationTimeoutWorkloadRecheck = 5 * time.Minute
 )
 
 // LaunchTimeout is a heuristic time that we expect to be able to launch within
@@ -135,6 +141,17 @@ func (l *Liveness) reconcileInitializationTimeout(ctx context.Context, nodeClaim
 	if timeUntilTimeout := initializationTimeout - l.clock.Since(registered.LastTransitionTime.Time); timeUntilTimeout > 0 {
 		return reconcile.Result{RequeueAfter: timeUntilTimeout}, nil
 	}
+	// A node that is serving workload pods is not stuck in the sense this timeout guards against, even if it never
+	// reports Initialized (e.g. an extended resource it advertised at launch never shows up). Deleting it would evict
+	// running workloads for no gain, so it is left alone and re-checked in case the pods later drain away.
+	hasWorkload, err := l.hasWorkloadPods(ctx, nodeClaim)
+	if err != nil {
+		return reconcile.Result{}, err
+	}
+	if hasWorkload {
+		log.FromContext(ctx).V(1).WithValues("timeout", initializationTimeout).Info("skipping initialization timeout for node running workload pods")
+		return reconcile.Result{RequeueAfter: initializationTimeoutWorkloadRecheck}, nil
+	}
 	if err := l.deleteNodeClaimForTimeout(ctx, initializationTimeout, initializationTimeoutReason, nodeClaim); err != nil {
 		if client.IgnoreNotFound(err) != nil {
 			return reconcile.Result{}, err
@@ -143,6 +160,26 @@ func (l *Liveness) reconcileInitializationTimeout(ctx context.Context, nodeClaim
 	return reconcile.Result{}, nil
 }
 
+// hasWorkloadPods reports whether the NodeClaim's node is running any pod that isn't part of the node's own
+// bootstrap, i.e. a non-DaemonSet pod that hasn't finished or started terminating. A NodeClaim whose node is not
+// found has no workload.
+func (l *Liveness) hasWorkloadPods(ctx context.Context, nodeClaim *v1.NodeClaim) (bool, error) {
+	node, err := nodeclaimutils.NodeForNodeClaim(ctx, l.kubeClient, nodeClaim)
+	if err != nil {
+		if nodeclaimutils.IsNodeNotFoundError(err) {
+			return false, nil
+		}
+		return false, err
+	}
+	pods, err := nodeutils.GetPods(ctx, l.kubeClient, node)
+	if err != nil {
+		return false, err
+	}
+	return lo.ContainsBy(pods, func(pod *corev1.Pod) bool {
+		return !podutils.IsOwnedByDaemonSet(pod) && !podutils.IsTerminal(pod) && !podutils.IsTerminating(pod)
+	}), nil
+}
+
 // updateNodePoolRegistrationHealth sets the NodeRegistrationHealthy=False
 // on the NodePool if the nodeClaim fails to launch/register
 func (l *Liveness) updateNodePoolRegistrationHealth(ctx context.Context, nodeClaim *v1.NodeClaim) error {

pkg/controllers/nodeclaim/lifecycle/liveness_test.go

@@ -29,6 +29,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 
 	v1 "sigs.k8s.io/karpenter/pkg/apis/v1"
 	"sigs.k8s.io/karpenter/pkg/cloudprovider/fake"
@@ -37,6 +38,7 @@ import (
 	"sigs.k8s.io/karpenter/pkg/operator/options"
 	"sigs.k8s.io/karpenter/pkg/test"
 	. "sigs.k8s.io/karpenter/pkg/test/expectations"
+	nodeclaimutils "sigs.k8s.io/karpenter/pkg/utils/nodeclaim"
 )
 
 var _ = Describe("Liveness", func() {
@@ -537,6 +539,40 @@ var _ = Describe("Liveness", func() {
 			ExpectExists(ctx, env.Client, nodeClaim)
 			ExpectExists(ctx, env.Client, node)
 		})
+		It("shouldn't delete a registered NodeClaim past the initialization timeout while its node runs workload pods", func() {
+			ctx = options.ToContext(ctx, test.Options(test.OptionsFields{NodeClaimInitializationTimeout: lo.ToPtr(time.Hour)}))
+			DeferCleanup(func() { ctx = options.ToContext(ctx, test.Options()) })
+			nodeClaim := registeredUninitializedNodeClaim()
+			node, err := nodeclaimutils.NodeForNodeClaim(ctx, env.Client, nodeClaim)
+			Expect(err).ToNot(HaveOccurred())
+
+			// DaemonSet pods are part of the node's bootstrap and don't count as workload
+			daemonSetPod := test.Pod(test.PodOptions{
+				ObjectMeta: metav1.ObjectMeta{OwnerReferences: []metav1.OwnerReference{{
+					APIVersion:         "apps/v1",
+					Kind:               "DaemonSet",
+					Name:               "bootstrap",
+					UID:                types.UID("bootstrap"),
+					Controller:         lo.ToPtr(true),
+					BlockOwnerDeletion: lo.ToPtr(true),
+				}}},
+				NodeName: node.Name,
+				Phase:    corev1.PodRunning,
+			})
+			workloadPod := test.Pod(test.PodOptions{NodeName: node.Name, Phase: corev1.PodRunning})
+			ExpectApplied(ctx, env.Client, daemonSetPod, workloadPod)
+
+			env.Clock.Step(2 * time.Hour)
+			result := ExpectObjectReconciled(ctx, env.Client, nodeClaimController, nodeClaim)
+			Expect(result.RequeueAfter).To(Equal(5 * time.Minute))
+			ExpectExists(ctx, env.Client, nodeClaim)
+
+			// Once the workload pod is gone, only the DaemonSet pod remains and the timeout applies
+			ExpectDeleted(ctx, env.Client, workloadPod)
+			ExpectObjectReconciled(ctx, env.Client, nodeClaimController, nodeClaim)
+			ExpectFinalizersRemoved(ctx, env.Client, nodeClaim)
+			ExpectNotFound(ctx, env.Client, nodeClaim)
+		})
 		It("should measure the initialization timeout from registration, not from the NodeClaim's creation", func() {
 			ctx = options.ToContext(ctx, test.Options(test.OptionsFields{NodeClaimInitializationTimeout: lo.ToPtr(time.Hour)}))
 			DeferCleanup(func() { ctx = options.ToContext(ctx, test.Options()) })