diff --git a/pkg/kube/statuswait.go b/pkg/kube/statuswait.go index dda1d2c40..f1ac06601 100644 --- a/pkg/kube/statuswait.go +++ b/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,8 +247,10 @@ func (w *statusWaiter) waitFor(ctx context.Context, resourceList ResourceList, s if rs.Status == status.CurrentStatus { continue } - if deletedIsDone && rs.Status == status.NotFoundStatus { - 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)) @@ -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,11 +320,18 @@ 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 { - 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 { diff --git a/pkg/kube/statuswait_test.go b/pkg/kube/statuswait_test.go index 6d2851ee5..cbfbfc2f4 100644 --- a/pkg/kube/statuswait_test.go +++ b/pkg/kube/statuswait_test.go @@ -1778,40 +1778,106 @@ 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() - 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, + 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", + }, + }, } - 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())) + 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()) - go func() { - time.Sleep(timeUntilJobDelete) - assert.NoError(t, fakeClient.Tracker().Delete(gvr, u.GetNamespace(), u.GetName())) - }() + // 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() - 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") + 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