changes to review

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

Author: gengur

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,16 +44,17 @@ 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()
 }
 
+// Stop stops the assembler and all its components.
 func (a *Assembler) Stop() {
 	a.netStopper.Stop()
 	a.prefetcher.Stop()
 	a.assembler.Index.Stop()
 	a.baReplicator.Stop()
 	a.assembler.WaitTermination()
+	a.metrics.Stop()
 	a.assembler.Ledger.Close()
 }
 
@@ -126,6 +128,9 @@ func NewDefaultAssembler(
 
 	assembler.assembler.Run()
 
+	assembler.metrics = NewAssemblerMetrics(assembler, al, index, shardIds, partyIds, logger, config.MetricsLogInterval)
+	assembler.metrics.Start()
+
 	return assembler
 }
 

node/assembler/assembler_metrics.go

@@ -0,0 +1,106 @@
+/*
+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
+	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,
+		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.assembler.Ledger.GetBlocksCount()
+		blocksSizeCommitted := m.assembler.assembler.Ledger.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 total committed transactions: %d, total blocks: %d, total size: %d, prefetch buffer size: %d(%s) bytes",
+			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.assembler.Ledger.GetBlocksCount()
+			blocksSizeCommitted := m.assembler.assembler.Ledger.GetBlocksSize()
+
+			m.logger.Infof("ASSEMBLER_METRICS: total committed transactions %d, new transactions in the last %.2f seconds: %d, total blocks: %d, new blocks in the last: %.2f seconds: %d, total size: %d, prefetch buffer size: %d(%s) bytes",
+				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

@@ -24,6 +24,9 @@ type AssemblerIndex interface {
 }
 
 type AssemblerLedgerWriter interface {
+	GetTxCount() uint64
+	GetBlocksCount() uint64
+	GetBlocksSize() uint64
 	Append(batch types.Batch, orderingInfo types.OrderingInfo)
 	AppendConfig(configBlock *common.Block, decisionNum types.DecisionNum)
 	Close()

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
@@ -99,7 +103,7 @@ func NewAssemblerLedger(logger types.Logger, ledgerPath string) (*AssemblerLedge
 		cancellationContext:  ctx,
 		cancelContextFunc:    cancel,
 	}
-	go al.trackThroughput()
+
 	return al, nil
 }
 
@@ -109,27 +113,20 @@ 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 {
+	return atomic.LoadUint64(&l.transactionCount)
 }
 
-func (l *AssemblerLedger) GetTxCount() uint64 {
-	c := atomic.LoadUint64(&l.transactionCount)
-	return c
+func (l *AssemblerLedger) GetBlocksCount() uint64 {
+	return atomic.LoadUint64(&l.blocksCount)
+}
+
+func (l *AssemblerLedger) GetBlocksSize() uint64 {
+	return atomic.LoadUint64(&l.blocksSize)
+}
+
+func blockSize(block *common.Block) uint64 {
+	return uint64(len(protoutil.MarshalOrPanic(block)))
 }
 
 func (l *AssemblerLedger) Append(batch types.Batch, orderingInfo types.OrderingInfo) {
@@ -194,6 +191,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 +206,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{