⚠ (feat) Introduce new feature-gated query endpoint (#1643)

* add a new feature-gated api/v1/query endpoint

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

* add singleflight for shared index access for concurrent requests

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

* a few improvements and optimizations

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

* another round of refactoring improvement

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

* Tests

* use io.MultiReader instead of manual implementation

* include checkPrecoditions check

* add comments

* code cleanup

* cleanup index.Get's signature

* only allow GET/HEAD methods

* refractor serverJSONLines

* replace static test variable

* fix unit test

* refractor getIndex() (and other lint issues)

---------

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

Author: anik120

catalogd/cmd/catalogd/main.go

@@ -303,9 +303,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

@@ -10,6 +10,7 @@ import (
 
 	"github.com/go-logr/logr"
 	"github.com/gorilla/handlers"
+	"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"
@@ -42,17 +43,12 @@ func AddCatalogServerToManager(mgr ctrl.Manager, cfg CatalogServerConfig, tlsFil
 	}
 
 	shutdownTimeout := 30 * time.Second
-
-	l := mgr.GetLogger().WithName("catalogd-http-server")
-	handler := catalogdmetrics.AddMetricsToHandler(cfg.LocalStorage.StorageServerHandler())
-	handler = logrLoggingHandler(l, handler)
-
 	catalogServer := manager.Server{
 		Name:                "catalogs",
 		OnlyServeWhenLeader: true,
 		Server: &http.Server{
 			Addr:        cfg.CatalogAddr,
-			Handler:     handler,
+			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
@@ -97,3 +93,12 @@ func logrLoggingHandler(l logr.Logger, handler http.Handler) http.Handler {
 		l.Info("handled request", "host", host, "username", username, "method", params.Request.Method, "uri", uri, "protocol", params.Request.Proto, "status", params.StatusCode, "size", params.Size)
 	})
 }
+
+func storageServerHandlerWrapped(l logr.Logger, cfg CatalogServerConfig) http.Handler {
+	handler := cfg.LocalStorage.StorageServerHandler()
+	handler = gzhttp.GzipHandler(handler)
+	handler = catalogdmetrics.AddMetricsToHandler(handler)
+
+	handler = logrLoggingHandler(l, handler)
+	return handler
+}

catalogd/internal/serverutil/serverutil_test.go

@@ -0,0 +1,128 @@
+package serverutil
+
+import (
+	"compress/gzip"
+	"context"
+	"io"
+	"io/fs"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/go-logr/logr"
+	"github.com/stretchr/testify/require"
+)
+
+func TestStorageServerHandlerWrapped_Gzip(t *testing.T) {
+	var generatedJSON = func(size int) string {
+		return "{\"data\":\"" + strings.Repeat("test data ", size) + "\"}"
+	}
+	tests := []struct {
+		name             string
+		acceptEncoding   string
+		responseContent  string
+		expectCompressed bool
+		expectedStatus   int
+	}{
+		{
+			name:             "compresses large response when client accepts gzip",
+			acceptEncoding:   "gzip",
+			responseContent:  generatedJSON(1000),
+			expectCompressed: true,
+			expectedStatus:   http.StatusOK,
+		},
+		{
+			name:             "does not compress small response even when client accepts gzip",
+			acceptEncoding:   "gzip",
+			responseContent:  `{"foo":"bar"}`,
+			expectCompressed: false,
+			expectedStatus:   http.StatusOK,
+		},
+		{
+			name:             "does not compress when client doesn't accept gzip",
+			acceptEncoding:   "",
+			responseContent:  generatedJSON(1000),
+			expectCompressed: false,
+			expectedStatus:   http.StatusOK,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Create a mock storage instance that returns our test content
+			mockStorage := &mockStorageInstance{
+				content: tt.responseContent,
+			}
+
+			cfg := CatalogServerConfig{
+				LocalStorage: mockStorage,
+			}
+			handler := storageServerHandlerWrapped(logr.Logger{}, cfg)
+
+			// Create test request
+			req := httptest.NewRequest("GET", "/test", nil)
+			if tt.acceptEncoding != "" {
+				req.Header.Set("Accept-Encoding", tt.acceptEncoding)
+			}
+
+			// Create response recorder
+			rec := httptest.NewRecorder()
+
+			// Handle the request
+			handler.ServeHTTP(rec, req)
+
+			// Check status code
+			require.Equal(t, tt.expectedStatus, rec.Code)
+
+			// Check if response was compressed
+			wasCompressed := rec.Header().Get("Content-Encoding") == "gzip"
+			require.Equal(t, tt.expectCompressed, wasCompressed)
+
+			// Get the response body
+			var responseBody []byte
+			if wasCompressed {
+				// Decompress the response
+				gzipReader, err := gzip.NewReader(rec.Body)
+				require.NoError(t, err)
+				responseBody, err = io.ReadAll(gzipReader)
+				require.NoError(t, err)
+				require.NoError(t, gzipReader.Close())
+			} else {
+				responseBody = rec.Body.Bytes()
+			}
+
+			// Verify the response content
+			require.Equal(t, tt.responseContent, string(responseBody))
+		})
+	}
+}
+
+// mockStorageInstance implements storage.Instance interface for testing
+type mockStorageInstance struct {
+	content string
+}
+
+func (m *mockStorageInstance) StorageServerHandler() http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, err := w.Write([]byte(m.content))
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+		}
+	})
+}
+
+func (m *mockStorageInstance) Store(ctx context.Context, catalogName string, fs fs.FS) error {
+	return nil
+}
+
+func (m *mockStorageInstance) Delete(catalogName string) error {
+	return nil
+}
+
+func (m *mockStorageInstance) ContentExists(catalog string) bool {
+	return true
+}
+func (m *mockStorageInstance) BaseURL(catalog string) string {
+	return ""
+}