From 11b9f12411c544f7af266df035b1c6e8c2084971 Mon Sep 17 00:00:00 2001 From: alok-ssv Date: Wed, 27 May 2026 17:13:58 +0530 Subject: [PATCH] chore: add more logs and metrics --- message/validation/consensus_validation.go | 18 ++-- message/validation/errors.go | 40 +++++++-- message/validation/observer.go | 19 ++++ message/validation/observer_event_test.go | 54 ++++++++++++ message/validation/partial_validation.go | 14 +-- message/validation/stage.go | 34 ++++++++ message/validation/validation.go | 28 +++--- network/peers/connections/conn_handler.go | 63 +++++++++++--- network/peers/connections/observability.go | 45 ++++++++++ .../peers/connections/observability_test.go | 87 +++++++++++++++++++ network/peers/peertrace/observer.go | 45 ++++++++++ network/peers/peertrace/observer_test.go | 84 +++++++++++++++++- network/streams/controller.go | 19 ++++ network/streams/observability.go | 33 +++++++ network/streams/observability_test.go | 54 ++++++++++++ network/topics/controller.go | 10 ++- network/topics/observability.go | 39 ++++++++- network/topics/observability_test.go | 64 ++++++++++++-- network/topics/tracer.go | 28 +++++- 19 files changed, 716 insertions(+), 62 deletions(-) create mode 100644 message/validation/observer_event_test.go create mode 100644 message/validation/stage.go create mode 100644 network/peers/connections/observability_test.go create mode 100644 network/streams/observability_test.go diff --git a/message/validation/consensus_validation.go b/message/validation/consensus_validation.go index 61ed77ecdf..9d0e756a57 100644 --- a/message/validation/consensus_validation.go +++ b/message/validation/consensus_validation.go @@ -34,32 +34,32 @@ func (mv *messageValidator) validateConsensusMessage( e := ErrSSVDataTooBig e.got = len(ssvMessage.Data) e.want = maxEncodedConsensusMsgSize - return nil, e + return nil, withValidationStage(SSVValidationStageConsensus, e) } consensusMessage, err := specqbft.DecodeMessage(ssvMessage.Data) if err != nil { e := ErrUndecodableMessageData e.innerErr = err - return nil, e + return nil, withValidationStage(SSVValidationStageConsensus, e) } if err := mv.validateConsensusMessageSemantics(signedSSVMessage, consensusMessage, committeeInfo.committee); err != nil { - return consensusMessage, err + return consensusMessage, withValidationStage(SSVValidationStageConsensus, err) } state := mv.validatorState(ssvMessage.GetID(), committeeInfo) if err := mv.validateQBFTLogic(signedSSVMessage, consensusMessage, committeeInfo, receivedFrom, receivedAt, state); err != nil { - return consensusMessage, err + return consensusMessage, withValidationStage(SSVValidationStageConsensus, err) } if err := mv.validateQBFTMessageByDutyLogic(signedSSVMessage, consensusMessage, committeeInfo, receivedAt, state); err != nil { - return consensusMessage, err + return consensusMessage, withValidationStage(SSVValidationStageConsensus, err) } if err := ctx.Err(); err != nil { - return consensusMessage, err + return consensusMessage, withValidationStage(SSVValidationStageContext, err) } for i := range signedSSVMessage.Signatures { @@ -67,18 +67,18 @@ func (mv *messageValidator) validateConsensusMessage( signature := signedSSVMessage.Signatures[i] if err := ctx.Err(); err != nil { - return consensusMessage, err + return consensusMessage, withValidationStage(SSVValidationStageContext, err) } if err := mv.signatureVerifier.VerifySignature(operatorID, ssvMessage, signature); err != nil { e := ErrSignatureVerification e.innerErr = fmt.Errorf("verify opid: %v signature: %w", operatorID, err) - return consensusMessage, e + return consensusMessage, withValidationStage(SSVValidationStageSignatureVerify, e) } } if err := mv.updateConsensusState(signedSSVMessage, consensusMessage, committeeInfo, receivedFrom, state); err != nil { - return consensusMessage, err + return consensusMessage, withValidationStage(SSVValidationStageStateUpdate, err) } return consensusMessage, nil diff --git a/message/validation/errors.go b/message/validation/errors.go index c389fa2a4a..18329ea4d5 100644 --- a/message/validation/errors.go +++ b/message/validation/errors.go @@ -143,8 +143,11 @@ var ( ErrZeroRound = Error{text: "zero round", reject: true} ) -func (mv *messageValidator) handleValidationError(ctx context.Context, peerID peer.ID, decodedMessage *queue.SSVMessage, err error) pubsub.ValidationResult { +func (mv *messageValidator) handleValidationError(ctx context.Context, peerID peer.ID, decodedMessage *queue.SSVMessage, pmsg *pubsub.Message, err error) pubsub.ValidationResult { loggerFields := mv.buildLoggerFields(decodedMessage) + stage := validationStageFromError(err) + topic := pubsubMessageTopic(pmsg) + payloadSize := pubsubMessagePayloadSize(pmsg) logger := mv.logger. With(loggerFields.AsZapFields()...). @@ -153,12 +156,12 @@ func (mv *messageValidator) handleValidationError(ctx context.Context, peerID pe switch { case errors.Is(err, context.DeadlineExceeded): recordIgnoredMessage(ctx, loggerFields.Role, validationTimeoutReason) - mv.observeSSVValidation(ctx, peerID, loggerFields, SSVValidationIgnored, validationTimeoutReason, err) + mv.observeSSVValidation(ctx, peerID, loggerFields, SSVValidationIgnored, validationTimeoutReason, stage, topic, payloadSize, err) logger.Debug("ignoring message due to validation timeout", zap.Error(err)) return pubsub.ValidationIgnore case errors.Is(err, context.Canceled): recordIgnoredMessage(ctx, loggerFields.Role, validationCanceledReason) - mv.observeSSVValidation(ctx, peerID, loggerFields, SSVValidationIgnored, validationCanceledReason, err) + mv.observeSSVValidation(ctx, peerID, loggerFields, SSVValidationIgnored, validationCanceledReason, stage, topic, payloadSize, err) logger.Debug("ignoring message due to validation cancellation", zap.Error(err)) return pubsub.ValidationIgnore } @@ -166,7 +169,7 @@ func (mv *messageValidator) handleValidationError(ctx context.Context, peerID pe var valErr Error if !errors.As(err, &valErr) { recordIgnoredMessage(ctx, loggerFields.Role, err.Error()) - mv.observeSSVValidation(ctx, peerID, loggerFields, SSVValidationIgnored, err.Error(), err) + mv.observeSSVValidation(ctx, peerID, loggerFields, SSVValidationIgnored, err.Error(), stage, topic, payloadSize, err) logger.Debug("ignoring invalid message", zap.Error(err)) return pubsub.ValidationIgnore } @@ -176,7 +179,7 @@ func (mv *messageValidator) handleValidationError(ctx context.Context, peerID pe logger.Debug("ignoring invalid message", zap.Error(valErr)) } recordIgnoredMessage(ctx, loggerFields.Role, valErr.Text()) - mv.observeSSVValidation(ctx, peerID, loggerFields, SSVValidationIgnored, valErr.Text(), valErr) + mv.observeSSVValidation(ctx, peerID, loggerFields, SSVValidationIgnored, valErr.Text(), stage, topic, payloadSize, valErr) return pubsub.ValidationIgnore } @@ -185,16 +188,17 @@ func (mv *messageValidator) handleValidationError(ctx context.Context, peerID pe } recordRejectedMessage(ctx, loggerFields.Role, valErr.Text()) - mv.observeSSVValidation(ctx, peerID, loggerFields, SSVValidationRejected, valErr.Text(), valErr) + mv.observeSSVValidation(ctx, peerID, loggerFields, SSVValidationRejected, valErr.Text(), stage, topic, payloadSize, valErr) return pubsub.ValidationReject } -func (mv *messageValidator) handleValidationSuccess(ctx context.Context, peerID peer.ID, decodedMessage *queue.SSVMessage) pubsub.ValidationResult { +func (mv *messageValidator) handleValidationSuccess(ctx context.Context, peerID peer.ID, decodedMessage *queue.SSVMessage, pmsg *pubsub.Message) pubsub.ValidationResult { recordAcceptedMessage(ctx, messageRole(decodedMessage)) if mv.observer == nil { return pubsub.ValidationAccept } - mv.observeSSVValidation(ctx, peerID, mv.buildLoggerFields(decodedMessage), SSVValidationAccepted, "valid", nil) + mv.observeSSVValidation(ctx, peerID, mv.buildLoggerFields(decodedMessage), SSVValidationAccepted, "valid", + SSVValidationStageComplete, pubsubMessageTopic(pmsg), pubsubMessagePayloadSize(pmsg), nil) return pubsub.ValidationAccept } @@ -204,6 +208,9 @@ func (mv *messageValidator) observeSSVValidation( loggerFields *LoggerFields, outcome string, reason string, + stage string, + topic string, + payloadSize int, err error, ) { if mv.observer == nil { @@ -217,6 +224,9 @@ func (mv *messageValidator) observeSSVValidation( PeerID: peerID, Outcome: outcome, Reason: reason, + Stage: stage, + Topic: topic, + PayloadSize: payloadSize, Role: loggerFields.Role, SSVMessageType: loggerFields.SSVMessageType, Slot: loggerFields.Slot, @@ -232,3 +242,17 @@ func (mv *messageValidator) observeSSVValidation( } mv.observer.ObserveSSVValidation(ctx, mv.logger, event) } + +func pubsubMessageTopic(pmsg *pubsub.Message) string { + if pmsg == nil { + return "" + } + return pmsg.GetTopic() +} + +func pubsubMessagePayloadSize(pmsg *pubsub.Message) int { + if pmsg == nil { + return 0 + } + return len(pmsg.GetData()) +} diff --git a/message/validation/observer.go b/message/validation/observer.go index 4c007420e8..919445f68f 100644 --- a/message/validation/observer.go +++ b/message/validation/observer.go @@ -16,12 +16,31 @@ const ( SSVValidationRejected = "rejected" ) +const ( + SSVValidationStageUnknown = "unknown" + SSVValidationStageContext = "context" + SSVValidationStagePubsubBasic = "pubsub_basic" + SSVValidationStageDecodeSigned = "decode_signed" + SSVValidationStageSignedSemantics = "signed_semantics" + SSVValidationStageSSVSemantics = "ssv_semantics" + SSVValidationStageCommitteeLookup = "committee_lookup" + SSVValidationStageCommitteeChecks = "committee_checks" + SSVValidationStageConsensus = "consensus_validation" + SSVValidationStagePartial = "partial_validation" + SSVValidationStageSignatureVerify = "signature_verification" + SSVValidationStageStateUpdate = "state_update" + SSVValidationStageComplete = "complete" +) + // SSVValidationEvent describes the SSV-level validation decision before it // is reduced to libp2p's accept/reject/ignore validation result. type SSVValidationEvent struct { PeerID peer.ID Outcome string Reason string + Stage string + Topic string + PayloadSize int Role spectypes.RunnerRole SSVMessageType spectypes.MsgType Slot phase0.Slot diff --git a/message/validation/observer_event_test.go b/message/validation/observer_event_test.go new file mode 100644 index 0000000000..d9b98ce86b --- /dev/null +++ b/message/validation/observer_event_test.go @@ -0,0 +1,54 @@ +package validation + +import ( + "context" + "testing" + + pubsub "github.com/libp2p/go-libp2p-pubsub" + pspb "github.com/libp2p/go-libp2p-pubsub/pb" + "github.com/libp2p/go-libp2p/core/peer" + "github.com/stretchr/testify/require" + "go.uber.org/zap" +) + +type captureSSVValidationObserver struct { + event SSVValidationEvent +} + +func (o *captureSSVValidationObserver) ObserveSSVValidation(_ context.Context, _ *zap.Logger, event SSVValidationEvent) { + o.event = event +} + +func TestHandleValidationErrorObservesStageTopicAndPayload(t *testing.T) { + observer := &captureSSVValidationObserver{} + mv := &messageValidator{ + logger: zap.NewNop(), + observer: observer, + } + + topic := "ssv.v2.42" + pmsg := &pubsub.Message{ + Message: &pspb.Message{ + Topic: &topic, + Data: []byte{1, 2, 3, 4}, + }, + } + pid := peer.ID("peer-a") + + result := mv.handleValidationError(t.Context(), pid, nil, pmsg, withValidationStage(SSVValidationStageDecodeSigned, ErrMalformedPubSubMessage)) + + require.Equal(t, pubsub.ValidationReject, result) + require.Equal(t, pid, observer.event.PeerID) + require.Equal(t, SSVValidationRejected, observer.event.Outcome) + require.Equal(t, ErrMalformedPubSubMessage.Text(), observer.event.Reason) + require.Equal(t, SSVValidationStageDecodeSigned, observer.event.Stage) + require.Equal(t, topic, observer.event.Topic) + require.Equal(t, 4, observer.event.PayloadSize) +} + +func TestWithValidationStagePreservesErrorMatching(t *testing.T) { + err := withValidationStage(SSVValidationStagePubsubBasic, ErrPubSubMessageHasNoData) + + require.ErrorIs(t, err, ErrPubSubMessageHasNoData) + require.Equal(t, SSVValidationStagePubsubBasic, validationStageFromError(err)) +} diff --git a/message/validation/partial_validation.go b/message/validation/partial_validation.go index 9bf209c532..a41c0ecd94 100644 --- a/message/validation/partial_validation.go +++ b/message/validation/partial_validation.go @@ -31,27 +31,27 @@ func (mv *messageValidator) validatePartialSignatureMessage( e := ErrSSVDataTooBig e.got = len(ssvMessage.Data) e.want = maxEncodedPartialSignatureSize - return nil, e + return nil, withValidationStage(SSVValidationStagePartial, e) } partialSignatureMessages := &spectypes.PartialSignatureMessages{} if err := partialSignatureMessages.Decode(ssvMessage.Data); err != nil { e := ErrUndecodableMessageData e.innerErr = err - return nil, e + return nil, withValidationStage(SSVValidationStagePartial, e) } if err := mv.validatePartialSignatureMessageSemantics(signedSSVMessage, partialSignatureMessages, committeeInfo.validatorIndices); err != nil { - return partialSignatureMessages, err + return partialSignatureMessages, withValidationStage(SSVValidationStagePartial, err) } state := mv.validatorState(ssvMessage.GetID(), committeeInfo) if err := mv.validatePartialSigMessagesByDutyLogic(signedSSVMessage, partialSignatureMessages, committeeInfo, receivedFrom, receivedAt, state); err != nil { - return partialSignatureMessages, err + return partialSignatureMessages, withValidationStage(SSVValidationStagePartial, err) } if err := ctx.Err(); err != nil { - return partialSignatureMessages, err + return partialSignatureMessages, withValidationStage(SSVValidationStageContext, err) } signature := signedSSVMessage.Signatures[0] @@ -59,11 +59,11 @@ func (mv *messageValidator) validatePartialSignatureMessage( if err := mv.signatureVerifier.VerifySignature(signer, ssvMessage, signature); err != nil { e := ErrSignatureVerification e.innerErr = fmt.Errorf("verify opid: %v signature: %w", signer, err) - return partialSignatureMessages, e + return partialSignatureMessages, withValidationStage(SSVValidationStageSignatureVerify, e) } if err := mv.updatePartialSignatureState(partialSignatureMessages, receivedFrom, state, signer, committeeInfo); err != nil { - return partialSignatureMessages, err + return partialSignatureMessages, withValidationStage(SSVValidationStageStateUpdate, err) } return partialSignatureMessages, nil diff --git a/message/validation/stage.go b/message/validation/stage.go new file mode 100644 index 0000000000..c7c7f4634a --- /dev/null +++ b/message/validation/stage.go @@ -0,0 +1,34 @@ +package validation + +import "errors" + +type validationStageError struct { + stage string + err error +} + +func (e validationStageError) Error() string { + return e.err.Error() +} + +func (e validationStageError) Unwrap() error { + return e.err +} + +func withValidationStage(stage string, err error) error { + if err == nil { + return nil + } + if validationStageFromError(err) != SSVValidationStageUnknown { + return err + } + return validationStageError{stage: stage, err: err} +} + +func validationStageFromError(err error) string { + var staged validationStageError + if errors.As(err, &staged) { + return staged.stage + } + return SSVValidationStageUnknown +} diff --git a/message/validation/validation.go b/message/validation/validation.go index 010ca25de8..255946188d 100644 --- a/message/validation/validation.go +++ b/message/validation/validation.go @@ -124,7 +124,7 @@ func (mv *messageValidator) Validate(ctx context.Context, peerID peer.ID, pmsg * } if err := ctx.Err(); err != nil { - return mv.handleValidationError(ctx, peerID, nil, err) + return mv.handleValidationError(ctx, peerID, nil, pmsg, withValidationStage(SSVValidationStageContext, err)) } validationStart := time.Now() @@ -136,12 +136,12 @@ func (mv *messageValidator) Validate(ctx context.Context, peerID peer.ID, pmsg * }() if err != nil { - return mv.handleValidationError(ctx, peerID, decodedMessage, err) + return mv.handleValidationError(ctx, peerID, decodedMessage, pmsg, err) } pmsg.ValidatorData = decodedMessage - return mv.handleValidationSuccess(ctx, peerID, decodedMessage) + return mv.handleValidationSuccess(ctx, peerID, decodedMessage, pmsg) } func messageRole(decodedMessage *queue.SSVMessage) spectypes.RunnerRole { @@ -153,16 +153,16 @@ func messageRole(decodedMessage *queue.SSVMessage) spectypes.RunnerRole { func (mv *messageValidator) handlePubsubMessage(ctx context.Context, pMsg *pubsub.Message, receivedAt time.Time) (*queue.SSVMessage, error) { if err := ctx.Err(); err != nil { - return nil, err + return nil, withValidationStage(SSVValidationStageContext, err) } if err := mv.validatePubSubMessage(pMsg); err != nil { - return nil, err + return nil, withValidationStage(SSVValidationStagePubsubBasic, err) } signedSSVMessage, err := mv.decodeSignedSSVMessage(pMsg) if err != nil { - return nil, err + return nil, withValidationStage(SSVValidationStageDecodeSigned, err) } return mv.handleSignedSSVMessage(ctx, signedSSVMessage, pMsg.GetTopic(), pMsg.ReceivedFrom, receivedAt) @@ -180,31 +180,31 @@ func (mv *messageValidator) handleSignedSSVMessage( } if err := ctx.Err(); err != nil { - return decodedMessage, err + return decodedMessage, withValidationStage(SSVValidationStageContext, err) } if err := mv.validateSignedSSVMessage(signedSSVMessage); err != nil { - return decodedMessage, err + return decodedMessage, withValidationStage(SSVValidationStageSignedSemantics, err) } decodedMessage.SSVMessage = signedSSVMessage.SSVMessage if err := mv.validateSSVMessage(signedSSVMessage.SSVMessage); err != nil { - return decodedMessage, err + return decodedMessage, withValidationStage(SSVValidationStageSSVSemantics, err) } committeeInfo, err := mv.getCommitteeAndValidatorIndices(signedSSVMessage.SSVMessage.GetID()) if err != nil { - return decodedMessage, err + return decodedMessage, withValidationStage(SSVValidationStageCommitteeLookup, err) } if err := mv.committeeChecks(signedSSVMessage, committeeInfo, topic); err != nil { - return decodedMessage, err + return decodedMessage, withValidationStage(SSVValidationStageCommitteeChecks, err) } // Bail out before we potentially wait on the per-message validation mutex. if err := ctx.Err(); err != nil { - return decodedMessage, err + return decodedMessage, withValidationStage(SSVValidationStageContext, err) } validationMu := mv.getValidationLock(signedSSVMessage.SSVMessage.GetID()) @@ -216,14 +216,14 @@ func (mv *messageValidator) handleSignedSSVMessage( consensusMessage, err := mv.validateConsensusMessage(ctx, signedSSVMessage, committeeInfo, receivedFrom, receivedAt) decodedMessage.Body = consensusMessage if err != nil { - return decodedMessage, err + return decodedMessage, withValidationStage(SSVValidationStageConsensus, err) } case spectypes.SSVPartialSignatureMsgType: partialSignatureMessages, err := mv.validatePartialSignatureMessage(ctx, signedSSVMessage, committeeInfo, receivedFrom, receivedAt) decodedMessage.Body = partialSignatureMessages if err != nil { - return decodedMessage, err + return decodedMessage, withValidationStage(SSVValidationStagePartial, err) } default: diff --git a/network/peers/connections/conn_handler.go b/network/peers/connections/conn_handler.go index 0b8cb3c877..640da2c9ba 100644 --- a/network/peers/connections/conn_handler.go +++ b/network/peers/connections/conn_handler.go @@ -39,6 +39,31 @@ type connHandler struct { peerObserver *peertrace.Observer } +type acceptConnectionError struct { + reason string + err error +} + +func (e acceptConnectionError) Error() string { + return e.err.Error() +} + +func (e acceptConnectionError) Unwrap() error { + return e.err +} + +func newAcceptConnectionError(reason string, err error) error { + return acceptConnectionError{reason: reason, err: err} +} + +func connectionReason(err error) string { + var acceptErr acceptConnectionError + if errors.As(err, &acceptErr) { + return acceptErr.reason + } + return connectionHandshakeReasonHandshakeError +} + // NewConnHandler creates a new connection handler func NewConnHandler( ctx context.Context, @@ -66,6 +91,18 @@ func NewConnHandler( // Handle configures a network notifications handler that handshakes and tracks all p2p connections func (ch *connHandler) Handle() *libp2pnetwork.NotifyBundle { + var ignoredConnection = errors.New("ignored connection") + connectionOutcome := func(err error) string { + switch { + case err == nil: + return connectionHandshakeOutcomeSuccess + case errors.Is(err, ignoredConnection): + return connectionHandshakeOutcomeIgnored + default: + return connectionHandshakeOutcomeFailure + } + } + disconnect := func(logger *zap.Logger, net libp2pnetwork.Network, conn libp2pnetwork.Conn) { id := conn.RemotePeer() errClose := net.ClosePeer(id) @@ -94,11 +131,10 @@ func (ch *connHandler) Handle() *libp2pnetwork.NotifyBundle { delete(ongoingHandshakes, pid) } - var ignoredConnection = errors.New("ignored connection") acceptConnection := func(logger *zap.Logger, net libp2pnetwork.Network, conn libp2pnetwork.Conn) (err error) { defer func() { if r := recover(); r != nil { - err = fmt.Errorf("panic: %v", r) + err = newAcceptConnectionError(connectionHandshakeReasonPanic, fmt.Errorf("panic: %v", r)) } }() @@ -107,7 +143,7 @@ func (ch *connHandler) Handle() *libp2pnetwork.NotifyBundle { if !beginHandshake(pid) { // Another connection with the same peer is already being handled. logger.Debug("peer is already being handled") - return ignoredConnection + return newAcceptConnectionError(connectionHandshakeReasonAlreadyHandling, ignoredConnection) } defer func() { // Unset this peer as being handled. @@ -117,7 +153,7 @@ func (ch *connHandler) Handle() *libp2pnetwork.NotifyBundle { switch ch.peerInfos.State(pid) { case peers.StateConnected, peers.StateConnecting: logger.Debug("peer is already connected or connecting") - return ignoredConnection + return newAcceptConnectionError(connectionHandshakeReasonAlreadyConnected, ignoredConnection) } ch.peerInfos.AddPeerInfo(pid, conn.RemoteMultiaddr(), conn.Stat().Direction) @@ -133,13 +169,13 @@ func (ch *connHandler) Handle() *libp2pnetwork.NotifyBundle { for { select { case <-deadline.C: - return errors.New("peer hasn't sent a handshake request") + return newAcceptConnectionError(connectionHandshakeReasonTimeoutNoHandshake, errors.New("peer hasn't sent a handshake request")) case <-ticker.C: // Check if peer has sent a handshake request. if pi := ch.peerInfos.PeerInfo(pid); pi != nil && pi.LastHandshake.After(start) { if pi.LastHandshakeError != nil { // Handshake failed. - return fmt.Errorf("peer handshake request failed: %w", pi.LastHandshakeError) + return newAcceptConnectionError(connectionHandshakeReasonHandshakeError, fmt.Errorf("peer handshake request failed: %w", pi.LastHandshakeError)) } // Handshake succeeded. @@ -147,13 +183,13 @@ func (ch *connHandler) Handle() *libp2pnetwork.NotifyBundle { } if net.Connectedness(pid) != libp2pnetwork.Connected { - return errors.New("lost connection") + return newAcceptConnectionError(connectionHandshakeReasonLostConnection, errors.New("lost connection")) } } } if !ch.sharesEnoughSubnets(conn) { - return errors.New("peer doesn't share enough subnets") + return newAcceptConnectionError(connectionHandshakeReasonSubnetsMismatch, errors.New("peer doesn't share enough subnets")) } return nil @@ -164,7 +200,7 @@ func (ch *connHandler) Handle() *libp2pnetwork.NotifyBundle { ch.peerInfos.SetState(pid, peers.StateConnecting) if err := ch.handshaker.Handshake(logger, conn); err != nil { - return fmt.Errorf("could not handshake: %w", err) + return newAcceptConnectionError(connectionHandshakeReasonHandshakeError, fmt.Errorf("could not handshake: %w", err)) } logger.Debug("handshake completed successfully") @@ -189,12 +225,19 @@ func (ch *connHandler) Handle() *libp2pnetwork.NotifyBundle { // Handle the connection (could be either incoming or outgoing) without blocking. go func() { logger := connLogger(conn) + start := time.Now() err := acceptConnection(logger, net, conn) if err == nil { if ch.connIdx.AtLimit(conn.Stat().Direction) { - err = errors.New("reached total connected peers limit") + err = newAcceptConnectionError(connectionHandshakeReasonMaxPeersLimit, errors.New("reached total connected peers limit")) } } + outcome := connectionOutcome(err) + reason := connectionHandshakeReasonSuccess + if err != nil { + reason = connectionReason(err) + } + recordConnectionHandshake(ch.ctx, conn.Stat().Direction, outcome, reason, time.Since(start)) if errors.Is(err, ignoredConnection) { return } diff --git a/network/peers/connections/observability.go b/network/peers/connections/observability.go index 39b92a32d3..3b27f30fa4 100644 --- a/network/peers/connections/observability.go +++ b/network/peers/connections/observability.go @@ -2,6 +2,7 @@ package connections import ( "context" + "time" "github.com/libp2p/go-libp2p/core/network" "go.opentelemetry.io/otel" @@ -15,6 +16,25 @@ import ( const ( observabilityComponentName = "github.com/ssvlabs/ssv/network/peers/connections" observabilityNamespace = "ssv.p2p.connections" + + connectionHandshakeOutcomeAttribute = "ssv.p2p.connection.handshake.outcome" + connectionHandshakeReasonAttribute = "ssv.p2p.connection.handshake.reason" +) + +const ( + connectionHandshakeOutcomeSuccess = "success" + connectionHandshakeOutcomeFailure = "failure" + connectionHandshakeOutcomeIgnored = "ignored" + + connectionHandshakeReasonSuccess = "success" + connectionHandshakeReasonAlreadyHandling = "already_handling" + connectionHandshakeReasonAlreadyConnected = "already_connected_or_connecting" + connectionHandshakeReasonTimeoutNoHandshake = "timeout_no_handshake" + connectionHandshakeReasonHandshakeError = "handshake_error" + connectionHandshakeReasonLostConnection = "lost_connection" + connectionHandshakeReasonSubnetsMismatch = "subnets_mismatch" + connectionHandshakeReasonMaxPeersLimit = "max_peers_limit" + connectionHandshakeReasonPanic = "panic" ) var ( @@ -43,6 +63,19 @@ var ( observability.InstrumentName(observabilityNamespace, "gater_decisions"), metric.WithUnit("{decision}"), metric.WithDescription("total number of connection gater decisions by phase, decision, reason, direction, and highlighted peer status"))) + + connectionHandshakesCounter = metrics.New( + meter.Int64Counter( + observability.InstrumentName(observabilityNamespace, "handshakes"), + metric.WithUnit("{handshake}"), + metric.WithDescription("total number of p2p connection handshake outcomes by direction, outcome, and reason"))) + + connectionHandshakeDurationHistogram = metrics.New( + meter.Float64Histogram( + observability.InstrumentName(observabilityNamespace, "handshake_duration"), + metric.WithUnit("s"), + metric.WithDescription("p2p connection handshake duration by direction and outcome"), + metric.WithExplicitBucketBoundaries(metrics.SecondsHistogramBuckets...))) ) func recordConnected(ctx context.Context, direction network.Direction) { @@ -76,3 +109,15 @@ func recordConnectionGaterDecision( attribute.Bool("ssv.p2p.connection.gater.highlighted_peer", highlighted), )) } + +func recordConnectionHandshake(ctx context.Context, direction network.Direction, outcome string, reason string, dur time.Duration) { + connectionHandshakesCounter.Add(ctx, 1, metric.WithAttributes( + observability.NetworkDirectionAttribute(direction), + attribute.String(connectionHandshakeOutcomeAttribute, outcome), + attribute.String(connectionHandshakeReasonAttribute, reason), + )) + connectionHandshakeDurationHistogram.Record(ctx, dur.Seconds(), metric.WithAttributes( + observability.NetworkDirectionAttribute(direction), + attribute.String(connectionHandshakeOutcomeAttribute, outcome), + )) +} diff --git a/network/peers/connections/observability_test.go b/network/peers/connections/observability_test.go new file mode 100644 index 0000000000..b4116dec2d --- /dev/null +++ b/network/peers/connections/observability_test.go @@ -0,0 +1,87 @@ +package connections + +import ( + "testing" + "time" + + "github.com/libp2p/go-libp2p/core/network" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" +) + +func TestRecordConnectionHandshake(t *testing.T) { + reader := metric.NewManualReader() + provider := metric.NewMeterProvider(metric.WithReader(reader)) + previousProvider := otel.GetMeterProvider() + otel.SetMeterProvider(provider) + t.Cleanup(func() { + otel.SetMeterProvider(previousProvider) + require.NoError(t, provider.Shutdown(t.Context())) + }) + + recordConnectionHandshake(t.Context(), network.DirOutbound, connectionHandshakeOutcomeFailure, connectionHandshakeReasonHandshakeError, 125*time.Millisecond) + + var rm metricdata.ResourceMetrics + require.NoError(t, reader.Collect(t.Context(), &rm)) + + requireConnectionMetricSum(t, rm, "ssv.p2p.connections.handshakes", map[string]string{ + "ssv.p2p.connection.direction": "outbound", + connectionHandshakeOutcomeAttribute: connectionHandshakeOutcomeFailure, + connectionHandshakeReasonAttribute: connectionHandshakeReasonHandshakeError, + }) + requireConnectionMetricHistogram(t, rm, "ssv.p2p.connections.handshake_duration", map[string]string{ + "ssv.p2p.connection.direction": "outbound", + connectionHandshakeOutcomeAttribute: connectionHandshakeOutcomeFailure, + }) +} + +func requireConnectionMetricSum(t *testing.T, rm metricdata.ResourceMetrics, metricName string, attrs map[string]string) { + t.Helper() + + for _, scopeMetrics := range rm.ScopeMetrics { + for _, metric := range scopeMetrics.Metrics { + if metric.Name != metricName { + continue + } + sum, ok := metric.Data.(metricdata.Sum[int64]) + require.True(t, ok) + require.Len(t, sum.DataPoints, 1) + require.EqualValues(t, 1, sum.DataPoints[0].Value) + for key, expected := range attrs { + value, ok := sum.DataPoints[0].Attributes.Value(attribute.Key(key)) + require.True(t, ok) + require.Equal(t, expected, value.AsString()) + } + return + } + } + + t.Fatalf("%s metric was not collected", metricName) +} + +func requireConnectionMetricHistogram(t *testing.T, rm metricdata.ResourceMetrics, metricName string, attrs map[string]string) { + t.Helper() + + for _, scopeMetrics := range rm.ScopeMetrics { + for _, metric := range scopeMetrics.Metrics { + if metric.Name != metricName { + continue + } + histogram, ok := metric.Data.(metricdata.Histogram[float64]) + require.True(t, ok) + require.Len(t, histogram.DataPoints, 1) + require.EqualValues(t, 1, histogram.DataPoints[0].Count) + for key, expected := range attrs { + value, ok := histogram.DataPoints[0].Attributes.Value(attribute.Key(key)) + require.True(t, ok) + require.Equal(t, expected, value.AsString()) + } + return + } + } + + t.Fatalf("%s metric was not collected", metricName) +} diff --git a/network/peers/peertrace/observer.go b/network/peers/peertrace/observer.go index 6ade8e5bf9..fb6491643f 100644 --- a/network/peers/peertrace/observer.go +++ b/network/peers/peertrace/observer.go @@ -46,6 +46,18 @@ var ( observability.InstrumentName(observabilityNamespace, "ssv_validations"), metric.WithUnit("{message}"), metric.WithDescription("total number of SSV-level validation decisions for messages from configured highlighted peers by outcome and reason"))) + + highlightedPeerPubsubRejectsCounter = metrics.New( + meter.Int64Counter( + observability.InstrumentName(observabilityNamespace, "pubsub_rejects"), + metric.WithUnit("{message}"), + metric.WithDescription("total number of pubsub reject trace events involving configured highlighted peers by topic and reason"))) + + highlightedPeerPubsubDropsCounter = metrics.New( + meter.Int64Counter( + observability.InstrumentName(observabilityNamespace, "pubsub_drops"), + metric.WithUnit("{message}"), + metric.WithDescription("total number of pubsub drop trace events involving configured highlighted peers by event type and topic"))) ) // Config defines peers that should be highlighted in p2p logs and metrics. @@ -191,6 +203,9 @@ func (o *Observer) ObserveSSVValidation(ctx context.Context, logger *zap.Logger, zap.String("peer_source", match.Source), zap.String("ssv_validation_result", event.Outcome), zap.String("ssv_validation_reason", event.Reason), + zap.String("ssv_validation_stage", event.Stage), + zap.String("topic", event.Topic), + zap.Int("payload_size", event.PayloadSize), zap.String("role", event.Role.String()), zap.Int32("role_id", int32(event.Role)), zap.String("ssv_message_type", ssvmessage.MsgTypeToString(event.SSVMessageType)), @@ -218,6 +233,8 @@ func (o *Observer) ObserveSSVValidation(ctx context.Context, logger *zap.Logger, attribute.String("ssv.p2p.highlight.label", o.label), attribute.String("ssv.p2p.ssv_validation.result", event.Outcome), attribute.String("ssv.p2p.ssv_validation.reason", event.Reason), + attribute.String("ssv.p2p.ssv_validation.stage", event.Stage), + attribute.String("ssv.p2p.pubsub.topic", event.Topic), attribute.String("ssv.p2p.message.role", event.Role.String()), attribute.String("ssv.p2p.message.type", ssvmessage.MsgTypeToString(event.SSVMessageType)), attribute.String("ssv.p2p.qbft.message.type", qbftMessageType), @@ -225,6 +242,34 @@ func (o *Observer) ObserveSSVValidation(ctx context.Context, logger *zap.Logger, )) } +func (o *Observer) ObservePubsubReject(ctx context.Context, pid peer.ID, topic string, reason string) { + match, ok := o.Match(pid) + if !ok { + return + } + + highlightedPeerPubsubRejectsCounter.Add(ctx, 1, metric.WithAttributes( + attribute.String("ssv.p2p.highlight.label", o.label), + attribute.String("ssv.p2p.pubsub.topic", topic), + attribute.String("ssv.p2p.pubsub.reject.reason", reason), + attribute.String("ssv.p2p.peer.id", match.ID.String()), + )) +} + +func (o *Observer) ObservePubsubDrop(ctx context.Context, pid peer.ID, eventType string, topic string) { + match, ok := o.Match(pid) + if !ok { + return + } + + highlightedPeerPubsubDropsCounter.Add(ctx, 1, metric.WithAttributes( + attribute.String("ssv.p2p.highlight.label", o.label), + attribute.String("ssv.p2p.pubsub.drop.event", eventType), + attribute.String("ssv.p2p.pubsub.topic", topic), + attribute.String("ssv.p2p.peer.id", match.ID.String()), + )) +} + func splitPeerList(value string) []string { return strings.FieldsFunc(value, func(r rune) bool { return r == ',' || r == ';' || r == '\n' || r == '\t' || r == ' ' diff --git a/network/peers/peertrace/observer_test.go b/network/peers/peertrace/observer_test.go index dff6f41534..ea116c95df 100644 --- a/network/peers/peertrace/observer_test.go +++ b/network/peers/peertrace/observer_test.go @@ -7,6 +7,10 @@ import ( "github.com/libp2p/go-libp2p/core/crypto" "github.com/libp2p/go-libp2p/core/peer" "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" "go.uber.org/zap" zapobserver "go.uber.org/zap/zaptest/observer" @@ -127,12 +131,84 @@ func TestObserveSSVValidation_UsesProvidedLogger(t *testing.T) { core, logs := zapobserver.New(zap.InfoLevel) logger := zap.New(core) observer.ObserveSSVValidation(t.Context(), logger, ssvvalidation.SSVValidationEvent{ - PeerID: pid, - Outcome: ssvvalidation.SSVValidationAccepted, - Reason: "valid", + PeerID: pid, + Outcome: ssvvalidation.SSVValidationAccepted, + Reason: "valid", + Stage: ssvvalidation.SSVValidationStageComplete, + Topic: "ssv.v2.42", + PayloadSize: 128, }) require.Len(t, logs.All(), 1) require.Equal(t, "p2p highlighted peer ssv validation", logs.All()[0].Message) - require.Equal(t, ssvvalidation.SSVValidationAccepted, logs.All()[0].ContextMap()["ssv_validation_result"]) + fields := logs.All()[0].ContextMap() + require.Equal(t, ssvvalidation.SSVValidationAccepted, fields["ssv_validation_result"]) + require.Equal(t, ssvvalidation.SSVValidationStageComplete, fields["ssv_validation_stage"]) + require.Equal(t, "ssv.v2.42", fields["topic"]) + require.Equal(t, int64(128), fields["payload_size"]) +} + +func TestObservePubsubRejectAndDrop_RecordHighlightedMetrics(t *testing.T) { + reader := metric.NewManualReader() + provider := metric.NewMeterProvider(metric.WithReader(reader)) + previousProvider := otel.GetMeterProvider() + otel.SetMeterProvider(provider) + t.Cleanup(func() { + otel.SetMeterProvider(previousProvider) + require.NoError(t, provider.Shutdown(t.Context())) + }) + + observer, err := New(Config{ + Label: "attack-simulator", + Peers: attackSimulatorPublicKey, + }) + require.NoError(t, err) + + var pid peer.ID + for highlightedPeer := range observer.peers { + pid = highlightedPeer + } + + observer.ObservePubsubReject(t.Context(), pid, "ssv.v2.42", "validation failed") + observer.ObservePubsubDrop(t.Context(), pid, "drop_rpc", "multiple") + + var rm metricdata.ResourceMetrics + require.NoError(t, reader.Collect(t.Context(), &rm)) + + requirePeertraceMetricSum(t, rm, "ssv.p2p.highlighted_peer.pubsub_rejects", map[string]string{ + "ssv.p2p.highlight.label": "attack-simulator", + "ssv.p2p.pubsub.topic": "ssv.v2.42", + "ssv.p2p.pubsub.reject.reason": "validation failed", + "ssv.p2p.peer.id": pid.String(), + }) + requirePeertraceMetricSum(t, rm, "ssv.p2p.highlighted_peer.pubsub_drops", map[string]string{ + "ssv.p2p.highlight.label": "attack-simulator", + "ssv.p2p.pubsub.drop.event": "drop_rpc", + "ssv.p2p.pubsub.topic": "multiple", + "ssv.p2p.peer.id": pid.String(), + }) +} + +func requirePeertraceMetricSum(t *testing.T, rm metricdata.ResourceMetrics, metricName string, attrs map[string]string) { + t.Helper() + + for _, scopeMetrics := range rm.ScopeMetrics { + for _, metric := range scopeMetrics.Metrics { + if metric.Name != metricName { + continue + } + sum, ok := metric.Data.(metricdata.Sum[int64]) + require.True(t, ok) + require.Len(t, sum.DataPoints, 1) + require.EqualValues(t, 1, sum.DataPoints[0].Value) + for key, expected := range attrs { + value, ok := sum.DataPoints[0].Attributes.Value(attribute.Key(key)) + require.True(t, ok) + require.Equal(t, expected, value.AsString()) + } + return + } + } + + t.Fatalf("%s metric was not collected", metricName) } diff --git a/network/streams/controller.go b/network/streams/controller.go index f834018535..c9e2040954 100644 --- a/network/streams/controller.go +++ b/network/streams/controller.go @@ -60,6 +60,7 @@ func (n *streamCtrl) Request(logger *zap.Logger, peerID peer.ID, protocol protoc stream, err := n.host.NewStream(ctx, peerID, protocol) if err != nil { + recordStreamError(n.ctx, protocol, streamOperationDial, streamErrorReason(err)) return nil, err } @@ -78,9 +79,11 @@ func (n *streamCtrl) Request(logger *zap.Logger, peerID peer.ID, protocol protoc }() if err := s.WriteWithTimeout(data, n.readWriteTimeout); err != nil { + recordStreamError(n.ctx, s.Protocol(), streamOperationWriteRequest, streamErrorReason(err)) return nil, fmt.Errorf("could not write to stream: %w", err) } if err := s.CloseWrite(); err != nil { + recordStreamError(n.ctx, s.Protocol(), streamOperationCloseWrite, streamErrorReason(err)) return nil, fmt.Errorf("could not close write stream: %w", err) } res, err := s.ReadWithTimeout(n.readWriteTimeout) @@ -88,6 +91,7 @@ func (n *streamCtrl) Request(logger *zap.Logger, peerID peer.ID, protocol protoc if errors.Is(err, ErrStreamMessageTooLarge) { n.observeOversizedPayload(logger, peerID, s.Protocol(), "response") } + recordStreamError(n.ctx, s.Protocol(), streamOperationReadResponse, streamErrorReason(err)) return nil, fmt.Errorf("could not read stream msg: %w", err) } @@ -117,6 +121,7 @@ func (n *streamCtrl) HandleStream(logger *zap.Logger, stream core.Stream) ([]byt if errors.Is(err, ErrStreamMessageTooLarge) { n.observeOversizedPayload(logger, s.Conn().RemotePeer(), s.Protocol(), "request") } + recordStreamError(n.ctx, s.Protocol(), streamOperationReadRequest, streamErrorReason(err)) return nil, nil, done, fmt.Errorf("could not read stream msg: %w", err) } n.peerObserver.Observe(n.ctx, logger, "stream_request_received", s.Conn().RemotePeer(), @@ -128,6 +133,7 @@ func (n *streamCtrl) HandleStream(logger *zap.Logger, stream core.Stream) ([]byt cp := make([]byte, len(res)) copy(cp, res) if err := s.WriteWithTimeout(cp, n.readWriteTimeout); err != nil { + recordStreamError(n.ctx, s.Protocol(), streamOperationWriteResponse, streamErrorReason(err)) return fmt.Errorf("could not write to stream: %w", err) } @@ -157,3 +163,16 @@ func (n *streamCtrl) observeOversizedPayload(logger *zap.Logger, peerID peer.ID, zap.String("direction", direction), ) } + +func streamErrorReason(err error) string { + switch { + case errors.Is(err, ErrStreamMessageTooLarge): + return streamErrorReasonOversizedPayload + case errors.Is(err, context.DeadlineExceeded): + return streamErrorReasonTimeout + case errors.Is(err, libp2pnetwork.ErrReset): + return streamErrorReasonReset + default: + return streamErrorReasonError + } +} diff --git a/network/streams/observability.go b/network/streams/observability.go index 892390980b..9e4eed3f2e 100644 --- a/network/streams/observability.go +++ b/network/streams/observability.go @@ -1,6 +1,8 @@ package streams import ( + "context" + "github.com/libp2p/go-libp2p/core/protocol" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/attribute" @@ -13,6 +15,23 @@ import ( const ( observabilityName = "github.com/ssvlabs/ssv/network/streams" observabilityNamespace = "ssv.p2p.stream" + + streamOperationAttribute = "ssv.p2p.stream.operation" + streamErrorReasonAttribute = "ssv.p2p.stream.error.reason" +) + +const ( + streamOperationDial = "dial" + streamOperationWriteRequest = "write_request" + streamOperationCloseWrite = "close_write" + streamOperationReadResponse = "read_response" + streamOperationReadRequest = "read_request" + streamOperationWriteResponse = "write_response" + + streamErrorReasonError = "error" + streamErrorReasonOversizedPayload = "oversized_payload" + streamErrorReasonReset = "reset" + streamErrorReasonTimeout = "timeout" ) var ( @@ -47,6 +66,12 @@ var ( observability.InstrumentName(observabilityNamespace, "payloads.oversized"), metric.WithUnit("{payload}"), metric.WithDescription("total number of oversized stream payloads rejected"))) + + streamErrorsCounter = metrics.New( + meter.Int64Counter( + observability.InstrumentName(observabilityNamespace, "errors"), + metric.WithUnit("{error}"), + metric.WithDescription("total number of p2p stream errors by protocol, operation, and reason"))) ) func protocolIDAttribute(id protocol.ID) attribute.KeyValue { @@ -58,3 +83,11 @@ func streamDirectionAttribute(direction string) attribute.KeyValue { const attrName = "ssv.p2p.stream.direction" return attribute.String(attrName, direction) } + +func recordStreamError(ctx context.Context, id protocol.ID, operation string, reason string) { + streamErrorsCounter.Add(ctx, 1, metric.WithAttributes( + protocolIDAttribute(id), + attribute.String(streamOperationAttribute, operation), + attribute.String(streamErrorReasonAttribute, reason), + )) +} diff --git a/network/streams/observability_test.go b/network/streams/observability_test.go new file mode 100644 index 0000000000..e931ba1d55 --- /dev/null +++ b/network/streams/observability_test.go @@ -0,0 +1,54 @@ +package streams + +import ( + "testing" + + "github.com/libp2p/go-libp2p/core/protocol" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" +) + +func TestRecordStreamError(t *testing.T) { + reader := metric.NewManualReader() + provider := metric.NewMeterProvider(metric.WithReader(reader)) + previousProvider := otel.GetMeterProvider() + otel.SetMeterProvider(provider) + t.Cleanup(func() { + otel.SetMeterProvider(previousProvider) + require.NoError(t, provider.Shutdown(t.Context())) + }) + + recordStreamError(t.Context(), protocol.ID("/ssv/test"), streamOperationReadResponse, streamErrorReasonTimeout) + + var rm metricdata.ResourceMetrics + require.NoError(t, reader.Collect(t.Context(), &rm)) + + for _, scopeMetrics := range rm.ScopeMetrics { + for _, metric := range scopeMetrics.Metrics { + if metric.Name != "ssv.p2p.stream.errors" { + continue + } + sum, ok := metric.Data.(metricdata.Sum[int64]) + require.True(t, ok) + require.Len(t, sum.DataPoints, 1) + require.EqualValues(t, 1, sum.DataPoints[0].Value) + requireStreamMetricAttribute(t, sum.DataPoints[0].Attributes, "ssv.p2p.protocol.id", "/ssv/test") + requireStreamMetricAttribute(t, sum.DataPoints[0].Attributes, streamOperationAttribute, streamOperationReadResponse) + requireStreamMetricAttribute(t, sum.DataPoints[0].Attributes, streamErrorReasonAttribute, streamErrorReasonTimeout) + return + } + } + + t.Fatal("stream error metric was not collected") +} + +func requireStreamMetricAttribute(t *testing.T, set attribute.Set, key string, expected string) { + t.Helper() + + value, ok := set.Value(attribute.Key(key)) + require.True(t, ok) + require.Equal(t, expected, value.AsString()) +} diff --git a/network/topics/controller.go b/network/topics/controller.go index a59921aa7a..f7523a7f18 100644 --- a/network/topics/controller.go +++ b/network/topics/controller.go @@ -300,6 +300,11 @@ func (ctrl *topicsCtrl) listen(sub *pubsub.Subscription) error { } if err := ctrl.msgHandler(ctx, topicNameFull, msg); err != nil { + recordPubsubMessageHandlerError(ctx, topicNameFull) + ctrl.peerObserver.Observe(ctx, logger, "pubsub_message_handler_error", msg.ReceivedFrom, + zap.String("topic", topicNameFull), + zap.Error(err), + ) logger.Debug("could not handle msg", zap.Error(err)) } } @@ -320,8 +325,11 @@ func (ctrl *topicsCtrl) setupTopicValidator(name string) error { validator := ctrl.msgValidator.ValidatorForTopic(name) wrappedValidator := func(ctx context.Context, p peer.ID, pmsg *pubsub.Message) pubsub.ValidationResult { recordPubsubMessageReceived(ctx, name) + start := time.Now() result := validator(ctx, p, pmsg) - ctrl.peerObserver.ObserveValidation(ctx, ctrl.logger, p, name, validationResultString(result), len(pmsg.Data), + resultString := validationResultString(result) + recordPubsubMessageValidated(ctx, name, resultString, time.Since(start)) + ctrl.peerObserver.ObserveValidation(ctx, ctrl.logger, p, name, resultString, len(pmsg.Data), zap.Int("validation_result_code", int(result)), ) return result diff --git a/network/topics/observability.go b/network/topics/observability.go index 5b357054cc..2b5adccf09 100644 --- a/network/topics/observability.go +++ b/network/topics/observability.go @@ -2,6 +2,7 @@ package topics import ( "context" + "time" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/attribute" @@ -15,8 +16,9 @@ const ( observabilityName = "github.com/ssvlabs/ssv/network/topics" observabilityNamespace = "ssv.p2p.messages" - pubsubObservabilityNamespace = "ssv.p2p.pubsub.messages" - pubsubTopicAttributeKey = "ssv.p2p.pubsub.topic" + pubsubObservabilityNamespace = "ssv.p2p.pubsub.messages" + pubsubTopicAttributeKey = "ssv.p2p.pubsub.topic" + pubsubValidationResultAttributeKey = "ssv.p2p.pubsub.validation.result" ) var ( @@ -40,6 +42,25 @@ var ( metric.WithUnit("{message}"), metric.WithDescription("total number of messages received by the pubsub topic validator"))) + pubsubMessagesValidatedCounter = metrics.New( + meter.Int64Counter( + observability.InstrumentName(pubsubObservabilityNamespace, "validated"), + metric.WithUnit("{message}"), + metric.WithDescription("total number of messages completed by the pubsub topic validator by result"))) + + pubsubMessageValidationDurationHistogram = metrics.New( + meter.Float64Histogram( + observability.InstrumentName(pubsubObservabilityNamespace, "validation_duration"), + metric.WithUnit("s"), + metric.WithDescription("pubsub topic validator duration by result"), + metric.WithExplicitBucketBoundaries(metrics.SecondsHistogramBuckets...))) + + pubsubMessageHandlerErrorsCounter = metrics.New( + meter.Int64Counter( + observability.InstrumentName(pubsubObservabilityNamespace, "handler_errors"), + metric.WithUnit("{error}"), + metric.WithDescription("total number of messages accepted by pubsub validation but failed by the topic message handler"))) + msgIDHandlerBufferFallbackCounter = metrics.New( meter.Int64Counter( observability.InstrumentName(observabilityNamespace, "msg_id_buffer_fallback"), @@ -65,3 +86,17 @@ func messageTypeAttribute(value uint64) attribute.KeyValue { func recordPubsubMessageReceived(ctx context.Context, topic string) { pubsubMessagesReceivedCounter.Add(ctx, 1, metric.WithAttributes(pubsubTopicAttribute(topic))) } + +func pubsubValidationResultAttribute(value string) attribute.KeyValue { + return attribute.String(pubsubValidationResultAttributeKey, value) +} + +func recordPubsubMessageValidated(ctx context.Context, topic string, result string, dur time.Duration) { + attrs := metric.WithAttributes(pubsubTopicAttribute(topic), pubsubValidationResultAttribute(result)) + pubsubMessagesValidatedCounter.Add(ctx, 1, attrs) + pubsubMessageValidationDurationHistogram.Record(ctx, dur.Seconds(), attrs) +} + +func recordPubsubMessageHandlerError(ctx context.Context, topic string) { + pubsubMessageHandlerErrorsCounter.Add(ctx, 1, metric.WithAttributes(pubsubTopicAttribute(topic))) +} diff --git a/network/topics/observability_test.go b/network/topics/observability_test.go index 41f1dc650a..028d5d2fd5 100644 --- a/network/topics/observability_test.go +++ b/network/topics/observability_test.go @@ -2,14 +2,16 @@ package topics import ( "testing" + "time" "github.com/stretchr/testify/require" "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/sdk/metric" "go.opentelemetry.io/otel/sdk/metric/metricdata" ) -func TestRecordPubsubMessageReceived(t *testing.T) { +func TestRecordPubsubMessageMetrics(t *testing.T) { reader := metric.NewManualReader() provider := metric.NewMeterProvider(metric.WithReader(reader)) previousProvider := otel.GetMeterProvider() @@ -22,27 +24,77 @@ func TestRecordPubsubMessageReceived(t *testing.T) { const topic = "ssv.v2.42" recordPubsubMessageReceived(t.Context(), topic) recordPubsubMessageReceived(t.Context(), topic) + recordPubsubMessageValidated(t.Context(), topic, "reject", 150*time.Millisecond) + recordPubsubMessageValidated(t.Context(), topic, "reject", 250*time.Millisecond) + recordPubsubMessageHandlerError(t.Context(), topic) var rm metricdata.ResourceMetrics require.NoError(t, reader.Collect(t.Context(), &rm)) + requireMetricSum(t, rm, "ssv.p2p.pubsub.messages.received", 2, map[string]string{ + pubsubTopicAttributeKey: topic, + }) + requireMetricSum(t, rm, "ssv.p2p.pubsub.messages.validated", 2, map[string]string{ + pubsubTopicAttributeKey: topic, + pubsubValidationResultAttributeKey: "reject", + }) + requireMetricHistogram(t, rm, "ssv.p2p.pubsub.messages.validation_duration", 2, map[string]string{ + pubsubTopicAttributeKey: topic, + pubsubValidationResultAttributeKey: "reject", + }) + requireMetricSum(t, rm, "ssv.p2p.pubsub.messages.handler_errors", 1, map[string]string{ + pubsubTopicAttributeKey: topic, + }) +} + +func requireMetricSum(t *testing.T, rm metricdata.ResourceMetrics, metricName string, value int64, attrs map[string]string) { + t.Helper() + for _, scopeMetrics := range rm.ScopeMetrics { for _, metric := range scopeMetrics.Metrics { - if metric.Name != "ssv.p2p.pubsub.messages.received" { + if metric.Name != metricName { continue } sum, ok := metric.Data.(metricdata.Sum[int64]) require.True(t, ok) require.Len(t, sum.DataPoints, 1) - require.EqualValues(t, 2, sum.DataPoints[0].Value) + require.EqualValues(t, value, sum.DataPoints[0].Value) + requireMetricAttributes(t, sum.DataPoints[0].Attributes, attrs) + return + } + } + + t.Fatalf("%s metric was not collected", metricName) +} + +func requireMetricHistogram(t *testing.T, rm metricdata.ResourceMetrics, metricName string, count uint64, attrs map[string]string) { + t.Helper() - topicAttr, ok := sum.DataPoints[0].Attributes.Value(pubsubTopicAttributeKey) + for _, scopeMetrics := range rm.ScopeMetrics { + for _, metric := range scopeMetrics.Metrics { + if metric.Name != metricName { + continue + } + + histogram, ok := metric.Data.(metricdata.Histogram[float64]) require.True(t, ok) - require.Equal(t, topic, topicAttr.AsString()) + require.Len(t, histogram.DataPoints, 1) + require.EqualValues(t, count, histogram.DataPoints[0].Count) + requireMetricAttributes(t, histogram.DataPoints[0].Attributes, attrs) return } } - t.Fatal("pubsub received metric was not collected") + t.Fatalf("%s metric was not collected", metricName) +} + +func requireMetricAttributes(t *testing.T, set attribute.Set, attrs map[string]string) { + t.Helper() + + for key, expected := range attrs { + value, ok := set.Value(attribute.Key(key)) + require.True(t, ok) + require.Equal(t, expected, value.AsString()) + } } diff --git a/network/topics/tracer.go b/network/topics/tracer.go index 6a23b0761e..b201b31381 100644 --- a/network/topics/tracer.go +++ b/network/topics/tracer.go @@ -45,6 +45,8 @@ func (pst *psTracer) log(logger *zap.Logger, evt *ps_pb.TraceEvent) { zap.String("type", evt.GetType().String()), } var highlightedPeer peer.ID + var rejectTopic, rejectReason string + var dropEventType, dropTopic string switch evt.GetType() { case ps_pb.TraceEvent_PUBLISH_MESSAGE: msg := evt.GetPublishMessage() @@ -60,6 +62,8 @@ func (pst *psTracer) log(logger *zap.Logger, evt *ps_pb.TraceEvent) { fields = append(fields, zap.String("msgID", hex.EncodeToString(msg.GetMessageID()))) fields = append(fields, zap.String("topic", msg.GetTopic())) fields = append(fields, zap.String("reason", msg.GetReason())) + rejectTopic = msg.GetTopic() + rejectReason = msg.GetReason() case ps_pb.TraceEvent_DUPLICATE_MESSAGE: msg := evt.GetDuplicateMessage() pid, err := peer.IDFromBytes(msg.GetReceivedFrom()) @@ -69,6 +73,8 @@ func (pst *psTracer) log(logger *zap.Logger, evt *ps_pb.TraceEvent) { } fields = append(fields, zap.String("msgID", hex.EncodeToString(msg.GetMessageID()))) fields = append(fields, zap.String("topic", msg.GetTopic())) + dropEventType = strings.ToLower(evt.GetType().String()) + dropTopic = msg.GetTopic() case ps_pb.TraceEvent_DELIVER_MESSAGE: msg := evt.GetDeliverMessage() pid, err := peer.IDFromBytes(msg.GetReceivedFrom()) @@ -141,6 +147,8 @@ func (pst *psTracer) log(logger *zap.Logger, evt *ps_pb.TraceEvent) { } if meta := msg.GetMeta(); meta != nil { fields = appendMessages(fields, meta.GetMessages()) + dropEventType = strings.ToLower(evt.GetType().String()) + dropTopic = topicFromMessageMeta(meta.GetMessages()) if ctrl := meta.Control; ctrl != nil { fields = appendIHave(fields, ctrl.GetIhave()) fields = appendIWant(fields, ctrl.GetIwant()) @@ -180,13 +188,31 @@ func (pst *psTracer) log(logger *zap.Logger, evt *ps_pb.TraceEvent) { return } if highlightedPeer != "" { - pst.peerObserver.Observe(context.Background(), logger, "pubsub_trace_"+strings.ToLower(evt.GetType().String()), highlightedPeer, fields...) + ctx := context.Background() + pst.peerObserver.Observe(ctx, logger, "pubsub_trace_"+strings.ToLower(evt.GetType().String()), highlightedPeer, fields...) + if rejectTopic != "" || rejectReason != "" { + pst.peerObserver.ObservePubsubReject(ctx, highlightedPeer, rejectTopic, rejectReason) + } + if dropEventType != "" { + pst.peerObserver.ObservePubsubDrop(ctx, highlightedPeer, dropEventType, dropTopic) + } } if pst.traceLog { logger.Debug("pubsub event", fields...) } } +func topicFromMessageMeta(messages []*ps_pb.TraceEvent_MessageMeta) string { + switch len(messages) { + case 0: + return "" + case 1: + return messages[0].GetTopic() + default: + return "multiple" + } +} + func appendMessages(fields []zap.Field, messages []*ps_pb.TraceEvent_MessageMeta) []zap.Field { if len(messages) == 0 { return fields