saluki_app/metrics/
api.rs

1use std::{sync::Arc, time::Duration};
2
3use async_trait::async_trait;
4use metrics::Level;
5use saluki_api::{
6    extract::{Query, State},
7    response::IntoResponse,
8    routing::{post, Router},
9    APIHandler, DynamicRoute, EndpointType, StatusCode,
10};
11use saluki_common::sync::shutdown::ShutdownHandle;
12use saluki_core::{
13    observability::metrics::FilterHandle,
14    runtime::{state::DataspaceRegistry, InitializationError, Supervisable, SupervisorFuture},
15};
16use saluki_error::generic_error;
17use serde::Deserialize;
18use tokio::{
19    pin, select,
20    sync::{mpsc, Mutex},
21    time::sleep,
22};
23use tracing::info;
24
25const MAXIMUM_OVERRIDE_LENGTH_SECS: u64 = 60 * 60;
26
27#[derive(Deserialize)]
28struct OverrideQueryParams {
29    time_secs: u64,
30}
31
32/// State used for the metrics API handler.
33#[derive(Clone)]
34pub struct MetricsHandlerState {
35    override_tx: mpsc::Sender<Option<(Duration, Level)>>,
36}
37
38/// An API handler for updating metric filtering directives at runtime.
39///
40/// This handler exposes two main routes -- `/metrics/override` and `/metrics/reset` -- which allow for
41/// overriding the default metric filtering directives (configured at startup) at runtime, and then resetting them once
42/// the override is no longer needed.
43///
44/// As this has the potential for incredibly high metrics cardinality at runtime, the override is set with a specific
45/// duration in which it will apply. Once an override has been active for the configured duration, it will automatically
46/// be reset unless the override is refreshed before the duration elapses.
47///
48/// The maximum duration for an override is 60 minutes.
49pub struct MetricsAPIHandler {
50    state: MetricsHandlerState,
51}
52
53impl MetricsAPIHandler {
54    async fn override_handler(
55        State(state): State<MetricsHandlerState>, params: Query<OverrideQueryParams>, body: String,
56    ) -> impl IntoResponse {
57        // Make sure the override length is within the acceptable range.
58        if params.time_secs > MAXIMUM_OVERRIDE_LENGTH_SECS {
59            return (
60                StatusCode::BAD_REQUEST,
61                format!(
62                    "override time cannot be greater than {} seconds",
63                    MAXIMUM_OVERRIDE_LENGTH_SECS
64                ),
65            );
66        }
67
68        // Parse the override duration and create a new filter from the body.
69        let duration = Duration::from_secs(params.time_secs);
70        let new_level = match Level::try_from(body.as_str()) {
71            Ok(level) => level,
72            Err(e) => {
73                return (
74                    StatusCode::BAD_REQUEST,
75                    format!("failed to parse override filter: {}", e),
76                )
77            }
78        };
79
80        // Instruct the override processor to apply the new metric filtering directives for the given duration.
81        let _ = state.override_tx.send(Some((duration, new_level))).await;
82
83        (StatusCode::OK, "acknowledged".to_string())
84    }
85
86    async fn reset_handler(State(state): State<MetricsHandlerState>) {
87        // Instruct the override processor to immediately reset back to the original metric filtering directives.
88        let _ = state.override_tx.send(None).await;
89    }
90}
91
92impl APIHandler for MetricsAPIHandler {
93    type State = MetricsHandlerState;
94
95    fn generate_initial_state(&self) -> Self::State {
96        self.state.clone()
97    }
98
99    fn generate_routes(&self) -> Router<Self::State> {
100        Router::new()
101            .route("/metrics/override", post(Self::override_handler))
102            .route("/metrics/reset", post(Self::reset_handler))
103    }
104}
105
106/// A worker that processes dynamic metric filter override requests.
107///
108/// When running, the worker asserts a set of routes (based on [`MetricsAPIHandler`]) that allow triggering
109/// an override of the current metrics filter directives as well as clearing (resetting) the active override.
110pub struct MetricsOverrideWorker {
111    handler: MetricsAPIHandler,
112    state: Arc<Mutex<MetricsOverrideWorkerState>>,
113}
114
115struct MetricsOverrideWorkerState {
116    filter_handle: FilterHandle,
117    override_rx: mpsc::Receiver<Option<(Duration, Level)>>,
118}
119
120impl MetricsOverrideWorker {
121    /// Creates a new `MetricsOverrideWorker` driving the given filter handle.
122    pub(super) fn new(filter_handle: FilterHandle) -> Self {
123        let (override_tx, override_rx) = mpsc::channel(1);
124        let handler = MetricsAPIHandler {
125            state: MetricsHandlerState { override_tx },
126        };
127        Self {
128            handler,
129            state: Arc::new(Mutex::new(MetricsOverrideWorkerState {
130                filter_handle,
131                override_rx,
132            })),
133        }
134    }
135}
136
137#[async_trait]
138impl Supervisable for MetricsOverrideWorker {
139    fn name(&self) -> &str {
140        "dynamic-metrics-override-processor"
141    }
142
143    async fn initialize(&self, process_shutdown: ShutdownHandle) -> Result<SupervisorFuture, InitializationError> {
144        let mut state = self.state.clone().lock_owned().await;
145        let metrics_route = DynamicRoute::http(EndpointType::Privileged, &self.handler);
146
147        Ok(Box::pin(async move {
148            DataspaceRegistry::try_current()
149                .ok_or_else(|| generic_error!("Dataspace not available."))?
150                .assert(metrics_route, "metrics-api");
151
152            process_override_requests(&mut state, process_shutdown).await;
153
154            Ok(())
155        }))
156    }
157}
158
159async fn process_override_requests(state: &mut MetricsOverrideWorkerState, process_shutdown: ShutdownHandle) {
160    let mut override_active = false;
161    let override_timeout = sleep(Duration::MAX);
162
163    pin!(override_timeout, process_shutdown);
164
165    loop {
166        select! {
167            _ = &mut process_shutdown => break,
168            maybe_override = state.override_rx.recv() => match maybe_override {
169                Some(Some((duration, new_level))) => {
170                    // TODO: Using the `Debug` representation of `Level` is noisy, and we should add a method upstream to
171                    // just get the stringified representation of the level instead.
172                    info!(level = ?new_level, "Overriding existing metric filtering directive for {} seconds...", duration.as_secs());
173
174                    state.filter_handle.override_filter(new_level);
175
176                    // Mark ourselves as having an active override and update the override timeout.
177                    override_active = true;
178                    override_timeout.as_mut().reset(tokio::time::Instant::now() + duration);
179                },
180
181                Some(None) => {
182                    // We've been instructed to immediately reset the filter back to the original one, so simply update
183                    // the override timeout to fire as soon as possible.
184                    override_timeout.as_mut().reset(tokio::time::Instant::now());
185                },
186
187                // Our sender has dropped, so there's no more override requests for us to handle.
188                None => break,
189            },
190            _ = &mut override_timeout => {
191                // Our override timeout has fired. If we have an active override, reset it.
192                //
193                // Otherwise, this just means that we've been running for a while without any overrides, so we can just
194                // reset the timeout with a long duration.
195                if override_active {
196                    override_active = false;
197
198                    state.filter_handle.reset_filter();
199
200                    info!("Restored original metric filtering directive.");
201                }
202
203                override_timeout.as_mut().reset(tokio::time::Instant::now() + Duration::from_secs(MAXIMUM_OVERRIDE_LENGTH_SECS));
204            }
205        }
206    }
207}
208
209// This module mirrors the well-tested override/reset worker in `logging/api.rs`. Unlike the logging worker -- whose
210// effect is observable through a locally-constructed `tracing` reload handle -- the metrics worker's only observable
211// effect is on a `FilterHandle`, which is created by installing the process-global metrics recorder. Because that
212// recorder can only be installed once per process, all of the worker's behaviors are exercised in a single test,
213// observed end-to-end through the real metrics pipeline.
214#[cfg(test)]
215mod tests {
216    use futures::StreamExt as _;
217    use metrics::{counter, Key};
218    use saluki_core::observability::metrics::{delete_counter, initialize_metrics, MetricsStream};
219
220    use super::*;
221
222    // Emits a DEBUG-level counter and immediately evicts it, returning whether the resulting snapshot carried the
223    // counter as an upsert -- i.e. whether a DEBUG-level metric currently passes the runtime filter level.
224    async fn debug_metric_passes_filter(stream: &mut MetricsStream, name: &'static str) -> bool {
225        counter!(level: Level::DEBUG, name).increment(1);
226        assert!(
227            delete_counter(Key::from_name(name)),
228            "the counter should exist before eviction"
229        );
230
231        let snapshot = tokio::time::timeout(Duration::from_secs(1), stream.next())
232            .await
233            .expect("a snapshot should be produced after eviction")
234            .expect("the metrics stream should not close");
235
236        !snapshot.upserts.is_empty()
237    }
238
239    async fn wait_until_filter_passes(stream: &mut MetricsStream, name: &'static str, expected: bool) {
240        let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
241
242        loop {
243            if debug_metric_passes_filter(stream, name).await == expected {
244                return;
245            }
246
247            assert!(
248                tokio::time::Instant::now() < deadline,
249                "filter never reached `passes={expected}`"
250            );
251
252            sleep(Duration::from_millis(10)).await;
253        }
254    }
255
256    #[tokio::test]
257    async fn override_reset_and_expiration_toggle_the_runtime_filter_level() {
258        let (filter_handle, _flusher) = initialize_metrics("test".to_string(), Level::INFO)
259            .await
260            .expect("metrics subsystem should initialize");
261
262        let (override_tx, override_rx) = mpsc::channel(1);
263        let mut state = MetricsOverrideWorkerState {
264            filter_handle,
265            override_rx,
266        };
267        let processor = tokio::spawn(async move {
268            process_override_requests(&mut state, ShutdownHandle::noop()).await;
269        });
270
271        let mut stream = MetricsStream::register();
272        const PROBE: &str = "override_reset_probe";
273
274        // With the default INFO filter, a DEBUG-level metric is filtered out.
275        assert!(!debug_metric_passes_filter(&mut stream, PROBE).await);
276
277        // Overriding to DEBUG lets the DEBUG-level metric through.
278        override_tx
279            .send(Some((Duration::from_secs(60), Level::DEBUG)))
280            .await
281            .expect("send override");
282        wait_until_filter_passes(&mut stream, PROBE, true).await;
283
284        // Resetting restores the INFO default, filtering the DEBUG-level metric out again.
285        override_tx.send(None).await.expect("send reset");
286        wait_until_filter_passes(&mut stream, PROBE, false).await;
287
288        // A time-boxed override auto-expires and restores the default filter without an explicit reset.
289        override_tx
290            .send(Some((Duration::from_millis(250), Level::DEBUG)))
291            .await
292            .expect("send time-boxed override");
293        wait_until_filter_passes(&mut stream, PROBE, true).await;
294        wait_until_filter_passes(&mut stream, PROBE, false).await;
295
296        // Dropping the controller side makes the worker exit cleanly.
297        drop(override_tx);
298        processor.await.expect("override processor should exit cleanly");
299    }
300}
saluki_app/metrics/api.rs

saluki_app/metrics/
api.rs
