Merge e1da86fe73 into 62c7e845d1

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,7 +97,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)
+	// 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 {
@@ -154,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, w.Logger()))
+	done := statusCollector.ListenWithObserver(eventCh, statusObserver(cancel, status.NotFoundStatus, nil, &observedResources{}, w.Logger()))
 	<-done
 
 	if statusCollector.Error != nil {
@@ -180,6 +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, nil)
+}
+
+// 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{}
@@ -198,7 +233,8 @@ 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()))
+	observed := &observedResources{}
+	done := statusCollector.ListenWithObserver(eventCh, statusObserver(cancel, status.CurrentStatus, deletedIsDone, observed, w.Logger()))
 	<-done
 
 	if statusCollector.Error != nil {
@@ -211,6 +247,11 @@ func (w *statusWaiter) wait(ctx context.Context, resourceList ResourceList, sw w
 		if rs.Status == status.CurrentStatus {
 			continue
 		}
+		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))
 	}
@@ -237,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, 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
@@ -255,6 +320,19 @@ func statusObserver(cancel context.CancelFunc, desired status.Status, logger *sl
 			if rs.Status == status.FailedStatus && desired == status.CurrentStatus {
 				continue
 			}
+			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

@@ -1777,3 +1777,136 @@ func TestWatchUntilReadyWithCustomReaders(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
+			fakeClient := dynamicfake.NewSimpleDynamicClient(scheme.Scheme)
+			fakeMapper := testutil.NewFakeRESTMapper(
+				batchv1.SchemeGroupVersion.WithKind("Job"),
+				v1.SchemeGroupVersion.WithKind("Pod"),
+			)
+			statusWaiter := statusWaiter{
+				restMapper: fakeMapper,
+				client:     fakeClient,
+			}
+			statusWaiter.SetLogger(slog.Default().Handler())
+
+			// 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(500 * time.Millisecond)
+					assert.NoError(t, fakeClient.Tracker().Delete(gvr, u.GetNamespace(), u.GetName()))
+				}()
+			}
+
+			resourceList := getResourceListFromRuntimeObjs(t, c, objs)
+			start := time.Now()
+			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
+// 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")
+}