fix(kube): don't hang waiting on CRDs that are still being established

Signed-off-by: MrJack <36191829+biagiopietro@users.noreply.github.com>

pkg/kube/statuswait.go

@@ -33,10 +33,14 @@ 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"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/dynamic"
 	watchtools "k8s.io/client-go/tools/watch"
+	"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
 
 	"helm.sh/helm/v4/internal/logging"
 	helmStatusReaders "helm.sh/helm/v4/internal/statusreaders"
@@ -194,6 +198,15 @@ func (w *statusWaiter) wait(ctx context.Context, resourceList ResourceList, sw w
 		resources = append(resources, obj)
 	}
 
+	// Ensure the RESTMapper can resolve every GroupKind before the status
+	// watcher starts its informers. Otherwise a CRD that is being created or
+	// established asynchronously would cause the watcher to permanently
+	// abandon the informer for that GroupKind, leaving the affected resources
+	// stuck in the Unknown status until the wait times out.
+	if err := w.ensureResourceMappingsReady(cancelCtx, resources); err != nil {
+		return err
+	}
+
 	eventCh := sw.Watch(cancelCtx, resources, watcher.Options{
 		RESTScopeStrategy: watcher.RESTScopeNamespace,
 	})
@@ -237,6 +250,105 @@ func contextWithTimeout(ctx context.Context, timeout time.Duration) (context.Con
 	return watchtools.ContextWithOptionalTimeout(ctx, timeout)
 }
 
+// ensureResourceMappingsReady blocks until the RESTMapper can resolve every
+// GroupKind present in resources, or until the context is done.
+//
+// The status watcher starts one informer per GroupKind. If a GroupKind cannot
+// be mapped when its informer starts (e.g. because the backing CRD was created
+// by the same release and is not established yet), the underlying watcher gives
+// up on that informer permanently, leaving the affected resources stuck in the
+// Unknown status until the wait times out. Resolving the mappings up front
+// avoids that race so informers always start successfully.
+func (w *statusWaiter) ensureResourceMappingsReady(ctx context.Context, resources []object.ObjMetadata) error {
+	// Do not do any work if the wait has already been cancelled.
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	groupKinds := make([]schema.GroupKind, 0, len(resources))
+	seen := make(map[schema.GroupKind]struct{}, len(resources))
+	for _, r := range resources {
+		if r.GroupKind.Kind == "" {
+			// A resource without a type cannot be resolved; leave it to the
+			// watcher to report as not ready.
+			continue
+		}
+		if _, ok := seen[r.GroupKind]; ok {
+			continue
+		}
+		seen[r.GroupKind] = struct{}{}
+		groupKinds = append(groupKinds, r.GroupKind)
+	}
+	if len(groupKinds) == 0 {
+		return nil
+	}
+
+	// Fast path: if every GroupKind is already known there is nothing to wait
+	// for and the status watcher can start immediately.
+	if mappingsResolved(w.restMapper, groupKinds) {
+		return nil
+	}
+	w.Logger().Debug("waiting for resource types to become available",
+		"groupKinds", groupKindNames(groupKinds))
+
+	err := wait.PollUntilContextCancel(ctx, time.Second, true, func(_ context.Context) (bool, error) {
+		for _, gk := range groupKinds {
+			if _, err := w.restMapper.RESTMapping(gk); err != nil {
+				if isRetryableMappingError(err) {
+					return false, nil
+				}
+				return false, err
+			}
+		}
+		return true, nil
+	})
+	if err != nil && !wait.Interrupted(err) {
+		return err
+	}
+	return nil
+}
+
+// mappingsResolved returns true when the RESTMapper can resolve every GroupKind.
+func mappingsResolved(mapper meta.RESTMapper, groupKinds []schema.GroupKind) bool {
+	for _, gk := range groupKinds {
+		if _, err := mapper.RESTMapping(gk); err != nil {
+			return false
+		}
+	}
+	return true
+}
+
+// groupKindNames returns the GroupKinds in a human-readable form.
+func groupKindNames(groupKinds []schema.GroupKind) []string {
+	names := make([]string, 0, len(groupKinds))
+	for _, gk := range groupKinds {
+		if gk.Group == "" {
+			names = append(names, gk.Kind)
+		} else {
+			names = append(names, gk.Group+"/"+gk.Kind)
+		}
+	}
+	return names
+}
+
+// isRetryableMappingError returns true if a RESTMapper lookup failed because a
+// resource type is not (yet) known to the API server, which is expected while a
+// CRD is being created or established.
+func isRetryableMappingError(err error) bool {
+	if meta.IsNoMatchError(err) {
+		return true
+	}
+	discoveryFailed, ok := errors.AsType[*apiutil.ErrResourceDiscoveryFailed](err)
+	if !ok {
+		return false
+	}
+	for _, gvErr := range *discoveryFailed {
+		if meta.IsNoMatchError(gvErr) || apierrors.IsNotFound(gvErr) {
+			return true
+		}
+	}
+	return false
+}
+
 func statusObserver(cancel context.CancelFunc, desired status.Status, logger *slog.Logger) collector.ObserverFunc {
 	return func(statusCollector *collector.ResourceStatusCollector, _ event.Event) {
 		var rss []*event.ResourceStatus

pkg/kube/statuswait_crd_test.go

@@ -0,0 +1,167 @@
+/*
+Copyright The Helm Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package kube
+
+import (
+	"context"
+	"log/slog"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"k8s.io/apimachinery/pkg/api/meta"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/cli-runtime/pkg/resource"
+	dynamicfake "k8s.io/client-go/dynamic/fake"
+	"k8s.io/kubectl/pkg/scheme"
+)
+
+const rolloutManifest = `
+apiVersion: argoproj.io/v1alpha1
+kind: Rollout
+metadata:
+  name: app
+  namespace: default
+  generation: 1
+status:
+  observedGeneration: 1
+  conditions:
+  - type: Promoted
+    status: "True"
+`
+
+var rolloutGVR = schema.GroupVersionResource{
+	Group:    "argoproj.io",
+	Version:  "v1alpha1",
+	Resource: "rollouts",
+}
+
+var rolloutGK = schema.GroupKind{Group: "argoproj.io", Kind: "Rollout"}
+
+// delayedMapper simulates a CRD whose GroupKind is not registered in the
+// RESTMapper until a short delay has passed (e.g. the CRD is created by the
+// same Helm release and takes a moment to be established). Until then
+// RESTMapping returns a NoMatchError.
+type delayedMapper struct {
+	meta.RESTMapper
+	rolloutGK schema.GroupKind
+	ready     atomic.Bool
+}
+
+func (d *delayedMapper) RESTMapping(gk schema.GroupKind, versions ...string) (*meta.RESTMapping, error) {
+	if gk == d.rolloutGK && !d.ready.Load() {
+		return nil, &meta.NoResourceMatchError{
+			PartialResource: schema.GroupVersionResource{Group: gk.Group, Resource: "rollouts"},
+		}
+	}
+	return d.RESTMapper.RESTMapping(gk, versions...)
+}
+
+func newDelayedMapper(ready bool) *delayedMapper {
+	delegate := meta.NewDefaultRESTMapper([]schema.GroupVersion{
+		{Group: "argoproj.io", Version: "v1alpha1"},
+	})
+	delegate.Add(schema.GroupVersionKind{Group: "argoproj.io", Version: "v1alpha1", Kind: "Rollout"}, meta.RESTScopeNamespace)
+	mapper := &delayedMapper{
+		RESTMapper: delegate,
+		rolloutGK:  rolloutGK,
+	}
+	mapper.ready.Store(ready)
+	return mapper
+}
+
+func newRolloutStatusWaiter(t *testing.T, mapper meta.RESTMapper) (*statusWaiter, ResourceList) {
+	t.Helper()
+	rollout := getRuntimeObjFromManifests(t, []string{rolloutManifest})[0].(*unstructured.Unstructured)
+
+	fakeClient := dynamicfake.NewSimpleDynamicClientWithCustomListKinds(
+		scheme.Scheme,
+		map[schema.GroupVersionResource]string{rolloutGVR: "RolloutList"},
+		rollout,
+	)
+
+	sw := &statusWaiter{
+		client:     fakeClient,
+		restMapper: mapper,
+	}
+	sw.SetLogger(slog.Default().Handler())
+
+	resourceList := ResourceList{
+		&resource.Info{
+			Object:    rollout,
+			Namespace: rollout.GetNamespace(),
+			Name:      rollout.GetName(),
+		},
+	}
+	return sw, resourceList
+}
+
+// TestStatusWaitCustomResource ensures that waiting on a custom resource (such
+// as an Argo Rollout) does not hang when its CRD is not yet registered in the
+// RESTMapper when the wait starts. The wait must succeed as soon as the CRD
+// becomes available, instead of leaving the resource in the Unknown status
+// until the timeout.
+func TestStatusWaitCustomResource(t *testing.T) {
+	t.Parallel()
+	tests := []struct {
+		name       string
+		initReady  bool
+		readyAfter time.Duration
+	}{
+		{
+			name:      "CRD is already registered",
+			initReady: true,
+		},
+		{
+			// Simulate the CRD becoming established shortly after the wait
+			// starts.
+			name:       "CRD is registered while waiting",
+			readyAfter: 500 * time.Millisecond,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			mapper := newDelayedMapper(tt.initReady)
+			if tt.readyAfter > 0 {
+				time.AfterFunc(tt.readyAfter, func() {
+					mapper.ready.Store(true)
+				})
+			}
+
+			sw, resourceList := newRolloutStatusWaiter(t, mapper)
+
+			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+			defer cancel()
+			done := make(chan error, 1)
+			go func() {
+				done <- sw.Wait(resourceList, 5*time.Second)
+			}()
+
+			select {
+			case err := <-done:
+				if err != nil {
+					t.Fatalf("Wait failed: %v", err)
+				}
+			case <-ctx.Done():
+				t.Fatal("Wait hung: custom resource remained in the Unknown status despite its CRD becoming available")
+			}
+		})
+	}
+}