add metric for pending tx / trust / virtual / verified batch

metrics/prometheus.go

@@ -13,6 +13,7 @@ var (
 	storageMutex  sync.RWMutex
 	registerer    prometheus.Registerer
 	gauges        map[string]prometheus.Gauge
+	gaugeVecs     map[string]*prometheus.GaugeVec
 	counters      map[string]prometheus.Counter
 	counterVecs   map[string]*prometheus.CounterVec
 	histograms    map[string]prometheus.Histogram
@@ -28,6 +29,11 @@ type CounterVecOpts struct {
 	Labels []string
 }
 
+type GaugeVecOpts struct {
+	prometheus.GaugeOpts
+	Labels []string
+}
+
 // HistogramVecOpts holds options for the HistogramVec type.
 type HistogramVecOpts struct {
 	prometheus.HistogramOpts
@@ -40,6 +46,7 @@ func Init() {
 		storageMutex = sync.RWMutex{}
 		registerer = prometheus.DefaultRegisterer
 		gauges = make(map[string]prometheus.Gauge)
+		gaugeVecs = make(map[string]*prometheus.GaugeVec)
 		counters = make(map[string]prometheus.Counter)
 		counterVecs = make(map[string]*prometheus.CounterVec)
 		histograms = make(map[string]prometheus.Histogram)
@@ -54,6 +61,62 @@ func Handler() http.Handler {
 	return promhttp.Handler()
 }
 
+// RegisterGaugeVecs registers the provided gauge vec metrics to the Prometheus
+// registerer.
+func RegisterGaugeVecs(opts ...GaugeVecOpts) {
+	if !initialized {
+		return
+	}
+
+	storageMutex.Lock()
+	defer storageMutex.Unlock()
+
+	for _, options := range opts {
+		registerGaugeVecIfNotExists(options)
+	}
+}
+
+// UnregisterGaugeVecs unregisters the provided gauge vec metrics from the Prometheus
+// registerer.
+func UnregisterGaugeVecs(names ...string) {
+	if !initialized {
+		return
+	}
+
+	storageMutex.Lock()
+	defer storageMutex.Unlock()
+
+	for _, name := range names {
+		unregisterGaugeVecIfExists(name)
+	}
+}
+
+// GaugeVec retrieves gauge ver metric by name
+func GaugeVec(name string) (gaugeVec *prometheus.GaugeVec, exist bool) {
+	if !initialized {
+		return
+	}
+
+	storageMutex.RLock()
+	defer storageMutex.RUnlock()
+
+	gaugeVec, exist = gaugeVecs[name]
+
+	return gaugeVec, exist
+}
+
+// GaugeVecSet sets the value for gauge vec with the given name and label.
+// name and label.
+func GaugeVecSet(name string, label string, value float64) {
+	if !initialized {
+		return
+	}
+
+	if cv, ok := GaugeVec(name); ok {
+		cv.WithLabelValues(label).Add(value)
+	}
+}
+
 // RegisterGauges registers the provided gauge metrics to the Prometheus
 // registerer.
 func RegisterGauges(opts ...prometheus.GaugeOpts) {
@@ -468,6 +531,48 @@ func unregisterGaugeIfExists(name string) {
 	log.Debug("Gauge Metric successfully unregistered!")
 }
 
+// registerGaugeVecIfNotExists registers single gauge vec metric if not exists
+func registerGaugeVecIfNotExists(opts GaugeVecOpts) {
+	log := log.WithFields("metricName", opts.Name)
+	if _, exist := gaugeVecs[opts.Name]; exist {
+		log.Warn("Gauge vec metric already exists.")
+		return
+	}
+
+	log.Debug("Creating Gauge Vec Metric...")
+	gauge := prometheus.NewGaugeVec(opts.GaugeOpts, opts.Labels)
+	log.Debugf("Gauge Vec Metric successfully created! Labels: %p", opts.ConstLabels)
+
+	log.Debug("Registering Gauge Vec Metric...")
+	registerer.MustRegister(gauge)
+	log.Debug("Gauge Vec Metric successfully registered!")
+
+	gaugeVecs[opts.Name] = gauge
+}
+
+// unregisterGaugeVecIfExists unregisters single gauge vec metric if exists
+func unregisterGaugeVecIfExists(name string) {
+	var (
+		gauge *prometheus.GaugeVec
+		ok    bool
+	)
+
+	log := log.WithFields("metricName", name)
+	if gauge, ok = gaugeVecs[name]; !ok {
+		log.Warn("Trying to delete non-existing Gauge metrics.")
+		return
+	}
+
+	log.Debug("Unregistering Gauge Vec Metric...")
+	ok = registerer.Unregister(gauge)
+	if !ok {
+		log.Error("Failed to unregister Gauge Vec Metric.")
+		return
+	}
+	delete(gauges, name)
+	log.Debug("Gauge Vec Metric successfully unregistered!")
+}
+
 // registerCounterIfNotExists registers single counter metric if not exists
 func registerCounterIfNotExists(opts prometheus.CounterOpts) {
 	log := log.WithFields("metricName", opts.Name)

sequencer/db_metric.go

@@ -0,0 +1,22 @@
+package sequencer
+
+import (
+	"github.com/0xPolygonHermez/zkevm-node/log"
+	"github.com/0xPolygonHermez/zkevm-node/sequencer/metrics"
+	"time"
+)
+
+func (d *dbManager) countPendingTx() {
+	ticker := time.NewTicker(time.Second * 10)
+	for {
+		select {
+		case <-ticker.C:
+			transactions, err := d.txPool.CountPendingTransactions(d.ctx)
+			if err != nil {
+				log.Errorf("load pending tx from pool: %v", err)
+				continue
+			}
+			metrics.PendingTxCount(int(transactions))
+		}
+	}
+}

sequencer/dbmanager.go

@@ -53,6 +53,7 @@ func newDBManager(ctx context.Context, config DBManagerCfg, txPool txPool, state
 // Start stars the dbManager routines
 func (d *dbManager) Start() {
 	go d.loadFromPool()
+	go d.countPendingTx()
 	go func() {
 		for {
 			time.Sleep(d.cfg.L2ReorgRetrievalInterval.Duration)

sequencer/finalizer.go

@@ -383,13 +383,17 @@ func (f *finalizer) finalizeBatches(ctx context.Context) {
 			log.Infof(metrics.GetLogStatistics().Summary())
 			metrics.GetLogStatistics().ResetStatistics()
 			metrics.GetLogStatistics().UpdateTimestamp(metrics.NewRound, time.Now())
+			metrics.BatchExecuteTime(metrics.BatchFinalizeTypeLabelDeadline, metrics.GetLogStatistics().GetStatistics(metrics.ProcessingTxCommit))
+			metrics.TrustBatchNum(f.batch.batchNumber - 1)
 		} else if f.isBatchFull() || f.isBatchAlmostFull() {
 			log.Infof("closing batch %d because it's almost full.", f.batch.batchNumber)
 			metrics.GetLogStatistics().SetTag(metrics.BatchCloseReason, "full")
 			f.finalizeBatch(ctx)
 			log.Infof(metrics.GetLogStatistics().Summary())
 			metrics.GetLogStatistics().ResetStatistics()
 			metrics.GetLogStatistics().UpdateTimestamp(metrics.NewRound, time.Now())
+			metrics.BatchExecuteTime(metrics.BatchFinalizeTypeLabelFullBatch, metrics.GetLogStatistics().GetStatistics(metrics.ProcessingTxCommit))
+			metrics.TrustBatchNum(f.batch.batchNumber - 1)
 		}
 
 		if err := ctx.Err(); err != nil {

sequencer/interfaces.go

@@ -23,6 +23,7 @@ type txPool interface {
 	DeleteTransactionByHash(ctx context.Context, hash common.Hash) error
 	MarkWIPTxsAsPending(ctx context.Context) error
 	GetNonWIPPendingTxs(ctx context.Context) ([]pool.Transaction, error)
+	CountPendingTransactions(ctx context.Context) (uint64, error)
 	UpdateTxStatus(ctx context.Context, hash common.Hash, newStatus pool.TxStatus, isWIP bool, failedReason *string) error
 	GetTxZkCountersByHash(ctx context.Context, hash common.Hash) (*state.ZKCounters, error)
 	UpdateTxWIPStatus(ctx context.Context, hash common.Hash, isWIP bool) error

sequencer/metrics/logstatistics.go

@@ -11,6 +11,8 @@ type LogStatistics interface {
 	CumulativeValue(tag LogTag, value int64)
 	CumulativeTiming(tag LogTag, duration time.Duration)
 	SetTag(tag LogTag, value string)
+	GetTag(tag LogTag) string
+	GetStatistics(tag LogTag) int64
 	Summary() string
 	ResetStatistics()
 

sequencer/metrics/logstatisticsimpl.go

@@ -45,6 +45,14 @@ func (l *logStatisticsInstance) SetTag(tag LogTag, value string) {
 	l.tags[tag] = value
 }
 
+func (l *logStatisticsInstance) GetTag(tag LogTag) string {
+	return l.tags[tag]
+}
+
+func (l *logStatisticsInstance) GetStatistics(tag LogTag) int64 {
+	return l.statistics[tag]
+}
+
 func (l *logStatisticsInstance) UpdateTimestamp(tag LogTag, tm time.Time) {
 	l.timestamp[tag] = tm
 }

sequencer/metrics/metrics.go

@@ -24,12 +24,20 @@ const (
 	SequenceRewardInMaticName = Prefix + "sequence_reward_in_matic"
 	// ProcessingTimeName is the name of the metric that shows the processing time.
 	ProcessingTimeName = Prefix + "processing_time"
+	// PendingTxCountName is the name of metric that shows the number of pending transactions.
+	PendingTxCountName = Prefix + "pending_tx_count"
+	// BatchExecuteTimeName is the name of the metric that shows the batch execution time.
+	BatchExecuteTimeName = Prefix + "batch_execute_time"
+	// TrustBatchNumName is the name of the metric that shows the trust batch num
+	TrustBatchNumName = Prefix + "trust_batch_num"
 	// WorkerPrefix is the prefix for the metrics of the worker.
 	WorkerPrefix = Prefix + "worker_"
 	// WorkerProcessingTimeName is the name of the metric that shows the worker processing time.
 	WorkerProcessingTimeName = WorkerPrefix + "processing_time"
 	// TxProcessedLabelName is the name of the label for the processed transactions.
 	TxProcessedLabelName = "status"
+	// BatchFinalizeTypeLabelName is the name of the label for the batch finalize type.
+	BatchFinalizeTypeLabelName = "batch_type"
 )
 
 // TxProcessedLabel represents the possible values for the
@@ -45,12 +53,20 @@ const (
 	TxProcessedLabelFailed TxProcessedLabel = "failed"
 )
 
+type BatchFinalizeTypeLabel string
+
+const (
+	BatchFinalizeTypeLabelDeadline  BatchFinalizeTypeLabel = "deadline"
+	BatchFinalizeTypeLabelFullBatch BatchFinalizeTypeLabel = "full_batch"
+)
+
 // Register the metrics for the sequencer package.
 func Register() {
 	var (
 		counters    []prometheus.CounterOpts
 		counterVecs []metrics.CounterVecOpts
 		gauges      []prometheus.GaugeOpts
+		gaugeVecs   []metrics.GaugeVecOpts
 		histograms  []prometheus.HistogramOpts
 	)
 
@@ -88,6 +104,24 @@ func Register() {
 			Name: SequenceRewardInMaticName,
 			Help: "[SEQUENCER] reward for a sequence in Matic",
 		},
+		{
+			Name: PendingTxCountName,
+			Help: "[SEQUENCER] number of pending transactions",
+		},
+		{
+			Name: TrustBatchNumName,
+			Help: "[SEQUENCER] trust batch num",
+		},
+	}
+
+	gaugeVecs = []metrics.GaugeVecOpts{
+		{
+			GaugeOpts: prometheus.GaugeOpts{
+				Name: BatchExecuteTimeName,
+				Help: "[SEQUENCER] batch execution time",
+			},
+			Labels: []string{BatchFinalizeTypeLabelName},
+		},
 	}
 
 	histograms = []prometheus.HistogramOpts{
@@ -104,9 +138,25 @@ func Register() {
 	metrics.RegisterCounters(counters...)
 	metrics.RegisterCounterVecs(counterVecs...)
 	metrics.RegisterGauges(gauges...)
+	metrics.RegisterGaugeVecs(gaugeVecs...)
 	metrics.RegisterHistograms(histograms...)
 }
 
+// PendingTxCount sets the gauge to the given number of pending transactions.
+func PendingTxCount(count int) {
+	metrics.GaugeSet(PendingTxCountName, float64(count))
+}
+
+// BatchExecuteTime sets the gauge vector to the given batch type and time.
+func BatchExecuteTime(batchType BatchFinalizeTypeLabel, time int64) {
+	metrics.GaugeVecSet(BatchExecuteTimeName, string(batchType), float64(time))
+}
+
+// TrustBatchNum set the gauge to the given trust batch num
+func TrustBatchNum(batchNum uint64) {
+	metrics.GaugeSet(TrustBatchNumName, float64(batchNum))
+}
+
 // AverageGasPrice sets the gauge to the given average gas price.
 func AverageGasPrice(price float64) {
 	metrics.GaugeSet(GasPriceEstimatedAverageName, price)

synchronizer/metrics/metrics.go

@@ -37,6 +37,12 @@ const (
 
 	// ProcessTrustedBatchTimeName is the name of the label to process trusted batch.
 	ProcessTrustedBatchTimeName = Prefix + "process_trusted_batch_time"
+
+	// VirtualBatchNumName is the name of the metric virtual batch number
+	VirtualBatchNumName = Prefix + "virtual_batch_num"
+
+	// VerifiedBatchNumName is the name of the metric verified batch number
+	VerifiedBatchNumName = Prefix + "verified_batch_num"
 )
 
 // Register the metrics for the synchronizer package.
@@ -80,7 +86,27 @@ func Register() {
 		},
 	}
 
+	gauge := []prometheus.GaugeOpts{
+		{
+			Name: VirtualBatchNumName,
+			Help: "[SYNCHRONIZER] virtual batch num",
+		},
+		{
+			Name: VerifiedBatchNumName,
+			Help: "[SYNCHRONIZER] verified batch num",
+		},
+	}
+
 	metrics.RegisterHistograms(histograms...)
+	metrics.RegisterGauges(gauge...)
+}
+
+func VirtualBatchNum(batchNum uint64) {
+	metrics.GaugeSet(VirtualBatchNumName, float64(batchNum))
+}
+
+func VerifiedBatchNum(batchNum uint64) {
+	metrics.GaugeSet(VerifiedBatchNumName, float64(batchNum))
 }
 
 // InitializationTime observes the time initializing the synchronizer on the histogram.

synchronizer/synchronizer.go

@@ -1001,6 +1001,7 @@ func (s *ClientSynchronizer) processSequenceBatches(sequencedBatches []etherman.
 			log.Errorf("error storing virtualBatch. BatchNumber: %d, BlockNumber: %d, error: %v", virtualBatch.BatchNumber, blockNumber, err)
 			return err
 		}
+		metrics.VirtualBatchNum(virtualBatch.BatchNumber)
 	}
 	// Insert the sequence to allow the aggregator verify the sequence batches
 	seq := state.Sequence{
@@ -1251,6 +1252,7 @@ func (s *ClientSynchronizer) processTrustedVerifyBatches(lastVerifiedBatch ether
 			log.Errorf("error storing the verifiedB in processTrustedVerifyBatches. BlockNumber: %d, error: %v", lastVerifiedBatch.BlockNumber, err)
 			return err
 		}
+		metrics.VerifiedBatchNum(verifiedB.BatchNumber)
 	}
 	return nil
 }