Merge pull request #8030 from onflow/yurii/7391-extend-collection-building-metrics

[Overload Resilience] Metrics for prioritized transactions

Author: durkmurder

module/builder/collection/builder.go

@@ -191,7 +191,7 @@ func (b *Builder) BuildOn(parentID flow.Identifier, setter func(*flow.HeaderBody
 	// STEP 2: build a payload of valid transactions, while at the same
 	// time figuring out the correct reference block ID for the collection.
 	span, _ = b.tracer.StartSpanFromContext(ctx, trace.COLBuildOnCreatePayload)
-	payload, err := b.buildPayload(buildCtx)
+	payload, priorityTransactionsCount, err := b.buildPayload(buildCtx)
 	span.End()
 	if err != nil {
 		return nil, fmt.Errorf("could not build payload: %w", err)
@@ -216,6 +216,8 @@ func (b *Builder) BuildOn(parentID flow.Identifier, setter func(*flow.HeaderBody
 		return nil, fmt.Errorf("could not build cluster block: %w", err)
 	}
 
+	b.metrics.ClusterBlockCreated(block, priorityTransactionsCount)
+
 	blockProposal, err := cluster.NewProposal(
 		cluster.UntrustedProposal{
 			Block:           *block,
@@ -403,8 +405,13 @@ func (b *Builder) populateFinalizedAncestryLookup(lctx lockctx.Proof, ctx *block
 
 // buildPayload constructs a valid payload based on transactions available in the mempool.
 // If the mempool is empty, an empty payload will be returned.
+// Return values:
+//   - *cluster.Payload: the payload that has been built.
+//   - uint: number of prioritized transactions included in the payload.
+//   - error: exception if failed to build the payload.
+//
 // No errors are expected during normal operation.
-func (b *Builder) buildPayload(buildCtx *blockBuildContext) (*cluster.Payload, error) {
+func (b *Builder) buildPayload(buildCtx *blockBuildContext) (*cluster.Payload, uint, error) {
 	lookup := buildCtx.lookup
 	limiter := buildCtx.limiter
 	config := buildCtx.config
@@ -469,7 +476,7 @@ func (b *Builder) buildPayload(buildCtx *blockBuildContext) (*cluster.Payload, e
 			continue // in case we are configured with liberal transaction ingest rules
 		}
 		if err != nil {
-			return nil, fmt.Errorf("could not retrieve reference header: %w", err)
+			return nil, 0, fmt.Errorf("could not retrieve reference header: %w", err)
 		}
 
 		// disallow un-finalized reference blocks, and reference blocks beyond the cluster's operating epoch
@@ -480,7 +487,7 @@ func (b *Builder) buildPayload(buildCtx *blockBuildContext) (*cluster.Payload, e
 		// make sure the reference block is finalized and not orphaned
 		blockIDFinalizedAtRefHeight, err := b.mainHeaders.BlockIDByHeight(refHeader.Height)
 		if err != nil {
-			return nil, fmt.Errorf("could not check that reference block (id=%x) for transaction (id=%x) is finalized: %w", tx.ReferenceBlockID, txID, err)
+			return nil, 0, fmt.Errorf("could not check that reference block (id=%x) for transaction (id=%x) is finalized: %w", tx.ReferenceBlockID, txID, err)
 		}
 		if blockIDFinalizedAtRefHeight != tx.ReferenceBlockID {
 			// the transaction references an orphaned block - it will never be valid
@@ -544,7 +551,7 @@ func (b *Builder) buildPayload(buildCtx *blockBuildContext) (*cluster.Payload, e
 	// build the payload from the transactions
 	collection, err := flow.NewCollection(flow.UntrustedCollection{Transactions: transactions})
 	if err != nil {
-		return nil, fmt.Errorf("could not build the collection from the transactions: %w", err)
+		return nil, 0, fmt.Errorf("could not build the collection from the transactions: %w", err)
 	}
 
 	payload, err := cluster.NewPayload(
@@ -554,9 +561,9 @@ func (b *Builder) buildPayload(buildCtx *blockBuildContext) (*cluster.Payload, e
 		},
 	)
 	if err != nil {
-		return nil, fmt.Errorf("could not build a payload: %w", err)
+		return nil, 0, fmt.Errorf("could not build a payload: %w", err)
 	}
-	return payload, nil
+	return payload, uint(len(priorityTransactions)), nil
 }
 
 // buildHeader constructs the header for the cluster block being built.

module/metrics.go

@@ -615,15 +615,16 @@ type CollectionMetrics interface {
 	// a tx->col span for the transaction.
 	TransactionIngested(txID flow.Identifier)
 
-	// ClusterBlockProposed is called when a new collection is proposed by us or
-	// any other node in the cluster.
-	ClusterBlockProposed(block *cluster.Block)
-
 	// ClusterBlockFinalized is called when a collection is finalized.
 	ClusterBlockFinalized(block *cluster.Block)
 
 	// CollectionMaxSize measures the current maximum size of a collection.
 	CollectionMaxSize(size uint)
+
+	// ClusterBlockCreated informs about cluster blocks being PROPOSED by THIS NODE.
+	// CAUTION: These metrics will represent a partial picture of cluster block creation across the network,
+	// as each node will only report on cluster blocks where they are the proposer.
+	ClusterBlockCreated(block *cluster.Block, priorityTxnsCount uint)
 }
 
 type ConsensusMetrics interface {

module/metrics/collection.go

@@ -15,8 +15,9 @@ type CollectionCollector struct {
 	transactionsIngested prometheus.Counter       // tracks the number of ingested transactions
 	finalizedHeight      *prometheus.GaugeVec     // tracks the finalized height
 	maxCollectionSize    prometheus.Gauge         // tracks the maximum collection size
-	proposals            *prometheus.HistogramVec // tracks the number/size of PROPOSED collections
 	guarantees           *prometheus.HistogramVec // counts the number/size of FINALIZED collections
+	collectionSize       *prometheus.HistogramVec // number of transactions included ONLY in the cluster blocks proposed by this node
+	priorityTxns         *prometheus.HistogramVec // number of priority transactions included ONLY in cluster blocks proposed by this node
 }
 
 var _ module.CollectionMetrics = (*CollectionCollector)(nil)
@@ -46,20 +47,28 @@ func NewCollectionCollector(tracer module.Tracer) *CollectionCollector {
 			Help:      "last used max collection size",
 		}),
 
-		proposals: promauto.NewHistogramVec(prometheus.HistogramOpts{
+		guarantees: promauto.NewHistogramVec(prometheus.HistogramOpts{
 			Namespace: namespaceCollection,
 			Subsystem: subsystemProposal,
 			Buckets:   []float64{1, 2, 5, 10, 20},
-			Name:      "proposals_size_transactions",
-			Help:      "size/number of proposed collections",
+			Name:      "guarantees_size_transactions",
+			Help:      "size/number of guaranteed/finalized collections",
 		}, []string{LabelChain}),
 
-		guarantees: promauto.NewHistogramVec(prometheus.HistogramOpts{
+		collectionSize: promauto.NewHistogramVec(prometheus.HistogramOpts{
 			Namespace: namespaceCollection,
 			Subsystem: subsystemProposal,
 			Buckets:   []float64{1, 2, 5, 10, 20},
-			Name:      "guarantees_size_transactions",
-			Help:      "size/number of guaranteed/finalized collections",
+			Name:      "collection_size",
+			Help:      "number of transactions included ONLY in the cluster blocks proposed by this node",
+		}, []string{LabelChain}),
+
+		priorityTxns: promauto.NewHistogramVec(prometheus.HistogramOpts{
+			Namespace: namespaceCollection,
+			Subsystem: subsystemProposal,
+			Buckets:   []float64{1, 2, 5, 10, 20},
+			Name:      "priority_transactions",
+			Help:      "number of priority transactions included ONLY in cluster blocks proposed by this node",
 		}, []string{LabelChain}),
 	}
 
@@ -72,33 +81,38 @@ func (cc *CollectionCollector) TransactionIngested(txID flow.Identifier) {
 	cc.transactionsIngested.Inc()
 }
 
-// ClusterBlockProposed tracks the size and number of proposals, as well as
-// starting the collection->guarantee span.
-func (cc *CollectionCollector) ClusterBlockProposed(block *cluster.Block) {
-	collection := block.Payload.Collection.Light()
-
-	cc.proposals.
-		With(prometheus.Labels{LabelChain: block.ChainID.String()}).
-		Observe(float64(collection.Len()))
-}
-
 // ClusterBlockFinalized updates the guaranteed collection size gauge and
 // finishes the tx->collection span for each constituent transaction.
 func (cc *CollectionCollector) ClusterBlockFinalized(block *cluster.Block) {
-	collection := block.Payload.Collection.Light()
-	chainID := block.ChainID
+	chainID := block.ChainID.String()
 
 	cc.finalizedHeight.
-		With(prometheus.Labels{LabelChain: chainID.String()}).
+		With(prometheus.Labels{LabelChain: chainID}).
 		Set(float64(block.Height))
 	cc.guarantees.
 		With(prometheus.Labels{
-			LabelChain: chainID.String(),
+			LabelChain: chainID,
 		}).
-		Observe(float64(collection.Len()))
+		Observe(float64(block.Payload.Collection.Len()))
 }
 
 // CollectionMaxSize measures the current maximum size of a collection.
 func (cc *CollectionCollector) CollectionMaxSize(size uint) {
 	cc.maxCollectionSize.Set(float64(size))
 }
+
+// ClusterBlockCreated informs about cluster block being proposed by this node.
+// CAUTION: These metrics will represent a partial picture of cluster block creation across the network,
+// as each node will only report on cluster blocks where they are the proposer.
+// It reports several metrics, specifically how many transactions have been included and how many of them are priority txns.
+func (cc *CollectionCollector) ClusterBlockCreated(block *cluster.Block, priorityTxnsCount uint) {
+	chainID := block.ChainID.String()
+
+	cc.collectionSize.
+		With(prometheus.Labels{LabelChain: chainID}).
+		Observe(float64(block.Payload.Collection.Len()))
+
+	cc.priorityTxns.
+		With(prometheus.Labels{LabelChain: chainID}).
+		Observe(float64(priorityTxnsCount))
+}

module/metrics/noop.go

@@ -125,9 +125,9 @@ func (nc *NoopCollector) ValidatorProcessingDuration(duration time.Duration)
 func (nc *NoopCollector) PayloadProductionDuration(duration time.Duration)               {}
 func (nc *NoopCollector) TimeoutCollectorsRange(uint64, uint64, int)                     {}
 func (nc *NoopCollector) TransactionIngested(txID flow.Identifier)                       {}
-func (nc *NoopCollector) ClusterBlockProposed(*cluster.Block)                            {}
 func (nc *NoopCollector) ClusterBlockFinalized(*cluster.Block)                           {}
 func (nc *NoopCollector) CollectionMaxSize(uint)                                         {}
+func (nc *NoopCollector) ClusterBlockCreated(*cluster.Block, uint)                       {}
 func (nc *NoopCollector) StartCollectionToFinalized(collectionID flow.Identifier)        {}
 func (nc *NoopCollector) FinishCollectionToFinalized(collectionID flow.Identifier)       {}
 func (nc *NoopCollector) StartBlockToSeal(blockID flow.Identifier)                       {}