Merge pull request #464 from MichaelMure/fix-metrics

metrics: record message/request event even in case of error
libp2p · Mar 3, 2020 · c2631d9 · c2631d9
2 parents 4d2a8a7 + faa5638
commit c2631d9
Showing 1 changed file with 36 additions and 24 deletions.
diff --git a/dht_net.go b/dht_net.go
@@ -82,8 +82,9 @@ func (dht *IpfsDHT) handleNewMessage(s network.Stream) bool {
 	for {
 		var req pb.Message
 		msgbytes, err := r.ReadMsg()
+		msgLen := len(msgbytes)
 		if err != nil {
-			defer r.ReleaseMsg(msgbytes)
+			r.ReleaseMsg(msgbytes)
 			if err == io.EOF {
 				return true
 			}
@@ -92,37 +93,39 @@ func (dht *IpfsDHT) handleNewMessage(s network.Stream) bool {
 			if err.Error() != "stream reset" {
 				logger.Debugf("error reading message: %#v", err)
 			}
-			stats.RecordWithTags(
-				ctx,
-				[]tag.Mutator{tag.Upsert(metrics.KeyMessageType, "UNKNOWN")},
-				metrics.ReceivedMessageErrors.M(1),
-			)
+			if msgLen > 0 {
+				stats.RecordWithTags(ctx,
+					[]tag.Mutator{tag.Upsert(metrics.KeyMessageType, "UNKNOWN")},
+					metrics.ReceivedMessages.M(1),
+					metrics.ReceivedMessageErrors.M(1),
+					metrics.ReceivedBytes.M(int64(msgLen)),
+				)
+			}
 			return false
 		}
 		err = req.Unmarshal(msgbytes)
 		r.ReleaseMsg(msgbytes)
 		if err != nil {
 			logger.Debugf("error unmarshalling message: %#v", err)
-			stats.RecordWithTags(
-				ctx,
+			stats.RecordWithTags(ctx,
 				[]tag.Mutator{tag.Upsert(metrics.KeyMessageType, "UNKNOWN")},
+				metrics.ReceivedMessages.M(1),
 				metrics.ReceivedMessageErrors.M(1),
+				metrics.ReceivedBytes.M(int64(msgLen)),
 			)
 			return false
 		}
 
 		timer.Reset(dhtStreamIdleTimeout)
 
 		startTime := time.Now()
-		ctx, _ := tag.New(
-			ctx,
+		ctx, _ := tag.New(ctx,
 			tag.Upsert(metrics.KeyMessageType, req.GetType().String()),
 		)
 
-		stats.Record(
-			ctx,
+		stats.Record(ctx,
 			metrics.ReceivedMessages.M(1),
-			metrics.ReceivedBytes.M(int64(req.Size())),
+			metrics.ReceivedBytes.M(int64(msgLen)),
 		)
 
 		handler := dht.handlerForMsgType(req.GetType())
@@ -166,28 +169,31 @@ func (dht *IpfsDHT) sendRequest(ctx context.Context, p peer.ID, pmes *pb.Message
 
 	ms, err := dht.messageSenderForPeer(ctx, p)
 	if err != nil {
-		stats.Record(ctx, metrics.SentRequestErrors.M(1))
+		stats.Record(ctx,
+			metrics.SentRequests.M(1),
+			metrics.SentRequestErrors.M(1),
+		)
 		return nil, err
 	}
 
 	start := time.Now()
 
 	rpmes, err := ms.SendRequest(ctx, pmes)
 	if err != nil {
-		stats.Record(ctx, metrics.SentRequestErrors.M(1))
+		stats.Record(ctx,
+			metrics.SentRequests.M(1),
+			metrics.SentRequestErrors.M(1),
+		)
 		return nil, err
 	}
 
 	// update the peer (on valid msgs only)
 	dht.updateFromMessage(ctx, p, rpmes)
 
-	stats.Record(
-		ctx,
+	stats.Record(ctx,
 		metrics.SentRequests.M(1),
 		metrics.SentBytes.M(int64(pmes.Size())),
-		metrics.OutboundRequestLatency.M(
-			float64(time.Since(start))/float64(time.Millisecond),
-		),
+		metrics.OutboundRequestLatency.M(float64(time.Since(start))/float64(time.Millisecond)),
 	)
 	dht.peerstore.RecordLatency(p, time.Since(start))
 	logger.Event(ctx, "dhtReceivedMessage", dht.self, p, rpmes)
@@ -200,20 +206,26 @@ func (dht *IpfsDHT) sendMessage(ctx context.Context, p peer.ID, pmes *pb.Message
 
 	ms, err := dht.messageSenderForPeer(ctx, p)
 	if err != nil {
-		stats.Record(ctx, metrics.SentMessageErrors.M(1))
+		stats.Record(ctx,
+			metrics.SentMessages.M(1),
+			metrics.SentMessageErrors.M(1),
+		)
 		return err
 	}
 
 	if err := ms.SendMessage(ctx, pmes); err != nil {
-		stats.Record(ctx, metrics.SentMessageErrors.M(1))
+		stats.Record(ctx,
+			metrics.SentMessages.M(1),
+			metrics.SentMessageErrors.M(1),
+		)
 		return err
 	}
 
-	stats.Record(
-		ctx,
+	stats.Record(ctx,
 		metrics.SentMessages.M(1),
 		metrics.SentBytes.M(int64(pmes.Size())),
 	)
+
 	logger.Event(ctx, "dhtSentMessage", dht.self, p, pmes)
 	return nil
 }