Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 12 additions & 3 deletions manager/state/raft/raft.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
// LargeSendTimeout is the timeout on the sending snapshots to other raft
// nodes. Leave this as 0 to get the default value.
LargeSendTimeout time.Duration
TLSCredentials credentials.TransportCredentials
KeyRotator EncryptionKeyRotator
// DisableStackDump prevents Run from dumping goroutine stacks when the
// store becomes stuck.
DisableStackDump bool
Expand All @@ -204,6 +207,11 @@ func NewNode(opts NodeOptions) *Node {
if opts.SendTimeout == 0 {
opts.SendTimeout = 2 * time.Second
}
if opts.LargeSendTimeout == 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.LargeSendTimeout = 20 * time.Second
}

raftStore := raft.NewMemoryStorage()

Expand Down Expand Up @@ -349,6 +357,7 @@ func (n *Node) initTransport() {
transportConfig := &transport.Config{
HeartbeatInterval: time.Duration(n.Config.ElectionTick) * n.opts.TickInterval,
SendTimeout: n.opts.SendTimeout,
LargeSendTimeout: n.opts.LargeSendTimeout,
Credentials: n.opts.TLSCredentials,
Raft: n,
}
Expand Down
1 change: 1 addition & 0 deletions manager/state/raft/transport/mock_raft_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,7 @@ func newMockRaft() (*mockRaft, error) {
cfg := &Config{
HeartbeatInterval: 3 * time.Second,
SendTimeout: 2 * time.Second,
LargeSendTimeout: 20 * time.Second,
Raft: mr,
}
tr := New(cfg)
Expand Down
9 changes: 8 additions & 1 deletion manager/state/raft/transport/peer.go
Original file line number Diff line number Diff line change
Expand Up @@ -133,7 +133,14 @@ 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 LargeSendTimeout because
// sending snapshots can take more time than other messages sent between peers.
// The same applies to AppendEntries as well, where messages can get large.
if m.Type == raftpb.MsgSnap || m.Type == raftpb.MsgApp {
timeout = p.tr.config.LargeSendTimeout
}
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() {
Expand Down
1 change: 1 addition & 0 deletions manager/state/raft/transport/transport.go
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@ type Raft interface {
type Config struct {
HeartbeatInterval time.Duration
SendTimeout time.Duration
LargeSendTimeout time.Duration
Credentials credentials.TransportCredentials
RaftID string

Expand Down