[full-ci] enhancement: use reva client pool selectors (#6452)

* enhancement: use reva client pool selectors

register mock service to registry and pass tests

* enhancement: bump reva

* Fix a couple of linter issues

---------

Co-authored-by: Ralf Haferkamp <rhaferkamp@owncloud.com>

services/search/pkg/content/cs3.go

@@ -11,24 +11,25 @@ import (
 	rpc "github.com/cs3org/go-cs3apis/cs3/rpc/v1beta1"
 	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
 	revactx "github.com/cs3org/reva/v2/pkg/ctx"
+	"github.com/cs3org/reva/v2/pkg/rgrpc/todo/pool"
 	"github.com/owncloud/ocis/v2/ocis-pkg/log"
 )
 
 type cs3 struct {
-	httpClient http.Client
-	gwClient   gateway.GatewayAPIClient
-	logger     log.Logger
+	httpClient      http.Client
+	gatewaySelector pool.Selectable[gateway.GatewayAPIClient]
+	logger          log.Logger
 }
 
-func newCS3Retriever(client gateway.GatewayAPIClient, logger log.Logger, insecure bool) cs3 {
+func newCS3Retriever(gatewaySelector pool.Selectable[gateway.GatewayAPIClient], logger log.Logger, insecure bool) cs3 {
 	return cs3{
 		httpClient: http.Client{
 			Transport: &http.Transport{
 				TLSClientConfig: &tls.Config{InsecureSkipVerify: insecure}, //nolint:gosec
 			},
 		},
-		gwClient: client,
-		logger:   logger,
+		gatewaySelector: gatewaySelector,
+		logger:          logger,
 	}
 }
 
@@ -40,7 +41,13 @@ func (s cs3) Retrieve(ctx context.Context, rID *provider.ResourceId) (io.ReadClo
 		return nil, fmt.Errorf("context without %s", revactx.TokenHeader)
 	}
 
-	res, err := s.gwClient.InitiateFileDownload(ctx, &provider.InitiateFileDownloadRequest{Ref: &provider.Reference{ResourceId: rID, Path: "."}})
+	gatewayClient, err := s.gatewaySelector.Next()
+	if err != nil {
+		s.logger.Error().Err(err).Msg("could not get reva gatewayClient")
+		return nil, err
+	}
+
+	res, err := gatewayClient.InitiateFileDownload(ctx, &provider.InitiateFileDownloadRequest{Ref: &provider.Reference{ResourceId: rID, Path: "."}})
 	if err != nil {
 		return nil, err
 	}

services/search/pkg/content/tika.go

@@ -8,6 +8,7 @@ import (
 	"github.com/bbalet/stopwords"
 	gateway "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1"
 	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
+	"github.com/cs3org/reva/v2/pkg/rgrpc/todo/pool"
 	"github.com/google/go-tika/tika"
 	"github.com/owncloud/ocis/v2/ocis-pkg/log"
 	"github.com/owncloud/ocis/v2/services/search/pkg/config"
@@ -23,7 +24,7 @@ type Tika struct {
 }
 
 // NewTikaExtractor creates a new Tika instance.
-func NewTikaExtractor(gw gateway.GatewayAPIClient, logger log.Logger, cfg *config.Config) (*Tika, error) {
+func NewTikaExtractor(gatewaySelector pool.Selectable[gateway.GatewayAPIClient], logger log.Logger, cfg *config.Config) (*Tika, error) {
 	basic, err := NewBasicExtractor(logger)
 	if err != nil {
 		return nil, err
@@ -38,7 +39,7 @@ func NewTikaExtractor(gw gateway.GatewayAPIClient, logger log.Logger, cfg *confi
 
 	return &Tika{
 		Basic:                      basic,
-		Retriever:                  newCS3Retriever(gw, logger, cfg.Extractor.CS3AllowInsecure),
+		Retriever:                  newCS3Retriever(gatewaySelector, logger, cfg.Extractor.CS3AllowInsecure),
 		tika:                       tika.NewClient(nil, cfg.Extractor.Tika.TikaURL),
 		contentExtractionSizeLimit: cfg.ContentExtractionSizeLimit,
 	}, nil

services/search/pkg/search/search.go

@@ -12,6 +12,7 @@ import (
 	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
 	ctxpkg "github.com/cs3org/reva/v2/pkg/ctx"
 	"github.com/cs3org/reva/v2/pkg/errtypes"
+	"github.com/cs3org/reva/v2/pkg/rgrpc/todo/pool"
 	"github.com/cs3org/reva/v2/pkg/utils"
 	"github.com/owncloud/ocis/v2/ocis-pkg/log"
 	searchmsg "github.com/owncloud/ocis/v2/protogen/gen/ocis/messages/search/v0"
@@ -20,12 +21,17 @@ import (
 )
 
 // ResolveReference makes sure the path is relative to the space root
-func ResolveReference(ctx context.Context, ref *provider.Reference, ri *provider.ResourceInfo, gw gateway.GatewayAPIClient) (*provider.Reference, error) {
+func ResolveReference(ctx context.Context, ref *provider.Reference, ri *provider.ResourceInfo, gatewaySelector pool.Selectable[gateway.GatewayAPIClient]) (*provider.Reference, error) {
 	if ref.GetResourceId().GetOpaqueId() == ref.GetResourceId().GetSpaceId() {
 		return ref, nil
 	}
 
-	gpRes, err := gw.GetPath(ctx, &provider.GetPathRequest{
+	gatewayClient, err := gatewaySelector.Next()
+	if err != nil {
+		return nil, err
+	}
+
+	gpRes, err := gatewayClient.GetPath(ctx, &provider.GetPathRequest{
 		ResourceId: ri.Id,
 	})
 	if err != nil || gpRes.Status.Code != rpc.Code_CODE_OK {
@@ -61,9 +67,15 @@ func logDocCount(engine engine.Engine, logger log.Logger) {
 	logger.Debug().Interface("count", c).Msg("new document count")
 }
 
-func getAuthContext(owner *user.User, gw gateway.GatewayAPIClient, secret string, logger log.Logger) (context.Context, error) {
+func getAuthContext(owner *user.User, gatewaySelector pool.Selectable[gateway.GatewayAPIClient], secret string, logger log.Logger) (context.Context, error) {
+	gatewayClient, err := gatewaySelector.Next()
+	if err != nil {
+		logger.Error().Err(err).Msg("could not get reva gatewayClient")
+		return nil, err
+	}
+
 	ownerCtx := ctxpkg.ContextSetUser(context.Background(), owner)
-	authRes, err := gw.Authenticate(ownerCtx, &gateway.AuthenticateRequest{
+	authRes, err := gatewayClient.Authenticate(ownerCtx, &gateway.AuthenticateRequest{
 		Type:         "machine",
 		ClientId:     "userid:" + owner.GetId().GetOpaqueId(),
 		ClientSecret: secret,
@@ -81,8 +93,13 @@ func getAuthContext(owner *user.User, gw gateway.GatewayAPIClient, secret string
 	return metadata.AppendToOutgoingContext(ownerCtx, ctxpkg.TokenHeader, authRes.Token), nil
 }
 
-func statResource(ctx context.Context, ref *provider.Reference, gw gateway.GatewayAPIClient, logger log.Logger) (*provider.StatResponse, error) {
-	res, err := gw.Stat(ctx, &provider.StatRequest{Ref: ref})
+func statResource(ctx context.Context, ref *provider.Reference, gatewaySelector pool.Selectable[gateway.GatewayAPIClient], logger log.Logger) (*provider.StatResponse, error) {
+	gatewayClient, err := gatewaySelector.Next()
+	if err != nil {
+		return nil, err
+	}
+
+	res, err := gatewayClient.Stat(ctx, &provider.StatRequest{Ref: ref})
 	if err != nil {
 		logger.Error().Err(err).Msg("failed to stat the moved resource")
 		return nil, err

services/search/pkg/search/search_suite_test.go

@@ -3,10 +3,23 @@ package search_test
 import (
 	"testing"
 
+	"github.com/owncloud/ocis/v2/ocis-pkg/registry"
+	mRegistry "go-micro.dev/v4/registry"
+
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 )
 
+func init() {
+	registry.Configure("memory")
+	r := registry.GetRegistry()
+	service := registry.BuildGRPCService("com.owncloud.api.gateway", "", "", "")
+	service.Nodes = []*mRegistry.Node{{
+		Address: "any",
+	}}
+
+	_ = r.Register(service)
+}
 func TestSearch(t *testing.T) {
 	RegisterFailHandler(Fail)
 	RunSpecs(t, "Search Suite")

services/search/pkg/search/service.go

@@ -13,6 +13,7 @@ import (
 	rpcv1beta1 "github.com/cs3org/go-cs3apis/cs3/rpc/v1beta1"
 	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
 	"github.com/cs3org/reva/v2/pkg/errtypes"
+	"github.com/cs3org/reva/v2/pkg/rgrpc/todo/pool"
 	sdk "github.com/cs3org/reva/v2/pkg/sdk/common"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/walker"
 	"github.com/cs3org/reva/v2/pkg/storagespace"
@@ -46,23 +47,23 @@ type Searcher interface {
 // Service is responsible for indexing spaces and pass on a search
 // to it's underlying engine.
 type Service struct {
-	logger    log.Logger
-	gateway   gateway.GatewayAPIClient
-	engine    engine.Engine
-	extractor content.Extractor
-	secret    string
+	logger          log.Logger
+	gatewaySelector pool.Selectable[gateway.GatewayAPIClient]
+	engine          engine.Engine
+	extractor       content.Extractor
+	secret          string
 }
 
 var errSkipSpace error
 
 // NewService creates a new Provider instance.
-func NewService(gw gateway.GatewayAPIClient, eng engine.Engine, extractor content.Extractor, logger log.Logger, cfg *config.Config) *Service {
+func NewService(gatewaySelector pool.Selectable[gateway.GatewayAPIClient], eng engine.Engine, extractor content.Extractor, logger log.Logger, cfg *config.Config) *Service {
 	var s = &Service{
-		gateway:   gw,
-		engine:    eng,
-		secret:    cfg.MachineAuthAPIKey,
-		logger:    logger,
-		extractor: extractor,
+		gatewaySelector: gatewaySelector,
+		engine:          eng,
+		secret:          cfg.MachineAuthAPIKey,
+		logger:          logger,
+		extractor:       extractor,
 	}
 
 	return s
@@ -75,7 +76,12 @@ func (s *Service) Search(ctx context.Context, req *searchsvc.SearchRequest) (*se
 	}
 	s.logger.Debug().Str("query", req.Query).Msg("performing a search")
 
-	listSpacesRes, err := s.gateway.ListStorageSpaces(ctx, &provider.ListStorageSpacesRequest{
+	gatewayClient, err := s.gatewaySelector.Next()
+	if err != nil {
+		return nil, err
+	}
+
+	listSpacesRes, err := gatewayClient.ListStorageSpaces(ctx, &provider.ListStorageSpacesRequest{
 		Filters: []*provider.ListStorageSpacesRequest_Filter{
 			{
 				Type: provider.ListStorageSpacesRequest_Filter_TYPE_SPACE_TYPE,
@@ -227,7 +233,13 @@ func (s *Service) searchIndex(ctx context.Context, req *searchsvc.SearchRequest,
 			s.logger.Warn().Interface("space", space).Msg("could not find mountpoint space for grant space")
 			return nil, errSkipSpace
 		}
-		gpRes, err := s.gateway.GetPath(ctx, &provider.GetPathRequest{
+
+		gatewayClient, err := s.gatewaySelector.Next()
+		if err != nil {
+			return nil, err
+		}
+
+		gpRes, err := gatewayClient.GetPath(ctx, &provider.GetPathRequest{
 			ResourceId: space.Root,
 		})
 		if err != nil {
@@ -296,7 +308,7 @@ func (s *Service) searchIndex(ctx context.Context, req *searchsvc.SearchRequest,
 
 // IndexSpace (re)indexes all resources of a given space.
 func (s *Service) IndexSpace(spaceID *provider.StorageSpaceId, uID *user.UserId) error {
-	ownerCtx, err := getAuthContext(&user.User{Id: uID}, s.gateway, s.secret, s.logger)
+	ownerCtx, err := getAuthContext(&user.User{Id: uID}, s.gatewaySelector, s.secret, s.logger)
 	if err != nil {
 		return err
 	}
@@ -312,7 +324,7 @@ func (s *Service) IndexSpace(spaceID *provider.StorageSpaceId, uID *user.UserId)
 	}
 	rootID.OpaqueId = rootID.SpaceId
 
-	w := walker.NewWalker(s.gateway)
+	w := walker.NewWalker(s.gatewaySelector)
 	err = w.Walk(ownerCtx, &rootID, func(wd string, info *provider.ResourceInfo, err error) error {
 		if err != nil {
 			s.logger.Error().Err(err).Msg("error walking the tree")
@@ -426,17 +438,17 @@ func (s *Service) MoveItem(ref *provider.Reference, uID *user.UserId) {
 }
 
 func (s *Service) resInfo(uID *user.UserId, ref *provider.Reference) (context.Context, *provider.StatResponse, string) {
-	ownerCtx, err := getAuthContext(&user.User{Id: uID}, s.gateway, s.secret, s.logger)
+	ownerCtx, err := getAuthContext(&user.User{Id: uID}, s.gatewaySelector, s.secret, s.logger)
 	if err != nil {
 		return nil, nil, ""
 	}
 
-	statRes, err := statResource(ownerCtx, ref, s.gateway, s.logger)
+	statRes, err := statResource(ownerCtx, ref, s.gatewaySelector, s.logger)
 	if err != nil {
 		return nil, nil, ""
 	}
 
-	r, err := ResolveReference(ownerCtx, ref, statRes.GetInfo(), s.gateway)
+	r, err := ResolveReference(ownerCtx, ref, statRes.GetInfo(), s.gatewaySelector)
 	if err != nil {
 		return nil, nil, ""
 	}

services/search/pkg/search/service_test.go

@@ -8,6 +8,7 @@ import (
 	sprovider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
 	typesv1beta1 "github.com/cs3org/go-cs3apis/cs3/types/v1beta1"
 	"github.com/cs3org/reva/v2/pkg/rgrpc/status"
+	"github.com/cs3org/reva/v2/pkg/rgrpc/todo/pool"
 	cs3mocks "github.com/cs3org/reva/v2/tests/cs3mocks/mocks"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
@@ -20,17 +21,19 @@ import (
 	engineMocks "github.com/owncloud/ocis/v2/services/search/pkg/engine/mocks"
 	"github.com/owncloud/ocis/v2/services/search/pkg/search"
 	"github.com/stretchr/testify/mock"
+	"google.golang.org/grpc"
 )
 
 var _ = Describe("Searchprovider", func() {
 	var (
-		s           search.Searcher
-		extractor   *contentMocks.Extractor
-		gw          *cs3mocks.GatewayAPIClient
-		indexClient *engineMocks.Engine
-		ctx         context.Context
-		logger      = log.NewLogger()
-		user        = &userv1beta1.User{
+		s               search.Searcher
+		extractor       *contentMocks.Extractor
+		gatewayClient   *cs3mocks.GatewayAPIClient
+		gatewaySelector pool.Selectable[gateway.GatewayAPIClient]
+		indexClient     *engineMocks.Engine
+		ctx             context.Context
+		logger          = log.NewLogger()
+		user            = &userv1beta1.User{
 			Id: &userv1beta1.UserId{
 				OpaqueId: "user",
 			},
@@ -70,22 +73,31 @@ var _ = Describe("Searchprovider", func() {
 	)
 
 	BeforeEach(func() {
+		pool.RemoveSelector("GatewaySelector" + "com.owncloud.api.gateway")
+		gatewayClient = &cs3mocks.GatewayAPIClient{}
+		gatewaySelector = pool.GetSelector[gateway.GatewayAPIClient](
+			"GatewaySelector",
+			"com.owncloud.api.gateway",
+			func(cc *grpc.ClientConn) gateway.GatewayAPIClient {
+				return gatewayClient
+			},
+		)
+
 		ctx = context.Background()
-		gw = &cs3mocks.GatewayAPIClient{}
 		indexClient = &engineMocks.Engine{}
 		extractor = &contentMocks.Extractor{}
 
-		s = search.NewService(gw, indexClient, extractor, logger, &config.Config{})
+		s = search.NewService(gatewaySelector, indexClient, extractor, logger, &config.Config{})
 
-		gw.On("Authenticate", mock.Anything, mock.Anything).Return(&gateway.AuthenticateResponse{
+		gatewayClient.On("Authenticate", mock.Anything, mock.Anything).Return(&gateway.AuthenticateResponse{
 			Status: status.NewOK(ctx),
 			Token:  "authtoken",
 		}, nil)
-		gw.On("Stat", mock.Anything, mock.Anything).Return(&sprovider.StatResponse{
+		gatewayClient.On("Stat", mock.Anything, mock.Anything).Return(&sprovider.StatResponse{
 			Status: status.NewOK(context.Background()),
 			Info:   ri,
 		}, nil)
-		gw.On("GetPath", mock.Anything, mock.MatchedBy(func(req *sprovider.GetPathRequest) bool {
+		gatewayClient.On("GetPath", mock.Anything, mock.MatchedBy(func(req *sprovider.GetPathRequest) bool {
 			return req.ResourceId.OpaqueId == ri.Id.OpaqueId
 		})).Return(&sprovider.GetPathResponse{
 			Status: status.NewOK(context.Background()),
@@ -96,14 +108,14 @@ var _ = Describe("Searchprovider", func() {
 
 	Describe("New", func() {
 		It("returns a new instance", func() {
-			s := search.NewService(gw, indexClient, extractor, logger, &config.Config{})
+			s := search.NewService(gatewaySelector, indexClient, extractor, logger, &config.Config{})
 			Expect(s).ToNot(BeNil())
 		})
 	})
 
 	Describe("IndexSpace", func() {
 		It("walks the space and indexes all files", func() {
-			gw.On("GetUserByClaim", mock.Anything, mock.Anything).Return(&userv1beta1.GetUserByClaimResponse{
+			gatewayClient.On("GetUserByClaim", mock.Anything, mock.Anything).Return(&userv1beta1.GetUserByClaimResponse{
 				Status: status.NewOK(context.Background()),
 				User:   user,
 			}, nil)
@@ -127,7 +139,7 @@ var _ = Describe("Searchprovider", func() {
 
 		Context("with a personal space", func() {
 			BeforeEach(func() {
-				gw.On("ListStorageSpaces", mock.Anything, mock.Anything).Return(&sprovider.ListStorageSpacesResponse{
+				gatewayClient.On("ListStorageSpaces", mock.Anything, mock.Anything).Return(&sprovider.ListStorageSpacesResponse{
 					Status:        status.NewOK(ctx),
 					StorageSpaces: []*sprovider.StorageSpace{personalSpace},
 				}, nil)
@@ -210,14 +222,14 @@ var _ = Describe("Searchprovider", func() {
 						},
 					},
 				}
-				gw.On("GetPath", mock.Anything, mock.Anything).Return(&sprovider.GetPathResponse{
+				gatewayClient.On("GetPath", mock.Anything, mock.Anything).Return(&sprovider.GetPathResponse{
 					Status: status.NewOK(ctx),
 					Path:   "/grant/path",
 				}, nil)
 			})
 
 			It("searches the received spaces", func() {
-				gw.On("ListStorageSpaces", mock.Anything, mock.Anything).Return(&sprovider.ListStorageSpacesResponse{
+				gatewayClient.On("ListStorageSpaces", mock.Anything, mock.Anything).Return(&sprovider.ListStorageSpacesResponse{
 					Status:        status.NewOK(ctx),
 					StorageSpaces: []*sprovider.StorageSpace{grantSpace, mountpointSpace},
 				}, nil)
@@ -259,7 +271,7 @@ var _ = Describe("Searchprovider", func() {
 
 			Context("when searching both spaces", func() {
 				BeforeEach(func() {
-					gw.On("ListStorageSpaces", mock.Anything, mock.Anything).Return(&sprovider.ListStorageSpacesResponse{
+					gatewayClient.On("ListStorageSpaces", mock.Anything, mock.Anything).Return(&sprovider.ListStorageSpacesResponse{
 						Status:        status.NewOK(ctx),
 						StorageSpaces: []*sprovider.StorageSpace{personalSpace, grantSpace, mountpointSpace},
 					}, nil)

services/search/pkg/service/grpc/v0/service.go

@@ -19,6 +19,7 @@ import (
 	"github.com/jellydator/ttlcache/v2"
 	ociscrypto "github.com/owncloud/ocis/v2/ocis-pkg/crypto"
 	"github.com/owncloud/ocis/v2/ocis-pkg/log"
+	"github.com/owncloud/ocis/v2/ocis-pkg/registry"
 	v0 "github.com/owncloud/ocis/v2/protogen/gen/ocis/messages/search/v0"
 	searchsvc "github.com/owncloud/ocis/v2/protogen/gen/ocis/services/search/v0"
 	"github.com/owncloud/ocis/v2/services/search/pkg/content"
@@ -55,9 +56,9 @@ func NewHandler(opts ...Option) (searchsvc.SearchProviderHandler, func(), error)
 	}
 
 	// initialize gateway
-	gw, err := pool.GetGatewayServiceClient(cfg.Reva.Address)
+	selector, err := pool.GatewaySelector(cfg.Reva.Address, pool.WithRegistry(registry.GetRegistry()))
 	if err != nil {
-		logger.Fatal().Err(err).Str("addr", cfg.Reva.Address).Msg("could not get reva client")
+		logger.Fatal().Err(err).Msg("could not get reva gateway selector")
 		return nil, teardown, err
 	}
 	// initialize search content extractor
@@ -68,7 +69,7 @@ func NewHandler(opts ...Option) (searchsvc.SearchProviderHandler, func(), error)
 			return nil, teardown, err
 		}
 	case "tika":
-		if extractor, err = content.NewTikaExtractor(gw, logger, cfg); err != nil {
+		if extractor, err = content.NewTikaExtractor(selector, logger, cfg); err != nil {
 			return nil, teardown, err
 		}
 	default:
@@ -106,7 +107,7 @@ func NewHandler(opts ...Option) (searchsvc.SearchProviderHandler, func(), error)
 		return nil, teardown, err
 	}
 
-	ss := search.NewService(gw, eng, extractor, logger, cfg)
+	ss := search.NewService(selector, eng, extractor, logger, cfg)
 
 	// setup event handling
 	if err := search.HandleEvents(ss, bus, logger, cfg); err != nil {