From 4a055f1b248dcd1ee0c22ab2ff950a7fc6c83991 Mon Sep 17 00:00:00 2001 From: Todd Short Date: Tue, 21 Jul 2026 11:01:24 -0400 Subject: [PATCH 1/2] UPSTREAM: : OCPBUGS-99298: increase catalog HTTP client timeout from 10s to 5m The 10-second timeout was too aggressive for large catalog responses. Increase it to 5 minutes to avoid timeouts when fetching catalogs. Signed-off-by: Todd Short --- internal/shared/util/http/httputil.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/shared/util/http/httputil.go b/internal/shared/util/http/httputil.go index f5a982d2de..7bc65dd534 100644 --- a/internal/shared/util/http/httputil.go +++ b/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 { From 3c66e84e3ca6e365932aa59e7aa6fe03c371b8f1 Mon Sep 17 00:00:00 2001 From: Todd Short Date: Thu, 23 Jul 2026 11:19:57 -0400 Subject: [PATCH 2/2] UPSTREAM: : Fix catalogd HA readiness and non-200 cache handling Backport of upstream #2674. In HA topology (replicas=2), the old catalogd serverutil used OnlyServeWhenLeader=true: the non-leader pod bound the TCP port via net.Listen so the OS accepted connections, but no HTTP server was running. Operator-controller hit the non-leader ~50% of the time and the connection hung silently. With the HTTP client timeout raised to 5m (OCPBUGS-92037), this hang lasted the full 5 minutes instead of 10s, causing catalog caches to never be populated within test windows. Fix: - catalogd serverutil: switch to NeedLeaderElection=false so all pods serve HTTP; non-leaders return 404 (empty local cache) instead of hanging. Register a readiness check so the pod is excluded from endpoints until the server is ready. - client.go: do not cache non-200 responses; return the error directly so the next reconcile retries a fresh HTTP request and eventually hits the leader. Upstream: https://github.com/operator-framework/operator-controller/pull/2674 Signed-off-by: Todd Short --- internal/catalogd/serverutil/serverutil.go | 123 ++++++++++++++---- .../catalogmetadata/client/client.go | 6 +- .../catalogmetadata/client/client_test.go | 7 - 3 files changed, 100 insertions(+), 36 deletions(-) diff --git a/internal/catalogd/serverutil/serverutil.go b/internal/catalogd/serverutil/serverutil.go index 143d4c8763..c29b5742f7 100644 --- a/internal/catalogd/serverutil/serverutil.go +++ b/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 != "" { diff --git a/internal/operator-controller/catalogmetadata/client/client.go b/internal/operator-controller/catalogmetadata/client/client.go index 0d08c40ef5..e79fdfbcc5 100644 --- a/internal/operator-controller/catalogmetadata/client/client.go +++ b/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) diff --git a/internal/operator-controller/catalogmetadata/client/client_test.go b/internal/operator-controller/catalogmetadata/client/client_test.go index 00a226873e..66817c504b 100644 --- a/internal/operator-controller/catalogmetadata/client/client_test.go +++ b/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")