larger refactor of catalogd's structure

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

Author: joelanford

cmd/catalogd/main.go

@@ -46,17 +46,16 @@ 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"
+	v1 "github.com/operator-framework/operator-controller/internal/catalogd/handlers/api/v1"
+	"github.com/operator-framework/operator-controller/internal/catalogd/handlers/middleware"
 	"github.com/operator-framework/operator-controller/internal/catalogd/serverutil"
 	"github.com/operator-framework/operator-controller/internal/catalogd/storage"
 	"github.com/operator-framework/operator-controller/internal/catalogd/webhook"
@@ -328,61 +327,55 @@ 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)
+	handler := configureHandler(configAPIV1Handler(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 := serverutil.ServerConfig{
+		Name:                "catalogs",
+		OnlyServeWhenLeader: true,
+		ListenAddr:          cfg.catalogServerAddr,
+		Server: &http.Server{
+			Handler:      handler,
+			ReadTimeout:  5 * time.Second,
+			WriteTimeout: 5 * time.Minute,
+		},
+	}
+	if cfg.certFile != "" && cfg.keyFile != "" {
+		catalogServerConfig.GetCertificate = cw.GetCertificate
 	}
 
-	err = serverutil.AddCatalogServerToManager(mgr, catalogServerConfig, cw)
+	catalogServer, err := serverutil.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 +445,54 @@ 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 configAPIV1Handler(baseURLPath string, si *storage.Instances) subPathedHandler {
+	metasEnabled := features.CatalogdFeatureGate.Enabled(features.APIV1MetasHandler)
+	graphqlEnabled := features.CatalogdFeatureGate.Enabled(features.APIV1GraphQLHandler)
+
+	// Setup API v1 handler
+	apiV1HandlerOpts := []v1.APIV1HandlerOption{}
+	apiV1HandlerOpts = append(apiV1HandlerOpts, v1.WithAllHandler(si.Files))
+
+	if metasEnabled {
+		apiV1HandlerOpts = append(apiV1HandlerOpts, v1.WithMetasHandler(si.Files, si.Indices))
+	}
+
+	if graphqlEnabled {
+		apiV1HandlerOpts = append(apiV1HandlerOpts, v1.WithGraphQLHandler(si.Files, si.Indices, si.GraphQLSchemas))
+	}
+
+	return v1.NewAPIV1Handler(baseURLPath, apiV1HandlerOpts...)
+}
+
+type subPathedHandler interface {
+	SubPath() string
+	http.Handler
+}
+
+func configureHandler(handlers ...subPathedHandler) http.Handler {
+	mux := http.NewServeMux()
+	for _, h := range handlers {
+		mux.Handle(h.SubPath(), h)
+	}
+	return middleware.Standard(mux)
+}

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))
+	}
+}

internal/catalogd/controllers/core/clustercatalog_controller_test.go renamed to internal/catalogd/controllers/clustercatalog_controller_test.go

@@ -1,4 +1,4 @@
-package core
+package controllers
 
 import (
 	"context"

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

@@ -1,35 +1,35 @@
-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/handlers/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
 		}
 
-		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)
 }

internal/catalogd/handlers/api/v1/graphql.go

@@ -0,0 +1,42 @@
+package v1
+
+import (
+	"net/http"
+
+	"github.com/graphql-go/handler"
+	"k8s.io/klog/v2"
+
+	"github.com/operator-framework/operator-controller/internal/catalogd/storage"
+)
+
+func apiV1GraphQLHandler(files *storage.Files, indices *storage.Indices, schemas *storage.GraphQLSchemas) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		catalog := r.PathValue("catalog")
+		logger := klog.FromContext(r.Context()).WithValues("catalog", catalog)
+
+		file, err := files.Get(catalog)
+		if err != nil {
+			logger.Error(err, "error getting catalog file")
+			http.Error(w, "Not found", http.StatusNotFound)
+		}
+		defer file.Close()
+
+		idx, err := indices.Get(catalog)
+		if err != nil {
+			logger.Error(err, "error getting catalog index")
+		}
+
+		schema, err := schemas.Get(catalog)
+		if err != nil {
+			logger.Error(err, "error getting catalog graphql schema")
+			http.Error(w, "Not found", http.StatusNotFound)
+		}
+
+		r = r.WithContext(storage.ContextWithCatalogData(r.Context(), file, idx))
+		h := handler.New(&handler.Config{
+			Schema:   schema,
+			GraphiQL: true,
+		})
+		h.ServeHTTP(w, r)
+	})
+}