Raise gRPC message size limits

Author: bootjp

adapter/test_util.go

@@ -13,6 +13,7 @@ import (
 	"github.com/Jille/raft-grpc-leader-rpc/leaderhealth"
 	transport "github.com/Jille/raft-grpc-transport"
 	"github.com/Jille/raftadmin"
+	internalutil "github.com/bootjp/elastickv/internal"
 	"github.com/bootjp/elastickv/kv"
 	pb "github.com/bootjp/elastickv/proto"
 	"github.com/bootjp/elastickv/store"
@@ -23,7 +24,6 @@ import (
 	"github.com/stretchr/testify/require"
 	"golang.org/x/sys/unix"
 	"google.golang.org/grpc"
-	"google.golang.org/grpc/credentials/insecure"
 )
 
 func shutdown(nodes []Node) {
@@ -350,7 +350,7 @@ func setupNodes(t *testing.T, ctx context.Context, n int, ports []portsAdress) (
 		r, tm, err := newRaft(strconv.Itoa(i), port.raftAddress, fsm, i == 0, cfg, electionTimeout)
 		assert.NoError(t, err)
 
-		s := grpc.NewServer()
+		s := grpc.NewServer(internalutil.GRPCServerOptions()...)
 		trx := kv.NewTransaction(r)
 		coordinator := kv.NewCoordinator(trx, r)
 		relay := NewRedisPubSubRelay()
@@ -416,9 +416,7 @@ func newRaft(myID string, myAddress string, fsm raft.FSM, bootstrap bool, cfg ra
 		Level: hclog.LevelFromString("WARN"),
 	})
 
-	tm := transport.New(raft.ServerAddress(myAddress), []grpc.DialOption{
-		grpc.WithTransportCredentials(insecure.NewCredentials()),
-	})
+	tm := transport.New(raft.ServerAddress(myAddress), internalutil.GRPCDialOptions())
 
 	r, err := raft.NewRaft(c, fsm, ldb, sdb, fss, tm.Transport())
 	if err != nil {

cmd/server/demo.go

@@ -30,7 +30,6 @@ import (
 	"github.com/hashicorp/raft"
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc"
-	"google.golang.org/grpc/credentials/insecure"
 )
 
 var (
@@ -206,7 +205,7 @@ func joinCluster(ctx context.Context, nodes []config) error {
 	}
 
 	// Connect to leader
-	conn, err := grpc.NewClient(leader.address, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	conn, err := grpc.NewClient(leader.address, internalutil.GRPCDialOptions()...)
 	if err != nil {
 		return fmt.Errorf("failed to dial leader: %w", err)
 	}
@@ -322,7 +321,7 @@ func setupStorage(dir string) (raft.LogStore, raft.StableStore, raft.SnapshotSto
 }
 
 func setupGRPC(r *raft.Raft, st store.MVCCStore, tm *transport.Manager, coordinator *kv.Coordinate, distServer *adapter.DistributionServer, relay *adapter.RedisPubSubRelay) (*grpc.Server, *adapter.GRPCServer) {
-	s := grpc.NewServer()
+	s := grpc.NewServer(internalutil.GRPCServerOptions()...)
 	trx := kv.NewTransaction(r)
 	routedStore := kv.NewLeaderRoutedStore(st, coordinator)
 	gs := adapter.NewGRPCServer(routedStore, coordinator, adapter.WithCloseStore())
@@ -381,9 +380,7 @@ func run(ctx context.Context, eg *errgroup.Group, cfg config) error {
 	})
 
 	// Transport
-	tm := transport.New(raft.ServerAddress(cfg.address), []grpc.DialOption{
-		grpc.WithTransportCredentials(insecure.NewCredentials()),
-	})
+	tm := transport.New(raft.ServerAddress(cfg.address), internalutil.GRPCDialOptions())
 
 	r, err := raft.NewRaft(c, fsm, ldb, sdb, fss, tm.Transport())
 	if err != nil {

internal/grpc.go

@@ -0,0 +1,29 @@
+package internal
+
+import (
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+const GRPCMaxMessageBytes = 64 << 20
+
+// GRPCServerOptions keeps Raft replication and the public/internal APIs aligned
+// on the same message-size budget.
+func GRPCServerOptions() []grpc.ServerOption {
+	return []grpc.ServerOption{
+		grpc.MaxRecvMsgSize(GRPCMaxMessageBytes),
+		grpc.MaxSendMsgSize(GRPCMaxMessageBytes),
+	}
+}
+
+// GRPCDialOptions returns the common insecure dial options used by node-local
+// and node-to-node traffic.
+func GRPCDialOptions() []grpc.DialOption {
+	return []grpc.DialOption{
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithDefaultCallOptions(
+			grpc.MaxCallRecvMsgSize(GRPCMaxMessageBytes),
+			grpc.MaxCallSendMsgSize(GRPCMaxMessageBytes),
+		),
+	}
+}

kv/grpc_conn_cache.go

@@ -3,11 +3,11 @@ package kv
 import (
 	"sync"
 
+	internalutil "github.com/bootjp/elastickv/internal"
 	"github.com/cockroachdb/errors"
 	"github.com/hashicorp/raft"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/connectivity"
-	"google.golang.org/grpc/credentials/insecure"
 )
 
 // GRPCConnCache reuses gRPC connections per address. gRPC itself handles
@@ -75,9 +75,9 @@ func (c *GRPCConnCache) ConnFor(addr raft.ServerAddress) (*grpc.ClientConn, erro
 		return conn, nil
 	}
 
-	conn, err := grpc.NewClient(string(addr),
-		grpc.WithTransportCredentials(insecure.NewCredentials()),
-		grpc.WithDefaultCallOptions(grpc.WaitForReady(true)),
+	conn, err := grpc.NewClient(
+		string(addr),
+		append(internalutil.GRPCDialOptions(), grpc.WithDefaultCallOptions(grpc.WaitForReady(true)))...,
 	)
 	if err != nil {
 		return nil, errors.WithStack(err)

main.go

@@ -287,7 +287,7 @@ func raftMonitorRuntimes(runtimes []*raftGroupRuntime) []monitoring.RaftRuntime
 
 func startRaftServers(ctx context.Context, lc *net.ListenConfig, eg *errgroup.Group, runtimes []*raftGroupRuntime, shardStore *kv.ShardStore, coordinate kv.Coordinator, distServer *adapter.DistributionServer, relay *adapter.RedisPubSubRelay) error {
 	for _, rt := range runtimes {
-		gs := grpc.NewServer()
+		gs := grpc.NewServer(internalutil.GRPCServerOptions()...)
 		trx := kv.NewTransaction(rt.raft)
 		grpcSvc := adapter.NewGRPCServer(shardStore, coordinate)
 		pb.RegisterRawKVServer(gs, grpcSvc)

multiraft_runtime.go

@@ -7,12 +7,11 @@ import (
 	"time"
 
 	transport "github.com/Jille/raft-grpc-transport"
+	internalutil "github.com/bootjp/elastickv/internal"
 	"github.com/bootjp/elastickv/internal/raftstore"
 	"github.com/bootjp/elastickv/store"
 	"github.com/cockroachdb/errors"
 	"github.com/hashicorp/raft"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/credentials/insecure"
 )
 
 type raftGroupRuntime struct {
@@ -117,9 +116,7 @@ func newRaftGroup(raftID string, group groupSpec, baseDir string, multi bool, bo
 		return nil, nil, nil, errors.WithStack(err)
 	}
 
-	tm = transport.New(raft.ServerAddress(group.address), []grpc.DialOption{
-		grpc.WithTransportCredentials(insecure.NewCredentials()),
-	})
+	tm = transport.New(raft.ServerAddress(group.address), internalutil.GRPCDialOptions())
 
 	r, err := raft.NewRaft(c, fsm, raftStore, raftStore, fss, tm.Transport())
 	if err != nil {