feat(api): instrument graphql task metrics

crates/synd_api/src/config.rs

@@ -23,4 +23,9 @@ pub mod serve {
 
     pub const HEALTH_CHECK_PATH: &str = "/health";
 }
+
+pub mod metrics {
+    use std::time::Duration;
+
+    pub const MONITOR_INTERVAL: Duration = Duration::from_secs(60);
 }

crates/synd_api/src/dependency.rs

@@ -29,7 +29,6 @@ impl Dependency {
         kvsd: KvsdOptions,
         tls: TlsOptions,
         serve_options: args::ServeOptions,
-        monitors: Monitors,
     ) -> anyhow::Result<Self> {
         let KvsdOptions {
             kvsd_host,
@@ -69,6 +68,8 @@ impl Dependency {
             .await
             .with_context(|| format!("tls options: {tls:?}"))?;
 
+        let monitors = Monitors::new();
+
         Ok(Dependency {
             authenticator,
             runtime,

crates/synd_api/src/main.rs

@@ -1,19 +1,14 @@
-use std::time::Duration;
-
 use fdlimit::Outcome;
 use synd_o11y::{
-    metric,
     opentelemetry::OpenTelemetryGuard,
     tracing_subscriber::otel_metrics::{self, metrics_event_filter},
 };
-use tokio_metrics::RuntimeMonitor;
 use tracing::{error, info};
 
 use synd_api::{
     args::{self, Args, ObservabilityOptions},
     config,
     dependency::Dependency,
-    monitor::Monitors,
     repository::kvsd::ConnectKvsdFailed,
     serve::listen_and_serve,
     shutdown::Shutdown,
@@ -90,9 +85,8 @@ async fn run(
         o11y,
     }: Args,
     shutdown: Shutdown,
-    monitors: Monitors,
 ) -> anyhow::Result<()> {
-    let dep = Dependency::new(kvsd, tls, serve, monitors).await?;
+    let dep = Dependency::new(kvsd, tls, serve).await?;
 
     info!(
         version = config::VERSION,
@@ -120,40 +114,15 @@ fn init_file_descriptor_limit() {
         .ok();
 }
 
-fn init_runtime_monitor() -> Monitors {
-    let handle = tokio::runtime::Handle::current();
-    let runtime_monitor = RuntimeMonitor::new(&handle);
-    let task_monitors = Monitors::new();
-    let intervals = runtime_monitor
-        .intervals()
-        .zip(task_monitors.gql.intervals());
-    tokio::spawn(async move {
-        for (runtime_metrics, gql_metrics) in intervals {
-            metric!(counter.runtime.poll = runtime_metrics.total_polls_count);
-            metric!(
-                counter.runtime.busy_duration = runtime_metrics.total_busy_duration.as_secs_f64()
-            );
-            metric!(
-                counter.task.graphql.idle_duration = gql_metrics.total_idle_duration.as_secs_f64()
-            );
-
-            tokio::time::sleep(Duration::from_secs(60)).await;
-        }
-    });
-
-    task_monitors
-}
-
 #[tokio::main]
 async fn main() {
     let args = args::parse();
     let _guard = init_tracing(&args.o11y);
     let shutdown = Shutdown::watch_signal();
-    let monitors = init_runtime_monitor();
 
     init_file_descriptor_limit();
 
-    if let Err(err) = run(args, shutdown, monitors).await {
+    if let Err(err) = run(args, shutdown).await {
         if let Some(err) = err.downcast_ref::<ConnectKvsdFailed>() {
             error!("{err}: make sure kvsd is running");
         } else {

crates/synd_api/src/monitor.rs

@@ -1,4 +1,7 @@
-use tokio_metrics::TaskMonitor;
+use std::time::Duration;
+
+use synd_o11y::metric;
+use tokio_metrics::{RuntimeMonitor, TaskMonitor};
 
 pub struct Monitors {
     pub gql: TaskMonitor,
@@ -10,4 +13,47 @@ impl Monitors {
             gql: TaskMonitor::new(),
         }
     }
+
+    pub async fn monitor(self, interval: Duration) {
+        let handle = tokio::runtime::Handle::current();
+        let runtime_monitor = RuntimeMonitor::new(&handle);
+        let intervals = runtime_monitor.intervals().zip(self.gql.intervals());
+
+        for (runtime_metrics, gql_metrics) in intervals {
+            // Runtime metrics
+            metric!(monotonic_counter.runtime.poll = runtime_metrics.total_polls_count);
+            metric!(
+                monotonic_counter.runtime.busy_duration =
+                    runtime_metrics.total_busy_duration.as_secs_f64()
+            );
+
+            // Tasks poll metrics
+            metric!(
+                counter.task.graphql.mean_poll_duration =
+                    gql_metrics.mean_poll_duration().as_secs_f64()
+            );
+            metric!(
+                counter.task.graphql.mean_slow_poll_duration =
+                    gql_metrics.mean_slow_poll_duration().as_secs_f64()
+            );
+
+            // Tasks schedule metrics
+            metric!(
+                counter.task.graphql.mean_first_poll_delay =
+                    gql_metrics.mean_first_poll_delay().as_secs_f64(),
+            );
+            metric!(
+                counter.task.graphql.mean_scheduled_duration =
+                    gql_metrics.mean_scheduled_duration().as_secs_f64(),
+            );
+
+            // Tasks idle metrics
+            metric!(
+                counter.task.graphql.mean_idle_duration =
+                    gql_metrics.mean_idle_duration().as_secs_f64(),
+            );
+
+            tokio::time::sleep(interval).await;
+        }
+    }
 }

crates/synd_api/src/serve/layer/request_metrics.rs

@@ -10,6 +10,8 @@ use futures_util::Future;
 use synd_o11y::metric;
 use tower::{Layer, Service};
 
+use crate::config;
+
 #[derive(Clone)]
 pub struct RequestMetricsLayer {}
 

crates/synd_api/src/serve/mod.rs

@@ -16,6 +16,7 @@ use tower_http::{
 use tracing::info;
 
 use crate::{
+    config,
     dependency::Dependency,
     gql::{self, SyndSchema},
     serve::layer::{authenticate, request_metrics::RequestMetricsLayer, trace},
@@ -76,10 +77,12 @@ pub async fn serve(
     } = dep;
 
     let cx = Context {
-        gql_monitor: monitors.gql,
+        gql_monitor: monitors.gql.clone(),
         schema: gql::schema_builder().data(runtime).finish(),
     };
 
+    tokio::spawn(monitors.monitor(config::metrics::MONITOR_INTERVAL));
+
     let service = Router::new()
         .route("/graphql", post(gql::handler::graphql))
         .layer(Extension(cx))