fix: make demo e2e catalog queries resilient to transient failures

The generate-demos CI job fails ~45% of the time with jq exit status 5
on the ClusterCatalog Quickstart scenario. Several issues contribute:

- jq -s (slurp mode) buffers the entire operatorhubio FBC response in
  memory before processing, risking system errors on large catalogs
- catalog content queries run exactly once with no retry, so any
  transient port-forward or network hiccup fails the step immediately
- bash() does not attach stderr to ExitError, making failures opaque
- with CatalogdHA, kubectl port-forward to the service deterministically
  picks the same pod via GetFirstPod sorting; if that pod is not the
  leader, it returns 404 (empty local cache) for every retry

Remove jq slurp mode so each JSON object is processed in constant
memory, prefixing filters with 'objects' to skip non-object values in
the FBC stream. Wrap CatalogContainsSomePackages, PackageHasSomeChannels,
and PackageHasSomeBundles in waitFor for retry on transient errors. Add
curl --compressed to handle gzip-encoded responses and --fail with
pipefail to detect HTTP errors. Resolve the catalogd leader pod via its
Lease and port-forward directly to it, falling back to the service when
the lease cannot be read. Reset port-forwards on query failure and
re-establish dead ones via liveness checks. Inject stderr into ExitError
in bash() to match k8sClient diagnostics. Log catalog query errors at
V(0) so CI timeout failures are diagnosable.

Co-Authored-By: Claude <noreply@anthropic.com>

Author: pedjak

test/e2e/steps/demo_steps.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"crypto/tls"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"net"
 	"net/http"
@@ -41,6 +42,10 @@ func bash(ctx context.Context, script string) (string, error) {
 
 	if err != nil {
 		logger.V(1).Info("Failed to run", "command", script, "stderr", stderr, "error", err)
+		var exitErr *exec.ExitError
+		if errors.As(err, &exitErr) {
+			exitErr.Stderr = stderrBuf.Bytes()
+		}
 	}
 	logger.V(1).Info("Output", "command", script, "output", stdout)
 
@@ -81,79 +86,144 @@ func CatalogReportsConditionWithoutReason(ctx context.Context, catalogUserName,
 func ensureCatalogPortForward(ctx context.Context) (string, error) {
 	sc := scenarioCtx(ctx)
 	if sc.catalogAddr != "" {
-		return sc.catalogAddr, nil
+		if catalogPortForwardAlive(sc.catalogAddr) {
+			return sc.catalogAddr, nil
+		}
+		logger.V(1).Info("Catalog port-forward is dead, re-establishing", "addr", sc.catalogAddr)
+		resetCatalogPortForward(ctx)
+	}
+
+	ns := componentNamespaces["catalogd"]
+	target, err := catalogdLeaderPod(ctx, ns)
+	if err != nil {
+		logger.V(1).Info("Could not resolve catalogd leader pod, falling back to service", "error", err)
+		target = "service/catalogd-service"
 	}
 
-	addr, cleanup, err := portForward(ctx, componentNamespaces["catalogd"], "service/catalogd-service", 443)
+	addr, cleanup, err := portForward(ctx, ns, target, 443)
 	if err != nil {
-		return "", fmt.Errorf("failed to start catalog port-forward: %w", err)
+		return "", fmt.Errorf("failed to start catalog port-forward to %s: %w", target, err)
 	}
 	sc.catalogAddr = addr
 	sc.catalogCleanup = cleanup
 
 	waitFor(ctx, func() bool {
-		client := &http.Client{
-			Timeout: 3 * time.Second,
-			Transport: &http.Transport{
-				TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, //nolint:gosec
-				DialContext:     (&net.Dialer{Timeout: 2 * time.Second}).DialContext,
-			},
-		}
-		resp, err := client.Get(fmt.Sprintf("https://%s/", addr))
-		if err != nil {
-			return false
-		}
-		resp.Body.Close()
-		return true
+		return catalogPortForwardAlive(addr)
 	})
 	return addr, nil
 }
 
+func catalogdLeaderPod(ctx context.Context, ns string) (string, error) {
+	holder, err := k8sClient(ctx, "get", "lease", "catalogd-operator-lock", "-n", ns,
+		"-o", "jsonpath={.spec.holderIdentity}")
+	if err != nil {
+		return "", fmt.Errorf("failed to get catalogd leader lease: %w", err)
+	}
+	holder = strings.TrimSpace(holder)
+	podName := holder
+	if idx := strings.LastIndex(holder, "_"); idx >= 0 {
+		podName = holder[:idx]
+	}
+	if podName == "" {
+		return "", fmt.Errorf("catalogd leader lease has empty holderIdentity")
+	}
+	logger.Info("Resolved catalogd leader pod", "holder", holder, "pod", podName)
+	return fmt.Sprintf("pod/%s", podName), nil
+}
+
+func catalogPortForwardAlive(addr string) bool {
+	client := &http.Client{
+		Timeout: 3 * time.Second,
+		Transport: &http.Transport{
+			TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, //nolint:gosec
+			DialContext:     (&net.Dialer{Timeout: 2 * time.Second}).DialContext,
+		},
+	}
+	resp, err := client.Get(fmt.Sprintf("https://%s/", addr))
+	if err != nil {
+		return false
+	}
+	resp.Body.Close()
+	return true
+}
+
+// resetCatalogPortForward tears down the cached port-forward so the next
+// call to ensureCatalogPortForward establishes a fresh connection.  With
+// CatalogdHA, non-leader pods return 404 (empty local cache); resetting
+// lets the next retry potentially reach the leader pod.
+func resetCatalogPortForward(ctx context.Context) {
+	sc := scenarioCtx(ctx)
+	if sc.catalogCleanup != nil {
+		sc.catalogCleanup()
+	}
+	sc.catalogAddr = ""
+	sc.catalogCleanup = nil
+}
+
 func catalogCurlJq(ctx context.Context, catalogName, jqFilter string) (string, error) {
 	addr, err := ensureCatalogPortForward(ctx)
 	if err != nil {
 		return "", err
 	}
 	script := fmt.Sprintf(
-		`curl -s -k https://%s/catalogs/%s/api/v1/all | jq -s '%s'`,
+		`set -o pipefail; curl -sS -k --compressed --fail https://%s/catalogs/%s/api/v1/all | jq '%s'`,
 		addr, catalogName, jqFilter,
 	)
-	return bash(ctx, script)
+	out, err := bash(ctx, script)
+	if err != nil {
+		resetCatalogPortForward(ctx)
+	}
+	return out, err
 }
 
 func CatalogContainsSomePackages(ctx context.Context, catalogName string) error {
-	out, err := catalogCurlJq(ctx, catalogName,
-		`.[] | select(.schema == "olm.package") | .name`)
-	if err != nil {
-		return err
-	}
-	if strings.TrimSpace(out) == "" {
-		return fmt.Errorf("catalog %q contains no packages", catalogName)
-	}
+	waitFor(ctx, func() bool {
+		out, err := catalogCurlJq(ctx, catalogName,
+			`objects | select(.schema == "olm.package") | .name`)
+		if err != nil {
+			logger.Info("Catalog query failed, retrying", "catalog", catalogName, "error", err, "stderr", stderrOutput(err))
+			return false
+		}
+		if strings.TrimSpace(out) == "" {
+			logger.Info("Catalog returned no packages, retrying", "catalog", catalogName)
+			return false
+		}
+		return true
+	})
 	return nil
 }
 
 func PackageHasSomeChannels(ctx context.Context, packageName, catalogName string) error {
-	out, err := catalogCurlJq(ctx, catalogName,
-		fmt.Sprintf(`.[] | select(.schema == "olm.channel") | select(.package == "%s") | .name`, packageName))
-	if err != nil {
-		return err
-	}
-	if strings.TrimSpace(out) == "" {
-		return fmt.Errorf("package %q in catalog %q has no channels", packageName, catalogName)
-	}
+	waitFor(ctx, func() bool {
+		out, err := catalogCurlJq(ctx, catalogName,
+			fmt.Sprintf(`objects | select(.schema == "olm.channel") | select(.package == "%s") | .name`, packageName))
+		if err != nil {
+			logger.Info("Catalog query failed, retrying", "catalog", catalogName, "package", packageName, "error", err, "stderr", stderrOutput(err))
+			return false
+		}
+		if strings.TrimSpace(out) == "" {
+			logger.Info("Package has no channels, retrying", "catalog", catalogName, "package", packageName)
+			return false
+		}
+		return true
+	})
 	return nil
 }
 
 func PackageHasSomeBundles(ctx context.Context, packageName, catalogName string) error {
-	out, err := catalogCurlJq(ctx, catalogName,
-		fmt.Sprintf(`.[] | select(.schema == "olm.bundle") | select(.package == "%s") | .name`, packageName))
-	if err != nil {
-		return err
-	}
-	if strings.TrimSpace(out) == "" {
-		return fmt.Errorf("package %q in catalog %q has no bundles", packageName, catalogName)
-	}
+	waitFor(ctx, func() bool {
+		out, err := catalogCurlJq(ctx, catalogName,
+			fmt.Sprintf(`objects | select(.schema == "olm.bundle") | select(.package == "%s") | .name`, packageName))
+		if err != nil {
+			logger.Info("Catalog query failed, retrying", "catalog", catalogName, "package", packageName, "error", err, "stderr", stderrOutput(err))
+			return false
+		}
+		if strings.TrimSpace(out) == "" {
+			logger.Info("Package has no bundles, retrying", "catalog", catalogName, "package", packageName)
+			return false
+		}
+		return true
+	})
 	return nil
 }