feat: add metrics for intent execution time

magicblock-committor-service/src/intent_execution_manager/intent_execution_engine.rs

@@ -1,4 +1,7 @@
-use std::sync::{Arc, Mutex};
+use std::{
+    sync::{Arc, Mutex},
+    time::{Duration, Instant},
+};
 
 use futures_util::{stream::FuturesUnordered, StreamExt};
 use log::{error, info, trace, warn};
@@ -242,23 +245,27 @@ where
         execution_permit: OwnedSemaphorePermit,
         result_sender: broadcast::Sender<BroadcastedIntentExecutionResult>,
     ) {
+        let instant = Instant::now();
         let result = executor
             .execute(intent.inner.clone(), persister)
             .await
             .inspect_err(|err| {
                 error!(
-                    "Failed to execute BaseIntent. id: {}. {:?}",
+                    "Failed to execute BaseIntent. id: {}. {}",
                     intent.id, err
                 )
-            })
+            });
+
+        // Metrics
+        Self::execution_metrics(instant.elapsed(), &intent, &result);
+
+        let result = result
             .map(|output| ExecutionOutputWrapper {
                 id: intent.id,
                 trigger_type: intent.trigger_type,
                 output,
             })
             .map_err(|err| {
-                // Increase failed intents metric as well
-                metrics::inc_committor_failed_intents_count();
                 (intent.inner.id, intent.trigger_type, Arc::new(err))
             });
 
@@ -280,6 +287,47 @@ where
         // Free worker
         drop(execution_permit);
     }
+
+    /// Records metrics related to intent execution
+    fn execution_metrics(
+        execution_time: Duration,
+        intent: &ScheduledBaseIntentWrapper,
+        result: &IntentExecutorResult<ExecutionOutput>,
+    ) {
+        const EXECUTION_TIME_THRESHOLD: f64 = 2.0;
+
+        let intent_execution_secs = execution_time.as_secs_f64();
+        metrics::observe_committor_intent_execution_time_histogram(
+            intent_execution_secs,
+            intent,
+            result,
+        );
+        if let Err(ref err) = result {
+            metrics::inc_committor_failed_intents_count(intent, err);
+        }
+
+        // Loki alerts
+        if intent_execution_secs >= EXECUTION_TIME_THRESHOLD {
+            info!(
+                "Intent took too long to execute: {}s. {}",
+                intent_execution_secs,
+                result
+                    .as_ref()
+                    .map(|_| "succeeded".to_string())
+                    .unwrap_or_else(|err| format!("{err:?}"))
+            );
+        } else {
+            trace!("Seconds took to execute intent: {}", intent_execution_secs);
+        }
+
+        // Alert
+        if intent.is_undelegate() && result.is_err() {
+            warn!(
+                "Intent execution resulted in stuck accounts: {:?}",
+                intent.get_committed_pubkeys()
+            );
+        }
+    }
 }
 
 /// Worker tests

magicblock-committor-service/src/intent_executor/error.rs

@@ -1,4 +1,5 @@
 use log::error;
+use magicblock_metrics::metrics;
 use magicblock_rpc_client::{
     utils::TransactionErrorMapper, MagicBlockRpcClientError,
 };
@@ -100,6 +101,17 @@ impl IntentExecutorError {
     }
 }
 
+impl metrics::LabelValue for IntentExecutorError {
+    fn value(&self) -> &str {
+        match self {
+            IntentExecutorError::ActionsError(_, _) => "actions_failed",
+            IntentExecutorError::CpiLimitError(_, _) => "cpi_limit_failed",
+            IntentExecutorError::CommitIDError(_, _) => "commit_nonce_failed",
+            _ => "failed",
+        }
+    }
+}
+
 /// Those are the errors that may occur during Commit/Finalize stages on Base layer
 #[derive(thiserror::Error, Debug)]
 pub enum TransactionStrategyExecutionError {

magicblock-committor-service/src/intent_executor/mod.rs

@@ -9,6 +9,7 @@ use std::{ops::ControlFlow, sync::Arc, time::Duration};
 use async_trait::async_trait;
 use futures_util::future::try_join_all;
 use log::{error, trace, warn};
+use magicblock_metrics::metrics;
 use magicblock_program::{
     magic_scheduled_base_intent::ScheduledBaseIntent,
     validator::validator_authority,
@@ -18,8 +19,8 @@ use magicblock_rpc_client::{
         decide_rpc_error_flow, map_magicblock_client_error,
         send_transaction_with_retries, SendErrorMapper, TransactionErrorMapper,
     },
-    MagicBlockSendTransactionConfig, MagicBlockSendTransactionOutcome,
-    MagicblockRpcClient,
+    MagicBlockRpcClientError, MagicBlockSendTransactionConfig,
+    MagicBlockSendTransactionOutcome, MagicblockRpcClient,
 };
 use solana_pubkey::Pubkey;
 use solana_sdk::{
@@ -68,6 +69,18 @@ pub enum ExecutionOutput {
     },
 }
 
+impl metrics::LabelValue for ExecutionOutput {
+    fn value(&self) -> &str {
+        match self {
+            Self::SingleStage(_) => "single_stage_succeeded",
+            Self::TwoStage {
+                commit_signature: _,
+                finalize_signature: _,
+            } => "two_stage_succeeded",
+        }
+    }
+}
+
 #[async_trait]
 pub trait IntentExecutor: Send + Sync + 'static {
     /// Executes Message on Base layer
@@ -663,6 +676,59 @@ where
         )
         .await
     }
+
+    async fn intent_metrics(
+        rpc_client: MagicblockRpcClient,
+        execution_outcome: ExecutionOutput,
+    ) {
+        use solana_transaction_status_client_types::EncodedTransactionWithStatusMeta;
+        fn extract_cu(tx: EncodedTransactionWithStatusMeta) -> Option<u64> {
+            let cu = tx.meta?.compute_units_consumed;
+            cu.into()
+        }
+
+        let cu_metrics = || async {
+            match execution_outcome {
+                ExecutionOutput::SingleStage(signature) => {
+                    let tx =
+                        rpc_client.get_transaction(&signature, None).await?;
+                    Ok::<_, MagicBlockRpcClientError>(extract_cu(
+                        tx.transaction,
+                    ))
+                }
+                ExecutionOutput::TwoStage {
+                    commit_signature,
+                    finalize_signature,
+                } => {
+                    let commit_tx = rpc_client
+                        .get_transaction(&commit_signature, None)
+                        .await?;
+                    let finalize_tx = rpc_client
+                        .get_transaction(&finalize_signature, None)
+                        .await?;
+                    let commit_cu = extract_cu(commit_tx.transaction);
+                    let finalize_cu = extract_cu(finalize_tx.transaction);
+                    let (Some(commit_cu), Some(finalize_cu)) =
+                        (commit_cu, finalize_cu)
+                    else {
+                        return Ok(None);
+                    };
+                    Ok(Some(commit_cu + finalize_cu))
+                }
+            }
+        };
+
+        match cu_metrics().await {
+            Ok(Some(cu)) => metrics::set_commmittor_intent_cu_usage(
+                i64::try_from(cu).unwrap_or(i64::MAX),
+            ),
+            Err(err) => warn!(
+                "Failed to fetch CUs for intent: {:?}. {:?}",
+                err, execution_outcome
+            ),
+            _ => {}
+        }
+    }
 }
 
 #[async_trait]
@@ -692,8 +758,12 @@ where
             // Write result of intent into Persister
             Self::persist_result(&persister, &result, message_id, &pubkeys);
         }
-
-        result
+        result.inspect(|output| {
+            tokio::spawn(Self::intent_metrics(
+                self.rpc_client.clone(),
+                *output,
+            ));
+        })
     }
 }
 

magicblock-committor-service/src/types.rs

@@ -1,6 +1,9 @@
 use std::ops::Deref;
 
-use magicblock_program::magic_scheduled_base_intent::ScheduledBaseIntent;
+use magicblock_metrics::metrics;
+use magicblock_program::magic_scheduled_base_intent::{
+    MagicBaseIntent, ScheduledBaseIntent,
+};
 
 // TODO: should be removed once cranks are supported
 // Ideally even now OffChain/"Manual" commits should be triggered via Tx
@@ -16,6 +19,16 @@ pub struct ScheduledBaseIntentWrapper {
     pub trigger_type: TriggerType,
 }
 
+impl metrics::LabelValue for ScheduledBaseIntentWrapper {
+    fn value(&self) -> &str {
+        match &self.inner.base_intent {
+            MagicBaseIntent::BaseActions(_) => "actions",
+            MagicBaseIntent::Commit(_) => "commit",
+            MagicBaseIntent::CommitAndUndelegate(_) => "commit_and_undelegate",
+        }
+    }
+}
+
 impl Deref for ScheduledBaseIntentWrapper {
     type Target = ScheduledBaseIntent;
 

magicblock-metrics/src/metrics/mod.rs

@@ -5,7 +5,8 @@ use prometheus::{
     Histogram, HistogramOpts, HistogramVec, IntCounter, IntCounterVec,
     IntGauge, IntGaugeVec, Opts, Registry,
 };
-pub use types::{AccountClone, AccountCommit, Outcome};
+pub use types::{AccountClone, AccountCommit, LabelValue, Outcome};
+
 mod types;
 
 // -----------------
@@ -181,13 +182,32 @@ lazy_static::lazy_static! {
         "committor_intent_backlog_count", "Number of intents in backlog",
     ).unwrap();
 
-    static ref COMMITTOR_FAILED_INTENTS_COUNT: IntCounter = IntCounter::new(
-        "committor_failed_intents_count", "Number of failed to be executed intents",
+    static ref COMMITTOR_FAILED_INTENTS_COUNT: IntCounterVec = IntCounterVec::new(
+        Opts::new("committor_failed_intents_count", "Number of failed to be executed intents"),
+        &["intent_kind", "error_kind"]
     ).unwrap();
 
     static ref COMMITTOR_EXECUTORS_BUSY_COUNT: IntGauge = IntGauge::new(
         "committor_executors_busy_count", "Number of busy intent executors"
     ).unwrap();
+
+    static ref COMMITTOR_INTENT_EXECUTION_TIME_HISTOGRAM: HistogramVec = HistogramVec::new(
+        HistogramOpts::new(
+            "committor_intent_execution_time_histogram",
+            "Time in seconds spent on intent execution"
+        )
+        .buckets(
+            MILLIS_1_9.iter()
+            .chain(MILLIS_10_90.iter())
+            .chain(MILLIS_100_900.iter())
+            .chain(SECONDS_1_9.iter()).cloned().collect(),
+        ),
+        &["intent_kind", "outcome_kind"],
+    ).unwrap();
+
+    static ref COMMITTOR_INTENT_CU_USAGE: IntGauge = IntGauge::new(
+        "committor_intent_cu_usage", "Compute units used for Intent"
+    ).unwrap();
 }
 
 pub(crate) fn register() {
@@ -223,6 +243,7 @@ pub(crate) fn register() {
         register!(COMMITTOR_INTENTS_BACKLOG_COUNT);
         register!(COMMITTOR_FAILED_INTENTS_COUNT);
         register!(COMMITTOR_EXECUTORS_BUSY_COUNT);
+        register!(COMMITTOR_INTENT_EXECUTION_TIME_HISTOGRAM);
         register!(ENSURE_ACCOUNTS_TIME);
         register!(RPC_REQUEST_HANDLING_TIME);
         register!(TRANSACTION_PROCESSING_TIME);
@@ -318,10 +339,29 @@ pub fn set_committor_intents_backlog_count(value: i64) {
     COMMITTOR_INTENTS_BACKLOG_COUNT.set(value)
 }
 
-pub fn inc_committor_failed_intents_count() {
-    COMMITTOR_FAILED_INTENTS_COUNT.inc()
+pub fn inc_committor_failed_intents_count(
+    intent_kind: &impl LabelValue,
+    error_kind: &impl LabelValue,
+) {
+    COMMITTOR_FAILED_INTENTS_COUNT
+        .with_label_values(&[intent_kind.value(), error_kind.value()])
+        .inc()
 }
 
 pub fn set_committor_executors_busy_count(value: i64) {
     COMMITTOR_EXECUTORS_BUSY_COUNT.set(value)
 }
+
+pub fn observe_committor_intent_execution_time_histogram(
+    seconds: f64,
+    kind: &impl LabelValue,
+    outcome: &impl LabelValue,
+) {
+    COMMITTOR_INTENT_EXECUTION_TIME_HISTOGRAM
+        .with_label_values(&[kind.value(), outcome.value()])
+        .observe(seconds);
+}
+
+pub fn set_commmittor_intent_cu_usage(value: i64) {
+    COMMITTOR_INTENT_CU_USAGE.set(value)
+}

magicblock-metrics/src/metrics/types.rs

@@ -68,3 +68,20 @@ pub enum AccountCommit<'a> {
     CommitOnly { pubkey: &'a str, outcome: Outcome },
     CommitAndUndelegate { pubkey: &'a str, outcome: Outcome },
 }
+
+pub trait LabelValue {
+    fn value(&self) -> &str;
+}
+
+impl<T, E> LabelValue for Result<T, E>
+where
+    T: LabelValue,
+    E: LabelValue,
+{
+    fn value(&self) -> &str {
+        match self {
+            Ok(ok) => ok.value(),
+            Err(err) => err.value(),
+        }
+    }
+}