From 487c816b4825bb017a9cdca5d351d3f27a1682d2 Mon Sep 17 00:00:00 2001 From: Natalie Morad Date: Tue, 7 Jul 2026 14:40:27 +0300 Subject: [PATCH] Add latency metrics to assembler Signed-off-by: Natalie Morad --- docs/monitoring/metrics.md | 13 +++- node/assembler/assembler.go | 3 +- node/assembler/assembler_test.go | 2 +- node/assembler/collator.go | 8 +- node/assembler/collator_test.go | 12 +++ node/assembler/metrics.go | 76 +++++++++++++++---- node/assembler/mocks/prefetcher_factory.go | 22 +++--- .../prefetch_benchmark_test.go | 14 ++++ node/assembler/prefetcher.go | 13 +++- node/assembler/prefetcher_test.go | 19 ++++- 10 files changed, 148 insertions(+), 34 deletions(-) diff --git a/docs/monitoring/metrics.md b/docs/monitoring/metrics.md index 2167bb723..76c157031 100644 --- a/docs/monitoring/metrics.md +++ b/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." diff --git a/node/assembler/assembler.go b/node/assembler/assembler.go index 241d3c15c..a5d59fc6c 100644 --- a/node/assembler/assembler.go +++ b/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) diff --git a/node/assembler/assembler_test.go b/node/assembler/assembler_test.go index 518012c3e..6d17d5809 100644 --- a/node/assembler/assembler_test.go +++ b/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 }) diff --git a/node/assembler/collator.go b/node/assembler/collator.go index f12332d55..7b792e535 100644 --- a/node/assembler/collator.go +++ b/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,8 +89,10 @@ 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()) @@ -97,7 +100,10 @@ func (c *Collator) processOrderedBatchAttestations() { } 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") } diff --git a/node/assembler/collator_test.go b/node/assembler/collator_test.go index bf1883598..4bc64b871 100644 --- a/node/assembler/collator_test.go +++ b/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 } diff --git a/node/assembler/metrics.go b/node/assembler/metrics.go index a6ce2ff04..2462c1a11 100644 --- a/node/assembler/metrics.go +++ b/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 diff --git a/node/assembler/mocks/prefetcher_factory.go b/node/assembler/mocks/prefetcher_factory.go index ff0999a5f..c5f52df1b 100644 --- a/node/assembler/mocks/prefetcher_factory.go +++ b/node/assembler/mocks/prefetcher_factory.go @@ -10,14 +10,15 @@ import ( ) type FakePrefetcherFactory struct { - CreateStub func([]types.ShardID, []types.PartyID, assembler.PrefetchIndexer, assembler.BatchBringer, *flogging.FabricLogger) assembler.PrefetcherController + CreateStub func([]types.ShardID, []types.PartyID, assembler.PrefetchIndexer, assembler.BatchBringer, *assembler.Metrics, *flogging.FabricLogger) assembler.PrefetcherController createMutex sync.RWMutex createArgsForCall []struct { arg1 []types.ShardID arg2 []types.PartyID arg3 assembler.PrefetchIndexer arg4 assembler.BatchBringer - arg5 *flogging.FabricLogger + arg5 *assembler.Metrics + arg6 *flogging.FabricLogger } createReturns struct { result1 assembler.PrefetcherController @@ -29,7 +30,7 @@ type FakePrefetcherFactory struct { invocationsMutex sync.RWMutex } -func (fake *FakePrefetcherFactory) Create(arg1 []types.ShardID, arg2 []types.PartyID, arg3 assembler.PrefetchIndexer, arg4 assembler.BatchBringer, arg5 *flogging.FabricLogger) assembler.PrefetcherController { +func (fake *FakePrefetcherFactory) Create(arg1 []types.ShardID, arg2 []types.PartyID, arg3 assembler.PrefetchIndexer, arg4 assembler.BatchBringer, arg5 *assembler.Metrics, arg6 *flogging.FabricLogger) assembler.PrefetcherController { var arg1Copy []types.ShardID if arg1 != nil { arg1Copy = make([]types.ShardID, len(arg1)) @@ -47,14 +48,15 @@ func (fake *FakePrefetcherFactory) Create(arg1 []types.ShardID, arg2 []types.Par arg2 []types.PartyID arg3 assembler.PrefetchIndexer arg4 assembler.BatchBringer - arg5 *flogging.FabricLogger - }{arg1Copy, arg2Copy, arg3, arg4, arg5}) + arg5 *assembler.Metrics + arg6 *flogging.FabricLogger + }{arg1Copy, arg2Copy, arg3, arg4, arg5, arg6}) stub := fake.CreateStub fakeReturns := fake.createReturns - fake.recordInvocation("Create", []interface{}{arg1Copy, arg2Copy, arg3, arg4, arg5}) + fake.recordInvocation("Create", []interface{}{arg1Copy, arg2Copy, arg3, arg4, arg5, arg6}) fake.createMutex.Unlock() if stub != nil { - return stub(arg1, arg2, arg3, arg4, arg5) + return stub(arg1, arg2, arg3, arg4, arg5, arg6) } if specificReturn { return ret.result1 @@ -68,17 +70,17 @@ func (fake *FakePrefetcherFactory) CreateCallCount() int { return len(fake.createArgsForCall) } -func (fake *FakePrefetcherFactory) CreateCalls(stub func([]types.ShardID, []types.PartyID, assembler.PrefetchIndexer, assembler.BatchBringer, *flogging.FabricLogger) assembler.PrefetcherController) { +func (fake *FakePrefetcherFactory) CreateCalls(stub func([]types.ShardID, []types.PartyID, assembler.PrefetchIndexer, assembler.BatchBringer, *assembler.Metrics, *flogging.FabricLogger) assembler.PrefetcherController) { fake.createMutex.Lock() defer fake.createMutex.Unlock() fake.CreateStub = stub } -func (fake *FakePrefetcherFactory) CreateArgsForCall(i int) ([]types.ShardID, []types.PartyID, assembler.PrefetchIndexer, assembler.BatchBringer, *flogging.FabricLogger) { +func (fake *FakePrefetcherFactory) CreateArgsForCall(i int) ([]types.ShardID, []types.PartyID, assembler.PrefetchIndexer, assembler.BatchBringer, *assembler.Metrics, *flogging.FabricLogger) { fake.createMutex.RLock() defer fake.createMutex.RUnlock() argsForCall := fake.createArgsForCall[i] - return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5 + return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5, argsForCall.arg6 } func (fake *FakePrefetcherFactory) CreateReturns(result1 assembler.PrefetcherController) { diff --git a/node/assembler/prefetch_benchmark/prefetch_benchmark_test.go b/node/assembler/prefetch_benchmark/prefetch_benchmark_test.go index 732b56cfa..8abd3c626 100644 --- a/node/assembler/prefetch_benchmark/prefetch_benchmark_test.go +++ b/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 diff --git a/node/assembler/prefetcher.go b/node/assembler/prefetcher.go index 2a15e84d9..76abc036b 100644 --- a/node/assembler/prefetcher.go +++ b/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") diff --git a/node/assembler/prefetcher_test.go b/node/assembler/prefetcher_test.go index a4a9bf021..af402dc46 100644 --- a/node/assembler/prefetcher_test.go +++ b/node/assembler/prefetcher_test.go @@ -8,10 +8,14 @@ package assembler_test import ( "testing" + "time" + "github.com/hyperledger/fabric-x-orderer/common/operations" "github.com/hyperledger/fabric-x-orderer/common/types" "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" + node_ledger "github.com/hyperledger/fabric-x-orderer/node/ledger" "github.com/hyperledger/fabric-x-orderer/testutil" "github.com/stretchr/testify/require" ) @@ -42,12 +46,25 @@ func setupPrefetcherTest(t *testing.T, shards []types.ShardID, parties []types.P vars.batchFetcherMock.ReplicateCalls(func(shard types.ShardID) <-chan types.Batch { return vars.shardToReplicationChan[shard] }) + + cfg := &config.AssemblerNodeConfig{ + PartyId: 1, + Metrics: &operations.Metrics{ + Provider: "disabled", + MetricsLogInterval: time.Second, + }, + } + logger := testutil.CreateLogger(t, 1) + + metrics := assembler.NewMetrics(cfg, &node_ledger.AssemblerLedgerMetrics{}, logger) + vars.prefetcher = assembler.NewPrefetcher( vars.shards, vars.parties, vars.prefetchIndexMock, vars.batchFetcherMock, - testutil.CreateLogger(t, 1), + metrics, + logger, ) return vars }