feat(metric): add AltDA metrics to batcher

* Adds a new `eigenda_failover_total` counter metric to batcher, tracking the total number of AltDA failovers.

* Adds `datype` label to the `batcherTxEvs` metrics.

Author: iquidus

op-batcher/batcher/channel.go

@@ -81,7 +81,6 @@ func (c *channel) rewindAltDAFrameCursor(txData txData) {
 // It rewinds the channelBuilder's frameCursor to the first frame of the failed txData,
 // so that the frames can be resubmitted. failoverToEthDA should be set to true when using altDA
 // and altDA is down. This will switch the channel to submit frames to ethDA instead.
-// TODO: add a metric for altDA submission failures.
 func (c *channel) AltDASubmissionFailed(id string, failoverToEthDA bool) {
 	// We coopt TxFailed to rewind the frame cursor.
 	// This will force a resubmit of all the following frames as well,
@@ -100,6 +99,7 @@ func (c *channel) AltDASubmissionFailed(id string, failoverToEthDA bool) {
 		// batcherService.initChannelConfig function stateless so that we can reuse it.
 		c.log.Info("Failing over to calldata txs", "id", c.ID())
 		c.cfg.DaType = DaTypeCalldata
+		c.metr.RecordFailoverToEthDA()
 	}
 }
 
@@ -124,13 +124,13 @@ func (c *channel) TxFailed(id string) {
 	} else {
 		c.log.Warn("unknown transaction marked as failed", "id", id)
 	}
-	c.metr.RecordBatchTxFailed()
+	c.metr.RecordBatchTxFailed(c.cfg.DaType.String())
 }
 
 // TxConfirmed marks a transaction as confirmed on L1. Returns a bool indicating
 // whether the channel timed out on chain.
 func (c *channel) TxConfirmed(id string, inclusionBlock eth.BlockID) bool {
-	c.metr.RecordBatchTxSuccess()
+	c.metr.RecordBatchTxSuccess(c.cfg.DaType.String())
 	c.log.Debug("marked transaction as confirmed", "id", id, "block", inclusionBlock)
 	if _, ok := c.pendingTransactions[id]; !ok {
 		c.log.Warn("unknown transaction marked as confirmed", "id", id, "block", inclusionBlock)

op-batcher/batcher/channel_manager.go

@@ -147,16 +147,18 @@ func (s *channelManager) TxFailed(_id txID) {
 func (s *channelManager) TxConfirmed(_id txID, inclusionBlock eth.BlockID) {
 
 	id := _id.String()
+	daType := DaTypeCalldata
 	if channel, ok := s.txChannels[id]; ok {
 		delete(s.txChannels, id)
 		if timedOut := channel.TxConfirmed(id, inclusionBlock); timedOut {
 			s.log.Warn("channel timed out on chain", "channel_id", channel.ID(), "tx_id", id)
 			s.handleChannelInvalidated(channel)
 		}
+		daType = channel.cfg.DaType
 	} else {
 		s.log.Warn("transaction from unknown channel marked as confirmed", "id", id)
 	}
-	s.metr.RecordBatchTxSubmitted()
+	s.metr.RecordBatchTxSubmitted(daType.String())
 	s.log.Debug("marked transaction as confirmed", "id", id, "block", inclusionBlock)
 }
 

op-batcher/metrics/metrics.go

@@ -59,15 +59,17 @@ type Metricer interface {
 	// It should be called when clearing the ChannelManager state.
 	ClearAllStateMetrics()
 
-	RecordBatchTxSubmitted()
-	RecordBatchTxSuccess()
-	RecordBatchTxFailed()
+	RecordBatchTxSubmitted(daType string)
+	RecordBatchTxSuccess(daType string)
+	RecordBatchTxFailed(daType string)
 
 	RecordBlobUsedBytes(num int)
 
 	Document() []opmetrics.DocumentedMetric
 
 	PendingDABytes() float64
+
+	RecordFailoverToEthDA()
 }
 
 type Metrics struct {
@@ -122,6 +124,8 @@ type Metrics struct {
 	pidControllerIntegral   prometheus.Gauge
 	pidControllerDerivative prometheus.Gauge
 	pidResponseTime         prometheus.Histogram
+
+	eigenDAFailoverToEthDA prometheus.Counter
 }
 
 var _ Metricer = (*Metrics)(nil)
@@ -235,7 +239,7 @@ func NewMetrics(procName string) *Metrics {
 			Buckets:   prometheus.LinearBuckets(0.0, eth.MaxBlobDataSize/13, 14),
 		}),
 
-		batcherTxEvs: opmetrics.NewEventVec(factory, ns, "", "batcher_tx", "BatcherTx", []string{"stage"}),
+		batcherTxEvs: opmetrics.NewEventVec(factory, ns, "", "batcher_tx", "BatcherTx", []string{"stage", "datype"}),
 
 		throttleIntensity: *factory.NewGaugeVec(prometheus.GaugeOpts{
 			Namespace: ns,
@@ -298,6 +302,11 @@ func NewMetrics(procName string) *Metrics {
 			Name:      "unsafe_da_bytes",
 			Help:      "The estimated number of unsafe DA bytes",
 		}),
+		eigenDAFailoverToEthDA: factory.NewCounter(prometheus.CounterOpts{
+			Namespace: ns,
+			Name:      "eigenda_failover_total",
+			Help:      "Total number of failovers to EthDA",
+		}),
 	}
 	m.pendingDABytesGaugeFunc = factory.NewGaugeFunc(prometheus.GaugeOpts{
 		Namespace: ns,
@@ -428,16 +437,16 @@ func (m *Metrics) RecordChannelTimedOut(id derive.ChannelID) {
 	m.channelEvs.Record(StageTimedOut)
 }
 
-func (m *Metrics) RecordBatchTxSubmitted() {
-	m.batcherTxEvs.Record(TxStageSubmitted)
+func (m *Metrics) RecordBatchTxSubmitted(daType string) {
+	m.batcherTxEvs.Record(TxStageSubmitted, daType)
 }
 
-func (m *Metrics) RecordBatchTxSuccess() {
-	m.batcherTxEvs.Record(TxStageSuccess)
+func (m *Metrics) RecordBatchTxSuccess(daType string) {
+	m.batcherTxEvs.Record(TxStageSuccess, daType)
 }
 
-func (m *Metrics) RecordBatchTxFailed() {
-	m.batcherTxEvs.Record(TxStageFailed)
+func (m *Metrics) RecordBatchTxFailed(daType string) {
+	m.batcherTxEvs.Record(TxStageFailed, daType)
 }
 
 func (m *Metrics) RecordBlobUsedBytes(num int) {
@@ -511,3 +520,8 @@ func (m *Metrics) RecordThrottleControllerState(error, integral, derivative floa
 func (m *Metrics) RecordThrottleResponseTime(duration time.Duration) {
 	m.pidResponseTime.Observe(duration.Seconds())
 }
+
+// RecordFailoverToEthDA records when the system fails over to EthDA
+func (m *Metrics) RecordFailoverToEthDA() {
+	m.eigenDAFailoverToEthDA.Inc()
+}

op-batcher/metrics/noop.go

@@ -56,10 +56,10 @@ func (*noopMetrics) RecordUnsafeDABytes(int64) {}
 func (*noopMetrics) RecordThrottleControllerState(error, integral, derivative float64) {}
 func (*noopMetrics) RecordThrottleResponseTime(duration time.Duration)                 {}
 
-func (*noopMetrics) RecordBatchTxSubmitted() {}
-func (*noopMetrics) RecordBatchTxSuccess()   {}
-func (*noopMetrics) RecordBatchTxFailed()    {}
-func (*noopMetrics) RecordBlobUsedBytes(int) {}
+func (*noopMetrics) RecordBatchTxSubmitted(string) {}
+func (*noopMetrics) RecordBatchTxSuccess(string)   {}
+func (*noopMetrics) RecordBatchTxFailed(string)    {}
+func (*noopMetrics) RecordBlobUsedBytes(int)       {}
 func (*noopMetrics) StartBalanceMetrics(log.Logger, *ethclient.Client, common.Address) io.Closer {
 	return nil
 }
@@ -78,3 +78,5 @@ func (nm *ThrottlingMetrics) PendingDABytes() float64 {
 }
 
 func (*noopMetrics) ClearAllStateMetrics() {}
+
+func (*noopMetrics) RecordFailoverToEthDA() {}