diff --git a/conn/node.go b/conn/node.go index 038b0776c9a..a33b9236d81 100644 --- a/conn/node.go +++ b/conn/node.go @@ -225,11 +225,21 @@ func (n *Node) SetPeer(pid uint64, addr string) { n.peers[pid] = addr } -func (n *Node) Send(m raftpb.Message) { - x.AssertTruef(n.Id != m.To, "Sending message to itself") - data, err := m.Marshal() +func (n *Node) Send(msg raftpb.Message) { + x.AssertTruef(n.Id != msg.To, "Sending message to itself") + data, err := msg.Marshal() x.Check(err) + if glog.V(2) { + switch msg.Type { + case raftpb.MsgHeartbeat, raftpb.MsgHeartbeatResp: + case raftpb.MsgReadIndex, raftpb.MsgReadIndexResp: + case raftpb.MsgApp, raftpb.MsgAppResp: + case raftpb.MsgProp: + default: + glog.Infof("RaftComm: [%#x] Sending message of type %s to %#x", msg.From, msg.Type, msg.To) + } + } // As long as leadership is stable, any attempted Propose() calls should be reflected in the // next raft.Ready.Messages. Leaders will send MsgApps to the followers; followers will send // MsgProp to the leader. It is up to the transport layer to get those messages to their @@ -243,7 +253,7 @@ func (n *Node) Send(m raftpb.Message) { // node. But, we shouldn't take the liberty to do that here. It would take us more time to // repropose these dropped messages anyway, than to block here a bit waiting for the messages // channel to clear out. - n.messages <- sendmsg{to: m.To, data: data} + n.messages <- sendmsg{to: msg.To, data: data} } func (n *Node) Snapshot() (raftpb.Snapshot, error) { diff --git a/conn/raft_server.go b/conn/raft_server.go index ac394f9aa11..03a04077d4c 100644 --- a/conn/raft_server.go +++ b/conn/raft_server.go @@ -225,6 +225,16 @@ func (w *RaftServer) RaftMessage(server pb.Raft_RaftMessageServer) error { // This should be done in order, and not via a goroutine. // Step can block forever. See: https://github.com/etcd-io/etcd/issues/10585 // So, add a context with timeout to allow it to get out of the blockage. + if glog.V(2) { + switch msg.Type { + case raftpb.MsgHeartbeat, raftpb.MsgHeartbeatResp: + case raftpb.MsgReadIndex, raftpb.MsgReadIndexResp: + case raftpb.MsgApp, raftpb.MsgAppResp: + case raftpb.MsgProp: + default: + glog.Infof("RaftComm: [%#x] Received msg of type: %s from %#x", msg.To, msg.Type, msg.From) + } + } if err := raft.Step(ctx, msg); err != nil { glog.Warningf("Error while raft.Step from %#x: %v. Closing RaftMessage stream.", rc.GetId(), err)