add more tracing data

consensus/msgs.go

@@ -72,7 +72,7 @@ func MsgToProto(msg Message) (*cmtcons.Message, error) {
 		return m.Wrap().(*cmtcons.Message), nil
 
 	case *HasBlockMessage:
-		m := &cmtcons.HasCompactBlock{
+		m := &cmtcons.HasBlock{
 			Height: msg.Height,
 			Round:  msg.Round,
 		}
@@ -216,7 +216,7 @@ func MsgFromProto(p *cmtcons.Message) (Message, error) {
 			Round: msg.Round,
 		}
 
-	case *cmtcons.HasCompactBlock:
+	case *cmtcons.HasBlock:
 		pb = &HasBlockMessage{
 			Height: msg.Height,
 			Round:  msg.Round,

consensus/reactor.go

@@ -397,8 +397,22 @@ func (conR *Reactor) ReceiveEnvelope(e p2p.Envelope) {
 		case *CompactBlockMessage:
 			ps.SetHasBlock(msg.Block.Height, msg.Round)
 			conR.conS.peerMsgQueue <- msgInfo{msg, e.Src.ID()}
+			schema.WriteCompactBlock(
+				conR.traceClient,
+				msg.Block.Height,
+				msg.Round,
+				string(e.Src.ID()),
+				schema.Download,
+			)
 		case *HasBlockMessage:
 			ps.SetHasBlock(msg.Height, msg.Round)
+			schema.WriteHasBlock(
+				conR.traceClient,
+				msg.Height,
+				msg.Round,
+				string(e.Src.ID()),
+				schema.Download,
+			)
 		case *BlockPartMessage:
 			ps.SetHasProposalBlockPart(msg.Height, msg.Round, int(msg.Part.Index))
 			conR.Metrics.BlockParts.With("peer_id", string(e.Src.ID())).Add(1)
@@ -609,11 +623,18 @@ func (conR *Reactor) broadcastHasVoteMessage(vote *types.Vote) {
 func (conR *Reactor) broadcastHasBlockMessage(data *types.EventDataCompleteProposal) {
 	conR.Switch.BroadcastEnvelope(p2p.Envelope{
 		ChannelID: DataChannel,
-		Message: &cmtcons.HasCompactBlock{
+		Message: &cmtcons.HasBlock{
 			Height: data.Height,
 			Round:  data.Round,
 		},
 	})
+	schema.WriteHasBlock(
+		conR.traceClient,
+		data.Height,
+		data.Round,
+		"",
+		schema.Upload,
+	)
 }
 
 func makeRoundStepMessage(rs *cstypes.RoundState) (nrsMsg *cmtcons.NewRoundStep) {
@@ -667,7 +688,7 @@ func (conR *Reactor) getRoundState() *cstypes.RoundState {
 }
 
 func (conR *Reactor) gossipDataRoutine(peer p2p.Peer, ps *PeerState) {
-	logger := conR.Logger.With("peer", peer)
+	logger := conR.Logger.With("peer", peer.ID())
 
 OUTER_LOOP:
 	for {
@@ -695,6 +716,13 @@ OUTER_LOOP:
 				}, logger) {
 					ps.SetHasBlock(prs.Height, prs.Round)
 					conR.conS.metrics.CompactBlocksSent.Add(1)
+					schema.WriteCompactBlock(
+						conR.traceClient,
+						prs.Height,
+						prs.Round,
+						string(peer.ID()),
+						schema.Upload,
+					)
 				}
 				continue OUTER_LOOP
 			}
@@ -1339,7 +1367,7 @@ func (ps *PeerState) SetHasProposalBlockPart(height int64, round int32, index in
 	ps.PRS.ProposalBlockParts.SetIndex(index, true)
 }
 
-// SetHasCompactBlock sets the given block part index as known for the peer.
+// SetHasBlock sets the given block part index as known for the peer.
 func (ps *PeerState) SetHasBlock(height int64, round int32) {
 	ps.mtx.Lock()
 	defer ps.mtx.Unlock()

mempool/cat/block_builder.go

@@ -45,6 +45,7 @@ func (memR *Reactor) FetchTxsFromKeys(ctx context.Context, blockID []byte, compa
 	if len(missingKeys) == 0 {
 		return txs, nil
 	}
+	initialNumMissing := len(missingKeys)
 
 	// setup a request for this block and begin to track and retrieve all missing transactions
 	request := memR.blockFetcher.newRequest(
@@ -54,7 +55,6 @@ func (memR *Reactor) FetchTxsFromKeys(ctx context.Context, blockID []byte, compa
 		txs,
 	)
 	defer func() {
-		initialNumMissing := len(missingKeys)
 		timeTaken := request.TimeTaken()
 		memR.Logger.Info("fetched txs", "timeTaken", timeTaken, "numRetrieved", initialNumMissing-len(request.missingKeys), "numMissing", len(request.missingKeys))
 	}()

pkg/trace/schema/consensus.go

@@ -15,6 +15,8 @@ func ConsensusTables() []string {
 		VoteTable,
 		ConsensusStateTable,
 		ProposalTable,
+		CompactBlockTable,
+		HasBlockTable,
 	}
 }
 
@@ -247,3 +249,60 @@ func WriteProposal(
 		TransferType: transferType,
 	})
 }
+
+const (
+	// CompactBlockTable is the name of the table that stores the consensus compact block
+	// parts.
+	CompactBlockTable = "consensus_compact_block"
+)
+
+type CompactBlock struct {
+	Height       int64        `json:"height"`
+	Round        int32        `json:"round"`
+	PeerID       string       `json:"peer_id"`
+	TransferType TransferType `json:"transfer_type"`
+}
+
+func (c CompactBlock) Table() string {
+	return CompactBlockTable
+}
+
+func WriteCompactBlock(
+	client trace.Tracer,
+	height int64,
+	round int32,
+	peerID string,
+	transferType TransferType,
+) {
+	client.Write(CompactBlock{
+		Height:       height,
+		Round:        round,
+		PeerID:       peerID,
+		TransferType: transferType,
+	})
+}
+
+const (
+	HasBlockTable = "consensus_has_block"
+)
+
+type HasBlock struct {
+	Height       int64        `json:"height"`
+	Round        int32        `json:"round"`
+	PeerID       string       `json:"peer_id"`
+	TransferType TransferType `json:"transfer_type"`
+}
+
+func (h HasBlock) Table() string {
+	return HasBlockTable
+}
+
+func WriteHasBlock(
+	client trace.Tracer,
+	height int64,
+	round int32,
+	peerID string,
+	transferType TransferType,
+) {
+	client.Write(HasBlock{Height: height, Round: round, PeerID: peerID, TransferType: transferType})
+}

proto/tendermint/consensus/message.go

@@ -49,9 +49,9 @@ func (m *CompactBlock) Wrap() proto.Message {
 	return cm
 }
 
-func (m *HasCompactBlock) Wrap() proto.Message {
+func (m *HasBlock) Wrap() proto.Message {
 	cm := &Message{}
-	cm.Sum = &Message_HasCompactBlock{HasCompactBlock: m}
+	cm.Sum = &Message_HasBlock{HasBlock: m}
 	return cm
 }
 
@@ -104,8 +104,8 @@ func (m *Message) Unwrap() (proto.Message, error) {
 	case *Message_CompactBlock:
 		return m.GetCompactBlock(), nil
 
-	case *Message_HasCompactBlock:
-		return m.GetHasCompactBlock(), nil
+	case *Message_HasBlock:
+		return m.GetHasBlock(), nil
 
 	case *Message_BlockPart:
 		return m.GetBlockPart(), nil