feat: treat "Bitswap" attempt as a parent, record total bytes

pkg/aggregateeventrecorder/aggregateeventrecorder.go

@@ -11,6 +11,7 @@ import (
 	"github.com/filecoin-project/lassie/pkg/events"
 	"github.com/filecoin-project/lassie/pkg/types"
 	"github.com/ipfs/go-log/v2"
+	"github.com/multiformats/go-multicodec"
 )
 
 var logger = log.Logger("lassie/aggregateeventrecorder")
@@ -209,6 +210,15 @@ func (a *aggregateEventRecorder) ingestEvents() {
 					tempData.retrievalAttempts[spid] = attempt
 				}
 				attempt.BytesTransferred += ret.ByteCount()
+				fmt.Println("data received", spid, ret.ByteCount())
+				if ret.Protocol() == multicodec.TransportBitswap {
+					// record the total under the bitswap identifier as well
+					if _, ok := tempData.retrievalAttempts[types.BitswapIndentifier]; ok {
+						tempData.retrievalAttempts[types.BitswapIndentifier].BytesTransferred += ret.ByteCount()
+						fmt.Println("data received", types.BitswapIndentifier, ret.ByteCount())
+					}
+				}
+
 			case events.FailedRetrievalEvent:
 				// Add an error message to the retrieval attempt
 				spid := events.Identifier(ret)

pkg/aggregateeventrecorder/aggregateeventrecorder_test.go

@@ -123,9 +123,10 @@ func TestAggregateEventRecorder(t *testing.T) {
 				verifyIntNode(t, sp2Attempt, "bytesTransferred", 5000)
 				bitswapAttempt, err := retrievalAttempts.LookupByString(types.BitswapIndentifier)
 				require.NoError(t, err)
-				require.Equal(t, int64(2), bitswapAttempt.Length())
+				require.Equal(t, int64(3), bitswapAttempt.Length())
 				verifyStringNode(t, bitswapAttempt, "timeToFirstByte", "20ms")
 				verifyStringNode(t, bitswapAttempt, "protocol", multicodec.TransportBitswap.String())
+				verifyIntNode(t, bitswapAttempt, "bytesTransferred", 10000)
 				bitswapPeer1Attempt, err := retrievalAttempts.LookupByString(bitswapCandidates[0].MinerPeer.ID.String())
 				require.NoError(t, err)
 				require.Equal(t, int64(2), bitswapPeer1Attempt.Length())