Merge pull request #6239 from The-K-R-O-K/AndriiSlisarchuk/6128-metri…

…cs-for-tx-validator

[Access] Added metrics for transaction validator

access/validator.go

@@ -5,18 +5,21 @@ import (
 	"errors"
 	"fmt"
 
+	"github.com/rs/zerolog/log"
+
 	"github.com/onflow/cadence"
 	jsoncdc "github.com/onflow/cadence/encoding/json"
 	"github.com/onflow/cadence/runtime/parser"
 	"github.com/onflow/crypto"
 	"github.com/onflow/flow-core-contracts/lib/go/templates"
-	"github.com/rs/zerolog/log"
 
 	cadenceutils "github.com/onflow/flow-go/access/utils"
 	"github.com/onflow/flow-go/fvm"
 	"github.com/onflow/flow-go/fvm/systemcontracts"
 	"github.com/onflow/flow-go/model/flow"
+	"github.com/onflow/flow-go/module"
 	"github.com/onflow/flow-go/module/execution"
+	"github.com/onflow/flow-go/module/metrics"
 	"github.com/onflow/flow-go/state"
 	"github.com/onflow/flow-go/state/protocol"
 )
@@ -87,19 +90,28 @@ type TransactionValidationOptions struct {
 	CheckPayerBalance            bool
 }
 
+type ValidationStep struct {
+	check      func(*flow.TransactionBody) error
+	failReason string
+}
+
 type TransactionValidator struct {
-	blocks                   Blocks     // for looking up blocks to check transaction expiry
-	chain                    flow.Chain // for checking validity of addresses
-	options                  TransactionValidationOptions
-	serviceAccountAddress    flow.Address
-	limiter                  RateLimiter
-	scriptExecutor           execution.ScriptExecutor
-	verifyPayerBalanceScript []byte
+	blocks                       Blocks     // for looking up blocks to check transaction expiry
+	chain                        flow.Chain // for checking validity of addresses
+	options                      TransactionValidationOptions
+	serviceAccountAddress        flow.Address
+	limiter                      RateLimiter
+	scriptExecutor               execution.ScriptExecutor
+	verifyPayerBalanceScript     []byte
+	transactionValidationMetrics module.TransactionValidationMetrics
+
+	validationSteps []ValidationStep
 }
 
 func NewTransactionValidator(
 	blocks Blocks,
 	chain flow.Chain,
+	transactionValidationMetrics module.TransactionValidationMetrics,
 	options TransactionValidationOptions,
 	executor execution.ScriptExecutor,
 ) (*TransactionValidator, error) {
@@ -109,80 +121,68 @@ func NewTransactionValidator(
 
 	env := systemcontracts.SystemContractsForChain(chain.ChainID()).AsTemplateEnv()
 
-	return &TransactionValidator{
-		blocks:                   blocks,
-		chain:                    chain,
-		options:                  options,
-		serviceAccountAddress:    chain.ServiceAddress(),
-		limiter:                  NewNoopLimiter(),
-		scriptExecutor:           executor,
-		verifyPayerBalanceScript: templates.GenerateVerifyPayerBalanceForTxExecution(env),
-	}, nil
+	txValidator := &TransactionValidator{
+		blocks:                       blocks,
+		chain:                        chain,
+		options:                      options,
+		serviceAccountAddress:        chain.ServiceAddress(),
+		limiter:                      NewNoopLimiter(),
+		scriptExecutor:               executor,
+		verifyPayerBalanceScript:     templates.GenerateVerifyPayerBalanceForTxExecution(env),
+		transactionValidationMetrics: transactionValidationMetrics,
+	}
+
+	txValidator.initValidationSteps()
+
+	return txValidator, nil
 }
 
 func NewTransactionValidatorWithLimiter(
 	blocks Blocks,
 	chain flow.Chain,
 	options TransactionValidationOptions,
+	transactionValidationMetrics module.TransactionValidationMetrics,
 	rateLimiter RateLimiter,
 ) *TransactionValidator {
-	return &TransactionValidator{
-		blocks:                blocks,
-		chain:                 chain,
-		options:               options,
-		serviceAccountAddress: chain.ServiceAddress(),
-		limiter:               rateLimiter,
-	}
-}
-
-func (v *TransactionValidator) Validate(ctx context.Context, tx *flow.TransactionBody) (err error) {
-	// rate limit transactions for specific payers.
-	// a short term solution to prevent attacks that send too many failed transactions
-	// if a transaction is from a payer that should be rate limited, all the following
-	// checks will be skipped
-	err = v.checkRateLimitPayer(tx)
-	if err != nil {
-		return err
-	}
-
-	err = v.checkTxSizeLimit(tx)
-	if err != nil {
-		return err
-	}
-
-	err = v.checkMissingFields(tx)
-	if err != nil {
-		return err
+	txValidator := &TransactionValidator{
+		blocks:                       blocks,
+		chain:                        chain,
+		options:                      options,
+		serviceAccountAddress:        chain.ServiceAddress(),
+		limiter:                      rateLimiter,
+		transactionValidationMetrics: transactionValidationMetrics,
 	}
 
-	err = v.checkGasLimit(tx)
-	if err != nil {
-		return err
-	}
-
-	err = v.checkExpiry(tx)
-	if err != nil {
-		return err
-	}
+	txValidator.initValidationSteps()
 
-	err = v.checkCanBeParsed(tx)
-	if err != nil {
-		return err
-	}
+	return txValidator
+}
 
-	err = v.checkAddresses(tx)
-	if err != nil {
-		return err
+func (v *TransactionValidator) initValidationSteps() {
+	v.validationSteps = []ValidationStep{
+		// rate limit transactions for specific payers.
+		// a short term solution to prevent attacks that send too many failed transactions
+		// if a transaction is from a payer that should be rate limited, all the following
+		// checks will be skipped
+		{v.checkRateLimitPayer, metrics.InvalidTransactionRateLimit},
+		{v.checkTxSizeLimit, metrics.InvalidTransactionByteSize},
+		{v.checkMissingFields, metrics.IncompleteTransaction},
+		{v.checkGasLimit, metrics.InvalidGasLimit},
+		{v.checkExpiry, metrics.ExpiredTransaction},
+		{v.checkCanBeParsed, metrics.InvalidScript},
+		{v.checkAddresses, metrics.InvalidAddresses},
+		{v.checkSignatureFormat, metrics.InvalidSignature},
+		{v.checkSignatureDuplications, metrics.DuplicatedSignature},
 	}
+}
 
-	err = v.checkSignatureFormat(tx)
-	if err != nil {
-		return err
-	}
+func (v *TransactionValidator) Validate(ctx context.Context, tx *flow.TransactionBody) (err error) {
 
-	err = v.checkSignatureDuplications(tx)
-	if err != nil {
-		return err
+	for _, step := range v.validationSteps {
+		if err = step.check(tx); err != nil {
+			v.transactionValidationMetrics.TransactionValidationFailed(step.failReason)
+			return err
+		}
 	}
 
 	err = v.checkSufficientBalanceToPayForTransaction(ctx, tx)
@@ -192,15 +192,19 @@ func (v *TransactionValidator) Validate(ctx context.Context, tx *flow.Transactio
 		// are 'internal' and related to script execution process. they shouldn't
 		// prevent the transaction from proceeding.
 		if IsInsufficientBalanceError(err) {
+			v.transactionValidationMetrics.TransactionValidationFailed(metrics.InsufficientBalance)
 			return err
 		}
 
 		// log and ignore all other errors
+		v.transactionValidationMetrics.TransactionValidationSkipped()
 		log.Info().Err(err).Msg("check payer validation skipped due to error")
 	}
 
 	// TODO replace checkSignatureFormat by verifying the account/payer signatures
 
+	v.transactionValidationMetrics.TransactionValidated()
+
 	return nil
 }
 
@@ -328,7 +332,6 @@ func (v *TransactionValidator) checkCanBeParsed(tx *flow.TransactionBody) error
 }
 
 func (v *TransactionValidator) checkAddresses(tx *flow.TransactionBody) error {
-
 	for _, address := range append(tx.Authorizers, tx.Payer) {
 		// we check whether this is a valid output of the address generator
 		if !v.chain.IsValid(address) {
@@ -356,7 +359,6 @@ func (v *TransactionValidator) checkSignatureDuplications(tx *flow.TransactionBo
 }
 
 func (v *TransactionValidator) checkSignatureFormat(tx *flow.TransactionBody) error {
-
 	for _, signature := range append(tx.PayloadSignatures, tx.EnvelopeSignatures...) {
 		// check the format of the signature is valid.
 		// a valid signature is an ECDSA signature of either P-256 or secp256k1 curve.

access/validator_test.go

@@ -17,7 +17,9 @@ import (
 	accessmock "github.com/onflow/flow-go/access/mock"
 	"github.com/onflow/flow-go/fvm"
 	"github.com/onflow/flow-go/model/flow"
+	"github.com/onflow/flow-go/module"
 	execmock "github.com/onflow/flow-go/module/execution/mock"
+	"github.com/onflow/flow-go/module/metrics"
 	"github.com/onflow/flow-go/utils/unittest"
 )
 
@@ -31,9 +33,11 @@ type TransactionValidatorSuite struct {
 	header           *flow.Header
 	chain            flow.Chain
 	validatorOptions access.TransactionValidationOptions
+	metrics          module.TransactionValidationMetrics
 }
 
 func (s *TransactionValidatorSuite) SetupTest() {
+	s.metrics = metrics.NewNoopCollector()
 	s.blocks = accessmock.NewBlocks(s.T())
 	assert.NotNil(s.T(), s.blocks)
 
@@ -89,7 +93,7 @@ func (s *TransactionValidatorSuite) TestTransactionValidator_ScriptExecutorInter
 		Return(nil, errors.New("script executor internal error")).
 		Once()
 
-	validator, err := access.NewTransactionValidator(s.blocks, s.chain, s.validatorOptions, scriptExecutor)
+	validator, err := access.NewTransactionValidator(s.blocks, s.chain, s.metrics, s.validatorOptions, scriptExecutor)
 	assert.NoError(s.T(), err)
 	assert.NotNil(s.T(), validator)
 
@@ -116,7 +120,7 @@ func (s *TransactionValidatorSuite) TestTransactionValidator_SufficientBalance()
 		Return(actualResponse, nil).
 		Once()
 
-	validator, err := access.NewTransactionValidator(s.blocks, s.chain, s.validatorOptions, scriptExecutor)
+	validator, err := access.NewTransactionValidator(s.blocks, s.chain, s.metrics, s.validatorOptions, scriptExecutor)
 	assert.NoError(s.T(), err)
 	assert.NotNil(s.T(), validator)
 
@@ -147,7 +151,7 @@ func (s *TransactionValidatorSuite) TestTransactionValidator_InsufficientBalance
 	assert.NoError(s.T(), err)
 	assert.NotNil(s.T(), actualAccountResponse)
 
-	validator, err := access.NewTransactionValidator(s.blocks, s.chain, s.validatorOptions, scriptExecutor)
+	validator, err := access.NewTransactionValidator(s.blocks, s.chain, s.metrics, s.validatorOptions, scriptExecutor)
 	assert.NoError(s.T(), err)
 	assert.NotNil(s.T(), validator)
 

cmd/access/node_builder/access_node_builder.go

@@ -287,43 +287,44 @@ type FlowAccessNodeBuilder struct {
 	*AccessNodeConfig
 
 	// components
-	FollowerState              protocol.FollowerState
-	SyncCore                   *chainsync.Core
-	RpcEng                     *rpc.Engine
-	FollowerDistributor        *consensuspubsub.FollowerDistributor
-	CollectionRPC              access.AccessAPIClient
-	TransactionTimings         *stdmap.TransactionTimings
-	CollectionsToMarkFinalized *stdmap.Times
-	CollectionsToMarkExecuted  *stdmap.Times
-	BlocksToMarkExecuted       *stdmap.Times
-	TransactionMetrics         *metrics.TransactionCollector
-	RestMetrics                *metrics.RestCollector
-	AccessMetrics              module.AccessMetrics
-	PingMetrics                module.PingMetrics
-	Committee                  hotstuff.DynamicCommittee
-	Finalized                  *flow.Header // latest finalized block that the node knows of at startup time
-	Pending                    []*flow.Header
-	FollowerCore               module.HotStuffFollower
-	Validator                  hotstuff.Validator
-	ExecutionDataDownloader    execution_data.Downloader
-	PublicBlobService          network.BlobService
-	ExecutionDataRequester     state_synchronization.ExecutionDataRequester
-	ExecutionDataStore         execution_data.ExecutionDataStore
-	ExecutionDataBlobstore     blobs.Blobstore
-	ExecutionDataCache         *execdatacache.ExecutionDataCache
-	ExecutionIndexer           *indexer.Indexer
-	ExecutionIndexerCore       *indexer.IndexerCore
-	ScriptExecutor             *backend.ScriptExecutor
-	RegistersAsyncStore        *execution.RegistersAsyncStore
-	Reporter                   *index.Reporter
-	EventsIndex                *index.EventsIndex
-	TxResultsIndex             *index.TransactionResultsIndex
-	IndexerDependencies        *cmd.DependencyList
-	collectionExecutedMetric   module.CollectionExecutedMetric
-	ExecutionDataPruner        *pruner.Pruner
-	ExecutionDatastoreManager  edstorage.DatastoreManager
-	ExecutionDataTracker       tracker.Storage
-	VersionControl             *version.VersionControl
+	FollowerState                protocol.FollowerState
+	SyncCore                     *chainsync.Core
+	RpcEng                       *rpc.Engine
+	FollowerDistributor          *consensuspubsub.FollowerDistributor
+	CollectionRPC                access.AccessAPIClient
+	TransactionTimings           *stdmap.TransactionTimings
+	CollectionsToMarkFinalized   *stdmap.Times
+	CollectionsToMarkExecuted    *stdmap.Times
+	BlocksToMarkExecuted         *stdmap.Times
+	TransactionMetrics           *metrics.TransactionCollector
+	TransactionValidationMetrics *metrics.TransactionValidationCollector
+	RestMetrics                  *metrics.RestCollector
+	AccessMetrics                module.AccessMetrics
+	PingMetrics                  module.PingMetrics
+	Committee                    hotstuff.DynamicCommittee
+	Finalized                    *flow.Header // latest finalized block that the node knows of at startup time
+	Pending                      []*flow.Header
+	FollowerCore                 module.HotStuffFollower
+	Validator                    hotstuff.Validator
+	ExecutionDataDownloader      execution_data.Downloader
+	PublicBlobService            network.BlobService
+	ExecutionDataRequester       state_synchronization.ExecutionDataRequester
+	ExecutionDataStore           execution_data.ExecutionDataStore
+	ExecutionDataBlobstore       blobs.Blobstore
+	ExecutionDataCache           *execdatacache.ExecutionDataCache
+	ExecutionIndexer             *indexer.Indexer
+	ExecutionIndexerCore         *indexer.IndexerCore
+	ScriptExecutor               *backend.ScriptExecutor
+	RegistersAsyncStore          *execution.RegistersAsyncStore
+	Reporter                     *index.Reporter
+	EventsIndex                  *index.EventsIndex
+	TxResultsIndex               *index.TransactionResultsIndex
+	IndexerDependencies          *cmd.DependencyList
+	collectionExecutedMetric     module.CollectionExecutedMetric
+	ExecutionDataPruner          *pruner.Pruner
+	ExecutionDatastoreManager    edstorage.DatastoreManager
+	ExecutionDataTracker         tracker.Storage
+	VersionControl               *version.VersionControl
 
 	// The sync engine participants provider is the libp2p peer store for the access node
 	// which is not available until after the network has started.
@@ -1665,6 +1666,10 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 			)
 			return nil
 		}).
+		Module("transaction validation metrics", func(node *cmd.NodeConfig) error {
+			builder.TransactionValidationMetrics = metrics.NewTransactionValidationCollector()
+			return nil
+		}).
 		Module("rest metrics", func(node *cmd.NodeConfig) error {
 			m, err := metrics.NewRestCollector(routes.URLToRoute, node.MetricsRegisterer)
 			if err != nil {
@@ -1676,6 +1681,7 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 		Module("access metrics", func(node *cmd.NodeConfig) error {
 			builder.AccessMetrics = metrics.NewAccessCollector(
 				metrics.WithTransactionMetrics(builder.TransactionMetrics),
+				metrics.WithTransactionValidationMetrics(builder.TransactionValidationMetrics),
 				metrics.WithBackendScriptsMetrics(builder.TransactionMetrics),
 				metrics.WithRestMetrics(builder.RestMetrics),
 			)