[full-ci] fix: always select next gateway client

changelog/unreleased/fix-select-next-gateway-client.md

@@ -0,0 +1,5 @@
+Bugfix: Always select next gateway client
+
+We now use the gateway selector to always select the next gateway client. This ensures that we can always connect to the gateway during up- and downscaling.
+
+https://github.com/owncloud/ocis/pull/10133

services/collaboration/.mockery.yaml

@@ -15,3 +15,10 @@ packages:
             dir: "mocks"
         interfaces:
             LockParser:
+    github.com/cs3org/reva/v2/pkg/rgrpc/todo/pool:
+        config:
+            dir: "mocks"
+        interfaces:
+            Selectable:
+                config:
+                    filename: "gateway_selector.go"

services/collaboration/pkg/command/server.go

@@ -5,8 +5,10 @@ import (
 	"fmt"
 	"net"
 
+	"github.com/cs3org/reva/v2/pkg/rgrpc/todo/pool"
 	"github.com/oklog/run"
 	"github.com/owncloud/ocis/v2/ocis-pkg/config/configlog"
+	registry "github.com/owncloud/ocis/v2/ocis-pkg/registry"
 	"github.com/owncloud/ocis/v2/ocis-pkg/tracing"
 	"github.com/owncloud/ocis/v2/services/collaboration/pkg/config"
 	"github.com/owncloud/ocis/v2/services/collaboration/pkg/config/parser"
@@ -44,7 +46,17 @@ func Server(cfg *config.Config) *cli.Command {
 				return err
 			}
 
-			gwc, err := helpers.GetCS3apiClient(cfg, false)
+			tm, err := pool.StringToTLSMode(cfg.CS3Api.GRPCClientTLS.Mode)
+			if err != nil {
+				return err
+			}
+			gatewaySelector, err := pool.GatewaySelector(
+				cfg.CS3Api.Gateway.Name,
+				pool.WithTLSCACert(cfg.CS3Api.GRPCClientTLS.CACert),
+				pool.WithTLSMode(tm),
+				pool.WithRegistry(registry.GetRegistry()),
+				pool.WithTracerProvider(traceProvider),
+			)
 			if err != nil {
 				return err
 			}
@@ -54,7 +66,7 @@ func Server(cfg *config.Config) *cli.Command {
 				return err
 			}
 
-			if err := helpers.RegisterAppProvider(ctx, cfg, logger, gwc, appUrls); err != nil {
+			if err := helpers.RegisterAppProvider(ctx, cfg, logger, gatewaySelector, appUrls); err != nil {
 				return err
 			}
 
@@ -101,7 +113,7 @@ func Server(cfg *config.Config) *cli.Command {
 
 			// start HTTP server
 			httpServer, err := http.Server(
-				http.Adapter(connector.NewHttpAdapter(gwc, cfg)),
+				http.Adapter(connector.NewHttpAdapter(gatewaySelector, cfg)),
 				http.Logger(logger),
 				http.Config(cfg),
 				http.Context(ctx),

services/collaboration/pkg/config/cs3api.go

@@ -1,9 +1,12 @@
 package config
 
+import "github.com/owncloud/ocis/v2/ocis-pkg/shared"
+
 // CS3Api defines the available configuration in order to access to the CS3 gateway.
 type CS3Api struct {
-	Gateway     Gateway     `yaml:"gateway"`
-	DataGateway DataGateway `yaml:"datagateway"`
+	Gateway       Gateway               `yaml:"gateway"`
+	DataGateway   DataGateway           `yaml:"datagateway"`
+	GRPCClientTLS *shared.GRPCClientTLS `yaml:"grpc_client_tls"`
 }
 
 // Gateway defines the available configuration for the CS3 API gateway

services/collaboration/pkg/config/defaults/defaultconfig.go

@@ -2,6 +2,7 @@ package defaults
 
 import (
 	"github.com/owncloud/ocis/v2/ocis-pkg/shared"
+	"github.com/owncloud/ocis/v2/ocis-pkg/structs"
 	"github.com/owncloud/ocis/v2/services/collaboration/pkg/config"
 )
 
@@ -93,6 +94,9 @@ func EnsureDefaults(cfg *config.Config) {
 	} else if cfg.TokenManager == nil {
 		cfg.TokenManager = &config.TokenManager{}
 	}
+	if cfg.CS3Api.GRPCClientTLS == nil && cfg.Commons != nil {
+		cfg.CS3Api.GRPCClientTLS = structs.CopyOrZeroValue(cfg.Commons.GRPCClientTLS)
+	}
 }
 
 // Sanitize sanitized the configuration

services/collaboration/pkg/connector/contentconnector.go

@@ -15,6 +15,7 @@ import (
 	providerv1beta1 "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
 	types "github.com/cs3org/go-cs3apis/cs3/types/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/tracing"
 	"github.com/owncloud/ocis/v2/services/collaboration/pkg/config"
 	"github.com/owncloud/ocis/v2/services/collaboration/pkg/middleware"
@@ -42,14 +43,14 @@ type ContentConnectorService interface {
 // uploads (PutFile)
 // Note that operations might return any kind of error, not just ConnectorError
 type ContentConnector struct {
-	gwc gatewayv1beta1.GatewayAPIClient
+	gws pool.Selectable[gatewayv1beta1.GatewayAPIClient]
 	cfg *config.Config
 }
 
 // NewContentConnector creates a new content connector
-func NewContentConnector(gwc gatewayv1beta1.GatewayAPIClient, cfg *config.Config) *ContentConnector {
+func NewContentConnector(gws pool.Selectable[gatewayv1beta1.GatewayAPIClient], cfg *config.Config) *ContentConnector {
 	return &ContentConnector{
-		gwc: gwc,
+		gws: gws,
 		cfg: cfg,
 	}
 }
@@ -76,7 +77,11 @@ func (c *ContentConnector) GetFile(ctx context.Context, w http.ResponseWriter) e
 		Logger()
 	logger.Debug().Msg("GetFile: start")
 
-	sResp, err := c.gwc.Stat(ctx, &providerv1beta1.StatRequest{
+	gwc, err := c.gws.Next()
+	if err != nil {
+		return err
+	}
+	sResp, err := gwc.Stat(ctx, &providerv1beta1.StatRequest{
 		Ref: wopiContext.FileReference,
 	})
 	if err != nil {
@@ -94,7 +99,11 @@ func (c *ContentConnector) GetFile(ctx context.Context, w http.ResponseWriter) e
 	if wopiContext.ViewMode == appproviderv1beta1.ViewMode_VIEW_MODE_VIEW_ONLY && wopiContext.ViewOnlyToken != "" {
 		ctx = revactx.ContextSetToken(ctx, wopiContext.ViewOnlyToken)
 	}
-	resp, err := c.gwc.InitiateFileDownload(ctx, req)
+	gwc, err = c.gws.Next()
+	if err != nil {
+		return err
+	}
+	resp, err := gwc.InitiateFileDownload(ctx, req)
 	if err != nil {
 		logger.Error().Err(err).Msg("GetFile: InitiateFileDownload failed")
 		return err
@@ -227,9 +236,13 @@ func (c *ContentConnector) PutFile(ctx context.Context, stream io.Reader, stream
 		Logger()
 	logger.Debug().Msg("PutFile: start")
 
+	gwc, err := c.gws.Next()
+	if err != nil {
+		return nil, err
+	}
 	// We need a stat call on the target file in order to get both the lock
 	// (if any) and the current size of the file
-	statRes, err := c.gwc.Stat(ctx, &providerv1beta1.StatRequest{
+	statRes, err := gwc.Stat(ctx, &providerv1beta1.StatRequest{
 		Ref: wopiContext.FileReference,
 	})
 	if err != nil {
@@ -286,8 +299,12 @@ func (c *ContentConnector) PutFile(ctx context.Context, stream io.Reader, stream
 		},
 	}
 
+	gwc, err = c.gws.Next()
+	if err != nil {
+		return nil, err
+	}
 	// Initiate the upload request
-	resp, err := c.gwc.InitiateFileUpload(ctx, req)
+	resp, err := gwc.InitiateFileUpload(ctx, req)
 	if err != nil {
 		logger.Error().Err(err).Msg("UploadHelper: InitiateFileUpload failed")
 		return nil, err
@@ -383,9 +400,13 @@ func (c *ContentConnector) PutFile(ctx context.Context, stream io.Reader, stream
 				Msg("UploadHelper: Put request to the upload endpoint failed with unexpected status")
 			return NewResponse(500), nil
 		}
+		gwc, err = c.gws.Next()
+		if err != nil {
+			return nil, err
+		}
 		// We need a stat call on the target file after the upload to get the
 		// new mtime
-		statResAfter, err := c.gwc.Stat(ctx, &providerv1beta1.StatRequest{
+		statResAfter, err := gwc.Stat(ctx, &providerv1beta1.StatRequest{
 			Ref: wopiContext.FileReference,
 		})
 		if err != nil {

services/collaboration/pkg/connector/contentconnector_test.go

@@ -11,6 +11,7 @@ import (
 	"github.com/cs3org/reva/v2/pkg/utils"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
+	"github.com/owncloud/ocis/v2/services/collaboration/mocks"
 	"github.com/stretchr/testify/mock"
 
 	appproviderv1beta1 "github.com/cs3org/go-cs3apis/cs3/app/provider/v1beta1"
@@ -27,10 +28,11 @@ import (
 
 var _ = Describe("ContentConnector", func() {
 	var (
-		cc            *connector.ContentConnector
-		gatewayClient *cs3mocks.GatewayAPIClient
-		cfg           *config.Config
-		wopiCtx       middleware.WopiContext
+		cc              *connector.ContentConnector
+		gatewayClient   *cs3mocks.GatewayAPIClient
+		gatewaySelector *mocks.Selectable[gateway.GatewayAPIClient]
+		cfg             *config.Config
+		wopiCtx         middleware.WopiContext
 
 		srv           *httptest.Server
 		srvReqHeader  http.Header
@@ -40,8 +42,11 @@ var _ = Describe("ContentConnector", func() {
 	BeforeEach(func() {
 		// contentConnector only uses "cfg.CS3Api.DataGateway.Insecure", which is irrelevant for the tests
 		cfg = &config.Config{}
-		gatewayClient = &cs3mocks.GatewayAPIClient{}
-		cc = connector.NewContentConnector(gatewayClient, cfg)
+		gatewayClient = cs3mocks.NewGatewayAPIClient(GinkgoT())
+
+		gatewaySelector = mocks.NewSelectable[gateway.GatewayAPIClient](GinkgoT())
+		gatewaySelector.On("Next").Return(gatewayClient, nil)
+		cc = connector.NewContentConnector(gatewaySelector, cfg)
 
 		wopiCtx = middleware.WopiContext{
 			AccessToken: "abcdef123456",
@@ -91,6 +96,8 @@ var _ = Describe("ContentConnector", func() {
 			}, nil)
 		})
 		It("No valid context", func() {
+			gatewaySelector.EXPECT().Next().Unset()
+			gatewayClient.EXPECT().Stat(mock.Anything, mock.Anything).Unset()
 			sb := httptest.NewRecorder()
 			ctx := context.Background()
 			err := cc.GetFile(ctx, sb)
@@ -226,6 +233,7 @@ var _ = Describe("ContentConnector", func() {
 
 	Describe("PutFile", func() {
 		It("No valid context", func() {
+			gatewaySelector.EXPECT().Next().Unset()
 			reader := strings.NewReader("Content to upload is here!")
 			ctx := context.Background()
 			response, err := cc.PutFile(ctx, reader, reader.Size(), "notARandomLockId")