feat(server): implement comprehensive log rotation based on size and retention

- Configurable maximum log size and retention period from server configuration
- Optimized single directory traversal for collecting file information
- Eliminated redundant metadata() calls for improved performance
- Enhanced error handling with detailed logging for all failure cases
- Merged directory traversals into a single operation to improve performance

Fixes apache#46, detail changes can be found at apache#2452.

Author: Svecco

Cargo.lock

@@ -661,6 +661,7 @@ rle-decode-fast: 1.0.3, "Apache-2.0 OR MIT",
 rmcp: 0.11.0, "MIT",
 rmcp-macros: 0.11.0, "MIT",
 roaring: 0.10.12, "Apache-2.0 OR MIT",
+rolling-file: 0.2.0, "Apache-2.0 OR MIT",
 route-recognizer: 0.3.1, "MIT",
 rsa: 0.9.9, "Apache-2.0 OR MIT",
 rust-ini: 0.21.3, "MIT",

DEPENDENCIES.md

@@ -100,6 +100,7 @@ rand = { workspace = true }
 reqwest = { workspace = true, features = ["rustls-tls-no-provider"] }
 ring = "0.17.14"
 ringbuffer = "0.16.0"
+rolling-file = "0.2.0"
 rustls = { workspace = true }
 rustls-pemfile = "2.2.0"
 send_wrapper = "0.6.0"

core/server/Cargo.toml

@@ -30,9 +30,11 @@ use opentelemetry_sdk::Resource;
 use opentelemetry_sdk::logs::log_processor_with_async_runtime;
 use opentelemetry_sdk::propagation::TraceContextPropagator;
 use opentelemetry_sdk::trace::span_processor_with_async_runtime;
+use rolling_file::{BasicRollingFileAppender, RollingConditionBasic};
 use std::io::{self, Write};
 use std::path::PathBuf;
 use std::sync::{Arc, Mutex};
+use std::time::Duration;
 use tracing::{info, trace};
 use tracing_appender::non_blocking::WorkerGuard;
 use tracing_opentelemetry::OpenTelemetryLayer;
@@ -253,8 +255,21 @@ impl Logging {
             let base_directory = PathBuf::from(base_directory);
             let logs_subdirectory = PathBuf::from(config.path.clone());
             let logs_path = base_directory.join(logs_subdirectory.clone());
-            let file_appender =
-                tracing_appender::rolling::hourly(logs_path.clone(), IGGY_LOG_FILE_PREFIX);
+            let max_files = Self::calculate_max_files(
+                config.max_size.as_bytes_u64(),
+                config.max_size.as_bytes_u64(),
+            );
+
+            let condition = RollingConditionBasic::new()
+                .max_size(config.max_size.as_bytes_u64())
+                .hourly();
+
+            let file_appender = BasicRollingFileAppender::new(
+                logs_path.join(IGGY_LOG_FILE_PREFIX),
+                condition,
+                max_files,
+            )
+            .map_err(|_| LogError::FileReloadFailure)?;
             let (mut non_blocking_file, file_guard) = tracing_appender::non_blocking(file_appender);
 
             self.dump_to_file(&mut non_blocking_file);
@@ -284,9 +299,11 @@ impl Logging {
             self.init_telemetry(telemetry_config)?;
         }
 
+        self._install_log_rotation_handler(config, logs_path.as_ref());
+
         if let Some(logs_path) = logs_path {
             info!(
-                "Logging initialized, logs will be stored at: {logs_path:?}. Logs will be rotated hourly. Log filter: {log_filter}."
+                "Logging initialized, logs will be stored at: {logs_path:?}. Logs will be rotated based on size. Log filter: {log_filter}."
             );
         } else {
             info!("Logging initialized (file output disabled). Log filter: {log_filter}.");
@@ -397,10 +414,6 @@ impl Logging {
         Format::default().with_thread_names(true)
     }
 
-    fn _install_log_rotation_handler(&self) {
-        todo!("Implement log rotation handler based on size and retention time");
-    }
-
     fn print_build_info() {
         if option_env!("IGGY_CI_BUILD") == Some("true") {
             let hash = option_env!("VERGEN_GIT_SHA").unwrap_or("unknown");
@@ -417,6 +430,160 @@ impl Logging {
             )
         }
     }
+
+    fn calculate_max_files(max_total_size_bytes: u64, max_file_size_bytes: u64) -> usize {
+        if max_file_size_bytes == 0 {
+            return 10;
+        }
+
+        let max_files = max_total_size_bytes / max_file_size_bytes;
+        max_files.clamp(1, 1000) as usize
+    }
+
+    // Use a mutex lock to ensure log rotation operations do not produce race conditions.
+    fn _install_log_rotation_handler(&self, config: &LoggingConfig, logs_path: Option<&PathBuf>) {
+        if let Some(logs_path) = logs_path {
+            let path = logs_path.to_path_buf();
+            let max_size = config.max_size.as_bytes_u64();
+            let retention = config.retention.get_duration();
+            let rotation_mutex = Arc::new(Mutex::new(()));
+            let rotation_mutex_clone = Arc::clone(&rotation_mutex);
+            std::thread::spawn(move || {
+                loop {
+                    std::thread::sleep(Duration::from_secs(3600));
+                    match rotation_mutex_clone.lock() {
+                        Ok(_guard) => {
+                            Self::cleanup_log_files(&path, retention, max_size);
+                        }
+                        Err(e) => {
+                            tracing::warn!("Failed to acquire log rotation lock: {:?}", e);
+                        }
+                    }
+                }
+            });
+        }
+    }
+
+    fn cleanup_log_files(logs_path: &PathBuf, retention: Duration, max_size_bytes: u64) {
+        use std::fs;
+        use std::time::{SystemTime, UNIX_EPOCH};
+
+        tracing::debug!(
+            "Starting log cleanup for directory: {:?}, retention: {:?}, max_size: {} bytes",
+            logs_path,
+            retention,
+            max_size_bytes
+        );
+
+        let entries = match fs::read_dir(logs_path) {
+            Ok(entries) => entries,
+            Err(e) => {
+                tracing::warn!("Failed to read log directory {:?}: {}", logs_path, e);
+                return;
+            }
+        };
+
+        let mut file_entries = Vec::new();
+
+        for entry in entries.flatten() {
+            let metadata = match entry.metadata() {
+                Ok(metadata) => metadata,
+                Err(e) => {
+                    tracing::warn!("Failed to get metadata for {:?}: {}", entry.path(), e);
+                    continue;
+                }
+            };
+
+            if !metadata.is_file() {
+                continue;
+            }
+
+            let modified = match metadata.modified() {
+                Ok(modified) => modified,
+                Err(e) => {
+                    tracing::warn!(
+                        "Failed to get modification time for {:?}: {}",
+                        entry.path(),
+                        e
+                    );
+                    continue;
+                }
+            };
+
+            let elapsed = match modified.duration_since(UNIX_EPOCH) {
+                Ok(elapsed) => elapsed,
+                Err(e) => {
+                    tracing::warn!(
+                        "Failed to calculate elapsed time for {:?}: {}",
+                        entry.path(),
+                        e
+                    );
+                    continue;
+                }
+            };
+
+            let file_size = metadata.len();
+            file_entries.push((entry, modified, elapsed, file_size));
+        }
+
+        tracing::debug!(
+            "Processed {} log files from directory: {:?}",
+            file_entries.len(),
+            logs_path
+        );
+
+        let mut removed_files_count = 0;
+
+        if !retention.is_zero() {
+            let cutoff = match SystemTime::now().duration_since(UNIX_EPOCH) {
+                Ok(now) => now - retention,
+                Err(e) => {
+                    tracing::warn!("Failed to get current time: {}", e);
+                    return;
+                }
+            };
+
+            for (entry, _, elapsed, _) in &file_entries {
+                if *elapsed < cutoff {
+                    if let Err(e) = fs::remove_file(entry.path()) {
+                        tracing::warn!("Failed to remove old log file {:?}: {}", entry.path(), e);
+                    } else {
+                        tracing::debug!("Removed old log file: {:?}", entry.path());
+                        removed_files_count += 1;
+                    }
+                }
+            }
+        }
+
+        if max_size_bytes > 0 {
+            let total_size: u64 = file_entries.iter().map(|(_, _, _, size)| *size).sum();
+
+            if total_size > max_size_bytes {
+                file_entries.sort_by_key(|(_, modified, _, _)| *modified);
+
+                let mut current_size = total_size;
+                for (entry, _, _, file_size) in file_entries {
+                    if current_size <= max_size_bytes {
+                        break;
+                    }
+
+                    if let Err(e) = fs::remove_file(entry.path()) {
+                        tracing::warn!("Failed to remove log file {:?}: {}", entry.path(), e);
+                    } else {
+                        tracing::debug!("Removed log file to control size: {:?}", entry.path());
+                        current_size = current_size.saturating_sub(file_size);
+                        removed_files_count += 1;
+                    }
+                }
+            }
+        }
+
+        tracing::info!(
+            "Completed log cleanup for directory: {:?}. Removed {} files.",
+            logs_path,
+            removed_files_count
+        );
+    }
 }
 
 impl Default for Logging {