feat: Add grace period for do-not-disrupt annotation (kubernetes-sigs#2874)

Author: AndrewMitchell25

designs/do-not-disrupt-grace-period.md

@@ -79,7 +79,7 @@ Extend the existing `karpenter.sh/do-not-disrupt` annotation to accept duration
 #### Annotation Values
 
 - `"true"`: Indefinite protection (existing behavior, backward compatible)
-- Duration string: Protection for the specified duration from pod binding time (e.g., `"4h"`, `"30m"`, `"1h30m"`)
+- Duration string: Protection for the specified duration from pod start time (e.g., `"4h"`, `"30m"`, `"1h30m"`)
   - Follows Go's `time.Duration` format
   - Common examples: `"30m"`, `"1h"`, `"4h"`, `"24h"`, `"1h30m"`
 
@@ -100,19 +100,19 @@ spec:
 
 1. **Indefinite Protection**: If `karpenter.sh/do-not-disrupt: "true"` is set, the behavior remains the same as today - indefinite protection (backward compatible)
 2. **Time-Limited Protection**: If set to a duration value (e.g., `"4h"`):
-   - The pod is protected from disruption for the specified duration starting from the pod's binding time
+   - The pod is protected from disruption for the specified duration starting from the pod's start time
    - After the grace period expires, the pod is treated as if it doesn't have the do-not-disrupt annotation
    - The node becomes eligible for disruption if no other constraints prevent it
-3. **Invalid Values**: If the value cannot be parsed as either `"true"` or a valid duration, it is treated as indefinite protection (backward compatible, fail-safe behavior)
+3. **Invalid Values**: If the value cannot be parsed as either `"true"` or a valid duration, the pod is treated as if it doesn't have the do-not-disrupt annotation, maintaining Karpenter's current behavior
 
 ### Time Calculation
 
 The grace period expiration time is calculated as:
 ```
-expiration_time = pod.BoundTimestamp + parsed_duration
+expiration_time = pod.Status.StartTime + parsed_duration
 ```
 
-For example, if a pod is bound at `2024-01-01T10:00:00Z` with `karpenter.sh/do-not-disrupt: "4h"`, it will be protected until `2024-01-01T14:00:00Z`.
+For example, if a pod starts at `2024-01-01T10:00:00Z` with `karpenter.sh/do-not-disrupt: "4h"`, it will be protected until `2024-01-01T14:00:00Z`.
 
 ### Implementation Details
 
@@ -127,13 +127,13 @@ func parseDoNotDisrupt(value string) (indefinite bool, duration time.Duration, e
 
     d, err := time.ParseDuration(value)
     if err != nil {
-        // Invalid format - treat as indefinite (fail-safe)
-        return true, 0, nil
+        // Invalid format - treat as if annotation doesn't exist
+        return false, 0, fmt.Errorf("failed to parse %q as a duration: %w", value, err)
     }
 
     if d <= 0 {
-        // Zero or negative - treat as indefinite (fail-safe)
-        return true, 0, nil
+        // Zero or negative - treat as if annotation doesn't exist
+        return false, 0, fmt.Errorf("duration %q must be positive", value)
     }
 
     return false, d, nil
@@ -175,7 +175,7 @@ spec:
 ```
 
 **Behavior**:
-- For the first 2 hours after pod binding, the node hosting this job will not be disrupted
+- For the first 2 hours after pod start, the node hosting this job will not be disrupted
 - After 2 hours, if the job hasn't completed, the node becomes eligible for disruption
 - This provides protection for normal execution while ensuring the node can eventually be disrupted
 
@@ -332,7 +332,7 @@ The annotation controls **when** a pod can be disrupted. Pod's `terminationGrace
 
 Users can determine the grace period status by:
 1. Checking pod annotations for the grace period value
-2. Comparing current time against `pod.BindingTimestamp + grace_period`
+2. Comparing current time against `pod.Status.StartTime + grace_period`
 3. Observing Karpenter events when nodes become eligible for disruption
 
 Potential future enhancements:
@@ -399,7 +399,6 @@ While ideally this would be a Kubernetes-native PDB feature, the immediate probl
 
 This feature is fully backward compatible:
 - Existing pods with only `karpenter.sh/do-not-disrupt: "true"` continue to work as before
-- Invalid grace period values default to indefinite protection
 - No changes to existing API contracts
 
 ## Rollout Plan

pkg/controllers/controllers.go

@@ -158,7 +158,7 @@ func NewControllers(
 
 	if options.FromContext(ctx).FeatureGates.StaticCapacity {
 		controllers = append(controllers, staticprovisioning.NewController(kubeClient, cluster, recorder, cloudProvider, p, clock))
-		controllers = append(controllers, staticdeprovisioning.NewController(kubeClient, cluster, cloudProvider, clock))
+		controllers = append(controllers, staticdeprovisioning.NewController(kubeClient, cluster, cloudProvider, clock, recorder))
 	}
 
 	if options.FromContext(ctx).FeatureGates.NodeOverlay {

pkg/controllers/disruption/consolidation.go

@@ -136,7 +136,7 @@ func (c *consolidation) sortCandidates(candidates []*Candidate) []*Candidate {
 func (c *consolidation) computeConsolidation(ctx context.Context, candidates ...*Candidate) (Command, error) {
 	var err error
 	// Run scheduling simulation to compute consolidation option
-	results, err := SimulateScheduling(ctx, c.kubeClient, c.cluster, c.provisioner, candidates...)
+	results, err := SimulateScheduling(ctx, c.kubeClient, c.cluster, c.provisioner, c.clock, c.recorder, candidates...)
 	if err != nil {
 		// if a candidate node is now deleting, just retry
 		if errors.Is(err, errCandidateDeleting) {

pkg/controllers/disruption/consolidation_test.go

@@ -2018,7 +2018,7 @@ var _ = Describe("Consolidation", func() {
 			Entry("if the candidate is on-demand node", false),
 			Entry("if the candidate is spot node", true),
 		)
-		DescribeTable("can replace nodes, considers karpenter.sh/do-not-disrupt on nodes",
+		DescribeTable("can replace nodes, considers karpenter.sh/do-not-disrupt set to true on nodes",
 			func(spotToSpot bool) {
 				nodeClaim = lo.Ternary(spotToSpot, spotNodeClaim, nodeClaim)
 				node = lo.Ternary(spotToSpot, spotNode, node)
@@ -2108,7 +2108,7 @@ var _ = Describe("Consolidation", func() {
 			Entry("if the candidate is on-demand node", false),
 			Entry("if the candidate is spot node", true),
 		)
-		DescribeTable("can replace nodes, considers karpenter.sh/do-not-disrupt on pods",
+		DescribeTable("can replace nodes, considers karpenter.sh/do-not-disrupt set to true on pods",
 			func(spotToSpot bool) {
 				nodeClaim = lo.Ternary(spotToSpot, spotNodeClaim, nodeClaim)
 				node = lo.Ternary(spotToSpot, spotNode, node)
@@ -2627,7 +2627,7 @@ var _ = Describe("Consolidation", func() {
 			// eviction
 			ExpectNotFound(ctx, env.Client, nodeClaims[0], nodes[0])
 		})
-		It("can delete nodes, considers karpenter.sh/do-not-disrupt on nodes", func() {
+		It("can delete nodes, considers karpenter.sh/do-not-disrupt set to true on nodes", func() {
 			// create our RS so we can link a pod to it
 			rs := test.ReplicaSet()
 			ExpectApplied(ctx, env.Client, rs)
@@ -2669,7 +2669,7 @@ var _ = Describe("Consolidation", func() {
 			Expect(ExpectNodes(ctx, env.Client)).To(HaveLen(1))
 			ExpectNotFound(ctx, env.Client, nodeClaims[0], nodes[0])
 		})
-		It("can delete nodes, considers karpenter.sh/do-not-disrupt on pods", func() {
+		It("can delete nodes, considers karpenter.sh/do-not-disrupt set to true on pods", func() {
 			// create our RS so we can link a pod to it
 			rs := test.ReplicaSet()
 			ExpectApplied(ctx, env.Client, rs)
@@ -2712,7 +2712,7 @@ var _ = Describe("Consolidation", func() {
 			Expect(ExpectNodes(ctx, env.Client)).To(HaveLen(1))
 			ExpectNotFound(ctx, env.Client, nodeClaims[0], nodes[0])
 		})
-		It("does not consolidate nodes with karpenter.sh/do-not-disrupt on pods when the NodePool's TerminationGracePeriod is not nil", func() {
+		It("does not consolidate nodes with karpenter.sh/do-not-disrupt set to true on pods when the NodePool's TerminationGracePeriod is not nil", func() {
 			// create our RS so we can link a pod to it
 			rs := test.ReplicaSet()
 			ExpectApplied(ctx, env.Client, rs)
@@ -2807,6 +2807,138 @@ var _ = Describe("Consolidation", func() {
 			Expect(ExpectNodeClaims(ctx, env.Client)).To(HaveLen(2))
 			Expect(ExpectNodes(ctx, env.Client)).To(HaveLen(2))
 		})
+		It("does not consolidate nodes with pods that have a duration-based do-not-disrupt annotation that is still active", func() {
+			rs := test.ReplicaSet()
+			ExpectApplied(ctx, env.Client, rs)
+			Expect(env.Client.Get(ctx, client.ObjectKeyFromObject(rs), rs)).To(Succeed())
+
+			pods := test.Pods(3, test.PodOptions{
+				ObjectMeta: metav1.ObjectMeta{Labels: labels,
+					OwnerReferences: []metav1.OwnerReference{
+						{
+							APIVersion:         "apps/v1",
+							Kind:               "ReplicaSet",
+							Name:               rs.Name,
+							UID:                rs.UID,
+							Controller:         lo.ToPtr(true),
+							BlockOwnerDeletion: lo.ToPtr(true),
+						},
+					}}})
+			// Set a 2m duration-based do-not-disrupt annotation on pod[2]
+			pods[2].Annotations = lo.Assign(pods[2].Annotations, map[string]string{v1.DoNotDisruptAnnotationKey: "2m"})
+			pods[2].Status.StartTime = &metav1.Time{Time: fakeClock.Now()}
+
+			ExpectApplied(ctx, env.Client, rs, pods[0], pods[1], pods[2], nodePool)
+			ExpectApplied(ctx, env.Client, nodeClaims[0], nodes[0], nodeClaims[1], nodes[1])
+
+			ExpectManualBinding(ctx, env.Client, pods[0], nodes[0])
+			ExpectManualBinding(ctx, env.Client, pods[1], nodes[0])
+			ExpectManualBinding(ctx, env.Client, pods[2], nodes[1])
+
+			ExpectMakeNodesAndNodeClaimsInitializedAndStateUpdated(ctx, env.Client, nodeStateController, nodeClaimStateController, []*corev1.Node{nodes[0], nodes[1]}, []*v1.NodeClaim{nodeClaims[0], nodeClaims[1]})
+			ExpectSingletonReconciled(ctx, disruptionController)
+
+			// Grace period is still active on pod[2], so nodes[1] should not be consolidated
+			// Only nodes[0] (no annotated pods) can be consolidated
+			cmds := queue.GetCommands()
+			Expect(cmds).To(HaveLen(1))
+			ExpectObjectReconciled(ctx, env.Client, queue, cmds[0].Candidates[0].NodeClaim)
+			ExpectNodeClaimsCascadeDeletion(ctx, env.Client, nodeClaims[0])
+
+			// nodes[0] deleted, nodes[1] still protected
+			Expect(ExpectNodeClaims(ctx, env.Client)).To(HaveLen(1))
+			Expect(ExpectNodes(ctx, env.Client)).To(HaveLen(1))
+			ExpectNotFound(ctx, env.Client, nodeClaims[0], nodes[0])
+		})
+		It("can consolidate nodes after duration-based do-not-disrupt annotation expires", func() {
+			rs := test.ReplicaSet()
+			ExpectApplied(ctx, env.Client, rs)
+			Expect(env.Client.Get(ctx, client.ObjectKeyFromObject(rs), rs)).To(Succeed())
+
+			pods := test.Pods(3, test.PodOptions{
+				ObjectMeta: metav1.ObjectMeta{Labels: labels,
+					OwnerReferences: []metav1.OwnerReference{
+						{
+							APIVersion:         "apps/v1",
+							Kind:               "ReplicaSet",
+							Name:               rs.Name,
+							UID:                rs.UID,
+							Controller:         lo.ToPtr(true),
+							BlockOwnerDeletion: lo.ToPtr(true),
+						},
+					}}})
+			// All pods have a 2m duration-based do-not-disrupt annotation
+			for _, p := range pods {
+				p.Annotations = lo.Assign(p.Annotations, map[string]string{v1.DoNotDisruptAnnotationKey: "2m"})
+				p.Status.StartTime = &metav1.Time{Time: fakeClock.Now()}
+			}
+
+			ExpectApplied(ctx, env.Client, rs, pods[0], pods[1], pods[2], nodePool)
+			ExpectApplied(ctx, env.Client, nodeClaims[0], nodes[0], nodeClaims[1], nodes[1])
+
+			ExpectManualBinding(ctx, env.Client, pods[0], nodes[0])
+			ExpectManualBinding(ctx, env.Client, pods[1], nodes[0])
+			ExpectManualBinding(ctx, env.Client, pods[2], nodes[1])
+
+			ExpectMakeNodesAndNodeClaimsInitializedAndStateUpdated(ctx, env.Client, nodeStateController, nodeClaimStateController, []*corev1.Node{nodes[0], nodes[1]}, []*v1.NodeClaim{nodeClaims[0], nodeClaims[1]})
+			ExpectSingletonReconciled(ctx, disruptionController)
+
+			// All pods have active grace periods, no consolidation should happen
+			cmds := queue.GetCommands()
+			Expect(cmds).To(HaveLen(0))
+			Expect(ExpectNodeClaims(ctx, env.Client)).To(HaveLen(2))
+			Expect(ExpectNodes(ctx, env.Client)).To(HaveLen(2))
+
+			// Advance clock past the 2m grace period
+			fakeClock.Step(3 * time.Minute)
+
+			ExpectMakeNodesAndNodeClaimsInitializedAndStateUpdated(ctx, env.Client, nodeStateController, nodeClaimStateController, []*corev1.Node{nodes[0], nodes[1]}, []*v1.NodeClaim{nodeClaims[0], nodeClaims[1]})
+			ExpectSingletonReconciled(ctx, disruptionController)
+
+			// Grace periods expired, consolidation should now proceed
+			cmds = queue.GetCommands()
+			Expect(cmds).To(HaveLen(1))
+		})
+		It("can delete nodes, considers invalid do-not-disrupt annotation format as not blocking", func() {
+			rs := test.ReplicaSet()
+			ExpectApplied(ctx, env.Client, rs)
+			Expect(env.Client.Get(ctx, client.ObjectKeyFromObject(rs), rs)).To(Succeed())
+
+			pods := test.Pods(3, test.PodOptions{
+				ObjectMeta: metav1.ObjectMeta{Labels: labels,
+					OwnerReferences: []metav1.OwnerReference{
+						{
+							APIVersion:         "apps/v1",
+							Kind:               "ReplicaSet",
+							Name:               rs.Name,
+							UID:                rs.UID,
+							Controller:         lo.ToPtr(true),
+							BlockOwnerDeletion: lo.ToPtr(true),
+						},
+					}}})
+			// Set an invalid format annotation - should not block consolidation
+			pods[2].Annotations = lo.Assign(pods[2].Annotations, map[string]string{v1.DoNotDisruptAnnotationKey: "invalid-format"})
+
+			ExpectApplied(ctx, env.Client, rs, pods[0], pods[1], pods[2], nodePool)
+			ExpectApplied(ctx, env.Client, nodeClaims[0], nodes[0], nodeClaims[1], nodes[1])
+
+			ExpectManualBinding(ctx, env.Client, pods[0], nodes[0])
+			ExpectManualBinding(ctx, env.Client, pods[1], nodes[0])
+			ExpectManualBinding(ctx, env.Client, pods[2], nodes[1])
+
+			ExpectMakeNodesAndNodeClaimsInitializedAndStateUpdated(ctx, env.Client, nodeStateController, nodeClaimStateController, []*corev1.Node{nodes[0], nodes[1]}, []*v1.NodeClaim{nodeClaims[0], nodeClaims[1]})
+			ExpectSingletonReconciled(ctx, disruptionController)
+
+			// Invalid annotation format should not block consolidation
+			cmds := queue.GetCommands()
+			Expect(cmds).To(HaveLen(1))
+			ExpectObjectReconciled(ctx, env.Client, queue, cmds[0].Candidates[0].NodeClaim)
+			ExpectNodeClaimsCascadeDeletion(ctx, env.Client, nodeClaims[1])
+
+			Expect(ExpectNodeClaims(ctx, env.Client)).To(HaveLen(1))
+			Expect(ExpectNodes(ctx, env.Client)).To(HaveLen(1))
+			ExpectNotFound(ctx, env.Client, nodeClaims[1], nodes[1])
+		})
 		It("can delete nodes, evicts pods without an ownerRef", func() {
 			// create our RS so we can link a pod to it
 			rs := test.ReplicaSet()
@@ -3443,7 +3575,7 @@ var _ = Describe("Consolidation", func() {
 			Expect(ExpectNodes(ctx, env.Client)).To(HaveLen(1))
 			ExpectExists(ctx, env.Client, nodeClaims[0])
 		})
-		It("should not replace node if a pod schedules with karpenter.sh/do-not-disrupt during the TTL wait", func() {
+		It("should not replace node if a pod schedules with karpenter.sh/do-not-disrupt set to true during the TTL wait", func() {
 			pod := test.Pod()
 			ExpectApplied(ctx, env.Client, nodePool, nodeClaim, node, pod)
 
@@ -3515,7 +3647,7 @@ var _ = Describe("Consolidation", func() {
 				},
 			)
 		})
-		It("should not delete node if pods schedule with karpenter.sh/do-not-disrupt during the TTL wait", func() {
+		It("should not delete node if pods schedule with karpenter.sh/do-not-disrupt set to true during the TTL wait", func() {
 			pods := test.Pods(2, test.PodOptions{})
 			ExpectApplied(ctx, env.Client, nodePool, nodeClaims[0], nodes[0], nodeClaims[1], nodes[1], pods[0], pods[1])
 

pkg/controllers/disruption/controller.go

@@ -103,7 +103,7 @@ func NewMethods(clk clock.Clock, cluster *state.Cluster, kubeClient client.Clien
 		// Terminate and create replacement for drifted NodeClaims in Static NodePool
 		NewStaticDrift(cluster, provisioner, cp),
 		// Terminate any NodeClaims that have drifted from provisioning specifications, allowing the pods to reschedule.
-		NewDrift(kubeClient, cluster, provisioner, recorder),
+		NewDrift(kubeClient, cluster, provisioner, recorder, clk),
 		// Attempt to identify multiple NodeClaims that we can consolidate simultaneously to reduce pod churn
 		NewMultiNodeConsolidation(c),
 		// And finally fall back our single NodeClaim consolidation to further reduce cluster cost.

pkg/controllers/disruption/drift.go

@@ -23,6 +23,7 @@ import (
 	"sort"
 
 	"github.com/samber/lo"
+	"k8s.io/utils/clock"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"sigs.k8s.io/karpenter/pkg/utils/pretty"
@@ -40,14 +41,16 @@ type Drift struct {
 	cluster     *state.Cluster
 	provisioner *provisioning.Provisioner
 	recorder    events.Recorder
+	clock       clock.Clock
 }
 
-func NewDrift(kubeClient client.Client, cluster *state.Cluster, provisioner *provisioning.Provisioner, recorder events.Recorder) *Drift {
+func NewDrift(kubeClient client.Client, cluster *state.Cluster, provisioner *provisioning.Provisioner, recorder events.Recorder, clk clock.Clock) *Drift {
 	return &Drift{
 		kubeClient:  kubeClient,
 		cluster:     cluster,
 		provisioner: provisioner,
 		recorder:    recorder,
+		clock:       clk,
 	}
 }
 
@@ -78,7 +81,7 @@ func (d *Drift) ComputeCommands(ctx context.Context, disruptionBudgetMapping map
 			continue
 		}
 		// Check if we need to create any NodeClaims.
-		results, err := SimulateScheduling(ctx, d.kubeClient, d.cluster, d.provisioner, candidate)
+		results, err := SimulateScheduling(ctx, d.kubeClient, d.cluster, d.provisioner, d.clock, d.recorder, candidate)
 		if err != nil {
 			// if a candidate is now deleting, just retry
 			if errors.Is(err, errCandidateDeleting) {