feat: Add cache auto-purge and --old arg to cache clean command (#76)

* Add cache auto-purge, add --old arg to cache clean command

* Apply various suggestions

* Apply more suggestions

* Add `from_bool` method to DeleteFilter enum

* Rework `auto_purge` method

Author: Techassi

extra/completions/_stackablectl

@@ -8,7 +8,10 @@ pub const DEFAULT_NAMESPACE: &str = "default";
 
 pub const DEFAULT_LOCAL_CLUSTER_NAME: &str = "stackable-data-platform";
 
+pub const DEFAULT_AUTO_PURGE_INTERVAL: Duration = Duration::from_secs(60 * 15); // 15 minutes
 pub const DEFAULT_CACHE_MAX_AGE: Duration = Duration::from_secs(60 * 60); // One hour
+pub const CACHE_LAST_AUTO_PURGE_FILEPATH: &str = ".cache-last-purge";
+pub const CACHE_PROTECTED_FILES: &[&str] = &[".cache-last-purge"];
 
 pub const HELM_REPO_NAME_STABLE: &str = "stackable-stable";
 pub const HELM_REPO_NAME_TEST: &str = "stackable-test";

extra/completions/stackablectl.bash

@@ -1,14 +1,20 @@
 use std::{
+    ffi::OsString,
+    num::ParseIntError,
     path::{Path, PathBuf},
-    time::{Duration, SystemTime, SystemTimeError},
+    time::{Duration, SystemTime, SystemTimeError, UNIX_EPOCH},
 };
 
 use sha2::{Digest, Sha256};
 use snafu::{ResultExt, Snafu};
 use tokio::{fs, io};
+use tracing::debug;
 use url::Url;
 
-use crate::constants::DEFAULT_CACHE_MAX_AGE;
+use crate::constants::{
+    CACHE_LAST_AUTO_PURGE_FILEPATH, CACHE_PROTECTED_FILES, DEFAULT_AUTO_PURGE_INTERVAL,
+    DEFAULT_CACHE_MAX_AGE,
+};
 
 pub type Result<T, E = Error> = std::result::Result<T, E>;
 
@@ -20,12 +26,16 @@ pub enum Error {
     #[snafu(display("system time error"))]
     SystemTimeError { source: SystemTimeError },
 
+    #[snafu(display("failed to parse string as integer"))]
+    ParseIntError { source: ParseIntError },
+
     #[snafu(display("tried to write file with disabled cache"))]
     WriteDisabled,
 }
 
 #[derive(Debug)]
 pub struct Cache {
+    pub(crate) auto_purge_interval: Duration,
     pub(crate) backend: CacheBackend,
     pub(crate) max_age: Duration,
 }
@@ -97,18 +107,16 @@ impl Cache {
                 let mut entries = fs::read_dir(base_path).await.context(CacheIoSnafu)?;
 
                 while let Some(entry) = entries.next_entry().await.context(CacheIoSnafu)? {
-                    files.push((
-                        entry.path(),
-                        entry
-                            .metadata()
-                            .await
-                            .context(CacheIoSnafu)?
-                            .modified()
-                            .context(CacheIoSnafu)?,
-                    ))
+                    let metadata = entry.metadata().await.context(CacheIoSnafu)?;
+
+                    // Skip protected files
+                    if is_protected_file(entry.file_name()) {
+                        continue;
+                    }
+
+                    files.push((entry.path(), metadata.modified().context(CacheIoSnafu)?))
                 }
 
-                files.sort();
                 Ok(files)
             }
             CacheBackend::Disabled => Ok(vec![]),
@@ -117,18 +125,80 @@ impl Cache {
 
     /// Removes all cached files by deleting the base cache folder and then
     /// recreating it.
-    pub async fn purge(&self) -> Result<()> {
+    pub async fn purge(&self, delete_filter: DeleteFilter) -> Result<()> {
         match &self.backend {
             CacheBackend::Disk { base_path } => {
-                fs::remove_dir_all(base_path).await.context(CacheIoSnafu)?;
-                fs::create_dir_all(base_path).await.context(CacheIoSnafu)
+                let mut entries = fs::read_dir(base_path).await.context(CacheIoSnafu)?;
+
+                while let Some(entry) = entries.next_entry().await.context(CacheIoSnafu)? {
+                    let metadata = entry.metadata().await.context(CacheIoSnafu)?;
+
+                    let should_delete_file = match delete_filter {
+                        // Skip protected files
+                        _ if is_protected_file(entry.file_name()) => false,
+
+                        // Without --old / --outdated
+                        DeleteFilter::All => true,
+                        // with --old/--outdated
+                        DeleteFilter::OnlyExpired => {
+                            metadata
+                                .modified()
+                                .context(CacheIoSnafu)?
+                                .elapsed()
+                                .context(SystemTimeSnafu)?
+                                > self.max_age
+                        }
+                    };
+
+                    if should_delete_file {
+                        fs::remove_file(entry.path()).await.context(CacheIoSnafu)?;
+                    }
+                }
+
+                Ok(())
             }
-            CacheBackend::Disabled => todo!(),
+            CacheBackend::Disabled => Ok(()),
         }
     }
 
-    fn new(backend: CacheBackend, max_age: Duration) -> Self {
-        Self { backend, max_age }
+    pub async fn auto_purge(&self) -> Result<()> {
+        match &self.backend {
+            CacheBackend::Disk { base_path } => {
+                let cache_auto_purge_filepath = base_path.join(CACHE_LAST_AUTO_PURGE_FILEPATH);
+
+                // Read and covert timestamp
+                let last_purged_at = match fs::read_to_string(&cache_auto_purge_filepath).await {
+                    Ok(timestamp) => Some(
+                        UNIX_EPOCH + Duration::from_secs(timestamp.parse().context(ParseIntSnafu)?),
+                    ),
+                    Err(err) if err.kind() == std::io::ErrorKind::NotFound => None,
+                    Err(err) => return Err(err).context(CacheIoSnafu),
+                };
+
+                // If the auto purge interval elapsed, run purge and write
+                // back the new timestamp
+                if last_purged_at
+                    .and_then(|ts| ts.elapsed().ok())
+                    .map_or(true, |elapsed| elapsed >= self.auto_purge_interval)
+                {
+                    debug!("Auto-purging outdated cache files");
+
+                    self.purge(DeleteFilter::OnlyExpired).await?;
+                    write_cache_auto_purge_file(&cache_auto_purge_filepath).await?;
+                }
+
+                Ok(())
+            }
+            CacheBackend::Disabled => Ok(()),
+        }
+    }
+
+    fn new(backend: CacheBackend, max_age: Duration, auto_purge_interval: Duration) -> Self {
+        Self {
+            auto_purge_interval,
+            backend,
+            max_age,
+        }
     }
 
     async fn read(file_path: PathBuf) -> Result<String> {
@@ -163,13 +233,15 @@ pub enum CacheStatus<T> {
 
 #[derive(Debug, Clone)]
 pub struct CacheSettings {
+    pub auto_purge_interval: Duration,
     pub backend: CacheBackend,
     pub max_age: Duration,
 }
 
 impl From<CacheBackend> for CacheSettings {
     fn from(backend: CacheBackend) -> Self {
         Self {
+            auto_purge_interval: DEFAULT_AUTO_PURGE_INTERVAL,
             max_age: DEFAULT_CACHE_MAX_AGE,
             backend,
         }
@@ -195,9 +267,18 @@ impl CacheSettings {
         match &self.backend {
             CacheBackend::Disk { base_path } => {
                 fs::create_dir_all(base_path).await.context(CacheIoSnafu)?;
-                Ok(Cache::new(self.backend, self.max_age))
+
+                Ok(Cache::new(
+                    self.backend,
+                    self.max_age,
+                    self.auto_purge_interval,
+                ))
             }
-            CacheBackend::Disabled => Ok(Cache::new(self.backend, self.max_age)),
+            CacheBackend::Disabled => Ok(Cache::new(
+                self.backend,
+                self.max_age,
+                self.auto_purge_interval,
+            )),
         }
     }
 }
@@ -207,3 +288,28 @@ pub enum CacheBackend {
     Disk { base_path: PathBuf },
     Disabled,
 }
+
+pub enum DeleteFilter {
+    All,
+    OnlyExpired,
+}
+
+async fn write_cache_auto_purge_file(path: &Path) -> Result<()> {
+    fs::write(
+        path,
+        SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .context(SystemTimeSnafu)?
+            .as_secs()
+            .to_string()
+            .as_bytes(),
+    )
+    .await
+    .context(CacheIoSnafu)
+}
+
+fn is_protected_file(filename: OsString) -> bool {
+    // Non-UTF-8 filenames can't possibly be on the protected list
+    let Some(filename) = filename.to_str() else { return false; };
+    CACHE_PROTECTED_FILES.contains(&filename)
+}

extra/completions/stackablectl.fish

@@ -53,6 +53,11 @@ impl FileTransferClient {
         Ok(Self { client, cache })
     }
 
+    pub fn new_with(cache: Cache) -> Self {
+        let client = reqwest::Client::new();
+        Self { client, cache }
+    }
+
     /// Retrieves data from `path_or_url` which can either be a [`PathBuf`]
     /// or a [`Url`]. The `processor` defines how the data is processed, for
     /// example as plain text data, YAML content or even templated.

rust/stackable-cockpit/src/constants.rs

@@ -3,9 +3,10 @@ use std::time::Duration;
 use clap::{Args, Subcommand};
 use comfy_table::{presets::UTF8_FULL, ColumnConstraint, Table, Width};
 use snafu::{ResultExt, Snafu};
-use stackable_cockpit::xfer::cache;
+use stackable_cockpit::xfer::cache::{self, Cache, DeleteFilter};
+use tracing::{info, instrument};
 
-use crate::cli::{CacheSettingsError, Cli};
+use crate::cli::CacheSettingsError;
 
 #[derive(Debug, Args)]
 pub struct CacheArgs {
@@ -21,7 +22,14 @@ pub enum CacheCommands {
 
     /// Clean cached files
     #[command(aliases(["rm", "purge"]))]
-    Clean,
+    Clean(CacheCleanArgs),
+}
+
+#[derive(Debug, Args)]
+pub struct CacheCleanArgs {
+    /// Only remove outdated files in the cache
+    #[arg(long = "old", visible_aliases(["outdated"]))]
+    only_remove_old_files: bool,
 }
 
 #[derive(Debug, Snafu)]
@@ -34,17 +42,17 @@ pub enum CacheCmdError {
 }
 
 impl CacheArgs {
-    pub async fn run(&self, common_args: &Cli) -> Result<String, CacheCmdError> {
-        match self.subcommand {
-            CacheCommands::List => list_cmd(common_args).await,
-            CacheCommands::Clean => clean_cmd(common_args).await,
+    pub async fn run(&self, cache: Cache) -> Result<String, CacheCmdError> {
+        match &self.subcommand {
+            CacheCommands::List => list_cmd(cache).await,
+            CacheCommands::Clean(args) => clean_cmd(args, cache).await,
         }
     }
 }
 
-async fn list_cmd(common_args: &Cli) -> Result<String, CacheCmdError> {
-    let cache_settings = common_args.cache_settings().context(CacheSettingsSnafu)?;
-    let cache = cache_settings.try_into_cache().await.context(CacheSnafu)?;
+#[instrument(skip_all)]
+async fn list_cmd(cache: Cache) -> Result<String, CacheCmdError> {
+    info!("Listing cached files");
 
     let files = cache.list().await.context(CacheSnafu)?;
 
@@ -72,10 +80,16 @@ async fn list_cmd(common_args: &Cli) -> Result<String, CacheCmdError> {
     Ok(table.to_string())
 }
 
-async fn clean_cmd(common_args: &Cli) -> Result<String, CacheCmdError> {
-    let cache_settings = common_args.cache_settings().context(CacheSettingsSnafu)?;
-    let cache = cache_settings.try_into_cache().await.context(CacheSnafu)?;
+#[instrument(skip_all)]
+async fn clean_cmd(args: &CacheCleanArgs, cache: Cache) -> Result<String, CacheCmdError> {
+    info!("Cleaning cached files");
+
+    let delete_filter = if args.only_remove_old_files {
+        DeleteFilter::OnlyExpired
+    } else {
+        DeleteFilter::All
+    };
 
-    cache.purge().await.context(CacheSnafu)?;
+    cache.purge(delete_filter).await.context(CacheSnafu)?;
     Ok("Cleaned cached files".into())
 }

rust/stackable-cockpit/src/xfer/cache.rs

@@ -13,11 +13,11 @@ use stackable_cockpit::{
         stack::{StackError, StackList},
     },
     utils::path::PathOrUrlParseError,
-    xfer::{FileTransferClient, FileTransferError},
+    xfer::{cache::Cache, FileTransferClient, FileTransferError},
 };
 use tracing::{debug, info, instrument};
 
-use crate::cli::{CacheSettingsError, Cli, CommonClusterArgs, CommonClusterArgsError, OutputType};
+use crate::cli::{Cli, CommonClusterArgs, CommonClusterArgsError, OutputType};
 
 #[derive(Debug, Args)]
 pub struct DemoArgs {
@@ -114,9 +114,6 @@ pub enum DemoCmdError {
     #[snafu(display("path/url parse error"))]
     PathOrUrlParseError { source: PathOrUrlParseError },
 
-    #[snafu(display("cache settings resolution error"), context(false))]
-    CacheSettingsError { source: CacheSettingsError },
-
     #[snafu(display("cluster argument error"))]
     CommonClusterArgsError { source: CommonClusterArgsError },
 
@@ -126,12 +123,10 @@ pub enum DemoCmdError {
 
 impl DemoArgs {
     #[instrument]
-    pub async fn run(&self, common_args: &Cli) -> Result<String, DemoCmdError> {
+    pub async fn run(&self, common_args: &Cli, cache: Cache) -> Result<String, DemoCmdError> {
         debug!("Handle demo args");
 
-        let transfer_client = FileTransferClient::new(common_args.cache_settings()?)
-            .await
-            .context(TransferSnafu)?;
+        let transfer_client = FileTransferClient::new_with(cache);
 
         // Build demo list based on the (default) remote demo file, and additional files provided by the
         // STACKABLE_DEMO_FILES env variable or the --demo-files CLI argument.