datadog_agent_runtime/control_plane/
config_runtime.rs

1//! Runtime configuration API handler.
2
3use std::sync::Arc;
4
5use arc_swap::ArcSwap;
6use async_trait::async_trait;
7use saluki_api::{
8    extract::State,
9    response::IntoResponse,
10    routing::{get, Router},
11    APIHandler, DynamicRoute, EndpointType, StatusCode,
12};
13use saluki_common::sync::shutdown::ShutdownHandle;
14use saluki_core::runtime::{state::DataspaceRegistry, InitializationError, Supervisable, SupervisorFuture};
15use saluki_error::generic_error;
16use serde::Serialize;
17
18/// Produces the current runtime configuration in its serialized JSON form.
19#[derive(Clone)]
20pub struct RuntimeConfigSnapshot {
21    serialize: Arc<dyn Fn() -> Result<String, serde_json::Error> + Send + Sync>,
22}
23
24impl RuntimeConfigSnapshot {
25    /// Creates a `RuntimeConfigSnapshot` that serializes whatever configuration `current` holds at the time of each
26    /// request.
27    pub fn from_shared<T>(current: Arc<ArcSwap<T>>) -> Self
28    where
29        T: Serialize + Send + Sync + 'static,
30    {
31        Self {
32            serialize: Arc::new(move || serde_json::to_string(&**current.load())),
33        }
34    }
35
36    fn to_json(&self) -> Result<String, serde_json::Error> {
37        (self.serialize)()
38    }
39}
40
41/// State used for the runtime configuration API handler.
42#[derive(Clone)]
43pub struct ConfigRuntimeState {
44    snapshot: RuntimeConfigSnapshot,
45}
46
47/// An API handler for returning the translated runtime configuration.
48///
49/// This handler exposes a single route -- `/config/runtime` -- that returns the current
50/// translated configuration in its serialized JSON form. It reflects any dynamic updates
51/// applied to the configuration since startup.
52pub struct ConfigRuntimeAPIHandler {
53    state: ConfigRuntimeState,
54}
55
56impl ConfigRuntimeAPIHandler {
57    fn new(snapshot: RuntimeConfigSnapshot) -> Self {
58        Self {
59            state: ConfigRuntimeState { snapshot },
60        }
61    }
62
63    async fn config_handler(State(state): State<ConfigRuntimeState>) -> impl IntoResponse {
64        match state.snapshot.to_json() {
65            Ok(body) => (StatusCode::OK, body).into_response(),
66            Err(e) => (
67                StatusCode::INTERNAL_SERVER_ERROR,
68                format!("Failed to serialize configuration: {}", e),
69            )
70                .into_response(),
71        }
72    }
73}
74
75impl APIHandler for ConfigRuntimeAPIHandler {
76    type State = ConfigRuntimeState;
77
78    fn generate_initial_state(&self) -> Self::State {
79        self.state.clone()
80    }
81
82    fn generate_routes(&self) -> Router<Self::State> {
83        Router::new().route("/config/runtime", get(Self::config_handler))
84    }
85}
86
87/// A worker for exposing an endpoint that returns the translated runtime configuration.
88///
89/// When running, the worker asserts a route (based on `ConfigRuntimeAPIHandler`) that returns
90/// the current translated configuration. As the configuration may contain sensitive data, this
91/// route is only present on the privileged API endpoint.
92pub struct ConfigRuntimeWorker {
93    handler: ConfigRuntimeAPIHandler,
94}
95
96impl ConfigRuntimeWorker {
97    /// Creates a new [`ConfigRuntimeWorker`] serving the given configuration.
98    pub fn new(snapshot: RuntimeConfigSnapshot) -> Self {
99        Self {
100            handler: ConfigRuntimeAPIHandler::new(snapshot),
101        }
102    }
103}
104
105#[async_trait]
106impl Supervisable for ConfigRuntimeWorker {
107    fn name(&self) -> &str {
108        "config-runtime-api"
109    }
110
111    async fn initialize(&self, process_shutdown: ShutdownHandle) -> Result<SupervisorFuture, InitializationError> {
112        let config_route = DynamicRoute::http(EndpointType::Privileged, &self.handler);
113
114        Ok(Box::pin(async move {
115            let dataspace =
116                DataspaceRegistry::try_current().ok_or_else(|| generic_error!("Dataspace not available."))?;
117
118            dataspace.assert(config_route, "config-runtime-api");
119
120            process_shutdown.await;
121            Ok(())
122        }))
123    }
124}
125
126#[cfg(test)]
127mod tests {
128    use super::*;
129
130    #[derive(Serialize)]
131    struct ExampleConfig {
132        log_level: String,
133    }
134
135    #[test]
136    fn shared_snapshot_follows_updates_to_the_shared_configuration() {
137        let current = Arc::new(ArcSwap::from_pointee(ExampleConfig {
138            log_level: "info".to_string(),
139        }));
140        let snapshot = RuntimeConfigSnapshot::from_shared(Arc::clone(&current));
141        assert_eq!(snapshot.to_json().unwrap(), r#"{"log_level":"info"}"#);
142
143        current.store(Arc::new(ExampleConfig {
144            log_level: "debug".to_string(),
145        }));
146        assert_eq!(snapshot.to_json().unwrap(), r#"{"log_level":"debug"}"#);
147    }
148}
datadog_agent_runtime/control_plane/config_runtime.rs

datadog_agent_runtime/control_plane/
config_runtime.rs
