Add prometheus client & seq metrics & Block/Batch Info

cmd/cdk-erigon/run_xlayer.go

@@ -1,15 +1,58 @@
 package main
 
 import (
+	"fmt"
 	"github.com/ledgerwatch/erigon/eth/ethconfig"
 	"github.com/ledgerwatch/erigon/node/nodecfg"
 	"github.com/ledgerwatch/erigon/zk/apollo"
+	"github.com/ledgerwatch/erigon/zk/metrics"
 	"github.com/ledgerwatch/log/v3"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"net"
+	"net/http"
+	"time"
+)
+
+const (
+	MetricsEndpoint = "/metrics"
 )
 
 func initRunForXLayer(ethCfg *ethconfig.Config, nodeCfg *nodecfg.Config) {
 	apolloClient := apollo.NewClient(ethCfg, nodeCfg)
 	if apolloClient.LoadConfig() {
 		log.Info("apollo config loaded")
 	}
+
+	// Start Metrics Server
+	if ethCfg.Zk.XLayer.Metrics.Enabled {
+		metrics.XLayerMetricsInit()
+		go startMetricsHttpServer(ethCfg.Zk.XLayer.Metrics)
+	}
+}
+
+func startMetricsHttpServer(c ethconfig.MetricsConfig) {
+	const ten = 10
+	mux := http.NewServeMux()
+	address := fmt.Sprintf("%s:%d", c.Host, c.Port)
+	lis, err := net.Listen("tcp", address)
+	if err != nil {
+		log.Error("failed to create tcp listener for metrics", "err", err)
+		return
+	}
+	mux.Handle(MetricsEndpoint, promhttp.Handler())
+
+	metricsServer := &http.Server{
+		Handler:           mux,
+		ReadHeaderTimeout: ten * time.Second,
+		ReadTimeout:       ten * time.Second,
+	}
+	log.Info("metrics server listening on port", c.Port)
+	if err := metricsServer.Serve(lis); err != nil {
+		if err == http.ErrServerClosed {
+			log.Warn("http server for metrics stopped")
+			return
+		}
+		log.Error("closed http connection for metrics server", "err", err)
+		return
+	}
 }

cmd/rpcdaemon/commands/send_transaction_zk.go

@@ -11,6 +11,7 @@ import (
 	"github.com/ledgerwatch/erigon/zk/sequencer"
 	"github.com/ledgerwatch/erigon/zk/zkchainconfig"
 	"github.com/ledgerwatch/erigon/zkevm/jsonrpc/client"
+	"github.com/pkg/errors"
 )
 
 func (api *APIImpl) isPoolManagerAddressSet() bool {
@@ -24,7 +25,7 @@ func (api *APIImpl) isZkNonSequencer(chainId *big.Int) bool {
 func (api *APIImpl) sendTxZk(rpcUrl string, encodedTx hexutility.Bytes, chainId uint64) (common.Hash, error) {
 	res, err := client.JSONRPCCall(rpcUrl, "eth_sendRawTransaction", encodedTx)
 	if err != nil {
-		return common.Hash{}, err
+		return common.Hash{}, errors.Wrap(err, "failed to send transaction with zk rpc call")
 	}
 
 	if res.Error != nil {

cmd/utils/flags_xlayer.go

@@ -163,6 +163,22 @@ var (
 		Usage: "Full batch sleep duration is the time the sequencer sleeps between each full batch iteration.",
 		Value: 0 * time.Second,
 	}
+	// X Layer Prometheus Metrics
+	XLMetricsHostFlag = cli.StringFlag{
+		Name:  "zkevm.metrics-host",
+		Usage: "prometheus metrics host",
+		Value: "",
+	}
+	XLMetricsPortFlag = cli.StringFlag{
+		Name:  "zkevm.metrics-port",
+		Usage: "prometheus metrics port",
+		Value: "",
+	}
+	XLMetricsEnabledFlag = cli.StringFlag{
+		Name:  "zkevm.metrics-enabled",
+		Usage: "prometheus metrics enabled",
+		Value: "",
+	}
 )
 
 func setGPOXLayer(ctx *cli.Context, cfg *gaspricecfg.Config) {

eth/ethconfig/config_xlayer.go

@@ -7,6 +7,7 @@ type XLayerConfig struct {
 	Apollo        ApolloClientConfig
 	Nacos         NacosConfig
 	EnableInnerTx bool
+	Metrics       MetricsConfig
 	// Sequencer
 	SequencerBatchSleepDuration time.Duration
 }
@@ -28,3 +29,13 @@ type ApolloClientConfig struct {
 	AppID         string
 	NamespaceName string
 }
+
+// MetricsConfig is the config for prometheus metrics
+type MetricsConfig struct {
+	// Host is the address to bind the metrics server
+	Host string
+	// Port is the port to bind the metrics server
+	Port int
+	// Enabled is the flag to enable/disable the metrics server
+	Enabled bool
+}

test/config/test.erigon.rpc.config.yaml

@@ -47,6 +47,21 @@ http.port: 8545
 http.vhosts: any
 http.corsdomain: any
 ws: true
+
+zkevm.apollo-enabled: false
+zkevm.apollo-ip-addr: ""
+zkevm.apollo-app-id: ""
+zkevm.apollo-namespace-name: ""
+
+zkevm.nacos-urls: ""
+zkevm.nacos-namespace-id: ""
+zkevm.nacos-application-name: ""
+zkevm.nacos-external-listen-addr: ""
+
+zkevm.metrics-host: "0.0.0.0"
+zkevm.metrics-port: "9095"
+zkevm.metrics-enabled: true
+
 db.read.concurrency: 20000
 txpool.globalslots: 100000
 networkid: 195
@@ -55,4 +70,4 @@ zkevm.pool-manager-url: http://erigon-pool-manager:8545
 
 http.methodratelimit: "{\"methods\":[\"eth_syncing\"],\"count\":10,\"bucket\":1}"
 #http.apikeys: |
-#  {"project":"project1","key":"944cd2a6939eb23053289d9b91d6c498","timeout":"2033-12-12","methods":["eth_syncing"],"count":5,"bucket":1}
+#  {"project":"project1","key":"944cd2a6939eb23053289d9b91d6c498","timeout":"2033-12-12","methods":["eth_syncing"],"count":5,"bucket":1}

test/config/test.erigon.seq.config.yaml

@@ -64,6 +64,10 @@ zkevm.nacos-namespace-id: ""
 zkevm.nacos-application-name: ""
 zkevm.nacos-external-listen-addr: ""
 
+zkevm.metrics-host: "0.0.0.0"
+zkevm.metrics-port: "9095"
+zkevm.metrics-enabled: true
+
 db.read.concurrency: 20000
 txpool.globalslots: 100000
 txpool.globalbasefeeslots: 100000

test/docker-compose.yml

@@ -116,6 +116,7 @@ services:
     ports:
       - 8123:8545
       - 6900:6900
+      - 9092:9095
     volumes:
       - ./config/test.erigon.seq.config.yaml:/usr/src/app/config.yaml
       - ./config/dynamic-mynetwork-allocs.json:/usr/src/app/dynamic-mynetwork-allocs.json
@@ -132,6 +133,7 @@ services:
     ports:
       - 8124:8545
       - 6901:6900
+      - 9091:9095
     volumes:
       - ./config/test.erigon.rpc.config.yaml:/usr/src/app/config.yaml
       - ./config/dynamic-mynetwork-allocs.json:/usr/src/app/dynamic-mynetwork-allocs.json

turbo/cli/default_flags.go

@@ -258,4 +258,7 @@ var DefaultFlags = []cli.Flag{
 	&utils.TxPoolWhiteList,
 	&utils.TxPoolBlockedList,
 	&utils.SequencerBatchSleepDuration,
+	&utils.XLMetricsEnabledFlag,
+	&utils.XLMetricsHostFlag,
+	&utils.XLMetricsPortFlag,
 }

turbo/cli/flags_xlayer.go

@@ -22,5 +22,10 @@ func ApplyFlagsForXLayerConfig(ctx *cli.Context, cfg *ethconfig.Config) {
 		},
 		EnableInnerTx:               ctx.Bool(utils.AllowInternalTransactions.Name),
 		SequencerBatchSleepDuration: ctx.Duration(utils.SequencerBatchSleepDuration.Name),
+		Metrics: ethconfig.MetricsConfig{
+			Enabled: ctx.Bool(utils.XLMetricsEnabledFlag.Name),
+			Host:    ctx.String(utils.XLMetricsHostFlag.Name),
+			Port:    ctx.Int(utils.XLMetricsPortFlag.Name),
+		},
 	}
 }

zk/datastream/server/datastream_populate.go

@@ -34,10 +34,6 @@ func (srv *DataStreamServer) WriteWholeBatchToStream(
 	batchNum uint64,
 ) error {
 	var err error
-	if err = srv.stream.StartAtomicOp(); err != nil {
-		return err
-	}
-	defer srv.stream.RollbackAtomicOp()
 
 	blocksForBatch, err := reader.GetL2BlockNosByBatch(batchNum)
 	if err != nil {
@@ -58,6 +54,11 @@ func (srv *DataStreamServer) WriteWholeBatchToStream(
 		return err
 	}
 
+	if err = srv.stream.StartAtomicOp(); err != nil {
+		return err
+	}
+	defer srv.stream.RollbackAtomicOp()
+
 	blocks := make([]eritypes.Block, 0)
 	txsPerBlock := make(map[uint64][]eritypes.Transaction)
 	for blockNumber := fromBlockNum; blockNumber <= toBlockNum; blockNumber++ {
@@ -116,14 +117,14 @@ func (srv *DataStreamServer) WriteBlocksToStreamConsecutively(
 		return err
 	}
 
-	if err = srv.stream.StartAtomicOp(); err != nil {
+	if err = srv.UnwindIfNecessary(logPrefix, reader, from, latestbatchNum, batchNum); err != nil {
 		return err
 	}
-	defer srv.stream.RollbackAtomicOp()
 
-	if err = srv.UnwindIfNecessary(logPrefix, reader, from, latestbatchNum, batchNum); err != nil {
+	if err = srv.stream.StartAtomicOp(); err != nil {
 		return err
 	}
+	defer srv.stream.RollbackAtomicOp()
 
 	// check if a new batch starts and the old needs closing before that
 	// if it is already closed with a batch end, do not add a new batch end
@@ -220,16 +221,15 @@ func (srv *DataStreamServer) WriteBlockWithBatchStartToStream(
 	t := utils.StartTimer("write-stream", "writeblockstostream")
 	defer t.LogTimer()
 
-	if err = srv.stream.StartAtomicOp(); err != nil {
-		return err
-	}
-	defer srv.stream.RollbackAtomicOp()
-
 	blockNum := block.NumberU64()
 
 	if err = srv.UnwindIfNecessary(logPrefix, reader, blockNum, prevBlockBatchNum, batchNum); err != nil {
 		return err
 	}
+	if err = srv.stream.StartAtomicOp(); err != nil {
+		return err
+	}
+	defer srv.stream.RollbackAtomicOp()
 
 	// if start of new batch add batch start entries
 	var batchStartEntries *DataStreamEntries

zk/metrics/metrics_xlayer.go

@@ -0,0 +1,65 @@
+package metrics
+
+import (
+	"fmt"
+	"github.com/ledgerwatch/log/v3"
+	"github.com/prometheus/client_golang/prometheus"
+	"time"
+)
+
+// BatchFinalizeTypeLabel batch finalize type label
+type BatchFinalizeTypeLabel string
+
+const (
+	// BatchFinalizeTypeLabelDeadline batch finalize type deadline label
+	BatchFinalizeTypeLabelDeadline BatchFinalizeTypeLabel = "deadline"
+	// BatchFinalizeTypeLabelFullBatch batch finalize type full batch label
+	BatchFinalizeTypeLabelFullBatch BatchFinalizeTypeLabel = "full_batch"
+)
+
+const (
+	CounterOverflow = "BatchCounterOverflow"
+	EmptyTimeOut    = "EmptyBatchTimeOut"
+	NonEmptyTimeOut = "NonEmptyBatchTimeOut"
+)
+
+var (
+	SeqPrefix = "sequencer_"
+	// BatchExecuteTimeName is the name of the metric that shows the batch execution time.
+	BatchExecuteTimeName = SeqPrefix + "batch_execute_time"
+	PoolTxCountName      = SeqPrefix + "pool_tx_count"
+)
+
+func XLayerMetricsInit() {
+	prometheus.MustRegister(BatchExecuteTimeGauge)
+	prometheus.MustRegister(PoolTxCount)
+}
+
+var BatchExecuteTimeGauge = prometheus.NewGaugeVec(
+	prometheus.GaugeOpts{
+		Name: BatchExecuteTimeName,
+		Help: "[SEQUENCER] batch execution time in second",
+	},
+	[]string{"closingReason"},
+)
+
+var PoolTxCount = prometheus.NewGaugeVec(
+	prometheus.GaugeOpts{
+		Name: PoolTxCountName,
+		Help: "[SEQUENCER] txcount of each pool in txpool",
+	},
+	[]string{"poolName"},
+)
+
+// BatchExecuteTime sets the gauge vector to the given batch type and time.
+func BatchExecuteTime(closingReason string, duration time.Duration) {
+	log.Info(fmt.Sprintf("[BatchExecuteTime] ClosingReason: %s, Duration: %.2fs", closingReason, duration.Seconds()))
+	BatchExecuteTimeGauge.WithLabelValues(closingReason).Set(duration.Seconds())
+}
+
+func AddPoolTxCount(pending, basefee, queued int) {
+	log.Info(fmt.Sprintf("[PoolTxCount] pending: %d, basefee: %d, queued: %d", pending, basefee, queued))
+	PoolTxCount.WithLabelValues("pending").Set(float64(pending))
+	PoolTxCount.WithLabelValues("basefee").Set(float64(basefee))
+	PoolTxCount.WithLabelValues("queued").Set(float64(queued))
+}