saluki_app/
config.rs

1//! Configuration API handler.
2
3use std::sync::Arc;
4
5use async_trait::async_trait;
6use http::StatusCode;
7use saluki_api::{
8    extract::State,
9    response::IntoResponse,
10    routing::{get, Router},
11    APIHandler, DynamicRoute, EndpointType,
12};
13use saluki_common::sync::shutdown::ShutdownHandle;
14use saluki_core::{
15    diagnostic::DiagnosticsEmitter,
16    runtime::{state::DataspaceRegistry, InitializationError, Supervisable, SupervisorFuture},
17    support::SubsystemIdentifier,
18};
19use saluki_error::generic_error;
20use serde_json::Value;
21
22/// Produces a fresh serialized configuration snapshot per call.
23pub type ConfigSnapshotFn = Arc<dyn Fn() -> Value + Send + Sync>;
24
25/// State used for the config API handler.
26#[derive(Clone)]
27pub struct ConfigState {
28    snapshot: ConfigSnapshotFn,
29}
30
31/// An API handler for returning the current configuration.
32///
33/// This handler exposes a single route -- `/config` -- that returns the current configuration in its serialized JSON
34/// form. This allows determining exactly how the process' configuration looks based on the various providers being
35/// used, including any dynamic changes being applied.
36pub struct ConfigAPIHandler {
37    state: ConfigState,
38}
39
40impl ConfigAPIHandler {
41    fn new(snapshot: ConfigSnapshotFn) -> Self {
42        Self {
43            state: ConfigState { snapshot },
44        }
45    }
46
47    async fn config_handler(State(state): State<ConfigState>) -> impl IntoResponse {
48        let config = (state.snapshot)();
49        (StatusCode::OK, serde_json::to_string(&config).unwrap()).into_response()
50    }
51}
52
53impl APIHandler for ConfigAPIHandler {
54    type State = ConfigState;
55
56    fn generate_initial_state(&self) -> Self::State {
57        self.state.clone()
58    }
59
60    fn generate_routes(&self) -> Router<Self::State> {
61        Router::new().route("/config", get(Self::config_handler))
62    }
63}
64
65/// A worker for exposing an endpoint that returns the current configuration.
66///
67/// When running, the worker asserts a set of routes (based on [`ConfigAPIHandler`]) that allow querying the current
68/// configuration. As the configuration may contain sensitive data, these routes are only present on the privileged API
69/// endpoint.
70pub struct ConfigWorker {
71    handler: ConfigAPIHandler,
72}
73
74impl ConfigWorker {
75    /// Creates a new [`ConfigWorker`] that serves the snapshots produced by the given closure.
76    pub fn new(snapshot: ConfigSnapshotFn) -> Self {
77        Self {
78            handler: ConfigAPIHandler::new(snapshot),
79        }
80    }
81}
82
83#[async_trait]
84impl Supervisable for ConfigWorker {
85    fn name(&self) -> &str {
86        "config-api"
87    }
88
89    async fn initialize(&self, process_shutdown: ShutdownHandle) -> Result<SupervisorFuture, InitializationError> {
90        let config_route = DynamicRoute::http(EndpointType::Privileged, &self.handler);
91
92        let snapshot = self.handler.state.snapshot.clone();
93
94        Ok(Box::pin(async move {
95            let dataspace =
96                DataspaceRegistry::try_current().ok_or_else(|| generic_error!("Dataspace not available."))?;
97
98            dataspace.assert(config_route, "config-api");
99
100            let diagnostics =
101                DiagnosticsEmitter::from_dataspace(SubsystemIdentifier::from_segments(["config-api"]), dataspace);
102            diagnostics.register_collector("runtime_config_dump.yaml", move || {
103                serde_json::to_vec_pretty(&snapshot()).unwrap_or_default()
104            });
105
106            process_shutdown.await;
107            Ok(())
108        }))
109    }
110}
111
112#[cfg(test)]
113mod tests {
114    use std::sync::atomic::{AtomicUsize, Ordering};
115
116    use http_body_util::BodyExt as _;
117    use serde_json::json;
118
119    use super::*;
120
121    async fn response_parts(handler: &ConfigAPIHandler) -> (StatusCode, String) {
122        let response = ConfigAPIHandler::config_handler(State(handler.state.clone()))
123            .await
124            .into_response();
125        let status = response.status();
126        let body = response.into_body().collect().await.expect("body collects").to_bytes();
127
128        (status, String::from_utf8(body.to_vec()).expect("body is UTF-8"))
129    }
130
131    #[tokio::test]
132    async fn config_endpoint_serves_a_fresh_snapshot_per_request() {
133        let calls = Arc::new(AtomicUsize::new(0));
134        let snapshot_calls = Arc::clone(&calls);
135        let handler = ConfigAPIHandler::new(Arc::new(
136            move || json!({ "revision": snapshot_calls.fetch_add(1, Ordering::Relaxed) }),
137        ));
138
139        let (status, body) = response_parts(&handler).await;
140        assert_eq!(status, StatusCode::OK);
141        assert_eq!(body, r#"{"revision":0}"#);
142
143        let (status, body) = response_parts(&handler).await;
144        assert_eq!(status, StatusCode::OK);
145        assert_eq!(body, r#"{"revision":1}"#);
146        assert_eq!(calls.load(Ordering::Relaxed), 2);
147    }
148}
saluki_app/config.rs

saluki_app/
config.rs
