scale-out: support for GQL scale-out with local metadata

Signed-off-by: Vishwas Rajashekar <[email protected]>

Author: vrajashkr

examples/scale-out-cluster-local/config-cluster-member0.json

@@ -0,0 +1,31 @@
+{
+  "distSpecVersion": "1.1.0",
+  "storage": {
+    "rootDirectory": "./workspace/zot/data/mem1",
+    "dedupe": false
+  },
+  "http": {
+    "address": "127.0.0.1",
+    "port": "9000"
+  },
+  "log": {
+    "level": "debug"
+  },
+  "cluster": {
+    "members": [
+      "127.0.0.1:9000",
+      "127.0.0.1:9001"
+    ],
+    "hashKey": "loremipsumdolors"
+  },
+  "extensions": {
+    "search": {
+      "cve": {
+        "updateInterval": "2h"
+      }
+    },
+    "ui": {
+      "enable": true
+    }
+  }
+}

examples/scale-out-cluster-local/config-cluster-member1.json

@@ -0,0 +1,31 @@
+{
+  "distSpecVersion": "1.1.0",
+  "storage": {
+    "rootDirectory": "./workspace/zot/data/mem2",
+    "dedupe": false
+  },
+  "http": {
+    "address": "127.0.0.1",
+    "port": "9001"
+  },
+  "log": {
+    "level": "debug"
+  },
+  "cluster": {
+    "members": [
+      "127.0.0.1:9000",
+      "127.0.0.1:9001"
+    ],
+    "hashKey": "loremipsumdolors"
+  },
+  "extensions": {
+    "search": {
+      "cve": {
+        "updateInterval": "2h"
+      }
+    },
+    "ui": {
+      "enable": true
+    }
+  }
+}

examples/scale-out-cluster-local/haproxy.cfg

@@ -0,0 +1,25 @@
+global
+        log /tmp/log local0
+        log /tmp/log local1 notice
+        maxconn 2000
+        stats timeout 30s
+        daemon
+
+defaults
+        log     global
+        mode    http
+        option  httplog
+        option  dontlognull
+        timeout connect 5000
+        timeout client  50000
+        timeout server  50000
+
+frontend zot
+    bind *:8080
+    default_backend zot-cluster
+
+backend zot-cluster
+    balance roundrobin
+    cookie SERVER insert indirect nocache
+    server zot0 127.0.0.1:9000 cookie zot0
+    server zot1 127.0.0.1:9001 cookie zot1

pkg/api/cluster_proxy.go

@@ -0,0 +1,85 @@
+package api
+
+import (
+	"fmt"
+	"io"
+	"net/http"
+
+	"github.com/gorilla/mux"
+
+	"zotregistry.dev/zot/pkg/api/constants"
+	"zotregistry.dev/zot/pkg/cluster"
+	"zotregistry.dev/zot/pkg/proxy"
+)
+
+// ClusterProxy wraps an http.HandlerFunc which requires proxying between zot instances to ensure
+// that a given repository only has a single writer and reader for dist-spec operations in a scale-out cluster.
+// based on the hash value of the repository name, the request will either be handled locally
+// or proxied to another zot member in the cluster to get the data before sending a response to the client.
+func ClusterProxy(ctrlr *Controller) func(http.HandlerFunc) http.HandlerFunc {
+	return func(next http.HandlerFunc) http.HandlerFunc {
+		return http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
+			config := ctrlr.Config
+			logger := ctrlr.Log
+
+			// if no cluster or single-node cluster, handle locally.
+			if config.Cluster == nil || len(config.Cluster.Members) == 1 {
+				next.ServeHTTP(response, request)
+
+				return
+			}
+
+			// since the handler has been wrapped, it should be possible to get the name
+			// of the repository from the mux.
+			vars := mux.Vars(request)
+			name, ok := vars["name"]
+
+			if !ok || name == "" {
+				response.WriteHeader(http.StatusNotFound)
+
+				return
+			}
+
+			// the target member is the only one which should do read/write for the dist-spec APIs
+			// for the given repository.
+			targetMemberIndex, targetMember := cluster.ComputeTargetMember(config.Cluster.HashKey, config.Cluster.Members, name)
+			logger.Debug().Str(constants.RepositoryLogKey, name).
+				Msg(fmt.Sprintf("target member socket: %s index: %d", targetMember, targetMemberIndex))
+
+			// if the target member is the same as the local member, the current member should handle the request.
+			// since the instances have the same config, a quick index lookup is sufficient
+			if targetMemberIndex == config.Cluster.Proxy.LocalMemberClusterSocketIndex {
+				logger.Debug().Str(constants.RepositoryLogKey, name).Msg("handling the request locally")
+				next.ServeHTTP(response, request)
+
+				return
+			}
+
+			// if the header contains a hop-count, return an error response as there should be no multi-hop
+			if request.Header.Get(constants.ScaleOutHopCountHeader) != "" {
+				logger.Fatal().Str("url", request.URL.String()).
+					Msg("failed to process request - cannot proxy an already proxied request")
+
+				return
+			}
+
+			logger.Debug().Str(constants.RepositoryLogKey, name).Msg("proxying the request")
+
+			proxyResponse, err := proxy.ProxyHTTPRequest(request.Context(), request, targetMember, ctrlr.Config)
+			if err != nil {
+				logger.Error().Err(err).Str(constants.RepositoryLogKey, name).Msg("failed to proxy the request")
+				http.Error(response, err.Error(), http.StatusInternalServerError)
+
+				return
+			}
+
+			defer func() {
+				_ = proxyResponse.Body.Close()
+			}()
+
+			proxy.CopyHeader(response.Header(), proxyResponse.Header)
+			response.WriteHeader(proxyResponse.StatusCode)
+			_, _ = io.Copy(response, proxyResponse.Body)
+		})
+	}
+}

pkg/api/config/config.go

@@ -531,3 +531,12 @@ func IsOauth2Supported(provider string) bool {
 
 	return false
 }
+
+func (c *Config) IsClusteringEnabled() bool {
+	return c.Cluster != nil
+}
+
+func (c *Config) IsSharedStorageEnabled() bool {
+	return c.Storage.RemoteCache &&
+		c.Storage.StorageDriver != nil && c.Storage.CacheDriver != nil
+}

pkg/api/config/config_test.go

@@ -129,4 +129,42 @@ func TestConfig(t *testing.T) {
 
 		So(conf.IsRetentionEnabled(), ShouldBeTrue)
 	})
+
+	Convey("Test IsClusteringEnabled()", t, func() {
+		conf := config.New()
+		So(conf.Cluster, ShouldBeNil)
+		So(conf.IsClusteringEnabled(), ShouldBeFalse)
+
+		conf.Cluster = &config.ClusterConfig{
+			Members: []string{
+				"127.0.0.1:9090",
+				"127.0.0.1:9001",
+			},
+			HashKey: "loremipsumdolors",
+		}
+
+		So(conf.IsClusteringEnabled(), ShouldBeTrue)
+	})
+
+	Convey("Test IsSharedStorageEnabled()", t, func() {
+		conf := config.New()
+		So(conf.Storage.RemoteCache, ShouldBeFalse)
+		So(conf.Storage.CacheDriver, ShouldBeNil)
+		So(conf.Storage.StorageDriver, ShouldBeNil)
+		So(conf.IsSharedStorageEnabled(), ShouldBeFalse)
+
+		conf.Storage.RemoteCache = true
+		So(conf.Storage.RemoteCache, ShouldBeTrue)
+		So(conf.IsSharedStorageEnabled(), ShouldBeFalse)
+
+		storageDriver := map[string]interface{}{"name": "s3"}
+		conf.Storage.StorageDriver = storageDriver
+		So(conf.Storage.StorageDriver, ShouldNotBeNil)
+		So(conf.IsSharedStorageEnabled(), ShouldBeFalse)
+
+		cacheDriver := map[string]interface{}{"name": "dynamodb"}
+		conf.Storage.CacheDriver = cacheDriver
+		So(conf.Storage.CacheDriver, ShouldNotBeNil)
+		So(conf.IsSharedStorageEnabled(), ShouldBeTrue)
+	})
 }

pkg/api/controller.go

@@ -25,6 +25,7 @@ import (
 	"zotregistry.dev/zot/pkg/log"
 	"zotregistry.dev/zot/pkg/meta"
 	mTypes "zotregistry.dev/zot/pkg/meta/types"
+	"zotregistry.dev/zot/pkg/proxy"
 	"zotregistry.dev/zot/pkg/scheduler"
 	"zotregistry.dev/zot/pkg/storage"
 	"zotregistry.dev/zot/pkg/storage/gc"
@@ -72,7 +73,7 @@ func NewController(appConfig *config.Config) *Controller {
 
 		// memberSocket is the local member's socket
 		// the index is also fetched for quick lookups during proxying
-		memberSocketIdx, memberSocket, err := GetLocalMemberClusterSocket(appConfig.Cluster.Members, localSockets)
+		memberSocketIdx, memberSocket, err := proxy.GetLocalMemberClusterSocket(appConfig.Cluster.Members, localSockets)
 		if err != nil {
 			logger.Error().Err(err).Msg("failed to get member socket")
 			panic("failed to get member socket")

pkg/extensions/extension_search.go

@@ -16,6 +16,7 @@ import (
 	"zotregistry.dev/zot/pkg/extensions/search"
 	cveinfo "zotregistry.dev/zot/pkg/extensions/search/cve"
 	"zotregistry.dev/zot/pkg/extensions/search/gql_generated"
+	gqlproxy "zotregistry.dev/zot/pkg/extensions/search/gql_proxy"
 	"zotregistry.dev/zot/pkg/log"
 	mTypes "zotregistry.dev/zot/pkg/meta/types"
 	"zotregistry.dev/zot/pkg/scheduler"
@@ -91,15 +92,18 @@ func SetupSearchRoutes(conf *config.Config, router *mux.Router, storeController
 	}
 
 	resConfig := search.GetResolverConfig(log, storeController, metaDB, cveInfo)
+	executableSchema := gql_generated.NewExecutableSchema(resConfig)
 
 	allowedMethods := zcommon.AllowedMethods(http.MethodGet, http.MethodPost)
 
+	gqlProxy := gqlproxy.GqlProxyRequestHandler(conf, log, executableSchema.Schema())
+
 	extRouter := router.PathPrefix(constants.ExtSearchPrefix).Subrouter()
 	extRouter.Use(zcommon.CORSHeadersMiddleware(conf.HTTP.AllowOrigin))
 	extRouter.Use(zcommon.ACHeadersMiddleware(conf, allowedMethods...))
 	extRouter.Use(zcommon.AddExtensionSecurityHeaders())
 	extRouter.Methods(allowedMethods...).
-		Handler(gqlHandler.NewDefaultServer(gql_generated.NewExecutableSchema(resConfig))) //nolint: staticcheck
+		Handler(gqlProxy(gqlHandler.NewDefaultServer(executableSchema)))
 
 	log.Info().Msg("finished setting up search routes")
 }

pkg/extensions/search/gql_proxy/generic_fan_out_gql_handler.go

@@ -0,0 +1,47 @@
+package gqlproxy
+
+import (
+	"encoding/json"
+	"io"
+	"net/http"
+
+	"zotregistry.dev/zot/pkg/api/config"
+	"zotregistry.dev/zot/pkg/proxy"
+)
+
+func fanOutGqlHandler(config *config.Config, response http.ResponseWriter, request *http.Request) {
+	// Proxy to all members including self in order to get the data as calling next() won't return the
+	// aggregated data to this handler.
+	finalMap := map[string]any{}
+
+	for _, targetMember := range config.Cluster.Members {
+		proxyResponse, err := proxy.ProxyHTTPRequest(request.Context(), request, targetMember, config)
+		if err != nil {
+			http.Error(response, "failed to process GQL request", http.StatusInternalServerError)
+
+			return
+		}
+
+		proxyBody, err := io.ReadAll(proxyResponse.Body)
+		_ = proxyResponse.Body.Close()
+
+		if err != nil {
+			http.Error(response, "failed to process GQL request", http.StatusInternalServerError)
+
+			return
+		}
+
+		responseResult := map[string]any{}
+
+		err = json.Unmarshal(proxyBody, &responseResult)
+		if err != nil {
+			http.Error(response, err.Error(), http.StatusInternalServerError)
+
+			return
+		}
+		// perform merge of fields
+		finalMap = deepMergeMaps(finalMap, responseResult)
+	}
+
+	prepareAndWriteResponse(finalMap, response)
+}