Merge pull request #54 from silicon-heaven/add-log-retention

Add log retention

Author: j4r0u53k

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "historyprovider"
-version = "2.2.1"
+version = "2.3.0"
 edition = "2024"
 
 [[bin]]

Cargo.toml

@@ -5,6 +5,8 @@ use tokio::io;
 use std::collections::HashMap;
 use std::path::{Path, PathBuf};
 
+use crate::util::msec_to_log2_filename;
+
 #[derive(Debug, Clone)]
 pub(crate) struct LogFile {
     pub(crate) name: PathBuf,
@@ -40,11 +42,11 @@ pub(crate) async fn collect_log2_files(dir: impl AsRef<Path>) -> io::Result<Vec<
 }
 
 /// Prune `.log2` files while keeping the newest one per directory
-pub(crate) async fn cleanup_log2_files(dir: impl AsRef<Path>, size_limit: u64) -> io::Result<()> {
+pub(crate) async fn cleanup_log2_files(dir: impl AsRef<Path>, size_limit: u64, days_to_keep: i64) -> io::Result<()> {
     let files = collect_log2_files(dir.as_ref()).await?;
     let mut files_size: u64 = files.iter().map(|f| f.size).sum();
 
-    info!("log2 files size: {files_size}, size limit: {size_limit}");
+    info!("log2 files size: {files_size}, size limit: {size_limit}, days_to_keep: {days_to_keep}");
     if files_size < size_limit {
         info!("Size limit not hit, nothing to cleanup");
         return Ok(());
@@ -58,11 +60,15 @@ pub(crate) async fn cleanup_log2_files(dir: impl AsRef<Path>, size_limit: u64) -
 
     let mut deletable_files = Vec::new();
 
+    // This file doesn't have to exist, I'm only constructing the filename for log retention.
+    let oldest_file_to_keep = PathBuf::from(msec_to_log2_filename(shvproto::DateTime::now().add_days(-days_to_keep).epoch_msec()));
+    info!("keeping files younger than {filename}", filename = oldest_file_to_keep.to_string_lossy());
+
     for (_dir, mut group) in grouped {
         // Sort descending (newest first)
         group.sort_by(|a, b| b.name.cmp(&a.name));
-        // Keep the newest file
-        deletable_files.extend(group.into_iter().skip(1));
+        // Keep the newest file, and keep files newer than oldest_file_to_delete
+        deletable_files.extend(group.into_iter().skip(1).filter(|log_file| log_file.name < oldest_file_to_keep));
     }
 
     // Sort deletable files (oldest first for deletion)

src/cleanup.rs

@@ -118,7 +118,7 @@ pub(crate) async fn dirtylog_task(
                     Ok(file) => file,
                     Err(err) => {
                         if err.kind() == std::io::ErrorKind::NotFound {
-                            info!("Trim dirty log done, no dirty log file for the site");
+                            info!("Trim dirty log done, no dirty log file for the site {site}");
                         } else {
                             error!("Cannot trim dirty log. Cannot open file {file_path}: {err}", file_path = dirty_log_path.to_string_lossy());
                         }

src/dirtylog.rs

@@ -27,8 +27,6 @@ mod util;
 pub mod typeinfo;
 pub mod alarm;
 
-const MAX_JOURNAL_DIR_SIZE_DEFAULT: usize = 30 * 1_000_000_000;
-
 fn default_journal_dir() -> String {
     "/tmp/hp-rs/shvjournal".into()
 }
@@ -40,6 +38,7 @@ pub struct HpConfig {
     max_sync_tasks: Option<usize>,
     max_journal_dir_size: Option<usize>,
     periodic_sync_interval: Option<u64>,
+    days_to_keep: Option<i64>,
 }
 
 impl HpConfig {

src/lib.rs

@@ -406,7 +406,6 @@ pub(crate) async fn sites_task(
 
                     let (sites_info, sub_hps) = match sites {
                         Ok(sites) => {
-                            eprintln!("sites = '{:?}'", sites);
                             if sites
                                 .get("_meta")
                                 .map(RpcValue::as_map)

src/sites.rs

@@ -24,8 +24,8 @@ use crate::journalentry::JournalEntry;
 use crate::journalrw::{GetLog2Params, GetLog2Since, JournalReaderLog2, JournalWriterLog2, Log2Reader};
 use crate::sites::{SitesData, SubHpInfo};
 use crate::tree::{FileType, LsFilesEntry, METH_READ};
-use crate::util::{get_files, is_log2_file, DedupReceiver};
-use crate::{State, MAX_JOURNAL_DIR_SIZE_DEFAULT};
+use crate::util::{get_files, is_log2_file, msec_to_log2_filename, DedupReceiver};
+use crate::{HpConfig, State};
 
 #[derive(Default)]
 pub(crate) struct SyncInfo {
@@ -542,13 +542,6 @@ async fn sync_site_legacy(
     const GETLOG_SINCE_DAYS_DEFAULT: i64 = 365;
     const RECORD_COUNT_LIMIT: i64 = 10000;
 
-    fn msec_to_log2_filename(msec: i64) -> String {
-        shvproto::DateTime::from_epoch_msec(msec)
-            .to_chrono_datetime()
-            .format("%Y-%m-%dT%H-%M-%S-%3f.log2")
-            .to_string()
-    }
-
     let (mut getlog_params, mut log_file_path, mut log_file_entries) = match newest_log {
         Some((newest_log_file, newest_log_entries)) => {
             let last_log_entry_msec = newest_log_entries.last().expect("The newest log is not empty").epoch_msec;
@@ -699,6 +692,11 @@ async fn sync_site_legacy(
 
 const MAX_SYNC_TASKS_DEFAULT: usize = 8;
 
+pub fn log_size_limit(config: &HpConfig) -> u64 {
+    const MAX_JOURNAL_DIR_SIZE_DEFAULT: usize = 30 * 1_000_000_000;
+    config.max_journal_dir_size.unwrap_or(MAX_JOURNAL_DIR_SIZE_DEFAULT) as u64
+}
+
 pub(crate) async fn sync_task(
     client_cmd_tx: ClientCommandSender,
     _client_evt_rx: ClientEventsReceiver,
@@ -723,7 +721,8 @@ pub(crate) async fn sync_task(
 
     // The download size limit should be lower than the max_journal_dir_size, because it doesn't
     // count in the files synced by getLog.
-    let max_journal_dir_size = app_state.config.max_journal_dir_size.unwrap_or(MAX_JOURNAL_DIR_SIZE_DEFAULT) as u64;
+    let max_journal_dir_size = log_size_limit(&app_state.config);
+    let days_to_keep = app_state.config.days_to_keep.unwrap_or(0);
 
     while let Some(cmd) = sync_cmd_rx.next().await {
         match cmd {
@@ -819,7 +818,7 @@ pub(crate) async fn sync_task(
                             panic!("Cannot send dirtylog Trim command for site {site}: {e}")
                         )
                     );
-                match cleanup_log2_files(&app_state.config.journal_dir, max_journal_dir_size).await {
+                match cleanup_log2_files(&app_state.config.journal_dir, max_journal_dir_size, days_to_keep).await {
                     Ok(_) => info!("Cleanup journal dir done"),
                     Err(err) => error!("Cleanup journal dir error: {err}"),
                 }
@@ -883,7 +882,7 @@ pub(crate) async fn sync_task(
                         .unwrap_or_else(|e|
                             panic!("Cannot send dirtylog Trim command for site {site}: {e}")
                         );
-                    match cleanup_log2_files(&app_state.config.journal_dir, max_journal_dir_size).await {
+                    match cleanup_log2_files(&app_state.config.journal_dir, max_journal_dir_size, days_to_keep).await {
                         Ok(_) => info!("Cleanup journal dir done"),
                         Err(err) => error!("Cleanup journal dir error: {err}"),
                     }
@@ -893,7 +892,7 @@ pub(crate) async fn sync_task(
             }
             SyncCommand::Cleanup => {
                 info!("Cleanup journal dir start");
-                match cleanup_log2_files(&app_state.config.journal_dir, max_journal_dir_size).await {
+                match cleanup_log2_files(&app_state.config.journal_dir, max_journal_dir_size, days_to_keep).await {
                     Ok(_) => info!("Cleanup journal dir done"),
                     Err(err) => error!("Cleanup journal dir error: {err}"),
                 }

src/sync.rs

@@ -26,7 +26,8 @@ use crate::getlog::getlog_handler;
 use crate::journalrw::{journal_entries_to_rpcvalue, GetLog2Params, Log2Header, Log2Reader};
 use crate::pushlog::pushlog_impl;
 use crate::sites::SubHpInfo;
-use crate::{AlarmWithTimestamp, HpConfig, State, MAX_JOURNAL_DIR_SIZE_DEFAULT};
+use crate::sync::log_size_limit;
+use crate::{AlarmWithTimestamp, HpConfig, State};
 
 // History site node methods
 const METH_GET_LOG: &str = "getLog";
@@ -333,10 +334,6 @@ impl From<LsFilesEntry> for RpcValue {
     }
 }
 
-fn log_size_limit(config: &HpConfig) -> i64 {
-    config.max_journal_dir_size.unwrap_or(MAX_JOURNAL_DIR_SIZE_DEFAULT) as i64
-}
-
 async fn total_log_size(config: &HpConfig) -> tokio::io::Result<i64> {
     collect_log2_files(&config.journal_dir)
         .await

src/tree.rs

@@ -134,6 +134,14 @@ pub(crate) fn dedup_channel<T: Eq + Hash + Clone>() -> (DedupSender<T>, DedupRec
     (DedupSender { pending: pending.clone(), sender }, DedupReceiver { pending, receiver })
 }
 
+pub(crate) fn msec_to_log2_filename(msec: i64) -> String {
+    shvproto::DateTime::from_epoch_msec(msec)
+        .to_chrono_datetime()
+        .format("%Y-%m-%dT%H-%M-%S-%3f.log2")
+        .to_string()
+}
+
+
 #[cfg(test)]
 pub mod testing {
     use crate::{State, dirtylog::DirtyLogCommand, sites::{SiteInfo, SitesData, SubHpInfo}, sync::SyncCommand, util::{DedupReceiver, dedup_channel}};
@@ -398,6 +406,7 @@ pub mod testing {
                 journal_dir: journal_dir.path().to_str().expect("path must work").to_string(),
                 max_sync_tasks: None,
                 max_journal_dir_size: None,
+                days_to_keep: None,
                 periodic_sync_interval: Some(3),
             },
             dirtylog_cmd_tx,