fix(kube): scope the hook wait exception to TTL Jobs

The first version accepted NotFound for any resource passed to
WatchUntilReady, which also accepted a Pod hook, a Job without a TTL, or a
hook deleted by an operator before it finished, and accepted a hook that was
already missing when the wait started.

The exception is now limited to the Jobs in the wait that set
.spec.ttlSecondsAfterFinished, which are the ones the TTL controller removes
when they complete, and only once the hook has actually been observed on the
cluster. Everything else that disappears, and a TTL Job that was never seen
running, still fail the wait.

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

pkg/kube/statuswait.go

@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"log/slog"
 	"sort"
+	"sync"
 	"time"
 
 	"github.com/fluxcd/cli-utils/pkg/kstatus/polling/aggregator"
@@ -33,6 +34,7 @@ import (
 	"github.com/fluxcd/cli-utils/pkg/kstatus/watcher"
 	"github.com/fluxcd/cli-utils/pkg/object"
 	appsv1 "k8s.io/api/apps/v1"
+	batchv1 "k8s.io/api/batch/v1"
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/client-go/dynamic"
@@ -95,11 +97,11 @@ func (w *statusWaiter) WatchUntilReady(resourceList ResourceList, timeout time.D
 		StatusReaders: append(w.readers, jobSR, podSR, genericSR),
 	}
 	sw.StatusReader = sr
-	// 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)
+	// A Job hook that sets .spec.ttlSecondsAfterFinished is removed by the TTL
+	// controller as soon as it completes, so it can disappear while Helm is
+	// still waiting for it. Only those hooks may report NotFound instead of
+	// completing; every other hook that goes away is still an error.
+	return w.waitFor(ctx, resourceList, sw, ttlJobs(resourceList))
 }
 
 func (w *statusWaiter) Wait(resourceList ResourceList, timeout time.Duration) error {
@@ -158,7 +160,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, false, w.Logger()))
+	done := statusCollector.ListenWithObserver(eventCh, statusObserver(cancel, status.NotFoundStatus, nil, &observedResources{}, w.Logger()))
 	<-done
 
 	if statusCollector.Error != nil {
@@ -184,13 +186,35 @@ 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)
+	return w.waitFor(ctx, resourceList, sw, nil)
 }
 
-// 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 {
+// ttlJobs returns the identities of the Jobs in resourceList that set
+// .spec.ttlSecondsAfterFinished, which the TTL controller deletes once they
+// finish.
+func ttlJobs(resourceList ResourceList) map[object.ObjMetadata]struct{} {
+	var ttl map[object.ObjMetadata]struct{}
+	for _, r := range resourceList {
+		job, ok := AsVersioned(r).(*batchv1.Job)
+		if !ok || job.Spec.TTLSecondsAfterFinished == nil {
+			continue
+		}
+		obj, err := object.RuntimeToObjMeta(r.Object)
+		if err != nil {
+			continue
+		}
+		if ttl == nil {
+			ttl = map[object.ObjMetadata]struct{}{}
+		}
+		ttl[obj] = struct{}{}
+	}
+	return ttl
+}
+
+// waitFor waits until every resource reaches the current status. Resources in
+// deletedIsDone are considered done when they are not found, instead of
+// blocking until the timeout expires.
+func (w *statusWaiter) waitFor(ctx context.Context, resourceList ResourceList, sw watcher.StatusWatcher, deletedIsDone map[object.ObjMetadata]struct{}) error {
 	cancelCtx, cancel := context.WithCancel(ctx)
 	defer cancel()
 	resources := []object.ObjMetadata{}
@@ -209,7 +233,8 @@ func (w *statusWaiter) waitFor(ctx context.Context, resourceList ResourceList, s
 		RESTScopeStrategy: watcher.RESTScopeNamespace,
 	})
 	statusCollector := collector.NewResourceStatusCollector(resources)
-	done := statusCollector.ListenWithObserver(eventCh, statusObserver(cancel, status.CurrentStatus, deletedIsDone, w.Logger()))
+	observed := &observedResources{}
+	done := statusCollector.ListenWithObserver(eventCh, statusObserver(cancel, status.CurrentStatus, deletedIsDone, observed, w.Logger()))
 	<-done
 
 	if statusCollector.Error != nil {
@@ -222,9 +247,11 @@ func (w *statusWaiter) waitFor(ctx context.Context, resourceList ResourceList, s
 		if rs.Status == status.CurrentStatus {
 			continue
 		}
-		if deletedIsDone && rs.Status == status.NotFoundStatus {
+		if rs.Status == status.NotFoundStatus && observed.wasPresent(id) {
+			if _, ok := deletedIsDone[id]; ok {
 				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))
 	}
@@ -251,7 +278,31 @@ func contextWithTimeout(ctx context.Context, timeout time.Duration) (context.Con
 	return watchtools.ContextWithOptionalTimeout(ctx, timeout)
 }
 
-func statusObserver(cancel context.CancelFunc, desired status.Status, deletedIsDone bool, logger *slog.Logger) collector.ObserverFunc {
+// observedResources records the resources that were seen on the cluster while
+// waiting, so that a resource which disappears can be told apart from one that
+// was never there.
+type observedResources struct {
+	mu      sync.Mutex
+	present map[object.ObjMetadata]struct{}
+}
+
+func (o *observedResources) markPresent(id object.ObjMetadata) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	if o.present == nil {
+		o.present = map[object.ObjMetadata]struct{}{}
+	}
+	o.present[id] = struct{}{}
+}
+
+func (o *observedResources) wasPresent(id object.ObjMetadata) bool {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	_, ok := o.present[id]
+	return ok
+}
+
+func statusObserver(cancel context.CancelFunc, desired status.Status, deletedIsDone map[object.ObjMetadata]struct{}, observed *observedResources, logger *slog.Logger) collector.ObserverFunc {
 	return func(statusCollector *collector.ResourceStatusCollector, _ event.Event) {
 		var rss []*event.ResourceStatus
 		var nonDesiredResources []*event.ResourceStatus
@@ -269,12 +320,19 @@ func statusObserver(cancel context.CancelFunc, desired status.Status, deletedIsD
 			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 {
+			if rs.Status != status.NotFoundStatus {
+				observed.markPresent(rs.Identifier)
+			}
+			// A Job hook that sets .spec.ttlSecondsAfterFinished is deleted by
+			// the TTL controller once it completes, so its disappearance ends
+			// the wait rather than blocking it. This only applies to a hook that
+			// was seen on the cluster first: one that is already gone when the
+			// wait starts was never observed running and is still an error.
+			if rs.Status == status.NotFoundStatus && observed.wasPresent(rs.Identifier) {
+				if _, ok := deletedIsDone[rs.Identifier]; ok {
 					continue
 				}
+			}
 			rss = append(rss, rs)
 			if rs.Status != desired {
 				nonDesiredResources = append(nonDesiredResources, rs)

pkg/kube/statuswait_test.go

@@ -1778,18 +1778,72 @@ 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) {
+var jobTTLNoStatusManifest = `
+apiVersion: batch/v1
+kind: Job
+metadata:
+   name: test
+   namespace: qual
+   generation: 1
+spec:
+   ttlSecondsAfterFinished: 0
+`
+
+// TestWatchUntilReadyHookDeleted covers hooks that disappear while Helm waits.
+// A Job that sets .spec.ttlSecondsAfterFinished is removed by the TTL
+// controller as soon as it completes, so its deletion ends the wait. Any other
+// hook that goes away, and a TTL Job that was never seen running, still fail.
+func TestWatchUntilReadyHookDeleted(t *testing.T) {
+	t.Parallel()
+	tests := []struct {
+		name          string
+		manifest      string
+		create        bool
+		expectErrStrs []string
+	}{
+		{
+			name:     "TTL Job deleted while waiting is done",
+			manifest: jobTTLNoStatusManifest,
+			create:   true,
+		},
+		{
+			name:     "Job without TTL deleted while waiting fails",
+			manifest: jobNoStatusManifest,
+			create:   true,
+			expectErrStrs: []string{
+				"resource Job/qual/test not ready. status: NotFound",
+				"context deadline exceeded",
+			},
+		},
+		{
+			name:     "Pod hook deleted while waiting fails",
+			manifest: podNoStatusManifest,
+			create:   true,
+			expectErrStrs: []string{
+				"resource Pod/ns/in-progress-pod not ready. status: NotFound",
+				"context deadline exceeded",
+			},
+		},
+		{
+			name:     "TTL Job that was never running fails",
+			manifest: jobTTLNoStatusManifest,
+			create:   false,
+			expectErrStrs: []string{
+				"resource Job/qual/test not ready.",
+				"context deadline exceeded",
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(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"),
+				v1.SchemeGroupVersion.WithKind("Pod"),
 			)
 			statusWaiter := statusWaiter{
 				restMapper: fakeMapper,
@@ -1797,21 +1851,33 @@ func TestWatchUntilReadyHookDeletedWhileWaiting(t *testing.T) {
 			}
 			statusWaiter.SetLogger(slog.Default().Handler())
 
-	// The Job never reports completion, so only its deletion can end the wait.
-	objs := getRuntimeObjFromManifests(t, []string{jobNoStatusManifest})
+			// The hook never reports completion, so only its deletion can end
+			// the wait.
+			objs := getRuntimeObjFromManifests(t, []string{tt.manifest})
 			u := objs[0].(*unstructured.Unstructured)
 			gvr := getGVR(t, fakeMapper, u)
+			if tt.create {
 				require.NoError(t, fakeClient.Tracker().Create(gvr, u, u.GetNamespace()))
-
 				go func() {
-		time.Sleep(timeUntilJobDelete)
+					time.Sleep(500 * time.Millisecond)
 					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))
+			err := statusWaiter.WatchUntilReady(resourceList, timeout)
+			if tt.expectErrStrs != nil {
+				require.Error(t, err)
+				for _, expectedErrStr := range tt.expectErrStrs {
+					require.ErrorContains(t, err, expectedErrStr)
+				}
+				return
+			}
+			require.NoError(t, err)
 			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