diff --git a/changelog.d/file_fingerprint_rotation.fix.md b/changelog.d/file_fingerprint_rotation.fix.md new file mode 100644 index 0000000000000..320efbd888339 --- /dev/null +++ b/changelog.d/file_fingerprint_rotation.fix.md @@ -0,0 +1,3 @@ +Fix a rotation race in the `file` and `kubernetes_logs` sources that could skip or duplicate logs and truncate the beginning of a record. No configuration changes are required. + +authors: pznamensky diff --git a/lib/file-source-common/src/checkpointer.rs b/lib/file-source-common/src/checkpointer.rs index dd9a01bd4c0af..72bad151c2a5e 100644 --- a/lib/file-source-common/src/checkpointer.rs +++ b/lib/file-source-common/src/checkpointer.rs @@ -485,7 +485,7 @@ mod test { .await .expect("writing test data"); - let new = fingerprinter + let (new, _) = fingerprinter .fingerprint(&log_path) .await .expect("getting new checksum"); diff --git a/lib/file-source-common/src/fingerprinter.rs b/lib/file-source-common/src/fingerprinter.rs index 9c17294197d7e..d2e79f52c1a04 100644 --- a/lib/file-source-common/src/fingerprinter.rs +++ b/lib/file-source-common/src/fingerprinter.rs @@ -175,32 +175,27 @@ impl Fingerprinter { } } - /// Returns the `FileFingerprint` of a file, depending on `Fingerprinter::strategy` - pub(crate) async fn fingerprint(&mut self, path: &Path) -> Result { - use FileFingerprint::{DevInode, FirstLinesChecksum}; - - match self.strategy { + pub(crate) async fn fingerprint(&mut self, path: &Path) -> Result<(FileFingerprint, File)> { + let mut file = File::open(path).await?; + let fingerprint = match self.strategy { FingerprintStrategy::DevInode => { - let file_handle = File::open(path).await?; - let file_info = file_handle.file_info().await?; - let dev = file_info.portable_dev(); - let ino = file_info.portable_ino(); - Ok(DevInode(dev, ino)) + let file_info = file.file_info().await?; + FileFingerprint::DevInode(file_info.portable_dev(), file_info.portable_ino()) } FingerprintStrategy::FirstLinesChecksum { ignored_header_bytes, lines, } => { let buffer = self.buffer.resize_slice_mut(self.max_line_length); - let mut fp = File::open(path).await?; - let mut reader = UncompressedReaderImpl::reader(&mut fp).await?; + let mut reader = UncompressedReaderImpl::reader(&mut file).await?; skip_first_n_bytes(&mut reader, ignored_header_bytes).await?; let bytes_read = fingerprinter_read_until(reader, b'\n', lines, buffer).await?; - let fingerprint = FINGERPRINT_CRC.checksum(&buffer[..bytes_read]); - Ok(FirstLinesChecksum(fingerprint)) + FileFingerprint::FirstLinesChecksum(FINGERPRINT_CRC.checksum(&buffer[..bytes_read])) } - } + }; + file.seek(SeekFrom::Start(0)).await?; + Ok((fingerprint, file)) } pub async fn fingerprint_or_emit( @@ -208,7 +203,7 @@ impl Fingerprinter { path: &Path, known_small_files: &mut HashMap, emitter: &impl FileSourceInternalEvents, - ) -> Option { + ) -> Option<(FileFingerprint, File)> { let metadata = match fs::metadata(path).await { Ok(metadata) => { if metadata.is_dir() { @@ -343,8 +338,8 @@ mod test { .is_err() ); assert_eq!( - fingerprinter.fingerprint(&full_line_path).await.unwrap(), - fingerprinter.fingerprint(&duplicate_path).await.unwrap(), + fingerprinter.fingerprint(&full_line_path).await.unwrap().0, + fingerprinter.fingerprint(&duplicate_path).await.unwrap().0, ); } @@ -408,7 +403,7 @@ mod test { max_line_length - 1, ); - let mut run = async |path| fingerprinter.fingerprint(path).await; + let mut run = async |path| fingerprinter.fingerprint(path).await.map(|(id, _)| id); assert!(run(&empty).await.is_err()); assert!(run(&incomplete_line).await.is_err()); @@ -503,7 +498,7 @@ mod test { b"line one\nline two\nine three\n", ); - let mut run = async move |path| fingerprinter.fingerprint(path).await; + let mut run = async move |path| fingerprinter.fingerprint(path).await.map(|(id, _)| id); assert!(run(&incomplete_lines).await.is_err()); @@ -570,7 +565,7 @@ mod test { // Finish the header and append a complete record to the same file. let mut file = fs::OpenOptions::new().append(true).open(&path).unwrap(); file.write_all(b"!record\n").unwrap(); - let fingerprint = fingerprinter.fingerprint(&path).await.unwrap(); + let (fingerprint, _) = fingerprinter.fingerprint(&path).await.unwrap(); // The expected checksum includes only the record, not its header. assert_eq!( fingerprint, @@ -616,7 +611,7 @@ mod test { &gzip(b"some-header-22\nhellow world\nfrom vector\n").await, ); - let mut run = async move |path| fingerprinter.fingerprint(path).await; + let mut run = async move |path| fingerprinter.fingerprint(path).await.map(|(id, _)| id); assert!(run(&two_lines).await.is_ok()); assert_eq!( @@ -654,8 +649,8 @@ mod test { assert!(fingerprinter.fingerprint(&empty_path).await.is_ok()); assert!(fingerprinter.fingerprint(&small_path).await.is_ok()); assert_ne!( - fingerprinter.fingerprint(&medium_path).await.unwrap(), - fingerprinter.fingerprint(&duplicate_path).await.unwrap() + fingerprinter.fingerprint(&medium_path).await.unwrap().0, + fingerprinter.fingerprint(&duplicate_path).await.unwrap().0 ); } diff --git a/lib/file-source/src/file_server.rs b/lib/file-source/src/file_server.rs index 782ea72fe53db..a5b85b67aa888 100644 --- a/lib/file-source/src/file_server.rs +++ b/lib/file-source/src/file_server.rs @@ -19,7 +19,7 @@ use futures::{ use futures_util::future::join_all; use indexmap::IndexMap; use tokio::{ - fs::{self, remove_file}, + fs::{self, File, remove_file}, time::sleep, }; @@ -118,42 +118,33 @@ where let mut known_small_files = HashMap::new(); - let mut existing_files = Vec::new(); - for path in self.paths_provider.paths() { - if let Some(file_id) = self - .fingerprinter - .fingerprint_or_emit(&path, &mut known_small_files, &self.emitter) - .await - { - existing_files.push((path, file_id)); - } - } - - let metadata = join_all( - existing_files - .iter() - .map(|(path, _file_id)| fs::metadata(path)), - ) - .await; + let paths: Vec<_> = self.paths_provider.paths().into_iter().collect(); + let metadata = join_all(paths.iter().map(fs::metadata)).await; let created = metadata.into_iter().map(|m| { m.and_then(|m| m.created()) .map_or_else(|_| Utc::now(), DateTime::::from) }); - let mut existing_files: Vec<(DateTime, PathBuf, FileFingerprint)> = existing_files + let mut existing_files: Vec<(DateTime, PathBuf)> = paths .into_iter() .zip(created) - .map(|((path, file_id), key)| (key, path, file_id)) + .map(|(path, key)| (key, path)) .collect(); - existing_files.sort_by_key(|(key, _, _)| *key); + existing_files.sort_by_key(|(key, _)| *key); let checkpoints = checkpointer.view(); - for (_key, path, file_id) in existing_files { - self.watch_new_file(path, file_id, &mut fp_map, &checkpoints, true) - .await; + for (_key, path) in existing_files { + if let Some((file_id, file)) = self + .fingerprinter + .fingerprint_or_emit(&path, &mut known_small_files, &self.emitter) + .await + { + self.watch_new_file(path, file_id, file, &mut fp_map, &checkpoints, true) + .await; + } } self.emitter.emit_files_open(fp_map.len()); @@ -198,7 +189,7 @@ where watcher.set_file_findable(false); // assume not findable until found } for path in self.paths_provider.paths() { - if let Some(file_id) = self + if let Some((file_id, file)) = self .fingerprinter .fingerprint_or_emit(&path, &mut known_small_files, &self.emitter) .await @@ -219,7 +210,9 @@ where path = ?path, old_path = ?watcher.path ); - if let Ok(Some(unwatch_info)) = watcher.update_path(path).await { + if let Ok(Some(unwatch_info)) = + watcher.update_path(path, file).await + { // Inode changed - emit metrics for the old file self.emitter.emit_file_unwatched( &unwatch_info.path, @@ -233,10 +226,10 @@ where path = ?path, old_path = ?watcher.path ); - let (old_path, new_path) = (&watcher.path, &path); + let old_path = &watcher.path; if let (Ok(old_modified_time), Ok(new_modified_time)) = ( fs::metadata(old_path).await.and_then(|m| m.modified()), - fs::metadata(new_path).await.and_then(|m| m.modified()), + file.metadata().await.and_then(|m| m.modified()), ) && old_modified_time < new_modified_time { info!( @@ -244,7 +237,8 @@ where new_modified_time = ?new_modified_time, old_modified_time = ?old_modified_time, ); - if let Ok(Some(unwatch_info)) = watcher.update_path(path).await + if let Ok(Some(unwatch_info)) = + watcher.update_path(path, file).await { // Inode changed - emit metrics for the old file self.emitter.emit_file_unwatched( @@ -257,8 +251,15 @@ where } } else { // untracked file fingerprint - self.watch_new_file(path, file_id, &mut fp_map, &checkpoints, false) - .await; + self.watch_new_file( + path, + file_id, + file, + &mut fp_map, + &checkpoints, + false, + ) + .await; self.emitter.emit_files_open(fp_map.len()); } } @@ -348,11 +349,12 @@ where && watcher.last_read_success().elapsed() >= grace_period { // Try to remove - match remove_file(&watcher.path).await { - Ok(()) => { + match watcher.remove_file().await { + Ok(true) => { self.emitter.emit_file_deleted(&watcher.path); watcher.set_dead(); } + Ok(false) => {} Err(error) => { // We will try again after some time. self.emitter.emit_file_delete_error(&watcher.path, error); @@ -457,6 +459,7 @@ where &self, path: PathBuf, file_id: FileFingerprint, + file: File, fp_map: &mut IndexMap, checkpoints: &CheckpointsView, startup: bool, @@ -488,6 +491,7 @@ where match FileWatcher::new( path.clone(), + file, read_from, self.ignore_before, self.max_line_bytes, @@ -630,3 +634,397 @@ pub struct Line { pub start_offset: u64, pub end_offset: u64, } + +#[cfg(test)] +mod tests { + use std::{ + fs as std_fs, io, + path::Path, + sync::atomic::{AtomicBool, Ordering}, + }; + + use async_compression::tokio::bufread::GzipEncoder; + use file_source_common::FingerprintStrategy; + use futures::{ + FutureExt as _, StreamExt as _, + channel::{mpsc, oneshot}, + }; + use tokio::{fs::File, io::AsyncReadExt as _}; + + use crate::file_watcher::RawLine; + + use super::*; + + #[derive(Clone, Default)] + struct NoErrors { + rotate_on_add: Option<(PathBuf, &'static [u8])>, + } + + impl FileSourceInternalEvents for NoErrors { + fn emit_file_added(&self, path: &Path) { + if let Some((rotated, replacement)) = &self.rotate_on_add { + std_fs::rename(path, rotated).unwrap(); + std_fs::write(path, replacement).unwrap(); + } + } + fn emit_file_resumed(&self, _: &Path, _: u64) {} + fn emit_file_watch_error(&self, path: &Path, error: io::Error) { + panic!("watch {path:?}: {error}"); + } + fn emit_file_unwatched(&self, _: &Path, _: bool, _: Option) {} + fn emit_file_deleted(&self, _: &Path) {} + fn emit_file_delete_error(&self, path: &Path, error: io::Error) { + panic!("delete {path:?}: {error}"); + } + fn emit_file_fingerprint_read_error(&self, path: &Path, error: io::Error) { + panic!("fingerprint {path:?}: {error}"); + } + fn emit_file_checkpointed(&self, _: usize, _: Duration) {} + fn emit_file_checksum_failed(&self, path: &Path) { + panic!("checksum {path:?}"); + } + fn emit_file_checkpoint_write_error(&self, error: io::Error) { + panic!("checkpoint: {error}"); + } + fn emit_files_open(&self, _: usize) {} + fn emit_path_globbing_failed(&self, path: &Path, error: &io::Error) { + panic!("glob {path:?}: {error}"); + } + fn emit_file_line_too_long(&self, _: &bytes::BytesMut, _: usize, _: usize) { + panic!("unexpected oversized line"); + } + } + + struct NoPaths; + + impl PathsProvider for NoPaths { + type IntoIter = Vec; + + fn paths(&self) -> Self::IntoIter { + Vec::new() + } + } + + struct DiscoverAfterStartup { + path: PathBuf, + startup: AtomicBool, + } + + impl PathsProvider for DiscoverAfterStartup { + type IntoIter = Vec; + + fn paths(&self) -> Self::IntoIter { + if self.startup.swap(false, Ordering::Relaxed) { + Vec::new() + } else { + vec![self.path.clone()] + } + } + } + + fn checksum_strategy() -> FingerprintStrategy { + FingerprintStrategy::FirstLinesChecksum { + ignored_header_bytes: 0, + lines: 1, + } + } + + fn test_server( + data_dir: &Path, + strategy: FingerprintStrategy, + ) -> FileServer { + test_server_with_paths(data_dir, strategy, NoPaths) + } + + fn test_server_with_paths( + data_dir: &Path, + strategy: FingerprintStrategy, + paths_provider: PP, + ) -> FileServer { + FileServer { + paths_provider, + max_read_bytes: 2048, + ignore_checkpoints: false, + read_from: ReadFrom::Beginning, + ignore_before: None, + max_line_bytes: 1024, + line_delimiter: Bytes::from_static(b"\n"), + data_dir: data_dir.to_path_buf(), + glob_minimum_cooldown: Duration::from_millis(25), + fingerprinter: Fingerprinter::new(strategy, 1024, false), + oldest_first: false, + remove_after: None, + emitter: NoErrors::default(), + rotate_wait: Duration::from_mins(1), + } + } + + async fn fingerprint( + server: &mut FileServer, + path: &Path, + ) -> (FileFingerprint, File) { + server + .fingerprinter + .fingerprint_or_emit(path, &mut HashMap::new(), &server.emitter) + .await + .unwrap() + } + + async fn next_line(watcher: &mut FileWatcher) -> RawLine { + watcher.read_line().await.unwrap().raw_line.unwrap() + } + + #[tokio::test] + async fn remove_after_preserves_replacement_file() { + const ORIGINAL: &[u8] = b"original event\n"; + const REPLACEMENT: &[u8] = b"unread replacement event\n"; + + for rotate in [false, true] { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("0.log"); + let rotated = dir.path().join("0.log.rotated"); + std_fs::write(&path, ORIGINAL).unwrap(); + let paths = DiscoverAfterStartup { + path: path.clone(), + startup: AtomicBool::new(true), + }; + let mut server = test_server_with_paths(dir.path(), checksum_strategy(), paths); + server.glob_minimum_cooldown = Duration::from_mins(1); + server.remove_after = Some(Duration::ZERO); + if rotate { + server.emitter.rotate_on_add = Some((rotated.clone(), REPLACEMENT)); + } + let (tx, mut rx) = mpsc::channel(1); + let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>(); + let shutdown = shutdown_rx.shared(); + let run = server.run( + tx, + shutdown.clone(), + shutdown, + Checkpointer::new(dir.path()), + ); + let consume = async { + let batch = rx.next().await.unwrap(); + assert_eq!(batch.len(), 1); + assert_eq!(batch[0].text.as_ref(), &ORIGINAL[..ORIGINAL.len() - 1]); + // Two idle batches allow the EOF read backoff to expire before cleanup. + for _ in 0..2 { + assert!(rx.next().await.unwrap().is_empty()); + } + if rotate { + assert_eq!(std_fs::read(&path).unwrap(), REPLACEMENT); + assert_eq!(std_fs::read(&rotated).unwrap(), ORIGINAL); + } else { + assert!(!path.exists()); + } + shutdown_tx.send(()).unwrap(); + while rx.next().await.is_some() {} + }; + tokio::time::timeout(Duration::from_secs(5), async { + let (result, ()) = futures::future::join(run, consume).await; + result.unwrap(); + }) + .await + .unwrap(); + } + } + + #[tokio::test] + async fn rotation_uses_fingerprinted_handle_and_checkpoint() { + const FIRST: &[u8] = b"2026-10-16T00:31:00.000Z stdout F first event\n"; + const TARGET: &[u8] = b"2026-10-16T00:32:52.539Z stdout F original event\n"; + const REPLACEMENT: &[u8] = b"2026-10-16T00:33:00.000Z stdout F replacement event\n"; + + for strategy in [checksum_strategy(), FingerprintStrategy::DevInode] { + for resume in [false, true] { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("0.log"); + let rotated = dir.path().join("0.log.rotated"); + let mut original = FIRST.to_vec(); + original.extend_from_slice(TARGET); + std_fs::write(&path, &original).unwrap(); + let mut server = test_server(dir.path(), strategy.clone()); + let (original_id, original_file) = fingerprint(&mut server, &path).await; + let checkpointer = Checkpointer::new(dir.path()); + let checkpoints = checkpointer.view(); + if resume { + checkpoints.update(original_id, u64::try_from(FIRST.len()).unwrap()); + } + + std_fs::rename(&path, &rotated).unwrap(); + std_fs::write(&path, REPLACEMENT).unwrap(); + let mut watchers = IndexMap::new(); + server + .watch_new_file( + path.clone(), + original_id, + original_file, + &mut watchers, + &checkpoints, + false, + ) + .await; + let original_watcher = watchers.get_mut(&original_id).unwrap(); + if !resume { + let line = next_line(original_watcher).await; + assert_eq!(line.bytes.as_ref(), &FIRST[..FIRST.len() - 1]); + assert_eq!(line.offset, 0); + } + let line = next_line(original_watcher).await; + assert_eq!(line.bytes.as_ref(), &TARGET[..TARGET.len() - 1]); + assert_eq!(line.offset, u64::try_from(FIRST.len()).unwrap()); + assert_eq!( + original_watcher.get_file_position(), + u64::try_from(original.len()).unwrap() + ); + + let (replacement_id, replacement_file) = fingerprint(&mut server, &path).await; + assert_ne!(replacement_id, original_id); + server + .watch_new_file( + path, + replacement_id, + replacement_file, + &mut watchers, + &checkpoints, + false, + ) + .await; + let line = next_line(watchers.get_mut(&replacement_id).unwrap()).await; + assert_eq!(line.bytes.as_ref(), &REPLACEMENT[..REPLACEMENT.len() - 1]); + assert_eq!(line.offset, 0); + assert_eq!(watchers.len(), 2); + assert_eq!(std_fs::read(rotated).unwrap(), original); + } + } + } + + #[tokio::test] + async fn replacement_uses_fingerprinted_handle_and_preserves_offset() { + const CONTENT: &[u8] = b"same first line\nsecond line\nthird line\n"; + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("original.log"); + let candidate = dir.path().join("candidate.log"); + let renamed = dir.path().join("candidate.log.rotated"); + std_fs::write(&path, CONTENT).unwrap(); + std_fs::write(&candidate, CONTENT).unwrap(); + let mut server = test_server(dir.path(), checksum_strategy()); + let (original_id, original_file) = fingerprint(&mut server, &path).await; + let mut watcher = FileWatcher::new( + path.clone(), + original_file, + ReadFrom::Beginning, + None, + 1024, + Bytes::from_static(b"\n"), + ) + .await + .unwrap(); + let first = next_line(&mut watcher).await; + assert_eq!(first.bytes.as_ref(), b"same first line"); + let position = watcher.get_file_position(); + let (candidate_id, candidate_file) = fingerprint(&mut server, &candidate).await; + assert_eq!(candidate_id, original_id); + + std_fs::rename(&candidate, &renamed).unwrap(); + std_fs::write(&candidate, b"unrelated replacement\n").unwrap(); + let old_info = watcher + .update_path(candidate, candidate_file) + .await + .unwrap() + .unwrap(); + assert_eq!(old_info.path, path); + assert_eq!(watcher.get_file_position(), position); + for expected in [b"second line".as_slice(), b"third line".as_slice()] { + let line = next_line(&mut watcher).await; + assert_eq!(line.bytes.as_ref(), expected); + } + assert!(watcher.read_line().await.unwrap().raw_line.is_none()); + assert_eq!(std_fs::read(renamed).unwrap(), CONTENT); + } + + #[tokio::test] + async fn fingerprinted_gzip_rewinds_and_keeps_skip_semantics() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("input.log.gz"); + let mut compressed = Vec::new(); + GzipEncoder::new(b"first\nsecond\n".as_slice()) + .read_to_end(&mut compressed) + .await + .unwrap(); + std_fs::write(&path, compressed).unwrap(); + let mut server = test_server(dir.path(), checksum_strategy()); + for read_from in [ReadFrom::Beginning, ReadFrom::End, ReadFrom::Checkpoint(6)] { + let (_, file) = fingerprint(&mut server, &path).await; + let mut watcher = FileWatcher::new( + path.clone(), + file, + read_from, + None, + 1024, + Bytes::from_static(b"\n"), + ) + .await + .unwrap(); + if read_from == ReadFrom::Beginning { + for expected in [b"first".as_slice(), b"second".as_slice()] { + let line = next_line(&mut watcher).await; + assert_eq!(line.bytes.as_ref(), expected); + } + } + assert!(watcher.read_line().await.unwrap().raw_line.is_none()); + } + } + + #[tokio::test] + async fn fingerprinted_handle_preserves_start_position_options() { + let cases = [ + ("startup/end", true, ReadFrom::End, false, None, 2), + ("new/end", false, ReadFrom::End, false, None, 0), + ("startup/start", true, ReadFrom::Beginning, false, None, 0), + ("checkpoint/end", true, ReadFrom::End, false, Some(6), 1), + ("checkpoint/new", false, ReadFrom::End, false, Some(6), 1), + ("ignored/end", true, ReadFrom::End, true, Some(6), 2), + ("ignored/new", false, ReadFrom::End, true, Some(6), 0), + ("ignored/start", true, ReadFrom::Beginning, true, Some(6), 0), + ]; + let lines = [b"first".as_slice(), b"second".as_slice()]; + for (name, startup, read_from, ignore_checkpoints, checkpoint, skip) in cases { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("input.log"); + std_fs::write(&path, b"first\nsecond\n").unwrap(); + let mut server = test_server(dir.path(), checksum_strategy()); + server.read_from = read_from; + server.ignore_checkpoints = ignore_checkpoints; + let (file_id, file) = fingerprint(&mut server, &path).await; + let checkpointer = Checkpointer::new(dir.path()); + let checkpoints = checkpointer.view(); + if let Some(position) = checkpoint { + checkpoints.update(file_id, position); + } + let mut watchers = IndexMap::new(); + server + .watch_new_file(path, file_id, file, &mut watchers, &checkpoints, startup) + .await; + let watcher = watchers.get_mut(&file_id).unwrap(); + let mut position = u64::try_from( + lines[..skip] + .iter() + .map(|line| line.len() + 1) + .sum::(), + ) + .unwrap(); + assert_eq!(watcher.get_file_position(), position, "{name}"); + for expected in &lines[skip..] { + let line = next_line(watcher).await; + assert_eq!(line.bytes.as_ref(), *expected, "{name}"); + assert_eq!(line.offset, position, "{name}"); + position += u64::try_from(expected.len() + 1).unwrap(); + } + assert!( + watcher.read_line().await.unwrap().raw_line.is_none(), + "{name}" + ); + } + } +} diff --git a/lib/file-source/src/file_watcher/mod.rs b/lib/file-source/src/file_watcher/mod.rs index 4dc8f8f69315a..753f9a1bc0064 100644 --- a/lib/file-source/src/file_watcher/mod.rs +++ b/lib/file-source/src/file_watcher/mod.rs @@ -158,23 +158,22 @@ impl FileWatcher { /// Create a new `FileWatcher` /// /// The input path will be used by `FileWatcher` to prime its state - /// machine. A `FileWatcher` tracks _only one_ file. This function returns - /// an error if the path does not exist or is not readable by the current process. + /// machine. A `FileWatcher` tracks _only one_ file. The open file must be + /// the same one used for fingerprinting to avoid a race with log rotation. /// /// # Errors - /// Returns an error if opening the file, reading its metadata, or checking - /// its compression header fails. + /// Returns an error if reading file metadata or checking its compression header fails. /// /// # Panics /// Panics if seeking to the initial position in an uncompressed file fails. pub async fn new( path: PathBuf, + f: File, read_from: ReadFrom, ignore_before: Option>, max_line_bytes: usize, line_delimiter: Bytes, ) -> Result { - let f = File::open(&path).await?; let file_info = f.file_info().await?; let (devno, ino) = (file_info.portable_dev(), file_info.portable_ino()); @@ -278,11 +277,12 @@ impl FileWatcher { /// containing metrics about the old file so the caller can emit appropriate events. /// /// # Errors - /// Returns an error if opening the replacement file, reading its metadata - /// or compression header, or restoring the file position fails. - pub async fn update_path(&mut self, path: PathBuf) -> io::Result> { - let new_file = File::open(&path).await?; - + /// Returns an error if reading metadata, checking compression, or restoring the position fails. + pub async fn update_path( + &mut self, + path: PathBuf, + new_file: File, + ) -> io::Result> { let file_info = new_file.file_info().await?; let unwatch_info = if (file_info.portable_dev(), file_info.portable_ino()) == (self.devno, self.inode) { @@ -317,6 +317,21 @@ impl FileWatcher { Ok(unwatch_info) } + /// Skip deletion when rotation has rebound the path to another file. + /// + /// # Errors + /// Returns an error if opening the path, reading metadata, or removing the file fails. + pub(super) async fn remove_file(&self) -> io::Result { + let file = File::open(&self.path).await?; + let file_info = file.file_info().await?; + if (file_info.portable_dev(), file_info.portable_ino()) != (self.devno, self.inode) { + return Ok(false); + } + + tokio::fs::remove_file(&self.path).await?; + Ok(true) + } + pub fn set_file_findable(&mut self, f: bool) { self.findable = f; if f { diff --git a/lib/file-source/src/file_watcher/tests/bytes_unread.rs b/lib/file-source/src/file_watcher/tests/bytes_unread.rs index dd2e5c8233d65..8d1f1d31d6b00 100644 --- a/lib/file-source/src/file_watcher/tests/bytes_unread.rs +++ b/lib/file-source/src/file_watcher/tests/bytes_unread.rs @@ -25,6 +25,7 @@ async fn test_bytes_unread_basic() { // Create watcher starting from beginning let mut fw = FileWatcher::new( path.clone(), + tokio::fs::File::open(&path).await.unwrap(), ReadFrom::Beginning, None, 100_000, @@ -76,6 +77,7 @@ async fn test_bytes_unread_after_delete() { // Create watcher starting from beginning let mut fw = FileWatcher::new( path.clone(), + tokio::fs::File::open(&path).await.unwrap(), ReadFrom::Beginning, None, 100_000, @@ -112,6 +114,7 @@ async fn test_bytes_unread_growing_file() { // Create watcher let mut fw = FileWatcher::new( path.clone(), + tokio::fs::File::open(&path).await.unwrap(), ReadFrom::Beginning, None, 100_000, @@ -154,6 +157,7 @@ async fn test_bytes_unread_gzip_is_unknown_even_at_eof() { write_gzip(&path).await; let mut fw = FileWatcher::new( path.clone(), + tokio::fs::File::open(&path).await.unwrap(), ReadFrom::Beginning, None, 100_000, @@ -177,7 +181,14 @@ async fn test_bytes_unread_gzip_is_unknown_even_at_eof() { // replacement plain reader can report the bytes beyond its inherited offset. let replacement = dir.path().join("replacement.log"); fs::write(&replacement, b"line1\nline2\nline3\n").unwrap(); - let old_info = fw.update_path(replacement).await.unwrap().unwrap(); + let old_info = fw + .update_path( + replacement.clone(), + tokio::fs::File::open(&replacement).await.unwrap(), + ) + .await + .unwrap() + .unwrap(); assert_eq!(old_info.path, path); assert_eq!(old_info.bytes_unread, None); assert!(old_info.reached_eof); @@ -197,6 +208,7 @@ async fn test_bytes_unread_skipped_gzip_is_unknown() { ] { let mut fw = FileWatcher::new( path.clone(), + tokio::fs::File::open(&path).await.unwrap(), read_from, ignore_before, 100_000, diff --git a/lib/file-source/src/file_watcher/tests/experiment.rs b/lib/file-source/src/file_watcher/tests/experiment.rs index d77bf26feaf01..c3390fcb4b046 100644 --- a/lib/file-source/src/file_watcher/tests/experiment.rs +++ b/lib/file-source/src/file_watcher/tests/experiment.rs @@ -30,6 +30,7 @@ async fn experiment(actions: Vec) { let mut rotation_count = 0; let mut fw = FileWatcher::new( path.clone(), + tokio::fs::File::open(&path).await.unwrap(), ReadFrom::Beginning, None, 100_000, diff --git a/lib/file-source/src/file_watcher/tests/experiment_no_truncations.rs b/lib/file-source/src/file_watcher/tests/experiment_no_truncations.rs index d68b8218731ca..0c7cb28550b12 100644 --- a/lib/file-source/src/file_watcher/tests/experiment_no_truncations.rs +++ b/lib/file-source/src/file_watcher/tests/experiment_no_truncations.rs @@ -18,6 +18,7 @@ async fn experiment_no_truncations(actions: Vec) { let mut rotation_count = 0; let mut fw = FileWatcher::new( path.clone(), + tokio::fs::File::open(&path).await.unwrap(), ReadFrom::Beginning, None, 100_000, diff --git a/lib/file-source/src/file_watcher/tests/mod.rs b/lib/file-source/src/file_watcher/tests/mod.rs index c999ec27f8074..dae958c72d02a 100644 --- a/lib/file-source/src/file_watcher/tests/mod.rs +++ b/lib/file-source/src/file_watcher/tests/mod.rs @@ -194,7 +194,8 @@ async fn gzip_multi_stream_reads_all_members() { fs::write(&path, &bytes).unwrap(); let mut fw = FileWatcher::new( - path, + path.clone(), + tokio::fs::File::open(&path).await.unwrap(), file_source_common::ReadFrom::Beginning, None, 100_000, @@ -289,6 +290,7 @@ async fn updating_path_resets_eof_backoff() { std::fs::write(&old_path, b"first\n").unwrap(); let mut watcher = FileWatcher::new( old_path.clone(), + tokio::fs::File::open(&old_path).await.unwrap(), file_source_common::ReadFrom::Beginning, None, 1024, @@ -307,7 +309,13 @@ async fn updating_path_resets_eof_backoff() { } else { std::fs::rename(&old_path, &new_path).unwrap(); } - let old_info = watcher.update_path(new_path).await.unwrap(); + let old_info = watcher + .update_path( + new_path.clone(), + tokio::fs::File::open(&new_path).await.unwrap(), + ) + .await + .unwrap(); assert_eq!(old_info.is_some(), replace_file); if let Some(old_info) = old_info { assert!(old_info.reached_eof);