feat(file source): notify-based discovery + idle handle closing (#3567)

Cargo.toml

@@ -188,6 +188,7 @@ metrics-tracing-context = { version = "0.17.0", default-features = false }
 metrics-util = { version = "0.18.0", default-features = false, features = ["registry"] }
 mlua = { version = "0.11", default-features = false, features = ["lua54", "send", "vendored"] }
 nom = { version = "8.0.0", default-features = false }
+notify = { version = "8.1.0", default-features = false, features = ["macos_fsevent"] }
 ordered-float = { version = "5.3.0", default-features = false }
 pastey = { version = "0.2", default-features = false }
 pin-project = { version = "1.1.11", default-features = false }
@@ -443,7 +444,7 @@ mongodb = { version = "3.7.0", default-features = false, optional = true, featur
 async-nats = { version = "0.49.0", default-features = false, optional = true, features = ["ring", "websockets", "jetstream", "nkeys"] }
 nkeys = { version = "0.4.5", default-features = false, optional = true }
 nom = { workspace = true, optional = true }
-notify = { version = "8.1.0", default-features = false, features = ["macos_fsevent"] }
+notify.workspace = true
 openssl = { version = "0.10.73", default-features = false, features = ["vendored"] }
 openssl-probe = { version = "0.1.6", default-features = false }
 ordered-float.workspace = true

changelog.d/3567_file_source_notify_discovery.enhancement.md

@@ -0,0 +1,13 @@
+The `file` source now supports an opt-in `file_discovery_mode: notify` setting that uses OS-level
+file system event notifications (inotify on Linux, FSEvents on macOS, `ReadDirectoryChangesW` on
+Windows) instead of periodic glob re-scanning to discover new files and wake up reads. This avoids
+the cost of re-globbing and re-fingerprinting every matched file on a fixed interval. A much less
+frequent periodic reconciliation pass (`reconcile_interval_secs`) still runs as a correctness
+backstop. The default remains the existing polling-based `file_discovery_mode: polling` behavior.
+
+Separately (and independently of `file_discovery_mode`), a new `idle_timeout_secs` option (default:
+60 seconds) closes a file's handle once it has reached EOF and received no new data for that long,
+avoiding holding a large number of open file handles for files that are being watched but aren't
+actively being written to. See the `idle_timeout_secs` documentation for details.
+
+authors: sashamelentiev

lib/file-source-common/Cargo.toml

@@ -11,7 +11,7 @@ workspace = true
 
 [target.'cfg(windows)'.dependencies]
 libc.workspace = true
-winapi = { version = "0.3", features = ["winioctl"] }
+winapi = { version = "0.3", features = ["winioctl", "ioapiset"] }
 
 [dependencies]
 chrono.workspace = true

lib/file-source-common/src/fingerprinter.rs

@@ -690,6 +690,8 @@ mod test {
 
         fn emit_files_open(&self, _: usize) {}
 
+        fn emit_files_idle(&self, _: usize) {}
+
         fn emit_path_globbing_failed(&self, _: &Path, _: &Error) {
             panic!()
         }

lib/file-source-common/src/internal_events.rs

@@ -25,8 +25,16 @@ pub trait FileSourceInternalEvents: Send + Sync + Clone + 'static {
 
     fn emit_file_checkpoint_write_error(&self, error: Error);
 
+    /// Number of files with an actually-open file handle (i.e. `Active`
+    /// watchers). Distinct from the total number of tracked files, which may
+    /// also include `Idle` watchers that hold no handle at all.
     fn emit_files_open(&self, count: usize);
 
+    /// Number of tracked files currently in the passive `Idle` state: no open
+    /// file handle, checkpoint retained, polled only via cheap `fs::metadata`
+    /// stats. See <https://github.com/vectordotdev/vector/issues/3567>.
+    fn emit_files_idle(&self, count: usize);
+
     fn emit_path_globbing_failed(&self, path: &Path, error: &Error);
 
     fn emit_file_line_too_long(
@@ -35,4 +43,20 @@ pub trait FileSourceInternalEvents: Send + Sync + Clone + 'static {
         configured_limit: usize,
         encountered_size_so_far: usize,
     );
+
+    /// Emitted when the OS-level filesystem event watcher (if in use) reports that its
+    /// internal event queue overflowed, meaning some events may have been silently dropped.
+    /// Implementors should log this loudly, since it means the event-driven discovery path
+    /// may have missed file creations/modifications until the next reconciliation pass.
+    fn emit_file_watch_events_overflowed(&self) {}
+
+    /// Emitted when the OS-level filesystem event watcher itself fails (e.g. the watched
+    /// directory disappears, or the OS notification API errors out). The event-driven
+    /// discovery path will keep relying on the periodic reconciliation pass until watching
+    /// can be re-established.
+    fn emit_file_watch_backend_error(&self, _error: &Error) {}
+
+    /// Emitted once at startup (or when watched directories change) to report how many
+    /// directories are being watched via OS-level notifications.
+    fn emit_file_watch_directories(&self, _count: usize) {}
 }

lib/file-source/Cargo.toml

@@ -20,6 +20,7 @@ futures-util.workspace = true
 vector-common = { path = "../vector-common", default-features = false }
 file-source-common = { path = "../file-source-common" }
 async-compression.workspace = true
+notify.workspace = true
 
 [dev-dependencies]
 tokio = { workspace = true, features = ["full"] }