changes to review

gengur · gengur · commit 392db26de728 · 2025-10-30T05:42:37.000-05:00
Signed-off-by: Genady Gurevich &lt;genadyg@il.ibm.com&gt;
diff --git a/config/config.go b/config/config.go
@@ -376,7 +376,6 @@ func (config *Configuration) ExtractAssemblerConfig() *nodeconfig.AssemblerNodeC
 		Consenter:                 consenterFromMyParty,
 		UseTLS:                    config.LocalConfig.TLSConfig.Enabled,
 		ClientAuthRequired:        config.LocalConfig.TLSConfig.ClientAuthRequired,
-		MonitoringListenAddress:   net.JoinHostPort(config.LocalConfig.NodeLocalConfig.GeneralConfig.MonitoringListenAddress, strconv.Itoa(int(config.LocalConfig.NodeLocalConfig.GeneralConfig.MonitoringListenPort))),
 		MetricsLogInterval:        config.LocalConfig.NodeLocalConfig.GeneralConfig.MetricsLogInterval,
 	}
 	return assemblerConfig
diff --git a/node/assembler/assembler.go b/node/assembler/assembler.go
@@ -32,6 +32,7 @@ type Assembler struct {
 	prefetcher   PrefetcherController
 	baReplicator delivery.ConsensusBringer
 	netStopper   NetStopper
+	metrics      *AssemblerMetrics
 }
 
 func (a *Assembler) Broadcast(server orderer.AtomicBroadcast_BroadcastServer) error {
@@ -43,8 +44,15 @@ func (a *Assembler) Deliver(server orderer.AtomicBroadcast_DeliverServer) error
 }
 
 func (a *Assembler) GetTxCount() uint64 {
-	// TODO do this in a cleaner fashion
-	return a.assembler.Ledger.(*node_ledger.AssemblerLedger).GetTxCount()
+	return a.assembler.Ledger.GetTxCount()
+}
+
+func (a *Assembler) GetBlocksCount() uint64 {
+	return a.assembler.Ledger.GetBlocksCount()
+}
+
+func (a *Assembler) GetBlocksSize() uint64 {
+	return a.assembler.Ledger.GetBlocksSize()
 }
 
 func (a *Assembler) Stop() {
@@ -53,6 +61,7 @@ func (a *Assembler) Stop() {
 	a.assembler.Index.Stop()
 	a.baReplicator.Stop()
 	a.assembler.WaitTermination()
+	a.metrics.Stop()
 	a.assembler.Ledger.Close()
 }
 
@@ -127,6 +136,9 @@ func NewDefaultAssembler(
 
 	assembler.assembler.Run()
 
+	assembler.metrics = NewAssemblerMetrics(assembler, al, index, shardIds, partyIds, logger, config.MetricsLogInterval)
+	assembler.metrics.Start()
+
 	return assembler
 }
 
diff --git a/node/assembler/assembler_metrics.go b/node/assembler/assembler_metrics.go
@@ -0,0 +1,108 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package assembler
+
+import (
+	"fmt"
+	"strings"
+	"sync"
+	"time"
+
+	arma_types "github.com/hyperledger/fabric-x-orderer/common/types"
+	node_ledger "github.com/hyperledger/fabric-x-orderer/node/ledger"
+)
+
+type AssemblerMetrics struct {
+	assembler *Assembler
+	al        node_ledger.AssemblerLedgerReaderWriter
+	index     PrefetchIndexer
+	shardIds  []arma_types.ShardID
+	partyIds  []arma_types.PartyID
+	logger    arma_types.Logger
+	interval  time.Duration
+	stopChan  chan struct{}
+	stopOnce  sync.Once
+}
+
+func NewAssemblerMetrics(assembler *Assembler, al node_ledger.AssemblerLedgerReaderWriter,
+	index PrefetchIndexer, shardIds []arma_types.ShardID, partyIds []arma_types.PartyID, logger arma_types.Logger, interval time.Duration,
+) *AssemblerMetrics {
+	return &AssemblerMetrics{
+		assembler: assembler,
+		al:        al,
+		index:     index,
+		shardIds:  shardIds,
+		partyIds:  partyIds,
+		interval:  interval,
+		logger:    logger,
+		stopChan:  make(chan struct{}),
+	}
+}
+
+func (m *AssemblerMetrics) Start() {
+	if m.interval > 0 {
+		go m.trackMetrics()
+	}
+}
+
+func (m *AssemblerMetrics) Stop() {
+	m.stopOnce.Do(func() {
+		close(m.stopChan)
+		txCommitted := m.assembler.GetTxCount()
+		blocksCommitted := m.assembler.GetBlocksCount()
+		blocksSizeCommitted := m.assembler.GetBlocksSize()
+
+		totalPbSz := 0
+		var sb strings.Builder
+
+		for _, shardId := range m.shardIds {
+			for _, partyId := range m.partyIds {
+				pbsz := m.index.PrefetchBufferSize(ShardPrimary{Shard: shardId, Primary: partyId})
+				sb.WriteString(fmt.Sprintf("<Sh: %d, Pr:%d>:%d; ", shardId, partyId, pbsz))
+				totalPbSz += pbsz
+			}
+		}
+
+		m.logger.Infof("ASSEMBLER_METRICS ledger height: %d, total committed transactions: %d, total blocks %d bytes, total size:%d, prefetch buffer size: %d(%s) bytes",
+			m.al.LedgerReader().Height(), txCommitted, blocksCommitted, blocksSizeCommitted, totalPbSz, strings.TrimRight(sb.String(), "; "))
+	})
+}
+
+func (m *AssemblerMetrics) trackMetrics() {
+	lastTxCommitted, lastBlocksCommitted := uint64(0), uint64(0)
+	sec := m.interval.Seconds()
+	t := time.NewTicker(m.interval)
+	defer t.Stop()
+
+	for {
+		select {
+		case <-t.C:
+			totalPbSz := 0
+			var sb strings.Builder
+
+			for _, shardId := range m.shardIds {
+				for _, partyId := range m.partyIds {
+					pbsz := m.index.PrefetchBufferSize(ShardPrimary{Shard: shardId, Primary: partyId})
+					sb.WriteString(fmt.Sprintf("<Sh: %d, Pr:%d>:%d; ", shardId, partyId, pbsz))
+					totalPbSz += pbsz
+				}
+			}
+
+			txCommitted := m.assembler.GetTxCount()
+			blocksCommitted := m.assembler.GetBlocksCount()
+			blocksSizeCommitted := m.assembler.GetBlocksSize()
+
+			m.logger.Infof("ASSEMBLER_METRICS ledger height: %d, total committed transactions: %d, new transactions in the last %.2f seconds: %d, total blocks %d bytes, new blocks in the last %.2f seconds: %d, total size:%d, prefetch buffer size: %d(%s) bytes",
+				m.al.LedgerReader().Height(), txCommitted, sec, (txCommitted - lastTxCommitted), blocksCommitted,
+				sec, (blocksCommitted - lastBlocksCommitted), blocksSizeCommitted, totalPbSz, strings.TrimRight(sb.String(), "; "))
+
+			lastTxCommitted, lastBlocksCommitted = txCommitted, blocksCommitted
+		case <-m.stopChan:
+			return
+		}
+	}
+}
diff --git a/node/assembler/assembler_role.go b/node/assembler/assembler_role.go
@@ -26,6 +26,9 @@ type AssemblerIndex interface {
 }
 
 type AssemblerLedgerWriter interface {
+	GetTxCount() uint64
+	GetBlocksCount() uint64
+	GetBlocksSize() uint64
 	Append(batch types.Batch, orderingInfo types.OrderingInfo)
 	Close()
 }
diff --git a/node/assembler/assembler_role_test.go b/node/assembler/assembler_role_test.go
@@ -78,6 +78,24 @@ func (n naiveAssemblerLedger) Close() {
 	close(n)
 }
 
+// GetBlocksCount implements the assembler.AssemblerLedgerWriter interface.
+func (n naiveAssemblerLedger) GetBlocksCount() uint64 {
+	// For testing, return 0 or a mock value as needed.
+	return 0
+}
+
+// GetTxCount implements the assembler.AssemblerLedgerWriter interface.
+func (n naiveAssemblerLedger) GetTxCount() uint64 {
+	// For testing, return 0 or a mock value as needed.
+	return 0
+}
+
+// GetBlocksSize implements the assembler.AssemblerLedgerWriter interface.
+func (n naiveAssemblerLedger) GetBlocksSize() uint64 {
+	// For testing, return 0 or a mock value as needed.
+	return 0
+}
+
 func TestAssembler(t *testing.T) {
 	shardCount := 4
 	batchNum := 20
diff --git a/node/assembler/batch_cache.go b/node/assembler/batch_cache.go
@@ -7,14 +7,16 @@ SPDX-License-Identifier: Apache-2.0
 package assembler
 
 import (
+	"sync/atomic"
+
 	"github.com/hyperledger/fabric-x-orderer/common/types"
 )
 
 type BatchCache struct {
 	tag              string
 	shardBatchMapper *BatchMapper[types.BatchID, types.Batch]
 	partition        ShardPrimary
-	sizeBytes        int
+	sizeBytes        uint64
 }
 
 //go:generate counterfeiter -o ./mocks/batch_cache_factory.go . BatchCacheFactory
@@ -47,7 +49,7 @@ func (bc *BatchCache) Pop(batchId types.BatchID) (types.Batch, error) {
 	if err != nil {
 		return nil, err
 	}
-	bc.sizeBytes -= batchSizeBytes(batch)
+	atomic.AddUint64(&bc.sizeBytes, ^(uint64(batchSizeBytes(batch)) - 1))
 	return batch, nil
 }
 
@@ -57,7 +59,7 @@ func (bc *BatchCache) Put(batch types.Batch) error {
 		return ErrBatchAlreadyExists
 	}
 	if inserted {
-		bc.sizeBytes += batchSizeBytes(batch)
+		atomic.AddUint64(&bc.sizeBytes, uint64(batchSizeBytes(batch)))
 	}
 	return nil
 }
@@ -71,5 +73,5 @@ func (bc *BatchCache) Get(batchId types.BatchID) (types.Batch, error) {
 }
 
 func (bc *BatchCache) SizeBytes() int {
-	return bc.sizeBytes
+	return int(atomic.LoadUint64(&bc.sizeBytes))
 }
diff --git a/node/assembler/mocks/partition_prefetch_index.go b/node/assembler/mocks/partition_prefetch_index.go
diff --git a/node/assembler/mocks/prefetch_index.go b/node/assembler/mocks/prefetch_index.go
diff --git a/node/assembler/partition_prefetch_index.go b/node/assembler/partition_prefetch_index.go
@@ -56,6 +56,7 @@ type PartitionPrefetchIndexer interface {
 	Put(batch types.Batch) error
 	PutForce(batch types.Batch) error
 	Stop()
+	PrefetchBufferSize() int
 }
 
 //go:generate counterfeiter -o ./mocks/partition_prefetch_index_factory.go . PartitionPrefetchIndexerFactory
@@ -192,6 +193,10 @@ func NewPartitionPrefetchIndex(partition ShardPrimary, logger types.Logger, defa
 	return pi
 }
 
+func (pi *PartitionPrefetchIndex) PrefetchBufferSize() int {
+	return pi.cache.SizeBytes() + pi.forcedPutCache.SizeBytes()
+}
+
 func (pi *PartitionPrefetchIndex) getName() string {
 	return fmt.Sprintf("partition <Sh: %d, Pr: %d>", pi.partition.Shard, pi.partition.Primary)
 }
diff --git a/node/assembler/prefetch_index.go b/node/assembler/prefetch_index.go
@@ -18,6 +18,7 @@ type PrefetchIndexer interface {
 	Put(batch types.Batch) error
 	PutForce(batch types.Batch) error
 	Requests() <-chan types.BatchID
+	PrefetchBufferSize(shardPrimary ShardPrimary) int
 	Stop()
 }
 
@@ -106,6 +107,11 @@ func NewPrefetchIndex(
 	return pi
 }
 
+func (pi *PrefetchIndex) PrefetchBufferSize(shardPrimary ShardPrimary) int {
+	partitionIndex := pi.partitionToIndex[shardPrimary]
+	return partitionIndex.PrefetchBufferSize()
+}
+
 func (pi *PrefetchIndex) PopOrWait(batchId types.BatchID) (types.Batch, error) {
 	t1 := time.Now()
 	defer func() {
diff --git a/node/ledger/assembler_ledger.go b/node/ledger/assembler_ledger.go
@@ -31,6 +31,8 @@ type (
 //go:generate counterfeiter -o ./mocks/assembler_ledger.go . AssemblerLedgerReaderWriter
 type AssemblerLedgerReaderWriter interface {
 	GetTxCount() uint64
+	GetBlocksCount() uint64
+	GetBlocksSize() uint64
 	Append(batch types.Batch, orderingInfo types.OrderingInfo)
 	AppendConfig(configBlock *common.Block, decisionNum types.DecisionNum)
 	LastOrderingInfo() (*state.OrderingInformation, error)
@@ -54,6 +56,8 @@ type AssemblerLedger struct {
 	Logger               types.Logger
 	Ledger               blockledger.ReadWriter
 	transactionCount     uint64
+	blocksCount          uint64
+	blocksSize           uint64
 	blockStorageProvider *blkstorage.BlockStoreProvider
 	blockStore           *blkstorage.BlockStore
 	cancellationContext  context.Context
@@ -99,7 +103,7 @@ func NewAssemblerLedger(logger types.Logger, ledgerPath string) (*AssemblerLedge
 		cancellationContext:  ctx,
 		cancelContextFunc:    cancel,
 	}
-	go al.trackThroughput()
+
 	return al, nil
 }
 
@@ -109,29 +113,25 @@ func (l *AssemblerLedger) Close() {
 	l.blockStorageProvider.Close()
 }
 
-func (l *AssemblerLedger) trackThroughput() {
-	firstProbe := true
-	lastTxCount := uint64(0)
-	for {
-		txCount := atomic.LoadUint64(&l.transactionCount)
-		if !firstProbe {
-			l.Logger.Infof("Tx Count: %d, Commit throughput: %.2f", txCount, float64(txCount-lastTxCount)/10.0)
-		}
-		lastTxCount = txCount
-		firstProbe = false
-		select {
-		case <-time.After(time.Second * 10):
-		case <-l.cancellationContext.Done():
-			return
-		}
-	}
-}
-
 func (l *AssemblerLedger) GetTxCount() uint64 {
 	c := atomic.LoadUint64(&l.transactionCount)
 	return c
 }
 
+func (l *AssemblerLedger) GetBlocksCount() uint64 {
+	c := atomic.LoadUint64(&l.blocksCount)
+	return c
+}
+
+func (l *AssemblerLedger) GetBlocksSize() uint64 {
+	c := atomic.LoadUint64(&l.blocksSize)
+	return c
+}
+
+func blockSize(block *common.Block) uint64 {
+	return uint64(len(protoutil.MarshalOrPanic(block)))
+}
+
 func (l *AssemblerLedger) Append(batch types.Batch, orderingInfo types.OrderingInfo) {
 	ordInfo := orderingInfo.(*state.OrderingInformation)
 	t1 := time.Now()
@@ -194,6 +194,8 @@ func (l *AssemblerLedger) Append(batch types.Batch, orderingInfo types.OrderingI
 	}
 
 	atomic.AddUint64(&l.transactionCount, uint64(len(batch.Requests())))
+	atomic.AddUint64(&l.blocksCount, uint64(1))
+	atomic.AddUint64(&l.blocksSize, blockSize(block))
 }
 
 func (l *AssemblerLedger) AppendConfig(configBlock *common.Block, decisionNum types.DecisionNum) {
@@ -207,6 +209,8 @@ func (l *AssemblerLedger) AppendConfig(configBlock *common.Block, decisionNum ty
 		l.Logger.Panicf("attempting to AppendConfig a block which is not a config block: %d", configBlock.GetHeader().GetNumber())
 	}
 
+	atomic.AddUint64(&l.blocksCount, uint64(1))
+	atomic.AddUint64(&l.blocksSize, blockSize(configBlock))
 	transactionCount := atomic.AddUint64(&l.transactionCount, 1) // len(configBlock.GetData().GetData()) = should always be a single TX
 	batchID := types.NewSimpleBatch(types.ShardIDConsensus, 0, 0, nil, 0)
 	ordInfo := &state.OrderingInformation{
diff --git a/node/ledger/mocks/assembler_ledger.go b/node/ledger/mocks/assembler_ledger.go

Original file line number	Diff line number	Diff line change
`@@ -376,7 +376,6 @@ func (config Configuration) ExtractAssemblerConfig() nodeconfig.AssemblerNodeC`
`376`	`376`	`Consenter: consenterFromMyParty,`
`377`	`377`	`UseTLS: config.LocalConfig.TLSConfig.Enabled,`
`378`	`378`	`ClientAuthRequired: config.LocalConfig.TLSConfig.ClientAuthRequired,`
`379`		`- MonitoringListenAddress: net.JoinHostPort(config.LocalConfig.NodeLocalConfig.GeneralConfig.MonitoringListenAddress, strconv.Itoa(int(config.LocalConfig.NodeLocalConfig.GeneralConfig.MonitoringListenPort))),`
`380`	`379`	`MetricsLogInterval: config.LocalConfig.NodeLocalConfig.GeneralConfig.MetricsLogInterval,`
`381`	`380`	`}`
`382`	`381`	`return assemblerConfig`
Original file line number	Diff line number	Diff line change
`@@ -26,6 +26,9 @@ type AssemblerIndex interface {`
`26`	`26`	`}`
`27`	`27`
`28`	`28`	`type AssemblerLedgerWriter interface {`
	`29`	`+ GetTxCount() uint64`
	`30`	`+ GetBlocksCount() uint64`
	`31`	`+ GetBlocksSize() uint64`
`29`	`32`	`Append(batch types.Batch, orderingInfo types.OrderingInfo)`
`30`	`33`	`Close()`
`31`	`34`	`}`
Original file line number	Diff line number	Diff line change
`@@ -7,14 +7,16 @@ SPDX-License-Identifier: Apache-2.0`
`7`	`7`	`package assembler`
`8`	`8`
`9`	`9`	`import (`
	`10`	`+ "sync/atomic"`
	`11`	`+`
`10`	`12`	`"github.com/hyperledger/fabric-x-orderer/common/types"`
`11`	`13`	`)`
`12`	`14`
`13`	`15`	`type BatchCache struct {`
`14`	`16`	`tag string`
`15`	`17`	`shardBatchMapper *BatchMapper[types.BatchID, types.Batch]`
`16`	`18`	`partition ShardPrimary`
`17`		`- sizeBytes int`
	`19`	`+ sizeBytes uint64`
`18`	`20`	`}`
`19`	`21`
`20`	`22`	`//go:generate counterfeiter -o ./mocks/batch_cache_factory.go . BatchCacheFactory`
`@@ -47,7 +49,7 @@ func (bc *BatchCache) Pop(batchId types.BatchID) (types.Batch, error) {`
`47`	`49`	`if err != nil {`
`48`	`50`	`return nil, err`
`49`	`51`	`}`
`50`		`- bc.sizeBytes -= batchSizeBytes(batch)`
	`52`	`+ atomic.AddUint64(&bc.sizeBytes, ^(uint64(batchSizeBytes(batch)) - 1))`
`51`	`53`	`return batch, nil`
`52`	`54`	`}`
`53`	`55`
`@@ -57,7 +59,7 @@ func (bc *BatchCache) Put(batch types.Batch) error {`
`57`	`59`	`return ErrBatchAlreadyExists`
`58`	`60`	`}`
`59`	`61`	`if inserted {`
`60`		`- bc.sizeBytes += batchSizeBytes(batch)`
	`62`	`+ atomic.AddUint64(&bc.sizeBytes, uint64(batchSizeBytes(batch)))`
`61`	`63`	`}`
`62`	`64`	`return nil`
`63`	`65`	`}`
`@@ -71,5 +73,5 @@ func (bc *BatchCache) Get(batchId types.BatchID) (types.Batch, error) {`
`71`	`73`	`}`
`72`	`74`
`73`	`75`	`func (bc *BatchCache) SizeBytes() int {`
`74`		`- return bc.sizeBytes`
	`76`	`+ return int(atomic.LoadUint64(&bc.sizeBytes))`
`75`	`77`	`}`
Original file line number	Diff line number	Diff line change
`@@ -56,6 +56,7 @@ type PartitionPrefetchIndexer interface {`
`56`	`56`	`Put(batch types.Batch) error`
`57`	`57`	`PutForce(batch types.Batch) error`
`58`	`58`	`Stop()`
	`59`	`+ PrefetchBufferSize() int`
`59`	`60`	`}`
`60`	`61`
`61`	`62`	`//go:generate counterfeiter -o ./mocks/partition_prefetch_index_factory.go . PartitionPrefetchIndexerFactory`
`@@ -192,6 +193,10 @@ func NewPartitionPrefetchIndex(partition ShardPrimary, logger types.Logger, defa`
`192`	`193`	`return pi`
`193`	`194`	`}`
`194`	`195`
	`196`	`+func (pi *PartitionPrefetchIndex) PrefetchBufferSize() int {`
	`197`	`+ return pi.cache.SizeBytes() + pi.forcedPutCache.SizeBytes()`
	`198`	`+}`
	`199`	`+`
`195`	`200`	`func (pi *PartitionPrefetchIndex) getName() string {`
`196`	`201`	`return fmt.Sprintf("partition <Sh: %d, Pr: %d>", pi.partition.Shard, pi.partition.Primary)`
`197`	`202`	`}`