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

txHandler: add more metric #4786

Merged
merged 14 commits into from
Nov 17, 2022
Merged

txHandler: add more metric #4786

Show file tree
Hide file tree
Changes from 9 commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
f79a445
txhandler: add more metric
algorandskiy Nov 14, 2022
6d1f0ad
add TxHandler backlog size gauge
cce Nov 14, 2022
f234c96
Make less allocations in postProcessReportErrors
algorandskiy Nov 14, 2022
a12fe42
Add unit tests
algorandskiy Nov 15, 2022
5fd63cb
Add txsync counters
algorandskiy Nov 15, 2022
96d46e5
Add generic algod_transaction_messages_backlog_err metric
algorandskiy Nov 16, 2022
e5bcf91
Add algod_transaction_messages_remember
algorandskiy Nov 16, 2022
b9a1d9b
Merge remote-tracking branch 'upstream/master' into pavel/telemetry
cce Nov 17, 2022
17a100b
add TX, AV, PP, MI counters to DisconnectPeerEventDetails and PeerCon…
cce Nov 17, 2022
9188690
use bool instead of seenPrioTags
cce Nov 17, 2022
937818e
fix
cce Nov 17, 2022
2c29d72
add containsPPTag
cce Nov 17, 2022
ab035ca
Fix tests
algorandskiy Nov 17, 2022
d998917
do no re-allocate error
algorandskiy Nov 17, 2022
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
103 changes: 80 additions & 23 deletions data/transactions/verify/txn.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,52 @@ type GroupContext struct {
ledger logic.LedgerForSignature
}

var errTxGroupInvalidFee = errors.New("txgroup fee requirement overflow")
var errTxnSigHasNoSig = errors.New("signedtxn has no sig")
var errTxnSigNotWellFormed = errors.New("signedtxn should only have one of Sig or Msig or LogicSig")
var errRekeyingNotSupported = errors.New("nonempty AuthAddr but rekeying is not supported")
var errUnknownSignature = errors.New("has one mystery sig. WAT?")

// TxGroupErrorReason is reason code for ErrTxGroupError
type TxGroupErrorReason int

const (
// TxGroupErrorReasonGeneric is a generic (not tracked) reason code
TxGroupErrorReasonGeneric TxGroupErrorReason = iota
// TxGroupErrorReasonNotWellFormed is txn.WellFormed failure
TxGroupErrorReasonNotWellFormed
// TxGroupErrorReasonInvalidFee is invalid fee pooling in transaction group
TxGroupErrorReasonInvalidFee
// TxGroupErrorReasonHasNoSig is for transaction without any signature
TxGroupErrorReasonHasNoSig
// TxGroupErrorReasonSigNotWellFormed defines signature format errors
TxGroupErrorReasonSigNotWellFormed
// TxGroupErrorReasonMsigNotWellFormed defines multisig format errors
TxGroupErrorReasonMsigNotWellFormed
// TxGroupErrorReasonLogicSigFailed defines logic sig validation errors
TxGroupErrorReasonLogicSigFailed

// TxGroupErrorReasonNumValues is number of enum values
TxGroupErrorReasonNumValues
)

// ErrTxGroupError is an error from txn pre-validation (well form-ness, signature format, etc).
// It can be unwrapped into underlying error, as well as has a specific failure reason code.
type ErrTxGroupError struct {
err error
Reason TxGroupErrorReason
}

// Error returns an error message from the underlying error
func (e *ErrTxGroupError) Error() string {
return e.err.Error()
}

// Unwrap returns an underlying error
func (e *ErrTxGroupError) Unwrap() error {
return e.err
}

// PrepareGroupContext prepares a verification group parameter object for a given transaction
// group.
func PrepareGroupContext(group []transactions.SignedTxn, contextHdr bookkeeping.BlockHeader, ledger logic.LedgerForSignature) (*GroupContext, error) {
Expand Down Expand Up @@ -101,14 +147,14 @@ func (g *GroupContext) Equal(other *GroupContext) bool {

// txnBatchPrep verifies a SignedTxn having no obviously inconsistent data.
// Block-assembly time checks of LogicSig and accounting rules may still block the txn.
// it is the caller responsibility to call batchVerifier.Verify()
func txnBatchPrep(s *transactions.SignedTxn, txnIdx int, groupCtx *GroupContext, verifier *crypto.BatchVerifier) error {
// It is the caller responsibility to call batchVerifier.Verify().
func txnBatchPrep(s *transactions.SignedTxn, txnIdx int, groupCtx *GroupContext, verifier *crypto.BatchVerifier) *ErrTxGroupError {
if !groupCtx.consensusParams.SupportRekeying && (s.AuthAddr != basics.Address{}) {
return errors.New("nonempty AuthAddr but rekeying is not supported")
return &ErrTxGroupError{err: errRekeyingNotSupported, Reason: TxGroupErrorReasonGeneric}
}

if err := s.Txn.WellFormed(groupCtx.specAddrs, groupCtx.consensusParams); err != nil {
return err
return &ErrTxGroupError{err: err, Reason: TxGroupErrorReasonNotWellFormed}
}

return stxnCoreChecks(s, txnIdx, groupCtx, verifier)
Expand Down Expand Up @@ -144,10 +190,14 @@ func txnGroupBatchPrep(stxs []transactions.SignedTxn, contextHdr bookkeeping.Blo
minFeeCount := uint64(0)
feesPaid := uint64(0)
for i, stxn := range stxs {
err = txnBatchPrep(&stxn, i, groupCtx, verifier)
if err != nil {
err = fmt.Errorf("transaction %+v invalid : %w", stxn, err)
return
prepErr := txnBatchPrep(&stxn, i, groupCtx, verifier)
if prepErr != nil {
// re-wrap the error, take underlying one and copy the reason code
err = &ErrTxGroupError{
err: fmt.Errorf("transaction %+v invalid : %w", stxn, prepErr.err),
Copy link
Contributor

Choose a reason for hiding this comment

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

I think txnBatchPrep is already returning ErrTxGroupError, maybe you would just want to keep the error it created (not make a new one), which preserves .Reason, and set .err on it to have this wrapped fmt?

Reason: prepErr.Reason,
}
return nil, err
}
if stxn.Txn.Type != protocol.StateProofTx {
minFeeCount++
Expand All @@ -156,22 +206,27 @@ func txnGroupBatchPrep(stxs []transactions.SignedTxn, contextHdr bookkeeping.Blo
}
feeNeeded, overflow := basics.OMul(groupCtx.consensusParams.MinTxnFee, minFeeCount)
if overflow {
err = fmt.Errorf("txgroup fee requirement overflow")
return
err = &ErrTxGroupError{err: errTxGroupInvalidFee, Reason: TxGroupErrorReasonInvalidFee}
return nil, err
}
// feesPaid may have saturated. That's ok. Since we know
// feeNeeded did not overflow, simple comparison tells us
// feesPaid was enough.
if feesPaid < feeNeeded {
err = fmt.Errorf("txgroup had %d in fees, which is less than the minimum %d * %d",
feesPaid, minFeeCount, groupCtx.consensusParams.MinTxnFee)
return
err = &ErrTxGroupError{
err: fmt.Errorf(
"txgroup had %d in fees, which is less than the minimum %d * %d",
feesPaid, minFeeCount, groupCtx.consensusParams.MinTxnFee),
Reason: TxGroupErrorReasonInvalidFee,
}
return nil, err
}

return
return groupCtx, nil
}

func stxnCoreChecks(s *transactions.SignedTxn, txnIdx int, groupCtx *GroupContext, batchVerifier *crypto.BatchVerifier) error {
// stxnCoreChecks runs signatures validity checks and enqueues signature into batchVerifier for verification.
func stxnCoreChecks(s *transactions.SignedTxn, txnIdx int, groupCtx *GroupContext, batchVerifier *crypto.BatchVerifier) *ErrTxGroupError {
numSigs := 0
hasSig := false
hasMsig := false
Expand All @@ -196,11 +251,10 @@ func stxnCoreChecks(s *transactions.SignedTxn, txnIdx int, groupCtx *GroupContex
if s.Txn.Sender == transactions.StateProofSender && s.Txn.Type == protocol.StateProofTx {
return nil
}

return errors.New("signedtxn has no sig")
return &ErrTxGroupError{err: errTxnSigHasNoSig, Reason: TxGroupErrorReasonHasNoSig}
}
if numSigs > 1 {
return errors.New("signedtxn should only have one of Sig or Msig or LogicSig")
return &ErrTxGroupError{err: errTxnSigNotWellFormed, Reason: TxGroupErrorReasonSigNotWellFormed}
}

if hasSig {
Expand All @@ -209,14 +263,17 @@ func stxnCoreChecks(s *transactions.SignedTxn, txnIdx int, groupCtx *GroupContex
}
if hasMsig {
if err := crypto.MultisigBatchPrep(s.Txn, crypto.Digest(s.Authorizer()), s.Msig, batchVerifier); err != nil {
return fmt.Errorf("multisig validation failed: %w", err)
return &ErrTxGroupError{err: fmt.Errorf("multisig validation failed: %w", err), Reason: TxGroupErrorReasonMsigNotWellFormed}
}
return nil
}
if hasLogicSig {
return logicSigVerify(s, txnIdx, groupCtx)
if err := logicSigVerify(s, txnIdx, groupCtx); err != nil {
return &ErrTxGroupError{err: err, Reason: TxGroupErrorReasonLogicSigFailed}
}
return nil
}
return errors.New("has one mystery sig. WAT?")
return &ErrTxGroupError{err: errUnknownSignature, Reason: TxGroupErrorReasonGeneric}
}

// LogicSigSanityCheck checks that the signature is valid and that the program is basically well formed.
Expand Down Expand Up @@ -254,7 +311,7 @@ func logicSigSanityCheckBatchPrep(txn *transactions.SignedTxn, groupIndex int, g
}

if groupIndex < 0 {
return errors.New("Negative groupIndex")
return errors.New("negative groupIndex")
}
txngroup := transactions.WrapSignedTxnsWithAD(groupCtx.signedGroupTxns)
ep := logic.EvalParams{
Expand Down Expand Up @@ -310,7 +367,7 @@ func logicSigVerify(txn *transactions.SignedTxn, groupIndex int, groupCtx *Group
}

if groupIndex < 0 {
return errors.New("Negative groupIndex")
return errors.New("negative groupIndex")
}
ep := logic.EvalParams{
Proto: &groupCtx.consensusParams,
Expand Down
80 changes: 73 additions & 7 deletions data/txHandler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,9 +19,11 @@ package data
import (
"bytes"
"context"
"errors"
"fmt"
"io"
"sync"
"time"

"github.com/algorand/go-algorand/config"
"github.com/algorand/go-algorand/crypto"
Expand All @@ -45,6 +47,19 @@ var txBacklogSize = config.Consensus[protocol.ConsensusCurrentVersion].MaxTxnByt
var transactionMessagesHandled = metrics.MakeCounter(metrics.TransactionMessagesHandled)
var transactionMessagesDroppedFromBacklog = metrics.MakeCounter(metrics.TransactionMessagesDroppedFromBacklog)
var transactionMessagesDroppedFromPool = metrics.MakeCounter(metrics.TransactionMessagesDroppedFromPool)
var transactionMessagesAlreadyCommitted = metrics.MakeCounter(metrics.TransactionMessagesAlreadyCommitted)
var transactionMessagesTxGroupInvalidFee = metrics.MakeCounter(metrics.TransactionMessagesTxGroupInvalidFee)
var transactionMessagesTxnNotWellFormed = metrics.MakeCounter(metrics.TransactionMessagesTxnNotWellFormed)
var transactionMessagesTxnSigNotWellFormed = metrics.MakeCounter(metrics.TransactionMessagesTxnSigNotWellFormed)
var transactionMessagesTxnMsigNotWellFormed = metrics.MakeCounter(metrics.TransactionMessagesTxnMsigNotWellFormed)
var transactionMessagesTxnLogicSig = metrics.MakeCounter(metrics.TransactionMessagesTxnLogicSig)
var transactionMessagesTxnSigVerificationFailed = metrics.MakeCounter(metrics.TransactionMessagesTxnSigVerificationFailed)
var transactionMessagesBacklogErr = metrics.MakeCounter(metrics.TransactionMessagesBacklogErr)
var transactionMessagesRemember = metrics.MakeCounter(metrics.TransactionMessagesRemember)
var transactionMessagesBacklogSizeGauge = metrics.MakeGauge(metrics.TransactionMessagesBacklogSize)

var transactionGroupTxSyncRemember = metrics.MakeCounter(metrics.TransactionGroupTxSyncRemember)
var transactionGroupTxSyncAlreadyCommitted = metrics.MakeCounter(metrics.TransactionGroupTxSyncAlreadyCommitted)

// The txBacklogMsg structure used to track a single incoming transaction from the gossip network,
type txBacklogMsg struct {
Expand Down Expand Up @@ -101,8 +116,9 @@ func (handler *TxHandler) Start() {
handler.net.RegisterHandlers([]network.TaggedMessageHandler{
{Tag: protocol.TxnTag, MessageHandler: network.HandlerFunc(handler.processIncomingTxn)},
})
handler.backlogWg.Add(1)
handler.backlogWg.Add(2)
go handler.backlogWorker()
go handler.backlogGaugeThread()
}

// Stop suspends the processing of incoming messages at the transaction handler
Expand All @@ -113,12 +129,26 @@ func (handler *TxHandler) Stop() {

func reencode(stxns []transactions.SignedTxn) []byte {
var result [][]byte
for _, stxn := range stxns {
result = append(result, protocol.Encode(&stxn))
for i := range stxns {
algorandskiy marked this conversation as resolved.
Show resolved Hide resolved
result = append(result, protocol.Encode(&stxns[i]))
}
return bytes.Join(result, nil)
}

func (handler *TxHandler) backlogGaugeThread() {
defer handler.backlogWg.Done()
ticker := time.NewTicker(5 * time.Second)
defer ticker.Stop()
for {
select {
case <-ticker.C:
transactionMessagesBacklogSizeGauge.Set(float64(len(handler.backlogQueue)))
case <-handler.ctx.Done():
return
}
}
}

// backlogWorker is the worker go routine that process the incoming messages from the postVerificationQueue and backlogQueue channels
// and dispatches them further.
func (handler *TxHandler) backlogWorker() {
Expand All @@ -132,7 +162,7 @@ func (handler *TxHandler) backlogWorker() {
if !ok {
return
}
handler.postprocessCheckedTxn(wi)
handler.postProcessCheckedTxn(wi)

// restart the loop so that we could empty out the post verification queue.
continue
Expand All @@ -146,6 +176,7 @@ func (handler *TxHandler) backlogWorker() {
return
}
if handler.checkAlreadyCommitted(wi) {
transactionMessagesAlreadyCommitted.Inc(nil)
continue
}

Expand All @@ -156,17 +187,47 @@ func (handler *TxHandler) backlogWorker() {
if !ok {
return
}
handler.postprocessCheckedTxn(wi)
handler.postProcessCheckedTxn(wi)

case <-handler.ctx.Done():
return
}
}
}

func (handler *TxHandler) postprocessCheckedTxn(wi *txBacklogMsg) {
func (handler *TxHandler) postProcessReportErrors(err error) {
algorandskiy marked this conversation as resolved.
Show resolved Hide resolved
if errors.Is(err, crypto.ErrBatchVerificationFailed) {
transactionMessagesTxnSigVerificationFailed.Inc(nil)
return
}

var txGroupErr *verify.ErrTxGroupError
if errors.As(err, &txGroupErr) {
switch txGroupErr.Reason {
case verify.TxGroupErrorReasonNotWellFormed:
transactionMessagesTxnNotWellFormed.Inc(nil)
case verify.TxGroupErrorReasonInvalidFee:
transactionMessagesTxGroupInvalidFee.Inc(nil)
case verify.TxGroupErrorReasonHasNoSig:
fallthrough
case verify.TxGroupErrorReasonSigNotWellFormed:
transactionMessagesTxnSigNotWellFormed.Inc(nil)
case verify.TxGroupErrorReasonMsigNotWellFormed:
transactionMessagesTxnMsigNotWellFormed.Inc(nil)
case verify.TxGroupErrorReasonLogicSigFailed:
transactionMessagesTxnLogicSig.Inc(nil)
default:
algorandskiy marked this conversation as resolved.
Show resolved Hide resolved
transactionMessagesBacklogErr.Inc(nil)
}
} else {
transactionMessagesBacklogErr.Inc(nil)
}
}

func (handler *TxHandler) postProcessCheckedTxn(wi *txBacklogMsg) {
if wi.verificationErr != nil {
// disconnect from peer.
handler.postProcessReportErrors(wi.verificationErr)
logging.Base().Warnf("Received a malformed tx group %v: %v", wi.unverifiedTxGroup, wi.verificationErr)
handler.net.Disconnect(wi.rawmsg.Sender)
return
Expand All @@ -185,6 +246,8 @@ func (handler *TxHandler) postprocessCheckedTxn(wi *txBacklogMsg) {
return
Copy link
Contributor

@cce cce Nov 17, 2022 •
edited
Loading

Choose a reason for hiding this comment

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

do we want to break out any of the errors where Remember() fails? Remember failing is transactionMessagesHandled - transactionMessagesRemember so we already know the total

Copy link
Contributor

Choose a reason for hiding this comment

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

follow-up: will do in separate PR

}

transactionMessagesRemember.Inc(nil)

// if we remembered without any error ( i.e. txpool wasn't full ), then we should pin these transactions.
err = handler.ledger.VerifiedTransactionCache().Pin(verifiedTxGroup)
if err != nil {
Expand All @@ -203,7 +266,7 @@ func (handler *TxHandler) asyncVerifySignature(arg interface{}) interface{} {
latest := handler.ledger.Latest()
latestHdr, err := handler.ledger.BlockHdr(latest)
if err != nil {
tx.verificationErr = fmt.Errorf("Could not get header for previous block %d: %w", latest, err)
tx.verificationErr = fmt.Errorf("could not get header for previous block %d: %w", latest, err)
logging.Base().Warnf("Could not get header for previous block %d: %v", latest, err)
} else {
// we can't use PaysetGroups here since it's using a execpool like this go-routine and we don't want to deadlock.
Expand Down Expand Up @@ -289,6 +352,7 @@ func (handler *TxHandler) processDecoded(unverifiedTxGroup []transactions.Signed
unverifiedTxGroup: unverifiedTxGroup,
}
if handler.checkAlreadyCommitted(tx) {
transactionGroupTxSyncAlreadyCommitted.Inc(nil)
return network.OutgoingMessage{}, true
}

Expand Down Expand Up @@ -319,6 +383,8 @@ func (handler *TxHandler) processDecoded(unverifiedTxGroup []transactions.Signed
return network.OutgoingMessage{}, true
}

transactionGroupTxSyncRemember.Inc(nil)

// if we remembered without any error ( i.e. txpool wasn't full ), then we should pin these transactions.
err = handler.ledger.VerifiedTransactionCache().Pin(verifiedTxGroup)
if err != nil {
Expand Down
Loading