bitswap/network: add new metrics to compare bsnet and httpnet

Author: hsanjuan

bitswap/message/message.go

@@ -392,26 +392,30 @@ func BlockPresenceSize(c cid.Cid) int {
 }
 
 // FromNet generates a new BitswapMessage from incoming data on an io.Reader.
-func FromNet(r io.Reader) (BitSwapMessage, error) {
+func FromNet(r io.Reader) (BitSwapMessage, int, error) {
 	reader := msgio.NewVarintReaderSize(r, network.MessageSizeMax)
 	return FromMsgReader(reader)
 }
 
 // FromPBReader generates a new Bitswap message from a gogo-protobuf reader
-func FromMsgReader(r msgio.Reader) (BitSwapMessage, error) {
+func FromMsgReader(r msgio.Reader) (BitSwapMessage, int, error) {
 	msg, err := r.ReadMsg()
 	if err != nil {
-		return nil, err
+		return nil, 0, err
 	}
 
 	var pb pb.Message
 	err = pb.Unmarshal(msg)
 	r.ReleaseMsg(msg)
 	if err != nil {
-		return nil, err
+		return nil, 0, err
 	}
 
-	return newMessageFromProto(pb)
+	m, err := newMessageFromProto(pb)
+	if err != nil {
+		return nil, 0, err
+	}
+	return m, len(msg), nil
 }
 
 func (m *impl) ToProtoV0() *pb.Message {

bitswap/message/message_test.go

@@ -110,7 +110,7 @@ func TestToNetFromNetPreservesWantList(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	copied, err := FromNet(buf)
+	copied, _, err := FromNet(buf)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -143,7 +143,7 @@ func TestToAndFromNetMessage(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	m2, err := FromNet(buf)
+	m2, _, err := FromNet(buf)
 	if err != nil {
 		t.Fatal(err)
 	}

bitswap/network/bsnet/ipfs_impl.go

@@ -45,6 +45,8 @@ func NewFromIpfsHost(host host.Host, opts ...NetOpt) iface.BitSwapNetwork {
 		protocolBitswap:        s.ProtocolPrefix + ProtocolBitswap,
 
 		supportedProtocols: s.SupportedProtocols,
+
+		metrics: newMetrics(),
 	}
 
 	return &bitswapNetwork
@@ -80,6 +82,8 @@ type impl struct {
 
 	// inbound messages from the network are forwarded to the receiver
 	receivers []iface.Receiver
+
+	metrics *metrics
 }
 
 // interfaceWrapper is concrete type that wraps an interface. Necessary because
@@ -166,6 +170,15 @@ func (s *streamMessageSender) SupportsHave() bool {
 
 // Send a message to the peer, attempting multiple times
 func (s *streamMessageSender) SendMsg(ctx context.Context, msg bsmsg.BitSwapMessage) error {
+	if n := len(msg.Wantlist()); n > 0 {
+		s.bsnet.metrics.WantlistsTotal.Inc()
+		s.bsnet.metrics.WantlistsItemsTotal.Add(float64(n))
+		now := time.Now()
+		defer func() {
+			s.bsnet.metrics.WantlistsSeconds.Add(float64(time.Since(now)) / float64(time.Second))
+		}()
+	}
+
 	return s.multiAttempt(ctx, func() error {
 		return s.send(ctx, msg)
 	})
@@ -417,7 +430,7 @@ func (bsnet *impl) handleNewStream(s network.Stream) {
 
 	reader := msgio.NewVarintReaderSize(s, network.MessageSizeMax)
 	for {
-		received, err := bsmsg.FromMsgReader(reader)
+		received, size, err := bsmsg.FromMsgReader(reader)
 		if err != nil {
 			if err != io.EOF {
 				_ = s.Reset()
@@ -429,6 +442,8 @@ func (bsnet *impl) handleNewStream(s network.Stream) {
 			return
 		}
 
+		bsnet.metrics.ResponseSizes.Observe(float64(size))
+		bsnet.metrics.ResponseTotalBytes.Add(float64(size))
 		p := s.Conn().RemotePeer()
 		ctx := context.Background()
 		log.Debugf("bitswap net handleNewStream from %s", s.Conn().RemotePeer())

bitswap/network/httpnet/metrics.go

@@ -22,10 +22,38 @@ func requestsFailure(ctx context.Context) imetrics.Counter {
 	return imetrics.NewCtx(ctx, "requests_failure", "Failed (no response, dial error etc) requests count").Counter()
 }
 
+func requestSentBytes(ctx context.Context) imetrics.Counter {
+	return imetrics.NewCtx(ctx, "request_sent_bytes", "Total bytes sent on requests").Counter()
+}
+
+func requestTime(ctx context.Context) imetrics.Histogram {
+	return imetrics.NewCtx(ctx, "request_duration_seconds", "Histogram of request durations").Histogram(durationHistogramBuckets)
+}
+
 func requestsBodyFailure(ctx context.Context) imetrics.Counter {
 	return imetrics.NewCtx(ctx, "requests_body_failure", "Failure count when reading response body").Counter()
 }
 
+func responseSizes(ctx context.Context) imetrics.Histogram {
+	return imetrics.NewCtx(ctx, "response_bytes", "Histogram of http response sizes").Histogram(blockSizesHistogramBuckets)
+}
+
+func responseTotalBytes(ctx context.Context) imetrics.Counter {
+	return imetrics.NewCtx(ctx, "response_total_bytes", "Accumulated response bytes").Counter()
+}
+
+func wantlistsTotal(ctx context.Context) imetrics.Counter {
+	return imetrics.NewCtx(ctx, "wantlists_total", "Total number of wantlists sent").Counter()
+}
+
+func wantlistsItemsTotal(ctx context.Context) imetrics.Counter {
+	return imetrics.NewCtx(ctx, "wantlists_items_total", "Total number of elements in sent wantlists").Counter()
+}
+
+func wantlistsSeconds(ctx context.Context) imetrics.Counter {
+	return imetrics.NewCtx(ctx, "wantlists_seconds", "Number of seconds spent sending wantlists").Counter()
+}
+
 func statusNotFound(ctx context.Context) imetrics.Counter {
 	return imetrics.NewCtx(ctx, "status_404", "Request count with NotFound status").Counter()
 }
@@ -62,18 +90,16 @@ func statusOthers(ctx context.Context) imetrics.Counter {
 	return imetrics.NewCtx(ctx, "status_others", "Request count with other status codes").Counter()
 }
 
-func requestTime(ctx context.Context) imetrics.Histogram {
-	return imetrics.NewCtx(ctx, "request_duration_seconds", "Histogram of request durations").Histogram(durationHistogramBuckets)
-}
-
-func responseSize(ctx context.Context) imetrics.Histogram {
-	return imetrics.NewCtx(ctx, "response_bytes", "Histogram of http response sizes").Histogram(blockSizesHistogramBuckets)
-}
-
 type metrics struct {
 	RequestsInFlight                 imetrics.Gauge
 	RequestsTotal                    imetrics.Counter
 	RequestsFailure                  imetrics.Counter
+	RequestsSentBytes                imetrics.Counter
+	WantlistsTotal                   imetrics.Counter
+	WantlistsItemsTotal              imetrics.Counter
+	WantlistsSeconds                 imetrics.Counter
+	ResponseSizes                    imetrics.Histogram
+	ResponseTotalBytes               imetrics.Counter
 	RequestsBodyFailure              imetrics.Counter
 	StatusNotFound                   imetrics.Counter
 	StatusGone                       imetrics.Counter
@@ -85,7 +111,6 @@ type metrics struct {
 	StatusInternalServerError        imetrics.Counter
 	StatusOthers                     imetrics.Counter
 	RequestTime                      imetrics.Histogram
-	ResponseSize                     imetrics.Histogram
 }
 
 func newMetrics() *metrics {
@@ -94,8 +119,14 @@ func newMetrics() *metrics {
 	return &metrics{
 		RequestsInFlight:                 requestsInFlight(ctx),
 		RequestsTotal:                    requestsTotal(ctx),
+		RequestsSentBytes:                requestSentBytes(ctx),
 		RequestsFailure:                  requestsFailure(ctx),
 		RequestsBodyFailure:              requestsBodyFailure(ctx),
+		WantlistsTotal:                   wantlistsTotal(ctx),
+		WantlistsItemsTotal:              wantlistsItemsTotal(ctx),
+		WantlistsSeconds:                 wantlistsSeconds(ctx),
+		ResponseSizes:                    responseSizes(ctx),
+		ResponseTotalBytes:               responseTotalBytes(ctx),
 		StatusNotFound:                   statusNotFound(ctx),
 		StatusGone:                       statusGone(ctx),
 		StatusForbidden:                  statusForbidden(ctx),
@@ -106,7 +137,6 @@ func newMetrics() *metrics {
 		StatusInternalServerError:        statusInternalServerError(ctx),
 		StatusOthers:                     statusOthers(ctx),
 		RequestTime:                      requestTime(ctx),
-		ResponseSize:                     responseSize(ctx),
 	}
 }
 

bitswap/network/httpnet/msg_sender.go

@@ -1,6 +1,7 @@
 package httpnet
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
@@ -231,6 +232,12 @@ func (sender *httpMsgSender) tryURL(ctx context.Context, u *senderURL, entry bsm
 		return serr
 	}
 
+	// Record request size
+	var buf bytes.Buffer
+	req.Write(&buf)
+	sender.ht.metrics.RequestsSentBytes.Add(float64((&buf).Len()))
+
+	// Handle responses
 	limReader := &io.LimitedReader{
 		R: resp.Body,
 		N: sender.ht.maxBlockSize,
@@ -248,10 +255,18 @@ func (sender *httpMsgSender) tryURL(ctx context.Context, u *senderURL, entry bsm
 			Err:  err,
 		}
 	}
-	reqDuration := time.Since(reqStart)
-	sender.ht.metrics.ResponseSize.Observe(float64(len(body)))
+
+	// Calculate full response size with headers and everything.
+	// So this is comparable to bitswap message response sizes.
+	resp.Body = nil
+	var respBuf bytes.Buffer
+	resp.Write(&respBuf)
+	respLen := (&respBuf).Len() + len(body)
+
+	sender.ht.metrics.ResponseSizes.Observe(float64(respLen))
+	sender.ht.metrics.ResponseTotalBytes.Add(float64(respLen))
 	sender.ht.metrics.RequestsInFlight.Dec()
-	sender.ht.metrics.RequestTime.Observe(float64(reqDuration) / float64(time.Second))
+	sender.ht.metrics.RequestTime.Observe(float64(time.Since(reqStart)) / float64(time.Second))
 	sender.ht.metrics.updateStatusCounter(resp.StatusCode)
 
 	sender.ht.connEvtMgr.OnMessage(sender.peer)
@@ -353,6 +368,14 @@ func (sender *httpMsgSender) SendMsg(ctx context.Context, msg bsmsg.BitSwapMessa
 		return nil
 	}
 
+	// Keep metrics of wantlists sent and how long it took
+	sender.ht.metrics.WantlistsTotal.Inc()
+	sender.ht.metrics.WantlistsItemsTotal.Add(float64(len(wantlist)))
+	now := time.Now()
+	defer func() {
+		sender.ht.metrics.WantlistsSeconds.Add(float64(time.Since(now)) / float64(time.Second))
+	}()
+
 	go func() {
 		select {
 		case <-sender.closing: