feat(cli): add support for poll watcher

changelog.d/poll-watcher-support.enhancement.md

@@ -0,0 +1,3 @@
+Support for watching config file changes by polling at certain interval.
+
+authors: amribm

src/app.rs

@@ -1,6 +1,11 @@
 #![allow(missing_docs)]
-use std::sync::atomic::{AtomicUsize, Ordering};
-use std::{num::NonZeroUsize, path::PathBuf, process::ExitStatus, time::Duration};
+use std::{
+    num::{NonZeroU64, NonZeroUsize},
+    path::PathBuf,
+    process::ExitStatus,
+    sync::atomic::{AtomicUsize, Ordering},
+    time::Duration,
+};
 
 use exitcode::ExitCode;
 use futures::StreamExt;
@@ -12,7 +17,7 @@ use crate::extra_context::ExtraContext;
 #[cfg(feature = "api")]
 use crate::{api, internal_events::ApiStarted};
 use crate::{
-    cli::{handle_config_errors, LogFormat, Opts, RootOpts},
+    cli::{handle_config_errors, LogFormat, Opts, RootOpts, WatchConfigMethod},
     config::{self, Config, ConfigPath},
     heartbeat,
     internal_events::{VectorConfigLoadError, VectorQuit, VectorStarted, VectorStopped},
@@ -63,9 +68,18 @@ impl ApplicationConfig {
         let graceful_shutdown_duration = (!opts.no_graceful_shutdown_limit)
             .then(|| Duration::from_secs(u64::from(opts.graceful_shutdown_limit_secs)));
 
+        let watcher_conf = if opts.watch_config {
+            Some(watcher_config(
+                opts.watch_config_method,
+                opts.watch_config_poll_interval_seconds,
+            ))
+        } else {
+            None
+        };
+
         let config = load_configs(
             &config_paths,
-            opts.watch_config,
+            watcher_conf,
             opts.require_healthy,
             opts.allow_empty_config,
             graceful_shutdown_duration,
@@ -470,17 +484,18 @@ pub fn build_runtime(threads: Option<usize>, thread_name: &str) -> Result<Runtim
 
 pub async fn load_configs(
     config_paths: &[ConfigPath],
-    watch_config: bool,
+    watcher_conf: Option<config::watcher::WatcherConfig>,
     require_healthy: Option<bool>,
     allow_empty_config: bool,
     graceful_shutdown_duration: Option<Duration>,
     signal_handler: &mut SignalHandler,
 ) -> Result<Config, ExitCode> {
     let config_paths = config::process_paths(config_paths).ok_or(exitcode::CONFIG)?;
 
-    if watch_config {
+    if let Some(watcher_conf) = watcher_conf {
         // Start listening for config changes immediately.
         config::watcher::spawn_thread(
+            watcher_conf,
             signal_handler.clone_tx(),
             config_paths.iter().map(Into::into),
             None,
@@ -530,3 +545,13 @@ pub fn init_logging(color: bool, format: LogFormat, log_level: &str, rate: u64)
     );
     info!(message = "Log level is enabled.", level = ?level);
 }
+
+pub fn watcher_config(
+    method: WatchConfigMethod,
+    interval: NonZeroU64,
+) -> config::watcher::WatcherConfig {
+    match method {
+        WatchConfigMethod::Recommended => config::watcher::WatcherConfig::RecommendedWatcher,
+        WatchConfigMethod::Poll => config::watcher::WatcherConfig::PollWatcher(interval.into()),
+    }
+}

src/cli.rs

@@ -152,6 +152,31 @@ pub struct RootOpts {
     #[arg(short, long, env = "VECTOR_WATCH_CONFIG")]
     pub watch_config: bool,
 
+    /// Method for configuration watching.
+    ///
+    /// By default, `vector` uses recommended watcher for host OS
+    /// - `inotify` for Linux-based systems.
+    /// - `kqueue` for unix/macos
+    /// - `ReadDirectoryChangesWatcher` for windows
+    ///
+    /// The `poll` watcher can be used in cases where `inotify` doesn't work, e.g., when attaching the configuration via NFS.
+    #[arg(
+        long,
+        default_value = "recommended",
+        env = "VECTOR_WATCH_CONFIG_METHOD"
+    )]
+    pub watch_config_method: WatchConfigMethod,
+
+    /// Poll for changes in the configuration file at the given interval.
+    ///
+    /// This setting is only applicable if `Poll` is set in `--watch-config-method`.
+    #[arg(
+        long,
+        env = "VECTOR_WATCH_CONFIG_POLL_INTERVAL_SECONDS",
+        default_value = "30"
+    )]
+    pub watch_config_poll_interval_seconds: NonZeroU64,
+
     /// Set the internal log rate limit
     #[arg(
         short,
@@ -354,6 +379,15 @@ pub enum LogFormat {
     Json,
 }
 
+#[derive(clap::ValueEnum, Debug, Clone, Copy, PartialEq, Eq)]
+pub enum WatchConfigMethod {
+    /// Recommended watcher for the current OS, usually `inotify` for Linux-based systems.
+    Recommended,
+    /// Poll-based watcher, typically used for watching files on EFS/NFS-like network storage systems.
+    /// The interval is determined by  [`RootOpts::watch_config_poll_interval_seconds`].
+    Poll,
+}
+
 pub fn handle_config_errors(errors: Vec<String>) -> exitcode::ExitCode {
     for error in errors {
         error!(message = "Configuration error.", %error);

src/config/watcher.rs

@@ -1,10 +1,13 @@
-use std::{path::PathBuf, time::Duration};
+use std::{
+    path::{Path, PathBuf},
+    time::Duration,
+};
 use std::{
     sync::mpsc::{channel, Receiver},
     thread,
 };
 
-use notify::{recommended_watcher, EventKind, RecommendedWatcher, RecursiveMode, Watcher};
+use notify::{recommended_watcher, EventKind, RecursiveMode};
 
 use crate::Error;
 
@@ -19,11 +22,49 @@ const CONFIG_WATCH_DELAY: std::time::Duration = std::time::Duration::from_secs(1
 
 const RETRY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
 
+/// Refer to [`crate::cli::WatchConfigMethod`] for details.
+pub enum WatcherConfig {
+    /// Recommended watcher for the current OS.
+    RecommendedWatcher,
+    /// A poll-based watcher that checks for file changes at regular intervals.
+    PollWatcher(u64),
+}
+
+enum Watcher {
+    /// recommended watcher for os, usually inotify for linux based systems
+    RecommendedWatcher(notify::RecommendedWatcher),
+    /// poll based watcher. for watching files from NFS.
+    PollWatcher(notify::PollWatcher),
+}
+
+impl Watcher {
+    fn add_paths(&mut self, config_paths: &[PathBuf]) -> Result<(), Error> {
+        for path in config_paths {
+            self.watch(path, RecursiveMode::Recursive)?;
+        }
+        Ok(())
+    }
+
+    fn watch(&mut self, path: &Path, recursive_mode: RecursiveMode) -> Result<(), Error> {
+        use notify::Watcher as NotifyWatcher;
+        match self {
+            Watcher::RecommendedWatcher(watcher) => {
+                watcher.watch(path, recursive_mode)?;
+            }
+            Watcher::PollWatcher(watcher) => {
+                watcher.watch(path, recursive_mode)?;
+            }
+        }
+        Ok(())
+    }
+}
+
 /// Sends a ReloadFromDisk on config_path changes.
 /// Accumulates file changes until no change for given duration has occurred.
 /// Has best effort guarantee of detecting all file changes from the end of
 /// this function until the main thread stops.
 pub fn spawn_thread<'a>(
+    watcher_conf: WatcherConfig,
     signal_tx: crate::signal::SignalTx,
     config_paths: impl IntoIterator<Item = &'a PathBuf> + 'a,
     delay: impl Into<Option<Duration>>,
@@ -33,7 +74,7 @@ pub fn spawn_thread<'a>(
 
     // Create watcher now so not to miss any changes happening between
     // returning from this function and the thread starting.
-    let mut watcher = Some(create_watcher(&config_paths)?);
+    let mut watcher = Some(create_watcher(&watcher_conf, &config_paths)?);
 
     info!("Watching configuration files.");
 
@@ -53,7 +94,7 @@ pub fn spawn_thread<'a>(
 
                     // We need to read paths to resolve any inode changes that may have happened.
                     // And we need to do it before raising sighup to avoid missing any change.
-                    if let Err(error) = add_paths(&mut watcher, &config_paths) {
+                    if let Err(error) = watcher.add_paths(&config_paths) {
                         error!(message = "Failed to read files to watch.", %error);
                         break;
                     }
@@ -72,7 +113,7 @@ pub fn spawn_thread<'a>(
 
         thread::sleep(RETRY_TIMEOUT);
 
-        watcher = create_watcher(&config_paths)
+        watcher = create_watcher(&watcher_conf, &config_paths)
             .map_err(|error| error!(message = "Failed to create file watcher.", %error))
             .ok();
 
@@ -91,28 +132,28 @@ pub fn spawn_thread<'a>(
 }
 
 fn create_watcher(
+    watcher_conf: &WatcherConfig,
     config_paths: &[PathBuf],
-) -> Result<
-    (
-        RecommendedWatcher,
-        Receiver<Result<notify::Event, notify::Error>>,
-    ),
-    Error,
-> {
+) -> Result<(Watcher, Receiver<Result<notify::Event, notify::Error>>), Error> {
     info!("Creating configuration file watcher.");
+
     let (sender, receiver) = channel();
-    let mut watcher = recommended_watcher(sender)?;
-    add_paths(&mut watcher, config_paths)?;
+    let mut watcher = match watcher_conf {
+        WatcherConfig::RecommendedWatcher => {
+            let recommended_watcher = recommended_watcher(sender)?;
+            Watcher::RecommendedWatcher(recommended_watcher)
+        }
+        WatcherConfig::PollWatcher(interval) => {
+            let config =
+                notify::Config::default().with_poll_interval(Duration::from_secs(*interval));
+            let poll_watcher = notify::PollWatcher::new(sender, config)?;
+            Watcher::PollWatcher(poll_watcher)
+        }
+    };
+    watcher.add_paths(config_paths)?;
     Ok((watcher, receiver))
 }
 
-fn add_paths(watcher: &mut RecommendedWatcher, config_paths: &[PathBuf]) -> Result<(), Error> {
-    for path in config_paths {
-        watcher.watch(path, RecursiveMode::Recursive)?;
-    }
-    Ok(())
-}
-
 #[cfg(all(test, unix, not(target_os = "macos")))] // https://github.com/vectordotdev/vector/issues/5000
 mod tests {
     use super::*;
@@ -140,12 +181,13 @@ mod tests {
         let delay = Duration::from_secs(3);
         let dir = temp_dir().to_path_buf();
         let file_path = dir.join("vector.toml");
+        let watcher_conf = WatcherConfig::RecommendedWatcher;
 
         std::fs::create_dir(&dir).unwrap();
         let mut file = File::create(&file_path).unwrap();
 
         let (signal_tx, signal_rx) = broadcast::channel(128);
-        spawn_thread(signal_tx, &[dir], delay).unwrap();
+        spawn_thread(watcher_conf, signal_tx, &[dir], delay).unwrap();
 
         if !test(&mut file, delay * 5, signal_rx).await {
             panic!("Test timed out");
@@ -159,9 +201,10 @@ mod tests {
         let delay = Duration::from_secs(3);
         let file_path = temp_file();
         let mut file = File::create(&file_path).unwrap();
+        let watcher_conf = WatcherConfig::RecommendedWatcher;
 
         let (signal_tx, signal_rx) = broadcast::channel(128);
-        spawn_thread(signal_tx, &[file_path], delay).unwrap();
+        spawn_thread(watcher_conf, signal_tx, &[file_path], delay).unwrap();
 
         if !test(&mut file, delay * 5, signal_rx).await {
             panic!("Test timed out");
@@ -179,8 +222,10 @@ mod tests {
         let mut file = File::create(&file_path).unwrap();
         std::os::unix::fs::symlink(&file_path, &sym_file).unwrap();
 
+        let watcher_conf = WatcherConfig::RecommendedWatcher;
+
         let (signal_tx, signal_rx) = broadcast::channel(128);
-        spawn_thread(signal_tx, &[sym_file], delay).unwrap();
+        spawn_thread(watcher_conf, signal_tx, &[sym_file], delay).unwrap();
 
         if !test(&mut file, delay * 5, signal_rx).await {
             panic!("Test timed out");
@@ -195,12 +240,13 @@ mod tests {
         let dir = temp_dir().to_path_buf();
         let sub_dir = dir.join("sources");
         let file_path = sub_dir.join("input.toml");
+        let watcher_conf = WatcherConfig::RecommendedWatcher;
 
         std::fs::create_dir_all(&sub_dir).unwrap();
         let mut file = File::create(&file_path).unwrap();
 
         let (signal_tx, signal_rx) = broadcast::channel(128);
-        spawn_thread(signal_tx, &[sub_dir], delay).unwrap();
+        spawn_thread(watcher_conf, signal_tx, &[sub_dir], delay).unwrap();
 
         if !test(&mut file, delay * 5, signal_rx).await {
             panic!("Test timed out");

website/content/en/docs/administration/management.md

@@ -218,6 +218,13 @@ As you can see above, many administrative interfaces for Vector enable you to tr
 
 You can make Vector automatically reload itself when its [configuration file][configuration] changes by setting the `--watch-config` or `-w` [flag][watch_config] when you first start your Vector instance.
 
+Additionally you can add method for watching config change by setting the `--watch-config-method` to `recommended` or `poll`.
+
+`recommended` is default and it uses file event listener for file change events.
+`poll` can used where the event listener won't work, eg. attaching config files by NFS/EFS, which will poll for file changes on certain interval.
+
+You can set the poll interval by setting flag `--watch-config-poll-interval-seconds`. which defaults to `30`.
+
 ## How it works
 
 Running Vector instances accept the IPC [signals](#signals) and produce the [exit codes](#exit-codes) listed below.

website/cue/reference/cli.cue

@@ -157,6 +157,18 @@ cli: {
 			env_var:     "VECTOR_GRACEFUL_SHUTDOWN_LIMIT_SECS"
 			type:        "integer"
 		}
+		"watch-config-poll-interval-seconds": {
+			description: env_vars.VECTOR_WATCH_CONFIG_POLL_INTERVAL_SECONDS.description
+			env_var:     "VECTOR_WATCH_CONFIG_POLL_INTERVAL_SECONDS"
+			default:     env_vars.VECTOR_GRACEFUL_SHUTDOWN_LIMIT_SECS.type.uint.default
+			type:        "integer"
+		}
+		"watch-config-method": {
+			description: env_vars.VECTOR_WATCH_CONFIG_METHOD.description
+			env_var:     "VECTOR_WATCH_CONFIG_METHOD"
+			default:     env_vars.VECTOR_WATCH_CONFIG_METHOD.type.string.default
+			type:        "string"
+		}
 	}
 
 	// Reusable options
@@ -624,6 +636,25 @@ cli: {
 			description: "Watch for changes in the configuration file and reload accordingly"
 			type: bool: default: false
 		}
+		VECTOR_WATCH_CONFIG_METHOD: {
+			description: """
+                Method for watching config files.
+
+                `recommend` - recommended event based watcher for host OS
+                `poll` - `poll` watcher can be used in cases where event based watcher doesn't work, e.g., when attaching the configuration via NFS.
+                """
+			type: string: default: "recommended"
+		}
+		VECTOR_WATCH_CONFIG_POLL_INTERVAL_SECONDS: {
+			description: """
+                Poll for config changes at given interval
+                only applicable if `poll` is set in `--watch-config-method`
+                """
+			type: uint: {
+				default: 30
+				unit:    "seconds"
+			}
+		}
 		VECTOR_INTERNAL_LOG_RATE_LIMIT: {
 			description: "Set the internal log rate limit. This limits Vector from emitting identical logs more than once over the given number of seconds."
 			type: uint: {