feat: add RPC specific metrics

magicblock-account-cloner/src/remote_account_cloner_worker.rs

@@ -789,17 +789,11 @@ where
         pubkey: &Pubkey,
         lamports: u64,
         owner: &Pubkey,
-        balance_pda: Option<&Pubkey>,
+        _balance_pda: Option<&Pubkey>,
     ) -> AccountClonerResult<Signature> {
         self.account_dumper
             .dump_feepayer_account(pubkey, lamports, owner)
             .map_err(AccountClonerError::AccountDumperError)
-            .inspect(|_| {
-                metrics::inc_account_clone(metrics::AccountClone::FeePayer {
-                    pubkey: &pubkey.to_string(),
-                    balance_pda: balance_pda.map(|p| p.to_string()).as_deref(),
-                });
-            })
     }
 
     fn do_clone_undelegated_account(
@@ -810,14 +804,6 @@ where
         self.account_dumper
             .dump_undelegated_account(pubkey, account)
             .map_err(AccountClonerError::AccountDumperError)
-            .inspect(|_| {
-                metrics::inc_account_clone(
-                    metrics::AccountClone::Undelegated {
-                        pubkey: &pubkey.to_string(),
-                        owner: &account.owner().to_string(),
-                    },
-                );
-            })
     }
 
     fn do_clone_delegated_account(
@@ -846,13 +832,6 @@ where
         self.account_dumper
             .dump_delegated_account(pubkey, account, &record.owner)
             .map_err(AccountClonerError::AccountDumperError)
-            .inspect(|_| {
-                metrics::inc_account_clone(metrics::AccountClone::Delegated {
-                    // TODO(bmuddha): optimize metrics, remove .to_string()
-                    pubkey: &pubkey.to_string(),
-                    owner: &record.owner.to_string(),
-                });
-            })
     }
 
     async fn do_clone_program_accounts(
@@ -908,11 +887,6 @@ where
                 idl_account,
             )
             .map_err(AccountClonerError::AccountDumperError)
-            .inspect(|_| {
-                metrics::inc_account_clone(metrics::AccountClone::Program {
-                    pubkey: &pubkey.to_string(),
-                });
-            })
     }
 
     async fn fetch_program_idl(

magicblock-aperture/Cargo.toml

@@ -33,6 +33,7 @@ magicblock-chainlink = { workspace = true }
 magicblock-config = { workspace = true }
 magicblock-core = { workspace = true }
 magicblock-ledger = { workspace = true }
+magicblock-metrics = { workspace = true }
 magicblock-version = { workspace = true }
 
 # solana

magicblock-aperture/src/requests/http/mod.rs

@@ -1,5 +1,6 @@
 use log::*;
 use magicblock_core::traits::AccountsBank;
+use magicblock_metrics::metrics::ENSURE_ACCOUNTS_TIME;
 use std::{mem::size_of, ops::Range};
 
 use base64::{prelude::BASE64_STANDARD, Engine};
@@ -85,15 +86,17 @@ impl HttpDispatcher {
         &self,
         pubkey: &Pubkey,
     ) -> Option<AccountSharedData> {
-        debug!("Ensuring account {pubkey}");
+        let _timer = ENSURE_ACCOUNTS_TIME
+            .with_label_values(&["account"])
+            .start_timer();
         let _ = self
             .chainlink
             .ensure_accounts(&[*pubkey], None)
             .await
             .inspect_err(|e| {
                 // There is nothing we can do if fetching the account fails
                 // Log the error and return whatever is in the accounts db
-                error!("Failed to ensure account {pubkey}: {e}");
+                warn!("Failed to ensure account {pubkey}: {e}");
             });
         self.accountsdb.get_account(pubkey)
     }
@@ -104,14 +107,9 @@ impl HttpDispatcher {
         &self,
         pubkeys: &[Pubkey],
     ) -> Vec<Option<AccountSharedData>> {
-        if log::log_enabled!(log::Level::Debug) {
-            let pubkeys = pubkeys
-                .iter()
-                .map(ToString::to_string)
-                .collect::<Vec<_>>()
-                .join(", ");
-            debug!("Ensuring accounts {pubkeys}");
-        }
+        let _timer = ENSURE_ACCOUNTS_TIME
+            .with_label_values(&["multi-account"])
+            .start_timer();
         let _ = self
             .chainlink
             .ensure_accounts(pubkeys, None)
@@ -177,6 +175,9 @@ impl HttpDispatcher {
         &self,
         transaction: &SanitizedTransaction,
     ) -> RpcResult<()> {
+        let _timer = ENSURE_ACCOUNTS_TIME
+            .with_label_values(&["transaction"])
+            .start_timer();
         match self
             .chainlink
             .ensure_transaction_accounts(transaction)

magicblock-aperture/src/requests/http/send_transaction.rs

@@ -1,4 +1,6 @@
-use log::*;
+use magicblock_metrics::metrics::{
+    TRANSACTION_PROCESSING_TIME, TRANSACTION_SKIP_PREFLIGHT,
+};
 use solana_rpc_client_api::config::RpcSendTransactionConfig;
 use solana_transaction_error::TransactionError;
 use solana_transaction_status::UiTransactionEncoding;
@@ -15,8 +17,10 @@ impl HttpDispatcher {
         &self,
         request: &mut JsonRequest,
     ) -> HandlerResult {
+        let _timer = TRANSACTION_PROCESSING_TIME.start_timer();
         let (transaction_str, config) =
             parse_params!(request.params()?, String, RpcSendTransactionConfig);
+
         let transaction_str: String = some_or_err!(transaction_str);
         let config = config.unwrap_or_default();
         let encoding = config.encoding.unwrap_or(UiTransactionEncoding::Base58);
@@ -32,16 +36,14 @@ impl HttpDispatcher {
         {
             return Err(TransactionError::AlreadyProcessed.into());
         }
-        debug!("Received transaction: {signature}, ensuring accounts");
         self.ensure_transaction_accounts(&transaction).await?;
 
         // Based on the preflight flag, either execute and await the result,
         // or schedule (fire-and-forget) for background processing.
         if config.skip_preflight {
-            debug!("Scheduling transaction: {signature}");
+            TRANSACTION_SKIP_PREFLIGHT.inc();
             self.transactions_scheduler.schedule(transaction).await?;
         } else {
-            debug!("Executing transaction: {signature}");
             self.transactions_scheduler.execute(transaction).await?;
         }
 

magicblock-aperture/src/requests/mod.rs

@@ -88,6 +88,84 @@ pub(crate) enum JsonRpcWsMethod {
     SlotUnsubscribe,
 }
 
+impl JsonRpcHttpMethod {
+    pub(crate) fn as_str(&self) -> &'static str {
+        match self {
+            JsonRpcHttpMethod::GetAccountInfo => "getAccountInfo",
+            JsonRpcHttpMethod::GetBalance => "getBalance",
+            JsonRpcHttpMethod::GetBlock => "getBlock",
+            JsonRpcHttpMethod::GetBlockCommitment => "getBlockCommitment",
+            JsonRpcHttpMethod::GetBlockHeight => "getBlockHeight",
+            JsonRpcHttpMethod::GetBlockTime => "getBlockTime",
+            JsonRpcHttpMethod::GetBlocks => "getBlocks",
+            JsonRpcHttpMethod::GetBlocksWithLimit => "getBlocksWithLimit",
+            JsonRpcHttpMethod::GetClusterNodes => "getClusterNodes",
+            JsonRpcHttpMethod::GetEpochInfo => "getEpochInfo",
+            JsonRpcHttpMethod::GetEpochSchedule => "getEpochSchedule",
+            JsonRpcHttpMethod::GetFeeForMessage => "getFeeForMessage",
+            JsonRpcHttpMethod::GetFirstAvailableBlock => {
+                "getFirstAvailableBlock"
+            }
+            JsonRpcHttpMethod::GetGenesisHash => "getGenesisHash",
+            JsonRpcHttpMethod::GetHealth => "getHealth",
+            JsonRpcHttpMethod::GetHighestSnapshotSlot => {
+                "getHighestSnapshotSlot"
+            }
+            JsonRpcHttpMethod::GetIdentity => "getIdentity",
+            JsonRpcHttpMethod::GetLargestAccounts => "getLargestAccounts",
+            JsonRpcHttpMethod::GetLatestBlockhash => "getLatestBlockhash",
+            JsonRpcHttpMethod::GetMultipleAccounts => "getMultipleAccounts",
+            JsonRpcHttpMethod::GetProgramAccounts => "getProgramAccounts",
+            JsonRpcHttpMethod::GetSignatureStatuses => "getSignatureStatuses",
+            JsonRpcHttpMethod::GetSignaturesForAddress => {
+                "getSignaturesForAddress"
+            }
+            JsonRpcHttpMethod::GetSlot => "getSlot",
+            JsonRpcHttpMethod::GetSlotLeader => "getSlotLeader",
+            JsonRpcHttpMethod::GetSlotLeaders => "getSlotLeaders",
+            JsonRpcHttpMethod::GetSupply => "getSupply",
+            JsonRpcHttpMethod::GetTokenAccountBalance => {
+                "getTokenAccountBalance"
+            }
+            JsonRpcHttpMethod::GetTokenAccountsByDelegate => {
+                "getTokenAccountsByDelegate"
+            }
+            JsonRpcHttpMethod::GetTokenAccountsByOwner => {
+                "getTokenAccountsByOwner"
+            }
+            JsonRpcHttpMethod::GetTokenLargestAccounts => {
+                "getTokenLargestAccounts"
+            }
+            JsonRpcHttpMethod::GetTokenSupply => "getTokenSupply",
+            JsonRpcHttpMethod::GetTransaction => "getTransaction",
+            JsonRpcHttpMethod::GetTransactionCount => "getTransactionCount",
+            JsonRpcHttpMethod::GetVersion => "getVersion",
+            JsonRpcHttpMethod::IsBlockhashValid => "isBlockhashValid",
+            JsonRpcHttpMethod::MinimumLedgerSlot => "minimumLedgerSlot",
+            JsonRpcHttpMethod::RequestAirdrop => "requestAirdrop",
+            JsonRpcHttpMethod::SendTransaction => "sendTransaction",
+            JsonRpcHttpMethod::SimulateTransaction => "simulateTransaction",
+        }
+    }
+}
+
+impl JsonRpcWsMethod {
+    pub(crate) fn as_str(&self) -> &'static str {
+        match self {
+            JsonRpcWsMethod::AccountSubscribe => "accountSubscribe",
+            JsonRpcWsMethod::AccountUnsubscribe => "accountUnsubscribe",
+            JsonRpcWsMethod::LogsSubscribe => "logsSubscribe",
+            JsonRpcWsMethod::LogsUnsubscribe => "logsUnsubscribe",
+            JsonRpcWsMethod::ProgramSubscribe => "programSubscribe",
+            JsonRpcWsMethod::ProgramUnsubscribe => "programUnsubscribe",
+            JsonRpcWsMethod::SignatureSubscribe => "signatureSubscribe",
+            JsonRpcWsMethod::SignatureUnsubscribe => "signatureUnsubscribe",
+            JsonRpcWsMethod::SlotSubscribe => "slotSubscribe",
+            JsonRpcWsMethod::SlotUnsubscribe => "slotUnsubscribe",
+        }
+    }
+}
+
 /// A helper macro for easily parsing positional parameters from a JSON-RPC request.
 ///
 /// This macro simplifies the process of extracting and deserializing parameters

magicblock-aperture/src/server/http/dispatch.rs

@@ -6,6 +6,9 @@ use magicblock_core::link::{
     transactions::TransactionSchedulerHandle, DispatchEndpoints,
 };
 use magicblock_ledger::Ledger;
+use magicblock_metrics::metrics::{
+    RPC_REQUESTS_COUNT, RPC_REQUEST_HANDLING_TIME,
+};
 
 use crate::{
     requests::{
@@ -111,6 +114,11 @@ impl HttpDispatcher {
     async fn process(&self, request: &mut JsonHttpRequest) -> HandlerResult {
         // 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();
         match request.method {
             GetAccountInfo => self.get_account_info(request).await,
             GetBalance => self.get_balance(request).await,

magicblock-aperture/src/server/websocket/dispatch.rs

@@ -15,6 +15,7 @@ use crate::{
 use super::connection::ConnectionID;
 use hyper::body::Bytes;
 use json::{Serialize, Value};
+use magicblock_metrics::metrics::RPC_REQUESTS_COUNT;
 use tokio::sync::mpsc;
 
 /// The sender half of an MPSC channel used to push subscription notifications
@@ -65,6 +66,9 @@ impl WsDispatcher {
         request: &mut JsonWsRequest,
     ) -> RpcResult<WsDispatchResult> {
         use JsonRpcWsMethod::*;
+        RPC_REQUESTS_COUNT
+            .with_label_values(&[request.method.as_str()])
+            .inc();
         let result = match request.method {
             AccountSubscribe => self.account_subscribe(request).await,
             ProgramSubscribe => self.program_subscribe(request).await,

magicblock-aperture/src/state/subscriptions.rs

@@ -8,6 +8,7 @@ use std::{
     },
 };
 
+use magicblock_metrics::metrics::RPC_WS_SUBSCRIPTIONS_COUNT;
 use parking_lot::RwLock;
 use solana_account::ReadableAccount;
 use solana_pubkey::Pubkey;
@@ -105,7 +106,7 @@ impl SubscriptionsDb {
             .await
             .or_insert_with(|| UpdateSubscribers(vec![]))
             .add_subscriber(chan, encoder.clone());
-
+        let metric = SubMetricGuard::new("account");
         // Create a cleanup future that will be executed when the handle is dropped.
         let accounts = self.accounts.clone();
         let callback = async move {
@@ -116,6 +117,7 @@ impl SubscriptionsDb {
             if entry.remove_subscriber(conid, &encoder) {
                 let _ = entry.remove();
             }
+            drop(metric)
         };
         let cleanup = CleanUp(Some(Box::pin(callback)));
         SubscriptionHandle { id, cleanup }
@@ -146,13 +148,15 @@ impl SubscriptionsDb {
             .add_subscriber(chan, encoder.clone());
 
         let programs = self.programs.clone();
+        let metric = SubMetricGuard::new("program");
         let callback = async move {
             let Some(mut entry) = programs.get_async(&pubkey).await else {
                 return;
             };
             if entry.remove_subscriber(conid, &encoder) {
                 let _ = entry.remove();
             }
+            drop(metric)
         };
         let cleanup = CleanUp(Some(Box::pin(callback)));
         SubscriptionHandle { id, cleanup }
@@ -212,8 +216,10 @@ impl SubscriptionsDb {
         let id = self.logs.write().add_subscriber(chan, encoder.clone());
 
         let logs = self.logs.clone();
+        let metric = SubMetricGuard::new("logs");
         let callback = async move {
             logs.write().remove_subscriber(conid, &encoder);
+            drop(metric)
         };
         let cleanup = CleanUp(Some(Box::pin(callback)));
         SubscriptionHandle { id, cleanup }
@@ -239,8 +245,10 @@ impl SubscriptionsDb {
         let id = subscriber.id;
 
         let slot = self.slot.clone();
+        let metric = SubMetricGuard::new("slot");
         let callback = async move {
             slot.write().txs.remove(&conid);
+            drop(metric)
         };
         let cleanup = CleanUp(Some(Box::pin(callback)));
         SubscriptionHandle { id, cleanup }
@@ -391,3 +399,20 @@ impl<E> Drop for UpdateSubscriber<E> {
         self.live.store(false, Ordering::Relaxed);
     }
 }
+
+pub(crate) struct SubMetricGuard(&'static str);
+
+impl SubMetricGuard {
+    pub(crate) fn new(name: &'static str) -> Self {
+        RPC_WS_SUBSCRIPTIONS_COUNT.with_label_values(&[name]).inc();
+        Self(name)
+    }
+}
+
+impl Drop for SubMetricGuard {
+    fn drop(&mut self) {
+        RPC_WS_SUBSCRIPTIONS_COUNT
+            .with_label_values(&[self.0])
+            .dec();
+    }
+}