Fix aggregated API server readiness race causing sporadic 503s in envtest suites

Signed-off-by: Andreas Fritzler <andreas.fritzler@sap.com>

Author: afritzler

utils/envtest/envtest.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"net/url"
 	"os"
 	"path/filepath"
 	"reflect"
@@ -18,6 +19,7 @@ import (
 	"github.com/ironcore-dev/ironcore/utils/envtest/internal/testing/addr"
 	"github.com/ironcore-dev/ironcore/utils/envtest/internal/testing/certs"
 	corev1 "k8s.io/api/core/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/runtime"
@@ -658,9 +660,126 @@ func WaitUntilAPIServicesReady(ctx context.Context, ext *EnvironmentExtensions,
 		return fmt.Errorf("error waiting for group versions to be discoverable: %w", err)
 	}
 
+	if err := waitUntilAPIGroupsServe(ctx, c, scheme, groupVersions...); err != nil {
+		return fmt.Errorf("error waiting for aggregated api server to serve requests: %w", err)
+	}
+
+	return nil
+}
+
+// serveReadinessPollInterval is the interval at which the aggregated API server
+// is probed for serving real resource round-trips.
+const serveReadinessPollInterval = 100 * time.Millisecond
+
+// serveReadinessStableDuration is how long the aggregated API server must
+// continuously serve real resource round-trips before startup is considered
+// complete.
+//
+// The APIService Available condition and group discovery can report ready
+// while the aggregated server still returns transient 503 ServiceUnavailable
+// responses for a short window afterwards. Requiring sustained success absorbs
+// that window so tests observe a genuinely serving API server.
+const serveReadinessStableDuration = 500 * time.Millisecond
+
+// waitUntilAPIGroupsServe waits until the aggregated API server actually serves
+// resource round-trips for each of the given group versions by issuing a real
+// List per group. Unlike the APIService Available condition and discovery
+// (which are reported by / served from the kube-apiserver), a List is proxied
+// to the aggregated server, so it observes the transient 503 startup window and
+// absorbs it before the test suite proceeds.
+func waitUntilAPIGroupsServe(ctx context.Context, c client.Client, scheme *runtime.Scheme, gvs ...schema.GroupVersion) error {
+	targets, err := newListRoundTripTargets(scheme, gvs)
+	if err != nil {
+		return fmt.Errorf("error determining list round-trip targets: %w", err)
+	}
+	if len(targets) == 0 {
+		return nil
+	}
+
+	requiredSuccesses := int(serveReadinessStableDuration / serveReadinessPollInterval)
+	if requiredSuccesses < 1 {
+		requiredSuccesses = 1
+	}
+
+	successes := 0
+	var lastErr error
+	if err := wait.PollUntilContextCancel(ctx, serveReadinessPollInterval, true, func(ctx context.Context) (bool, error) {
+		for _, target := range targets {
+			if err := c.List(ctx, target.list); err != nil {
+				if !isTransientServeError(err) {
+					return false, fmt.Errorf("unexpected error listing %s: %w", target.gv, err)
+				}
+				lastErr = err
+				successes = 0
+				return false, nil
+			}
+		}
+		successes++
+		return successes >= requiredSuccesses, nil
+	}); err != nil {
+		if lastErr != nil {
+			return fmt.Errorf("%w (last error: %v)", err, lastErr)
+		}
+		return err
+	}
 	return nil
 }
 
+// listRoundTripTarget pairs a group version with a reusable list object used to
+// issue a real List round-trip against the aggregated API server.
+type listRoundTripTarget struct {
+	gv   schema.GroupVersion
+	list client.ObjectList
+}
+
+func newListRoundTripTargets(scheme *runtime.Scheme, gvs []schema.GroupVersion) ([]listRoundTripTarget, error) {
+	var targets []listRoundTripTarget
+	for _, gv := range gvs {
+		list, ok := newListForGroupVersion(scheme, gv)
+		if !ok {
+			continue
+		}
+		targets = append(targets, listRoundTripTarget{gv: gv, list: list})
+	}
+	return targets, nil
+}
+
+// newListForGroupVersion returns a fresh list object for the first list kind
+// registered for the given group version (e.g. *VolumeList), constructed via the
+// scheme rather than reflect so the scheme stays the single source of truth for
+// type instantiation.
+func newListForGroupVersion(scheme *runtime.Scheme, gv schema.GroupVersion) (client.ObjectList, bool) {
+	for kind := range scheme.KnownTypes(gv) {
+		obj, err := scheme.New(gv.WithKind(kind))
+		if err != nil {
+			continue
+		}
+		if list, ok := obj.(client.ObjectList); ok {
+			return list, true
+		}
+	}
+	return nil, false
+}
+
+// isTransientServeError reports whether an error returned while probing the
+// aggregated API server is a transient startup condition worth retrying.
+func isTransientServeError(err error) bool {
+	if err == nil {
+		return false
+	}
+	if apierrors.IsServiceUnavailable(err) ||
+		apierrors.IsServerTimeout(err) ||
+		apierrors.IsTimeout(err) ||
+		apierrors.IsTooManyRequests(err) ||
+		apierrors.IsInternalError(err) {
+		return true
+	}
+	// Transport-level errors (connection refused, EOF, reset) while the
+	// aggregated server is still coming up.
+	var urlErr *url.Error
+	return errors.As(err, &urlErr)
+}
+
 func WaitUntilAPIServicesReadyWithTimeout(timeout time.Duration, ext *EnvironmentExtensions, cfg *rest.Config, c client.Client, scheme *runtime.Scheme) error {
 	ctx, cancel := context.WithTimeout(context.Background(), timeout)
 	defer cancel()