vector/config/loading/
mod.rs

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

vector/config/loading/
mod.rs
