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, GenericError};
20use serde_json::Value;
21
22/// Produces a fresh serialized configuration snapshot per call.
23pub type ConfigSnapshotFn = Arc<dyn Fn() -> Result<Value, GenericError> + 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        match (state.snapshot)() {
49            Ok(config) => (StatusCode::OK, serde_json::to_string(&config).unwrap()).into_response(),
50            Err(e) => (
51                StatusCode::INTERNAL_SERVER_ERROR,
52                format!("Failed to get configuration: {}", e),
53            )
54                .into_response(),
55        }
56    }
57}
58
59impl APIHandler for ConfigAPIHandler {
60    type State = ConfigState;
61
62    fn generate_initial_state(&self) -> Self::State {
63        self.state.clone()
64    }
65
66    fn generate_routes(&self) -> Router<Self::State> {
67        Router::new().route("/config", get(Self::config_handler))
68    }
69}
70
71/// A worker for exposing an endpoint that returns the current configuration.
72///
73/// When running, the worker asserts a set of routes (based on [`ConfigAPIHandler`]) that allow querying the current
74/// configuration. As the configuration may contain sensitive data, these routes are only present on the privileged API
75/// endpoint.
76pub struct ConfigWorker {
77    handler: ConfigAPIHandler,
78}
79
80impl ConfigWorker {
81    /// Creates a new [`ConfigWorker`] that serves the snapshots produced by the given closure.
82    pub fn new(snapshot: ConfigSnapshotFn) -> Self {
83        Self {
84            handler: ConfigAPIHandler::new(snapshot),
85        }
86    }
87}
88
89#[async_trait]
90impl Supervisable for ConfigWorker {
91    fn name(&self) -> &str {
92        "config-api"
93    }
94
95    async fn initialize(&self, process_shutdown: ShutdownHandle) -> Result<SupervisorFuture, InitializationError> {
96        let config_route = DynamicRoute::http(EndpointType::Privileged, &self.handler);
97
98        let snapshot = self.handler.state.snapshot.clone();
99
100        Ok(Box::pin(async move {
101            let dataspace =
102                DataspaceRegistry::try_current().ok_or_else(|| generic_error!("Dataspace not available."))?;
103
104            dataspace.assert(config_route, "config-api");
105
106            let diagnostics =
107                DiagnosticsEmitter::from_dataspace(SubsystemIdentifier::from_segments(["config-api"]), dataspace);
108            diagnostics.register_collector("runtime_config_dump.yaml", move || {
109                snapshot()
110                    .map(|v| serde_json::to_vec_pretty(&v).unwrap_or_default())
111                    .unwrap_or_default()
112            });
113
114            process_shutdown.await;
115            Ok(())
116        }))
117    }
118}
119
120#[cfg(test)]
121mod tests {
122    use std::sync::atomic::{AtomicUsize, Ordering};
123
124    use http_body_util::BodyExt as _;
125    use saluki_error::generic_error;
126    use serde_json::json;
127
128    use super::*;
129
130    async fn response_parts(handler: &ConfigAPIHandler) -> (StatusCode, String) {
131        let response = ConfigAPIHandler::config_handler(State(handler.state.clone()))
132            .await
133            .into_response();
134        let status = response.status();
135        let body = response.into_body().collect().await.expect("body collects").to_bytes();
136
137        (status, String::from_utf8(body.to_vec()).expect("body is UTF-8"))
138    }
139
140    #[tokio::test]
141    async fn config_endpoint_serves_a_fresh_snapshot_per_request() {
142        let calls = Arc::new(AtomicUsize::new(0));
143        let snapshot_calls = Arc::clone(&calls);
144        let handler = ConfigAPIHandler::new(Arc::new(move || {
145            Ok(json!({ "revision": snapshot_calls.fetch_add(1, Ordering::Relaxed) }))
146        }));
147
148        let (status, body) = response_parts(&handler).await;
149        assert_eq!(status, StatusCode::OK);
150        assert_eq!(body, r#"{"revision":0}"#);
151
152        let (status, body) = response_parts(&handler).await;
153        assert_eq!(status, StatusCode::OK);
154        assert_eq!(body, r#"{"revision":1}"#);
155        assert_eq!(calls.load(Ordering::Relaxed), 2);
156    }
157
158    #[tokio::test]
159    async fn config_endpoint_reports_a_failed_snapshot() {
160        let handler = ConfigAPIHandler::new(Arc::new(|| Err(generic_error!("cannot serialize"))));
161
162        let (status, body) = response_parts(&handler).await;
163        assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR);
164        assert!(body.contains("cannot serialize"), "unexpected body: {body}");
165    }
166}
saluki_app/config.rs

saluki_app/
config.rs
