Add message size and time to receive to RPC trace events

comm.go

@@ -63,6 +63,10 @@ func (p *PubSub) handleNewStream(s network.Stream) {
 
 	r := msgio.NewVarintReaderSize(s, p.maxMessageSize)
 	for {
+		// ignore the values. We only want to know when the first bytes came in.
+		_, _ = r.NextMsgLen()
+		// Start the time once we've received the message length
+		start := time.Now()
 		msgbytes, err := r.ReadMsg()
 		if err != nil {
 			r.ReleaseMsg(msgbytes)
@@ -84,6 +88,7 @@ func (p *PubSub) handleNewStream(s network.Stream) {
 		rpc := new(RPC)
 		err = rpc.Unmarshal(msgbytes)
 		r.ReleaseMsg(msgbytes)
+		rpc.timeToReceive = time.Since(start)
 		if err != nil {
 			s.Reset()
 			log.Warnf("bogus rpc from %s: %s", s.Conn().RemotePeer(), err)

pb/trace.proto

@@ -107,11 +107,13 @@ message TraceEvent {
     repeated MessageMeta messages = 1;
     repeated SubMeta subscription = 2;
     optional ControlMeta control = 3;
+    optional uint64 millisecondsToReceive = 4;
   }
 
   message MessageMeta {
     optional bytes messageID = 1;
     optional string topic = 2;
+    optional uint64 messageSize = 3;
   }
 
   message SubMeta {

pubsub.go

@@ -255,6 +255,9 @@ type RPC struct {
 
 	// unexported on purpose, not sending this over the wire
 	from peer.ID
+
+	// timeToReceive is how long it took to receive the message.
+	timeToReceive time.Duration
 }
 
 // split splits the given RPC If a sub RPC is too large and can't be split