synd_api/
monitor.rs

1use std::time::Duration;
2
3use synd_support::o11y::metric;
4use tokio_metrics::{RuntimeMetrics, RuntimeMonitor, TaskMetrics, TaskMonitor};
5use tokio_util::sync::CancellationToken;
6
7struct Metrics {
8    runtime_busy_duration: f64,
9    gql_mean_poll_duration: f64,
10    gql_mean_slow_poll_duration: f64,
11    gql_mean_first_poll_delay: f64,
12    gql_mean_scheduled_duration: f64,
13    gql_mean_idle_duration: f64,
14}
15
16pub struct Monitors {
17    gql: TaskMonitor,
18}
19
20impl Monitors {
21    pub fn new() -> Self {
22        Self {
23            gql: TaskMonitor::new(),
24        }
25    }
26
27    pub(crate) fn graphql_task_monitor(&self) -> TaskMonitor {
28        self.gql.clone()
29    }
30
31    pub async fn emit_metrics(self, interval: Duration, ct: CancellationToken) {
32        let handle = tokio::runtime::Handle::current();
33        let runtime_monitor = RuntimeMonitor::new(&handle);
34        let intervals = runtime_monitor.intervals().zip(self.gql.intervals());
35
36        for (runtime_metrics, gql_metrics) in intervals {
37            let Metrics {
38                runtime_busy_duration,
39                gql_mean_poll_duration,
40                gql_mean_slow_poll_duration,
41                gql_mean_first_poll_delay,
42                gql_mean_scheduled_duration,
43                gql_mean_idle_duration,
44            } = Self::collect_metrics(&runtime_metrics, &gql_metrics);
45
46            // Runtime metrics
47            metric!(monotonic_counter.runtime.busy_duration = runtime_busy_duration);
48
49            // Tasks poll metrics
50            metric!(monotonic_counter.task.graphql.mean_poll_duration = gql_mean_poll_duration);
51            metric!(
52                monotonic_counter.task.graphql.mean_slow_poll_duration =
53                    gql_mean_slow_poll_duration
54            );
55
56            // Tasks schedule metrics
57            metric!(
58                monotonic_counter.task.graphql.mean_first_poll_delay = gql_mean_first_poll_delay,
59            );
60            metric!(
61                monotonic_counter.task.graphql.mean_scheduled_duration =
62                    gql_mean_scheduled_duration,
63            );
64
65            // Tasks idle metrics
66            metric!(monotonic_counter.task.graphql.mean_idle_duration = gql_mean_idle_duration);
67
68            tokio::select! {
69                biased;
70                // Make sure to respect cancellation
71                () = ct.cancelled() => break,
72                () = tokio::time::sleep(interval) => ()
73            }
74        }
75    }
76
77    fn collect_metrics(runtime_metrics: &RuntimeMetrics, gql_metrics: &TaskMetrics) -> Metrics {
78        Metrics {
79            runtime_busy_duration: runtime_metrics.total_busy_duration.as_secs_f64(),
80            gql_mean_poll_duration: gql_metrics.mean_poll_duration().as_secs_f64(),
81            gql_mean_slow_poll_duration: gql_metrics.mean_slow_poll_duration().as_secs_f64(),
82            gql_mean_first_poll_delay: gql_metrics.mean_first_poll_delay().as_secs_f64(),
83            gql_mean_scheduled_duration: gql_metrics.mean_scheduled_duration().as_secs_f64(),
84            gql_mean_idle_duration: gql_metrics.mean_idle_duration().as_secs_f64(),
85        }
86    }
87}
88
89#[cfg(test)]
90mod tests {
91    use std::sync::{Arc, Mutex};
92
93    use tracing::{Event, Subscriber, instrument::WithSubscriber};
94    use tracing_subscriber::{
95        Layer, Registry,
96        layer::{Context, SubscriberExt as _},
97        registry::LookupSpan,
98    };
99
100    use super::*;
101
102    struct TestLayer<F> {
103        on_event: F,
104    }
105
106    impl<S, F> Layer<S> for TestLayer<F>
107    where
108        S: Subscriber + for<'span> LookupSpan<'span>,
109        F: Fn(&Event<'_>) + 'static,
110    {
111        fn on_event(&self, event: &Event<'_>, _ctx: Context<'_, S>) {
112            (self.on_event)(event);
113        }
114    }
115
116    #[tokio::test]
117    async fn emit_metrics() {
118        let events = Arc::new(Mutex::new(Vec::new()));
119        let events_cloned = events.clone();
120        let on_event = move |event: &Event<'_>| {
121            let field = event.fields().next().unwrap().name();
122            events_cloned.lock().unwrap().push(field);
123        };
124        let layer = TestLayer { on_event };
125        let registry = Registry::default().with(layer);
126        let m = Monitors::new();
127        let ct = CancellationToken::new();
128        ct.cancel();
129
130        m.emit_metrics(Duration::from_millis(0), ct)
131            .with_subscriber(registry)
132            .await;
133
134        let events = events.lock().unwrap().clone();
135        insta::with_settings!({
136            description => "metrics which monitor emits",
137            omit_expression => true ,
138        }, {
139            insta::assert_yaml_snapshot!(events);
140        });
141    }
142}
synd_api/monitor.rs

synd_api/
monitor.rs
