[release-4.21] OCPBUGS-99756: increase catalog HTTP client timeout from 10s to 5m

internal/catalogd/serverutil/serverutil.go

@@ -1,7 +1,9 @@
 package serverutil
 
 import (
+	"context"
 	"crypto/tls"
+	"errors"
 	"fmt"
 	"io"
 	"net"
@@ -13,7 +15,7 @@ import (
 	"github.com/klauspost/compress/gzhttp"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/certwatcher"
-	"sigs.k8s.io/controller-runtime/pkg/manager"
+	"sigs.k8s.io/controller-runtime/pkg/healthz"
 
 	catalogdmetrics "github.com/operator-framework/operator-controller/internal/catalogd/metrics"
 	"github.com/operator-framework/operator-controller/internal/catalogd/storage"
@@ -27,49 +29,116 @@ type CatalogServerConfig struct {
 	LocalStorage storage.Instance
 }
 
-func AddCatalogServerToManager(mgr ctrl.Manager, cfg CatalogServerConfig, tlsFileWatcher *certwatcher.CertWatcher) error {
-	listener, err := net.Listen("tcp", cfg.CatalogAddr)
+// AddCatalogServerToManager adds the catalog HTTP server to the manager and registers
+// a readiness check that passes once the server has started serving.  Because
+// NeedLeaderElection returns false, Start() is called on every pod immediately, so all
+// replicas bind the catalog port and become ready.  Non-leader pods serve requests but
+// return 404 (empty local cache); callers are expected to retry.
+func AddCatalogServerToManager(mgr ctrl.Manager, cfg CatalogServerConfig, cw *certwatcher.CertWatcher) error {
+	shutdownTimeout := 30 * time.Second
+	r := &catalogServerRunnable{
+		cfg: cfg,
+		cw:  cw,
+		server: &http.Server{
+			Addr:         cfg.CatalogAddr,
+			Handler:      storageServerHandlerWrapped(mgr.GetLogger().WithName("catalogd-http-server"), cfg),
+			ReadTimeout:  5 * time.Second,
+			WriteTimeout: 5 * time.Minute,
+		},
+		shutdownTimeout: shutdownTimeout,
+		ready:           make(chan struct{}),
+	}
+
+	if err := mgr.Add(r); err != nil {
+		return fmt.Errorf("error adding catalog server to manager: %w", err)
+	}
+
+	// Register a readiness check that passes once Start() has been called and the
+	// server is actively serving.  All pods reach Start() (NeedLeaderElection=false),
+	// so all replicas become ready and receive traffic; non-leaders return 404 until
+	// they win the leader lease and populate their local cache.
+	if err := mgr.AddReadyzCheck("catalog-server", r.readyzCheck()); err != nil {
+		return fmt.Errorf("error adding catalog server readiness check: %w", err)
+	}
+
+	return nil
+}
+
+// catalogServerRunnable is a Runnable that binds the catalog HTTP port on every pod.
+// Because NeedLeaderElection returns false, Start() is called on all replicas immediately;
+// non-leader pods serve the catalog port but return 404 (empty local cache).
+type catalogServerRunnable struct {
+	cfg             CatalogServerConfig
+	cw              *certwatcher.CertWatcher
+	server          *http.Server
+	shutdownTimeout time.Duration
+	// ready is closed by Start() once the server is about to begin serving.
+	ready chan struct{}
+}
+
+// NeedLeaderElection returns false so the catalog server starts on every pod
+// immediately, regardless of leadership.  This is required for rolling updates:
+// if Start() were gated on leadership, a new pod could not win the leader lease
+// (held by the still-running old pod) and therefore could never pass the
+// catalog-server readiness check, deadlocking the rollout.
+//
+// Non-leader pods serve the catalog HTTP port but have an empty local cache
+// (only the leader's reconciler downloads catalog content), so requests to a
+// non-leader return 404.  Callers are expected to retry.
+func (r *catalogServerRunnable) NeedLeaderElection() bool { return false }
+
+func (r *catalogServerRunnable) Start(ctx context.Context) error {
+	listener, err := net.Listen("tcp", r.cfg.CatalogAddr)
 	if err != nil {
 		return fmt.Errorf("error creating catalog server listener: %w", err)
 	}
 
-	if cfg.CertFile != "" && cfg.KeyFile != "" {
-		// Use the passed certificate watcher instead of creating a new one
+	if r.cfg.CertFile != "" && r.cfg.KeyFile != "" {
 		config := &tls.Config{
-			GetCertificate: tlsFileWatcher.GetCertificate,
+			GetCertificate: r.cw.GetCertificate,
 			MinVersion:     tls.VersionTLS12,
 		}
 		listener = tls.NewListener(listener, config)
 	}
 
-	shutdownTimeout := 30 * time.Second
-	catalogServer := manager.Server{
-		Name:                "catalogs",
-		OnlyServeWhenLeader: true,
-		Server: &http.Server{
-			Addr:        cfg.CatalogAddr,
-			Handler:     storageServerHandlerWrapped(mgr.GetLogger().WithName("catalogd-http-server"), cfg),
-			ReadTimeout: 5 * time.Second,
-			// TODO: Revert this to 10 seconds if/when the API
-			// evolves to have significantly smaller responses
-			WriteTimeout: 5 * time.Minute,
-		},
-		ShutdownTimeout: &shutdownTimeout,
-		Listener:        listener,
-	}
+	// Signal readiness before blocking on Serve so the readiness probe passes promptly.
+	close(r.ready)
 
-	err = mgr.Add(&catalogServer)
-	if err != nil {
-		return fmt.Errorf("error adding catalog server to manager: %w", err)
-	}
+	go func() {
+		<-ctx.Done()
+		shutdownCtx := context.Background()
+		if r.shutdownTimeout > 0 {
+			var cancel context.CancelFunc
+			shutdownCtx, cancel = context.WithTimeout(shutdownCtx, r.shutdownTimeout)
+			defer cancel()
+		}
+		if err := r.server.Shutdown(shutdownCtx); err != nil {
+			// Shutdown errors (e.g. context deadline exceeded) are not actionable;
+			// the process is terminating regardless.
+			_ = err
+		}
+	}()
 
+	if err := r.server.Serve(listener); err != nil && !errors.Is(err, http.ErrServerClosed) {
+		return err
+	}
 	return nil
 }
 
+// readyzCheck returns a healthz.Checker that passes once Start() has been called.
+func (r *catalogServerRunnable) readyzCheck() healthz.Checker {
+	return func(_ *http.Request) error {
+		select {
+		case <-r.ready:
+			return nil
+		default:
+			return fmt.Errorf("catalog server not yet started")
+		}
+	}
+}
+
 func logrLoggingHandler(l logr.Logger, handler http.Handler) http.Handler {
 	return handlers.CustomLoggingHandler(nil, handler, func(_ io.Writer, params handlers.LogFormatterParams) {
-		// extract parameters used in apache common log format, but then log using `logr` to remain consistent
-		// with other loggers used in this codebase.
 		username := "-"
 		if params.URL.User != nil {
 			if name := params.URL.User.Username(); name != "" {

internal/operator-controller/catalogmetadata/client/client.go

@@ -106,8 +106,10 @@ func (c *Client) PopulateCache(ctx context.Context, catalog *ocv1.ClusterCatalog
 	defer resp.Body.Close()
 
 	if resp.StatusCode != http.StatusOK {
-		errToCache := fmt.Errorf("error: received unexpected response status code %d", resp.StatusCode)
-		return c.cache.Put(catalog.Name, catalog.Status.ResolvedSource.Image.Ref, nil, errToCache)
+		// Do not cache non-200 responses (e.g. 404 from a non-leader catalogd pod).
+		// Returning the error directly lets the next reconcile retry a fresh HTTP
+		// request and eventually hit the leader.
+		return nil, fmt.Errorf("error: received unexpected response status code %d", resp.StatusCode)
 	}
 
 	return c.cache.Put(catalog.Name, catalog.Status.ResolvedSource.Image.Ref, resp.Body, nil)

internal/operator-controller/catalogmetadata/client/client_test.go

@@ -234,13 +234,6 @@ func TestClientPopulateCache(t *testing.T) {
 					}},
 				}, nil
 			},
-			putFuncConstructor: func(t *testing.T) func(source string, errToCache error) (fs.FS, error) {
-				return func(source string, errToCache error) (fs.FS, error) {
-					assert.Empty(t, source)
-					assert.Error(t, errToCache)
-					return nil, errToCache
-				}
-			},
 			assert: func(t *testing.T, fs fs.FS, err error) {
 				assert.Nil(t, fs)
 				assert.ErrorContains(t, err, "received unexpected response status code 500")

internal/shared/util/http/httputil.go

@@ -7,7 +7,7 @@ import (
 )
 
 func BuildHTTPClient(cpw *CertPoolWatcher) (*http.Client, error) {
-	httpClient := &http.Client{Timeout: 10 * time.Second}
+	httpClient := &http.Client{Timeout: 5 * time.Minute}
 
 	pool, _, err := cpw.Get()
 	if err != nil {