feat: Collects flashblock metrics during execution (#156)

* feat: Collects flashbot metrics during execution

We now collect aggregated (by block) flashblock metrics

* chore: removes unused flashblock metrics

Reth supports flashblocks, but metrics are only available to base-reth-node for now.

Author: mw2000

runner/clients/baserethnode/metrics.go

@@ -48,6 +48,19 @@ func (r *metricsCollector) GetMetricTypes() map[string]bool {
 		"reth_sync_state_provider_total_storage_fetch_latency": true,
 		"reth_sync_state_provider_total_account_fetch_latency": true,
 		"reth_sync_state_provider_total_code_fetch_latency":    true,
+		"reth_reth_flashblocks_upstream_errors":                true,
+		"reth_reth_flashblocks_upstream_messages":              true,
+		"reth_reth_flashblocks_block_processing_duration":      true,
+		"reth_reth_flashblocks_sender_recovery_duration":       true,
+		"reth_reth_flashblocks_unexpected_block_order":         true,
+		"reth_reth_flashblocks_flashblocks_in_block":           true,
+		"reth_reth_flashblocks_block_processing_error":         true,
+		"reth_reth_flashblocks_pending_clear_catchup":          true,
+		"reth_reth_flashblocks_pending_clear_reorg":            true,
+		"reth_reth_flashblocks_pending_snapshot_fb_index":      true,
+		"reth_reth_flashblocks_pending_snapshot_height":        true,
+		"reth_reth_flashblocks_bundle_state_clone_duration":    true,
+		"reth_reth_flashblocks_bundle_state_clone_size":        true,
 	}
 }
 
@@ -66,23 +79,25 @@ func (r *metricsCollector) Collect(ctx context.Context, m *metrics.BlockMetrics)
 	}
 
 	txtParser := expfmt.NewTextParser(model.LegacyValidation)
-	metrics, err := txtParser.TextToMetricFamilies(bytes.NewReader(body))
+	parsedMetrics, err := txtParser.TextToMetricFamilies(bytes.NewReader(body))
 	if err != nil {
 		return fmt.Errorf("failed to parse metrics: %w", err)
 	}
 
 	metricTypes := r.GetMetricTypes()
 
-	for _, metric := range metrics {
+	for _, metric := range parsedMetrics {
 		name := metric.GetName()
 		if metricTypes[name] {
 			metricVal := metric.GetMetric()
 			if len(metricVal) != 1 {
-				r.log.Warn("expected 1 metric, got %d for metric %s", len(metricVal), name)
+				r.log.Warn("expected 1 metric value", "got", len(metricVal), "metric", name)
 			}
-			err = m.UpdatePrometheusMetric(name, metricVal[0])
-			if err != nil {
-				r.log.Warn("failed to add metric %s: %s", name, err)
+			if len(metricVal) > 0 {
+				err = m.UpdatePrometheusMetric(name, metricVal[0])
+				if err != nil {
+					r.log.Warn("failed to add metric", "name", name, "error", err)
+				}
 			}
 		}
 	}

runner/network/types/types.go

@@ -134,13 +134,18 @@ func getAverage(metrics []metrics.BlockMetrics, metricName string) float64 {
 }
 
 const (
-	UpdateForkChoiceLatencyMetric = "latency/update_fork_choice"
-	NewPayloadLatencyMetric       = "latency/new_payload"
-	GetPayloadLatencyMetric       = "latency/get_payload"
-	SendTxsLatencyMetric          = "latency/send_txs"
-	GasPerBlockMetric             = "gas/per_block"
-	GasPerSecondMetric            = "gas/per_second"
-	TransactionsPerBlockMetric    = "transactions/per_block"
+	UpdateForkChoiceLatencyMetric      = "latency/update_fork_choice"
+	NewPayloadLatencyMetric            = "latency/new_payload"
+	GetPayloadLatencyMetric            = "latency/get_payload"
+	SendTxsLatencyMetric               = "latency/send_txs"
+	GasPerBlockMetric                  = "gas/per_block"
+	GasPerSecondMetric                 = "gas/per_second"
+	TransactionsPerBlockMetric         = "transactions/per_block"
+	FlashblockProcessingDurationMetric = "reth_flashblocks_block_processing_duration"
+	FlashblockSenderRecoveryMetric     = "reth_flashblocks_sender_recovery_duration"
+	FlashblocksInBlockMetric           = "reth_flashblocks_flashblocks_in_block"
+	FlashblockUpstreamMessagesMetric   = "reth_flashblocks_upstream_messages"
+	FlashblockBundleStateCloneDuration = "reth_flashblocks_bundle_state_clone_duration"
 )
 
 type SequencerKeyMetrics struct {
@@ -152,7 +157,9 @@ type SequencerKeyMetrics struct {
 
 type ValidatorKeyMetrics struct {
 	CommonKeyMetrics
-	AverageNewPayloadLatency float64 `json:"newPayload"`
+	AverageNewPayloadLatency            float64 `json:"newPayload"`
+	AverageFlashblockProcessingDuration float64 `json:"flashblockProcessingDuration,omitempty"`
+	AverageFlashblocksInBlock           float64 `json:"flashblocksInBlock,omitempty"`
 }
 
 type CommonKeyMetrics struct {
@@ -163,9 +170,13 @@ type CommonKeyMetrics struct {
 func BlockMetricsToValidatorSummary(metrics []metrics.BlockMetrics) *ValidatorKeyMetrics {
 	averageNewPayloadLatency := getAverage(metrics, NewPayloadLatencyMetric)
 	averageGasPerSecond := getAverage(metrics, GasPerSecondMetric)
+	averageFlashblockProcessingDuration := getAverage(metrics, FlashblockProcessingDurationMetric)
+	averageFlashblocksInBlock := getAverage(metrics, FlashblocksInBlockMetric)
 
 	return &ValidatorKeyMetrics{
-		AverageNewPayloadLatency: averageNewPayloadLatency,
+		AverageNewPayloadLatency:            averageNewPayloadLatency,
+		AverageFlashblockProcessingDuration: averageFlashblockProcessingDuration,
+		AverageFlashblocksInBlock:           averageFlashblocksInBlock,
 		CommonKeyMetrics: CommonKeyMetrics{
 			AverageGasPerSecond: averageGasPerSecond,
 		},