fixed the assembler_test issue

Signed-off-by: Genady Gurevich <[email protected]>

Author: gengur

node/assembler/assembler.go

@@ -9,6 +9,7 @@ package assembler
 import (
 	"encoding/hex"
 	"fmt"
+	"time"
 
 	"github.com/hyperledger/fabric/protoutil"
 
@@ -32,6 +33,7 @@ type Assembler struct {
 	prefetcher   PrefetcherController
 	baReplicator delivery.ConsensusBringer
 	netStopper   NetStopper
+	metrics      *AssemblerMetrics
 }
 
 func (a *Assembler) Broadcast(server orderer.AtomicBroadcast_BroadcastServer) error {
@@ -43,8 +45,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 +62,7 @@ func (a *Assembler) Stop() {
 	a.assembler.Index.Stop()
 	a.baReplicator.Stop()
 	a.assembler.WaitTermination()
+	a.metrics.Stop()
 	a.assembler.Ledger.Close()
 }
 
@@ -127,6 +137,9 @@ func NewDefaultAssembler(
 
 	assembler.assembler.Run()
 
+	assembler.metrics = NewAssemblerMetrics(assembler, al, index, shardIds, partyIds, logger, 10*time.Second)
+	assembler.metrics.Start()
+
 	return assembler
 }
 

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
+		}
+	}
+}

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()
 }

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

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))
 }

node/assembler/mocks/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)
 }

node/assembler/mocks/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() {

node/assembler/partition_prefetch_index.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
@@ -132,6 +136,20 @@ func (l *AssemblerLedger) GetTxCount() uint64 {
 	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()
@@ -198,6 +216,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) {
@@ -211,6 +231,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{