fluencelabs/go-libp2p-kad-dht
1
0
Fork 0
mirror of https://github.com/fluencelabs/go-libp2p-kad-dht synced 2025-04-24 14:22:13 +00:00
Code Issues Projects Releases Wiki Activity

net: address PR comments regarding metrics

Browse Source
This commit is contained in:
Michael Muré 2020-03-02 13:11:56 +01:00
parent 979ed6ab6c
commit faa56381cb
No known key found for this signature in database
GPG Key ID: A4457C029293126F
1 changed files with 40 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

77
dht_net.go
View File

@ -82,6 +82,7 @@ func (dht *IpfsDHT) handleNewMessage(s network.Stream) bool {
for {
var req pb.Message
msgbytes, err := r.ReadMsg()
msgLen := len(msgbytes)
if err != nil {
r.ReleaseMsg(msgbytes)
if err == io.EOF {
@ -92,25 +93,25 @@ 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),
metrics.ReceivedMessages.M(1),
metrics.ReceivedBytes.M(int64(req.Size())),
)
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.ReceivedMessageErrors.M(1),
metrics.ReceivedMessages.M(1),
metrics.ReceivedBytes.M(int64(req.Size())),
metrics.ReceivedMessageErrors.M(1),
metrics.ReceivedBytes.M(int64(msgLen)),
)
return false
}
@ -118,15 +119,13 @@ func (dht *IpfsDHT) handleNewMessage(s network.Stream) bool {
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())
@ -168,15 +167,12 @@ func (dht *IpfsDHT) handleNewMessage(s network.Stream) bool {
func (dht *IpfsDHT) sendRequest(ctx context.Context, p peer.ID, pmes *pb.Message) (*pb.Message, error) {
ctx, _ = tag.New(ctx, metrics.UpsertMessageType(pmes))
stats.Record(
ctx,
metrics.SentRequests.M(1),
metrics.SentBytes.M(int64(pmes.Size())),
)
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
}
@ -184,18 +180,20 @@ func (dht *IpfsDHT) sendRequest(ctx context.Context, p peer.ID, pmes *pb.Message
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,
metrics.OutboundRequestLatency.M(
float64(time.Since(start))/float64(time.Millisecond),
),
stats.Record(ctx,
metrics.SentRequests.M(1),
metrics.SentBytes.M(int64(pmes.Size())),
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)
@ -206,23 +204,28 @@ func (dht *IpfsDHT) sendRequest(ctx context.Context, p peer.ID, pmes *pb.Message
func (dht *IpfsDHT) sendMessage(ctx context.Context, p peer.ID, pmes *pb.Message) error {
ctx, _ = tag.New(ctx, metrics.UpsertMessageType(pmes))
stats.Record(
ctx,
metrics.SentMessages.M(1),
metrics.SentBytes.M(int64(pmes.Size())),
)
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,
metrics.SentMessages.M(1),
metrics.SentBytes.M(int64(pmes.Size())),
)
logger.Event(ctx, "dhtSentMessage", dht.self, p, pmes)
return nil
}