Add configuration option to set the number of event logs retained on disk

app/buck2_client_ctx/src/streaming.rs

@@ -300,6 +300,10 @@ fn get_event_log_subscriber<T: StreamingCommand>(
         T::COMMAND_NAME.to_owned(),
         ctx.start_time,
         log_size_counter_bytes,
+        ctx.immediate_config
+            .daemon_startup_config()
+            .map(|daemon_startup_config| daemon_startup_config.retained_event_logs)
+            .unwrap(),
     );
     Box::new(log)
 }

app/buck2_client_ctx/src/subscribers/event_log.rs

@@ -39,6 +39,7 @@ impl EventLog {
         command_name: String,
         start_time: SystemTime,
         log_size_counter_bytes: Option<Arc<AtomicU64>>,
+        retained_event_logs: usize,
     ) -> EventLog {
         Self {
             writer: WriteEventLog::new(
@@ -50,6 +51,7 @@ impl EventLog {
                 command_name,
                 start_time,
                 log_size_counter_bytes,
+                retained_event_logs,
             ),
         }
     }

app/buck2_common/src/init.rs

@@ -22,6 +22,8 @@ use serde::Serialize;
 use crate::legacy_configs::configs::LegacyBuckConfig;
 use crate::legacy_configs::key::BuckconfigKeyRef;
 
+const DEFAULT_RETAINED_EVENT_LOGS: usize = 12;
+
 /// Helper enum to categorize the kind of timeout we get from the startup config.
 #[derive(Clone, Debug)]
 pub enum Timeout {
@@ -437,6 +439,7 @@ pub struct DaemonStartupConfig {
     pub resource_control: ResourceControlConfig,
     pub log_download_method: LogDownloadMethod,
     pub health_check_config: HealthCheckConfig,
+    pub retained_event_logs: usize,
 }
 
 impl DaemonStartupConfig {
@@ -512,6 +515,13 @@ impl DaemonStartupConfig {
             resource_control: ResourceControlConfig::from_config(config)?,
             log_download_method,
             health_check_config: HealthCheckConfig::from_config(config)?,
+            retained_event_logs: config
+                .get(BuckconfigKeyRef {
+                    section: "buck2",
+                    property: "retained_event_logs",
+                })
+                .and_then(|s| s.parse::<usize>().ok())
+                .unwrap_or(DEFAULT_RETAINED_EVENT_LOGS),
         })
     }
 
@@ -539,6 +549,7 @@ impl DaemonStartupConfig {
                 LogDownloadMethod::None
             },
             health_check_config: HealthCheckConfig::default(),
+            retained_event_logs: DEFAULT_RETAINED_EVENT_LOGS,
         }
     }
 }

app/buck2_event_log/src/file_names.rs

@@ -44,11 +44,9 @@ pub(crate) fn get_logfile_name(
     ))
 }
 
-pub(crate) async fn remove_old_logs(logdir: &AbsNormPath) {
-    const N_LOGS_RETAINED: usize = 12;
-
+pub(crate) async fn remove_old_logs(logdir: &AbsNormPath, retained_event_logs: usize) {
     if let Ok(logfiles) = get_files_in_log_dir(logdir) {
-        futures::stream::iter(logfiles.into_iter().rev().skip(N_LOGS_RETAINED - 1))
+        futures::stream::iter(logfiles.into_iter().rev().skip(retained_event_logs - 1))
             .then(|file| async move {
                 // The oldest logs might be open from another concurrent build, so suppress error.
                 tokio::fs::remove_file(file).await.ok()
@@ -74,7 +72,10 @@ pub fn get_local_logs(logdir: &AbsNormPath) -> buck2_error::Result<Vec<EventLogP
 }
 
 fn sort_logs(dir: fs_util::ReadDir) -> Vec<AbsNormPathBuf> {
-    let mut logfiles = dir.filter_map(Result::ok).collect::<Vec<_>>();
+    let mut logfiles = dir
+        .filter_map(Result::ok)
+        .filter(|entry| entry.file_type().ok().map_or(false, |ft| ft.is_file()))
+        .collect::<Vec<_>>();
     logfiles.sort_by_cached_key(|file| {
         // Return Unix epoch if unable to get creation time.
         if let Ok(metadata) = file.metadata() {
@@ -128,3 +129,69 @@ pub fn retrieve_all_logs(paths: &InvocationPaths) -> buck2_error::Result<Vec<Eve
     let log_dir = paths.log_dir();
     get_local_logs(&log_dir)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use buck2_core::fs::fs_util;
+    use buck2_core::fs::paths::abs_norm_path::AbsNormPath;
+    use buck2_core::fs::paths::abs_path::AbsPath;
+    use std::fs::File;
+    use std::time::{Duration, SystemTime};
+
+    #[tokio::test]
+    async fn test_remove_old_logs_with_mix_of_files_and_folders() -> buck2_error::Result<()> {
+        let logdir = tempfile::tempdir()?;
+        let logdir_path = AbsPath::new(logdir.path())?;
+        let logdir_norm = AbsNormPath::new(logdir_path)?;
+
+        // Create 5 subfolders in logdir, each with a file inside
+        let mut subdirs = Vec::new();
+        for i in 0..5 {
+            let subdir_path = logdir.path().join(format!("subdir{}", i));
+            let subdir = AbsPath::new(&subdir_path)?;
+            fs_util::create_dir_all(&subdir)?;
+            let inside_file = subdir.join("inside.txt");
+            fs_util::write(&inside_file, format!("content in subdir{}", i))?;
+            subdirs.push((subdir.to_owned(), inside_file));
+        }
+
+        let base_time = SystemTime::now().checked_sub(Duration::from_secs(100)).unwrap();
+
+        // Create 5 log files directly in logdir with incrementing modification times
+        let mut log_paths = Vec::new();
+        for i in 0..5 {
+            let log_path = logdir_path.join(format!("buck-log-{}.zst", i));
+            fs_util::write(&log_path, format!("log content {}", i))?;
+
+            let file = File::open(&log_path)?;
+            let mod_time = base_time + Duration::from_secs((i as u64) * 10);
+            let times = std::fs::FileTimes::new().set_modified(mod_time);
+            file.set_times(times)?;
+
+            log_paths.push(log_path.clone());
+        }
+
+        // Call the function to keep 3 logs (should delete 3 oldest, leave 2 newest)
+        remove_old_logs(&logdir_norm, 3).await;
+
+        // Check that the 3 oldest logs are removed (indices 0,1,2 - earliest created)
+        for path in &log_paths[0..3] {
+            assert!(!path.exists(), "{} should be removed", path.display());
+        }
+
+        // Check that the 2 newest logs remain (indices 3,4 - latest created)
+        for path in &log_paths[3..5] {
+            assert!(path.exists(), "{} should remain", path.display());
+        }
+
+        // Check that all subfolders remain
+        for (subdir, inside_file) in subdirs {
+            assert!(subdir.exists(), "{} should remain", subdir.display());
+            // Ensure the file inside subdirectory is still there
+            assert!(inside_file.exists(), "{} should remain", inside_file.display());
+        }
+
+        Ok(())
+    }
+}

app/buck2_event_log/src/write.rs

@@ -63,6 +63,7 @@ pub struct WriteEventLog {
     /// Allocation cache. Must be cleaned before use.
     buf: Vec<u8>,
     log_size_counter_bytes: Option<Arc<AtomicU64>>,
+    retained_event_logs: usize,
 }
 
 impl WriteEventLog {
@@ -75,6 +76,7 @@ impl WriteEventLog {
         command_name: String,
         start_time: SystemTime,
         log_size_counter_bytes: Option<Arc<AtomicU64>>,
+        retained_event_logs: usize,
     ) -> Self {
         Self {
             state: LogWriterState::Unopened {
@@ -88,6 +90,7 @@ impl WriteEventLog {
             start_time,
             buf: Vec::new(),
             log_size_counter_bytes,
+            retained_event_logs,
         }
     }
 
@@ -165,7 +168,7 @@ impl WriteEventLog {
             .with_buck_error_context(|| {
                 format!("Error creating event log directory: `{logdir}`")
             })?;
-        remove_old_logs(logdir).await;
+        remove_old_logs(logdir, self.retained_event_logs).await;
 
         let encoding = Encoding::PROTO_ZSTD;
         let file_name = &get_logfile_name(event, encoding, &self.command_name)?;
@@ -479,6 +482,7 @@ mod tests {
                 buf: Vec::new(),
                 log_size_counter_bytes: None,
                 start_time: SystemTime::UNIX_EPOCH,
+                retained_event_logs: 5,
             })
         }
     }