Repository: matrixorigin/matrixone
Pull request #6289 (2022-11-04 10:37:03)

[feat] logservice: make RPC message max size configurable
Configure it by new parameter max-message-size in log.toml.

## What type of PR is this?

- [ ] API-change
- [ ] BUG
- [ ] Improvement
- [ ] Documentation
- [x] Feature
- [ ] Test and CI
- [ ] Code Refactoring

## Which issue(s) this PR fixes:

issue #6238 

## What this PR does / why we need it:

Add a new configuration for logservice to set RPC message size max-message-size, whose default value is 100M.


Patch:
diff --git a/pkg/cnservice/tae.go b/pkg/cnservice/tae.go
index 2af1157f1..da85035be 100644
--- a/pkg/cnservice/tae.go
+++ b/pkg/cnservice/tae.go
@@ -55,6 +55,7 @@ func initTAE(
 				LogShardID:       pu.SV.LogShardID,
 				DNReplicaID:      pu.SV.DNReplicaID,
 				ServiceAddresses: cfg.HAKeeper.ClientConfig.ServiceAddresses,
+				MaxMessageSize:   int(cfg.RPC.MaxMessageSize),
 			})
 			cancel()
 			return lc, err
diff --git a/pkg/dnservice/factory.go b/pkg/dnservice/factory.go
index 6e2ce1576..2272047e7 100644
--- a/pkg/dnservice/factory.go
+++ b/pkg/dnservice/factory.go
@@ -106,6 +106,7 @@ func (s *store) newLogServiceClient(shard metadata.DNShard) (logservice.Client,
 		LogShardID:       shard.LogShardID,
 		DNReplicaID:      shard.ReplicaID,
 		ServiceAddresses: s.cfg.HAKeeper.ClientConfig.ServiceAddresses,
+		MaxMessageSize:   int(s.cfg.RPC.MaxMessageSize),
 	})
 }
 
diff --git a/pkg/logservice/client.go b/pkg/logservice/client.go
index 24a6315d7..84fe83290 100644
--- a/pkg/logservice/client.go
+++ b/pkg/logservice/client.go
@@ -311,7 +311,7 @@ func connectToLogService(ctx context.Context,
 		addresses[i], addresses[j] = addresses[j], addresses[i]
 	})
 	for _, addr := range addresses {
-		cc, err := getRPCClient(ctx, addr, c.respPool)
+		cc, err := getRPCClient(ctx, addr, c.respPool, c.cfg.MaxMessageSize)
 		if err != nil {
 			e = err
 			continue
@@ -497,7 +497,7 @@ func (c *client) doGetTruncatedLsn(ctx context.Context) (Lsn, error) {
 	return resp.LogResponse.Lsn, nil
 }
 
-func getRPCClient(ctx context.Context, target string, pool *sync.Pool) (morpc.RPCClient, error) {
+func getRPCClient(ctx context.Context, target string, pool *sync.Pool, maxMessageSize int) (morpc.RPCClient, error) {
 	mf := func() morpc.Message {
 		return pool.Get().(*RPCResponse)
 	}
@@ -524,7 +524,8 @@ func getRPCClient(ctx context.Context, target string, pool *sync.Pool) (morpc.RP
 	// to be attempted
 	codec := morpc.NewMessageCodec(mf,
 		morpc.WithCodecPayloadCopyBufferSize(defaultWriteSocketSize),
-		morpc.WithCodecEnableChecksum())
+		morpc.WithCodecEnableChecksum(),
+		morpc.WithCodecMaxBodySize(maxMessageSize))
 	bf := morpc.NewGoettyBasedBackendFactory(codec, backendOpts...)
 	return morpc.NewClient(bf, clientOpts...)
 }
diff --git a/pkg/logservice/config.go b/pkg/logservice/config.go
index 4bc563115..400fa4fb1 100644
--- a/pkg/logservice/config.go
+++ b/pkg/logservice/config.go
@@ -41,6 +41,7 @@ const (
 	defaultLogDBBufferSize     = 768 * 1024
 	defaultTruncateInterval    = 10 * time.Second
 	defaultMaxExportedSnapshot = 20
+	defaultMaxMessageSize      = 1024 * 1024 * 100
 )
 
 // Config defines the Configurations supported by the Log Service.
@@ -106,6 +107,11 @@ type Config struct {
 	// process truncate.
 	TruncateInterval toml.Duration `toml:"truncate-interval"`
 
+	RPC struct {
+		// MaxMessageSize is the max size for RPC message. The default value is 10MiB.
+		MaxMessageSize toml.ByteSize `toml:"max-message-size"`
+	}
+
 	// BootstrapConfig is the configuration specified for the bootstrapping
 	// procedure. It only needs to be specified for Log Stores selected to host
 	// initial HAKeeper replicas during bootstrapping.
@@ -271,6 +277,9 @@ func (c *Config) Validate() error {
 	if c.TruncateInterval.Duration == 0 {
 		return moerr.NewBadConfig("TruncateInterval not set")
 	}
+	if c.RPC.MaxMessageSize == 0 {
+		return moerr.NewBadConfig("MaxMessageSize not set")
+	}
 	// validate BootstrapConfig
 	if c.BootstrapConfig.BootstrapCluster {
 		if c.BootstrapConfig.NumOfLogShards == 0 {
@@ -364,6 +373,9 @@ func (c *Config) Fill() {
 	if c.TruncateInterval.Duration == 0 {
 		c.TruncateInterval.Duration = defaultTruncateInterval
 	}
+	if c.RPC.MaxMessageSize == 0 {
+		c.RPC.MaxMessageSize = toml.ByteSize(defaultMaxMessageSize)
+	}
 }
 
 // HAKeeperClientConfig is the config for HAKeeper clients.
@@ -401,6 +413,8 @@ type ClientConfig struct {
 	// LogService nodes service addresses. This field is provided for testing
 	// purposes only.
 	ServiceAddresses []string
+	// MaxMessageSize is the max message size for RPC.
+	MaxMessageSize int
 }
 
 // Validate validates the ClientConfig.
diff --git a/pkg/logservice/hakeeper_client.go b/pkg/logservice/hakeeper_client.go
index 0b2659856..6df9b9225 100644
--- a/pkg/logservice/hakeeper_client.go
+++ b/pkg/logservice/hakeeper_client.go
@@ -370,7 +370,7 @@ func connectToHAKeeper(ctx context.Context,
 		addresses[i], addresses[j] = addresses[j], addresses[i]
 	})
 	for _, addr := range addresses {
-		cc, err := getRPCClient(ctx, addr, c.respPool)
+		cc, err := getRPCClient(ctx, addr, c.respPool, defaultMaxMessageSize)
 		if err != nil {
 			e = err
 			continue
diff --git a/pkg/logservice/service.go b/pkg/logservice/service.go
index b57808d71..ab63c6287 100644
--- a/pkg/logservice/service.go
+++ b/pkg/logservice/service.go
@@ -126,7 +126,8 @@ func NewService(
 	// TODO: check and fix all these magic numbers
 	codec := morpc.NewMessageCodec(mf,
 		morpc.WithCodecPayloadCopyBufferSize(16*1024),
-		morpc.WithCodecEnableChecksum())
+		morpc.WithCodecEnableChecksum(),
+		morpc.WithCodecMaxBodySize(int(cfg.RPC.MaxMessageSize)))
 	server, err := morpc.NewRPCServer(LogServiceRPCName, cfg.ServiceListenAddress, codec,
 		morpc.WithServerGoettyOptions(goetty.WithSessionReleaseMsgFunc(func(i interface{}) {
 			respPool.Put(i.(morpc.RPCMessage).Message)
diff --git a/pkg/logservice/shardinfo.go b/pkg/logservice/shardinfo.go
index 452488dbd..fd224165b 100644
--- a/pkg/logservice/shardinfo.go
+++ b/pkg/logservice/shardinfo.go
@@ -43,7 +43,7 @@ func GetShardInfo(address string, shardID uint64) (ShardInfo, bool, error) {
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 	defer cancel()
-	cc, err := getRPCClient(ctx, address, respPool)
+	cc, err := getRPCClient(ctx, address, respPool, defaultMaxMessageSize)
 	if err != nil {
 		return ShardInfo{}, false, err
 	}


Test patch:
diff --git a/pkg/logservice/client_test.go b/pkg/logservice/client_test.go
index 3d424939e..d407f377c 100644
--- a/pkg/logservice/client_test.go
+++ b/pkg/logservice/client_test.go
@@ -37,8 +37,21 @@ import (
 
 var testLogger = logutil.GetGlobalLogger().Named("logservice-test")
 
-func runClientTest(t *testing.T,
-	readOnly bool, fn func(*testing.T, *Service, ClientConfig, Client)) {
+var getClientConfig = func(readOnly bool) ClientConfig {
+	return ClientConfig{
+		ReadOnly:         readOnly,
+		LogShardID:       1,
+		DNReplicaID:      2,
+		ServiceAddresses: []string{testServiceAddress},
+		MaxMessageSize:   defaultMaxMessageSize,
+	}
+}
+
+func runClientTest(
+	t *testing.T,
+	readOnly bool,
+	cCfgFn func(bool) ClientConfig,
+	fn func(*testing.T, *Service, ClientConfig, Client)) {
 	defer leaktest.AfterTest(t)()
 	cfg := getServiceTestConfig()
 	defer vfs.ReportLeakedFD(cfg.FS, t)
@@ -57,12 +70,10 @@ func runClientTest(t *testing.T,
 	init[2] = service.ID()
 	assert.NoError(t, service.store.startReplica(1, 2, init, false))
 
-	scfg := ClientConfig{
-		ReadOnly:         readOnly,
-		LogShardID:       1,
-		DNReplicaID:      2,
-		ServiceAddresses: []string{testServiceAddress},
+	if cCfgFn == nil {
+		cCfgFn = getClientConfig
 	}
+	scfg := cCfgFn(readOnly)
 
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 	defer cancel()
@@ -88,7 +99,7 @@ func TestClientCanBeReset(t *testing.T) {
 		client.resetClient()
 		assert.Nil(t, client.client)
 	}
-	runClientTest(t, false, fn)
+	runClientTest(t, false, nil, fn)
 }
 
 func TestPrepareClient(t *testing.T) {
@@ -102,7 +113,7 @@ func TestPrepareClient(t *testing.T) {
 		assert.NoError(t, client.prepareClient(ctx))
 		assert.NotNil(t, client.client)
 	}
-	runClientTest(t, false, fn)
+	runClientTest(t, false, nil, fn)
 }
 
 func TestLogShardNotFoundErrorIsConsideredAsTempError(t *testing.T) {
@@ -115,14 +126,14 @@ func TestLogShardNotFoundErrorIsConsideredAsTempError(t *testing.T) {
 		client := c.(*managedClient)
 		assert.True(t, client.isRetryableError(err))
 	}
-	runClientTest(t, false, fn)
+	runClientTest(t, false, nil, fn)
 }
 
 func TestClientCanBeCreated(t *testing.T) {
 	fn := func(t *testing.T, s *Service, cfg ClientConfig, c Client) {
 	}
-	runClientTest(t, false, fn)
-	runClientTest(t, true, fn)
+	runClientTest(t, false, nil, fn)
+	runClientTest(t, true, nil, fn)
 }
 
 func TestClientCanBeConnectedByReverseProxy(t *testing.T) {
@@ -196,7 +207,7 @@ func TestClientGetTSOTimestamp(t *testing.T) {
 		require.NoError(t, err)
 		assert.Equal(t, uint64(1101), v)
 	}
-	runClientTest(t, false, fn)
+	runClientTest(t, false, nil, fn)
 }
 
 func TestClientAppend(t *testing.T) {
@@ -218,7 +229,7 @@ func TestClientAppend(t *testing.T) {
 		_, err = c.Append(ctx, pb.LogRecord{Data: cmd})
 		assert.True(t, moerr.IsMoErrCode(err, moerr.ErrNotLeaseHolder))
 	}
-	runClientTest(t, false, fn)
+	runClientTest(t, false, nil, fn)
 }
 
 // FIXME: actually enforce allowed allocation
@@ -234,7 +245,7 @@ func TestClientAppendAlloc(t *testing.T) {
 		})
 		testLogger.Info(fmt.Sprintf("ac: %f", ac))
 	}
-	runClientTest(t, false, fn)
+	runClientTest(t, false, nil, fn)
 }
 
 func TestClientRead(t *testing.T) {
@@ -264,7 +275,7 @@ func TestClientRead(t *testing.T) {
 		_, _, err = c.Read(ctx, 6, math.MaxUint64)
 		assert.True(t, errors.Is(err, dragonboat.ErrInvalidRange))
 	}
-	runClientTest(t, false, fn)
+	runClientTest(t, false, nil, fn)
 }
 
 func TestClientTruncate(t *testing.T) {
@@ -285,7 +296,7 @@ func TestClientTruncate(t *testing.T) {
 		err = c.Truncate(ctx, 3)
 		assert.True(t, moerr.IsMoErrCode(err, moerr.ErrInvalidTruncateLsn))
 	}
-	runClientTest(t, false, fn)
+	runClientTest(t, false, nil, fn)
 }
 
 func TestReadOnlyClientRejectWriteRequests(t *testing.T) {
@@ -299,5 +310,50 @@ func TestReadOnlyClientRejectWriteRequests(t *testing.T) {
 		err = c.Truncate(ctx, 4)
 		require.True(t, moerr.IsMoErrCode(err, moerr.ErrInvalidInput))
 	}
-	runClientTest(t, true, fn)
+	runClientTest(t, true, nil, fn)
+}
+
+func TestClientSendWithMsgSize(t *testing.T) {
+	cFn := func(readOnly bool) ClientConfig {
+		return ClientConfig{
+			ReadOnly:         readOnly,
+			LogShardID:       1,
+			DNReplicaID:      2,
+			ServiceAddresses: []string{testServiceAddress},
+			MaxMessageSize:   testServerMaxMsgSize - 100,
+		}
+	}
+
+	fn := func(t *testing.T, s *Service, cfg ClientConfig, c Client) {
+		rec := c.GetLogRecord(testServerMaxMsgSize - 80)
+		rand.Read(rec.Payload())
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+		_, err := c.Append(ctx, rec)
+		require.NoError(t, err)
+
+		// client only accepts message whose size less than 190
+		_, _, err = c.Read(ctx, 4, math.MaxUint64)
+		require.Error(t, err)
+	}
+	runClientTest(t, false, cFn, fn)
+}
+
+func TestServerReceiveWithMsgSize(t *testing.T) {
+	fn := func(t *testing.T, s *Service, cfg ClientConfig, c Client) {
+		rec := c.GetLogRecord(16)
+		rand.Read(rec.Payload())
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+		_, err := c.Append(ctx, rec)
+		require.NoError(t, err)
+
+		rec = c.GetLogRecord(testServerMaxMsgSize + 20)
+		rand.Read(rec.Payload())
+		ctx, cancel = context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+		_, err = c.Append(ctx, rec)
+		require.Error(t, err)
+	}
+	runClientTest(t, false, nil, fn)
 }
diff --git a/pkg/logservice/hakeeper_client_test.go b/pkg/logservice/hakeeper_client_test.go
index 999cbae63..3310463e4 100644
--- a/pkg/logservice/hakeeper_client_test.go
+++ b/pkg/logservice/hakeeper_client_test.go
@@ -349,7 +349,7 @@ func testNotHAKeeperErrorIsHandled(t *testing.T, fn func(*testing.T, *managedHAK
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 	defer cancel()
-	cc, err := getRPCClient(ctx, cfg1.ServiceAddress, c.respPool)
+	cc, err := getRPCClient(ctx, cfg1.ServiceAddress, c.respPool, defaultMaxMessageSize)
 	require.NoError(t, err)
 	c.addr = cfg1.ServiceAddress
 	c.client = cc
diff --git a/pkg/logservice/service_test.go b/pkg/logservice/service_test.go
index 0d685ca00..d118c21b5 100644
--- a/pkg/logservice/service_test.go
+++ b/pkg/logservice/service_test.go
@@ -39,6 +39,7 @@ const (
 	testServiceAddress     = "127.0.0.1:9000"
 	testGossipAddress      = "127.0.0.1:9010"
 	dummyGossipSeedAddress = "127.0.0.1:9100"
+	testServerMaxMsgSize   = 200
 )
 
 func getServiceTestConfig() Config {
@@ -55,6 +56,7 @@ func getServiceTestConfig() Config {
 		DisableWorkers:       true,
 		UseTeeLogDB:          true,
 	}
+	c.RPC.MaxMessageSize = testServerMaxMsgSize
 	c.Fill()
 	return c
 }