fix: Helm doesn't wait for Custom Resources on new releases using `upgrade --

Fixes helm/helm#32066

Signed-off-by: Pratheek Nathani <181894361+pratheeknathani@users.noreply.github.com>

pkg/kube/statuswait.go

@@ -207,11 +207,15 @@ func (w *statusWaiter) wait(ctx context.Context, resourceList ResourceList, sw w
 	errs := []error{}
 	for _, id := range resources {
 		rs := statusCollector.ResourceStatuses[id]
-		if rs.Status == status.CurrentStatus {
+		reportedStatus := rs.Status
+		if reportedStatus == status.CurrentStatus && customResourceStatusPending(rs.Resource) {
+			reportedStatus = status.InProgressStatus
+		}
+		if reportedStatus == status.CurrentStatus {
 			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))
+			rs.Identifier.GroupKind.Kind, rs.Identifier.Namespace, rs.Identifier.Name, reportedStatus, rs.Message))
 	}
 	if err := ctx.Err(); err != nil {
 		errs = append(errs, err)
@@ -254,6 +258,15 @@ func statusObserver(cancel context.CancelFunc, desired status.Status, logger *sl
 			if rs.Status == status.FailedStatus && desired == status.CurrentStatus {
 				continue
 			}
+			// A freshly created custom resource whose controller has not yet observed
+			// it has an empty status, which kstatus reports as Current. Treat such a
+			// resource as still in progress so we keep waiting until its controller has
+			// reconciled it. See https://github.com/helm/helm/issues/32066.
+			if desired == status.CurrentStatus && rs.Status == status.CurrentStatus && customResourceStatusPending(rs.Resource) {
+				rsCopy := *rs
+				rsCopy.Status = status.InProgressStatus
+				rs = &rsCopy
+			}
 			rss = append(rss, rs)
 			if rs.Status != desired {
 				nonDesiredResources = append(nonDesiredResources, rs)
@@ -277,6 +290,44 @@ func statusObserver(cancel context.CancelFunc, desired status.Status, logger *sl
 	}
 }
 
+// customResourceStatusPending reports whether u is a freshly created custom
+// resource whose controller has not yet written any status. kstatus reports
+// such a resource as Current because its .status is still empty, which can
+// cause Helm to stop waiting before the controller has reconciled the resource
+// (helm/helm#32066). We only return true when there is positive evidence that a
+// controller is expected to populate status:
+//   - The resource is not one of the built-in types kstatus already understands.
+//   - metadata.generation is non-zero (the API server tracks the spec
+//     generation for the object).
+//   - The .status object is still effectively empty (the controller has not
+//     written anything to it yet).
+//
+// As soon as the controller writes any status we defer to the normal kstatus
+// computation, so a resource is never blocked forever on account of this
+// heuristic.
+func customResourceStatusPending(u *unstructured.Unstructured) bool {
+	if u == nil {
+		return false
+	}
+	if status.GetLegacyConditionsFn(u) != nil {
+		return false
+	}
+	generation, found, err := unstructured.NestedInt64(u.Object, "metadata", "generation")
+	if err != nil || !found || generation == 0 {
+		return false
+	}
+	// If the controller has written anything to .status, defer to the normal
+	// kstatus computation rather than overriding its result.
+	statusMap, found, err := unstructured.NestedMap(u.Object, "status")
+	if err != nil {
+		return false
+	}
+	if found && len(statusMap) > 0 {
+		return false
+	}
+	return true
+}
+
 type hookOnlyWaiter struct {
 	sw *statusWaiter
 }

pkg/kube/statuswait_test.go

@@ -43,6 +43,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/yaml"
 	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/cli-runtime/pkg/resource"
 	dynamicfake "k8s.io/client-go/dynamic/fake"
 	clienttesting "k8s.io/client-go/testing"
 	"k8s.io/kubectl/pkg/scheme"
@@ -61,6 +62,29 @@ status:
   phase: Running
 `
 
+var customResourceNoStatusManifest = `
+apiVersion: serving.knative.dev/v1
+kind: Service
+metadata:
+  name: hello-knative
+  namespace: ns
+  generation: 1
+`
+
+var customResourceReadyManifest = `
+apiVersion: serving.knative.dev/v1
+kind: Service
+metadata:
+  name: hello-knative
+  namespace: ns
+  generation: 1
+status:
+  observedGeneration: 1
+  conditions:
+  - type: Ready
+    status: "True"
+`
+
 var podNoStatusManifest = `
 apiVersion: v1
 kind: Pod
@@ -450,6 +474,86 @@ func TestStatusWait(t *testing.T) {
 	}
 }
 
+func TestStatusWaitCustomResourcePending(t *testing.T) {
+	t.Parallel()
+	customResourceGVK := schema.GroupVersionKind{Group: "serving.knative.dev", Version: "v1", Kind: "Service"}
+	customResourceGVR := schema.GroupVersionResource{Group: "serving.knative.dev", Version: "v1", Resource: "services"}
+	customListKinds := map[schema.GroupVersionResource]string{customResourceGVR: "ServiceList"}
+
+	resourceListFor := func(objs []runtime.Object) ResourceList {
+		rl := ResourceList{}
+		for _, obj := range objs {
+			u := obj.(*unstructured.Unstructured)
+			rl = append(rl, &resource.Info{
+				Object:    u,
+				Name:      u.GetName(),
+				Namespace: u.GetNamespace(),
+			})
+		}
+		return rl
+	}
+
+	t.Run("freshly created custom resource is not reported ready before its controller reconciles it", func(t *testing.T) {
+		t.Parallel()
+		// Use a scheme local to this subtest. NewSimpleDynamicClientWithCustomListKinds
+		// registers the custom list kind into the scheme it is given, so sharing the
+		// global scheme.Scheme across parallel tests would be a data race.
+		fakeClient := dynamicfake.NewSimpleDynamicClientWithCustomListKinds(runtime.NewScheme(), customListKinds)
+		fakeMapper := testutil.NewFakeRESTMapper(customResourceGVK)
+		statusWaiter := statusWaiter{
+			client:     fakeClient,
+			restMapper: fakeMapper,
+		}
+		statusWaiter.SetLogger(slog.Default().Handler())
+		objs := getRuntimeObjFromManifests(t, []string{customResourceNoStatusManifest})
+		for _, obj := range objs {
+			u := obj.(*unstructured.Unstructured)
+			gvr := getGVR(t, fakeMapper, u)
+			require.NoError(t, fakeClient.Tracker().Create(gvr, u, u.GetNamespace()))
+		}
+		err := statusWaiter.Wait(resourceListFor(objs), time.Second*2)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "resource Service/ns/hello-knative not ready. status: InProgress")
+		assert.Contains(t, err.Error(), "context deadline exceeded")
+	})
+
+	t.Run("custom resource is reported ready once its controller has reconciled it", func(t *testing.T) {
+		t.Parallel()
+		// Use a scheme local to this subtest (see note above) to avoid racing on
+		// the global scheme.Scheme with the other parallel subtest.
+		fakeClient := dynamicfake.NewSimpleDynamicClientWithCustomListKinds(runtime.NewScheme(), customListKinds)
+		fakeMapper := testutil.NewFakeRESTMapper(customResourceGVK)
+		statusWaiter := statusWaiter{
+			client:     fakeClient,
+			restMapper: fakeMapper,
+		}
+		statusWaiter.SetLogger(slog.Default().Handler())
+		objs := getRuntimeObjFromManifests(t, []string{customResourceNoStatusManifest})
+		u := objs[0].(*unstructured.Unstructured)
+		gvr := getGVR(t, fakeMapper, u)
+		require.NoError(t, fakeClient.Tracker().Create(gvr, u, u.GetNamespace()))
+
+		// Build the "reconciled" object on the test goroutine; only the tracker
+		// update runs in the background goroutine, and its error is reported back
+		// over a channel so we never touch t or testify from another goroutine.
+		ready := getRuntimeObjFromManifests(t, []string{customResourceReadyManifest})[0].(*unstructured.Unstructured)
+		const reconcileDelay = 500 * time.Millisecond
+		updateErrCh := make(chan error, 1)
+		go func() {
+			time.Sleep(reconcileDelay)
+			updateErrCh <- fakeClient.Tracker().Update(gvr, ready, ready.GetNamespace())
+		}()
+
+		start := time.Now()
+		err := statusWaiter.Wait(resourceListFor(objs), time.Second*10)
+		require.NoError(t, err)
+		require.NoError(t, <-updateErrCh)
+		// Wait must not succeed before the controller reconciled the resource;
+		// otherwise it would be reporting success prematurely (the bug in #32066).
+		require.GreaterOrEqual(t, time.Since(start), reconcileDelay)
+	})
+}
+
 func TestWaitForJobComplete(t *testing.T) {
 	t.Parallel()
 	tests := []struct {