fix(kube): finish hook waits when the hook is deleted

The legacy waiter ended a hook wait as soon as it saw a delete event for the
watched resource. The kstatus watcher, which is the default in Helm 4, has no
equivalent rule: it waits for the Current status, so a hook that is removed
while Helm is waiting is reported as NotFound and blocks until the timeout
expires, failing the release.

That is the normal lifecycle of a Job hook that sets
.spec.ttlSecondsAfterFinished: the TTL controller removes the Job as soon as it
completes. The ingress-nginx chart, for example, sets that field on its
admission webhook patch Jobs, so upgrades hang for the whole hook timeout and
then fail.

WatchUntilReady is only used for hooks, so it now treats a resource that is not
found as done. Wait and WaitWithJobs are unchanged: for regular chart resources
a disappearing resource is still an error.

Closes #31786

Signed-off-by: ChadiDridi <dridichady@gmail.com>

pkg/kube/statuswait.go

@@ -95,7 +95,11 @@ func (w *statusWaiter) WatchUntilReady(resourceList ResourceList, timeout time.D
 		StatusReaders: append(w.readers, jobSR, podSR, genericSR),
 	}
 	sw.StatusReader = sr
-	return w.wait(ctx, resourceList, sw)
+	// Hook resources may legitimately disappear while Helm is waiting for them.
+	// A Job hook with .spec.ttlSecondsAfterFinished is removed by the TTL
+	// controller as soon as it completes, so a hook that is gone is treated as
+	// done rather than as a resource that never became ready.
+	return w.waitFor(ctx, resourceList, sw, true)
 }
 
 func (w *statusWaiter) Wait(resourceList ResourceList, timeout time.Duration) error {
@@ -154,7 +158,7 @@ func (w *statusWaiter) waitForDelete(ctx context.Context, resourceList ResourceL
 		RESTScopeStrategy: watcher.RESTScopeNamespace,
 	})
 	statusCollector := collector.NewResourceStatusCollector(resources)
-	done := statusCollector.ListenWithObserver(eventCh, statusObserver(cancel, status.NotFoundStatus, w.Logger()))
+	done := statusCollector.ListenWithObserver(eventCh, statusObserver(cancel, status.NotFoundStatus, false, w.Logger()))
 	<-done
 
 	if statusCollector.Error != nil {
@@ -180,6 +184,13 @@ func (w *statusWaiter) waitForDelete(ctx context.Context, resourceList ResourceL
 }
 
 func (w *statusWaiter) wait(ctx context.Context, resourceList ResourceList, sw watcher.StatusWatcher) error {
+	return w.waitFor(ctx, resourceList, sw, false)
+}
+
+// waitFor waits until every resource reaches the current status. When
+// deletedIsDone is set, a resource that is not found is considered done
+// instead of blocking until the timeout expires.
+func (w *statusWaiter) waitFor(ctx context.Context, resourceList ResourceList, sw watcher.StatusWatcher, deletedIsDone bool) error {
 	cancelCtx, cancel := context.WithCancel(ctx)
 	defer cancel()
 	resources := []object.ObjMetadata{}
@@ -198,7 +209,7 @@ func (w *statusWaiter) wait(ctx context.Context, resourceList ResourceList, sw w
 		RESTScopeStrategy: watcher.RESTScopeNamespace,
 	})
 	statusCollector := collector.NewResourceStatusCollector(resources)
-	done := statusCollector.ListenWithObserver(eventCh, statusObserver(cancel, status.CurrentStatus, w.Logger()))
+	done := statusCollector.ListenWithObserver(eventCh, statusObserver(cancel, status.CurrentStatus, deletedIsDone, w.Logger()))
 	<-done
 
 	if statusCollector.Error != nil {
@@ -211,6 +222,9 @@ func (w *statusWaiter) wait(ctx context.Context, resourceList ResourceList, sw w
 		if rs.Status == status.CurrentStatus {
 			continue
 		}
+		if deletedIsDone && rs.Status == status.NotFoundStatus {
+			continue
+		}
 		errs = append(errs, fmt.Errorf("resource %s/%s/%s not ready. status: %s, message: %s",
 			rs.Identifier.GroupKind.Kind, rs.Identifier.Namespace, rs.Identifier.Name, rs.Status, rs.Message))
 	}
@@ -237,7 +251,7 @@ func contextWithTimeout(ctx context.Context, timeout time.Duration) (context.Con
 	return watchtools.ContextWithOptionalTimeout(ctx, timeout)
 }
 
-func statusObserver(cancel context.CancelFunc, desired status.Status, logger *slog.Logger) collector.ObserverFunc {
+func statusObserver(cancel context.CancelFunc, desired status.Status, deletedIsDone bool, logger *slog.Logger) collector.ObserverFunc {
 	return func(statusCollector *collector.ResourceStatusCollector, _ event.Event) {
 		var rss []*event.ResourceStatus
 		var nonDesiredResources []*event.ResourceStatus
@@ -255,6 +269,12 @@ func statusObserver(cancel context.CancelFunc, desired status.Status, logger *sl
 			if rs.Status == status.FailedStatus && desired == status.CurrentStatus {
 				continue
 			}
+			// A hook that is gone has finished its job: the TTL controller
+			// removes completed Jobs that set .spec.ttlSecondsAfterFinished, and
+			// the legacy waiter treated a delete event as the end of the wait.
+			if deletedIsDone && rs.Status == status.NotFoundStatus {
+				continue
+			}
 			rss = append(rss, rs)
 			if rs.Status != desired {
 				nonDesiredResources = append(nonDesiredResources, rs)

pkg/kube/statuswait_test.go

@@ -1777,3 +1777,70 @@ func TestWatchUntilReadyWithCustomReaders(t *testing.T) {
 		})
 	}
 }
+
+// TestWatchUntilReadyHookDeletedWhileWaiting covers a Job hook that sets
+// .spec.ttlSecondsAfterFinished: the TTL controller removes the Job as soon as
+// it completes, so the hook can disappear while Helm is still waiting for it.
+// The wait has to end there instead of running until the timeout expires.
+func TestWatchUntilReadyHookDeletedWhileWaiting(t *testing.T) {
+	t.Parallel()
+	c := newTestClient(t)
+	timeout := 3 * time.Second
+	timeUntilJobDelete := 500 * time.Millisecond
+	fakeClient := dynamicfake.NewSimpleDynamicClient(scheme.Scheme)
+	fakeMapper := testutil.NewFakeRESTMapper(
+		batchv1.SchemeGroupVersion.WithKind("Job"),
+	)
+	statusWaiter := statusWaiter{
+		restMapper: fakeMapper,
+		client:     fakeClient,
+	}
+	statusWaiter.SetLogger(slog.Default().Handler())
+
+	// The Job never reports completion, so only its deletion can end the wait.
+	objs := getRuntimeObjFromManifests(t, []string{jobNoStatusManifest})
+	u := objs[0].(*unstructured.Unstructured)
+	gvr := getGVR(t, fakeMapper, u)
+	require.NoError(t, fakeClient.Tracker().Create(gvr, u, u.GetNamespace()))
+
+	go func() {
+		time.Sleep(timeUntilJobDelete)
+		assert.NoError(t, fakeClient.Tracker().Delete(gvr, u.GetNamespace(), u.GetName()))
+	}()
+
+	resourceList := getResourceListFromRuntimeObjs(t, c, objs)
+	start := time.Now()
+	require.NoError(t, statusWaiter.WatchUntilReady(resourceList, timeout))
+	assert.Less(t, time.Since(start), timeout, "wait should end when the hook is deleted, not on timeout")
+}
+
+// TestStatusWaitDeletedResourceStillFails makes sure the hook behaviour above
+// does not leak into Wait, where a resource that goes away is still an error.
+func TestStatusWaitDeletedResourceStillFails(t *testing.T) {
+	t.Parallel()
+	c := newTestClient(t)
+	fakeClient := dynamicfake.NewSimpleDynamicClient(scheme.Scheme)
+	fakeMapper := testutil.NewFakeRESTMapper(
+		batchv1.SchemeGroupVersion.WithKind("Job"),
+	)
+	statusWaiter := statusWaiter{
+		restMapper: fakeMapper,
+		client:     fakeClient,
+	}
+	statusWaiter.SetLogger(slog.Default().Handler())
+
+	objs := getRuntimeObjFromManifests(t, []string{jobNoStatusManifest})
+	u := objs[0].(*unstructured.Unstructured)
+	gvr := getGVR(t, fakeMapper, u)
+	require.NoError(t, fakeClient.Tracker().Create(gvr, u, u.GetNamespace()))
+
+	go func() {
+		time.Sleep(200 * time.Millisecond)
+		assert.NoError(t, fakeClient.Tracker().Delete(gvr, u.GetNamespace(), u.GetName()))
+	}()
+
+	resourceList := getResourceListFromRuntimeObjs(t, c, objs)
+	err := statusWaiter.Wait(resourceList, time.Second)
+	require.Error(t, err)
+	require.ErrorContains(t, err, "resource Job/qual/test not ready. status: NotFound")
+}