diff --git a/pkg/kube/statuswait.go b/pkg/kube/statuswait.go index adc7a425a..eccd4d6ec 100644 --- a/pkg/kube/statuswait.go +++ b/pkg/kube/statuswait.go @@ -33,7 +33,9 @@ import ( "github.com/fluxcd/cli-utils/pkg/kstatus/watcher" "github.com/fluxcd/cli-utils/pkg/object" appsv1 "k8s.io/api/apps/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/client-go/dynamic" watchtools "k8s.io/client-go/tools/watch" @@ -163,6 +165,14 @@ func (w *statusWaiter) waitForDelete(ctx context.Context, resourceList ResourceL if rs.Status == status.NotFoundStatus || rs.Status == status.UnknownStatus { continue } + // The watcher may have missed the deletion event (e.g. due to a + // connection drop or informer lag). Verify with a live GET before + // reporting the resource as still existing. + if w.isResourceGone(id) { + w.Logger().Debug("watcher reported resource as existing but live GET confirms deletion", + "kind", id.GroupKind.Kind, "namespace", id.Namespace, "name", id.Name) + continue + } errs = append(errs, fmt.Errorf("resource %s/%s/%s still exists. status: %s, message: %s", rs.Identifier.GroupKind.Kind, rs.Identifier.Namespace, rs.Identifier.Name, rs.Status, rs.Message)) } @@ -183,6 +193,17 @@ func (w *statusWaiter) waitForDelete(ctx context.Context, resourceList ResourceL return nil } +func (w *statusWaiter) isResourceGone(id object.ObjMetadata) bool { + mapping, err := w.restMapper.RESTMapping(id.GroupKind) + if err != nil { + return false + } + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + _, err = w.client.Resource(mapping.Resource).Namespace(id.Namespace).Get(ctx, id.Name, metav1.GetOptions{}) + return apierrors.IsNotFound(err) +} + func (w *statusWaiter) wait(ctx context.Context, resourceList ResourceList, sw watcher.StatusWatcher) error { cancelCtx, cancel := context.WithCancel(ctx) defer cancel() diff --git a/pkg/kube/statuswait_test.go b/pkg/kube/statuswait_test.go index 73a424720..6951edebb 100644 --- a/pkg/kube/statuswait_test.go +++ b/pkg/kube/statuswait_test.go @@ -379,6 +379,68 @@ func TestStatusWaitForDeleteNonExistentObject(t *testing.T) { assert.NoError(t, err) } +func TestWaitForDeleteWithMissedWatchEvent(t *testing.T) { + t.Parallel() + c := newTestClient(t) + fakeClient := dynamicfake.NewSimpleDynamicClient(scheme.Scheme) + fakeMapper := testutil.NewFakeRESTMapper( + v1.SchemeGroupVersion.WithKind("Pod"), + ) + // Return a watcher with no events to simulate a missed deletion notification. + fakeClient.PrependWatchReactor("pods", func(action clienttesting.Action) (bool, watch.Interface, error) { + return true, watch.NewFake(), nil + }) + sw := statusWaiter{ + restMapper: fakeMapper, + client: fakeClient, + } + sw.SetLogger(slog.Default().Handler()) + objs := getRuntimeObjFromManifests(t, []string{podCurrentManifest}) + for _, obj := range objs { + u := obj.(*unstructured.Unstructured) + gvr := getGVR(t, fakeMapper, u) + err := fakeClient.Tracker().Create(gvr, u, u.GetNamespace()) + require.NoError(t, err) + } + // Delete the resource after a brief delay. The watcher will miss this + // deletion because watch events are suppressed, but a live GET should + // confirm the resource is gone. + go func() { + time.Sleep(50 * time.Millisecond) + for _, obj := range objs { + u := obj.(*unstructured.Unstructured) + gvr := getGVR(t, fakeMapper, u) + err := fakeClient.Tracker().Delete(gvr, u.GetNamespace(), u.GetName()) + assert.NoError(t, err) + } + }() + resourceList := getResourceListFromRuntimeObjs(t, c, objs) + err := sw.WaitForDelete(resourceList, 500*time.Millisecond) + assert.NoError(t, err) +} + +func TestIsResourceGoneTimeoutError(t *testing.T) { + t.Parallel() + fakeClient := dynamicfake.NewSimpleDynamicClient(scheme.Scheme) + fakeMapper := testutil.NewFakeRESTMapper( + v1.SchemeGroupVersion.WithKind("Pod"), + ) + fakeClient.PrependReactor("get", "pods", func(action clienttesting.Action) (bool, runtime.Object, error) { + return true, nil, context.DeadlineExceeded + }) + sw := statusWaiter{ + restMapper: fakeMapper, + client: fakeClient, + } + sw.SetLogger(slog.Default().Handler()) + id := object.ObjMetadata{ + GroupKind: v1.SchemeGroupVersion.WithKind("Pod").GroupKind(), + Namespace: "ns", + Name: "timeout-pod", + } + assert.False(t, sw.isResourceGone(id)) +} + func TestStatusWait(t *testing.T) { t.Parallel() tests := []struct {