agent_data_plane_config_system/
loaded.rs

1//! Loads local configuration before selecting its runtime authority.
2//!
3//! [`LoadedConfiguration::load`] prepares a typed snapshot of the local file and environment, which
4//! [`LoadedConfiguration::local`] exposes before values from the Datadog Agent stream are applied.
5//! [`LoadedConfiguration::run`] layers the Agent's configuration stream over the local sources,
6//! while [`LoadedConfiguration::standalone`] keeps the local sources authoritative. Both methods
7//! consume the loaded sources and return a [`ConfigurationSystem`].
8
9use std::path::Path;
10
11use agent_data_plane_config::SalukiConfiguration;
12use datadog_agent_config::apply_datadog_env;
13use saluki_config::dynamic::ConfigUpdate;
14use saluki_config::{ConfigurationLoader, GenericConfiguration};
15use serde_json::Value;
16use tokio::sync::mpsc;
17
18use crate::env_provider::EnvironmentProvider;
19use crate::saluki_env_overlay;
20use crate::source::SourceTree;
21use crate::system::{translate_strict, validate, ConfigurationSystem, Error};
22
23// The environment-variable prefix ADP reads (`DD_`). Mirrors
24// `PlatformSettings::get_env_var_prefix()`; hardcoded so the configuration system need not depend on
25// `datadog-agent-commons` for a single constant.
26const ENV_VAR_PREFIX: &str = "DD";
27
28// Bound on the internal channel that forwards Agent updates into the compatibility map. Matches the
29// Agent stream's own channel depth (`RemoteAgentBootstrap::create_config_stream`).
30const COMPAT_FORWARD_CHANNEL_SIZE: usize = 100;
31
32/// Where environment variables sit relative to the configuration file.
33///
34/// One setting, applied identically to the Figment provider order for the by-key view and to the
35/// order the typed base is composed in.
36#[derive(Clone, Copy, Debug, Eq, PartialEq)]
37pub enum EnvPrecedence {
38    /// Environment variables are read below the file: the file wins.
39    ///
40    /// The file takes precedence over environment variables.
41    BeforeFile,
42    /// Environment variables are read after the file: the environment wins. Matches the Datadog
43    /// Agent's precedence and is the value ADP uses.
44    AfterFile,
45    /// Environment variables are not read.
46    Disabled,
47}
48
49/// Local configuration prepared before a runtime authority is selected.
50///
51/// Retains a nested base for the typed path and a loader for the legacy by-key path. Both use the
52/// same source path and precedence but build their representations independently.
53pub struct LoadedConfiguration {
54    loader: ConfigurationLoader,
55    // Nested local base used for typed translation and Agent-layer merges.
56    base: SourceTree,
57    // Strictly translated local snapshot exposed before authority selection and used by standalone
58    // mode.
59    local: SalukiConfiguration,
60}
61
62impl LoadedConfiguration {
63    /// Loads and strictly translates the local file and environment using the requested precedence.
64    ///
65    /// The snapshot is translated but not validated: it is not yet authoritative. Under the Datadog
66    /// Agent the stream still has settings to contribute, so a local snapshot that could not run on
67    /// its own is normal here. Whichever authority [`run`](Self::run) or
68    /// [`standalone`](Self::standalone) selects applies validation.
69    ///
70    /// # Errors
71    ///
72    /// Returns an error if a local source cannot be read, decoded, deserialized, or translated.
73    pub async fn load(path: impl AsRef<Path>, env: EnvPrecedence) -> Result<Self, Error> {
74        let loader = build_loader(path.as_ref(), env)?;
75        // Every value the local sources supply was set explicitly: the file set it or an environment
76        // variable supplied it.
77        let base = SourceTree::all_explicit(build_base(path.as_ref(), env)?);
78        let local = translate_strict(&base)?;
79        Ok(Self { loader, base, local })
80    }
81
82    /// Returns the typed snapshot of the local file and environment.
83    ///
84    /// Values from the Datadog Agent stream have not been applied to this snapshot.
85    pub fn local(&self) -> &SalukiConfiguration {
86        &self.local
87    }
88
89    /// Returns the local file and environment through the legacy by-key configuration API.
90    // TODO: Remove this compatibility view once bootstrap consumers use `local`.
91    pub fn raw_config(&self) -> GenericConfiguration {
92        self.loader.bootstrap_generic()
93    }
94
95    /// Uses the Datadog Agent's configuration stream as the runtime authority.
96    ///
97    /// Waits for the initial Agent snapshot, layers it over the local sources, strictly translates
98    /// the result, and then starts the update task.
99    ///
100    /// # Errors
101    ///
102    /// Returns an error if the compatibility map cannot be built, the stream closes before its
103    /// initial snapshot, or the merged configuration cannot be deserialized or translated.
104    pub async fn run(self, config_stream: mpsc::Receiver<ConfigUpdate>) -> Result<ConfigurationSystem, Error> {
105        // The configuration system owns the Agent stream and forwards each update into this
106        // compatibility map, so `raw_map()` keeps serving un-migrated components.
107        let (compat_tx, compat_rx) = mpsc::channel(COMPAT_FORWARD_CHANNEL_SIZE);
108        let compat_map = self.loader.with_dynamic_configuration(compat_rx).into_generic().await?;
109
110        ConfigurationSystem::connected(config_stream, compat_tx, compat_map, self.base).await
111    }
112
113    /// Uses the translated local configuration as the runtime authority.
114    ///
115    /// No configuration stream or update task is created. Because nothing further will be layered on,
116    /// the local snapshot is validated here, where [`run`](Self::run) instead validates the merged
117    /// result of the Agent's initial snapshot.
118    ///
119    /// # Errors
120    ///
121    /// Returns an error if the compatibility map cannot be built from the local sources, or if the
122    /// local configuration fails validation.
123    pub async fn standalone(self) -> Result<ConfigurationSystem, Error> {
124        validate(&self.local)?;
125        let compat_map = self.loader.into_generic().await?;
126        Ok(ConfigurationSystem::standalone(compat_map, self.local))
127    }
128}
129
130/// Builds the typed base: the configuration file parsed to its nested shape, with environment
131/// variables read directly and decoded into the schema's shapes on top.
132///
133/// It reads the same file as the by-key compatibility view, normalizes it the same way (drop
134/// null-valued keys; an empty file is an empty object), and then overlays the environment via the
135/// generated Datadog reader, the Saluki-only reader, and the
136/// canonical proxy variables. `env` sets whether the environment overwrites the file (`AfterFile`)
137/// or only fills absent keys (`BeforeFile`); `Disabled` skips the environment entirely.
138fn build_base(path: &Path, env: EnvPrecedence) -> Result<Value, Error> {
139    let text = std::fs::read_to_string(path).map_err(|e| Error::Base {
140        message: format!("read `{}`: {e}", path.display()),
141    })?;
142    let mut base: Value = serde_yaml::from_str(&text).map_err(|e| Error::Base {
143        message: format!("parse `{}`: {e}", path.display()),
144    })?;
145    drop_nulls(&mut base);
146    if base.is_null() {
147        base = Value::Object(serde_json::Map::new());
148    }
149
150    let overwrite = match env {
151        EnvPrecedence::Disabled => return Ok(base),
152        EnvPrecedence::AfterFile => true,
153        EnvPrecedence::BeforeFile => false,
154    };
155    apply_datadog_env(&mut base, overwrite).map_err(|message| Error::Base { message })?;
156    saluki_env_overlay::apply_env(&mut base, overwrite).map_err(|message| Error::Base { message })?;
157    Ok(base)
158}
159
160/// Recursively removes object entries whose value is JSON null, mirroring the compatibility loader's
161/// file normalization: an explicitly null YAML key must not override a model default with null.
162fn drop_nulls(value: &mut Value) {
163    if let Value::Object(map) = value {
164        map.retain(|_, v| !v.is_null());
165        for v in map.values_mut() {
166            drop_nulls(v);
167        }
168    }
169}
170
171/// Builds the by-key configuration view at the given precedence, with later sources overriding
172/// earlier ones.
173///
174/// Two environment providers are used together. [`ConfigurationLoader::from_environment`] scans the
175/// `DD_` prefix and contributes every variable as a flat key, which is what a key not covered by
176/// either source model needs. [`EnvironmentProvider`] then contributes the modeled keys at their
177/// canonical paths, so a nested key such as `proxy.http` is reachable in the shape the Datadog Agent
178/// itself uses. It sits at the higher precedence of the two because it knows a key's real shape,
179/// while the scanning provider can only guess from the variable's name.
180fn build_loader(path: &Path, env: EnvPrecedence) -> Result<ConfigurationLoader, Error> {
181    let loader = ConfigurationLoader::default();
182    let loader = match env {
183        EnvPrecedence::AfterFile => loader
184            .from_yaml(path)?
185            .from_environment(ENV_VAR_PREFIX)?
186            .add_providers([schema_env_provider()?]),
187        EnvPrecedence::BeforeFile => loader
188            .from_environment(ENV_VAR_PREFIX)?
189            .add_providers([schema_env_provider()?])
190            .from_yaml(path)?,
191        EnvPrecedence::Disabled => loader.from_yaml(path)?,
192    };
193    Ok(loader)
194}
195
196/// Builds the schema-driven environment provider, reporting a malformed value the same way the typed
197/// base does.
198fn schema_env_provider() -> Result<EnvironmentProvider, Error> {
199    EnvironmentProvider::new().map_err(|message| Error::Base { message })
200}
201
202#[cfg(test)]
203mod tests {
204    use bytesize::ByteSize;
205    use saluki_config::test_env_lock;
206    use serde_json::json;
207
208    use super::*;
209
210    #[test]
211    fn build_base_composes_file_and_environment_by_precedence() {
212        let _guard = test_env_lock();
213        let path = std::env::temp_dir().join(format!("adp_build_base_{}.yaml", std::process::id()));
214        std::fs::write(
215            &path,
216            "dogstatsd_port: 8125\ndogstatsd_non_local_traffic: false\nempty_key:\n",
217        )
218        .unwrap();
219        std::env::set_var("DD_DOGSTATSD_PORT", "9125");
220
221        // AfterFile: the environment wins over the file, the decoded value is a real number, an
222        // explicitly null key is dropped, and an unrelated file key is preserved.
223        let base = build_base(&path, EnvPrecedence::AfterFile).expect("base builds");
224        assert_eq!(base.get("dogstatsd_port"), Some(&json!(9125)));
225        assert_eq!(base.get("dogstatsd_non_local_traffic"), Some(&json!(false)));
226        assert!(base.get("empty_key").is_none());
227
228        // BeforeFile: the file wins over the environment.
229        let base = build_base(&path, EnvPrecedence::BeforeFile).expect("base builds");
230        assert_eq!(base.get("dogstatsd_port"), Some(&json!(8125)));
231
232        // Disabled: the environment is ignored entirely.
233        let base = build_base(&path, EnvPrecedence::Disabled).expect("base builds");
234        assert_eq!(base.get("dogstatsd_port"), Some(&json!(8125)));
235
236        std::env::remove_var("DD_DOGSTATSD_PORT");
237        std::fs::remove_file(&path).ok();
238    }
239
240    #[tokio::test]
241    async fn local_exposes_translated_configuration() {
242        // Disable environment reads so this test does not need `ENV_MUTEX`.
243        let path = std::env::temp_dir().join(format!("adp_local_{}.yaml", std::process::id()));
244        std::fs::write(&path, "log_level: warn\ndogstatsd_port: 9125\n").unwrap();
245
246        let loaded = LoadedConfiguration::load(&path, EnvPrecedence::Disabled)
247            .await
248            .expect("local sources load");
249        let config = loaded.local();
250
251        assert_eq!(config.control.logging.level, "warn");
252        assert_eq!(config.domains.dogstatsd.listeners.port, 9125);
253
254        std::fs::remove_file(&path).ok();
255    }
256
257    #[tokio::test]
258    async fn load_rejects_translation_invalid_local_sources() {
259        let path = std::env::temp_dir().join(format!("adp_local_bad_{}.yaml", std::process::id()));
260        // The compatibility loader accepts this value, but typed translation rejects it.
261        std::fs::write(&path, "dogstatsd_tag_cardinality: bogus\n").unwrap();
262
263        let result = LoadedConfiguration::load(&path, EnvPrecedence::Disabled).await;
264
265        std::fs::remove_file(&path).ok();
266        assert!(matches!(result, Err(Error::Translate { .. })));
267    }
268
269    #[tokio::test]
270    async fn load_leaves_an_incomplete_local_snapshot_to_the_selected_authority() {
271        // A local snapshot with no API key is normal: under the Datadog Agent the key arrives over the
272        // configuration stream, and the CLI subcommands read this snapshot without submitting
273        // anything. Validation therefore belongs to whichever authority the caller then selects.
274        let path = std::env::temp_dir().join(format!("adp_no_api_key_{}.yaml", std::process::id()));
275        std::fs::write(&path, "log_level: warn\n").unwrap();
276
277        let loaded = LoadedConfiguration::load(&path, EnvPrecedence::Disabled)
278            .await
279            .expect("a local snapshot without an API key loads");
280        assert_eq!("", loaded.local().shared.endpoints.api_key);
281
282        // Standalone mode makes that same snapshot authoritative, so the missing key is fatal there.
283        let result = loaded.standalone().await;
284
285        std::fs::remove_file(&path).ok();
286        assert!(matches!(result, Err(Error::MissingApiKey)));
287    }
288
289    // `LoadedConfiguration::load` is `async` only for symmetry with the rest of the API; it awaits
290    // nothing. The environment tests below drive it on a local runtime rather than with
291    // `#[tokio::test]`, so the blocking environment guard is never held across an await point.
292    fn block_on<F: std::future::Future>(future: F) -> F::Output {
293        tokio::runtime::Builder::new_current_thread()
294            .build()
295            .expect("runtime builds")
296            .block_on(future)
297    }
298
299    #[test]
300    fn a_saluki_only_environment_variable_reaches_the_model() {
301        // End to end for a key the Datadog schema does not declare: `DD_DATA_PLANE_STANDALONE_MODE`
302        // is read at its canonical path by the Saluki-only reader and seeds `control.standalone_mode`.
303        let _guard = test_env_lock();
304        let path = std::env::temp_dir().join(format!("adp_saluki_env_{}.yaml", std::process::id()));
305        std::fs::write(&path, "{}\n").unwrap();
306        std::env::set_var("DD_DATA_PLANE_STANDALONE_MODE", "true");
307
308        let loaded = block_on(LoadedConfiguration::load(&path, EnvPrecedence::AfterFile)).expect("local sources load");
309
310        std::env::remove_var("DD_DATA_PLANE_STANDALONE_MODE");
311        std::fs::remove_file(&path).ok();
312        assert!(loaded.local().control.standalone_mode);
313    }
314
315    #[test]
316    fn ottl_filter_error_mode_environment_variable_reaches_the_model() {
317        // `DD_OTTL_FILTER_CONFIG_ERROR_MODE` exercises the enum leaf: the environment value is a
318        // plain string, decoded as a Saluki-only leaf, then validated during configuration
319        // deserialization.
320        use agent_data_plane_config::domains::traces::OttlErrorMode;
321
322        let _guard = test_env_lock();
323        let path = std::env::temp_dir().join(format!("adp_ottl_env_{}.yaml", std::process::id()));
324        std::fs::write(&path, "{}\n").unwrap();
325        std::env::set_var("DD_OTTL_FILTER_CONFIG_ERROR_MODE", "silent");
326
327        let loaded = block_on(LoadedConfiguration::load(&path, EnvPrecedence::AfterFile)).expect("local sources load");
328
329        std::env::remove_var("DD_OTTL_FILTER_CONFIG_ERROR_MODE");
330        std::fs::remove_file(&path).ok();
331        assert_eq!(
332            loaded.local().domains.traces.ottl_filter.error_mode,
333            OttlErrorMode::Silent
334        );
335    }
336
337    #[test]
338    fn a_structured_saluki_only_environment_variable_reaches_the_model() {
339        let _guard = test_env_lock();
340        let path = std::env::temp_dir().join(format!("adp_saluki_structured_env_{}.yaml", std::process::id()));
341        std::fs::write(&path, "{}\n").unwrap();
342        std::env::set_var(
343            "DD_METRIC_TAG_VALUE_ALLOWLIST",
344            r#"[{"metric_prefix":"requests.","tag_name":"customer_id","values":["customer-1"],"on_miss":"replace","replacement":"other"}]"#,
345        );
346
347        let loaded = block_on(LoadedConfiguration::load(&path, EnvPrecedence::AfterFile)).expect("local sources load");
348
349        std::env::remove_var("DD_METRIC_TAG_VALUE_ALLOWLIST");
350        std::fs::remove_file(&path).ok();
351        let entries = &loaded.local().domains.dogstatsd.tag_value_allowlist;
352        assert_eq!(entries.len(), 1);
353        assert_eq!(entries[0].metric_prefix, "requests.");
354        assert_eq!(entries[0].tag_name, "customer_id");
355        assert_eq!(entries[0].values, ["customer-1"]);
356        assert_eq!(
357            entries[0].on_miss,
358            agent_data_plane_config::domains::dogstatsd::TagValueMismatchAction::Replace
359        );
360        assert_eq!(entries[0].replacement, "other");
361    }
362
363    #[test]
364    fn metric_tag_value_allowlist_from_file_preserves_whitespace() {
365        let path = std::env::temp_dir().join(format!("adp_tag_value_allowlist_{}.yaml", std::process::id()));
366        std::fs::write(
367            &path,
368            "metric_tag_value_allowlist:\n  - metric_prefix: ' requests. '\n    tag_name: ' customer_id '\n    values: [' customer-1 ']\n    on_miss: replace\n    replacement: ' other '\n",
369        )
370        .unwrap();
371
372        let loaded = block_on(LoadedConfiguration::load(&path, EnvPrecedence::Disabled))
373            .expect("allow-list strings containing whitespace should load");
374
375        std::fs::remove_file(&path).ok();
376        let entries = &loaded.local().domains.dogstatsd.tag_value_allowlist;
377        assert_eq!(entries.len(), 1);
378        assert_eq!(entries[0].metric_prefix, " requests. ");
379        assert_eq!(entries[0].tag_name, " customer_id ");
380        assert_eq!(entries[0].values, [" customer-1 "]);
381        assert_eq!(entries[0].replacement, " other ");
382    }
383
384    #[test]
385    fn invalid_metric_tag_value_allowlist_from_environment_fails_load() {
386        let _guard = test_env_lock();
387        let path = std::env::temp_dir().join(format!("adp_bad_tag_value_allowlist_env_{}.yaml", std::process::id()));
388        std::fs::write(&path, "{}\n").unwrap();
389        std::env::set_var(
390            "DD_METRIC_TAG_VALUE_ALLOWLIST",
391            r#"[{"metric_prefix":"requests.","tag_name":"customer_id"},{"metric_prefix":"requests.api.","tag_name":"customer_id"}]"#,
392        );
393
394        let result = block_on(LoadedConfiguration::load(&path, EnvPrecedence::AfterFile));
395
396        std::env::remove_var("DD_METRIC_TAG_VALUE_ALLOWLIST");
397        std::fs::remove_file(&path).ok();
398        let error = match result {
399            Err(error) => error,
400            Ok(_) => panic!("overlapping environment allow-list should fail loading"),
401        };
402        assert!(matches!(error, Error::Deserialize { .. }));
403        assert!(error.to_string().contains("overlapping metric prefixes"));
404    }
405
406    #[test]
407    fn a_nested_datadog_environment_variable_reaches_the_by_key_view() {
408        // `DD_PROXY_HTTP` names a nested key, which Figment's prefix scan cannot place. The
409        // schema-driven provider resolves it, so the by-key view serves it at `proxy.http`.
410        let _guard = test_env_lock();
411        let path = std::env::temp_dir().join(format!("adp_bykey_env_{}.yaml", std::process::id()));
412        std::fs::write(&path, "{}\n").unwrap();
413        std::env::set_var("DD_PROXY_HTTP", "http://proxy.example.com");
414
415        let loaded = block_on(LoadedConfiguration::load(&path, EnvPrecedence::AfterFile)).expect("local sources load");
416        let raw = loaded.raw_config();
417
418        std::env::remove_var("DD_PROXY_HTTP");
419        std::fs::remove_file(&path).ok();
420        assert_eq!(
421            raw.try_get_typed::<String>("proxy.http").expect("key reads"),
422            Some("http://proxy.example.com".to_string())
423        );
424    }
425
426    #[test]
427    fn the_adp_zstd_override_reaches_both_views_from_the_environment() {
428        // The documented environment variable must produce the same canonical nested path in the
429        // by-key view and the typed model.
430        let _guard = test_env_lock();
431        let path = std::env::temp_dir().join(format!("adp_zstd_env_{}.yaml", std::process::id()));
432        std::fs::write(&path, "{}\n").unwrap();
433        std::env::set_var("DD_DATA_PLANE_SERIALIZER_ZSTD_COMPRESSOR_LEVEL", "7");
434
435        let loaded = block_on(LoadedConfiguration::load(&path, EnvPrecedence::AfterFile)).expect("local sources load");
436        let from_by_key = loaded
437            .raw_config()
438            .try_get_typed::<i32>("data_plane.serializer_zstd_compressor_level")
439            .expect("key reads");
440        let from_typed = loaded.local().shared.endpoints.compression.effective_zstd_level();
441
442        std::env::remove_var("DD_DATA_PLANE_SERIALIZER_ZSTD_COMPRESSOR_LEVEL");
443        std::fs::remove_file(&path).ok();
444        assert_eq!(from_by_key, Some(7));
445        assert_eq!(from_typed, 7);
446    }
447
448    #[test]
449    fn build_base_rejects_a_malformed_environment_value() {
450        let _guard = test_env_lock();
451        let path = std::env::temp_dir().join(format!("adp_build_base_bad_{}.yaml", std::process::id()));
452        std::fs::write(&path, "dogstatsd_port: 8125\n").unwrap();
453        std::env::set_var("DD_DOGSTATSD_PORT", "not-a-number");
454
455        let result = build_base(&path, EnvPrecedence::AfterFile);
456
457        std::env::remove_var("DD_DOGSTATSD_PORT");
458        std::fs::remove_file(&path).ok();
459        assert!(matches!(result, Err(Error::Base { .. })));
460    }
461
462    #[test]
463    fn build_base_accepts_a_human_readable_dogstatsd_interner_size() {
464        let _guard = test_env_lock();
465        let path = std::env::temp_dir().join(format!("adp_build_base_interner_{}.yaml", std::process::id()));
466        std::fs::write(&path, "{}\n").unwrap();
467        std::env::set_var("DD_DOGSTATSD_STRING_INTERNER_SIZE_BYTES", "12MiB");
468
469        let base = SourceTree::all_explicit(build_base(&path, EnvPrecedence::AfterFile).expect("base builds"));
470        let config = translate_strict(&base).expect("human-readable byte size translates");
471
472        std::env::remove_var("DD_DOGSTATSD_STRING_INTERNER_SIZE_BYTES");
473        std::fs::remove_file(&path).ok();
474        assert_eq!(
475            config.domains.dogstatsd.contexts.string_interner_size_bytes,
476            Some(ByteSize::mib(12).as_u64())
477        );
478    }
479}
agent_data_plane_config_system/loaded.rs

agent_data_plane_config_system/
loaded.rs
