Query api

catalogd/cmd/catalogd/main.go

@@ -294,9 +294,13 @@ func main() {
 		os.Exit(1)
 	}
 
-	localStorage = storage.LocalDirV1{RootDir: storeDir, RootURL: baseStorageURL}
+	localStorage = &storage.LocalDirV1{
+		RootDir:            storeDir,
+		RootURL:            baseStorageURL,
+		EnableQueryHandler: features.CatalogdFeatureGate.Enabled(features.APIV1QueryHandler),
+	}
 
-	// Config for the the catalogd web server
+	// Config for the catalogd web server
 	catalogServerConfig := serverutil.CatalogServerConfig{
 		ExternalAddr: externalAddr,
 		CatalogAddr:  catalogServerAddr,

catalogd/internal/features/features.go

@@ -5,7 +5,13 @@ import (
 	"k8s.io/component-base/featuregate"
 )
 
-var catalogdFeatureGates = map[featuregate.Feature]featuregate.FeatureSpec{}
+const (
+	APIV1QueryHandler = featuregate.Feature("APIV1QueryHandler")
+)
+
+var catalogdFeatureGates = map[featuregate.Feature]featuregate.FeatureSpec{
+	APIV1QueryHandler: {Default: false, PreRelease: featuregate.Alpha},
+}
 
 var CatalogdFeatureGate featuregate.MutableFeatureGate = featuregate.NewFeatureGate()
 

catalogd/internal/serverutil/serverutil.go

@@ -1,15 +1,18 @@
 package serverutil
 
 import (
+	"context"
 	"crypto/tls"
 	"fmt"
 	"net"
 	"net/http"
 	"time"
 
+	"github.com/go-logr/logr"
+	"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/log"
 
 	catalogdmetrics "github.com/operator-framework/operator-controller/catalogd/internal/metrics"
 	"github.com/operator-framework/operator-controller/catalogd/internal/storage"
@@ -39,13 +42,20 @@ func AddCatalogServerToManager(mgr ctrl.Manager, cfg CatalogServerConfig, tlsFil
 	}
 
 	shutdownTimeout := 30 * time.Second
+	handler := cfg.LocalStorage.StorageServerHandler()
+	handler = gzhttp.GzipHandler(handler)
+	handler = catalogdmetrics.AddMetricsToHandler(handler)
+	handler = newLoggingMiddleware(handler)
 
 	catalogServer := manager.Server{
 		Name:                "catalogs",
 		OnlyServeWhenLeader: true,
 		Server: &http.Server{
-			Addr:        cfg.CatalogAddr,
-			Handler:     catalogdmetrics.AddMetricsToHandler(cfg.LocalStorage.StorageServerHandler()),
+			Addr:    cfg.CatalogAddr,
+			Handler: handler,
+			BaseContext: func(_ net.Listener) context.Context {
+				return log.IntoContext(context.Background(), mgr.GetLogger().WithName("http.catalogs"))
+			},
 			ReadTimeout: 5 * time.Second,
 			// TODO: Revert this to 10 seconds if/when the API
 			// evolves to have significantly smaller responses
@@ -62,3 +72,31 @@ func AddCatalogServerToManager(mgr ctrl.Manager, cfg CatalogServerConfig, tlsFil
 
 	return nil
 }
+
+func newLoggingMiddleware(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		logger := logr.FromContextOrDiscard(r.Context())
+
+		start := time.Now()
+		lrw := &loggingResponseWriter{ResponseWriter: w, statusCode: http.StatusOK}
+		next.ServeHTTP(lrw, r)
+
+		logger.WithValues(
+			"method", r.Method,
+			"url", r.URL.String(),
+			"status", lrw.statusCode,
+			"duration", time.Since(start),
+			"remoteAddr", r.RemoteAddr,
+		).Info("HTTP request processed")
+	})
+}
+
+type loggingResponseWriter struct {
+	http.ResponseWriter
+	statusCode int
+}
+
+func (w *loggingResponseWriter) WriteHeader(code int) {
+	w.statusCode = code
+	w.ResponseWriter.WriteHeader(code)
+}

catalogd/internal/storage/index.go

@@ -0,0 +1,133 @@
+package storage
+
+import (
+	"cmp"
+	"encoding/json"
+	"fmt"
+	"io"
+	"slices"
+
+	"k8s.io/apimachinery/pkg/util/sets"
+
+	"github.com/operator-framework/operator-registry/alpha/declcfg"
+)
+
+type index struct {
+	BySchema  map[string][]section `json:"by_schema"`
+	ByPackage map[string][]section `json:"by_package"`
+	ByName    map[string][]section `json:"by_name"`
+}
+
+type section struct {
+	offset int64
+	length int64
+}
+
+func (s *section) MarshalJSON() ([]byte, error) {
+	return []byte(fmt.Sprintf(`[%d,%d]`, s.offset, s.length)), nil
+}
+
+func (s *section) UnmarshalJSON(b []byte) error {
+	vals := [2]int64{}
+	if err := json.Unmarshal(b, &vals); err != nil {
+		return err
+	}
+	s.offset = vals[0]
+	s.length = vals[1]
+	return nil
+}
+
+func (i index) Size() int64 {
+	size := 0
+	for k, v := range i.BySchema {
+		size += len(k) + len(v)*16
+	}
+	for k, v := range i.ByPackage {
+		size += len(k) + len(v)*16
+	}
+	for k, v := range i.ByName {
+		size += len(k) + len(v)*16
+	}
+	return int64(size)
+}
+
+func (i index) Get(r io.ReaderAt, schema, packageName, name string) (io.ReadSeeker, bool) {
+	sectionSet := i.getSectionSet(schema, packageName, name)
+
+	sections := sectionSet.UnsortedList()
+	slices.SortFunc(sections, func(a, b section) int {
+		return cmp.Compare(a.offset, b.offset)
+	})
+
+	srs := make([]io.ReadSeeker, 0, len(sections))
+	for _, s := range sections {
+		sr := io.NewSectionReader(r, s.offset, s.length)
+		srs = append(srs, sr)
+	}
+	return newMultiReadSeeker(srs...), true
+}
+
+func (i *index) getSectionSet(schema, packageName, name string) sets.Set[section] {
+	if schema == "" {
+		if packageName == "" {
+			if name == "" {
+				sectionSet := sets.New[section]()
+				for _, s := range i.BySchema {
+					sectionSet.Insert(s...)
+				}
+				return sectionSet
+			} else {
+				return sets.New[section](i.ByName[name]...)
+			}
+		} else {
+			sectionSet := sets.New[section](i.ByPackage[packageName]...)
+			if name == "" {
+				return sectionSet
+			} else {
+				return sectionSet.Intersection(sets.New[section](i.ByName[name]...))
+			}
+		}
+	} else {
+		sectionSet := sets.New[section](i.BySchema[schema]...)
+		if packageName == "" {
+			if name == "" {
+				return sectionSet
+			} else {
+				return sectionSet.Intersection(sets.New[section](i.ByName[name]...))
+			}
+		} else {
+			sectionSet = sectionSet.Intersection(sets.New[section](i.ByPackage[packageName]...))
+			if name == "" {
+				return sectionSet
+			} else {
+				return sectionSet.Intersection(sets.New[section](i.ByName[name]...))
+			}
+		}
+	}
+}
+
+func newIndex(metasChan <-chan *declcfg.Meta) (*index, error) {
+	idx := &index{
+		BySchema:  make(map[string][]section),
+		ByPackage: make(map[string][]section),
+		ByName:    make(map[string][]section),
+	}
+	offset := int64(0)
+	for meta := range metasChan {
+		start := offset
+		length := int64(len(meta.Blob))
+		offset += length
+
+		s := section{offset: start, length: length}
+		if meta.Schema != "" {
+			idx.BySchema[meta.Schema] = append(idx.BySchema[meta.Schema], s)
+		}
+		if meta.Package != "" {
+			idx.ByPackage[meta.Package] = append(idx.ByPackage[meta.Package], s)
+		}
+		if meta.Name != "" {
+			idx.ByName[meta.Name] = append(idx.ByName[meta.Name], s)
+		}
+	}
+	return idx, nil
+}