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 serde_json::Value;
15use tokio::sync::mpsc;
16
17use crate::saluki_env_overlay;
18use crate::source::SourceTree;
19use crate::system::{translate_strict, validate, ConfigurationSystem, ConfigurationUpdates, Error};
20
21/// Where environment variables sit relative to the configuration file.
22#[derive(Clone, Copy, Debug, Eq, PartialEq)]
23pub enum EnvPrecedence {
24    /// Environment variables are read below the file: the file wins.
25    ///
26    /// The file takes precedence over environment variables.
27    BeforeFile,
28    /// Environment variables are read after the file: the environment wins. Matches the Datadog
29    /// Agent's precedence and is the value ADP uses.
30    AfterFile,
31    /// Environment variables are not read.
32    Disabled,
33}
34
35/// Local configuration prepared before a runtime authority is selected.
36pub struct LoadedConfiguration {
37    // Nested local base used for typed translation and Agent-layer merges.
38    base: SourceTree,
39    // Strictly translated local snapshot exposed before authority selection and used by standalone
40    // mode.
41    local: SalukiConfiguration,
42}
43
44impl LoadedConfiguration {
45    /// Loads and strictly translates the local file and environment using the requested precedence.
46    ///
47    /// The snapshot is translated but not validated: it is not yet authoritative. Under the Datadog
48    /// Agent the stream still has settings to contribute, so a local snapshot that could not run on
49    /// its own is normal here. Whichever authority [`run`](Self::run) or
50    /// [`standalone`](Self::standalone) selects applies validation.
51    ///
52    /// # Errors
53    ///
54    /// Returns an error if a local source cannot be read, decoded, deserialized, or translated.
55    pub async fn load(path: impl AsRef<Path>, env: EnvPrecedence) -> Result<Self, Error> {
56        // Every value the local sources supply was set explicitly: the file set it or an environment
57        // variable supplied it.
58        let base = SourceTree::all_explicit(build_base(path.as_ref(), env)?);
59        let local = translate_strict(&base)?;
60        Ok(Self { base, local })
61    }
62
63    /// Returns the typed snapshot of the local file and environment.
64    ///
65    /// Values from the Datadog Agent stream have not been applied to this snapshot.
66    pub fn local(&self) -> &SalukiConfiguration {
67        &self.local
68    }
69
70    /// Uses the Datadog Agent's configuration stream as the runtime authority.
71    ///
72    /// Waits for the initial Agent snapshot, layers it over the local sources, and strictly
73    /// translates the result. Returns the system together with the [`ConfigurationUpdates`] that
74    /// apply each later update. The caller must run them.
75    ///
76    /// # Errors
77    ///
78    /// Returns an error if the stream closes before its initial snapshot, or the merged configuration
79    /// cannot be deserialized or translated.
80    pub async fn run(
81        self, config_stream: mpsc::Receiver<ConfigUpdate>,
82    ) -> Result<(ConfigurationSystem, ConfigurationUpdates), Error> {
83        ConfigurationSystem::connected(config_stream, self.base).await
84    }
85
86    /// Uses the translated local configuration as the runtime authority.
87    ///
88    /// No configuration stream or update task is created. Because nothing further will be layered on,
89    /// the local snapshot is validated here, where [`run`](Self::run) instead validates the merged
90    /// result of the Agent's initial snapshot.
91    ///
92    /// # Errors
93    ///
94    /// Returns an error if the local configuration fails validation.
95    pub async fn standalone(self) -> Result<ConfigurationSystem, Error> {
96        validate(&self.local)?;
97        Ok(ConfigurationSystem::standalone(self.local, self.base))
98    }
99}
100
101/// Builds the typed base: the configuration file parsed to its nested shape, with environment
102/// variables read directly and decoded into the schema's shapes on top.
103///
104/// A null-valued key is dropped and an empty file is an empty object. The environment is then
105/// overlaid via the generated Datadog reader, the Saluki-only reader, and the canonical proxy
106/// variables. `env` sets whether the environment overwrites the file (`AfterFile`) or only fills
107/// absent keys (`BeforeFile`); `Disabled` skips the environment entirely.
108fn build_base(path: &Path, env: EnvPrecedence) -> Result<Value, Error> {
109    let text = std::fs::read_to_string(path).map_err(|e| Error::Base {
110        message: format!("read `{}`: {e}", path.display()),
111    })?;
112    let mut base: Value = saluki_config::yaml::from_str(&text).map_err(|e| Error::Base {
113        message: format!("parse `{}`: {e}", path.display()),
114    })?;
115    drop_nulls(&mut base);
116    if base.is_null() {
117        base = Value::Object(serde_json::Map::new());
118    }
119
120    let overwrite = match env {
121        EnvPrecedence::Disabled => return Ok(base),
122        EnvPrecedence::AfterFile => true,
123        EnvPrecedence::BeforeFile => false,
124    };
125    apply_datadog_env(&mut base, overwrite).map_err(|message| Error::Base { message })?;
126    saluki_env_overlay::apply_env(&mut base, overwrite).map_err(|message| Error::Base { message })?;
127    Ok(base)
128}
129
130/// Recursively removes object entries whose value is JSON null: an explicitly null YAML key must not
131/// override a model default with null.
132fn drop_nulls(value: &mut Value) {
133    if let Value::Object(map) = value {
134        map.retain(|_, v| !v.is_null());
135        for v in map.values_mut() {
136            drop_nulls(v);
137        }
138    }
139}
140
141#[cfg(test)]
142mod tests {
143    use bytesize::ByteSize;
144    use saluki_config::dynamic::ConfigSetting;
145    use saluki_config::test_env_lock;
146    use serde_json::json;
147
148    use super::*;
149
150    #[test]
151    fn build_base_composes_file_and_environment_by_precedence() {
152        let _guard = test_env_lock();
153        let path = std::env::temp_dir().join(format!("adp_build_base_{}.yaml", std::process::id()));
154        std::fs::write(
155            &path,
156            "dogstatsd_port: 8125\ndogstatsd_non_local_traffic: false\nempty_key:\n",
157        )
158        .unwrap();
159        std::env::set_var("DD_DOGSTATSD_PORT", "9125");
160
161        // AfterFile: the environment wins over the file, the decoded value is a real number, an
162        // explicitly null key is dropped, and an unrelated file key is preserved.
163        let base = build_base(&path, EnvPrecedence::AfterFile).expect("base builds");
164        assert_eq!(base.get("dogstatsd_port"), Some(&json!(9125)));
165        assert_eq!(base.get("dogstatsd_non_local_traffic"), Some(&json!(false)));
166        assert!(base.get("empty_key").is_none());
167
168        // BeforeFile: the file wins over the environment.
169        let base = build_base(&path, EnvPrecedence::BeforeFile).expect("base builds");
170        assert_eq!(base.get("dogstatsd_port"), Some(&json!(8125)));
171
172        // Disabled: the environment is ignored entirely.
173        let base = build_base(&path, EnvPrecedence::Disabled).expect("base builds");
174        assert_eq!(base.get("dogstatsd_port"), Some(&json!(8125)));
175
176        std::env::remove_var("DD_DOGSTATSD_PORT");
177        std::fs::remove_file(&path).ok();
178    }
179
180    #[tokio::test]
181    async fn local_exposes_translated_configuration() {
182        // Disable environment reads so this test does not need `ENV_MUTEX`.
183        let path = std::env::temp_dir().join(format!("adp_local_{}.yaml", std::process::id()));
184        std::fs::write(&path, "log_level: warn\ndogstatsd_port: 9125\n").unwrap();
185
186        let loaded = LoadedConfiguration::load(&path, EnvPrecedence::Disabled)
187            .await
188            .expect("local sources load");
189        let config = loaded.local();
190
191        assert_eq!(config.control.logging.level, "warn");
192        assert_eq!(config.domains.dogstatsd.listeners.port, 9125);
193
194        std::fs::remove_file(&path).ok();
195    }
196
197    #[tokio::test]
198    async fn load_rejects_translation_invalid_local_sources() {
199        let path = std::env::temp_dir().join(format!("adp_local_bad_{}.yaml", std::process::id()));
200        // The file parses, but typed translation rejects the value.
201        std::fs::write(&path, "dogstatsd_tag_cardinality: bogus\n").unwrap();
202
203        let result = LoadedConfiguration::load(&path, EnvPrecedence::Disabled).await;
204
205        std::fs::remove_file(&path).ok();
206        assert!(matches!(result, Err(Error::Translate { .. })));
207    }
208
209    #[tokio::test]
210    async fn load_leaves_an_incomplete_local_snapshot_to_the_selected_authority() {
211        // A local snapshot with no API key is normal: under the Datadog Agent the key arrives over the
212        // configuration stream, and the CLI subcommands read this snapshot without submitting
213        // anything. Validation therefore belongs to whichever authority the caller then selects.
214        let path = std::env::temp_dir().join(format!("adp_no_api_key_{}.yaml", std::process::id()));
215        std::fs::write(&path, "log_level: warn\n").unwrap();
216
217        let loaded = LoadedConfiguration::load(&path, EnvPrecedence::Disabled)
218            .await
219            .expect("a local snapshot without an API key loads");
220        assert_eq!("", loaded.local().shared.endpoints.api_key);
221
222        // Standalone mode makes that same snapshot authoritative, so the missing key is fatal there.
223        let result = loaded.standalone().await;
224
225        std::fs::remove_file(&path).ok();
226        assert!(matches!(result, Err(Error::MissingApiKey)));
227    }
228
229    /// Loads the local sources from a temporary configuration file with the given contents.
230    ///
231    /// `test` names the file, so that tests which run at the same time do not share one.
232    async fn load_from_file(test: &str, contents: &str) -> LoadedConfiguration {
233        let path = std::env::temp_dir().join(format!("adp_{test}_{}.yaml", std::process::id()));
234        std::fs::write(&path, contents).unwrap();
235        let loaded = LoadedConfiguration::load(&path, EnvPrecedence::Disabled)
236            .await
237            .expect("local sources load");
238        std::fs::remove_file(&path).ok();
239        loaded
240    }
241
242    /// Waits until `system` reports `level` as the log level.
243    async fn await_log_level(system: &ConfigurationSystem, level: &str) {
244        tokio::time::timeout(std::time::Duration::from_secs(2), async {
245            while system.config().control.logging.level != level {
246                tokio::time::sleep(std::time::Duration::from_millis(5)).await;
247            }
248        })
249        .await
250        .unwrap_or_else(|_| panic!("the configuration should report the log level `{level}`"));
251    }
252
253    fn log_level_update(level: &str) -> ConfigUpdate {
254        ConfigUpdate::Partial(ConfigSetting::explicit("log_level", json!(level)))
255    }
256
257    #[tokio::test]
258    async fn run_applies_later_updates_only_once_the_updates_run() {
259        let loaded = load_from_file("run_updates", "api_key: test-api-key\nlog_level: info\n").await;
260
261        // `run` itself applies the initial snapshot. This is the startup gate.
262        let (agent_tx, agent_rx) = mpsc::channel(8);
263        agent_tx.send(ConfigUpdate::snapshot([])).await.unwrap();
264        let (system, updates) = loaded.run(agent_rx).await.expect("the initial snapshot is accepted");
265        assert_eq!(system.config().control.logging.level, "info");
266
267        // A later update waits until the updates run. Nothing applies it in the background before
268        // then.
269        agent_tx.send(log_level_update("warn")).await.unwrap();
270        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
271        assert_eq!(system.config().control.logging.level, "info");
272
273        tokio::spawn(updates.run());
274        await_log_level(&system, "warn").await;
275    }
276
277    #[tokio::test]
278    async fn updates_continue_where_a_stopped_run_left_off() {
279        let loaded = load_from_file("run_restart", "api_key: test-api-key\nlog_level: info\n").await;
280
281        let (agent_tx, agent_rx) = mpsc::channel(8);
282        agent_tx.send(ConfigUpdate::snapshot([])).await.unwrap();
283        let (system, updates) = loaded.run(agent_rx).await.expect("the initial snapshot is accepted");
284
285        let first_run = tokio::spawn(updates.run());
286        agent_tx.send(log_level_update("warn")).await.unwrap();
287        await_log_level(&system, "warn").await;
288
289        // Stop the run, as a supervisor does when it restarts a worker. The next run reads the same
290        // stream, so an update sent in between is not lost.
291        first_run.abort();
292        let _ = first_run.await;
293        agent_tx.send(log_level_update("error")).await.unwrap();
294
295        tokio::spawn(updates.run());
296        await_log_level(&system, "error").await;
297    }
298
299    #[tokio::test]
300    async fn updates_fail_once_the_stream_closes() {
301        let loaded = load_from_file("run_ended", "api_key: test-api-key\n").await;
302
303        let (agent_tx, agent_rx) = mpsc::channel(8);
304        agent_tx.send(ConfigUpdate::snapshot([])).await.unwrap();
305        let (_system, updates) = loaded.run(agent_rx).await.expect("the initial snapshot is accepted");
306        drop(agent_tx);
307
308        // The stream does not reopen, so running the updates again fails in the same way.
309        assert!(matches!(updates.run().await, Err(Error::UpdateStreamClosed)));
310        assert!(matches!(updates.run().await, Err(Error::UpdateStreamClosed)));
311    }
312
313    #[tokio::test]
314    async fn run_fails_if_the_stream_closes_before_the_initial_snapshot() {
315        let loaded = load_from_file("run_closed", "api_key: test-api-key\n").await;
316
317        let (agent_tx, agent_rx) = mpsc::channel::<ConfigUpdate>(1);
318        drop(agent_tx);
319
320        let result = loaded.run(agent_rx).await;
321        assert!(matches!(result, Err(Error::StreamClosed)));
322    }
323
324    // `LoadedConfiguration::load` is `async` only for symmetry with the rest of the API; it awaits
325    // nothing. The environment tests below drive it on a local runtime rather than with
326    // `#[tokio::test]`, so the blocking environment guard is never held across an await point.
327    fn block_on<F: std::future::Future>(future: F) -> F::Output {
328        tokio::runtime::Builder::new_current_thread()
329            .enable_time()
330            .build()
331            .expect("runtime builds")
332            .block_on(future)
333    }
334
335    #[test]
336    fn a_saluki_only_environment_variable_reaches_the_model() {
337        // End to end for a key the Datadog schema does not declare: `DD_DATA_PLANE_STANDALONE_MODE`
338        // is read at its canonical path by the Saluki-only reader and seeds `control.standalone_mode`.
339        let _guard = test_env_lock();
340        let path = std::env::temp_dir().join(format!("adp_saluki_env_{}.yaml", std::process::id()));
341        std::fs::write(&path, "{}\n").unwrap();
342        std::env::set_var("DD_DATA_PLANE_STANDALONE_MODE", "true");
343
344        let loaded = block_on(LoadedConfiguration::load(&path, EnvPrecedence::AfterFile)).expect("local sources load");
345
346        std::env::remove_var("DD_DATA_PLANE_STANDALONE_MODE");
347        std::fs::remove_file(&path).ok();
348        assert!(loaded.local().control.standalone_mode);
349    }
350
351    #[test]
352    fn trace_env_environment_variables_reach_the_model() {
353        // The trace environment picks its source by whether a key was configured. An empty variable is
354        // unset, as in the Agent, so it must not block a lower-precedence source. Global tags reach
355        // their lists unchanged; nothing here derives an environment from them.
356        let _guard = test_env_lock();
357        let path = std::env::temp_dir().join(format!("adp_trace_env_{}.yaml", std::process::id()));
358        std::fs::write(&path, "{}\n").unwrap();
359        std::env::set_var("DD_TAGS", "a:b env:tagged");
360        std::env::set_var("DD_EXTRA_TAGS", "env:extra");
361
362        std::env::set_var("DD_APM_ENV", "apm");
363        std::env::set_var("DD_ENV", "prod");
364        let set = block_on(LoadedConfiguration::load(&path, EnvPrecedence::AfterFile)).expect("local sources load");
365
366        std::env::set_var("DD_APM_ENV", "");
367        std::env::set_var("DD_ENV", "");
368        let empty = block_on(LoadedConfiguration::load(&path, EnvPrecedence::AfterFile)).expect("local sources load");
369
370        for var in ["DD_APM_ENV", "DD_ENV", "DD_TAGS", "DD_EXTRA_TAGS"] {
371            std::env::remove_var(var);
372        }
373        std::fs::remove_file(&path).ok();
374
375        let traces = &set.local().domains.traces;
376        assert_eq!(traces.apm_env.as_deref(), Some("apm"));
377        assert_eq!(traces.env.as_deref(), Some("prod"));
378
379        let traces = &empty.local().domains.traces;
380        assert_eq!(traces.apm_env, None);
381        assert_eq!(traces.env, None);
382        let tags = &empty.local().shared.tags;
383        assert_eq!(tags.tags, ["a:b", "env:tagged"]);
384        assert_eq!(tags.extra_tags, ["env:extra"]);
385    }
386
387    #[test]
388    fn ottl_filter_error_mode_environment_variable_reaches_the_model() {
389        // `DD_OTTL_FILTER_CONFIG_ERROR_MODE` exercises the enum leaf: the environment value is a
390        // plain string, decoded as a Saluki-only leaf, then validated during configuration
391        // deserialization.
392        use agent_data_plane_config::domains::traces::OttlErrorMode;
393
394        let _guard = test_env_lock();
395        let path = std::env::temp_dir().join(format!("adp_ottl_env_{}.yaml", std::process::id()));
396        std::fs::write(&path, "{}\n").unwrap();
397        std::env::set_var("DD_OTTL_FILTER_CONFIG_ERROR_MODE", "silent");
398
399        let loaded = block_on(LoadedConfiguration::load(&path, EnvPrecedence::AfterFile)).expect("local sources load");
400
401        std::env::remove_var("DD_OTTL_FILTER_CONFIG_ERROR_MODE");
402        std::fs::remove_file(&path).ok();
403        assert_eq!(
404            loaded.local().domains.traces.ottl_filter.error_mode,
405            OttlErrorMode::Silent
406        );
407    }
408
409    #[test]
410    fn a_structured_saluki_only_environment_variable_reaches_the_model() {
411        let _guard = test_env_lock();
412        let path = std::env::temp_dir().join(format!("adp_saluki_structured_env_{}.yaml", std::process::id()));
413        std::fs::write(&path, "{}\n").unwrap();
414        std::env::set_var(
415            "DD_METRIC_TAG_VALUE_ALLOWLIST",
416            r#"[{"metric_prefix":"requests.","tag_name":"customer_id","values":["customer-1"],"on_miss":"replace","replacement":"other"}]"#,
417        );
418
419        let loaded = block_on(LoadedConfiguration::load(&path, EnvPrecedence::AfterFile)).expect("local sources load");
420
421        std::env::remove_var("DD_METRIC_TAG_VALUE_ALLOWLIST");
422        std::fs::remove_file(&path).ok();
423        let entries = &loaded.local().domains.dogstatsd.tag_value_allowlist;
424        assert_eq!(entries.len(), 1);
425        assert_eq!(entries[0].metric_prefix, "requests.");
426        assert_eq!(entries[0].tag_name, "customer_id");
427        assert_eq!(entries[0].values, ["customer-1"]);
428        assert_eq!(
429            entries[0].on_miss,
430            agent_data_plane_config::domains::dogstatsd::TagValueMismatchAction::Replace
431        );
432        assert_eq!(entries[0].replacement, "other");
433    }
434
435    #[test]
436    fn metric_tag_value_allowlist_from_file_preserves_whitespace() {
437        let path = std::env::temp_dir().join(format!("adp_tag_value_allowlist_{}.yaml", std::process::id()));
438        std::fs::write(
439            &path,
440            "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",
441        )
442        .unwrap();
443
444        let loaded = block_on(LoadedConfiguration::load(&path, EnvPrecedence::Disabled))
445            .expect("allow-list strings containing whitespace should load");
446
447        std::fs::remove_file(&path).ok();
448        let entries = &loaded.local().domains.dogstatsd.tag_value_allowlist;
449        assert_eq!(entries.len(), 1);
450        assert_eq!(entries[0].metric_prefix, " requests. ");
451        assert_eq!(entries[0].tag_name, " customer_id ");
452        assert_eq!(entries[0].values, [" customer-1 "]);
453        assert_eq!(entries[0].replacement, " other ");
454    }
455
456    #[test]
457    fn invalid_metric_tag_value_allowlist_from_environment_fails_load() {
458        let _guard = test_env_lock();
459        let path = std::env::temp_dir().join(format!("adp_bad_tag_value_allowlist_env_{}.yaml", std::process::id()));
460        std::fs::write(&path, "{}\n").unwrap();
461        std::env::set_var(
462            "DD_METRIC_TAG_VALUE_ALLOWLIST",
463            r#"[{"metric_prefix":"requests.","tag_name":"customer_id"},{"metric_prefix":"requests.api.","tag_name":"customer_id"}]"#,
464        );
465
466        let result = block_on(LoadedConfiguration::load(&path, EnvPrecedence::AfterFile));
467
468        std::env::remove_var("DD_METRIC_TAG_VALUE_ALLOWLIST");
469        std::fs::remove_file(&path).ok();
470        let error = match result {
471            Err(error) => error,
472            Ok(_) => panic!("overlapping environment allow-list should fail loading"),
473        };
474        assert!(matches!(error, Error::Deserialize { .. }));
475        assert!(error.to_string().contains("overlapping metric prefixes"));
476    }
477
478    #[test]
479    fn the_adp_zstd_override_reaches_the_model_from_the_environment() {
480        // The documented environment variable for a Saluki-only key must reach the model, since no
481        // Datadog schema key covers it.
482        let _guard = test_env_lock();
483        let path = std::env::temp_dir().join(format!("adp_zstd_env_{}.yaml", std::process::id()));
484        std::fs::write(&path, "{}\n").unwrap();
485        std::env::set_var("DD_DATA_PLANE_SERIALIZER_ZSTD_COMPRESSOR_LEVEL", "7");
486
487        let loaded = block_on(LoadedConfiguration::load(&path, EnvPrecedence::AfterFile)).expect("local sources load");
488        let level = loaded.local().shared.endpoints.compression.effective_zstd_level();
489
490        std::env::remove_var("DD_DATA_PLANE_SERIALIZER_ZSTD_COMPRESSOR_LEVEL");
491        std::fs::remove_file(&path).ok();
492        assert_eq!(level, 7);
493    }
494
495    #[test]
496    fn the_adp_stop_timeout_override_reaches_the_model_from_the_environment() {
497        let _guard = test_env_lock();
498        let path = std::env::temp_dir().join(format!("adp_stop_timeout_env_{}.yaml", std::process::id()));
499        std::fs::write(&path, "{}\n").unwrap();
500        std::env::set_var("DD_DATA_PLANE_STOP_TIMEOUT", "45");
501
502        let loaded = block_on(LoadedConfiguration::load(&path, EnvPrecedence::AfterFile)).expect("local sources load");
503        let stop_timeout = loaded.local().control.stop_timeout;
504
505        std::env::remove_var("DD_DATA_PLANE_STOP_TIMEOUT");
506        std::fs::remove_file(&path).ok();
507        assert_eq!(stop_timeout, Some(std::time::Duration::from_secs(45)));
508    }
509
510    /// Sets process environment variables for the duration of a scope, removing them when it drops.
511    ///
512    /// The caller holds [`test_env_lock`] for at least as long.
513    struct ScopedEnv(Vec<&'static str>);
514
515    impl ScopedEnv {
516        fn set(vars: &[(&'static str, &str)]) -> Self {
517            for (name, value) in vars {
518                std::env::set_var(name, value);
519            }
520            Self(vars.iter().map(|(name, _)| *name).collect())
521        }
522    }
523
524    impl Drop for ScopedEnv {
525        fn drop(&mut self) {
526            for name in &self.0 {
527                std::env::remove_var(name);
528            }
529        }
530    }
531
532    /// Loads a temporary configuration file with the environment applied after it.
533    fn load_with_environment(
534        test: &str, contents: &str, vars: &[(&'static str, &str)],
535    ) -> Result<LoadedConfiguration, Error> {
536        let _guard = test_env_lock();
537        let path = std::env::temp_dir().join(format!("adp_{test}_{}.yaml", std::process::id()));
538        std::fs::write(&path, contents).unwrap();
539        let _env = ScopedEnv::set(vars);
540
541        let loaded = block_on(LoadedConfiguration::load(&path, EnvPrecedence::AfterFile));
542
543        std::fs::remove_file(&path).ok();
544        loaded
545    }
546
547    #[test]
548    fn agent_spellings_of_valid_local_inputs_load_and_survive_the_stream() {
549        // Each input is one the Agent accepts and ADP once rejected or misread: base-prefixed integers
550        // and a JSON list in the environment, and a quoted zero-padded integer in the file.
551        let loaded = load_with_environment(
552            "agent_spellings",
553            "api_key: test-api-key\ndogstatsd_buffer_size: \"010\"\n",
554            &[
555                ("DD_DOGSTATSD_PORT", "0x10"),
556                ("DD_AGENT_IPC_GRPC_MAX_MESSAGE_SIZE", "0x10"),
557                ("DD_CMD_PORT", "0x1F90"),
558                ("DD_APM_OBFUSCATION_ELASTICSEARCH_KEEP_VALUES", r#"["cr-a","cr-b"]"#),
559            ],
560        )
561        .expect("the Agent's spellings load");
562
563        let local = loaded.local();
564        assert_eq!(local.domains.dogstatsd.listeners.port, 16);
565        assert_eq!(local.control.ipc.grpc_max_message_size, 16);
566        assert_eq!(local.control.ipc.cmd_port, 8080);
567        assert_eq!(
568            local.domains.traces.obfuscation.elasticsearch.keep_values,
569            ["cr-a", "cr-b"]
570        );
571        assert_eq!(local.domains.dogstatsd.listeners.buffer_size, 8);
572
573        // The Agent's stream then connects and replaces a local value with its own typed one.
574        block_on(async {
575            let (agent_tx, agent_rx) = mpsc::channel(8);
576            agent_tx
577                .send(ConfigUpdate::snapshot([ConfigSetting::explicit(
578                    "dogstatsd_port",
579                    json!(9125),
580                )]))
581                .await
582                .unwrap();
583            let (system, _updates) = loaded.run(agent_rx).await.expect("the initial snapshot is accepted");
584
585            let config = system.config();
586            assert_eq!(config.domains.dogstatsd.listeners.port, 9125);
587            assert_eq!(config.control.ipc.cmd_port, 8080);
588            assert_eq!(config.domains.dogstatsd.listeners.buffer_size, 8);
589        });
590    }
591
592    /// Waits until `system` reports `port` as the DogStatsD port.
593    async fn await_dogstatsd_port(system: &ConfigurationSystem, port: u16) {
594        tokio::time::timeout(std::time::Duration::from_secs(2), async {
595            while system.config().domains.dogstatsd.listeners.port != port {
596                tokio::time::sleep(std::time::Duration::from_millis(5)).await;
597            }
598        })
599        .await
600        .unwrap_or_else(|_| panic!("the configuration should report the DogStatsD port {port}"));
601    }
602
603    #[test]
604    fn an_invalid_octal_integer_reads_as_zero_from_every_source() {
605        // The Agent starts with these values and its integer getters read them as 0, so ADP must
606        // neither stop at startup nor reject the stream's raw string.
607        let loaded = load_with_environment(
608            "invalid_octal",
609            "api_key: test-api-key\ndogstatsd_buffer_size: \"0128\"\n",
610            &[
611                ("DD_DOGSTATSD_PORT", "08"),
612                ("DD_CMD_PORT", "08.0"),
613                ("DD_AGENT_IPC_GRPC_MAX_MESSAGE_SIZE", "-09"),
614            ],
615        )
616        .expect("an invalid octal integer loads");
617
618        let local = loaded.local();
619        assert_eq!(local.domains.dogstatsd.listeners.port, 0);
620        assert_eq!(local.control.ipc.cmd_port, 0);
621        assert_eq!(local.control.ipc.grpc_max_message_size, 0);
622        assert_eq!(local.domains.dogstatsd.listeners.buffer_size, 0);
623
624        block_on(async {
625            // The stream forwards the Agent's raw strings, which connected startup accepts.
626            let (agent_tx, agent_rx) = mpsc::channel(8);
627            agent_tx
628                .send(ConfigUpdate::snapshot([
629                    ConfigSetting::explicit("dogstatsd_port", json!("010")),
630                    ConfigSetting::explicit("dogstatsd_buffer_size", json!("09")),
631                ]))
632                .await
633                .unwrap();
634            let (system, updates) = loaded.run(agent_rx).await.expect("the initial snapshot is accepted");
635            assert_eq!(system.config().domains.dogstatsd.listeners.port, 8);
636            assert_eq!(system.config().domains.dogstatsd.listeners.buffer_size, 0);
637
638            // A later update applies too. A rejected update would leave the last value in place.
639            tokio::spawn(updates.run());
640            agent_tx
641                .send(ConfigUpdate::Partial(ConfigSetting::explicit(
642                    "dogstatsd_port",
643                    json!("0x10"),
644                )))
645                .await
646                .unwrap();
647            await_dogstatsd_port(&system, 16).await;
648            agent_tx
649                .send(ConfigUpdate::Partial(ConfigSetting::explicit(
650                    "dogstatsd_port",
651                    json!("-08"),
652                )))
653                .await
654                .unwrap();
655            await_dogstatsd_port(&system, 0).await;
656        });
657    }
658
659    #[test]
660    fn a_malformed_integer_outside_the_invalid_octal_family_still_stops_startup() {
661        let result = load_with_environment("invalid_octal_boundary", "{}\n", &[("DD_DOGSTATSD_PORT", "08.")]);
662        match result {
663            Err(Error::Base { message }) => assert_eq!(
664                message,
665                "environment variable `DD_DOGSTATSD_PORT` (dogstatsd_port): invalid integer `08.`"
666            ),
667            Err(error) => panic!("expected a base error, got {error}"),
668            Ok(_) => panic!("`08.` should be rejected"),
669        }
670    }
671
672    #[test]
673    fn a_saluki_only_integer_keeps_its_decimal_grammar() {
674        // `dogstatsd_tcp_port` has no Agent counterpart, so a leading zero stays decimal.
675        let loaded = load_with_environment("saluki_integer", "{}\n", &[("DD_DOGSTATSD_TCP_PORT", "010")])
676            .expect("a zero-padded decimal loads");
677        assert_eq!(loaded.local().domains.dogstatsd.listeners.tcp_port, 10);
678
679        let loaded = load_with_environment("saluki_invalid_octal", "{}\n", &[("DD_DOGSTATSD_TCP_PORT", "08")])
680            .expect("a zero-padded decimal loads");
681        assert_eq!(loaded.local().domains.dogstatsd.listeners.tcp_port, 8);
682
683        let result = load_with_environment("saluki_hex", "{}\n", &[("DD_DOGSTATSD_TCP_PORT", "0x10")]);
684        match result {
685            Err(Error::Base { message }) => assert_eq!(
686                message,
687                "environment variable `DD_DOGSTATSD_TCP_PORT` (dogstatsd_tcp_port): invalid integer `0x10`"
688            ),
689            Err(error) => panic!("expected a base error, got {error}"),
690            Ok(_) => panic!("a hexadecimal Saluki-only integer should be rejected"),
691        }
692    }
693
694    #[test]
695    fn a_saluki_only_float_rejects_go_only_spellings() {
696        // `memory_slop_factor` has no Agent counterpart, so Go's hexadecimal float form stays invalid.
697        let result = load_with_environment("saluki_float", "{}\n", &[("DD_MEMORY_SLOP_FACTOR", "0x1p-2")]);
698        match result {
699            Err(Error::Base { message }) => assert_eq!(
700                message,
701                "environment variable `DD_MEMORY_SLOP_FACTOR` (memory_slop_factor): invalid number `0x1p-2`"
702            ),
703            Err(error) => panic!("expected a base error, got {error}"),
704            Ok(_) => panic!("a hexadecimal Saluki-only float should be rejected"),
705        }
706    }
707
708    #[test]
709    fn a_saluki_only_sequence_still_splits_on_whitespace() {
710        let loaded = load_with_environment(
711            "saluki_sequence",
712            "{}\n",
713            &[("DD_OTTL_FILTER_CONFIG_TRACES_SPAN", r#"["a","b"] c"#)],
714        )
715        .expect("a whitespace-separated sequence loads");
716        assert_eq!(
717            loaded.local().domains.traces.ottl_filter.span_conditions,
718            [r#"["a","b"]"#, "c"]
719        );
720    }
721
722    #[test]
723    fn build_base_rejects_a_malformed_environment_value() {
724        let _guard = test_env_lock();
725        let path = std::env::temp_dir().join(format!("adp_build_base_bad_{}.yaml", std::process::id()));
726        std::fs::write(&path, "dogstatsd_port: 8125\n").unwrap();
727        std::env::set_var("DD_DOGSTATSD_PORT", "not-a-number");
728
729        let result = build_base(&path, EnvPrecedence::AfterFile);
730
731        std::env::remove_var("DD_DOGSTATSD_PORT");
732        std::fs::remove_file(&path).ok();
733        assert!(matches!(result, Err(Error::Base { .. })));
734    }
735
736    #[test]
737    fn build_base_accepts_a_human_readable_dogstatsd_interner_size() {
738        let _guard = test_env_lock();
739        let path = std::env::temp_dir().join(format!("adp_build_base_interner_{}.yaml", std::process::id()));
740        std::fs::write(&path, "{}\n").unwrap();
741        std::env::set_var("DD_DOGSTATSD_STRING_INTERNER_SIZE_BYTES", "12MiB");
742
743        let base = SourceTree::all_explicit(build_base(&path, EnvPrecedence::AfterFile).expect("base builds"));
744        let config = translate_strict(&base).expect("human-readable byte size translates");
745
746        std::env::remove_var("DD_DOGSTATSD_STRING_INTERNER_SIZE_BYTES");
747        std::fs::remove_file(&path).ok();
748        assert_eq!(
749            config.domains.dogstatsd.contexts.string_interner_size_bytes,
750            Some(ByteSize::mib(12).as_u64())
751        );
752    }
753}
agent_data_plane_config_system/loaded.rs

agent_data_plane_config_system/
loaded.rs
