diff --git a/pkg/kube/statuswait.go b/pkg/kube/statuswait.go index 91eec30ef..634faff3d 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,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,9 @@ func (w *statusWaiter) wait(ctx context.Context, resourceList ResourceList, sw w if rs.Status == status.CurrentStatus { continue } + if rs.Status == status.NotFoundStatus && observed.deletionIsSuccess(id, deletedIsDone) { + 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 +276,60 @@ 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 what was seen on the cluster while waiting, so that +// a resource which disappears can be told apart from one that was never there, +// and so that a failure is not forgotten when the resource is deleted +// afterwards. +type observedResources struct { + mu sync.Mutex + present map[object.ObjMetadata]struct{} + failed 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 (o *observedResources) markFailed(id object.ObjMetadata) { + o.mu.Lock() + defer o.mu.Unlock() + if o.failed == nil { + o.failed = map[object.ObjMetadata]struct{}{} + } + o.failed[id] = struct{}{} +} + +func (o *observedResources) hasFailed(id object.ObjMetadata) bool { + o.mu.Lock() + defer o.mu.Unlock() + _, ok := o.failed[id] + return ok +} + +// deletionIsSuccess reports whether a resource that is no longer found should +// end the wait successfully: it has to be one of the resources deletion is +// expected for, it has to have been seen running, and it must not have failed +// while it was. +func (o *observedResources) deletionIsSuccess(id object.ObjMetadata, deletedIsDone map[object.ObjMetadata]struct{}) bool { + if _, ok := deletedIsDone[id]; !ok { + return false + } + return o.wasPresent(id) && !o.hasFailed(id) +} + +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 @@ -250,11 +342,29 @@ func statusObserver(cancel context.CancelFunc, desired status.Status, logger *sl if rs.Status == status.UnknownStatus && desired == status.NotFoundStatus { continue } + if rs.Status != status.NotFoundStatus { + observed.markPresent(rs.Identifier) + } + // Remember the failure: the resource may be deleted shortly after, + // and the delete event would otherwise replace Failed with NotFound + // and hide it. + if rs.Status == status.FailedStatus { + observed.markFailed(rs.Identifier) + } // Failed is a terminal state. This check ensures we don't wait forever for a resource // that has already failed, as intervention is required to resolve the failure. if rs.Status == status.FailedStatus && desired == status.CurrentStatus { continue } + // 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 running on the cluster and did not fail: one that is + // already gone when the wait starts, or that failed before being + // deleted, is still an error. + if rs.Status == status.NotFoundStatus && observed.deletionIsSuccess(rs.Identifier, deletedIsDone) { + continue + } rss = append(rss, rs) if rs.Status != desired { nonDesiredResources = append(nonDesiredResources, rs) diff --git a/pkg/kube/statuswait_test.go b/pkg/kube/statuswait_test.go index 5f5f5d051..8df0d2029 100644 --- a/pkg/kube/statuswait_test.go +++ b/pkg/kube/statuswait_test.go @@ -1777,3 +1777,203 @@ func TestWatchUntilReadyWithCustomReaders(t *testing.T) { }) } } + +var jobTTLNoStatusManifest = ` +apiVersion: batch/v1 +kind: Job +metadata: + name: test + namespace: qual + generation: 1 +spec: + ttlSecondsAfterFinished: 0 +` + +// TestDeletionIsSuccess covers which disappearing resources may end a wait +// successfully. A failed hook must not pass just because the TTL controller +// removed it afterwards: the delete event replaces Failed with NotFound in the +// collector, so the failure has to be remembered. +func TestDeletionIsSuccess(t *testing.T) { + t.Parallel() + id := object.ObjMetadata{ + GroupKind: batchv1.SchemeGroupVersion.WithKind("Job").GroupKind(), + Namespace: "qual", + Name: "test", + } + other := id + other.Name = "other" + + tests := []struct { + name string + present bool + failed bool + ttl bool + expected bool + }{ + { + name: "TTL Job seen running and then deleted", + present: true, + ttl: true, + expected: true, + }, + { + name: "TTL Job that failed before being deleted", + present: true, + failed: true, + ttl: true, + expected: false, + }, + { + name: "TTL Job that was never seen running", + ttl: true, + expected: false, + }, + { + name: "resource deletion is not expected for", + present: true, + expected: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + observed := &observedResources{} + if tt.present { + observed.markPresent(id) + } + if tt.failed { + observed.markFailed(id) + } + deletedIsDone := map[object.ObjMetadata]struct{}{} + if tt.ttl { + deletedIsDone[id] = struct{}{} + } + assert.Equal(t, tt.expected, observed.deletionIsSuccess(id, deletedIsDone)) + // An unrelated resource is never covered by the exception. + assert.False(t, observed.deletionIsSuccess(other, deletedIsDone)) + }) + } +} + +// 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") +}