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

synd_api/
monitor.rs
