feat: add request duration tracking for Fortuna callbacks (#2197)

* feat: add request_duration_ms histogram metric to track callback processing time

Co-Authored-By: Jayant Krishnamurthy <jayant@dourolabs.xyz>

* feat: implement request duration tracking in process_event_with_backoff

Co-Authored-By: Jayant Krishnamurthy <jayant@dourolabs.xyz>

---------

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: Jayant Krishnamurthy <jayant@dourolabs.xyz>

apps/fortuna/Cargo.lock

@@ -1503,7 +1503,7 @@ dependencies = [
 
 [[package]]
 name = "fortuna"
-version = "6.6.0"
+version = "6.7.0"
 dependencies = [
  "anyhow",
  "axum",

apps/fortuna/Cargo.toml

@@ -1,40 +1,43 @@
 [package]
-name    = "fortuna"
-version = "6.6.0"
+name = "fortuna"
+version = "6.7.0"
 edition = "2021"
 
 [dependencies]
-anyhow      = "1.0.75"
-axum        = { version = "0.6.20", features = ["json", "ws", "macros"] }
-axum-macros        = { version = "0.3.8" }
-base64             = { version = "0.21.0" }
+anyhow = "1.0.75"
+axum = { version = "0.6.20", features = ["json", "ws", "macros"] }
+axum-macros = { version = "0.3.8" }
+base64 = { version = "0.21.0" }
 bincode = "1.3.3"
-byteorder   = "1.5.0"
-clap        = { version = "4.4.6", features = ["derive", "cargo", "env"] }
-ethabi      = "18.0.0"
-ethers      = { version = "2.0.14", features = ["ws"] }
-futures            = { version = "0.3.28" }
+byteorder = "1.5.0"
+clap = { version = "4.4.6", features = ["derive", "cargo", "env"] }
+ethabi = "18.0.0"
+ethers = { version = "2.0.14", features = ["ws"] }
+futures = { version = "0.3.28" }
 hex = "0.4.3"
-prometheus-client  = { version = "0.21.2" }
+prometheus-client = { version = "0.21.2" }
 pythnet-sdk = { path = "../../pythnet/pythnet_sdk", features = ["strum"] }
-rand        = "0.8.5"
-reqwest     = { version = "0.11.22", features = ["json", "blocking"] }
-serde       = { version = "1.0.188", features = ["derive"] }
-serde_qs           = { version = "0.12.0", features = ["axum"] }
-serde_json  = "1.0.107"
+rand = "0.8.5"
+reqwest = { version = "0.11.22", features = ["json", "blocking"] }
+serde = { version = "1.0.188", features = ["derive"] }
+serde_qs = { version = "0.12.0", features = ["axum"] }
+serde_json = "1.0.107"
 serde_with = { version = "3.4.0", features = ["hex", "base64"] }
 serde_yaml = "0.9.25"
-sha3        = "0.10.8"
-tokio       = { version = "1.33.0", features = ["full"] }
-tower-http         = { version = "0.4.0", features = ["cors"] }
-tracing            = { version = "0.1.37", features = ["log"] }
+sha3 = "0.10.8"
+tokio = { version = "1.33.0", features = ["full"] }
+tower-http = { version = "0.4.0", features = ["cors"] }
+tracing = { version = "0.1.37", features = ["log"] }
 tracing-subscriber = { version = "0.3.17", features = ["env-filter"] }
-utoipa             = { version = "3.4.0", features = ["axum_extras"] }
-utoipa-swagger-ui  = { version = "3.1.4", features = ["axum"] }
+utoipa = { version = "3.4.0", features = ["axum_extras"] }
+utoipa-swagger-ui = { version = "3.1.4", features = ["axum"] }
 once_cell = "1.18.0"
 lazy_static = "1.4.0"
 url = "2.5.0"
-chrono = { version = "0.4.38", features = ["clock", "std"] , default-features = false}
+chrono = { version = "0.4.38", features = [
+  "clock",
+  "std",
+], default-features = false }
 backoff = { version = "0.4.0", features = ["futures", "tokio"] }
 thiserror = "1.0.61"
 futures-locks = "0.7.1"

apps/fortuna/src/keeper.rs

@@ -20,7 +20,7 @@ use {
     futures::StreamExt,
     prometheus_client::{
         encoding::EncodeLabelSet,
-        metrics::{counter::Counter, family::Family, gauge::Gauge},
+        metrics::{counter::Counter, family::Family, gauge::Gauge, histogram::Histogram},
         registry::Registry,
     },
     std::{
@@ -62,7 +62,6 @@ pub struct AccountLabel {
     pub address: String,
 }
 
-#[derive(Default)]
 pub struct KeeperMetrics {
     pub current_sequence_number: Family<AccountLabel, Gauge>,
     pub end_sequence_number: Family<AccountLabel, Gauge>,
@@ -74,6 +73,33 @@ pub struct KeeperMetrics {
     pub requests_processed: Family<AccountLabel, Counter>,
     pub requests_reprocessed: Family<AccountLabel, Counter>,
     pub reveals: Family<AccountLabel, Counter>,
+    pub request_duration_ms: Family<AccountLabel, Histogram>,
+}
+
+impl Default for KeeperMetrics {
+    fn default() -> Self {
+        Self {
+            current_sequence_number: Family::default(),
+            end_sequence_number: Family::default(),
+            balance: Family::default(),
+            collected_fee: Family::default(),
+            current_fee: Family::default(),
+            total_gas_spent: Family::default(),
+            requests: Family::default(),
+            requests_processed: Family::default(),
+            requests_reprocessed: Family::default(),
+            reveals: Family::default(),
+            request_duration_ms: Family::new_with_constructor(|| {
+                Histogram::new(
+                    vec![
+                        1000.0, 2500.0, 5000.0, 7500.0, 10000.0, 20000.0, 30000.0, 40000.0,
+                        50000.0, 60000.0,
+                    ]
+                    .into_iter(),
+                )
+            }),
+        }
+    }
 }
 
 impl KeeperMetrics {
@@ -141,6 +167,12 @@ impl KeeperMetrics {
             keeper_metrics.requests_reprocessed.clone(),
         );
 
+        writable_registry.register(
+            "request_duration_ms",
+            "Time taken to process each callback request in milliseconds",
+            keeper_metrics.request_duration_ms.clone(),
+        );
+
         keeper_metrics
     }
 }
@@ -342,6 +374,8 @@ pub async fn process_event_with_backoff(
     gas_limit: U256,
     metrics: Arc<KeeperMetrics>,
 ) {
+    let start_time = std::time::Instant::now();
+
     metrics
         .requests
         .get_or_create(&AccountLabel {
@@ -372,6 +406,16 @@ pub async fn process_event_with_backoff(
             tracing::error!("Failed to process event: {:?}", e);
         }
     }
+
+    let duration_ms = start_time.elapsed().as_millis() as f64;
+    metrics
+        .request_duration_ms
+        .get_or_create(&AccountLabel {
+            chain_id: chain_state.id.clone(),
+            address: chain_state.provider_address.to_string(),
+        })
+        .observe(duration_ms);
+
     metrics
         .requests_processed
         .get_or_create(&AccountLabel {