feat(metrics): add Prometheus metrics support for beacon client

* Implement core beacon metrics (head slot, epochs, active validators, reorgs, deposits)

* Expose configurable metrics address and port in the beacon node CLI

* Integrate with Grafana dashboard via `beaconMetrics.json`

* Update ream_launcher.star + Kurtosis integration

Closes #1526

Author: tosynthegeek

Cargo.lock

@@ -8,9 +8,9 @@ use ream_p2p::bootnodes::Bootnodes;
 use url::Url;
 
 use crate::cli::constants::{
-    DEFAULT_DISABLE_DISCOVERY, DEFAULT_DISCOVERY_PORT, DEFAULT_HTTP_ADDRESS,
-    DEFAULT_HTTP_ALLOW_ORIGIN, DEFAULT_HTTP_PORT, DEFAULT_NETWORK, DEFAULT_SOCKET_ADDRESS,
-    DEFAULT_SOCKET_PORT,
+    DEFAULT_BEACON_METRICS_ADDRESS, DEFAULT_BEACON_METRICS_PORT, DEFAULT_DISABLE_DISCOVERY,
+    DEFAULT_DISCOVERY_PORT, DEFAULT_HTTP_ADDRESS, DEFAULT_HTTP_ALLOW_ORIGIN, DEFAULT_HTTP_PORT,
+    DEFAULT_METRICS_ENABLED, DEFAULT_NETWORK, DEFAULT_SOCKET_ADDRESS, DEFAULT_SOCKET_PORT,
 };
 
 #[derive(Debug, Parser)]
@@ -89,6 +89,15 @@ pub struct BeaconNodeConfig {
         help = "Number of epochs to retain blob sidecars. Defaults to network spec value (4096 epochs for mainnet, ~18 days)"
     )]
     pub blob_retention_epochs: Option<u64>,
+
+    #[arg(long = "metrics", help = "Enable metrics", default_value_t = DEFAULT_METRICS_ENABLED)]
+    pub enable_metrics: bool,
+
+    #[arg(long, help = "Set metrics address", default_value_t = DEFAULT_BEACON_METRICS_ADDRESS)]
+    pub metrics_address: IpAddr,
+
+    #[arg(long, help = "Set metrics port", default_value_t = DEFAULT_BEACON_METRICS_PORT)]
+    pub metrics_port: u16,
 }
 
 impl From<BeaconNodeConfig> for ManagerConfig {

bin/ream/src/cli/beacon_node.rs

@@ -8,6 +8,8 @@ pub const DEFAULT_HTTP_ADDRESS: IpAddr = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1))
 pub const DEFAULT_HTTP_ALLOW_ORIGIN: bool = false;
 pub const DEFAULT_HTTP_PORT: u16 = 5052;
 pub const DEFAULT_KEY_MANAGER_HTTP_PORT: u16 = 8008;
+pub const DEFAULT_BEACON_METRICS_ADDRESS: IpAddr = IpAddr::V4(Ipv4Addr::UNSPECIFIED);
+pub const DEFAULT_BEACON_METRICS_PORT: u16 = 8008;
 pub const DEFAULT_METRICS_ENABLED: bool = false;
 pub const DEFAULT_METRICS_ADDRESS: IpAddr = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1));
 pub const DEFAULT_METRICS_PORT: u16 = 8080;

bin/ream/src/cli/constants.rs

@@ -54,7 +54,8 @@ use ream_fork_choice_lean::{
 };
 use ream_keystore::keystore::EncryptedKeystore;
 use ream_metrics::{
-    ATTESTATION_COMMITTEE_SUBNET, NODE_INFO, NODE_START_TIME_SECONDS, set_int_gauge_vec,
+    ATTESTATION_COMMITTEE_SUBNET, NODE_INFO, NODE_START_TIME_SECONDS, init_node_metrics,
+    set_int_gauge_vec,
 };
 use ream_network_manager::service::NetworkManagerService;
 use ream_network_spec::networks::{
@@ -495,6 +496,17 @@ async fn run_beacon_node_inner(
 ) {
     info!("starting up beacon node...");
 
+    if config.enable_metrics {
+        let address = SocketAddr::new(config.metrics_address, config.metrics_port);
+        prometheus_exporter::start(address).expect("Failed to start prometheus exporter");
+        info!(
+            "Metrics started on {}:{}",
+            config.metrics_address, config.metrics_port
+        );
+
+        init_node_metrics();
+    }
+
     if initialize_globals {
         set_beacon_network_spec(config.network.clone());
     }

bin/ream/src/main.rs

@@ -23,6 +23,7 @@ ream-consensus-misc.workspace = true
 ream-events-beacon.workspace = true
 ream-execution-engine.workspace = true
 ream-fork-choice-beacon.workspace = true
+ream-metrics.workspace = true
 ream-network-spec.workspace = true
 ream-operation-pool.workspace = true
 ream-req-resp.workspace = true

book/cli/ream/beacon_node.md

@@ -1,17 +1,21 @@
 use std::sync::Arc;
 
-use anyhow::bail;
+use anyhow::{anyhow, bail};
 use ream_consensus_beacon::{
     attestation::Attestation, attester_slashing::AttesterSlashing,
     electra::beacon_block::SignedBeaconBlock,
 };
-use ream_consensus_misc::constants::beacon::{FULU_FORK_EPOCH, genesis_validators_root};
+use ream_consensus_misc::{
+    constants::beacon::{FULU_FORK_EPOCH, genesis_validators_root},
+    misc::compute_epoch_at_slot,
+};
 use ream_events_beacon::{BeaconEvent, BeaconEventSender, event::chain::BlockEvent};
 use ream_execution_engine::ExecutionEngine;
 use ream_fork_choice_beacon::{
     handlers::{on_attestation, on_attester_slashing, on_block, on_tick},
     store::Store,
 };
+use ream_metrics::{BEACON_HEAD_EPOCH, BEACON_HEAD_SLOT, BEACON_REORGS_TOTAL};
 use ream_network_spec::networks::beacon_network_spec;
 use ream_operation_pool::OperationPool;
 use ream_req_resp::beacon::messages::status::Status;
@@ -48,6 +52,7 @@ impl BeaconChain {
 
     pub async fn process_block(&self, signed_block: SignedBeaconBlock) -> anyhow::Result<()> {
         let mut store = self.store.lock().await;
+        let previous_head = store.get_head().ok();
 
         on_block(
             &mut store,
@@ -57,7 +62,34 @@ impl BeaconChain {
         )
         .await?;
 
-        // Build and Emit Block event
+        let new_head = store.get_head()?;
+        let new_head_block = store
+            .db
+            .block_provider()
+            .get(new_head)?
+            .ok_or_else(|| anyhow!("head block not found in store"))?;
+        let new_head_slot = new_head_block.message.slot;
+
+        BEACON_HEAD_SLOT.set(new_head_slot as i64);
+        BEACON_HEAD_EPOCH.set(compute_epoch_at_slot(new_head_slot) as i64);
+
+        // Detect canonical chain reorgs for beacon_reorgs_total.
+        // A head change alone is insufficient because normal chain extensions
+        // also change the head. We only count a reorg when the previous head
+        // is no longer an ancestor of the new head.
+        if let Some(previous_head) = previous_head
+            && previous_head != new_head
+            && let Some(previous_head_block) = store.db.block_provider().get(previous_head)?
+        {
+            let previous_head_slot = previous_head_block.message.slot;
+            let still_ancestor =
+                store.get_ancestor(new_head, previous_head_slot).ok() == Some(previous_head);
+
+            if !still_ancestor {
+                BEACON_REORGS_TOTAL.inc();
+            }
+        }
+
         let finalized_checkpoint = store.db.finalized_checkpoint_provider().get().ok();
         let block_event =
             BlockEvent::from_block(&signed_block, finalized_checkpoint, |block_root, epoch| {

crates/common/chain/beacon/Cargo.toml

@@ -43,6 +43,7 @@ ream-consensus-misc.workspace = true
 ream-execution-engine.workspace = true
 ream-execution-rpc-types.workspace = true
 ream-merkle.workspace = true
+ream-metrics.workspace = true
 ream-network-spec.workspace = true
 
 [dev-dependencies]

crates/common/chain/beacon/src/beacon_chain.rs

@@ -5,6 +5,10 @@ use ream_consensus_misc::{
     polynomial_commitments::{kzg_commitment::KZGCommitment, kzg_proof::KZGProof},
 };
 use ream_merkle::is_valid_merkle_branch;
+use ream_metrics::{
+    BEACON_DATA_COLUMN_SIDECAR_COMPUTATION_SECONDS,
+    BEACON_DATA_COLUMN_SIDECAR_INCLUSION_PROOF_VERIFICATION_SECONDS,
+};
 use ream_network_spec::networks::beacon_network_spec;
 use serde::{Deserialize, Serialize};
 use ssz_derive::{Decode, Encode};
@@ -64,6 +68,7 @@ impl DataColumnSidecar {
 
     /// Verifies that the kzg_commitments list is included in the block body
     pub fn verify_inclusion_proof(&self) -> bool {
+        let _timer = BEACON_DATA_COLUMN_SIDECAR_INCLUSION_PROOF_VERIFICATION_SECONDS.start_timer();
         is_valid_merkle_branch(
             self.kzg_commitments.tree_hash_root(),
             &self.kzg_commitments_inclusion_proof,
@@ -111,6 +116,8 @@ pub fn get_data_column_sidecars(
     kzg_commitments_inclusion_proof: FixedVector<B256, typenum::U4>,
     cells_and_kzg_proofs: Vec<(Vec<Cell>, Vec<KZGProof>)>,
 ) -> Result<Vec<DataColumnSidecar>, DataColumnSidecarError> {
+    let _timer = BEACON_DATA_COLUMN_SIDECAR_COMPUTATION_SECONDS.start_timer();
+
     if cells_and_kzg_proofs.len() != kzg_commitments.len() {
         return Err(DataColumnSidecarError::CommitmentCountMismatch {
             actual: cells_and_kzg_proofs.len(),

crates/common/consensus/beacon/Cargo.toml

@@ -3,6 +3,10 @@ use ream_consensus_misc::{
     constants::beacon::CELLS_PER_EXT_BLOB, polynomial_commitments::kzg_proof::KZGProof,
 };
 use ream_execution_rpc_types::get_blobs::Blob;
+use ream_metrics::{
+    BEACON_DATA_AVAILABILITY_RECONSTRUCTED_COLUMNS_TOTAL,
+    BEACON_DATA_AVAILABILITY_RECONSTRUCTION_TIME_SECONDS,
+};
 use rust_eth_kzg::{Cell as KZGCell, DASContext, KZGProof as Proof};
 use ssz_types::FixedVector;
 
@@ -40,7 +44,9 @@ pub fn recover_matrix(
     blob_count: u64,
     das_context: &DASContext,
 ) -> Result<Vec<MatrixEntry>> {
+    let _timer = BEACON_DATA_AVAILABILITY_RECONSTRUCTION_TIME_SECONDS.start_timer();
     let mut matrix = Vec::new();
+    let mut reconstructed_count: u64 = 0;
 
     for blob_index in 0..blob_count {
         let (cell_indices, cells): (Vec<u64>, Vec<Cell>) = partial_matrix
@@ -49,9 +55,14 @@ pub fn recover_matrix(
             .map(|entry| (entry.column_index, entry.cell.clone()))
             .unzip();
 
+        let known_count = cell_indices.len() as u64;
+
         let (recovered_cells, recovered_proofs) =
             recover_cells_and_kzg_proofs(cell_indices, cells, das_context)?;
 
+        let newly_reconstructed = recovered_cells.len() as u64 - known_count;
+        reconstructed_count += newly_reconstructed;
+
         for (cell_index, (cell, kzg_proof)) in recovered_cells
             .into_iter()
             .zip(recovered_proofs)
@@ -66,6 +77,8 @@ pub fn recover_matrix(
         }
     }
 
+    BEACON_DATA_AVAILABILITY_RECONSTRUCTED_COLUMNS_TOTAL.inc_by(reconstructed_count);
+
     Ok(matrix)
 }