✨ prototyping dynamic graphql handler for catalogd

cmd/catalogd/main.go

@@ -22,6 +22,7 @@ import (
 	"errors"
 	"flag"
 	"fmt"
+	"net/http"
 	"net/url"
 	"os"
 	"path/filepath"
@@ -45,21 +46,21 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/healthz"
 	"sigs.k8s.io/controller-runtime/pkg/log"
-	"sigs.k8s.io/controller-runtime/pkg/metrics"
 	"sigs.k8s.io/controller-runtime/pkg/metrics/filters"
 	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
 	crwebhook "sigs.k8s.io/controller-runtime/pkg/webhook"
 
 	ocv1 "github.com/operator-framework/operator-controller/api/v1"
-	corecontrollers "github.com/operator-framework/operator-controller/internal/catalogd/controllers/core"
+	"github.com/operator-framework/operator-controller/internal/catalogd/controllers"
 	"github.com/operator-framework/operator-controller/internal/catalogd/features"
 	"github.com/operator-framework/operator-controller/internal/catalogd/garbagecollection"
-	catalogdmetrics "github.com/operator-framework/operator-controller/internal/catalogd/metrics"
-	"github.com/operator-framework/operator-controller/internal/catalogd/serverutil"
+	"github.com/operator-framework/operator-controller/internal/catalogd/handler"
+	v1 "github.com/operator-framework/operator-controller/internal/catalogd/handler/api/v1"
 	"github.com/operator-framework/operator-controller/internal/catalogd/storage"
 	"github.com/operator-framework/operator-controller/internal/catalogd/webhook"
 	sharedcontrollers "github.com/operator-framework/operator-controller/internal/shared/controllers"
 	fsutil "github.com/operator-framework/operator-controller/internal/shared/util/fs"
+	http2 "github.com/operator-framework/operator-controller/internal/shared/util/http"
 	imageutil "github.com/operator-framework/operator-controller/internal/shared/util/image"
 	"github.com/operator-framework/operator-controller/internal/shared/util/pullsecretcache"
 	sautil "github.com/operator-framework/operator-controller/internal/shared/util/sa"
@@ -326,47 +327,60 @@ func run(ctx context.Context) error {
 		},
 	}
 
-	var localStorage storage.Instance
-	metrics.Registry.MustRegister(catalogdmetrics.RequestDurationMetric)
-
 	storeDir := filepath.Join(cfg.cacheDir, storageDir)
 	if err := os.MkdirAll(storeDir, 0700); err != nil {
 		setupLog.Error(err, "unable to create storage directory for catalogs")
 		return err
 	}
 
-	baseStorageURL, err := url.Parse(fmt.Sprintf("%s/catalogs/", cfg.externalAddr))
+	const catalogsSubPath = "catalogs"
+	baseCatalogsURL, err := url.Parse(fmt.Sprintf("%s/%s", cfg.externalAddr, catalogsSubPath))
 	if err != nil {
 		setupLog.Error(err, "unable to create base storage URL")
 		return err
 	}
 
-	localStorage = &storage.LocalDirV1{
-		RootDir:            storeDir,
-		RootURL:            baseStorageURL,
-		EnableMetasHandler: features.CatalogdFeatureGate.Enabled(features.APIV1MetasHandler),
-	}
+	storageInstances := configureStorage(storeDir)
+	catalogdHandler := handler.NewStandardHandler(
+		newAPIV1Handler(catalogsSubPath, storageInstances),
+	)
 
 	// Config for the catalogd web server
-	catalogServerConfig := serverutil.CatalogServerConfig{
-		ExternalAddr: cfg.externalAddr,
-		CatalogAddr:  cfg.catalogServerAddr,
-		CertFile:     cfg.certFile,
-		KeyFile:      cfg.keyFile,
-		LocalStorage: localStorage,
+	catalogServerConfig := http2.ServerConfig{
+		Name:                "catalogs",
+		OnlyServeWhenLeader: true,
+		ListenAddr:          cfg.catalogServerAddr,
+		Server: &http.Server{
+			Handler:      catalogdHandler,
+			ReadTimeout:  5 * time.Second,
+			WriteTimeout: 5 * time.Minute,
+		},
+	}
+	if cfg.certFile != "" && cfg.keyFile != "" {
+		catalogServerConfig.TLSConfig = &tls.Config{
+			GetCertificate: cw.GetCertificate,
+			MinVersion:     tls.VersionTLS12,
+		}
 	}
 
-	err = serverutil.AddCatalogServerToManager(mgr, catalogServerConfig, cw)
+	catalogServer, err := http2.NewManagerServer(catalogServerConfig)
 	if err != nil {
 		setupLog.Error(err, "unable to configure catalog server")
 		return err
 	}
+	if err := mgr.Add(catalogServer); err != nil {
+		setupLog.Error(err, "unable to add catalog server to manager")
+		return err
+	}
 
-	if err = (&corecontrollers.ClusterCatalogReconciler{
+	if err = (&controllers.ClusterCatalogReconciler{
 		Client:      mgr.GetClient(),
 		ImageCache:  imageCache,
 		ImagePuller: imagePuller,
-		Storage:     localStorage,
+		Storage:     storageInstances,
+		GetBaseURL: func(catalogName string) string {
+			return fmt.Sprintf("%s/%s", baseCatalogsURL, catalogName)
+		},
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "ClusterCatalog")
 		return err
@@ -436,3 +450,25 @@ func podNamespace() string {
 	}
 	return string(namespace)
 }
+
+func configureStorage(storeDir string) *storage.Instances {
+	metasEnabled := features.CatalogdFeatureGate.Enabled(features.APIV1MetasHandler)
+	graphqlEnabled := features.CatalogdFeatureGate.Enabled(features.APIV1GraphQLHandler)
+
+	return storage.NewInstances(
+		storage.WithFiles(true, storeDir),
+		storage.WithIndices(metasEnabled || graphqlEnabled, storeDir),
+		storage.WithGraphQLSchemas(graphqlEnabled),
+	)
+}
+
+func newAPIV1Handler(baseURLPath string, si *storage.Instances) *v1.APIV1Handler {
+	metasEnabled := features.CatalogdFeatureGate.Enabled(features.APIV1MetasHandler)
+	graphqlEnabled := features.CatalogdFeatureGate.Enabled(features.APIV1GraphQLHandler)
+
+	return v1.NewAPIV1Handler(baseURLPath, si,
+		v1.WithAllHandler(true),
+		v1.WithMetasHandler(metasEnabled),
+		v1.WithGraphQLHandler(graphqlEnabled),
+	)
+}

config/components/base/experimental/kustomization.yaml

@@ -12,6 +12,7 @@ components:
 - ../../features/single-own-namespace
 - ../../features/preflight-permissions
 - ../../features/apiv1-metas-handler
+- ../../features/apiv1-graphql-handler
 - ../../features/helm-chart
 # This one is downstream only, so we shant use it
 # - ../../features/webhook-provider-openshift-serviceca

config/components/features/apiv1-graphql-handler/kustomization.yaml

@@ -0,0 +1,9 @@
+# kustomization file for catalogd APIv1 metas handler
+# DO NOT ADD A NAMESPACE HERE
+apiVersion: kustomize.config.k8s.io/v1alpha1
+kind: Component
+patches:
+ - target:
+      kind: Deployment
+      name: catalogd-controller-manager
+   path: patches/enable-featuregate.yaml

config/components/features/apiv1-graphql-handler/patches/enable-featuregate.yaml

@@ -0,0 +1,4 @@
+# enable APIv1 meta handler feature gate
+- op: add
+  path: /spec/template/spec/containers/0/args/-
+  value: "--feature-gates=APIV1GraphQLHandler=true"

config/overlays/tilt-local-dev/patches/catalogd.yaml

@@ -7,4 +7,4 @@
   value: null
 - op: remove
   # remove --leader-elect so container doesn't restart during breakpoints
-  path: /spec/template/spec/containers/0/args/2
+  path: /spec/template/spec/containers/0/args/0

go.mod

@@ -16,6 +16,9 @@ require (
 	github.com/google/go-containerregistry v0.20.6
 	github.com/google/renameio/v2 v2.0.0
 	github.com/gorilla/handlers v1.5.2
+	github.com/graphql-go/graphql v0.8.1
+	github.com/graphql-go/handler v0.2.4
+	github.com/itchyny/gojq v0.12.17
 	github.com/klauspost/compress v1.18.0
 	github.com/opencontainers/go-digest v1.0.0
 	github.com/opencontainers/image-spec v1.1.1
@@ -28,6 +31,7 @@ require (
 	golang.org/x/exp v0.0.0-20250620022241-b7579e27df2b
 	golang.org/x/mod v0.26.0
 	golang.org/x/sync v0.16.0
+	golang.org/x/text v0.27.0
 	golang.org/x/tools v0.35.0
 	gopkg.in/yaml.v2 v2.4.0
 	helm.sh/helm/v3 v3.18.4
@@ -133,6 +137,7 @@ require (
 	github.com/hashicorp/go-multierror v1.1.1 // indirect
 	github.com/huandu/xstrings v1.5.0 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
+	github.com/itchyny/timefmt-go v0.1.6 // indirect
 	github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99 // indirect
 	github.com/jmoiron/sqlx v1.4.0 // indirect
 	github.com/joelanford/ignore v0.1.1 // indirect
@@ -221,7 +226,6 @@ require (
 	golang.org/x/oauth2 v0.30.0 // indirect
 	golang.org/x/sys v0.34.0 // indirect
 	golang.org/x/term v0.33.0 // indirect
-	golang.org/x/text v0.27.0 // indirect
 	golang.org/x/time v0.12.0 // indirect
 	golang.org/x/tools/go/packages/packagestest v0.1.1-deprecated // indirect
 	gomodules.xyz/jsonpatch/v2 v2.4.0 // indirect

go.sum

@@ -246,6 +246,10 @@ github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 h1:JeSE6pjso5T
 github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674/go.mod h1:r4w70xmWCQKmi1ONH4KIaBptdivuRPyosB9RmPlGEwA=
 github.com/gosuri/uitable v0.0.4 h1:IG2xLKRvErL3uhY6e1BylFzG+aJiwQviDDTfOKeKTpY=
 github.com/gosuri/uitable v0.0.4/go.mod h1:tKR86bXuXPZazfOTG1FIzvjIdXzd0mo4Vtn16vt0PJo=
+github.com/graphql-go/graphql v0.8.1 h1:p7/Ou/WpmulocJeEx7wjQy611rtXGQaAcXGqanuMMgc=
+github.com/graphql-go/graphql v0.8.1/go.mod h1:nKiHzRM0qopJEwCITUuIsxk9PlVlwIiiI8pnJEhordQ=
+github.com/graphql-go/handler v0.2.4 h1:gz9q11TUHPNUpqzV8LMa+rkqM5NUuH/nkE3oF2LS3rI=
+github.com/graphql-go/handler v0.2.4/go.mod h1:gsQlb4gDvURR0bgN8vWQEh+s5vJALM2lYL3n3cf6OxQ=
 github.com/gregjones/httpcache v0.0.0-20190611155906-901d90724c79 h1:+ngKgrYPPJrOjhax5N+uePQ0Fh1Z7PheYoUI/0nzkPA=
 github.com/gregjones/httpcache v0.0.0-20190611155906-901d90724c79/go.mod h1:FecbI9+v66THATjSRHfNgh1IVFe/9kFxbXtjV0ctIMA=
 github.com/grpc-ecosystem/go-grpc-prometheus v1.2.1-0.20210315223345-82c243799c99 h1:JYghRBlGCZyCF2wNUJ8W0cwaQdtpcssJ4CgC406g+WU=
@@ -269,6 +273,10 @@ github.com/huandu/xstrings v1.5.0 h1:2ag3IFq9ZDANvthTwTiqSSZLjDc+BedvHPAp5tJy2TI
 github.com/huandu/xstrings v1.5.0/go.mod h1:y5/lhBue+AyNmUVz9RLU9xbLR0o4KIIExikq4ovT0aE=
 github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
 github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
+github.com/itchyny/gojq v0.12.17 h1:8av8eGduDb5+rvEdaOO+zQUjA04MS0m3Ps8HiD+fceg=
+github.com/itchyny/gojq v0.12.17/go.mod h1:WBrEMkgAfAGO1LUcGOckBl5O726KPp+OlkKug0I/FEY=
+github.com/itchyny/timefmt-go v0.1.6 h1:ia3s54iciXDdzWzwaVKXZPbiXzxxnv1SPGFfM/myJ5Q=
+github.com/itchyny/timefmt-go v0.1.6/go.mod h1:RRDZYC5s9ErkjQvTvvU7keJjxUYzIISJGxm9/mAERQg=
 github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99 h1:BQSFePA1RWJOlocH6Fxy8MmwDt+yVQYULKfN0RoTN8A=
 github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99/go.mod h1:1lJo3i6rXxKeerYnT8Nvf0QmHCRC1n8sfWVwXF2Frvo=
 github.com/jmhodges/clock v1.2.0 h1:eq4kys+NI0PLngzaHEe7AmPT90XMGIEySD1JfV1PDIs=

internal/catalogd/controllers/core/clustercatalog_controller.go → internal/catalogd/controllers/clustercatalog_controller.go

@@ -14,12 +14,14 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package core
+package controllers
 
 import (
 	"context" // #nosec
 	"errors"
 	"fmt"
+	"io/fs"
+	"iter"
 	"slices"
 	"sync"
 	"time"
@@ -38,6 +40,8 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
+	"github.com/operator-framework/operator-registry/alpha/declcfg"
+
 	ocv1 "github.com/operator-framework/operator-controller/api/v1"
 	"github.com/operator-framework/operator-controller/internal/catalogd/storage"
 	imageutil "github.com/operator-framework/operator-controller/internal/shared/util/image"
@@ -57,7 +61,8 @@ type ClusterCatalogReconciler struct {
 	ImageCache  imageutil.Cache
 	ImagePuller imageutil.Puller
 
-	Storage storage.Instance
+	Storage    storage.Instance
+	GetBaseURL func(catalogName string) string
 
 	finalizers crfinalizer.Finalizers
 
@@ -224,7 +229,7 @@ func (r *ClusterCatalogReconciler) reconcile(ctx context.Context, catalog *ocv1.
 	case !hasStoredCatalog:
 		l.Info("unpack required: no cached catalog metadata found for this catalog")
 		needsUnpack = true
-	case !r.Storage.ContentExists(catalog.Name):
+	case !r.Storage.Exists(catalog.Name):
 		l.Info("unpack required: no stored content found for this catalog")
 		needsUnpack = true
 	case !equality.Semantic.DeepEqual(catalog.Status, *expectedStatus):
@@ -265,12 +270,12 @@ func (r *ClusterCatalogReconciler) reconcile(ctx context.Context, catalog *ocv1.
 	// TODO: We should check to see if the unpacked result has the same content
 	//   as the already unpacked content. If it does, we should skip this rest
 	//   of the unpacking steps.
-	if err := r.Storage.Store(ctx, catalog.Name, fsys); err != nil {
+	if err := r.Storage.Store(ctx, catalog.Name, walkMetasFSIterator(ctx, fsys)); err != nil {
 		storageErr := fmt.Errorf("error storing fbc: %v", err)
 		updateStatusProgressing(&catalog.Status, catalog.GetGeneration(), storageErr)
 		return ctrl.Result{}, storageErr
 	}
-	baseURL := r.Storage.BaseURL(catalog.Name)
+	baseURL := r.GetBaseURL(catalog.Name)
 
 	updateStatusProgressing(&catalog.Status, catalog.GetGeneration(), nil)
 	updateStatusServing(&catalog.Status, canonicalRef, unpackTime, baseURL, catalog.GetGeneration())
@@ -296,8 +301,8 @@ func (r *ClusterCatalogReconciler) getCurrentState(catalog *ocv1.ClusterCatalog)
 
 	// Set expected status based on what we see in the stored catalog
 	clearUnknownConditions(expectedStatus)
-	if hasStoredCatalog && r.Storage.ContentExists(catalog.Name) {
-		updateStatusServing(expectedStatus, storedCatalog.ref, storedCatalog.lastUnpack, r.Storage.BaseURL(catalog.Name), storedCatalog.observedGeneration)
+	if hasStoredCatalog && r.Storage.Exists(catalog.Name) {
+		updateStatusServing(expectedStatus, storedCatalog.ref, storedCatalog.lastUnpack, r.GetBaseURL(catalog.Name), storedCatalog.observedGeneration)
 		updateStatusProgressing(expectedStatus, storedCatalog.observedGeneration, nil)
 	}
 
@@ -458,7 +463,7 @@ func (r *ClusterCatalogReconciler) deleteStoredCatalog(catalogName string) {
 }
 
 func (r *ClusterCatalogReconciler) deleteCatalogCache(ctx context.Context, catalog *ocv1.ClusterCatalog) error {
-	if err := r.Storage.Delete(catalog.Name); err != nil {
+	if err := r.Storage.Delete(ctx, catalog.Name); err != nil {
 		updateStatusProgressing(&catalog.Status, catalog.GetGeneration(), err)
 		return err
 	}
@@ -470,3 +475,12 @@ func (r *ClusterCatalogReconciler) deleteCatalogCache(ctx context.Context, catal
 	r.deleteStoredCatalog(catalog.Name)
 	return nil
 }
+
+func walkMetasFSIterator(ctx context.Context, fsys fs.FS) iter.Seq2[*declcfg.Meta, error] {
+	return func(yield func(*declcfg.Meta, error) bool) {
+		_ = declcfg.WalkMetasFS(ctx, fsys, func(path string, meta *declcfg.Meta, err error) error {
+			yield(meta, err)
+			return nil
+		}, declcfg.WithConcurrency(1))
+	}
+}

internal/catalogd/controllers/core/clustercatalog_controller_test.go → internal/catalogd/controllers/clustercatalog_controller_test.go

@@ -1,11 +1,10 @@
-package core
+package controllers
 
 import (
 	"context"
 	"errors"
 	"fmt"
-	"io/fs"
-	"net/http"
+	"iter"
 	"testing"
 	"testing/fstest"
 	"time"
@@ -21,6 +20,8 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
+	"github.com/operator-framework/operator-registry/alpha/declcfg"
+
 	ocv1 "github.com/operator-framework/operator-controller/api/v1"
 	"github.com/operator-framework/operator-controller/internal/catalogd/storage"
 	imageutil "github.com/operator-framework/operator-controller/internal/shared/util/image"
@@ -32,29 +33,21 @@ type MockStore struct {
 	shouldError bool
 }
 
-func (m MockStore) Store(_ context.Context, _ string, _ fs.FS) error {
+func (m MockStore) Store(_ context.Context, _ string, _ iter.Seq2[*declcfg.Meta, error]) error {
 	if m.shouldError {
 		return errors.New("mockstore store error")
 	}
 	return nil
 }
 
-func (m MockStore) Delete(_ string) error {
+func (m MockStore) Delete(_ context.Context, _ string) error {
 	if m.shouldError {
 		return errors.New("mockstore delete error")
 	}
 	return nil
 }
 
-func (m MockStore) BaseURL(_ string) string {
-	return "URL"
-}
-
-func (m MockStore) StorageServerHandler() http.Handler {
-	panic("not needed")
-}
-
-func (m MockStore) ContentExists(_ string) bool {
+func (m MockStore) Exists(_ string) bool {
 	return true
 }
 
@@ -807,6 +800,7 @@ func TestCatalogdControllerReconcile(t *testing.T) {
 				ImagePuller:    tt.puller,
 				ImageCache:     tt.cache,
 				Storage:        tt.store,
+				GetBaseURL:     func(catalogName string) string { return "URL" },
 				storedCatalogs: map[string]storedCatalogData{},
 			}
 			if reconciler.ImageCache == nil {
@@ -915,7 +909,8 @@ func TestPollingRequeue(t *testing.T) {
 					ImageFS: &fstest.MapFS{},
 					Ref:     ref,
 				},
-				Storage: &MockStore{},
+				Storage:    &MockStore{},
+				GetBaseURL: func(catalogName string) string { return "URL" },
 				storedCatalogs: map[string]storedCatalogData{
 					tc.catalog.Name: {
 						ref:                ref,
@@ -1140,6 +1135,7 @@ func TestPollingReconcilerUnpack(t *testing.T) {
 				Client:         nil,
 				ImagePuller:    &imageutil.MockPuller{Error: errors.New("mockpuller error")},
 				Storage:        &MockStore{},
+				GetBaseURL:     func(catalogName string) string { return "URL" },
 				storedCatalogs: scd,
 			}
 			require.NoError(t, reconciler.setupFinalizers())