vector/config/loading/
mod.rs

1mod config_builder;
2mod loader;
3mod secret;
4mod source;
5
6use std::{
7    collections::HashMap,
8    fmt::Debug,
9    fs::{File, ReadDir},
10    path::{Path, PathBuf},
11    sync::{Mutex, OnceLock},
12};
13
14pub use config_builder::ConfigBuilderLoader;
15use glob::glob;
16use loader::process::Process;
17pub use loader::*;
18pub use secret::*;
19pub use source::*;
20use vector_lib::configurable::NamedComponent;
21
22use super::{
23    Config, ConfigPath, Format, FormatHint, ProviderConfig, builder::ConfigBuilder, format,
24    validation, vars,
25};
26use crate::signal;
27
28pub static CONFIG_PATHS: Mutex<Vec<ConfigPath>> = Mutex::new(Vec::new());
29
30static ALLOW_ENV_VAR_INTERPOLATION: OnceLock<bool> = OnceLock::new();
31
32/// Sets whether environment variable interpolation is enabled for the process.
33/// Must be called exactly once at startup before any config loading.
34pub fn set_env_var_interpolation(allow: bool) {
35    ALLOW_ENV_VAR_INTERPOLATION
36        .set(allow)
37        .expect("set_env_var_interpolation must only be called once");
38}
39
40/// Returns whether environment variable interpolation is currently enabled.
41/// Defaults to `false` if [`set_env_var_interpolation`] has not been called.
42pub fn env_var_interpolation_enabled() -> bool {
43    *ALLOW_ENV_VAR_INTERPOLATION.get().unwrap_or(&false)
44}
45
46pub(super) fn read_dir<P: AsRef<Path> + Debug>(path: P) -> Result<ReadDir, Vec<String>> {
47    path.as_ref()
48        .read_dir()
49        .map_err(|err| vec![format!("Could not read config dir: {:?}, {}.", path, err)])
50}
51
52pub(super) fn component_name<P: AsRef<Path> + Debug>(path: P) -> Result<String, Vec<String>> {
53    path.as_ref()
54        .file_stem()
55        .and_then(|name| name.to_str())
56        .map(|name| name.to_string())
57        .ok_or_else(|| vec![format!("Couldn't get component name for file: {:?}", path)])
58}
59
60pub(super) fn open_file<P: AsRef<Path> + Debug>(path: P) -> Option<File> {
61    match File::open(&path) {
62        Ok(f) => Some(f),
63        Err(error) => {
64            if let std::io::ErrorKind::NotFound = error.kind() {
65                error!(
66                    message = "Config file not found in path.",
67                    ?path,
68                    internal_log_rate_limit = false
69                );
70                None
71            } else {
72                error!(message = "Error opening config file.", %error, ?path, internal_log_rate_limit = false);
73                None
74            }
75        }
76    }
77}
78
79/// Merge the paths coming from different cli flags with different formats into
80/// a unified list of paths with formats.
81pub fn merge_path_lists(
82    path_lists: Vec<(&[PathBuf], FormatHint)>,
83) -> impl Iterator<Item = (PathBuf, FormatHint)> + '_ {
84    path_lists
85        .into_iter()
86        .flat_map(|(paths, format)| paths.iter().cloned().map(move |path| (path, format)))
87}
88
89/// Expand a list of paths (potentially containing glob patterns) into real
90/// config paths, replacing it with the default paths when empty.
91pub fn process_paths(config_paths: &[ConfigPath]) -> Option<Vec<ConfigPath>> {
92    let starting_paths = if !config_paths.is_empty() {
93        config_paths.to_owned()
94    } else {
95        default_config_paths()
96    };
97
98    let mut paths = Vec::new();
99
100    for config_path in &starting_paths {
101        let config_pattern: &PathBuf = config_path.into();
102
103        let matches: Vec<PathBuf> = match glob(config_pattern.to_str().expect("No ability to glob"))
104        {
105            Ok(glob_paths) => glob_paths.filter_map(Result::ok).collect(),
106            Err(err) => {
107                error!(message = "Failed to read glob pattern.", path = ?config_pattern, error = ?err);
108                return None;
109            }
110        };
111
112        if matches.is_empty() {
113            error!(message = "Config file not found in path.", path = ?config_pattern, internal_log_rate_limit = false);
114            std::process::exit(exitcode::CONFIG);
115        }
116
117        match config_path {
118            ConfigPath::File(_, format) => {
119                for path in matches {
120                    paths.push(ConfigPath::File(path, *format));
121                }
122            }
123            ConfigPath::Dir(_) => {
124                for path in matches {
125                    paths.push(ConfigPath::Dir(path))
126                }
127            }
128        }
129    }
130
131    paths.sort();
132    paths.dedup();
133    // Ignore poison error and let the current main thread continue running to do the cleanup.
134    drop(
135        CONFIG_PATHS
136            .lock()
137            .map(|mut guard| guard.clone_from(&paths)),
138    );
139
140    Some(paths)
141}
142
143pub fn load_from_paths(config_paths: &[ConfigPath]) -> Result<Config, Vec<String>> {
144    let builder = ConfigBuilderLoader::default().load_from_paths(config_paths)?;
145    let (config, build_warnings) = builder.build_with_warnings()?;
146
147    for warning in build_warnings {
148        warn!("{}", warning);
149    }
150
151    Ok(config)
152}
153
154/// Loads a configuration from paths. Handle secret replacement and if a provider is present
155/// in the builder, the config is used as bootstrapping for a remote source. Otherwise,
156/// provider instantiation is skipped.
157pub async fn load_from_paths_with_provider_and_secrets(
158    config_paths: &[ConfigPath],
159    signal_handler: &mut signal::SignalHandler,
160    allow_empty: bool,
161) -> Result<Config, Vec<String>> {
162    let secrets_backends_loader = loader_from_paths(SecretBackendLoader::default(), config_paths)?;
163    let secrets = secrets_backends_loader
164        .retrieve_secrets(signal_handler)
165        .await
166        .map_err(|e| vec![e])?;
167
168    let mut builder = ConfigBuilderLoader::default()
169        .allow_empty(allow_empty)
170        .secrets(secrets)
171        .load_from_paths(config_paths)?;
172
173    validation::check_provider(&builder)?;
174    signal_handler.clear();
175
176    // If there's a provider, overwrite the existing config builder with the remote variant.
177    if let Some(mut provider) = builder.provider {
178        builder = provider.build(signal_handler).await?;
179        debug!(message = "Provider configured.", provider = ?provider.get_component_name());
180    }
181
182    finalize_config(builder).await
183}
184
185pub async fn load_from_str_with_secrets(
186    input: &str,
187    format: Format,
188    signal_handler: &mut signal::SignalHandler,
189    allow_empty: bool,
190) -> Result<Config, Vec<String>> {
191    let secrets_backends_loader =
192        loader_from_input(SecretBackendLoader::default(), input.as_bytes(), format)?;
193    let secrets = secrets_backends_loader
194        .retrieve_secrets(signal_handler)
195        .await
196        .map_err(|e| vec![e])?;
197
198    let builder = ConfigBuilderLoader::default()
199        .allow_empty(allow_empty)
200        .secrets(secrets)
201        .load_from_input(input.as_bytes(), format)?;
202    signal_handler.clear();
203
204    finalize_config(builder).await
205}
206
207async fn finalize_config(builder: ConfigBuilder) -> Result<Config, Vec<String>> {
208    let (new_config, build_warnings) = builder.build_with_warnings()?;
209
210    validation::check_buffer_preconditions(&new_config).await?;
211
212    for warning in build_warnings {
213        warn!("{}", warning);
214    }
215
216    Ok(new_config)
217}
218
219pub(super) fn loader_from_input<T, L, R>(
220    mut loader: L,
221    input: R,
222    format: Format,
223) -> Result<T, Vec<String>>
224where
225    T: serde::de::DeserializeOwned,
226    L: Loader<T> + Process,
227    R: std::io::Read,
228{
229    loader.load_from_str(input, format).map(|_| loader.take())
230}
231
232/// Iterators over `ConfigPaths`, and processes a file/dir according to a provided `Loader`.
233pub(super) fn loader_from_paths<T, L>(
234    mut loader: L,
235    config_paths: &[ConfigPath],
236) -> Result<T, Vec<String>>
237where
238    T: serde::de::DeserializeOwned,
239    L: Loader<T> + Process,
240{
241    let mut errors = Vec::new();
242
243    for config_path in config_paths {
244        match config_path {
245            ConfigPath::File(path, format_hint) => {
246                match loader.load_from_file(
247                    path,
248                    format_hint
249                        .or_else(move || Format::from_path(&path).ok())
250                        .unwrap_or_default(),
251                ) {
252                    Ok(()) => {}
253                    Err(errs) => errors.extend(errs),
254                };
255            }
256            ConfigPath::Dir(path) => {
257                match loader.load_from_dir(path) {
258                    Ok(()) => {}
259                    Err(errs) => errors.extend(errs),
260                };
261            }
262        }
263    }
264
265    if errors.is_empty() {
266        Ok(loader.take())
267    } else {
268        Err(errors)
269    }
270}
271
272/// Uses `SourceLoader` to process `ConfigPaths`, deserializing to a toml `SourceMap`.
273pub fn load_source_from_paths(
274    config_paths: &[ConfigPath],
275) -> Result<toml::value::Table, Vec<String>> {
276    loader_from_paths(SourceLoader::new(), config_paths)
277}
278
279pub fn load_from_str(input: &str, format: Format) -> Result<Config, Vec<String>> {
280    let builder = load_from_inputs(std::iter::once((input.as_bytes(), format)))?;
281    let (config, build_warnings) = builder.build_with_warnings()?;
282
283    for warning in build_warnings {
284        warn!("{}", warning);
285    }
286
287    Ok(config)
288}
289
290fn load_from_inputs(
291    inputs: impl IntoIterator<Item = (impl std::io::Read, Format)>,
292) -> Result<ConfigBuilder, Vec<String>> {
293    let mut config = Config::builder();
294    let mut errors = Vec::new();
295
296    for (input, format) in inputs {
297        if let Err(errs) = load(input, format).and_then(|n| config.append(n)) {
298            // TODO: add back paths
299            errors.extend(errs.iter().map(|e| e.to_string()));
300        }
301    }
302
303    if errors.is_empty() {
304        Ok(config)
305    } else {
306        Err(errors)
307    }
308}
309
310pub fn prepare_input<R: std::io::Read>(
311    mut input: R,
312    interpolate_env: bool,
313) -> Result<String, Vec<String>> {
314    let mut source_string = String::new();
315    input
316        .read_to_string(&mut source_string)
317        .map_err(|e| vec![e.to_string()])?;
318
319    if interpolate_env {
320        let mut vars: HashMap<String, String> = std::env::vars_os()
321            .filter_map(|(k, v)| match (k.into_string(), v.into_string()) {
322                (Ok(k), Ok(v)) => Some((k, v)),
323                _ => None,
324            })
325            .collect();
326
327        if !vars.contains_key("HOSTNAME")
328            && let Ok(hostname) = crate::get_hostname()
329        {
330            vars.insert("HOSTNAME".into(), hostname);
331        }
332        vars::interpolate(&source_string, &vars)
333    } else {
334        Ok(source_string)
335    }
336}
337
338pub fn load<R: std::io::Read, T>(input: R, format: Format) -> Result<T, Vec<String>>
339where
340    T: serde::de::DeserializeOwned,
341{
342    // Via configurations that load from raw string, skip interpolation of env
343    let with_vars = prepare_input(input, false)?;
344
345    format::deserialize(&with_vars, format)
346}
347
348#[cfg(not(windows))]
349fn default_path() -> PathBuf {
350    "/etc/vector/vector.yaml".into()
351}
352
353#[cfg(windows)]
354fn default_path() -> PathBuf {
355    let program_files =
356        std::env::var("ProgramFiles").expect("%ProgramFiles% environment variable must be defined");
357    format!("{}\\Vector\\config\\vector.yaml", program_files).into()
358}
359
360fn default_config_paths() -> Vec<ConfigPath> {
361    #[cfg(not(windows))]
362    let default_path = default_path();
363    #[cfg(windows)]
364    let default_path = default_path();
365
366    vec![ConfigPath::File(default_path, Some(Format::Yaml))]
367}
vector/config/loading/mod.rs

vector/config/loading/
mod.rs
