feat: added new grpc sync config option to allow setting max receive …

…message size. (#1358)

## This PR
Added a new config option `maxMsgSize` which will
allow users to override default message size 4Mb.

- allow override default max message size 4Mb

### Related Issues

Fixes #1357 

### How to test
To test this feature you need a flag source with large number of flags >
4Mb, e.g. below we are running with max size of 5Mb+
 
```
./flagd start --port 8013 --sync-port 8017 --sources='[{"uri": "localhost:8015", "provider": "grpc", "tls": false, "providerID": "flagd-sidecar", "selector": "all-flags", "maxMsgSize": 5728474}]'

		 ______   __       ________   _______    ______
		/_____/\ /_/\     /_______/\ /______/\  /_____/\
		\::::_\/_\:\ \    \::: _  \ \\::::__\/__\:::_ \ \
		 \:\/___/\\:\ \    \::(_)  \ \\:\ /____/\\:\ \ \ \
		  \:::._\/ \:\ \____\:: __  \ \\:\\_  _\/ \:\ \ \ \
		   \:\ \    \:\/___/\\:.\ \  \ \\:\_\ \ \  \:\/.:| |
		    \_\/     \_____\/ \__\/\__\/ \_____\/   \____/_/

2024-07-11T11:31:57.024+0200	info	cmd/start.go:107	flagd version: dev (da01e08), built at: 2024-07-11T11:14:44Z	{"component": "start"}
2024-07-11T11:31:57.026+0200	info	flag-sync/sync_service.go:54	starting flag sync service on port 8017	{"component": "FlagSyncService"}
2024-07-11T11:31:57.027+0200	info	grpc/grpc_sync.go:70	setting max receive message size 5728474 bytes default 4MB	{"component": "sync", "sync": "grpc"}

```

---------

Signed-off-by: Pradeep Mishra <pradeepbbl@gmail.com>
Signed-off-by: pradeepbbl <pradeepbbl@gmail.com>
Signed-off-by: Pradeep Mishra <pradeepbbl@users.noreply.github.com>
Co-authored-by: Michael Beemer <beeme1mr@users.noreply.github.com>

core/pkg/sync/builder/syncbuilder.go

@@ -166,6 +166,7 @@ func (sb *SyncBuilder) newGRPC(config sync.SourceConfig, logger *logger.Logger)
 		ProviderID:        config.ProviderID,
 		Secure:            config.TLS,
 		Selector:          config.Selector,
+		MaxMsgSize:        config.MaxMsgSize,
 	}
 }
 

core/pkg/sync/builder/syncbuilder_test.go

@@ -173,6 +173,26 @@ func Test_SyncsFromFromConfig(t *testing.T) {
 			},
 			wantErr: false,
 		},
+		{
+			name: "grpc-with-msg-size",
+			args: args{
+				logger: lg,
+				sources: []sync.SourceConfig{
+					{
+						URI:        "grpc://host:port",
+						Provider:   syncProviderGrpc,
+						ProviderID: "myapp",
+						CertPath:   "/tmp/ca.cert",
+						Selector:   "source=database",
+						MaxMsgSize: 10,
+					},
+				},
+			},
+			wantSyncs: []sync.ISync{
+				&grpc.Sync{},
+			},
+			wantErr: false,
+		},
 		{
 			name: "combined",
 			injectFunc: func(builder *SyncBuilder) {

core/pkg/sync/grpc/grpc_sync.go

@@ -48,6 +48,7 @@ type Sync struct {
 	Secure            bool
 	Selector          string
 	URI               string
+	MaxMsgSize        int
 
 	client FlagSyncServiceClient
 	ready  bool
@@ -62,7 +63,17 @@ func (g *Sync) Init(_ context.Context) error {
 	}
 
 	// Derive reusable client connection
-	rpcCon, err := grpc.NewClient(g.URI, grpc.WithTransportCredentials(tCredentials))
+	// Set MaxMsgSize if passed
+	var rpcCon *grpc.ClientConn
+
+	if g.MaxMsgSize > 0 {
+		g.Logger.Info(fmt.Sprintf("setting max receive message size %d bytes default 4MB", g.MaxMsgSize))
+		dialOptions := grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(g.MaxMsgSize))
+		rpcCon, err = grpc.NewClient(g.URI, grpc.WithTransportCredentials(tCredentials), dialOptions)
+	} else {
+		rpcCon, err = grpc.NewClient(g.URI, grpc.WithTransportCredentials(tCredentials))
+	}
+
 	if err != nil {
 		err := fmt.Errorf("error initiating grpc client connection: %w", err)
 		g.Logger.Error(err.Error())

core/pkg/sync/grpc/grpc_sync_test.go

@@ -19,6 +19,8 @@ import (
 	grpcmock "github.com/open-feature/flagd/core/pkg/sync/grpc/mock"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/mock/gomock"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zaptest/observer"
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
@@ -78,6 +80,30 @@ func Test_InitWithMockCredentialBuilder(t *testing.T) {
 	}
 }
 
+func Test_InitWithSizeOverride(t *testing.T) {
+	observedZapCore, observedLogs := observer.New(zap.InfoLevel)
+	observedLogger := zap.New(observedZapCore)
+
+	mockCtrl := gomock.NewController(t)
+	mockCredentialBulder := credendialsmock.NewMockBuilder(mockCtrl)
+
+	mockCredentialBulder.EXPECT().
+		Build(gomock.Any(), gomock.Any()).
+		Return(insecure.NewCredentials(), nil)
+
+	grpcSync := Sync{
+		URI:               "grpc-target",
+		Logger:            logger.NewLogger(observedLogger, false),
+		CredentialBuilder: mockCredentialBulder,
+		MaxMsgSize:        10,
+	}
+
+	err := grpcSync.Init(context.Background())
+
+	require.Nilf(t, err, "%s: expected no error, but got non nil error", t.Name())
+	require.Equal(t, "setting max receive message size 10 bytes default 4MB", observedLogs.All()[0].Message)
+}
+
 func Test_ReSyncTests(t *testing.T) {
 	const target = "localBufCon"
 

core/pkg/sync/isync.go

@@ -73,4 +73,5 @@ type SourceConfig struct {
 	ProviderID  string `json:"providerID,omitempty"`
 	Selector    string `json:"selector,omitempty"`
 	Interval    uint32 `json:"interval,omitempty"`
+	MaxMsgSize  int    `json:"maxMsgSize,omitempty"`
 }

docs/reference/sync-configuration.md

@@ -38,6 +38,7 @@ Alternatively, these configurations can be passed to flagd via config file, spec
 | providerID  | optional `string`  | Value binds to grpc connection's providerID field. gRPC server implementations may use this to identify connecting flagd instance                                                                                |
 | selector    | optional `string`  | Value binds to grpc connection's selector field. gRPC server implementations may use this to filter flag configurations                                                                                          |
 | certPath    | optional `string`  | Used for grpcs sync when TLS certificate is needed. If not provided, system certificates will be used for TLS connection                                                                                         |
+| maxMsgSize  | optional `int`     | Used for gRPC sync to set max receive message size (in bytes) e.g. 5242880 for 5MB. If not provided, the default is [4MB](https://pkg.go.dev/google.golang.org#grpc#MaxCallRecvMsgSize)                       |
 
 The `uri` field values **do not** follow the [URI patterns](#uri-patterns). The provider type is instead derived
 from the `provider` field. Only exception is the remote provider where `http(s)://` is expected by default. Incorrect
@@ -64,6 +65,7 @@ Startup command:
             {"uri":"https://secure-remote/basic-auth","provider":"http","authHeader":"Basic dXNlcjpwYXNz"},
             {"uri":"default/my-flag-config","provider":"kubernetes"},
             {"uri":"grpc-source:8080","provider":"grpc"},
+            {"uri":"my-flag-source:8080","provider":"grpc", "maxMsgSize": 5242880},
             {"uri":"my-flag-source:8080","provider":"grpc", "certPath": "/certs/ca.cert", "tls": true, "providerID": "flagd-weatherapp-sidecar", "selector": "source=database,app=weatherapp"}]'
 ```
 
@@ -80,6 +82,9 @@ sources:
     provider: kubernetes
   - uri: my-flag-source:8080
     provider: grpc
+  - uri: my-flag-source:8080
+    provider: grpc
+    maxMsgSize: 5242880
   - uri: my-flag-source:8080
     provider: grpc
     certPath: /certs/ca.cert