mirror/prysm-pulse
1
0
Fork 0
mirror of https://gitlab.com/pulsechaincom/prysm-pulse.git synced 2024-12-24 20:37:17 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

add peer to span and log (#3838)

Browse Source
This commit is contained in:
Preston Van Loon 2019-10-23 18:35:08 -07:00 committed by terence tsao
parent 9a0fb5dca1
commit 544ce2b4ed
1 changed files with 7 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
beacon-chain/sync/rpc.go
View File

@ -9,6 +9,7 @@ import (
"github.com/libp2p/go-libp2p-core/network" "github.com/libp2p/go-libp2p-core/network"
pb "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1" pb "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
"github.com/prysmaticlabs/prysm/shared/roughtime" "github.com/prysmaticlabs/prysm/shared/roughtime"
"github.com/prysmaticlabs/prysm/shared/traceutil"
"go.opencensus.io/trace" "go.opencensus.io/trace"
) )
@ -62,6 +63,8 @@ func (r *RegularSync) registerRPC(topic string, base interface{}, handle rpcHand
ctx, span := trace.StartSpan(ctx, "sync.rpc") ctx, span := trace.StartSpan(ctx, "sync.rpc")
defer span.End() defer span.End()
span.AddAttributes(trace.StringAttribute("topic", topic)) span.AddAttributes(trace.StringAttribute("topic", topic))
span.AddAttributes(trace.StringAttribute("peer", stream.Conn().RemotePeer().Pretty()))
log := log.WithField("peer", stream.Conn().RemotePeer().Pretty())
if err := stream.SetReadDeadline(roughtime.Now().Add(ttfbTimeout)); err != nil { if err := stream.SetReadDeadline(roughtime.Now().Add(ttfbTimeout)); err != nil {
log.WithError(err).Error("Could not set stream read deadline") log.WithError(err).Error("Could not set stream read deadline")
@ -76,21 +79,25 @@ func (r *RegularSync) registerRPC(topic string, base interface{}, handle rpcHand
msg := reflect.New(t.Elem()) msg := reflect.New(t.Elem())
if err := r.p2p.Encoding().DecodeWithLength(stream, msg.Interface()); err != nil { if err := r.p2p.Encoding().DecodeWithLength(stream, msg.Interface()); err != nil {
log.WithError(err).Error("Failed to decode stream message") log.WithError(err).Error("Failed to decode stream message")
traceutil.AnnotateError(span, err)
return return
} }
if err := handle(ctx, msg.Interface(), stream); err != nil { if err := handle(ctx, msg.Interface(), stream); err != nil {
messageFailedProcessingCounter.WithLabelValues(topic).Inc() messageFailedProcessingCounter.WithLabelValues(topic).Inc()
log.WithError(err).Error("Failed to handle p2p RPC") log.WithError(err).Error("Failed to handle p2p RPC")
traceutil.AnnotateError(span, err)
} }
} else { } else {
msg := reflect.New(t) msg := reflect.New(t)
if err := r.p2p.Encoding().DecodeWithLength(stream, msg.Interface()); err != nil { if err := r.p2p.Encoding().DecodeWithLength(stream, msg.Interface()); err != nil {
log.WithError(err).Error("Failed to decode stream message") log.WithError(err).Error("Failed to decode stream message")
traceutil.AnnotateError(span, err)
return return
} }
if err := handle(ctx, msg.Elem().Interface(), stream); err != nil { if err := handle(ctx, msg.Elem().Interface(), stream); err != nil {
messageFailedProcessingCounter.WithLabelValues(topic).Inc() messageFailedProcessingCounter.WithLabelValues(topic).Inc()
log.WithError(err).Error("Failed to handle p2p RPC") log.WithError(err).Error("Failed to handle p2p RPC")
traceutil.AnnotateError(span, err)
} }
} }