add additional status update loop metrics (#575)

Author: JannikSt

crates/orchestrator/src/main.rs

@@ -381,6 +381,7 @@ async fn main() -> Result<()> {
 
         let status_update_store_context = store_context.clone();
         let status_update_heartbeats = heartbeats.clone();
+        let status_update_metrics = metrics_context.clone();
         tasks.spawn({
             let contracts = contracts.clone();
             async move {
@@ -393,6 +394,7 @@ async fn main() -> Result<()> {
                     args.disable_ejection,
                     status_update_heartbeats.clone(),
                     status_updater_plugins,
+                    status_update_metrics,
                 );
                 status_updater.run().await
             }

crates/orchestrator/src/metrics/mod.rs

@@ -1,4 +1,4 @@
-use prometheus::{CounterVec, GaugeVec, Opts, Registry, TextEncoder};
+use prometheus::{CounterVec, GaugeVec, HistogramOpts, HistogramVec, Opts, Registry, TextEncoder};
 pub mod sync_service;
 pub mod webhook_sender;
 
@@ -14,6 +14,7 @@ pub struct MetricsContext {
     pub heartbeat_requests_total: CounterVec,
     pub nodes_per_task: GaugeVec,
     pub task_state: GaugeVec,
+    pub status_update_execution_time: HistogramVec,
 }
 
 impl MetricsContext {
@@ -90,7 +91,6 @@ impl MetricsContext {
             &["task_id", "task_name", "pool_id"],
         )
         .unwrap();
-
         let task_state = GaugeVec::new(
             Opts::new(
                 "orchestrator_task_state",
@@ -100,6 +100,19 @@ impl MetricsContext {
         )
         .unwrap();
 
+        let status_update_execution_time = HistogramVec::new(
+            HistogramOpts::new(
+                "orchestrator_status_update_execution_time_seconds",
+                "Duration of status update execution",
+            )
+            .buckets(vec![
+                0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.0, 5.0, 10.0, 15.0, 30.0, 45.0, 60.0, 90.0,
+                120.0,
+            ]),
+            &["node_address", "pool_id"],
+        )
+        .unwrap();
+
         let registry = Registry::new();
         let _ = registry.register(Box::new(compute_task_gauges.clone()));
         let _ = registry.register(Box::new(task_info.clone()));
@@ -110,6 +123,7 @@ impl MetricsContext {
         let _ = registry.register(Box::new(heartbeat_requests_total.clone()));
         let _ = registry.register(Box::new(nodes_per_task.clone()));
         let _ = registry.register(Box::new(task_state.clone()));
+        let _ = registry.register(Box::new(status_update_execution_time.clone()));
 
         Self {
             compute_task_gauges,
@@ -123,6 +137,7 @@ impl MetricsContext {
             heartbeat_requests_total,
             nodes_per_task,
             task_state,
+            status_update_execution_time,
         }
     }
 
@@ -218,6 +233,12 @@ impl MetricsContext {
         self.compute_task_gauges.reset();
     }
 
+    pub fn record_status_update_execution_time(&self, node_address: &str, duration: f64) {
+        self.status_update_execution_time
+            .with_label_values(&[node_address, &self.pool_id])
+            .observe(duration);
+    }
+
     /// Clear all orchestrator statistics metrics
     pub fn clear_orchestrator_statistics(&self) {
         self.nodes_total.reset();

crates/orchestrator/src/status_update/mod.rs

@@ -1,3 +1,4 @@
+use crate::metrics::MetricsContext;
 use crate::models::node::{NodeStatus, OrchestratorNode};
 use crate::plugins::StatusUpdatePlugin;
 use crate::store::core::StoreContext;
@@ -7,7 +8,7 @@ use shared::web3::contracts::core::builder::Contracts;
 use shared::web3::wallet::WalletProvider;
 use std::result::Result;
 use std::sync::Arc;
-use std::time::Duration;
+use std::time::{Duration, Instant};
 use tokio::time::interval;
 
 pub struct NodeStatusUpdater {
@@ -19,6 +20,7 @@ pub struct NodeStatusUpdater {
     disable_ejection: bool,
     heartbeats: Arc<LoopHeartbeats>,
     plugins: Vec<Box<dyn StatusUpdatePlugin>>,
+    metrics: Arc<MetricsContext>,
 }
 
 impl NodeStatusUpdater {
@@ -32,6 +34,7 @@ impl NodeStatusUpdater {
         disable_ejection: bool,
         heartbeats: Arc<LoopHeartbeats>,
         plugins: Vec<Box<dyn StatusUpdatePlugin>>,
+        metrics: Arc<MetricsContext>,
     ) -> Self {
         Self {
             store_context,
@@ -42,6 +45,7 @@ impl NodeStatusUpdater {
             disable_ejection,
             heartbeats,
             plugins,
+            metrics,
         }
     }
 
@@ -133,6 +137,7 @@ impl NodeStatusUpdater {
     pub async fn process_nodes(&self) -> Result<(), anyhow::Error> {
         let nodes = self.store_context.node_store.get_nodes().await?;
         for node in nodes {
+            let start_time = Instant::now();
             let node = node.clone();
             let old_status = node.status.clone();
             let heartbeat = self
@@ -313,6 +318,12 @@ impl NodeStatusUpdater {
                     }
                 }
             }
+            // Record status update execution time
+            let duration = start_time.elapsed();
+            self.metrics.record_status_update_execution_time(
+                &node.address.to_string(),
+                duration.as_secs_f64(),
+            );
         }
         Ok(())
     }
@@ -346,6 +357,7 @@ mod tests {
             false,
             Arc::new(LoopHeartbeats::new(&mode)),
             vec![],
+            app_state.metrics.clone(),
         );
         let node = OrchestratorNode {
             address: Address::from_str("0x0000000000000000000000000000000000000000").unwrap(),
@@ -450,6 +462,7 @@ mod tests {
             false,
             Arc::new(LoopHeartbeats::new(&mode)),
             vec![],
+            app_state.metrics.clone(),
         );
         tokio::spawn(async move {
             updater
@@ -502,6 +515,7 @@ mod tests {
             false,
             Arc::new(LoopHeartbeats::new(&mode)),
             vec![],
+            app_state.metrics.clone(),
         );
         tokio::spawn(async move {
             updater
@@ -570,6 +584,7 @@ mod tests {
             false,
             Arc::new(LoopHeartbeats::new(&mode)),
             vec![],
+            app_state.metrics.clone(),
         );
         tokio::spawn(async move {
             updater
@@ -648,6 +663,7 @@ mod tests {
             false,
             Arc::new(LoopHeartbeats::new(&mode)),
             vec![],
+            app_state.metrics.clone(),
         );
         tokio::spawn(async move {
             updater
@@ -734,6 +750,7 @@ mod tests {
             false,
             Arc::new(LoopHeartbeats::new(&mode)),
             vec![],
+            app_state.metrics.clone(),
         );
         tokio::spawn(async move {
             updater
@@ -817,6 +834,7 @@ mod tests {
             false,
             Arc::new(LoopHeartbeats::new(&mode)),
             vec![],
+            app_state.metrics.clone(),
         );
         tokio::spawn(async move {
             updater
@@ -894,6 +912,7 @@ mod tests {
             false,
             Arc::new(LoopHeartbeats::new(&mode)),
             vec![],
+            app_state.metrics.clone(),
         );
         tokio::spawn(async move {
             updater
@@ -962,6 +981,7 @@ mod tests {
             false,
             Arc::new(LoopHeartbeats::new(&mode)),
             vec![],
+            app_state.metrics.clone(),
         );
         tokio::spawn(async move {
             updater