Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

metrics: record message/request event even in case of error #464

Merged
merged 3 commits into from
Mar 3, 2020
Merged

metrics: record message/request event even in case of error #464

Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
fe2d21a
immediate buffer release instead of defering
MichaelMure Feb 27, 2020
979ed6a
metrics: also record message/request event on error
MichaelMure Feb 27, 2020
faa5638
net: address PR comments regarding metrics
MichaelMure Mar 2, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 17 additions & 8 deletions dht_net.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,7 +83,7 @@ func (dht *IpfsDHT) handleNewMessage(s network.Stream) bool {
var req pb.Message
msgbytes, err := r.ReadMsg()
if err != nil {
defer r.ReleaseMsg(msgbytes)
r.ReleaseMsg(msgbytes)
if err == io.EOF {
return true
}
Expand All @@ -96,6 +96,8 @@ func (dht *IpfsDHT) handleNewMessage(s network.Stream) bool {
ctx,
[]tag.Mutator{tag.Upsert(metrics.KeyMessageType, "UNKNOWN")},
metrics.ReceivedMessageErrors.M(1),
metrics.ReceivedMessages.M(1),
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hm, we probably shouldn't record either a received message or a message error unless we receive a non-empty message. If the stream's broken for some reason, that's not something the DHT is really concerned about.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Would it be fair to not record anything at all ? My understanding is that if the stream is broken we don't get io.EOF but we don't have a message either.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The more I read your message, the more I get confused ;-). It seems to me that there is three thing we can do:

  • valid message: record metrics.ReceivedMessages.M(1)
  • broken message: record both metrics.ReceivedMessages.M(1) and metrics.ReceivedMessageErrors.M(1)
  • no message: don't record anything

So what about:

  • io.EOF --> no message
  • len(msgbytes)==0 --> no message
  • len(msgbytes)!=0 --> broken message

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I did that, let me know if that doesn't make sense.

Copy link
Contributor Author

@MichaelMure MichaelMure Mar 2, 2020 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We might also return early on err.Error() == "stream reset" but that's quite brittle.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

👍

metrics.ReceivedBytes.M(int64(req.Size())),
MichaelMure marked this conversation as resolved.
Show resolved Hide resolved
)
return false
}
Expand All @@ -107,6 +109,8 @@ func (dht *IpfsDHT) handleNewMessage(s network.Stream) bool {
ctx,
[]tag.Mutator{tag.Upsert(metrics.KeyMessageType, "UNKNOWN")},
metrics.ReceivedMessageErrors.M(1),
metrics.ReceivedMessages.M(1),
metrics.ReceivedBytes.M(int64(req.Size())),
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should be len(msgbytes) but we need to record that before we release the message.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

it could always be len(msgbytes), right ? Does it ever make sense to use req.Size() instead?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I did that, let me know if that doesn't make sense.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Does it ever make sense to use

For requests, no.

)
return false
}
Expand Down Expand Up @@ -164,6 +168,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())),
MichaelMure marked this conversation as resolved.
Show resolved Hide resolved
)

ms, err := dht.messageSenderForPeer(ctx, p)
if err != nil {
stats.Record(ctx, metrics.SentRequestErrors.M(1))
Expand All @@ -183,8 +193,6 @@ func (dht *IpfsDHT) sendRequest(ctx context.Context, p peer.ID, pmes *pb.Message

stats.Record(
ctx,
metrics.SentRequests.M(1),
metrics.SentBytes.M(int64(pmes.Size())),
MichaelMure marked this conversation as resolved.
Show resolved Hide resolved
metrics.OutboundRequestLatency.M(
float64(time.Since(start))/float64(time.Millisecond),
),
Expand All @@ -198,6 +206,12 @@ 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(
MichaelMure marked this conversation as resolved.
Show resolved Hide resolved
ctx,
metrics.SentMessages.M(1),
metrics.SentBytes.M(int64(pmes.Size())),
MichaelMure marked this conversation as resolved.
Show resolved Hide resolved
)

ms, err := dht.messageSenderForPeer(ctx, p)
if err != nil {
stats.Record(ctx, metrics.SentMessageErrors.M(1))
Expand All @@ -209,11 +223,6 @@ func (dht *IpfsDHT) sendMessage(ctx context.Context, p peer.ID, pmes *pb.Message
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
}
Expand Down