add prometheus metric support (#1432)

Fixes #1393 

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

- **New Features**
- Integrated metrics tracking for block synchronization, block
management, P2P networking, and block execution.
	- Added block size calculation to provide insights into block data.

- **Enhancements**
	- Improved node configuration with new instrumentation options.
- Enhanced existing functions to utilize metrics, improving
observability.

- **Bug Fixes**
	- Adjusted test cases to align with the updated metrics integration.

- **Documentation**
- Updated internal documentation to reflect metrics integration in
various components.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Co-authored-by: Ganesha Upadhyaya <gupadhyaya@Ganeshas-MacBook-Pro-2.local>

Author: gupadhyaya

block/block_sync.go

@@ -57,7 +57,11 @@ func NewBlockSyncService(ctx context.Context, store ds.TxnDatastore, conf config
 	if !ok {
 		return nil, errors.New("failed to access the datastore")
 	}
-	ss, err := goheaderstore.NewStore[*types.Block](storeBatch, goheaderstore.WithStorePrefix("blockSync"))
+	ss, err := goheaderstore.NewStore[*types.Block](
+		storeBatch,
+		goheaderstore.WithStorePrefix("blockSync"),
+		goheaderstore.WithMetrics(),
+	)
 	if err != nil {
 		return nil, fmt.Errorf("failed to initialize the block store: %w", err)
 	}
@@ -129,6 +133,7 @@ func (bSyncService *BlockSyncService) Start() error {
 		ps,
 		pubsub.DefaultMsgIdFn,
 		goheaderp2p.WithSubscriberNetworkID(chainIDBlock),
+		goheaderp2p.WithSubscriberMetrics(),
 	)
 	if err != nil {
 		return err
@@ -229,6 +234,7 @@ func newBlockP2PServer(
 ) (*goheaderp2p.ExchangeServer[*types.Block], error) {
 	opts = append(opts,
 		goheaderp2p.WithNetworkID[goheaderp2p.ServerParameters](network),
+		goheaderp2p.WithMetrics[goheaderp2p.ServerParameters](),
 	)
 	return goheaderp2p.NewExchangeServer[*types.Block](host, store, opts...)
 }
@@ -243,6 +249,7 @@ func newBlockP2PExchange(
 	opts = append(opts,
 		goheaderp2p.WithNetworkID[goheaderp2p.ClientParameters](network),
 		goheaderp2p.WithChainID(chainID),
+		goheaderp2p.WithMetrics[goheaderp2p.ClientParameters](),
 	)
 	return goheaderp2p.NewExchange[*types.Block](host, peers, conngater, opts...)
 }
@@ -254,6 +261,9 @@ func newBlockSyncer(
 	sub header.Subscriber[*types.Block],
 	opts []goheadersync.Option,
 ) (*goheadersync.Syncer[*types.Block], error) {
+	opts = append(opts,
+		goheadersync.WithMetrics(),
+	)
 	return goheadersync.NewSyncer[*types.Block](ex, store, sub, opts...)
 }
 

block/header_sync.go

@@ -58,7 +58,11 @@ func NewHeaderSyncService(ctx context.Context, store ds.TxnDatastore, conf confi
 	if !ok {
 		return nil, errors.New("failed to access the datastore")
 	}
-	ss, err := goheaderstore.NewStore[*types.SignedHeader](storeBatch, goheaderstore.WithStorePrefix("headerSync"))
+	ss, err := goheaderstore.NewStore[*types.SignedHeader](
+		storeBatch,
+		goheaderstore.WithStorePrefix("headerSync"),
+		goheaderstore.WithMetrics(),
+	)
 	if err != nil {
 		return nil, fmt.Errorf("failed to initialize the header store: %w", err)
 	}
@@ -127,6 +131,7 @@ func (hSyncService *HeaderSyncService) Start() error {
 		ps,
 		pubsub.DefaultMsgIdFn,
 		goheaderp2p.WithSubscriberNetworkID(hSyncService.genesis.ChainID),
+		goheaderp2p.WithSubscriberMetrics(),
 	)
 	if err != nil {
 		return err
@@ -226,6 +231,7 @@ func newP2PServer(
 ) (*goheaderp2p.ExchangeServer[*types.SignedHeader], error) {
 	opts = append(opts,
 		goheaderp2p.WithNetworkID[goheaderp2p.ServerParameters](network),
+		goheaderp2p.WithMetrics[goheaderp2p.ServerParameters](),
 	)
 	return goheaderp2p.NewExchangeServer[*types.SignedHeader](host, store, opts...)
 }
@@ -240,6 +246,7 @@ func newP2PExchange(
 	opts = append(opts,
 		goheaderp2p.WithNetworkID[goheaderp2p.ClientParameters](network),
 		goheaderp2p.WithChainID(chainID),
+		goheaderp2p.WithMetrics[goheaderp2p.ClientParameters](),
 	)
 
 	return goheaderp2p.NewExchange[*types.SignedHeader](host, peers, conngater, opts...)
@@ -252,6 +259,9 @@ func newSyncer(
 	sub header.Subscriber[*types.SignedHeader],
 	opts []goheadersync.Option,
 ) (*goheadersync.Syncer[*types.SignedHeader], error) {
+	opts = append(opts,
+		goheadersync.WithMetrics(),
+	)
 	return goheadersync.NewSyncer[*types.SignedHeader](ex, store, sub, opts...)
 }
 

block/manager.go

@@ -97,6 +97,9 @@ type Manager struct {
 	doneBuildingBlock chan struct{}
 
 	pendingBlocks *PendingBlocks
+
+	// for reporting metrics
+	metrics *Metrics
 }
 
 // getInitialState tries to load lastState from Store, and if it's not available it reads GenesisDoc.
@@ -120,6 +123,8 @@ func NewManager(
 	eventBus *cmtypes.EventBus,
 	logger log.Logger,
 	blockStore *goheaderstore.Store[*types.Block],
+	seqMetrics *Metrics,
+	execMetrics *state.Metrics,
 ) (*Manager, error) {
 	s, err := getInitialState(store, genesis)
 	if err != nil {
@@ -148,7 +153,7 @@ func NewManager(
 		return nil, err
 	}
 
-	exec := state.NewBlockExecutor(proposerAddress, genesis.ChainID, mempool, proxyApp, eventBus, logger)
+	exec := state.NewBlockExecutor(proposerAddress, genesis.ChainID, mempool, proxyApp, eventBus, logger, execMetrics)
 	if s.LastBlockHeight+1 == uint64(genesis.InitialHeight) {
 		res, err := exec.InitChain(genesis)
 		if err != nil {
@@ -192,6 +197,7 @@ func NewManager(
 		doneBuildingBlock: make(chan struct{}),
 		buildingBlock:     false,
 		pendingBlocks:     NewPendingBlocks(),
+		metrics:           seqMetrics,
 	}
 	return agg, nil
 }
@@ -780,6 +786,7 @@ func (m *Manager) publishBlock(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
+	m.recordMetrics(block)
 	// Check for shut down event prior to sending the header and block to
 	// their respective channels. The reason for checking for the shutdown
 	// event separately is due to the inconsistent nature of the select
@@ -801,6 +808,13 @@ func (m *Manager) publishBlock(ctx context.Context) error {
 	return nil
 }
 
+func (m *Manager) recordMetrics(block *types.Block) {
+	m.metrics.NumTxs.Set(float64(len(block.Data.Txs)))
+	m.metrics.TotalTxs.Add(float64(len(block.Data.Txs)))
+	m.metrics.BlockSizeBytes.Set(float64(block.Size()))
+	m.metrics.CommittedHeight.Set(float64(block.Height()))
+}
+
 func (m *Manager) submitBlocksToDA(ctx context.Context) error {
 	submitted := false
 	backoff := initialBackoff
@@ -849,6 +863,7 @@ func (m *Manager) updateState(s types.State) error {
 		return err
 	}
 	m.lastState = s
+	m.metrics.Height.Set(float64(s.LastBlockHeight))
 	return nil
 }
 

block/manager_test.go

@@ -14,6 +14,7 @@ import (
 	goDATest "github.com/rollkit/go-da/test"
 	"github.com/rollkit/rollkit/config"
 	"github.com/rollkit/rollkit/da"
+	"github.com/rollkit/rollkit/state"
 	"github.com/rollkit/rollkit/store"
 	test "github.com/rollkit/rollkit/test/log"
 	"github.com/rollkit/rollkit/types"
@@ -78,7 +79,7 @@ func TestInitialState(t *testing.T) {
 			assert := assert.New(t)
 			logger := test.NewFileLoggerCustom(t, test.TempLogFileName(t, c.name))
 			dalc := &da.DAClient{DA: goDATest.NewDummyDA(), GasPrice: -1, Logger: logger}
-			agg, err := NewManager(key, conf, c.genesis, c.store, nil, nil, dalc, nil, logger, nil)
+			agg, err := NewManager(key, conf, c.genesis, c.store, nil, nil, dalc, nil, logger, nil, NopMetrics(), state.NopMetrics())
 			assert.NoError(err)
 			assert.NotNil(agg)
 			agg.lastStateMtx.RLock()

block/metrics.go

@@ -0,0 +1,82 @@
+package block
+
+import (
+	"github.com/go-kit/kit/metrics"
+	"github.com/go-kit/kit/metrics/discard"
+	"github.com/go-kit/kit/metrics/prometheus"
+	stdprometheus "github.com/prometheus/client_golang/prometheus"
+)
+
+const (
+	// MetricsSubsystem is a subsystem shared by all metrics exposed by this
+	// package.
+	MetricsSubsystem = "sequencer"
+)
+
+// Metrics contains metrics exposed by this package.
+type Metrics struct {
+	// Height of the chain.
+	Height metrics.Gauge
+
+	// Number of transactions.
+	NumTxs metrics.Gauge
+	// Size of the block.
+	BlockSizeBytes metrics.Gauge
+	// Total number of transactions.
+	TotalTxs metrics.Gauge
+	// The latest block height.
+	CommittedHeight metrics.Gauge `metrics_name:"latest_block_height"`
+}
+
+// PrometheusMetrics returns Metrics build using Prometheus client library.
+// Optionally, labels can be provided along with their values ("foo",
+// "fooValue").
+func PrometheusMetrics(namespace string, labelsAndValues ...string) *Metrics {
+	labels := []string{}
+	for i := 0; i < len(labelsAndValues); i += 2 {
+		labels = append(labels, labelsAndValues[i])
+	}
+	return &Metrics{
+		Height: prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
+			Namespace: namespace,
+			Subsystem: MetricsSubsystem,
+			Name:      "height",
+			Help:      "Height of the chain.",
+		}, labels).With(labelsAndValues...),
+		NumTxs: prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
+			Namespace: namespace,
+			Subsystem: MetricsSubsystem,
+			Name:      "num_txs",
+			Help:      "Number of transactions.",
+		}, labels).With(labelsAndValues...),
+		BlockSizeBytes: prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
+			Namespace: namespace,
+			Subsystem: MetricsSubsystem,
+			Name:      "block_size_bytes",
+			Help:      "Size of the block.",
+		}, labels).With(labelsAndValues...),
+		TotalTxs: prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
+			Namespace: namespace,
+			Subsystem: MetricsSubsystem,
+			Name:      "total_txs",
+			Help:      "Total number of transactions.",
+		}, labels).With(labelsAndValues...),
+		CommittedHeight: prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
+			Namespace: namespace,
+			Subsystem: MetricsSubsystem,
+			Name:      "latest_block_height",
+			Help:      "The latest block height.",
+		}, labels).With(labelsAndValues...),
+	}
+}
+
+// NopMetrics returns no-op Metrics.
+func NopMetrics() *Metrics {
+	return &Metrics{
+		Height:          discard.NewGauge(),
+		NumTxs:          discard.NewGauge(),
+		BlockSizeBytes:  discard.NewGauge(),
+		TotalTxs:        discard.NewGauge(),
+		CommittedHeight: discard.NewGauge(),
+	}
+}

config/config.go

@@ -34,8 +34,9 @@ type NodeConfig struct {
 	DAAddress          string `mapstructure:"da_address"`
 	Light              bool   `mapstructure:"light"`
 	HeaderConfig       `mapstructure:",squash"`
-	LazyAggregator     bool    `mapstructure:"lazy_aggregator"`
-	DAGasPrice         float64 `mapstructure:"da_gas_price"`
+	LazyAggregator     bool                         `mapstructure:"lazy_aggregator"`
+	Instrumentation    *cmcfg.InstrumentationConfig `mapstructure:"instrumentation"`
+	DAGasPrice         float64                      `mapstructure:"da_gas_price"`
 }
 
 // HeaderConfig allows node to pass the initial trusted header hash to start the header exchange service
@@ -74,6 +75,9 @@ func GetNodeConfig(nodeConf *NodeConfig, cmConf *cmcfg.Config) {
 			nodeConf.RPC.TLSCertFile = cmConf.RPC.TLSCertFile
 			nodeConf.RPC.TLSKeyFile = cmConf.RPC.TLSKeyFile
 		}
+		if cmConf.Instrumentation != nil {
+			nodeConf.Instrumentation = cmConf.Instrumentation
+		}
 	}
 }
 

config/defaults.go

@@ -2,6 +2,8 @@ package config
 
 import (
 	"time"
+
+	"github.com/cometbft/cometbft/config"
 )
 
 const (
@@ -30,4 +32,5 @@ var DefaultNodeConfig = NodeConfig{
 	HeaderConfig: HeaderConfig{
 		TrustedHash: "",
 	},
+	Instrumentation: config.DefaultInstrumentationConfig(),
 }