use less selectors

changelog/unreleased/use-less-selectors.md

@@ -0,0 +1,5 @@
+Bugfix: use less selectors that watch the registry
+
+The proxy now shares the service selector for all host lookups.
+
+https://github.com/owncloud/ocis/pull/9741

services/proxy/pkg/command/server.go

@@ -15,11 +15,6 @@ import (
 	chimiddleware "github.com/go-chi/chi/v5/middleware"
 	"github.com/justinas/alice"
 	"github.com/oklog/run"
-	"github.com/urfave/cli/v2"
-	microstore "go-micro.dev/v4/store"
-	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
-	"go.opentelemetry.io/otel/trace"
-
 	"github.com/owncloud/ocis/v2/ocis-pkg/config/configlog"
 	"github.com/owncloud/ocis/v2/ocis-pkg/log"
 	pkgmiddleware "github.com/owncloud/ocis/v2/ocis-pkg/middleware"
@@ -43,6 +38,11 @@ import (
 	"github.com/owncloud/ocis/v2/services/proxy/pkg/user/backend"
 	"github.com/owncloud/ocis/v2/services/proxy/pkg/userroles"
 	ocisstore "github.com/owncloud/ocis/v2/services/store/pkg/store"
+	"github.com/urfave/cli/v2"
+	"go-micro.dev/v4/selector"
+	microstore "go-micro.dev/v4/store"
+	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
+	"go.opentelemetry.io/otel/trace"
 )
 
 // Server is the entrypoint for the server command.
@@ -133,23 +133,28 @@ func Server(cfg *config.Config) *cli.Command {
 				return fmt.Errorf("failed to initialize reverse proxy: %w", err)
 			}
 
+			reg := registry.GetRegistry()
+
 			gatewaySelector, err := pool.GatewaySelector(
 				cfg.Reva.Address,
 				append(
 					cfg.Reva.GetRevaOptions(),
-					pool.WithRegistry(registry.GetRegistry()),
+					pool.WithRegistry(reg),
 					pool.WithTracerProvider(traceProvider),
 				)...)
 			if err != nil {
 				logger.Fatal().Err(err).Msg("Failed to get gateway selector")
 			}
 
+			serviceSelector := selector.NewSelector(selector.Registry(reg))
+
 			var userProvider backend.UserBackend
 			switch cfg.AccountBackend {
 			case "cs3":
 				userProvider = backend.NewCS3UserBackend(
 					backend.WithLogger(logger),
 					backend.WithRevaGatewaySelector(gatewaySelector),
+					backend.WithSelector(serviceSelector),
 					backend.WithMachineAuthAPIKey(cfg.MachineAuthAPIKey),
 					backend.WithOIDCissuer(cfg.OIDC.Issuer),
 					backend.WithServiceAccount(cfg.ServiceAccount),
@@ -187,7 +192,7 @@ func Server(cfg *config.Config) *cli.Command {
 			}
 
 			{
-				middlewares := loadMiddlewares(logger, cfg, userInfoCache, signingKeyStore, traceProvider, *m, userProvider, gatewaySelector)
+				middlewares := loadMiddlewares(logger, cfg, userInfoCache, signingKeyStore, traceProvider, *m, userProvider, gatewaySelector, serviceSelector)
 
 				server, err := proxyHTTP.Server(
 					proxyHTTP.Handler(lh.Handler()),
@@ -242,7 +247,7 @@ func Server(cfg *config.Config) *cli.Command {
 
 func loadMiddlewares(logger log.Logger, cfg *config.Config,
 	userInfoCache, signingKeyStore microstore.Store, traceProvider trace.TracerProvider, metrics metrics.Metrics,
-	userProvider backend.UserBackend, gatewaySelector pool.Selectable[gateway.GatewayAPIClient]) alice.Chain {
+	userProvider backend.UserBackend, gatewaySelector pool.Selectable[gateway.GatewayAPIClient], serviceSelector selector.Selector) alice.Chain {
 
 	rolesClient := settingssvc.NewRoleService("com.owncloud.api.settings", cfg.GrpcClient)
 	policiesProviderClient := policiessvc.NewPoliciesProviderService("com.owncloud.api.policies", cfg.GrpcClient)
@@ -342,7 +347,7 @@ func loadMiddlewares(logger log.Logger, cfg *config.Config,
 		middleware.AccessLog(logger),
 		middleware.HTTPSRedirect,
 		middleware.Security(cspConfig),
-		router.Middleware(cfg.PolicySelector, cfg.Policies, logger),
+		router.Middleware(serviceSelector, cfg.PolicySelector, cfg.Policies, logger),
 		middleware.Authentication(
 			authenticators,
 			middleware.CredentialsByUserAgent(cfg.AuthMiddleware.CredentialsByUserAgent),

services/proxy/pkg/proxy/proxy_integration_test.go

@@ -10,8 +10,10 @@ import (
 	"testing"
 
 	"github.com/owncloud/ocis/v2/ocis-pkg/log"
+	"github.com/owncloud/ocis/v2/ocis-pkg/registry"
 	"github.com/owncloud/ocis/v2/services/proxy/pkg/config"
 	"github.com/owncloud/ocis/v2/services/proxy/pkg/router"
+	"go-micro.dev/v4/selector"
 )
 
 func TestProxyIntegration(t *testing.T) {
@@ -111,12 +113,15 @@ func TestProxyIntegration(t *testing.T) {
 			expectProxyTo("http://users.example.com/user/1234"),
 	}
 
+	reg := registry.GetRegistry()
+	sel := selector.NewSelector(selector.Registry(reg))
+
 	for k := range tests {
 		t.Run(tests[k].id, func(t *testing.T) {
 			t.Parallel()
 			tc := tests[k]
 
-			rt := router.Middleware(nil, tc.conf, log.NewLogger())
+			rt := router.Middleware(sel, nil, tc.conf, log.NewLogger())
 			rp := newTestProxy(testConfig(tc.conf), func(req *http.Request) *http.Response {
 				if got, want := req.URL.String(), tc.expect.String(); got != want {
 					t.Errorf("Proxied url should be %v got %v", want, got)

services/proxy/pkg/router/router.go

@@ -9,7 +9,6 @@ import (
 	"strings"
 
 	"github.com/owncloud/ocis/v2/ocis-pkg/log"
-	"github.com/owncloud/ocis/v2/ocis-pkg/registry"
 	"github.com/owncloud/ocis/v2/services/proxy/pkg/config"
 	"github.com/owncloud/ocis/v2/services/proxy/pkg/proxy/policy"
 	"go-micro.dev/v4/selector"
@@ -20,8 +19,8 @@ type routingInfoCtxKey struct{}
 var noInfo = RoutingInfo{}
 
 // Middleware returns a HTTP middleware containing the router.
-func Middleware(policySelector *config.PolicySelector, policies []config.Policy, logger log.Logger) func(http.Handler) http.Handler {
-	router := New(policySelector, policies, logger)
+func Middleware(serviceSelector selector.Selector, policySelectorCfg *config.PolicySelector, policies []config.Policy, logger log.Logger) func(http.Handler) http.Handler {
+	router := New(serviceSelector, policySelectorCfg, policies, logger)
 	return func(next http.Handler) http.Handler {
 		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 			ri, ok := router.Route(r)
@@ -36,30 +35,31 @@ func Middleware(policySelector *config.PolicySelector, policies []config.Policy,
 
 // New creates a new request router.
 // It initializes the routes before returning the router.
-func New(policySelector *config.PolicySelector, policies []config.Policy, logger log.Logger) Router {
-	if policySelector == nil {
+func New(serviceSelector selector.Selector, policySelectorCfg *config.PolicySelector, policies []config.Policy, logger log.Logger) Router {
+	if policySelectorCfg == nil {
 		firstPolicy := policies[0].Name
 		logger.Warn().Str("policy", firstPolicy).Msg("policy-selector not configured. Will always use first policy")
-		policySelector = &config.PolicySelector{
+		policySelectorCfg = &config.PolicySelector{
 			Static: &config.StaticSelectorConf{
 				Policy: firstPolicy,
 			},
 		}
 	}
 
 	logger.Debug().
-		Interface("selector_config", policySelector).
+		Interface("selector_config", policySelectorCfg).
 		Msg("loading policy-selector")
 
-	selector, err := policy.LoadSelector(policySelector)
+	policySelector, err := policy.LoadSelector(policySelectorCfg)
 	if err != nil {
 		logger.Fatal().Err(err).Msg("Could not load policy-selector")
 	}
 
 	r := Router{
-		logger:         logger,
-		rewriters:      make(map[string]map[config.RouteType]map[string][]RoutingInfo),
-		policySelector: selector,
+		logger:          logger,
+		rewriters:       make(map[string]map[config.RouteType]map[string][]RoutingInfo),
+		policySelector:  policySelector,
+		serviceSelector: serviceSelector,
 	}
 	for _, pol := range policies {
 		for _, route := range pol.Routes {
@@ -103,9 +103,10 @@ func (r RoutingInfo) IsRouteUnprotected() bool {
 
 // Router handles the routing of HTTP requests according to the given policies.
 type Router struct {
-	logger         log.Logger
-	rewriters      map[string]map[config.RouteType]map[string][]RoutingInfo
-	policySelector policy.Selector
+	logger          log.Logger
+	rewriters       map[string]map[config.RouteType]map[string][]RoutingInfo
+	policySelector  policy.Selector
+	serviceSelector selector.Selector
 }
 
 func (rt Router) addHost(policy string, target *url.URL, route config.Route) {
@@ -124,16 +125,13 @@ func (rt Router) addHost(policy string, target *url.URL, route config.Route) {
 		rt.rewriters[policy][routeType][route.Method] = make([]RoutingInfo, 0)
 	}
 
-	reg := registry.GetRegistry()
-	sel := selector.NewSelector(selector.Registry(reg))
-
 	rt.rewriters[policy][routeType][route.Method] = append(rt.rewriters[policy][routeType][route.Method], RoutingInfo{
 		endpoint:    route.Endpoint,
 		unprotected: route.Unprotected,
 		rewrite: func(req *httputil.ProxyRequest) {
 			if route.Service != "" {
 				// select next node
-				next, err := sel.Select(route.Service)
+				next, err := rt.serviceSelector.Select(route.Service)
 				if err != nil {
 					rt.logger.Error().Err(err).
 						Str("service", route.Service).

services/proxy/pkg/router/router_test.go

@@ -10,8 +10,10 @@ import (
 	"testing"
 
 	"github.com/owncloud/ocis/v2/ocis-pkg/log"
+	"github.com/owncloud/ocis/v2/ocis-pkg/registry"
 	"github.com/owncloud/ocis/v2/services/proxy/pkg/config"
 	"github.com/owncloud/ocis/v2/services/proxy/pkg/config/defaults"
+	"go-micro.dev/v4/selector"
 )
 
 type matchertest struct {
@@ -69,7 +71,9 @@ func TestQueryRouteMatcher(t *testing.T) {
 func TestRegexRouteMatcher(t *testing.T) {
 	cfg := defaults.DefaultConfig()
 	cfg.Policies = defaults.DefaultPolicies()
-	rt := New(cfg.PolicySelector, cfg.Policies, log.NewLogger())
+	reg := registry.GetRegistry()
+	sel := selector.NewSelector(selector.Registry(reg))
+	rt := New(sel, cfg.PolicySelector, cfg.Policies, log.NewLogger())
 
 	table := []matchertest{
 		{endpoint: ".*some\\/url.*parameter=true", target: "/foobar/baz/some/url?parameter=true", matches: true},
@@ -112,7 +116,7 @@ func TestRouter(t *testing.T) {
 	}))
 	defer svr.Close()
 
-	selector := &config.PolicySelector{
+	policySelectorCfg := &config.PolicySelector{
 		Static: &config.StaticSelectorConf{
 			Policy: "default",
 		},
@@ -129,7 +133,9 @@ func TestRouter(t *testing.T) {
 		},
 	}
 
-	router := New(selector, policies, log.NewLogger())
+	reg := registry.GetRegistry()
+	sel := selector.NewSelector(selector.Registry(reg))
+	router := New(sel, policySelectorCfg, policies, log.NewLogger())
 
 	table := []matchertest{
 		{method: "PROPFIND", endpoint: "/dav/files/demo/", target: "ocdav"},

services/proxy/pkg/user/backend/cs3.go

@@ -15,13 +15,11 @@ import (
 	"github.com/cs3org/reva/v2/pkg/rgrpc/todo/pool"
 	utils "github.com/cs3org/reva/v2/pkg/utils"
 	libregraph "github.com/owncloud/libre-graph-api-go"
-	"go-micro.dev/v4/selector"
-
 	"github.com/owncloud/ocis/v2/ocis-pkg/log"
 	"github.com/owncloud/ocis/v2/ocis-pkg/oidc"
-	"github.com/owncloud/ocis/v2/ocis-pkg/registry"
 	"github.com/owncloud/ocis/v2/services/graph/pkg/errorcode"
 	"github.com/owncloud/ocis/v2/services/proxy/pkg/config"
+	"go-micro.dev/v4/selector"
 )
 
 type cs3backend struct {
@@ -36,6 +34,7 @@ type Option func(o *Options)
 type Options struct {
 	logger              log.Logger
 	gatewaySelector     pool.Selectable[gateway.GatewayAPIClient]
+	selector            selector.Selector
 	machineAuthAPIKey   string
 	oidcISS             string
 	serviceAccount      config.ServiceAccount
@@ -60,6 +59,13 @@ func WithRevaGatewaySelector(selectable pool.Selectable[gateway.GatewayAPIClient
 	}
 }
 
+// WithSelector set the Selector option
+func WithSelector(selector selector.Selector) Option {
+	return func(o *Options) {
+		o.selector = selector
+	}
+}
+
 // WithMachineAuthAPIKey configures the machine auth API key
 func WithMachineAuthAPIKey(ma string) Option {
 	return func(o *Options) {
@@ -94,12 +100,9 @@ func NewCS3UserBackend(opts ...Option) UserBackend {
 		o(&opt)
 	}
 
-	reg := registry.GetRegistry()
-	sel := selector.NewSelector(selector.Registry(reg))
-
 	b := cs3backend{
 		Options:       opt,
-		graphSelector: sel,
+		graphSelector: opt.selector,
 	}
 
 	return &b