refactor: move metrics into their owning crates

Cargo.toml

@@ -12,7 +12,6 @@ members = [
     "bins/validator-tui",
     "bins/verifier",
     "aperture",
-    "metrics",
     "chainsync",
     "programs/committor-interface",
     "programs/committor-program",
@@ -62,7 +61,6 @@ unwrap_used = "warn"
 wildcard_dependencies = "deny"
 
 [workspace.dependencies]
-metrics = { package = "magicblock-metrics", path = "metrics" }
 aperture = { package = "magicblock-aperture", path = "aperture" }
 chainsync = { package = "magicblock-chainsync", path = "chainsync" }
 committor-interface = { package = "magicblock-committor-interface", path = "programs/committor-interface" }

aperture/Cargo.toml

@@ -18,7 +18,7 @@ hyper-util = { workspace = true, features = ["http1", "http2", "server"] }
 futures = { workspace = true }
 libloading = { workspace = true }
 num_cpus = { workspace = true }
-tokio = { workspace = true }
+tokio = { workspace = true, features = ["rt-multi-thread"] }
 tokio-util = { workspace = true }
 
 # magicblock
@@ -28,8 +28,8 @@ chainsync = { workspace = true }
 config = { workspace = true }
 mbv-core = { workspace = true }
 ledger-deprecated = { workspace = true }
-metrics = { workspace = true }
-nucleus = { workspace = true, features = ["runtime"] }
+prometheus = { workspace = true }
+nucleus = { workspace = true, features = ["metrics", "runtime"] }
 
 # solana
 agave-geyser-plugin-interface = { workspace = true }

aperture/src/account.rs

@@ -1,7 +1,6 @@
 use chainsync::{AccountProperty, ChainSync, ChainSyncAccount, Result as ChainSyncResult};
 use engine::Engine;
 use json::{Deserialize, Serialize};
-use metrics::metrics::ENSURE_ACCOUNTS_TIME;
 use nucleus::runtime::TransactionView;
 use serde::{Deserializer, de::Error as _};
 use solana_account::{AccountMode, AccountSharedData};
@@ -72,7 +71,6 @@ impl Accounts {
         fetches: &mut u64,
     ) -> RpcResult<TransactionView> {
         let transaction = decode_transaction(transaction, encoding)?;
-        let _timer = ENSURE_ACCOUNTS_TIME.with_label_values(&["transaction"]).start_timer();
         *fetches += self
             .chainsync
             .ensure_transaction_accounts(&transaction)
@@ -92,7 +90,6 @@ impl Accounts {
 
     /// Account for successful synchronization before any later projection or admission error.
     async fn fetch(&self, pubkeys: &[Pubkey], fetches: &mut u64) -> ChainSyncResult<()> {
-        let _timer = ENSURE_ACCOUNTS_TIME.with_label_values(&["account"]).start_timer();
         let requests = pubkeys.iter().map(|&pubkey| ChainSyncAccount {
             pubkey,
             property: AccountProperty::Readonly,

aperture/src/lib.rs

@@ -126,3 +126,8 @@ mod requests;
 mod server;
 mod state;
 mod transaction;
+
+mod metrics;
+
+/// Standalone Prometheus scrape server for leaders and verifiers.
+pub use server::metrics::MetricsService;

aperture/src/metrics.rs

@@ -0,0 +1,49 @@
+//! Event-driven aperture metrics registered in the default Prometheus registry.
+
+use nucleus::metrics::{
+    HistogramOpts, HistogramTimer, HistogramVec, IntCounter, IntCounterVec, Opts, counter,
+    counter_vec, histogram_vec,
+};
+use std::sync::OnceLock;
+
+static METRICS: OnceLock<Metrics> = OnceLock::new();
+
+struct Metrics {
+    requests: IntCounterVec,
+    duration: HistogramVec,
+    skip_preflight: IntCounter,
+}
+
+fn metrics() -> &'static Metrics {
+    METRICS.get_or_init(|| Metrics {
+        requests: counter_vec(
+            Opts::new("aperture_rpc_requests_total", "RPC requests by method"),
+            &["method"],
+        ),
+        duration: histogram_vec(
+            HistogramOpts::new(
+                "aperture_rpc_request_duration_seconds",
+                "HTTP RPC request duration in seconds",
+            )
+            .buckets(vec![0.001, 0.01, 0.1, 1.0, 5.0, 10.0, 30.0, 60.0]),
+            &["method"],
+        ),
+        skip_preflight: counter(
+            Opts::new(
+                "aperture_transactions_skip_preflight_total",
+                "Transactions submitted with skipPreflight",
+            ),
+            0,
+        ),
+    })
+}
+
+pub(crate) fn request(method: &str) {
+    metrics().requests.with_label_values(&[method]).inc();
+}
+pub(crate) fn time_request(method: &str) -> HistogramTimer {
+    metrics().duration.with_label_values(&[method]).start_timer()
+}
+pub(crate) fn skip_preflight() {
+    metrics().skip_preflight.inc();
+}

aperture/src/requests/http/dispatch.rs

@@ -1,8 +1,8 @@
 use super::{
     HandlerResult, RpcHandlers, get_blocks::BlockRange, get_token_accounts::TokenAccountAuthority,
 };
+use crate::metrics;
 use crate::{error::RpcError, requests::JsonHttpRequest};
-use ::metrics::metrics::{RPC_REQUEST_HANDLING_TIME, RPC_REQUESTS_COUNT};
 
 impl RpcHandlers {
     pub(crate) async fn process(
@@ -13,8 +13,8 @@ impl RpcHandlers {
         // Route the request to the correct handler based on the method name.
         use crate::requests::JsonRpcHttpMethod::*;
         let method = request.method.as_str();
-        RPC_REQUESTS_COUNT.with_label_values(&[method]).inc();
-        let _timer = RPC_REQUEST_HANDLING_TIME.with_label_values(&[method]).start_timer();
+        metrics::request(method);
+        let _timer = metrics::time_request(method);
 
         match request.method {
             GetAccountInfo => self.get_account_info(request, fetches).await,

aperture/src/requests/http/send_transaction.rs

@@ -1,8 +1,8 @@
 use super::{HandlerResult, RpcHandlers};
+use crate::metrics;
 use crate::requests::{
     JsonHttpRequest as JsonRequest, params::SerdeSignature, payload::ResponsePayload,
 };
-use ::metrics::metrics::{TRANSACTION_PROCESSING_TIME, TRANSACTION_SKIP_PREFLIGHT};
 use solana_rpc_client_api::config::RpcSendTransactionConfig;
 use solana_transaction_status::UiTransactionEncoding;
 impl RpcHandlers {
@@ -11,7 +11,6 @@ impl RpcHandlers {
         request: &JsonRequest,
         fetches: &mut u64,
     ) -> HandlerResult {
-        let _timer = TRANSACTION_PROCESSING_TIME.start_timer();
         let transaction_str = request.required::<&str>(0)?;
         let config = request.optional::<RpcSendTransactionConfig>(1)?.unwrap_or_default();
         let encoding = config.encoding.unwrap_or(UiTransactionEncoding::Base58);
@@ -21,7 +20,7 @@ impl RpcHandlers {
         let signature = transaction.signatures()[0];
 
         if config.skip_preflight {
-            TRANSACTION_SKIP_PREFLIGHT.inc();
+            metrics::skip_preflight();
             self.engine.transaction(transaction)?.schedule().await?;
         } else {
             self.engine.transaction(transaction)?.execute().await??;

aperture/src/server/metrics.rs

@@ -0,0 +1,99 @@
+use std::net::SocketAddr;
+
+use http_body_util::{BodyExt, Empty, Full, combinators::BoxBody};
+use hyper::{
+    Method, Request, Response, StatusCode, body::Bytes, server::conn::http1, service::service_fn,
+};
+use hyper_util::rt::TokioIo;
+use nucleus::shutdown::{ShutdownHandle, ShutdownReason};
+use prometheus::TextEncoder;
+use tokio::{net::TcpListener, select, task::JoinSet};
+use tracing::{info, warn};
+
+pub struct MetricsService {
+    listener: TcpListener,
+}
+
+impl MetricsService {
+    pub async fn bind(addr: SocketAddr) -> std::io::Result<Self> {
+        let listener = TcpListener::bind(addr).await?;
+        Ok(Self { listener })
+    }
+
+    /// Serves metrics until its shutdown tier is cancelled.
+    pub async fn run(self, mut shutdown: ShutdownHandle) {
+        let reason = match serve(self.listener, &shutdown).await {
+            Err(error) => ShutdownReason::Error(Box::new(error)),
+            Ok(()) if shutdown.requested() => ShutdownReason::Signalled,
+            Ok(()) => ShutdownReason::Unexpected,
+        };
+        shutdown.terminate(reason);
+    }
+}
+
+async fn serve(listener: TcpListener, shutdown: &ShutdownHandle) -> std::io::Result<()> {
+    info!("Metrics server started");
+    let mut connections = JoinSet::new();
+
+    let result = loop {
+        select!(
+            _ = shutdown.signalled() => {
+                break Ok(());
+            }
+            _ = connections.join_next(), if !connections.is_empty() => {}
+            result = listener.accept() => {
+                let (stream, _) = match result {
+                    Ok(connection) => connection,
+                    Err(error) => break Err(error),
+                };
+                let io = TokioIo::new(stream);
+                connections.spawn(async move {
+                    let con = http1::Builder::new();
+                    let _ = con.serve_connection(io, service_fn(metrics_service_router)).await;
+                });
+            }
+        );
+    };
+
+    connections.shutdown().await;
+    info!("Metrics server shutdown");
+    result
+}
+
+async fn metrics_service_router(
+    req: Request<hyper::body::Incoming>,
+) -> Result<Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> {
+    // Record optional headers
+    let result = match (req.method(), req.uri().path()) {
+        (&Method::GET, "/metrics") => {
+            let metric_families = prometheus::gather();
+            let metrics =
+                TextEncoder::new().encode_to_string(&metric_families).unwrap_or_else(|error| {
+                    warn!(error = %error, "Failed to encode metrics");
+                    String::new()
+                });
+            Ok(Response::new(full(metrics)))
+        }
+        _ => {
+            let mut not_found = Response::new(empty());
+            *not_found.status_mut() = StatusCode::NOT_FOUND;
+            Ok(not_found)
+        }
+    };
+    // We must consume the body fully to keep the connection alive. We
+    // iterate over all chunks and simply drop them. This prevents garbage
+    // data of previous requests from being stuck in connection buffer.
+    let mut body = req.into_body();
+    while (body.frame().await).is_some() {}
+
+    result
+}
+
+fn full<T: Into<Bytes>>(chunk: T) -> BoxBody<Bytes, hyper::Error> {
+    Full::new(chunk.into()).map_err(|never| match never {}).boxed()
+}
+
+fn empty() -> BoxBody<Bytes, hyper::Error> {
+    let map_err = Empty::<Bytes>::new().map_err(|never| match never {});
+    map_err.boxed()
+}

aperture/src/server/mod.rs

@@ -1,2 +1,4 @@
 pub(crate) mod http;
 pub(crate) mod websocket;
+
+pub(crate) mod metrics;

aperture/src/server/websocket/dispatch.rs

@@ -1,6 +1,6 @@
 use std::collections::HashMap;
 
-use ::metrics::metrics::RPC_REQUESTS_COUNT;
+use crate::metrics;
 use engine::Engine;
 use hyper::body::Bytes;
 use json::Serialize;
@@ -48,7 +48,7 @@ impl WsDispatcher {
     /// Routes an incoming JSON-RPC request to the appropriate subscription handler.
     pub(crate) async fn dispatch(&mut self, request: &JsonWsRequest) -> RpcResult<SubResult> {
         use JsonRpcWsMethod::*;
-        RPC_REQUESTS_COUNT.with_label_values(&[request.method.as_str()]).inc();
+        metrics::request(request.method.as_str());
         match request.method {
             AccountSubscribe => self.account_subscribe(request),
             ProgramSubscribe => self.program_subscribe(request),

bins/validator/Cargo.toml

@@ -19,13 +19,12 @@ config = { workspace = true }
 mbv-core = { workspace = true }
 delegation-program-api = { workspace = true }
 ledger-deprecated = { workspace = true }
-metrics = { workspace = true }
 program = { workspace = true }
 runtime = { workspace = true }
 services = { workspace = true }
 task-scheduler = { workspace = true }
 mimalloc = { workspace = true }
-nucleus = { workspace = true }
+nucleus = { workspace = true, features = ["metrics"] }
 num_cpus = { workspace = true }
 replicator = { workspace = true }
 solana-commitment-config = { workspace = true }