Merge 33076b3bd5 into 62c7e845d1

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"
@@ -167,6 +169,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))
 	}
@@ -179,6 +189,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()

pkg/kube/statuswait_test.go

@@ -375,6 +375,68 @@ func TestStatusWaitForDeleteNonExistentObject(t *testing.T) {
 	assert.NoError(t, statusWaiter.WaitForDelete(resourceList, timeout))
 }
 
+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 {