changes to review

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

Author: gengur

config/config.go

@@ -346,6 +346,7 @@ func (config *Configuration) ExtractAssemblerConfig() *nodeconfig.AssemblerNodeC
 		Consenter:                 consenterFromMyParty,
 		UseTLS:                    config.LocalConfig.TLSConfig.Enabled,
 		ClientAuthRequired:        config.LocalConfig.TLSConfig.ClientAuthRequired,
+		MetricsLogInterval:        config.LocalConfig.NodeLocalConfig.GeneralConfig.MetricsLogInterval,
 	}
 	return assemblerConfig
 }

config/generate/local_config_gen.go

@@ -57,6 +57,7 @@ type GeneralConfigParams struct {
 	shardID            types.ShardID
 	tlsEnabled         bool
 	clientAuthRequired bool
+	metricsLogInterval time.Duration
 }
 
 // CreateArmaLocalConfig creates a config directory that includes the local config yaml files for all nodes for all parties.
@@ -94,9 +95,9 @@ func createNetworkLocalConfig(network Network, cryptoBaseDir string, configBaseD
 
 	redundantShardID := types.ShardID(0)
 	for _, party := range network.Parties {
-		routerGeneralParams := NewGeneralConfigParams(party.ID, redundantShardID, "router", trimPortFromEndpoint(party.RouterEndpoint), getPortFromEndpoint(party.RouterEndpoint), useTLSRouter, clientAuthRequiredRouter, "info", cryptoBaseDir, configBaseDir)
-		consensusGeneralParams := NewGeneralConfigParams(party.ID, redundantShardID, "consenter", trimPortFromEndpoint(party.ConsenterEndpoint), getPortFromEndpoint(party.ConsenterEndpoint), true, false, "info", cryptoBaseDir, configBaseDir)
-		assemblerGeneralParams := NewGeneralConfigParams(party.ID, redundantShardID, "assembler", trimPortFromEndpoint(party.AssemblerEndpoint), getPortFromEndpoint(party.AssemblerEndpoint), useTLSAssembler, clientAuthRequiredAssembler, "info", cryptoBaseDir, configBaseDir)
+		routerGeneralParams := NewGeneralConfigParams(party.ID, redundantShardID, "router", trimPortFromEndpoint(party.RouterEndpoint), getPortFromEndpoint(party.RouterEndpoint), DefaultMetricsLogInterval, useTLSRouter, clientAuthRequiredRouter, "info", cryptoBaseDir, configBaseDir)
+		consensusGeneralParams := NewGeneralConfigParams(party.ID, redundantShardID, "consenter", trimPortFromEndpoint(party.ConsenterEndpoint), getPortFromEndpoint(party.ConsenterEndpoint), DefaultMetricsLogInterval, true, false, "info", cryptoBaseDir, configBaseDir)
+		assemblerGeneralParams := NewGeneralConfigParams(party.ID, redundantShardID, "assembler", trimPortFromEndpoint(party.AssemblerEndpoint), getPortFromEndpoint(party.AssemblerEndpoint), 10*time.Second, useTLSAssembler, clientAuthRequiredAssembler, "info", cryptoBaseDir, configBaseDir)
 		partyLocalConfig := PartyLocalConfig{
 			RouterLocalConfig:    NewRouterLocalConfig(routerGeneralParams),
 			BatchersLocalConfig:  NewBatchersLocalConfigPerParty(party.ID, party.BatchersEndpoints, cryptoBaseDir, configBaseDir),
@@ -156,7 +157,7 @@ func createPartyConfigFiles(partyLocalConfig PartyLocalConfig, configBaseDir str
 	return nil
 }
 
-func NewGeneralConfigParams(partyID types.PartyID, shardID types.ShardID, role string, listenAddress string, listenPort uint32, tlsEnabled bool, clientAuthRequired bool, logLevel string, cryptoBaseDir string, configBaseDir string) GeneralConfigParams {
+func NewGeneralConfigParams(partyID types.PartyID, shardID types.ShardID, role string, listenAddress string, listenPort uint32, metricsLogInterval time.Duration, tlsEnabled bool, clientAuthRequired bool, logLevel string, cryptoBaseDir string, configBaseDir string) GeneralConfigParams {
 	return GeneralConfigParams{
 		partyID:            partyID,
 		shardID:            shardID,
@@ -168,6 +169,7 @@ func NewGeneralConfigParams(partyID types.PartyID, shardID types.ShardID, role s
 		listenPort:         listenPort,
 		tlsEnabled:         tlsEnabled,
 		clientAuthRequired: clientAuthRequired,
+		metricsLogInterval: metricsLogInterval,
 	}
 }
 
@@ -203,7 +205,7 @@ func NewGeneralConfig(generalConfigParams GeneralConfigParams) *config.GeneralCo
 		BCCSP:                               &factory.FactoryOpts{},
 		LogSpec:                             generalConfigParams.logLevel,
 		ClientSignatureVerificationRequired: DefaultClientSignatureVerificationRequired,
-		MetricsLogInterval:                  DefaultMetricsLogInterval,
+		MetricsLogInterval:                  generalConfigParams.metricsLogInterval,
 	}
 
 	if generalConfigParams.role == "consenter" {
@@ -244,7 +246,7 @@ func createBatcherLocalConfig(batcherGeneralParams GeneralConfigParams) *config.
 func NewBatchersLocalConfigPerParty(partyID types.PartyID, batcherEndpoints []string, cryptoBaseDir string, configBaseDir string) []*config.NodeLocalConfig {
 	var batchers []*config.NodeLocalConfig
 	for i, batcherEndpoint := range batcherEndpoints {
-		batcherGeneralParams := NewGeneralConfigParams(partyID, types.ShardID(uint16(i+1)), "batcher", trimPortFromEndpoint(batcherEndpoint), getPortFromEndpoint(batcherEndpoint), true, false, "info", cryptoBaseDir, configBaseDir)
+		batcherGeneralParams := NewGeneralConfigParams(partyID, types.ShardID(uint16(i+1)), "batcher", trimPortFromEndpoint(batcherEndpoint), getPortFromEndpoint(batcherEndpoint), DefaultMetricsLogInterval, true, false, "info", cryptoBaseDir, configBaseDir)
 		batcher := createBatcherLocalConfig(batcherGeneralParams)
 		batchers = append(batchers, batcher)
 	}

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
 }
 

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