larger refactor of catalogd's structure

Signed-off-by: Joe Lanford <[email protected]>

Author: joelanford

cmd/catalogd/main.go

@@ -46,22 +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"
-	"github.com/operator-framework/operator-controller/internal/catalogd/handlers"
-	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"
@@ -328,61 +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
 	}
 
-	indexer := storage.NewIndexer()
-	handlersMap := map[string]http.Handler{
-		"/all": handlers.V1AllHandler(indexer),
-	}
-
-	if features.CatalogdFeatureGate.Enabled(features.APIV1MetasHandler) {
-		handlersMap["/metas"] = handlers.V1MetasHandler(indexer)
-	}
-
-	if features.CatalogdFeatureGate.Enabled(features.APIV1GraphQLHandler) {
-		handlersMap["/graphql"] = handlers.V1GraphQLHandler()
-	}
-
-	localStorage = &storage.LocalDirV1{
-		Indexer:  indexer,
-		Handlers: handlersMap,
-		RootDir:  storeDir,
-		RootURL:  baseStorageURL,
-	}
+	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
@@ -452,3 +450,33 @@ func podNamespace() string {
 	}
 	return string(namespace)
 }
+
+func configureStorage(storeDir string) *storage.Instances {
+	metasEnabled := features.CatalogdFeatureGate.Enabled(features.APIV1MetasHandler)
+	graphqlEnabled := features.CatalogdFeatureGate.Enabled(features.APIV1GraphQLHandler)
+	needsIndices := metasEnabled || graphqlEnabled
+
+	// Setup storage instances
+	storageInstances := storage.Instances{}
+	storageInstances.Files = storage.NewFiles(storeDir)
+
+	if needsIndices {
+		storageInstances.Indices = storage.NewIndices(storeDir)
+	}
+	if graphqlEnabled {
+		storageInstances.GraphQLSchemas = storage.NewGraphQLSchemas()
+	}
+
+	return &storageInstances
+}
+
+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),
+	)
+}

…ollers/core/clustercatalog_controller.go …controllers/clustercatalog_controller.gointernal/catalogd/controllers/core/clustercatalog_controller.go renamed to internal/catalogd/controllers/clustercatalog_controller.go internal/catalogd/controllers/core/clustercatalog_controller.go renamed to 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))
+	}
+}

…s/core/clustercatalog_controller_test.go …ollers/clustercatalog_controller_test.gointernal/catalogd/controllers/core/clustercatalog_controller_test.go renamed to internal/catalogd/controllers/clustercatalog_controller_test.go internal/catalogd/controllers/core/clustercatalog_controller_test.go renamed to 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())

internal/catalogd/handlers/all.go internal/catalogd/handler/api/v1/all.gointernal/catalogd/handlers/all.go renamed to internal/catalogd/handler/api/v1/all.go

@@ -1,35 +1,36 @@
-package handlers
+package v1
 
 import (
-	"github.com/operator-framework/operator-controller/internal/catalogd/handlers/internal/handlerutil"
 	"net/http"
 
 	"k8s.io/klog/v2"
 
+	"github.com/operator-framework/operator-controller/internal/catalogd/handler/internal/handlerutil"
 	"github.com/operator-framework/operator-controller/internal/catalogd/storage"
 )
 
-func V1AllHandler(indexer *storage.Indexer) http.Handler {
-	return handlerutil.AllowedMethodsHandler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+func apiV1AllHandler(files *storage.Files) http.Handler {
+	allHandler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		catalog := r.PathValue("catalog")
 		logger := klog.FromContext(r.Context()).WithValues("catalog", catalog)
 
-		idx, err := indexer.GetIndex(catalog)
+		catalogFile, err := files.Get(catalog)
 		if err != nil {
-			logger.Error(err, "error getting index")
+			logger.Error(err, "error getting catalog file")
 			http.Error(w, "Not found", http.StatusNotFound)
 			return
 		}
+		defer catalogFile.Close()
 
-		catalogFile := idx.All()
-		catalogStat, err := idx.Stat()
+		catalogStat, err := catalogFile.Stat()
 		if err != nil {
-			logger.Error(err, "error stat-ing index")
+			logger.Error(err, "error stat-ing catalog file")
 			http.Error(w, "Internal server error", http.StatusInternalServerError)
 			return
 		}
 
 		w.Header().Add("Content-Type", "application/jsonl")
 		http.ServeContent(w, r, "", catalogStat.ModTime(), catalogFile)
-	}), http.MethodGet, http.MethodHead)
+	})
+	return handlerutil.AllowedMethodsHandler(allHandler, http.MethodGet, http.MethodHead)
 }