Add metric for active time of monitored futures (#19226)

crates/mysten-metrics/src/lib.rs

@@ -4,6 +4,7 @@
 use axum::{extract::Extension, http::StatusCode, routing::get, Router};
 use dashmap::DashMap;
 use parking_lot::Mutex;
+use prometheus::core::{AtomicI64, GenericGauge};
 use simple_server_timing_header::Timer;
 use std::future::Future;
 use std::net::SocketAddr;
@@ -63,6 +64,7 @@ pub struct Metrics {
     pub channel_inflight: IntGaugeVec,
     pub channel_sent: IntGaugeVec,
     pub channel_received: IntGaugeVec,
+    pub future_active_duration_ns: IntGaugeVec,
     pub scope_iterations: IntGaugeVec,
     pub scope_duration_ns: IntGaugeVec,
     pub scope_entrance: IntGaugeVec,
@@ -107,6 +109,13 @@ impl Metrics {
                 registry,
             )
             .unwrap(),
+            future_active_duration_ns: register_int_gauge_vec_with_registry!(
+                "monitored_future_active_duration_ns",
+                "Total duration in nanosecs where the monitored future is active (consuming CPU time)",
+                &["name"],
+                registry,
+            )
+            .unwrap(),
             scope_entrance: register_int_gauge_vec_with_registry!(
                 "monitored_scope_entrance",
                 "Number of entrance in the scope.",
@@ -349,21 +358,29 @@ impl<F: Future> MonitoredFutureExt for F {
     fn in_monitored_scope(self, name: &'static str) -> MonitoredScopeFuture<Self> {
         MonitoredScopeFuture {
             f: Box::pin(self),
+            active_duration_metric: get_metrics()
+                .map(|m| m.future_active_duration_ns.with_label_values(&[name])),
             _scope: monitored_scope(name),
         }
     }
 }
 
 pub struct MonitoredScopeFuture<F: Sized> {
     f: Pin<Box<F>>,
+    active_duration_metric: Option<GenericGauge<AtomicI64>>,
     _scope: Option<MonitoredScopeGuard>,
 }
 
 impl<F: Future> Future for MonitoredScopeFuture<F> {
     type Output = F::Output;
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        self.f.as_mut().poll(cx)
+        let active_timer = Instant::now();
+        let ret = self.f.as_mut().poll(cx);
+        if let Some(m) = &self.active_duration_metric {
+            m.add(active_timer.elapsed().as_nanos() as i64);
+        }
+        ret
     }
 }