feat(rollup-relayer): add sync height and throughput metrics (#1520)

common/version/version.go

@@ -5,7 +5,7 @@ import (
 	"runtime/debug"
 )
 
-var tag = "v4.4.59"
+var tag = "v4.4.62"
 
 var commit = func() string {
 	if info, ok := debug.ReadBuildInfo(); ok {

database/config.go

@@ -4,6 +4,8 @@ import (
 	"encoding/json"
 	"os"
 	"path/filepath"
+
+	"scroll-tech/common/utils"
 )
 
 // DBConfig db config
@@ -29,5 +31,11 @@ func NewConfig(file string) (*DBConfig, error) {
 		return nil, err
 	}
 
+	// Override config with environment variables
+	err = utils.OverrideConfigWithEnv(cfg, "SCROLL_ROLLUP_DB_CONFIG")
+	if err != nil {
+		return nil, err
+	}
+
 	return cfg, nil
 }

rollup/internal/controller/relayer/l2_relayer.go

@@ -450,6 +450,18 @@ func (r *Layer2Relayer) ProcessPendingBatches() {
 			log.Error("UpdateCommitTxHashAndRollupStatus failed", "hash", dbBatch.Hash, "index", dbBatch.Index, "err", err)
 			return
 		}
+
+		var maxBlockHeight uint64
+		var totalGasUsed uint64
+		for _, dbChunk := range dbChunks {
+			if dbChunk.EndBlockNumber > maxBlockHeight {
+				maxBlockHeight = dbChunk.EndBlockNumber
+			}
+			totalGasUsed += dbChunk.TotalL2TxGas
+		}
+		r.metrics.rollupL2RelayerCommitBlockHeight.Set(float64(maxBlockHeight))
+		r.metrics.rollupL2RelayerCommitThroughput.Add(float64(totalGasUsed))
+
 		r.metrics.rollupL2RelayerProcessPendingBatchSuccessTotal.Inc()
 		log.Info("Sent the commitBatch tx to layer1", "batch index", dbBatch.Index, "batch hash", dbBatch.Hash, "tx hash", txHash.String())
 	}

rollup/internal/controller/relayer/l2_relayer_metrics.go

@@ -29,6 +29,9 @@ type l2RelayerMetrics struct {
 	rollupL2RelayerProcessPendingBundlesFinalizedSuccessTotal       prometheus.Counter
 	rollupL2BundlesFinalizedConfirmedTotal                          prometheus.Counter
 	rollupL2BundlesFinalizedConfirmedFailedTotal                    prometheus.Counter
+
+	rollupL2RelayerCommitBlockHeight prometheus.Gauge
+	rollupL2RelayerCommitThroughput  prometheus.Counter
 }
 
 var (
@@ -123,6 +126,14 @@ func initL2RelayerMetrics(reg prometheus.Registerer) *l2RelayerMetrics {
 				Name: "rollup_layer2_bundles_finalized_confirmed_failed_total",
 				Help: "Total number of failed confirmations for finalized bundles on layer2.",
 			}),
+			rollupL2RelayerCommitBlockHeight: promauto.With(reg).NewGauge(prometheus.GaugeOpts{
+				Name: "rollup_l2_relayer_commit_block_height",
+				Help: "The latest block height committed by the L2 relayer",
+			}),
+			rollupL2RelayerCommitThroughput: promauto.With(reg).NewCounter(prometheus.CounterOpts{
+				Name: "rollup_l2_relayer_commit_throughput",
+				Help: "The cumulative gas used in blocks committed by the L2 relayer",
+			}),
 		}
 	})
 	return l2RelayerMetric

rollup/internal/controller/watcher/batch_proposer.go

@@ -53,6 +53,9 @@ type BatchProposer struct {
 
 	// total number of times that batch proposer stops early due to compressed data compatibility breach
 	compressedDataCompatibilityBreachTotal prometheus.Counter
+
+	batchProposeBlockHeight prometheus.Gauge
+	batchProposeThroughput  prometheus.Counter
 }
 
 // NewBatchProposer creates a new BatchProposer instance.
@@ -134,6 +137,14 @@ func NewBatchProposer(ctx context.Context, cfg *config.BatchProposerConfig, chai
 			Name: "rollup_propose_batch_estimate_blob_size_time",
 			Help: "Time taken to estimate blob size for the chunk.",
 		}),
+		batchProposeBlockHeight: promauto.With(reg).NewGauge(prometheus.GaugeOpts{
+			Name: "rollup_batch_propose_block_height",
+			Help: "The block height of the latest proposed batch",
+		}),
+		batchProposeThroughput: promauto.With(reg).NewCounter(prometheus.CounterOpts{
+			Name: "rollup_batch_propose_throughput",
+			Help: "The total gas used in proposed batches",
+		}),
 	}
 
 	return p
@@ -195,6 +206,18 @@ func (p *BatchProposer) updateDBBatchInfo(batch *encoding.Batch, codecVersion en
 		p.recordAllBatchMetrics(metrics)
 	}
 
+	if len(batch.Chunks) > 0 && len(batch.Chunks[len(batch.Chunks)-1].Blocks) > 0 {
+		lastChunk := batch.Chunks[len(batch.Chunks)-1]
+		lastBlock := lastChunk.Blocks[len(lastChunk.Blocks)-1]
+		p.batchProposeBlockHeight.Set(float64(lastBlock.Header.Number.Uint64()))
+	}
+
+	var totalGasUsed uint64
+	for _, chunk := range batch.Chunks {
+		totalGasUsed += chunk.L2GasUsed()
+	}
+	p.batchProposeThroughput.Add(float64(totalGasUsed))
+
 	p.proposeBatchUpdateInfoTotal.Inc()
 	err := p.db.Transaction(func(dbTX *gorm.DB) error {
 		dbBatch, dbErr := p.batchOrm.InsertBatch(p.ctx, batch, codecConfig, *metrics, dbTX)

rollup/internal/controller/watcher/chunk_proposer.go

@@ -56,6 +56,9 @@ type ChunkProposer struct {
 
 	// total number of times that chunk proposer stops early due to compressed data compatibility breach
 	compressedDataCompatibilityBreachTotal prometheus.Counter
+
+	chunkProposeBlockHeight prometheus.Gauge
+	chunkProposeThroughput  prometheus.Counter
 }
 
 // NewChunkProposer creates a new ChunkProposer instance.
@@ -150,6 +153,14 @@ func NewChunkProposer(ctx context.Context, cfg *config.ChunkProposerConfig, chai
 			Name: "rollup_propose_chunk_estimate_blob_size_time",
 			Help: "Time taken to estimate blob size for the chunk.",
 		}),
+		chunkProposeBlockHeight: promauto.With(reg).NewGauge(prometheus.GaugeOpts{
+			Name: "rollup_chunk_propose_block_height",
+			Help: "The block height of the latest proposed chunk",
+		}),
+		chunkProposeThroughput: promauto.With(reg).NewCounter(prometheus.CounterOpts{
+			Name: "rollup_chunk_propose_throughput",
+			Help: "The total gas used in proposed chunks",
+		}),
 	}
 
 	return p
@@ -214,6 +225,11 @@ func (p *ChunkProposer) updateDBChunkInfo(chunk *encoding.Chunk, codecVersion en
 		p.recordAllChunkMetrics(metrics)
 	}
 
+	if len(chunk.Blocks) > 0 {
+		p.chunkProposeBlockHeight.Set(float64(chunk.Blocks[len(chunk.Blocks)-1].Header.Number.Uint64()))
+	}
+	p.chunkProposeThroughput.Add(float64(chunk.L2GasUsed()))
+
 	p.proposeChunkUpdateInfoTotal.Inc()
 	err := p.db.Transaction(func(dbTX *gorm.DB) error {
 		dbChunk, err := p.chunkOrm.InsertChunk(p.ctx, chunk, codecConfig, *metrics, dbTX)

rollup/internal/controller/watcher/l2_watcher.go

@@ -127,6 +127,7 @@ func (w *L2WatcherClient) getAndStoreBlocks(ctx context.Context, from, to uint64
 			return fmt.Errorf("failed to GetBlockByNumberOrHash: %v. number: %v", err, number)
 		}
 		if block.RowConsumption == nil {
+			w.metrics.fetchNilRowConsumptionBlockTotal.Inc()
 			return fmt.Errorf("fetched block does not contain RowConsumption. number: %v", number)
 		}
 
@@ -151,6 +152,7 @@ func (w *L2WatcherClient) getAndStoreBlocks(ctx context.Context, from, to uint64
 				return fmt.Errorf("failed to estimate block L1 commit calldata size: %v", err)
 			}
 			w.metrics.rollupL2BlockL1CommitCalldataSize.Set(float64(blockL1CommitCalldataSize))
+			w.metrics.rollupL2WatcherSyncThroughput.Add(float64(block.Header.GasUsed))
 		}
 		if err := w.l2BlockOrm.InsertL2Blocks(w.ctx, blocks); err != nil {
 			return fmt.Errorf("failed to batch insert BlockTraces: %v", err)

rollup/internal/controller/watcher/l2_watcher_metrics.go

@@ -12,6 +12,9 @@ type l2WatcherMetrics struct {
 	fetchRunningMissingBlocksHeight   prometheus.Gauge
 	rollupL2BlocksFetchedGap          prometheus.Gauge
 	rollupL2BlockL1CommitCalldataSize prometheus.Gauge
+	fetchNilRowConsumptionBlockTotal  prometheus.Counter
+
+	rollupL2WatcherSyncThroughput prometheus.Counter
 }
 
 var (
@@ -38,6 +41,14 @@ func initL2WatcherMetrics(reg prometheus.Registerer) *l2WatcherMetrics {
 				Name: "rollup_l2_block_l1_commit_calldata_size",
 				Help: "The l1 commitBatch calldata size of the l2 block",
 			}),
+			fetchNilRowConsumptionBlockTotal: promauto.With(reg).NewCounter(prometheus.CounterOpts{
+				Name: "rollup_l2_watcher_fetch_nil_row_consumption_block_total",
+				Help: "The total number of occurrences where a fetched block has nil RowConsumption",
+			}),
+			rollupL2WatcherSyncThroughput: prometheus.NewCounter(prometheus.CounterOpts{
+				Name: "rollup_l2_watcher_sync_throughput",
+				Help: "The cumulative gas used in blocks that L2 watcher sync",
+			}),
 		}
 	})
 	return l2WatcherMetric