diff --git a/changelog.d/9766_data_dir_cli_env.feature.md b/changelog.d/9766_data_dir_cli_env.feature.md new file mode 100644 index 0000000000000..e9cabc57a18f5 --- /dev/null +++ b/changelog.d/9766_data_dir_cli_env.feature.md @@ -0,0 +1,3 @@ +Added a `--data-dir` command-line flag (and corresponding `VECTOR_DATA_DIR` environment variable) to override the `data_dir` global configuration option. When set, it takes precedence over any `data_dir` value in the configuration file. This is useful for keeping deployment-specific paths out of the configuration file, for example when validating a configuration in a CI environment where the configured `data_dir` may not exist. + +authors: xfocus3 diff --git a/src/app.rs b/src/app.rs index 15e267e68dc9c..551b5ad67559d 100644 --- a/src/app.rs +++ b/src/app.rs @@ -6,7 +6,7 @@ use std::os::unix::process::ExitStatusExt; use std::os::windows::process::ExitStatusExt; use std::{ num::{NonZeroU64, NonZeroUsize}, - path::PathBuf, + path::{Path, PathBuf}, process::ExitStatus, sync::atomic::{AtomicUsize, Ordering}, time::Duration, @@ -90,6 +90,7 @@ impl ApplicationConfig { &config_paths, watcher_conf, opts.require_healthy, + opts.data_dir.clone(), opts.allow_empty_config, graceful_shutdown_duration, signal_handler, @@ -267,7 +268,11 @@ impl Application { opts.root.dangerously_allow_env_var_interpolation || sub_command.dangerously_allow_env_var_interpolation(), ); - return Err(runtime.block_on(sub_command.execute(signals, color))); + return Err(runtime.block_on(sub_command.execute( + signals, + color, + opts.root.data_dir.as_deref(), + ))); } config::set_env_var_interpolation(opts.root.dangerously_allow_env_var_interpolation); @@ -323,6 +328,7 @@ impl Application { signals, topology_controller, allow_empty_config: root_opts.allow_empty_config, + data_dir: root_opts.data_dir, }) } } @@ -334,6 +340,7 @@ pub struct StartedApplication { pub signals: SignalPair, pub topology_controller: SharedTopologyController, pub allow_empty_config: bool, + pub data_dir: Option, } impl StartedApplication { @@ -353,6 +360,7 @@ impl StartedApplication { topology_controller, internal_topologies, allow_empty_config, + data_dir, } = self; let mut graceful_crash = UnboundedReceiverStream::new(graceful_crash_receiver); @@ -369,6 +377,7 @@ impl StartedApplication { &config_paths, &mut signal_handler, allow_empty_config, + data_dir.as_deref(), ).await { break signal; }, @@ -399,6 +408,7 @@ async fn handle_signal( config_paths: &[ConfigPath], signal_handler: &mut SignalHandler, allow_empty_config: bool, + data_dir: Option<&Path>, ) -> Option { match signal { Ok(SignalTo::ReloadComponents(components_to_reload)) => { @@ -417,12 +427,16 @@ async fn handle_signal( &topology_controller.config_paths, signal_handler, allow_empty_config, + data_dir, ) .await; reload_config_from_result(topology_controller, new_config).await } - Ok(SignalTo::ReloadFromConfigBuilder(config_builder)) => { + Ok(SignalTo::ReloadFromConfigBuilder(mut config_builder)) => { + if let Some(data_dir) = data_dir { + config_builder.set_data_dir(data_dir); + } let topology_controller = topology_controller.lock().await; reload_config_from_result(topology_controller, config_builder.build()).await } @@ -439,6 +453,7 @@ async fn handle_signal( &topology_controller.config_paths, signal_handler, allow_empty_config, + data_dir, ) .await; @@ -645,10 +660,12 @@ pub fn build_runtime( /// Returns a configuration exit code if paths, configuration loading, or watcher setup fail. // https://github.com/vectordotdev/vector/issues/23659 #[allow(clippy::large_futures, reason = "Boxing needs profiling")] +#[allow(clippy::too_many_arguments)] pub async fn load_configs( config_paths: &[ConfigPath], watcher_conf: Option, require_healthy: Option, + data_dir: Option, allow_empty_config: bool, graceful_shutdown_duration: Option, signal_handler: &mut SignalHandler, @@ -669,6 +686,7 @@ pub async fn load_configs( &config_paths, signal_handler, allow_empty_config, + data_dir.as_deref(), ) .await .map_err(handle_config_errors)?; @@ -776,3 +794,142 @@ pub fn watcher_config( WatchConfigMethod::Poll => config::watcher::WatcherConfig::PollWatcher(interval.into()), } } + +#[cfg(all(test, feature = "sources-demo_logs", feature = "sinks-blackhole"))] +mod data_dir_tests { + use clap::Parser; + + use super::*; + + fn config_text(data_dir: &Path, sink: &str) -> String { + let data_dir = serde_json::to_string(data_dir).unwrap(); + format!( + "data_dir: {data_dir}\nsources:\n input:\n type: demo_logs\n format: shuffle\n lines: [\"log\"]\nsinks:\n {sink}:\n type: blackhole\n inputs: [input]\n" + ) + } + + fn run_reload_test_in_subprocess() -> bool { + const CHILD: &str = "VECTOR_TEST_DATA_DIR_RELOAD_CHILD"; + if std::env::var_os(CHILD).is_some() { + return false; + } + let output = std::process::Command::new(std::env::current_exe().unwrap()) + .env(CHILD, "1") + .env_remove("VECTOR_DATA_DIR") + .args([ + "--exact", + "app::data_dir_tests::cli_data_dir_survives_topology_reload_signals", + "--nocapture", + ]) + .output() + .unwrap(); + assert!( + output.status.success(), + "{}\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr), + ); + assert!( + String::from_utf8_lossy(&output.stdout) + .contains("topology reload assertions completed") + ); + true + } + + #[tokio::test] + async fn cli_data_dir_survives_topology_reload_signals() { + if run_reload_test_in_subprocess() { + return; + } + + crate::test_util::trace_init(); + let directory = tempfile::tempdir().unwrap(); + let config_file = directory.path().join("vector.yaml"); + let override_dir = directory.path().join("override-state"); + std::fs::create_dir(&override_dir).unwrap(); + std::fs::write( + &config_file, + config_text(&directory.path().join("configured-state"), "initial"), + ) + .unwrap(); + let opts = RootOpts::try_parse_from([ + "vector", + "--config", + config_file.to_str().unwrap(), + "--data-dir", + override_dir.to_str().unwrap(), + ]) + .unwrap(); + let (mut signals, _receiver) = SignalHandler::new(); + let application = + ApplicationConfig::from_opts(&opts, &mut signals, ExtraContext::default()) + .await + .unwrap(); + let paths = application.config_paths.clone(); + let controller = SharedTopologyController::new(TopologyController { + topology: application.topology, + config_paths: paths.clone(), + require_healthy: opts.require_healthy, + #[cfg(feature = "api")] + api_server: None, + extra_context: ExtraContext::default(), + }); + + for replacement in ["disk", "provider", "components"] { + let text = config_text(&directory.path().join(replacement), replacement); + let signal = if replacement == "provider" { + let builder = config::loading::ConfigBuilderLoader::default() + .load_from_input(text.as_bytes(), config::Format::Yaml) + .unwrap(); + SignalTo::ReloadFromConfigBuilder(builder) + } else { + std::fs::write(&config_file, text).unwrap(); + if replacement == "disk" { + SignalTo::ReloadFromDisk + } else { + SignalTo::ReloadComponents(std::collections::HashSet::new()) + } + }; + let outcome = tokio::time::timeout( + Duration::from_secs(10), + handle_signal( + Ok(signal), + &controller, + &paths, + &mut signals, + false, + opts.data_dir.as_deref(), + ), + ) + .await + .expect("topology reload timed out"); + assert!(outcome.is_none()); + let current = controller.lock().await; + // A retained data_dir alone is insufficient: the new topology must actually + // have replaced the old one, rather than rolling back a rejected reload. + assert!( + current + .topology + .config + .sink(&config::ComponentKey::from(replacement)) + .is_some() + ); + assert_eq!( + current.topology.config.global.data_dir.as_ref(), + Some(&override_dir) + ); + } + let controller = controller + .try_into_inner() + .expect("no other controller owner") + .into_inner(); + assert!(controller.stop().await); + #[allow( + clippy::print_stdout, + reason = "The parent verifies the isolated reload assertions ran." + )] + { + println!("topology reload assertions completed"); + } + } +} diff --git a/src/cli.rs b/src/cli.rs index 7899b08e6638c..69a4e192f64d4 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -136,6 +136,15 @@ pub struct RootOpts { #[arg(short, long, env = "VECTOR_REQUIRE_HEALTHY")] pub require_healthy: Option, + /// The directory used for persisting Vector state data. + /// + /// This overrides the `data_dir` global option set in the configuration file. It is + /// useful for keeping deployment-specific paths out of the configuration file, for + /// example when validating a configuration in a CI environment where the configured + /// `data_dir` may not exist. + #[arg(long, env = "VECTOR_DATA_DIR", global = true)] + pub data_dir: Option, + /// Number of threads to use for processing (default is number of available cores) #[arg(short, long, env = "VECTOR_THREADS")] pub threads: Option, @@ -510,22 +519,23 @@ impl SubCommand { &self, mut signals: signal::SignalPair, color: bool, + data_dir: Option<&std::path::Path>, ) -> exitcode::ExitCode { match self { Self::Completion(s) => completion::cmd(s), Self::ConvertConfig(opts) => convert_config::cmd(opts), Self::Generate(g) => generate::cmd(g), Self::GenerateSchema(opts) => generate_schema::cmd(opts), - Self::Graph(g) => graph::cmd(g), + Self::Graph(g) => graph::cmd(g, data_dir), Self::List(l) => list::cmd(l), #[cfg(windows)] - Self::Service(s) => service::cmd(s), + Self::Service(s) => service::cmd(s, data_dir), #[cfg(feature = "api-client")] Self::Tap(t) => tap::cmd(t, signals.receiver).await, - Self::Test(t) => unit_test::cmd(t, &mut signals.handler).await, + Self::Test(t) => unit_test::cmd(t, &mut signals.handler, data_dir).await, #[cfg(feature = "top")] Self::Top(t) => top::cmd(t).await, - Self::Validate(v) => validate::validate(v, &mut signals.handler, color).await, + Self::Validate(v) => validate::validate(v, &mut signals.handler, color, data_dir).await, Self::Vrl(s) => vrl::cli::cmd::cmd(s, vector_vrl_functions::all()), } } @@ -584,6 +594,138 @@ pub fn handle_config_errors(errors: Vec) -> exitcode::ExitCode { #[cfg(test)] mod tests { + use std::path::PathBuf; + + #[cfg(unix)] + use std::os::unix::ffi::OsStringExt; + + use clap::Parser; + + use super::RootOpts; + + #[test] + fn data_dir_defaults_to_none() { + if data_dir_test_in_subprocess("cli::tests::data_dir_defaults_to_none", None) { + return; + } + let opts = RootOpts::try_parse_from(["vector"]).unwrap(); + assert_eq!(opts.data_dir, None); + #[allow( + clippy::print_stdout, + reason = "The parent verifies that the isolated test actually ran." + )] + { + println!("data-dir test completed"); + } + } + + #[test] + fn data_dir_flag_overrides_environment() { + if data_dir_test_in_subprocess( + "cli::tests::data_dir_flag_overrides_environment", + Some("environment-state"), + ) { + return; + } + let opts = super::Opts::try_parse_from([ + "vector", + "validate", + "--data-dir", + "flag-state", + "vector.yaml", + ]) + .unwrap(); + assert_eq!(opts.root.data_dir, Some(PathBuf::from("flag-state"))); + #[allow( + clippy::print_stdout, + reason = "The parent verifies that the isolated test actually ran." + )] + { + println!("data-dir test completed"); + } + } + + fn data_dir_test_in_subprocess(test_name: &str, value: Option<&str>) -> bool { + const CHILD: &str = "VECTOR_DATA_DIR_TEST_CHILD"; + if std::env::var(CHILD).as_deref() == Ok(test_name) { + return false; + } + let mut command = std::process::Command::new(std::env::current_exe().unwrap()); + command.env(CHILD, test_name).env_remove("VECTOR_DATA_DIR"); + if let Some(value) = value { + command.env("VECTOR_DATA_DIR", value); + } + let output = command + .args(["--exact", test_name, "--nocapture"]) + .output() + .unwrap(); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stdout) + ); + assert!(String::from_utf8_lossy(&output.stdout).contains("data-dir test completed")); + true + } + + #[test] + fn data_dir_rejects_empty_flag() { + let error = RootOpts::try_parse_from(["vector", "--data-dir", ""]).unwrap_err(); + assert_eq!(error.kind(), clap::error::ErrorKind::InvalidValue); + } + + #[test] + fn data_dir_rejects_empty_environment() { + if data_dir_test_in_subprocess("cli::tests::data_dir_rejects_empty_environment", Some("")) { + return; + } + let error = RootOpts::try_parse_from(["vector"]).unwrap_err(); + assert_eq!(error.kind(), clap::error::ErrorKind::InvalidValue); + let opts = RootOpts::try_parse_from(["vector", "--data-dir", "flag-state"]).unwrap(); + assert_eq!(opts.data_dir, Some(PathBuf::from("flag-state"))); + #[allow( + clippy::print_stdout, + reason = "The parent verifies that the isolated test actually ran." + )] + { + println!("data-dir test completed"); + } + } + + #[cfg(unix)] + #[test] + fn data_dir_parser_preserves_non_unicode_path() { + let path = std::ffi::OsString::from_vec(b"state-\xff".to_vec()); + let opts = RootOpts::try_parse_from([ + std::ffi::OsString::from("vector"), + std::ffi::OsString::from("--data-dir"), + path.clone(), + ]) + .unwrap(); + assert_eq!(opts.data_dir, Some(PathBuf::from(path))); + } + + #[test] + fn data_dir_parsed_from_flag() { + let opts = RootOpts::try_parse_from(["vector", "--data-dir", "/tmp/vector"]).unwrap(); + assert_eq!(opts.data_dir, Some(PathBuf::from("/tmp/vector"))); + } + + #[test] + fn data_dir_is_global_for_validate() { + for args in [ + ["vector", "--data-dir", "state", "validate", "vector.yaml"], + ["vector", "validate", "--data-dir", "state", "vector.yaml"], + ] { + let opts = super::Opts::try_parse_from(args).unwrap(); + assert_eq!(opts.root.data_dir, Some(PathBuf::from("state"))); + assert!(matches!( + opts.sub_command, + Some(super::SubCommand::Validate(_)) + )); + } + } + #[cfg(unix)] use nix::sys::resource::{Resource, getrlimit, setrlimit}; diff --git a/src/config/loading/config_builder.rs b/src/config/loading/config_builder.rs index 98fb16e1311f7..0aca2a44062ea 100644 --- a/src/config/loading/config_builder.rs +++ b/src/config/loading/config_builder.rs @@ -1,4 +1,8 @@ -use std::{collections::HashMap, io::Read}; +use std::{ + collections::HashMap, + io::Read, + path::{Path, PathBuf}, +}; use indexmap::IndexMap; @@ -16,9 +20,20 @@ pub struct ConfigBuilderLoader { builder: ConfigBuilder, secrets: HashMap, interpolate_env: bool, + data_dir: Option, } impl ConfigBuilderLoader { + /// Override the data directory before merging individual configuration files. + #[must_use] + pub fn data_dir(mut self, data_dir: Option<&Path>) -> Self { + self.data_dir = data_dir.map(Path::to_path_buf); + if let Some(data_dir) = data_dir { + self.builder.set_data_dir(data_dir); + } + self + } + /// Sets whether to interpolate environment variables in the config. #[must_use] pub const fn interpolate_env(mut self, interpolate: bool) -> Self { @@ -68,6 +83,7 @@ impl Default for ConfigBuilderLoader { builder: ConfigBuilder::default(), secrets: HashMap::new(), interpolate_env: super::env_var_interpolation_enabled(), + data_dir: None, } } } @@ -118,7 +134,11 @@ impl Process for ConfigBuilderLoader { ); } None => { - self.builder.append(deserialize_config_map(map)?)?; + let mut builder: ConfigBuilder = deserialize_config_map(map)?; + if let Some(data_dir) = &self.data_dir { + builder.set_data_dir(data_dir); + } + self.builder.append(builder)?; } } @@ -227,3 +247,54 @@ mod tests { .unwrap(); } } + +#[cfg(test)] +mod data_dir_tests { + use super::*; + use crate::config::ConfigPath; + + #[test] + fn data_dir_override_precedes_split_config_merge() { + let directory = tempfile::tempdir().unwrap(); + let first = directory.path().join("first.yaml"); + let second = directory.path().join("second.json"); + std::fs::write(&first, "data_dir: first-state\n").unwrap(); + std::fs::write(&second, r#"{"data_dir":"second-state"}"#).unwrap(); + let paths = [ + ConfigPath::File(first, None), + ConfigPath::File(second, None), + ]; + let override_dir = directory.path().join("override-state"); + let builder = ConfigBuilderLoader::default() + .data_dir(Some(&override_dir)) + .load_from_paths(&paths) + .unwrap(); + assert_eq!(builder.global.data_dir.as_ref(), Some(&override_dir)); + assert!( + ConfigBuilderLoader::default() + .load_from_paths(&paths) + .is_err() + ); + } + + #[tokio::test] + async fn data_dir_override_survives_repeated_loads() { + let directory = tempfile::tempdir().unwrap(); + let config_file = directory.path().join("vector.yaml"); + let paths = [ConfigPath::File(config_file.clone(), None)]; + let override_dir = directory.path().join("override-state"); + let (mut signals, _receiver) = crate::signal::SignalHandler::new(); + for configured_dir in ["first-state", "second-state"] { + std::fs::write(&config_file, format!("data_dir: {configured_dir}\n")).unwrap(); + let config = crate::config::load_from_paths_with_provider_and_secrets( + &paths, + &mut signals, + true, + Some(&override_dir), + ) + .await + .unwrap(); + assert_eq!(config.global.data_dir.as_ref(), Some(&override_dir)); + } + } +} diff --git a/src/config/loading/mod.rs b/src/config/loading/mod.rs index 30f129890499c..62c0aaceeaa2e 100644 --- a/src/config/loading/mod.rs +++ b/src/config/loading/mod.rs @@ -152,7 +152,16 @@ pub fn process_paths(config_paths: &[ConfigPath]) -> Option> { // https://github.com/vectordotdev/vector/issues/23659 #[allow(clippy::missing_errors_doc, reason = "Error documentation deferred")] pub fn load_from_paths(config_paths: &[ConfigPath]) -> Result> { - let builder = ConfigBuilderLoader::default().load_from_paths(config_paths)?; + load_from_paths_with_data_dir(config_paths, None) +} + +pub(crate) fn load_from_paths_with_data_dir( + config_paths: &[ConfigPath], + data_dir: Option<&Path>, +) -> Result> { + let builder = ConfigBuilderLoader::default() + .data_dir(data_dir) + .load_from_paths(config_paths)?; let (config, build_warnings) = builder.build_with_warnings()?; for warning in build_warnings { @@ -171,9 +180,11 @@ pub async fn load_from_paths_with_provider_and_secrets( config_paths: &[ConfigPath], signal_handler: &mut signal::SignalHandler, allow_empty: bool, + data_dir: Option<&Path>, ) -> Result> { let mut builder = - load_builder_from_paths_with_secrets(config_paths, signal_handler, allow_empty).await?; + load_builder_from_paths_with_secrets(config_paths, signal_handler, allow_empty, data_dir) + .await?; validation::check_provider(&builder)?; signal_handler.clear(); @@ -181,6 +192,9 @@ pub async fn load_from_paths_with_provider_and_secrets( // If there's a provider, overwrite the existing config builder with the remote variant. if let Some(mut provider) = builder.provider { builder = provider.build(signal_handler).await?; + if let Some(data_dir) = data_dir { + builder.set_data_dir(data_dir); + } debug!(message = "Provider configured.", provider = ?provider.get_component_name()); } @@ -193,6 +207,7 @@ pub(crate) async fn load_builder_from_paths_with_secrets( config_paths: &[ConfigPath], signal_handler: &mut signal::SignalHandler, allow_empty: bool, + data_dir: Option<&Path>, ) -> Result> { let secrets_backends_loader = loader_from_paths(SecretBackendLoader::default(), config_paths)?; let secrets = secrets_backends_loader @@ -201,6 +216,7 @@ pub(crate) async fn load_builder_from_paths_with_secrets( .map_err(|e| vec![e])?; ConfigBuilderLoader::default() + .data_dir(data_dir) .allow_empty(allow_empty) .secrets(secrets) .load_from_paths(config_paths) diff --git a/src/config/unit_test/mod.rs b/src/config/unit_test/mod.rs index 0f6da7e9ea57b..171c04c330c2e 100644 --- a/src/config/unit_test/mod.rs +++ b/src/config/unit_test/mod.rs @@ -15,6 +15,7 @@ mod unit_test_components; use std::{ collections::{BTreeMap, HashMap, HashSet}, + path::Path, sync::Arc, }; @@ -100,8 +101,11 @@ impl UnitTest { fn init_log_schema_from_paths( config_paths: &[ConfigPath], deny_if_set: bool, + data_dir: Option<&Path>, ) -> Result<(), Vec> { - let builder = ConfigBuilderLoader::default().load_from_paths(config_paths)?; + let builder = ConfigBuilderLoader::default() + .data_dir(data_dir) + .load_from_paths(config_paths)?; vector_lib::config::init_log_schema(builder.global.log_schema, deny_if_set); Ok(()) } @@ -111,8 +115,9 @@ fn init_log_schema_from_paths( pub async fn build_unit_tests_main( paths: &[ConfigPath], signal_handler: &mut signal::SignalHandler, + data_dir: Option<&Path>, ) -> Result, Vec> { - init_log_schema_from_paths(paths, false)?; + init_log_schema_from_paths(paths, false, data_dir)?; let secrets_backends_loader = loading::loader_from_paths(loading::SecretBackendLoader::default(), paths)?; let secrets = secrets_backends_loader @@ -121,6 +126,7 @@ pub async fn build_unit_tests_main( .map_err(|e| vec![e])?; let config_builder = ConfigBuilderLoader::default() + .data_dir(data_dir) .secrets(secrets) .load_from_paths(paths)?; diff --git a/src/graph.rs b/src/graph.rs index 1efe7aa3aa721..047856dbdb007 100644 --- a/src/graph.rs +++ b/src/graph.rs @@ -2,7 +2,7 @@ use std::{ collections::{HashMap, HashSet}, fmt::Write as _, - path::PathBuf, + path::{Path, PathBuf}, }; use clap::Parser; @@ -115,13 +115,13 @@ fn edge_attributes_to_string(attributes: &EdgeAttributes, default_label: Option< attrs.iter().map(|(k, v)| format!("{k}=\"{v}\"")).join(" ") } -pub(crate) fn cmd(opts: &Opts) -> exitcode::ExitCode { +pub(crate) fn cmd(opts: &Opts, data_dir: Option<&Path>) -> exitcode::ExitCode { let paths = opts.paths_with_formats(); let Some(paths) = config::process_paths(&paths) else { return exitcode::CONFIG; }; - let config = match config::load_from_paths(&paths) { + let config = match config::loading::load_from_paths_with_data_dir(&paths, data_dir) { Ok(config) => config, Err(errs) => { #[allow(clippy::print_stderr)] diff --git a/src/lib.rs b/src/lib.rs index 38fe0b53a82c1..f75bc4ddbdf64 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -102,6 +102,11 @@ pub mod secrets; pub mod serde; #[cfg(windows)] pub mod service; +#[cfg(any( + windows, + all(test, feature = "sources-demo_logs", feature = "sinks-blackhole") +))] +mod service_arguments; pub mod signal; pub(crate) mod sink_ext; #[allow(unreachable_pub)] diff --git a/src/service.rs b/src/service.rs index bee5c202e10a7..d088370ae3445 100644 --- a/src/service.rs +++ b/src/service.rs @@ -1,9 +1,13 @@ #![allow(missing_docs)] -use std::{ffi::OsString, path::PathBuf, time::Duration}; +use std::{ + ffi::OsString, + path::{Path, PathBuf}, + time::Duration, +}; use clap::Parser; -use crate::{cli::handle_config_errors, config}; +use crate::{config, service_arguments::create_service_arguments}; const DEFAULT_SERVICE_NAME: &str = crate::built_info::PKG_NAME; @@ -75,22 +79,22 @@ struct InstallOpts { } impl InstallOpts { - fn service_info(&self) -> ServiceInfo { + fn service_info(&self, data_dir: Option<&Path>) -> Option { let service_name = self.name.as_deref().unwrap_or(DEFAULT_SERVICE_NAME); let display_name = self.display_name.as_deref().unwrap_or("Vector Service"); let description = crate::built_info::PKG_DESCRIPTION; let current_exe = ::std::env::current_exe().unwrap(); let config_paths = self.config_paths_with_formats(); - let arguments = create_service_arguments(&config_paths).unwrap(); + let arguments = create_service_arguments(&config_paths, data_dir)?; - ServiceInfo { + Some(ServiceInfo { name: OsString::from(service_name), display_name: OsString::from(display_name), description: OsString::from(description), executable_path: current_exe, launch_arguments: arguments, - } + }) } fn config_paths_with_formats(&self) -> Vec { @@ -241,12 +245,15 @@ enum ControlAction { Restart { stop_timeout: Duration }, } -pub fn cmd(opts: &Opts) -> exitcode::ExitCode { +pub fn cmd(opts: &Opts, data_dir: Option<&Path>) -> exitcode::ExitCode { let sub_command = &opts.sub_command; match sub_command { Some(s) => match s { SubCommand::Install(opts) => { - control_service(&opts.service_info(), ControlAction::Install) + let Some(service_info) = opts.service_info(data_dir) else { + return exitcode::CONFIG; + }; + control_service(&service_info, ControlAction::Install) } SubCommand::Uninstall(opts) => { let stop_timeout = Duration::from_secs(opts.stop_timeout as u64); @@ -320,35 +327,29 @@ fn control_service(service: &ServiceInfo, action: ControlAction) -> exitcode::Ex } } -fn create_service_arguments(config_paths: &[config::ConfigPath]) -> Option> { - let config_paths = config::process_paths(config_paths)?; - match config::load_from_paths(&config_paths) { - Ok(_) => { - let mut args: Vec = config_paths - .iter() - .flat_map(|config_path| match config_path { - config::ConfigPath::File(path, format) => { - let key = match format { - None => "--config", - Some(config::Format::Toml) => "--config-toml", - Some(config::Format::Json) => "--config-json", - Some(config::Format::Yaml) => "--config-yaml", - }; - vec![OsString::from(key), path.as_os_str().into()] - } - config::ConfigPath::Dir(path) => { - vec![OsString::from("--config-dir"), path.as_os_str().into()] - } - }) - .collect(); - if config::env_var_interpolation_enabled() { - args.push(OsString::from("--dangerously-allow-env-var-interpolation")); - } - Some(args) - } - Err(errs) => { - handle_config_errors(errs); - None - } +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn service_rejects_invalid_config_glob_without_installing() { + let opts = Opts::try_parse_from(["service", "install", "--config", "["]).unwrap(); + let Some(SubCommand::Install(install)) = &opts.sub_command else { + panic!("expected install options"); + }; + assert!(install.service_info(None).is_none()); + // Invalid configuration is rejected before the service control manager is contacted. + assert_eq!(cmd(&opts, None), exitcode::CONFIG); + } + + #[test] + fn service_data_dir_rejects_empty_override_without_installing() { + let opts = Opts::try_parse_from(["service", "install"]).unwrap(); + let Some(SubCommand::Install(install)) = &opts.sub_command else { + panic!("expected install options"); + }; + assert!(install.service_info(Some(Path::new(""))).is_none()); + // Argument construction fails before the service control manager is contacted. + assert_eq!(cmd(&opts, Some(Path::new(""))), exitcode::CONFIG); } } diff --git a/src/service_arguments.rs b/src/service_arguments.rs new file mode 100644 index 0000000000000..f657165a9af01 --- /dev/null +++ b/src/service_arguments.rs @@ -0,0 +1,313 @@ +use std::{ffi::OsString, path::Path}; + +use crate::{cli::handle_config_errors, config}; + +pub(super) fn create_service_arguments( + config_paths: &[config::ConfigPath], + data_dir: Option<&Path>, +) -> Option> { + // The service may start in a different working directory from the installer. + // Resolve the override once for both configuration loading and persisted arguments, + // without requiring the directory to exist or resolving filesystem symlinks. + let data_dir = match data_dir + .map(|path| { + if path.is_absolute() { + Ok(path.to_path_buf()) + } else { + std::path::absolute(path) + } + }) + .transpose() + { + Ok(path) => path, + Err(error) => { + error!(message = "Failed to resolve service data directory.", %error); + return None; + } + }; + let data_dir = data_dir.as_deref(); + let config_paths = config::process_paths(config_paths)?; + match config::loading::load_from_paths_with_data_dir(&config_paths, data_dir) { + Ok(_) => { + let mut args: Vec = config_paths + .iter() + .flat_map(|config_path| match config_path { + config::ConfigPath::File(path, format) => { + let key = match format { + None => "--config", + Some(config::Format::Toml) => "--config-toml", + Some(config::Format::Json) => "--config-json", + Some(config::Format::Yaml) => "--config-yaml", + }; + vec![OsString::from(key), path.as_os_str().into()] + } + config::ConfigPath::Dir(path) => { + vec![OsString::from("--config-dir"), path.as_os_str().into()] + } + }) + .collect(); + if config::env_var_interpolation_enabled() { + args.push(OsString::from("--dangerously-allow-env-var-interpolation")); + } + if let Some(data_dir) = data_dir { + // Keep option and path as separate arguments. The service library owns + // Windows command-line quoting, including spaces and trailing backslashes. + args.push(OsString::from("--data-dir")); + args.push(data_dir.as_os_str().to_owned()); + } + Some(args) + } + Err(errs) => { + handle_config_errors(errs); + None + } + } +} + +#[cfg(all(test, feature = "sources-demo_logs", feature = "sinks-blackhole"))] +mod tests { + #[cfg(unix)] + use std::os::unix::ffi::OsStringExt; + #[cfg(windows)] + use std::os::windows::ffi::OsStringExt; + + use clap::Parser; + + use super::*; + + fn write_config(path: &Path, data_dir: &Path, include_components: bool) { + let data_dir = serde_json::to_string(data_dir).unwrap(); + let mut text = format!("data_dir: {data_dir}\n"); + if include_components { + text.push_str("sources:\n input:\n type: demo_logs\n format: shuffle\n lines: [\"log\"]\nsinks:\n output:\n type: blackhole\n inputs: [input]\n"); + } + std::fs::write(path, text).unwrap(); + } + + #[test] + fn service_data_dir_invalid_config_glob_returns_none() { + let paths = [config::ConfigPath::File("[".into(), None)]; + assert!(create_service_arguments(&paths, None).is_none()); + assert!(create_service_arguments(&paths, Some(Path::new("state"))).is_none()); + } + + #[test] + fn service_arguments_leave_unset_data_dir_unchanged() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("vector config.yaml"); + write_config(&path, &directory.path().join("configured-state"), true); + let args = create_service_arguments(&[config::ConfigPath::File(path.clone(), None)], None) + .unwrap(); + let mut expected = vec![OsString::from("--config"), path.into_os_string()]; + if config::env_var_interpolation_enabled() { + expected.push(OsString::from("--dangerously-allow-env-var-interpolation")); + } + assert_eq!(args, expected); + } + + #[test] + fn service_data_dir_keeps_spaces_and_backslashes_in_one_argument() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("vector config.yaml"); + write_config(&path, &directory.path().join("configured-state"), true); + let mut data_dir = directory.path().join("state with spaces"); + std::fs::create_dir(&data_dir).unwrap(); + data_dir.as_mut_os_string().push("\\"); + let args = + create_service_arguments(&[config::ConfigPath::File(path, None)], Some(&data_dir)) + .unwrap(); + assert_eq!(args[args.len() - 2], "--data-dir"); + assert_eq!(args.last().unwrap().as_os_str(), data_dir.as_os_str()); + let parsed = crate::cli::RootOpts::try_parse_from( + std::iter::once(OsString::from("vector")).chain(args), + ) + .unwrap(); + assert_eq!(parsed.data_dir, Some(data_dir)); + } + + #[test] + fn service_data_dir_override_precedes_split_config_merge() { + let directory = tempfile::tempdir().unwrap(); + let first = directory.path().join("first.yaml"); + let second = directory.path().join("second.yaml"); + write_config(&first, &directory.path().join("first-state"), true); + write_config(&second, &directory.path().join("second-state"), false); + let paths = [ + config::ConfigPath::File(first, None), + config::ConfigPath::File(second, None), + ]; + assert!(create_service_arguments(&paths, None).is_none()); + let data_dir = directory.path().join("override-state"); + let args = create_service_arguments(&paths, Some(&data_dir)).unwrap(); + assert_eq!(args.last().unwrap().as_os_str(), data_dir.as_os_str()); + } + + fn run_in_installer_directory(test_name: &str) -> bool { + const CHILD: &str = "VECTOR_TEST_SERVICE_DATA_DIR_CHILD"; + if std::env::var(CHILD).as_deref() == Ok(test_name) { + return false; + } + let directory = tempfile::tempdir().unwrap(); + let output = std::process::Command::new(std::env::current_exe().unwrap()) + .env(CHILD, test_name) + .env_remove("VECTOR_DATA_DIR") + .env_remove("VECTOR_TEST_SERVICE_DATA_DIR_ARGUMENT") + .current_dir(directory.path()) + .args(["--exact", test_name, "--nocapture"]) + .output() + .unwrap(); + assert_subprocess_completed(&output); + true + } + + fn assert_subprocess_completed(output: &std::process::Output) { + assert!( + output.status.success(), + "{}\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr), + ); + assert!( + String::from_utf8_lossy(&output.stdout) + .contains("service data-dir assertions completed") + ); + } + + #[test] + fn service_data_dir_resolves_relative_path_before_launch() { + const TEST: &str = + "service_arguments::tests::service_data_dir_resolves_relative_path_before_launch"; + const ARGUMENT: &str = "VECTOR_TEST_SERVICE_DATA_DIR_ARGUMENT"; + const EXPECTED: &str = "VECTOR_TEST_SERVICE_DATA_DIR_EXPECTED"; + const CONFIG: &str = "VECTOR_TEST_SERVICE_DATA_DIR_CONFIG"; + if run_in_installer_directory(TEST) { + return; + } + + if let Some(argument) = std::env::var_os(ARGUMENT) { + // Replay the persisted argument from a different service working directory. + let expected = std::path::PathBuf::from(std::env::var_os(EXPECTED).unwrap()); + assert_ne!(std::env::current_dir().unwrap(), expected.parent().unwrap()); + let opts = crate::cli::RootOpts::try_parse_from([ + OsString::from("vector"), + OsString::from("--data-dir"), + argument, + ]) + .unwrap(); + assert_eq!(opts.data_dir.as_ref(), Some(&expected)); + let paths = [config::ConfigPath::File( + std::env::var_os(CONFIG).unwrap().into(), + None, + )]; + let loaded = + config::loading::load_from_paths_with_data_dir(&paths, opts.data_dir.as_deref()) + .unwrap(); + assert_eq!(loaded.global.data_dir, Some(expected)); + } else { + let installer_directory = std::env::current_dir().unwrap(); + let config = installer_directory.join("vector config.yaml"); + write_config(&config, &installer_directory.join("configured-state"), true); + let relative = Path::new(".").join("state with spaces"); + let expected = installer_directory.join("state with spaces"); + assert!(!expected.exists()); + let args = create_service_arguments( + &[config::ConfigPath::File(config.clone(), None)], + Some(&relative), + ) + .unwrap(); + let argument = args.last().unwrap(); + assert!(Path::new(argument).is_absolute()); + assert_eq!(Path::new(argument), expected); + assert_eq!(args[args.len() - 2], "--data-dir"); + assert!(!expected.exists()); + + let service_directory = installer_directory.join("service-working-directory"); + std::fs::create_dir(&service_directory).unwrap(); + let output = std::process::Command::new(std::env::current_exe().unwrap()) + .env(ARGUMENT, argument) + .env(EXPECTED, expected) + .env(CONFIG, config) + .current_dir(service_directory) + .args(["--exact", TEST, "--nocapture"]) + .output() + .unwrap(); + assert_subprocess_completed(&output); + } + #[allow( + clippy::print_stdout, + reason = "The parent verifies that each isolated service-path test ran." + )] + { + println!("service data-dir assertions completed"); + } + } + + #[test] + fn service_data_dir_preserves_non_unicode_path() { + #[cfg(unix)] + let relative = OsString::from_vec(b"state-\xff".to_vec()); + #[cfg(windows)] + let relative = OsString::from_wide(&[0x73, 0x74, 0x61, 0x74, 0x65, 0x2d, 0xd800]); + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("vector.yaml"); + write_config(&path, &directory.path().join("configured-state"), true); + let paths = [config::ConfigPath::File(path, None)]; + let expected = std::env::current_dir().unwrap().join(&relative); + for data_dir in [Path::new(&relative), expected.as_path()] { + let args = create_service_arguments(&paths, Some(data_dir)).unwrap(); + assert_eq!(args.last().unwrap(), expected.as_os_str()); + let parsed = crate::cli::RootOpts::try_parse_from( + std::iter::once(OsString::from("vector")).chain(args), + ) + .unwrap(); + assert_eq!(parsed.data_dir.as_ref(), Some(&expected)); + } + } + + #[test] + fn service_data_dir_rejects_empty_override() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("vector.yaml"); + write_config(&path, &directory.path().join("configured-state"), true); + assert!( + create_service_arguments(&[config::ConfigPath::File(path, None)], Some(Path::new("")),) + .is_none() + ); + } + + #[cfg(windows)] + #[test] + fn service_data_dir_resolves_windows_relative_forms() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("vector.yaml"); + write_config(&path, &directory.path().join("configured-state"), true); + let paths = [config::ConfigPath::File(path, None)]; + for relative in [ + r".\state with spaces\", + r"\state with spaces\", + r"C:state with spaces\", + ] { + let expected = std::path::absolute(relative).unwrap(); + let args = create_service_arguments(&paths, Some(Path::new(relative))).unwrap(); + assert!(Path::new(args.last().unwrap()).is_absolute()); + assert_eq!(args.last().unwrap(), expected.as_os_str()); + } + } + + #[cfg(windows)] + #[test] + fn service_data_dir_keeps_absolute_windows_paths_unchanged() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("vector.yaml"); + write_config(&path, &directory.path().join("configured-state"), true); + let paths = [config::ConfigPath::File(path, None)]; + for absolute in [ + r"C:\state with spaces\", + r"\\server\share\state\", + r"\\?\C:\state with spaces\", + ] { + let args = create_service_arguments(&paths, Some(Path::new(absolute))).unwrap(); + assert_eq!(args.last().unwrap(), absolute); + } + } +} diff --git a/src/unit_test.rs b/src/unit_test.rs index e2f0c08a50173..7163052f749c8 100644 --- a/src/unit_test.rs +++ b/src/unit_test.rs @@ -3,7 +3,7 @@ use std::{ fs::File, io::prelude::*, - path::PathBuf, + path::{Path, PathBuf}, time::{Duration, Instant}, }; @@ -147,7 +147,11 @@ impl<'a> JUnitReporter<'a> { } } -pub async fn cmd(opts: &Opts, signal_handler: &mut signal::SignalHandler) -> exitcode::ExitCode { +pub async fn cmd( + opts: &Opts, + signal_handler: &mut signal::SignalHandler, + data_dir: Option<&Path>, +) -> exitcode::ExitCode { let mut aggregated_test_errors: Vec<(String, Vec)> = Vec::new(); let paths = opts.paths_with_formats(); @@ -161,7 +165,7 @@ pub async fn cmd(opts: &Opts, signal_handler: &mut signal::SignalHandler) -> exi { println!("Running tests"); } - match config::build_unit_tests_main(&paths, signal_handler).await { + match config::build_unit_tests_main(&paths, signal_handler, data_dir).await { Ok(tests) => { if tests.is_empty() { #[allow(clippy::print_stdout)] diff --git a/src/validate.rs b/src/validate.rs index 9a00fd16e5e7d..af4c2261b1dcb 100644 --- a/src/validate.rs +++ b/src/validate.rs @@ -182,12 +182,13 @@ pub async fn validate( opts: &Opts, signal_handler: &mut crate::signal::SignalHandler, color: bool, + data_dir: Option<&Path>, ) -> ExitCode { let mut fmt = Formatter::new(color); let mut validated = true; - let Some(mut config) = validate_config(opts, signal_handler, &mut fmt).await else { + let Some(mut config) = validate_config(opts, signal_handler, &mut fmt, data_dir).await else { return exitcode::CONFIG; }; @@ -215,6 +216,7 @@ pub async fn validate_config( opts: &Opts, signal_handler: &mut crate::signal::SignalHandler, fmt: &mut Formatter, + data_dir: Option<&Path>, ) -> Option { // Prepare paths let paths = opts.paths_with_formats(); @@ -236,9 +238,17 @@ pub async fn validate_config( // `SECRET[...]` placeholders stay in place. Otherwise resolve them like // the run path, so validation checks the config that would actually run. let builder = if opts.no_environment && !opts.resolve_secrets { - ConfigBuilderLoader::default().load_from_paths(&paths) + ConfigBuilderLoader::default() + .data_dir(data_dir) + .load_from_paths(&paths) } else { - config::loading::load_builder_from_paths_with_secrets(&paths, signal_handler, false).await + config::loading::load_builder_from_paths_with_secrets( + &paths, + signal_handler, + false, + data_dir, + ) + .await } .map_err(&mut report_error) .ok()?; diff --git a/tests/integration/cli.rs b/tests/integration/cli.rs index 26f003daba74a..4e69c082434b9 100644 --- a/tests/integration/cli.rs +++ b/tests/integration/cli.rs @@ -342,3 +342,95 @@ fn validate_output_with_args(config: &str, args: &[&str]) -> std::process::Outpu ); output } + +#[test] +fn validate_data_dir_override_from_flag_or_environment() { + let directory = tempfile::tempdir().unwrap(); + let config_path = directory.path().join("vector.yaml"); + let second_path = directory.path().join("global.yaml"); + let missing_dir = directory.path().join("missing-config-state"); + let other_missing_dir = directory.path().join("other-missing-config-state"); + std::fs::write( + &config_path, + formatdoc! {r#" + data_dir: {missing_dir:?} + sources: + in: + type: demo_logs + format: shuffle + lines: ["log"] + sinks: + out: + inputs: ["in"] + type: blackhole + "#}, + ) + .unwrap(); + std::fs::write(&second_path, format!("data_dir: {other_missing_dir:?}\n")).unwrap(); + + for from_flag in [true, false] { + for no_environment in [true, false] { + let mut cmd = Command::cargo_bin("vector").unwrap(); + cmd.env_remove("VECTOR_DATA_DIR").arg("validate"); + if from_flag { + cmd.arg("--data-dir").arg(directory.path()); + } else { + cmd.env("VECTOR_DATA_DIR", directory.path()); + } + if no_environment { + cmd.arg("--no-environment"); + } + cmd.arg(&config_path).arg(&second_path).assert().success(); + } + } + assert!(!missing_dir.exists()); + assert!(!other_missing_dir.exists()); +} + +#[test] +fn graph_and_test_data_dir_override_precedes_split_config_merge() { + let directory = tempfile::tempdir().unwrap(); + let first = directory.path().join("first.yaml"); + let second = directory.path().join("second.yaml"); + std::fs::write( + &first, + indoc! {r#" + data_dir: first-state + sources: + in: + type: demo_logs + format: shuffle + lines: ["log"] + sinks: + out: + inputs: ["in"] + type: blackhole + "#}, + ) + .unwrap(); + std::fs::write(&second, "data_dir: second-state\n").unwrap(); + for subcommand in ["graph", "test"] { + for override_enabled in [true, false] { + let mut command = Command::cargo_bin("vector").unwrap(); + command.env_remove("VECTOR_DATA_DIR").arg(subcommand); + if override_enabled { + command.arg("--data-dir").arg(directory.path()); + } + if subcommand == "graph" { + command + .arg("--config") + .arg(&first) + .arg("--config") + .arg(&second); + } else { + command.arg(&first).arg(&second); + } + let result = command.assert(); + if override_enabled { + result.success(); + } else { + result.failure(); + } + } + } +} diff --git a/website/cue/reference/cli.cue b/website/cue/reference/cli.cue index a4b1c775d2d66..d542e1aab7fce 100644 --- a/website/cue/reference/cli.cue +++ b/website/cue/reference/cli.cue @@ -135,6 +135,12 @@ cli: { } _core_config_options: { + "data-dir": { + description: env_vars.VECTOR_DATA_DIR.description + type: "string" + env_var: "VECTOR_DATA_DIR" + example: "vector --data-dir ./state --config vector.yaml" + } "config": { _short: "c" description: env_vars.VECTOR_CONFIG.description @@ -302,6 +308,7 @@ cli: { flags: _default_flags & _core_flags options: { + "data-dir": _core_config_options["data-dir"] "config-toml": { description: env_vars.VECTOR_CONFIG_TOML.description type: "string" @@ -474,6 +481,7 @@ cli: { } & _core_flags options: { + "data-dir": _core_config_options["data-dir"] "config-yaml": { description: """ Any number of Vector config files to validate. @@ -590,6 +598,24 @@ cli: { } } } + VECTOR_DATA_DIR: { + description: """ + Optionally override the global `data_dir` before configuration files are merged. + The `--data-dir` CLI option takes precedence over `VECTOR_DATA_DIR`, which takes + precedence over configuration-file values. If neither is set, Vector uses the + configured `data_dir` or its default. + + This is a global option, accepted before or after a subcommand. For example, + `vector validate --data-dir ./state vector.yaml` or + `VECTOR_DATA_DIR=./state vector validate vector.yaml` uses `./state` without + editing the configuration. The override is retained through configuration reloads + and Windows service installation. When installing a Windows service, relative + overrides are resolved against the installer's working directory and stored as + absolute paths. Validation still checks whether the directory is usable unless + the relevant environment checks are disabled. + """ + type: string: default: null + } VECTOR_CONFIG: { description: """ Read configuration from one or more files. Wildcard paths are supported. If no files are