Add latency metrics to assembler

docs/monitoring/metrics.md

@@ -44,6 +44,17 @@ It queries Prometheus for metric data and displays it as dashboards, graphs, and
   **Name:** "blocks_count_total"  
   **Help:** "The total number of blocks committed to the ledger."
 
+- **Namespace:** "assembler"  
+  **Name:** "batch_fetch_latency_seconds"
+  **Help:** "The latency to fetch a requested batch from the batcher."
+
+- **Namespace:** "assembler"  
+  **Name:** "ba_to_batch_latency_seconds"
+  **Help:** "The latency from receiving a batch attestation until the matching batch is available."
+
+- **Namespace:** "assembler"  
+  **Name:** "batch_ledger_append_latency_seconds"
+  **Help:** "The latency to append a batch to the ledger."
 ---
 
 ### Batcher
@@ -88,7 +99,7 @@ It queries Prometheus for metric data and displays it as dashboards, graphs, and
   **Name:** "batch_mempool_next_requests_latency_seconds"  
   **Help:** "The latency for the primary to retrieve the next batch from the mempool."
 
-- **Namespace:** "batcher"
+- **Namespace:** "batcher"  
   **Name:** "batch_verify_latency_seconds"
   **Help:** "The latency from receiving a batch on the secondary until it is verified."
 

node/assembler/assembler.go

@@ -199,7 +199,7 @@ func (a *Assembler) initFromConfig(
 	index := prefetchIndexFactory.Create(shardIds, partyIds, a.logger, nodeConfig.PrefetchEvictionTtl, nodeConfig.PrefetchBufferMemoryBytes, nodeConfig.BatchRequestsChannelSize, &DefaultTimerFactory{}, &DefaultBatchCacheFactory{}, &DefaultPartitionPrefetchIndexerFactory{}, nodeConfig.PopWaitMonitorTimeout)
 	br := batchBringerFactory.Create(batchFrontier, nodeConfig, a.logger)
 
-	a.prefetcher = prefetcherFactory.Create(shardIds, partyIds, index, br, a.logger)
+	a.prefetcher = prefetcherFactory.Create(shardIds, partyIds, index, br, a.metrics, a.logger)
 
 	channelID := nodeConfig.Bundle.ConfigtxValidator().ChannelID()
 	baReplicator := consensusBringerFactory.Create(channelID, nodeConfig.Consenter.TLSCACerts, nodeConfig.TLSPrivateKeyFile, nodeConfig.TLSCertificateFile, nodeConfig.Consenter.Endpoint, a.ledger, a.logger)
@@ -211,6 +211,7 @@ func (a *Assembler) initFromConfig(
 		Ledger:                            a.ledger,
 		ShardCount:                        len(nodeConfig.Shards),
 		ConfigProcessor:                   a,
+		Metrics:                           a.metrics,
 	}
 
 	a.ds = NewAssemblerDeliverService(a.ledger.LedgerReader(), a.logger, nodeConfig, a.metrics.deliverMetrics)

node/assembler/assembler_test.go

@@ -166,7 +166,7 @@ func (at *assemblerTest) StartAssembler() {
 	prefetchIndexerFactory := &assembler.DefaultPrefetchIndexerFactory{}
 
 	prefetcherFactoryMock := &assembler_mocks.FakePrefetcherFactory{}
-	prefetcherFactoryMock.CreateCalls(func(si []types.ShardID, pi1 []types.PartyID, pi2 assembler.PrefetchIndexer, bb assembler.BatchBringer, l *flogging.FabricLogger) assembler.PrefetcherController {
+	prefetcherFactoryMock.CreateCalls(func(si []types.ShardID, pi1 []types.PartyID, pi2 assembler.PrefetchIndexer, bb assembler.BatchBringer, m *assembler.Metrics, l *flogging.FabricLogger) assembler.PrefetcherController {
 		return at.prefetcherMock
 	})
 

node/assembler/collator.go

@@ -50,6 +50,7 @@ type Collator struct {
 	Shards                            []types.ShardID
 	runningWG                         sync.WaitGroup
 	ConfigProcessor                   ConfigProcessor
+	Metrics                           *Metrics
 }
 
 // Run starts a go routine which processes incoming ordered batch attestations from consensus
@@ -88,16 +89,21 @@ func (c *Collator) processOrderedBatchAttestations() {
 
 			continue // skip collating for BA's with config blocks
 		}
-
+		baToBatchStart := time.Now()
 		batch, err := c.collateAttestationWithBatch(oba.BatchAttestation())
+		c.Metrics.baToBatchLatency.Observe(time.Since(baToBatchStart).Seconds())
+
 		if err != nil {
 			if errors.Is(err, utils.ErrOperationCancelled) {
 				c.Logger.Warnf("Collating Attestation with batch %v was cancelled.", oba.BatchAttestation())
 				break
 			}
 			c.Logger.Panicf("Something went wrong while fetching the batch %v", oba.BatchAttestation())
 		}
+
+		appendStart := time.Now()
 		c.Ledger.Append(batch, oba.OrderingInformation)
+		c.Metrics.batchLedgerAppendLatency.Observe(time.Since(appendStart).Seconds())
 	}
 	c.Logger.Infof("Finished processing incoming OrderedBatchAttestations from consensus")
 }

node/assembler/collator_test.go

@@ -15,11 +15,13 @@ import (
 	"github.com/hyperledger/fabric-protos-go-apiv2/common"
 	"github.com/hyperledger/fabric-x-common/protoutil"
 	"github.com/hyperledger/fabric-x-orderer/common/monitoring"
+	"github.com/hyperledger/fabric-x-orderer/common/operations"
 	"github.com/hyperledger/fabric-x-orderer/common/types"
 	"github.com/hyperledger/fabric-x-orderer/common/utils"
 	"github.com/hyperledger/fabric-x-orderer/config/generate"
 	"github.com/hyperledger/fabric-x-orderer/node/assembler"
 	assembler_mocks "github.com/hyperledger/fabric-x-orderer/node/assembler/mocks"
+	"github.com/hyperledger/fabric-x-orderer/node/config"
 	"github.com/hyperledger/fabric-x-orderer/node/consensus/state"
 	node_ledger "github.com/hyperledger/fabric-x-orderer/node/ledger"
 	"github.com/hyperledger/fabric-x-orderer/testutil"
@@ -266,6 +268,15 @@ func createCollator(t *testing.T, shardCount int, ConfigProcessor assembler.Conf
 
 	ordBARep := make(naiveOrderedBatchAttestationReplicator)
 
+	metricsCfg := &config.AssemblerNodeConfig{
+		PartyId: 1,
+		Metrics: &operations.Metrics{
+			Provider:           "disabled",
+			MetricsLogInterval: time.Second,
+		},
+	}
+	metrics := assembler.NewMetrics(metricsCfg, &node_ledger.AssemblerLedgerMetrics{}, logger)
+
 	collator := &assembler.Collator{
 		Shards:                            shards,
 		Logger:                            testutil.CreateLogger(t, 0),
@@ -274,6 +285,7 @@ func createCollator(t *testing.T, shardCount int, ConfigProcessor assembler.Conf
 		OrderedBatchAttestationReplicator: ordBARep,
 		Index:                             index,
 		ConfigProcessor:                   ConfigProcessor,
+		Metrics:                           metrics,
 	}
 	return index, ledger, ordBARep, collator
 }

node/assembler/metrics.go

@@ -12,6 +12,7 @@ import (
 	"time"
 
 	"github.com/hyperledger/fabric-lib-go/common/flogging"
+	"github.com/hyperledger/fabric-lib-go/common/metrics"
 	"github.com/hyperledger/fabric-x-orderer/common/deliver"
 	"github.com/hyperledger/fabric-x-orderer/common/monitoring"
 	arma_types "github.com/hyperledger/fabric-x-orderer/common/types"
@@ -21,15 +22,44 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 )
 
+var (
+	batchFetchLatencyOpts = metrics.HistogramOpts{
+		Namespace:  "assembler",
+		Name:       "batch_fetch_latency_seconds",
+		Help:       "The latency to fetch a requested batch from the batcher.",
+		LabelNames: []string{"party_id"},
+		Buckets:    []float64{.0001, .001, .002, .003, .004, .005, .01, .03, .05, .1, .3, .5, 1}, // TODO: adjust buckets after reviewing Grafana
+	}
+
+	baToBatchLatencyOpts = metrics.HistogramOpts{
+		Namespace:  "assembler",
+		Name:       "ba_to_batch_latency_seconds",
+		Help:       "The latency from receiving a batch attestation until the matching batch is available.",
+		LabelNames: []string{"party_id"},
+		Buckets:    []float64{.0001, .001, .002, .003, .004, .005, .01, .03, .05, .1, .3, .5, 1}, // TODO: adjust buckets after reviewing Grafana
+	}
+
+	batchLedgerAppendLatencyOpts = metrics.HistogramOpts{
+		Namespace:  "assembler",
+		Name:       "batch_ledger_append_latency_seconds",
+		Help:       "The latency to append a batch to the ledger.",
+		LabelNames: []string{"party_id"},
+		Buckets:    []float64{.0001, .001, .002, .003, .004, .005, .01, .03, .05, .1, .3, .5, 1}, // TODO: adjust buckets after reviewing Grafana
+	}
+)
+
 type Metrics struct {
-	ledgerMetrics  *node_ledger.AssemblerLedgerMetrics
-	deliverMetrics *deliver.Metrics
-	logger         *flogging.FabricLogger
-	interval       time.Duration
-	stopChan       chan struct{}
-	stopOnce       sync.Once
-	startOnce      sync.Once
-	partyID        arma_types.PartyID
+	ledgerMetrics            *node_ledger.AssemblerLedgerMetrics
+	deliverMetrics           *deliver.Metrics
+	batchFetchLatency        metrics.Histogram
+	baToBatchLatency         metrics.Histogram
+	batchLedgerAppendLatency metrics.Histogram
+	logger                   *flogging.FabricLogger
+	interval                 time.Duration
+	stopChan                 chan struct{}
+	stopOnce                 sync.Once
+	startOnce                sync.Once
+	partyID                  arma_types.PartyID
 }
 
 func NewMetrics(assemblerNodeConfig *config.AssemblerNodeConfig, ledgerMetrics *node_ledger.AssemblerLedgerMetrics, logger *flogging.FabricLogger) *Metrics {
@@ -43,13 +73,20 @@ func NewMetrics(assemblerNodeConfig *config.AssemblerNodeConfig, ledgerMetrics *
 	ledgerMetrics.NewAssemblerLedgerMetrics(provider, partyID, logger)
 	deliverMetrics := deliver.NewMetrics(provider)
 
+	batchFetchLatency := provider.NewHistogram(batchFetchLatencyOpts).With([]string{partyID}...)
+	baToBatchLatency := provider.NewHistogram(baToBatchLatencyOpts).With([]string{partyID}...)
+	batchLedgerAppendLatency := provider.NewHistogram(batchLedgerAppendLatencyOpts).With([]string{partyID}...)
+
 	return &Metrics{
-		ledgerMetrics:  ledgerMetrics,
-		deliverMetrics: deliverMetrics,
-		interval:       assemblerNodeConfig.Metrics.MetricsLogInterval,
-		logger:         logger,
-		stopChan:       make(chan struct{}),
-		partyID:        assemblerNodeConfig.PartyId,
+		ledgerMetrics:            ledgerMetrics,
+		deliverMetrics:           deliverMetrics,
+		interval:                 assemblerNodeConfig.Metrics.MetricsLogInterval,
+		logger:                   logger,
+		stopChan:                 make(chan struct{}),
+		partyID:                  assemblerNodeConfig.PartyId,
+		batchFetchLatency:        batchFetchLatency,
+		baToBatchLatency:         baToBatchLatency,
+		batchLedgerAppendLatency: batchLedgerAppendLatency,
 	}
 }
 
@@ -70,7 +107,11 @@ func (m *Metrics) StopMetricsTracker() {
 		blocksCommitted := uint64(monitoring.GetMetricValue(m.ledgerMetrics.BlocksCount.(prometheus.Counter), m.logger))
 		blocksSizeCommitted := uint64(monitoring.GetMetricValue(m.ledgerMetrics.BlocksSize.(prometheus.Counter), m.logger))
 
-		m.logger.Infof("ASSEMBLER_METRICS: party_id=%d, total: TXs=%d, blocks=%d, estimated_block_size=%d", m.partyID, txCommitted, blocksCommitted, blocksSizeCommitted)
+		batchFetchLatencyAvg := monitoring.GetHistogramAverage(m.batchFetchLatency.(prometheus.Metric), m.logger)
+		baToBatchLatencyAvg := monitoring.GetHistogramAverage(m.baToBatchLatency.(prometheus.Metric), m.logger)
+		batchLedgerAppendLatencyAvg := monitoring.GetHistogramAverage(m.batchLedgerAppendLatency.(prometheus.Metric), m.logger)
+
+		m.logger.Infof("ASSEMBLER_METRICS: party_id=%d, total: TXs=%d, blocks=%d, estimated_block_size=%d, batch_fetch_latency_avg_seconds=%.6f, ba_to_batch_latency_avg_seconds=%.6f, batch_ledger_append_latency_avg_seconds=%.6f", m.partyID, txCommitted, blocksCommitted, blocksSizeCommitted, batchFetchLatencyAvg, baToBatchLatencyAvg, batchLedgerAppendLatencyAvg)
 	})
 }
 
@@ -98,8 +139,11 @@ func (m *Metrics) trackMetrics() {
 				newTXs = txCommitted - lastTxCommitted
 			}
 
-			m.logger.Infof("ASSEMBLER_METRICS: total: party_id=%d, TXs=%d, blocks=%d, estimated_block_size=%d, in the last %.2f seconds: TXs=%d, block=%d", m.partyID, txCommitted, blocksCommitted, blocksSizeCommitted, sec, newTXs, newBlocks)
+			batchFetchLatencyAvg := monitoring.GetHistogramAverage(m.batchFetchLatency.(prometheus.Metric), m.logger)
+			baToBatchLatencyAvg := monitoring.GetHistogramAverage(m.baToBatchLatency.(prometheus.Metric), m.logger)
+			batchLedgerAppendLatencyAvg := monitoring.GetHistogramAverage(m.batchLedgerAppendLatency.(prometheus.Metric), m.logger)
 
+			m.logger.Infof("ASSEMBLER_METRICS: total: party_id=%d, TXs=%d, blocks=%d, estimated_block_size=%d, batch_fetch_latency_avg_seconds=%.6f, ba_to_batch_latency_avg_seconds=%.6f, batch_ledger_append_latency_avg_seconds=%.6f, in the last %.2f seconds: TXs=%d, blocks=%d", m.partyID, txCommitted, blocksCommitted, blocksSizeCommitted, batchFetchLatencyAvg, baToBatchLatencyAvg, batchLedgerAppendLatencyAvg, sec, newTXs, newBlocks)
 			lastTxCommitted, lastBlocksCommitted = txCommitted, blocksCommitted
 		case <-m.stopChan:
 			return

node/assembler/prefetch_benchmark/prefetch_benchmark_test.go

@@ -13,9 +13,12 @@ import (
 	"time"
 
 	"github.com/hyperledger/fabric-lib-go/common/flogging"
+	"github.com/hyperledger/fabric-x-orderer/common/operations"
 	"github.com/hyperledger/fabric-x-orderer/common/types"
 	"github.com/hyperledger/fabric-x-orderer/node/assembler"
 	"github.com/hyperledger/fabric-x-orderer/node/assembler/mocks"
+	"github.com/hyperledger/fabric-x-orderer/node/config"
+	node_ledger "github.com/hyperledger/fabric-x-orderer/node/ledger"
 	"github.com/hyperledger/fabric-x-orderer/testutil"
 
 	"github.com/stretchr/testify/require"
@@ -132,11 +135,22 @@ func createPrefetcherBenchmarkSetup(
 		time.Second,
 	)
 	test.prefetchIndex = prefetchIndex
+
+	metricsCfg := &config.AssemblerNodeConfig{
+		PartyId: 1,
+		Metrics: &operations.Metrics{
+			Provider:           "disabled",
+			MetricsLogInterval: time.Second,
+		},
+	}
+	metrics := assembler.NewMetrics(metricsCfg, &node_ledger.AssemblerLedgerMetrics{}, logger)
+
 	test.prefetcher = assembler.NewPrefetcher(
 		params.shards,
 		params.parties,
 		prefetchIndex,
 		test.batchFetcherMock,
+		metrics,
 		testutil.CreateLoggerForModule(t, "Assembler Prefetcher", logLevel),
 	)
 	return test

node/assembler/prefetcher.go

@@ -10,6 +10,7 @@ import (
 	"context"
 	"errors"
 	"sync"
+	"time"
 
 	"github.com/hyperledger/fabric-lib-go/common/flogging"
 	"github.com/hyperledger/fabric-x-orderer/common/types"
@@ -23,13 +24,13 @@ type PrefetcherController interface {
 
 //go:generate counterfeiter -o ./mocks/prefetcher_factory.go . PrefetcherFactory
 type PrefetcherFactory interface {
-	Create(shards []types.ShardID, parties []types.PartyID, prefetchIndex PrefetchIndexer, batchFetcher BatchBringer, logger *flogging.FabricLogger) PrefetcherController
+	Create(shards []types.ShardID, parties []types.PartyID, prefetchIndex PrefetchIndexer, batchFetcher BatchBringer, metrics *Metrics, logger *flogging.FabricLogger) PrefetcherController
 }
 
 type DefaultPrefetcherFactory struct{}
 
-func (f *DefaultPrefetcherFactory) Create(shards []types.ShardID, parties []types.PartyID, prefetchIndex PrefetchIndexer, batchFetcher BatchBringer, logger *flogging.FabricLogger) PrefetcherController {
-	return NewPrefetcher(shards, parties, prefetchIndex, batchFetcher, logger)
+func (f *DefaultPrefetcherFactory) Create(shards []types.ShardID, parties []types.PartyID, prefetchIndex PrefetchIndexer, batchFetcher BatchBringer, metrics *Metrics, logger *flogging.FabricLogger) PrefetcherController {
+	return NewPrefetcher(shards, parties, prefetchIndex, batchFetcher, metrics, logger)
 }
 
 type Prefetcher struct {
@@ -40,6 +41,7 @@ type Prefetcher struct {
 	parties             []types.PartyID
 	cancellationContext context.Context
 	cancelContextFunc   context.CancelFunc
+	metrics             *Metrics
 	// in order to wait for all the go routines to finish
 	wg sync.WaitGroup
 }
@@ -49,6 +51,7 @@ func NewPrefetcher(
 	parties []types.PartyID,
 	prefetchIndex PrefetchIndexer,
 	batchFetcher BatchBringer,
+	metrics *Metrics,
 	logger *flogging.FabricLogger,
 ) *Prefetcher {
 	ctx, cancel := context.WithCancel(context.Background())
@@ -60,6 +63,7 @@ func NewPrefetcher(
 		parties:             parties,
 		cancellationContext: ctx,
 		cancelContextFunc:   cancel,
+		metrics:             metrics,
 	}
 	return p
 }
@@ -114,7 +118,10 @@ func (p *Prefetcher) handleBatchRequests() {
 			return
 		case batchId := <-p.prefetchIndex.Requests():
 			go func(batchId types.BatchID) {
+				fetchStart := time.Now()
 				batch, err := p.batchFetcher.GetBatch(batchId)
+				p.metrics.batchFetchLatency.Observe(time.Since(fetchStart).Seconds())
+
 				if err != nil {
 					if errors.Is(err, context.Canceled) {
 						p.logger.Infof("fetch canceled")