microsandbox

//! Log rotation implementation for the Microsandbox runtime. //! //! This module provides a rotating log implementation that automatically rotates log files //! when they reach a specified size. The rotation process involves: //! 1. Renaming the current log file to .old extension //! 2. Creating a new empty log file //! 3. Continuing writing to the new file //! //! The implementation is fully asynchronous and implements AsyncWrite. use futures::future::BoxFuture; use std::{ io::{self, Write}, path::{Path, PathBuf}, pin::Pin, sync::{ atomic::{AtomicU64, Ordering}, Arc, }, task::{Context, Poll}, }; use tokio::{ fs::{remove_file, rename, File, OpenOptions}, io::{AsyncWrite, AsyncWriteExt}, sync::mpsc::{self, UnboundedReceiver, UnboundedSender}, task::JoinHandle, }; use crate::DEFAULT_LOG_MAX_SIZE; //-------------------------------------------------------------------------------------------------- // Types //-------------------------------------------------------------------------------------------------- /// A rotating log file that automatically rotates when reaching a maximum size. /// /// The log rotation process preserves the last full log file with a ".old" extension /// while continuing to write to a new log file with the original name. /// /// # Example /// /// ```no_run /// use microsandbox_utils::log::RotatingLog; /// /// #[tokio::main] /// async fn main() -> std::io::Result<()> { /// let log = RotatingLog::new("app.log").await?; // 1MB max size /// Ok(()) /// } /// ``` pub struct RotatingLog { /// The current log file being written to file: File, /// Path to the current log file path: PathBuf, /// Maximum size in bytes before rotation max_size: u64, /// Current size of the log file (shared between sync and async paths) current_size: Arc<AtomicU64>, /// Current state of the log rotation state: State, /// Channel for sending data to sync writer tx: UnboundedSender<Vec<u8>>, /// Background task handle _background_task: JoinHandle<()>, } /// Internal state machine for managing log rotation enum State { /// Normal operation, ready to accept writes Idle, /// Currently performing log rotation Rotating(RotationFuture), /// Currently writing data Writing, } /// A sync writer that sends all written data to a channel. pub struct SyncChannelWriter { tx: UnboundedSender<Vec<u8>>, } type RotationFuture = BoxFuture<'static, io::Result<(File, PathBuf)>>; //-------------------------------------------------------------------------------------------------- // Methods //-------------------------------------------------------------------------------------------------- impl RotatingLog { /// Creates a new rotating log file with the default maximum size. /// /// This is a convenience wrapper around [`with_max_size`] that uses the default /// maximum log file size defined in `DEFAULT_LOG_MAX_SIZE`. /// /// ## Arguments /// /// * `path` - Path to the log file /// /// ## Errors /// /// Will return an error if: /// * The file cannot be created or opened /// * File metadata cannot be read pub async fn new(path: impl AsRef<Path>) -> io::Result<Self> { Self::with_max_size(path, DEFAULT_LOG_MAX_SIZE).await } /// Creates a new rotating log file. /// /// ## Errors /// /// Will return an error if: /// * The file cannot be created or opened /// * File metadata cannot be read pub async fn with_max_size(path: impl AsRef<Path>, max_size: u64) -> io::Result<Self> { let path = path.as_ref().to_path_buf(); let file = OpenOptions::new() .create(true) .append(true) .open(&path) .await?; let metadata = file.metadata().await?; let (tx, rx) = mpsc::unbounded_channel(); // Create shared atomic counter for current size let current_size = Arc::new(AtomicU64::new(metadata.len())); // Create a clone of the file and size counter for the background task let bg_file = file.try_clone().await?; let bg_path = path.clone(); let bg_max_size = max_size; let bg_size = Arc::clone(&current_size); // Spawn background task to handle channel data let background_task = tokio::spawn(async move { handle_channel_data(rx, bg_file, bg_path, bg_max_size, bg_size).await }); Ok(Self { file, path, max_size, current_size, state: State::Idle, tx, _background_task: background_task, }) } /// Get a sync writer that implements std::io::Write pub fn get_sync_writer(&self) -> SyncChannelWriter { SyncChannelWriter::new(self.tx.clone()) } } impl SyncChannelWriter { /// Creates a new `SyncChannelWriter` with the given channel sender. pub fn new(tx: UnboundedSender<Vec<u8>>) -> Self { Self { tx } } } //-------------------------------------------------------------------------------------------------- // Functions //-------------------------------------------------------------------------------------------------- /// Performs the actual log rotation operation. /// /// # Arguments /// /// * `file` - The current log file to be rotated /// * `path` - Path to the current log file /// /// # Returns /// /// Returns a tuple containing: /// * The newly created log file /// * The path to the new log file /// /// # Errors /// /// Will return an error if: /// * File synchronization fails /// * Old backup file cannot be removed /// * File rename operation fails /// * New log file cannot be created async fn do_rotation(file: File, path: PathBuf) -> io::Result<(File, PathBuf)> { file.sync_all().await?; let backup_path = path.with_extension("old"); if backup_path.exists() { remove_file(&backup_path).await?; } rename(&path, &backup_path).await?; let new_file = OpenOptions::new() .create(true) .append(true) .open(&path) .await?; Ok((new_file, path)) } /// Background task that handles data from the sync channel async fn handle_channel_data( mut rx: UnboundedReceiver<Vec<u8>>, mut file: File, path: PathBuf, max_size: u64, current_size: Arc<AtomicU64>, ) { while let Some(data) = rx.recv().await { let data_len = data.len() as u64; let size = current_size.fetch_add(data_len, Ordering::Relaxed); if size + data_len > max_size { // Clone the file handle before rotation if let Ok(file_clone) = file.try_clone().await { match do_rotation(file_clone, path.clone()).await { Ok((new_file, _)) => { file = new_file; current_size.store(0, Ordering::Relaxed); } Err(e) => { tracing::error!("failed to rotate log file: {}", e); continue; } } } else { tracing::error!("failed to clone file handle for rotation"); continue; } } if let Err(e) = file.write_all(&data).await { tracing::error!("failed to write to log file: {}", e); // On write error, subtract the size we added current_size.fetch_sub(data_len, Ordering::Relaxed); } } } //-------------------------------------------------------------------------------------------------- // Trait Implementations //-------------------------------------------------------------------------------------------------- impl AsyncWrite for RotatingLog { fn poll_write( mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8], ) -> Poll<io::Result<usize>> { let this = &mut *self; let buf_len = buf.len() as u64; loop { match &mut this.state { State::Idle => { let size = this.current_size.fetch_add(buf_len, Ordering::Relaxed); if size + buf_len > this.max_size { let old_file = std::mem::replace( &mut this.file, File::from_std(std::fs::File::open("/dev/null").unwrap()), ); let old_path = this.path.clone(); let fut = Box::pin(do_rotation(old_file, old_path)); this.state = State::Rotating(fut); } else { this.state = State::Writing; } } State::Rotating(fut) => { match fut.as_mut().poll(cx) { Poll::Pending => return Poll::Pending, Poll::Ready(Err(e)) => { this.state = State::Idle; // On rotation error, subtract the size we added this.current_size.fetch_sub(buf_len, Ordering::Relaxed); return Poll::Ready(Err(e)); } Poll::Ready(Ok((new_file, new_path))) => { this.file = new_file; this.path = new_path; this.current_size.store(0, Ordering::Relaxed); this.state = State::Writing; } } } State::Writing => { let pinned_file = Pin::new(&mut this.file); match pinned_file.poll_write(cx, buf) { Poll::Ready(Ok(written)) => { this.state = State::Idle; return Poll::Ready(Ok(written)); } Poll::Ready(Err(e)) => { this.state = State::Idle; // On write error, subtract the size we added this.current_size.fetch_sub(buf_len, Ordering::Relaxed); return Poll::Ready(Err(e)); } Poll::Pending => return Poll::Pending, } } } } } fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> { Pin::new(&mut self.file).poll_flush(cx) } fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> { Pin::new(&mut self.file).poll_shutdown(cx) } } impl Write for SyncChannelWriter { fn write(&mut self, buf: &[u8]) -> io::Result<usize> { let data = buf.to_vec(); self.tx.send(data).map_err(|_| { io::Error::new(io::ErrorKind::Other, "failed to send log data to channel") })?; Ok(buf.len()) } fn flush(&mut self) -> io::Result<()> { Ok(()) } } //-------------------------------------------------------------------------------------------------- // Tests //-------------------------------------------------------------------------------------------------- #[cfg(test)] mod tests { use super::*; use std::fs; use tempfile::tempdir; use tokio::io::AsyncWriteExt; #[tokio::test] async fn test_create_new_log() -> io::Result<()> { let dir = tempdir()?; let log_path = dir.path().join("test.log"); let log = RotatingLog::with_max_size(&log_path, 1024).await?; assert!(log_path.exists()); assert_eq!(log.max_size, 1024); assert_eq!(log.current_size.load(Ordering::Relaxed), 0); Ok(()) } #[tokio::test] async fn test_write_to_log() -> io::Result<()> { let dir = tempdir()?; let log_path = dir.path().join("test.log"); let mut log = RotatingLog::with_max_size(&log_path, 1024).await?; let test_data = b"test log entry\n"; log.write_all(test_data).await?; log.flush().await?; let content = fs::read_to_string(&log_path)?; assert_eq!(content, String::from_utf8_lossy(test_data)); assert_eq!( log.current_size.load(Ordering::Relaxed), test_data.len() as u64 ); Ok(()) } #[tokio::test] async fn test_log_rotation() -> io::Result<()> { let dir = tempdir()?; let log_path = dir.path().join("test.log"); let max_size = 20; // Small size to trigger rotation let mut log = RotatingLog::with_max_size(&log_path, max_size).await?; // Write data until we trigger rotation let first_entry = b"first entry\n"; log.write_all(first_entry).await?; log.flush().await?; let second_entry = b"second entry\n"; log.write_all(second_entry).await?; log.flush().await?; // Give some time for rotation to complete tokio::time::sleep(tokio::time::Duration::from_millis(100)).await; // Check that both current and old log files exist assert!(log_path.exists()); assert!(log_path.with_extension("old").exists()); // Verify old file contains our first entry let old_content = fs::read_to_string(log_path.with_extension("old"))?; assert_eq!(old_content, String::from_utf8_lossy(first_entry)); // Verify new file contains our second entry let new_content = fs::read_to_string(&log_path)?; assert_eq!(new_content, String::from_utf8_lossy(second_entry)); Ok(()) } #[tokio::test] async fn test_oversized_write() -> io::Result<()> { let dir = tempdir()?; let log_path = dir.path().join("test.log"); let max_size = 10; // Small size let mut log = RotatingLog::with_max_size(&log_path, max_size).await?; // Write data much larger than max_size let large_entry = b"this is a very large log entry that exceeds the maximum size\n"; log.write_all(large_entry).await?; log.flush().await?; // Give some time for rotation to complete tokio::time::sleep(tokio::time::Duration::from_millis(100)).await; // Verify the content was written (even though it exceeds max_size) assert!(log_path.exists()); let content = fs::read_to_string(&log_path)?; assert_eq!(content, String::from_utf8_lossy(large_entry)); Ok(()) } #[tokio::test] async fn test_sync_writer() -> io::Result<()> { let dir = tempdir()?; let log_path = dir.path().join("test.log"); let log = RotatingLog::with_max_size(&log_path, 1024).await?; let mut sync_writer = log.get_sync_writer(); let test_data = b"sync writer test\n"; sync_writer.write_all(test_data)?; sync_writer.flush()?; // Give some time for async processing tokio::time::sleep(tokio::time::Duration::from_millis(100)).await; let content = fs::read_to_string(&log_path)?; assert_eq!(content, String::from_utf8_lossy(test_data)); Ok(()) } #[tokio::test] async fn test_multiple_rotations() -> io::Result<()> { let dir = tempdir()?; let log_path = dir.path().join("test.log"); let max_size = 20; let mut log = RotatingLog::with_max_size(&log_path, max_size).await?; // Perform multiple writes to trigger multiple rotations for i in 0..3 { let test_data = format!("rotation test {}\n", i).into_bytes(); log.write_all(&test_data).await?; log.flush().await?; // Give time for rotation tokio::time::sleep(tokio::time::Duration::from_millis(100)).await; } // Verify only one .old file exists (latest rotation) assert!(log_path.exists()); assert!(log_path.with_extension("old").exists()); Ok(()) } }