Increase gRPC request timeout to 20 seconds when sending snapshots

Signed-off-by: Nishant Totla <nishanttotla@gmail.com>

manager/state/raft/raft.go

@@ -180,9 +180,12 @@ type NodeOptions struct {
 	ClockSource clock.Clock
 	// SendTimeout is the timeout on the sending messages to other raft
 	// nodes. Leave this as 0 to get the default value.
-	SendTimeout    time.Duration
-	TLSCredentials credentials.TransportCredentials
-	KeyRotator     EncryptionKeyRotator
+	SendTimeout time.Duration
+	// SendSnapshotTimeout is the timeout on the sending snapshots to other raft
+	// nodes. Leave this as 0 to get the default value.
+	SendSnapshotTimeout time.Duration
+	TLSCredentials      credentials.TransportCredentials
+	KeyRotator          EncryptionKeyRotator
 	// DisableStackDump prevents Run from dumping goroutine stacks when the
 	// store becomes stuck.
 	DisableStackDump bool
@@ -204,6 +207,11 @@ func NewNode(opts NodeOptions) *Node {
 	if opts.SendTimeout == 0 {
 		opts.SendTimeout = 2 * time.Second
 	}
+	if opts.SendSnapshotTimeout == 0 {
+		// a "slow" 100Mbps connection can send over 240MB data in 20 seconds
+		// which is well over the gRPC message limit of 128MB allowed by SwarmKit
+		opts.SendSnapshotTimeout = 20 * time.Second
+	}
 
 	raftStore := raft.NewMemoryStorage()
 
@@ -347,10 +355,11 @@ func (n *Node) WithContext(ctx context.Context) (context.Context, context.Cancel
 
 func (n *Node) initTransport() {
 	transportConfig := &transport.Config{
-		HeartbeatInterval: time.Duration(n.Config.ElectionTick) * n.opts.TickInterval,
-		SendTimeout:       n.opts.SendTimeout,
-		Credentials:       n.opts.TLSCredentials,
-		Raft:              n,
+		HeartbeatInterval:   time.Duration(n.Config.ElectionTick) * n.opts.TickInterval,
+		SendTimeout:         n.opts.SendTimeout,
+		SendSnapshotTimeout: n.opts.SendSnapshotTimeout,
+		Credentials:         n.opts.TLSCredentials,
+		Raft:                n,
 	}
 	n.transport = transport.New(transportConfig)
 }

manager/state/raft/transport/mock_raft_test.go

@@ -58,9 +58,10 @@ func newMockRaft() (*mockRaft, error) {
 		updatedNodes:         make(chan updateInfo, 4096),
 	}
 	cfg := &Config{
-		HeartbeatInterval: 3 * time.Second,
-		SendTimeout:       2 * time.Second,
-		Raft:              mr,
+		HeartbeatInterval:   3 * time.Second,
+		SendTimeout:         2 * time.Second,
+		SendSnapshotTimeout: 20 * time.Second,
+		Raft:                mr,
 	}
 	tr := New(cfg)
 	mr.tr = tr

manager/state/raft/transport/peer.go

@@ -133,7 +133,13 @@ func (p *peer) resolveAddr(ctx context.Context, id uint64) (string, error) {
 }
 
 func (p *peer) sendProcessMessage(ctx context.Context, m raftpb.Message) error {
-	ctx, cancel := context.WithTimeout(ctx, p.tr.config.SendTimeout)
+	timeout := p.tr.config.SendTimeout
+	// if a snapshot is being sent, set timeout to SendSnapshotTimeout because
+	// sending snapshots can take more time than other messages sent between peers
+	if m.Type == raftpb.MsgSnap {
+		timeout = p.tr.config.SendSnapshotTimeout
+	}
+	ctx, cancel := context.WithTimeout(ctx, timeout)
 	defer cancel()
 	_, err := api.NewRaftClient(p.conn()).ProcessRaftMessage(ctx, &api.ProcessRaftMessageRequest{Message: &m})
 	if grpc.Code(err) == codes.NotFound && grpc.ErrorDesc(err) == membership.ErrMemberRemoved.Error() {

manager/state/raft/transport/transport.go

@@ -33,10 +33,11 @@ type Raft interface {
 
 // Config for Transport
 type Config struct {
-	HeartbeatInterval time.Duration
-	SendTimeout       time.Duration
-	Credentials       credentials.TransportCredentials
-	RaftID            string
+	HeartbeatInterval   time.Duration
+	SendTimeout         time.Duration
+	SendSnapshotTimeout time.Duration
+	Credentials         credentials.TransportCredentials
+	RaftID              string
 
 	Raft
 }