Rpc/execution related metrics

Cargo.toml

@@ -91,7 +91,7 @@ jsonrpc-pubsub = "18.0.0"
 jsonrpc-ws-server = "18.0.0"
 lazy_static = "1.4.0"
 libc = "0.2.153"
-log = "0.4.20"
+log = { version = "0.4.20", features = ["release_max_level_info"] }
 lru = "0.16.0"
 macrotest = "1"
 magic-domain-program = { git = "https://github.com/magicblock-labs/magic-domain-program.git", rev = "ea04d46", default-features = false }

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

@@ -10,6 +10,7 @@ use log::*;
 use magicblock_core::{
     link::transactions::SanitizeableTransaction, traits::AccountsBank,
 };
+use magicblock_metrics::metrics::ENSURE_ACCOUNTS_TIME;
 use prelude::JsonBody;
 use solana_account::AccountSharedData;
 use solana_pubkey::Pubkey;
@@ -101,15 +102,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)
     }
@@ -120,22 +123,18 @@ 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}");
-        }
+        trace!("Ensuring accounts {pubkeys:?}");
+        let _timer = ENSURE_ACCOUNTS_TIME
+            .with_label_values(&["multi-account"])
+            .start_timer();
         let _ = self
             .chainlink
             .ensure_accounts(pubkeys, None)
             .await
             .inspect_err(|e| {
                 // There is nothing we can do if fetching the accounts fails
                 // Log the error and return whatever is in the accounts db
-                error!("Failed to ensure accounts: {e}");
+                warn!("Failed to ensure accounts: {e}");
             });
         pubkeys
             .iter()
@@ -195,6 +194,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,7 @@
-use log::*;
+use log::{trace, warn};
+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 +18,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);
@@ -33,16 +38,16 @@ 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 {
-            trace!("Scheduling transaction: {signature}");
+            TRANSACTION_SKIP_PREFLIGHT.inc();
             self.transactions_scheduler.schedule(transaction).await?;
+            trace!("Scheduling transaction {signature}");
         } else {
-            trace!("Executing transaction: {signature}");
+            trace!("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

@@ -2,6 +2,7 @@ use std::collections::HashMap;
 
 use hyper::body::Bytes;
 use json::{Serialize, Value};
+use magicblock_metrics::metrics::RPC_REQUESTS_COUNT;
 use tokio::sync::mpsc;
 
 use super::connection::ConnectionID;
@@ -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

@@ -15,6 +15,7 @@ use magicblock_core::{
     },
     Slot,
 };
+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-subscribe");
         // 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-subscribe");
         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-subscribe");
         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-subscribe");
         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();
+    }
+}