feat: add object storage usage view and manual log cleanup (#8724)

2026-04-05 09:10:48 -04:00
parent dd39c110a8
commit 02d0ee9198
10 changed files with 908 additions and 32 deletions
--- a/backend/.sqlx/query-b3c84299ed9872960c2287ef6e3a9dca85eeb2d22e7004021c987d08598b5585.json
+++ b/backend/.sqlx/query-b3c84299ed9872960c2287ef6e3a9dca85eeb2d22e7004021c987d08598b5585.json
@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT COUNT(*) FROM v2_job_completed\n         WHERE completed_at <= now() - ($1::bigint::text || ' s')::interval",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "count",
+        "type_info": "Int8"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Int8"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "b3c84299ed9872960c2287ef6e3a9dca85eeb2d22e7004021c987d08598b5585"
+}
--- a/backend/.sqlx/query-c5e72bcfd4d389fce281811d7a731711d89fb79f7b2adcfbdf33c070b3adaf7d.json
+++ b/backend/.sqlx/query-c5e72bcfd4d389fce281811d7a731711d89fb79f7b2adcfbdf33c070b3adaf7d.json
@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT COUNT(*) FROM log_file WHERE log_ts <= now() - ($1::bigint::text || ' s')::interval",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "count",
+        "type_info": "Int8"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Int8"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "c5e72bcfd4d389fce281811d7a731711d89fb79f7b2adcfbdf33c070b3adaf7d"
+}
--- a/backend/.sqlx/query-d21464c0e980b9d3a5c8f940c2abe43275843f33fb1998d9c8bd940ef127bfbf.json
+++ b/backend/.sqlx/query-d21464c0e980b9d3a5c8f940c2abe43275843f33fb1998d9c8bd940ef127bfbf.json
@@ -0,0 +1,29 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM log_file WHERE (file_path, hostname) IN (\n                 SELECT file_path, hostname FROM log_file\n                 WHERE log_ts <= now() - ($1::bigint::text || ' s')::interval\n                 LIMIT $2\n             ) RETURNING file_path, hostname",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "file_path",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "hostname",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Int8",
+        "Int8"
+      ]
+    },
+    "nullable": [
+      false,
+      false
+    ]
+  },
+  "hash": "d21464c0e980b9d3a5c8f940c2abe43275843f33fb1998d9c8bd940ef127bfbf"
+}
--- a/backend/src/monitor.rs
+++ b/backend/src/monitor.rs
@@ -1352,18 +1352,37 @@ async fn delete_log_files_from_disk_and_store(
    tmp_dir: &str,
    _s3_prefix: &str,
 ) {
+    // S3 bulk delete (batched via delete_stream — on S3 this uses the DeleteObjects
+    // API, up to 1000 objects per request).
    #[cfg(feature = "parquet")]
-    let os = windmill_object_store::get_object_store().await;
-    #[cfg(not(feature = "parquet"))]
-    let os: Option<()> = None;
-
-    let _should_del_from_store = MONITOR_LOGS_ON_OBJECT_STORE.read().await.clone();
+    {
+        let should_del_from_store = *MONITOR_LOGS_ON_OBJECT_STORE.read().await;
+        if should_del_from_store {
+            if let Some(os) = windmill_object_store::get_object_store().await {
+                let s3_paths: Vec<_> = paths_to_delete
+                    .iter()
+                    .map(|p| {
+                        windmill_object_store::object_store_reexports::Path::from(format!(
+                            "{}{}",
+                            _s3_prefix, p
+                        ))
+                    })
+                    .map(Ok)
+                    .collect();
+                let stream = futures::stream::iter(s3_paths).boxed();
+                let mut result = os.delete_stream(stream);
+                while let Some(r) = result.next().await {
+                    if let Err(e) = r {
+                        tracing::error!("Failed to delete from object store: {e}");
+                    }
+                }
+            }
+        }
+    }

+    // Disk delete in parallel.
    let delete_futures = FuturesUnordered::new();
-
    for path in paths_to_delete {
-        let _os2 = &os;
-
        delete_futures.push(async move {
            let disk_path = std::path::Path::new(tmp_dir).join(&path);
            if tokio::fs::metadata(&disk_path).await.is_ok() {
@@ -1372,31 +1391,10 @@ async fn delete_log_files_from_disk_and_store(
                        "Failed to delete from disk {}: {e}",
                        disk_path.to_string_lossy()
                    );
-                } else {
-                    tracing::debug!(
-                        "Succesfully deleted {} from disk",
-                        disk_path.to_string_lossy()
-                    );
-                }
-            }
-
-            #[cfg(feature = "parquet")]
-            if _should_del_from_store {
-                if let Some(os) = _os2 {
-                    let p = windmill_object_store::object_store_reexports::Path::from(format!(
-                        "{}{}",
-                        _s3_prefix, path
-                    ));
-                    if let Err(e) = os.delete(&p).await {
-                        tracing::error!("Failed to delete from object store {}: {e}", p.to_string())
-                    } else {
-                        tracing::debug!("Succesfully deleted {} from object store", p.to_string());
-                    }
                }
            }
        });
    }
-
    let _: Vec<_> = delete_futures.collect().await;
 }

--- a/backend/windmill-api-settings/src/lib.rs
+++ b/backend/windmill-api-settings/src/lib.rs
@@ -11,6 +11,8 @@ use std::{collections::HashMap, time::Duration};
 #[cfg(feature = "private")]
 mod ee;
 pub mod ee_oss;
+#[cfg(feature = "parquet")]
+mod log_cleanup;

 use windmill_api_auth::{require_devops_role, require_super_admin, ApiAuthed};
 use windmill_common::utils::HTTP_CLIENT_PERMISSIVE as HTTP_CLIENT;
@@ -129,7 +131,11 @@ pub fn global_service() -> Router {

    #[cfg(feature = "parquet")]
    {
-        return r.route("/test_object_storage_config", post(test_s3_bucket));
+        return r
+            .route("/test_object_storage_config", post(test_s3_bucket))
+            .route("/object_storage_usage", get(object_storage_usage))
+            .route("/run_log_cleanup", post(run_log_cleanup))
+            .route("/log_cleanup_status", get(log_cleanup_status));
    }

    #[cfg(not(feature = "parquet"))]
@@ -227,6 +233,99 @@ pub async fn test_s3_bucket(
    Ok("Tested blob storage successfully".to_string())
 }

+#[cfg(feature = "parquet")]
+#[derive(Serialize)]
+struct FolderUsage {
+    prefix: String,
+    size: u64,
+}
+
+#[cfg(feature = "parquet")]
+async fn object_storage_usage(
+    Extension(db): Extension<DB>,
+    authed: ApiAuthed,
+) -> error::JsonResult<Vec<FolderUsage>> {
+    use futures::StreamExt;
+
+    require_super_admin(&db, &authed.email).await?;
+
+    let client = windmill_object_store::get_object_store()
+        .await
+        .ok_or_else(|| error::Error::BadRequest("Object storage is not configured".to_string()))?;
+
+    // Overall cap on the whole listing operation — a bucket with millions of
+    // objects under a single prefix would otherwise hold an HTTP connection
+    // and memory for minutes. On timeout we return 504 and let the caller retry.
+    let work = async {
+        let list_result = client.list_with_delimiter(None).await?;
+
+        let mut usage: Vec<FolderUsage> = Vec::new();
+
+        let root_size: u64 = list_result.objects.iter().map(|o| o.size).sum();
+        if root_size > 0 {
+            usage.push(FolderUsage { prefix: "(root files)".to_string(), size: root_size });
+        }
+
+        for prefix in list_result.common_prefixes {
+            let prefix_str = prefix.to_string();
+            let mut total_size: u64 = 0;
+            let mut stream = client.list(Some(&prefix));
+            while let Some(item) = stream.next().await {
+                match item {
+                    Ok(meta) => total_size += meta.size,
+                    Err(e) => {
+                        tracing::warn!("Error listing objects under {prefix_str}: {e:#}");
+                        break;
+                    }
+                }
+            }
+            usage.push(FolderUsage { prefix: prefix_str, size: total_size });
+        }
+
+        usage.sort_by(|a, b| b.size.cmp(&a.size));
+        Ok::<_, windmill_object_store::object_store_reexports::ObjectStoreError>(usage)
+    };
+
+    let usage = tokio::time::timeout(std::time::Duration::from_secs(60), work)
+        .await
+        .map_err(|_| {
+            error::Error::internal_err(
+                "Listing object storage timed out after 60s — bucket may be too large".to_string(),
+            )
+        })?
+        .map_err(|e| error::Error::internal_err(format!("Failed to list objects: {e:#}")))?;
+
+    Ok(Json(usage))
+}
+
+#[cfg(feature = "parquet")]
+async fn run_log_cleanup(
+    Extension(db): Extension<DB>,
+    authed: ApiAuthed,
+) -> error::Result<axum::http::StatusCode> {
+    require_super_admin(&db, &authed.email).await?;
+
+    match log_cleanup::try_start().await {
+        Ok(()) => {
+            log_cleanup::spawn_cleanup(db.clone());
+            Ok(axum::http::StatusCode::ACCEPTED)
+        }
+        Err(_) => Err(error::Error::BadRequest(
+            "Log cleanup is already running".to_string(),
+        )),
+    }
+}
+
+#[cfg(feature = "parquet")]
+async fn log_cleanup_status(
+    Extension(db): Extension<DB>,
+    authed: ApiAuthed,
+) -> error::JsonResult<Option<log_cleanup::LogCleanupProgress>> {
+    require_super_admin(&db, &authed.email).await?;
+    let guard = log_cleanup::LOG_CLEANUP_STATUS.read().await;
+    Ok(Json(guard.clone()))
+}
+
 #[derive(Deserialize)]
 pub struct TestKey {
    pub license_key: String,
--- a/backend/windmill-api-settings/src/log_cleanup.rs
+++ b/backend/windmill-api-settings/src/log_cleanup.rs
@@ -0,0 +1,395 @@
+#![cfg(feature = "parquet")]
+/*
+ * Manual trigger for cleaning up expired log files from object storage.
+ *
+ * Mirrors the periodic cleanup done in backend/src/monitor.rs::delete_expired_items,
+ * but runs on demand from the UI with progress reporting and uses
+ * ObjectStore::delete_stream for batched S3 deletes (up to 1000 per request).
+ *
+ * Note: unlike the periodic cleanup (which only hits S3 when MONITOR_LOGS_ON_OBJECT_STORE
+ * is enabled), this manual path ALWAYS issues S3 deletes. That is intentional: operators
+ * who previously ran with the setting OFF may have orphan log files in their bucket and
+ * need a way to reclaim that space. Do not add a MONITOR_LOGS_ON_OBJECT_STORE guard here
+ * without first considering that use case.
+ */
+
+use std::sync::Arc;
+
+use chrono::{DateTime, Utc};
+use futures::stream::StreamExt;
+use serde::Serialize;
+use tokio::sync::RwLock;
+use uuid::Uuid;
+
+use windmill_common::error::{self};
+use windmill_common::tracing_init::{LOGS_SERVICE, TMP_WINDMILL_LOGS_SERVICE};
+use windmill_common::worker::WINDMILL_DIR;
+use windmill_common::{DB, JOB_RETENTION_SECS, SERVICE_LOG_RETENTION_SECS};
+
+use windmill_object_store::object_store_reexports::{ObjectStore, Path as ObjectPath};
+
+const SERVICE_LOG_BATCH: i64 = 2_000;
+const JOB_BATCH: i64 = 1_000;
+
+#[derive(Clone, Serialize)]
+pub struct LogCleanupProgress {
+    pub running: bool,
+    pub started_at: DateTime<Utc>,
+    pub finished_at: Option<DateTime<Utc>>,
+    pub total_service: u64,
+    pub processed_service: u64,
+    pub total_jobs: u64,
+    pub processed_jobs: u64,
+    pub s3_deleted: u64,
+    pub errors: u64,
+    pub last_error: Option<String>,
+}
+
+lazy_static::lazy_static! {
+    pub static ref LOG_CLEANUP_STATUS: Arc<RwLock<Option<LogCleanupProgress>>> =
+        Arc::new(RwLock::new(None));
+}
+
+/// Attempt to mark the cleanup as running. Returns `Err` if it is already running.
+pub async fn try_start() -> Result<(), &'static str> {
+    let mut guard = LOG_CLEANUP_STATUS.write().await;
+    if let Some(p) = guard.as_ref() {
+        if p.running {
+            return Err("Log cleanup is already running");
+        }
+    }
+    *guard = Some(LogCleanupProgress {
+        running: true,
+        started_at: Utc::now(),
+        finished_at: None,
+        total_service: 0,
+        processed_service: 0,
+        total_jobs: 0,
+        processed_jobs: 0,
+        s3_deleted: 0,
+        errors: 0,
+        last_error: None,
+    });
+    Ok(())
+}
+
+async fn update<F: FnOnce(&mut LogCleanupProgress)>(f: F) {
+    let mut guard = LOG_CLEANUP_STATUS.write().await;
+    if let Some(p) = guard.as_mut() {
+        f(p);
+    }
+}
+
+async fn finish() {
+    update(|p| {
+        p.running = false;
+        p.finished_at = Some(Utc::now());
+    })
+    .await;
+}
+
+async fn record_error(msg: String) {
+    tracing::error!("log cleanup: {msg}");
+    update(|p| {
+        p.errors = p.errors.saturating_add(1);
+        p.last_error = Some(msg);
+    })
+    .await;
+}
+
+/// Delete the given object paths from S3 in batches (uses ObjectStore::delete_stream
+/// which on S3 issues a single DeleteObjects request per 1000 paths).
+async fn s3_bulk_delete(
+    store: &Arc<dyn ObjectStore>,
+    paths: Vec<ObjectPath>,
+) -> (u64 /* deleted */, u64 /* errors */) {
+    let stream = futures::stream::iter(paths.into_iter().map(Ok)).boxed();
+    let mut deleted = 0u64;
+    let mut errors = 0u64;
+    let mut res = store.delete_stream(stream);
+    while let Some(r) = res.next().await {
+        match r {
+            Ok(_) => deleted += 1,
+            Err(e) => {
+                errors += 1;
+                tracing::warn!("log cleanup: failed to delete object: {e:#}");
+            }
+        }
+    }
+    (deleted, errors)
+}
+
+/// Delete the given relative paths from the local filesystem under `base_dir`.
+async fn disk_bulk_delete(base_dir: &str, rel_paths: &[String]) {
+    let futs = rel_paths.iter().map(|p| async move {
+        let full = std::path::Path::new(base_dir).join(p);
+        if tokio::fs::metadata(&full).await.is_ok() {
+            if let Err(e) = tokio::fs::remove_file(&full).await {
+                tracing::warn!(
+                    "log cleanup: failed to delete {}: {e}",
+                    full.to_string_lossy()
+                );
+            }
+        }
+    });
+    futures::future::join_all(futs).await;
+}
+
+async fn cleanup_service_logs(db: &DB, store: &Arc<dyn ObjectStore>) -> error::Result<()> {
+    // Count candidates upfront for progress reporting.
+    let total: i64 = sqlx::query_scalar!(
+        "SELECT COUNT(*) FROM log_file WHERE log_ts <= now() - ($1::bigint::text || ' s')::interval",
+        SERVICE_LOG_RETENTION_SECS,
+    )
+    .fetch_one(db)
+    .await?
+    .unwrap_or(0);
+
+    update(|p| p.total_service = total as u64).await;
+
+    if total <= 0 {
+        return Ok(());
+    }
+
+    struct LogFileRow {
+        file_path: String,
+        hostname: String,
+    }
+
+    loop {
+        // DELETE a batch of expired service log rows, returning their (file_path, hostname).
+        let rows = sqlx::query_as!(
+            LogFileRow,
+            "DELETE FROM log_file WHERE (file_path, hostname) IN (
+                 SELECT file_path, hostname FROM log_file
+                 WHERE log_ts <= now() - ($1::bigint::text || ' s')::interval
+                 LIMIT $2
+             ) RETURNING file_path, hostname",
+            SERVICE_LOG_RETENTION_SECS,
+            SERVICE_LOG_BATCH,
+        )
+        .fetch_all(db)
+        .await?;
+
+        if rows.is_empty() {
+            break;
+        }
+
+        let rel_paths: Vec<String> = rows
+            .iter()
+            .map(|r| format!("{}/{}", r.hostname, r.file_path))
+            .collect();
+
+        let batch_len = rel_paths.len() as u64;
+
+        // S3 delete (batched).
+        let s3_paths: Vec<ObjectPath> = rel_paths
+            .iter()
+            .map(|p| ObjectPath::from(format!("{}{}", LOGS_SERVICE, p)))
+            .collect();
+        let (deleted, errors) = s3_bulk_delete(store, s3_paths).await;
+
+        // Disk delete in parallel (best-effort).
+        disk_bulk_delete(&*TMP_WINDMILL_LOGS_SERVICE, &rel_paths).await;
+
+        update(|p| {
+            p.processed_service = p.processed_service.saturating_add(batch_len);
+            // total_service is sampled once upfront; rows that become expired during
+            // the run must not make processed > total.
+            if p.processed_service > p.total_service {
+                p.total_service = p.processed_service;
+            }
+            p.s3_deleted = p.s3_deleted.saturating_add(deleted);
+            p.errors = p.errors.saturating_add(errors);
+        })
+        .await;
+    }
+
+    Ok(())
+}
+
+async fn cleanup_job_logs(db: &DB, store: &Arc<dyn ObjectStore>) -> error::Result<()> {
+    let retention_secs = *JOB_RETENTION_SECS.read().await;
+    if retention_secs <= 0 {
+        return Ok(());
+    }
+
+    // Upper bound on the number of expired jobs.
+    let total: i64 = sqlx::query_scalar!(
+        "SELECT COUNT(*) FROM v2_job_completed
+         WHERE completed_at <= now() - ($1::bigint::text || ' s')::interval",
+        retention_secs,
+    )
+    .fetch_one(db)
+    .await?
+    .unwrap_or(0);
+
+    update(|p| p.total_jobs = total as u64).await;
+
+    if total <= 0 {
+        return Ok(());
+    }
+
+    loop {
+        let (deleted_count, rel_paths) =
+            delete_expired_jobs_batch(db, retention_secs, JOB_BATCH).await?;
+
+        if deleted_count == 0 {
+            break;
+        }
+
+        // S3 delete (batched).
+        let s3_paths: Vec<ObjectPath> = rel_paths
+            .iter()
+            .map(|p| ObjectPath::from(p.clone()))
+            .collect();
+        let (deleted, errors) = s3_bulk_delete(store, s3_paths).await;
+
+        // Disk delete in parallel (best-effort).
+        disk_bulk_delete(&*WINDMILL_DIR, &rel_paths).await;
+
+        update(|p| {
+            p.processed_jobs = p.processed_jobs.saturating_add(deleted_count as u64);
+            if p.processed_jobs > p.total_jobs {
+                p.total_jobs = p.processed_jobs;
+            }
+            p.s3_deleted = p.s3_deleted.saturating_add(deleted);
+            p.errors = p.errors.saturating_add(errors);
+        })
+        .await;
+    }
+
+    Ok(())
+}
+
+/// Mirrors backend/src/monitor.rs::delete_expired_jobs_batch but returns the
+/// log paths instead of deleting them from storage itself, so the caller can
+/// issue a single batched S3 delete across many batches via delete_stream.
+async fn delete_expired_jobs_batch(
+    db: &DB,
+    job_retention_secs: i64,
+    batch_size: i64,
+) -> error::Result<(usize, Vec<String>)> {
+    let mut tx = db.begin().await?;
+
+    let active_root_job_ids: Vec<Uuid> = sqlx::query_scalar!(
+        "SELECT q.id FROM v2_job_queue q
+         JOIN v2_job j ON j.id = q.id
+         WHERE j.parent_job IS NULL
+           AND j.created_at <= now() - ($1::bigint::text || ' s')::interval",
+        job_retention_secs
+    )
+    .fetch_all(&mut *tx)
+    .await?;
+
+    let deleted_jobs: Vec<Uuid> = sqlx::query_scalar!(
+        "DELETE FROM v2_job_completed
+         WHERE id IN (
+             SELECT jc.id FROM v2_job_completed jc
+             LEFT JOIN v2_job j ON j.id = jc.id
+             WHERE jc.completed_at <= now() - ($1::bigint::text || ' s')::interval
+               AND COALESCE(j.root_job, j.flow_innermost_root_job, jc.id) != ALL($3)
+             ORDER BY jc.completed_at ASC
+             LIMIT $2
+             FOR UPDATE OF jc SKIP LOCKED
+         )
+         RETURNING id",
+        job_retention_secs,
+        batch_size,
+        &active_root_job_ids
+    )
+    .fetch_all(&mut *tx)
+    .await?;
+
+    let deleted_count = deleted_jobs.len();
+    if deleted_count == 0 {
+        tx.commit().await?;
+        return Ok((0, Vec::new()));
+    }
+
+    if let Err(e) = sqlx::query!(
+        "DELETE FROM job_stats WHERE job_id = ANY($1)",
+        &deleted_jobs
+    )
+    .execute(&mut *tx)
+    .await
+    {
+        tracing::error!("log cleanup: error deleting job stats: {e:?}");
+    }
+
+    let log_paths: Vec<String> = match sqlx::query_scalar!(
+        "DELETE FROM job_logs WHERE job_id = ANY($1) RETURNING log_file_index",
+        &deleted_jobs
+    )
+    .fetch_all(&mut *tx)
+    .await
+    {
+        Ok(log_file_index) => log_file_index
+            .into_iter()
+            .filter_map(|opt| opt)
+            .flat_map(|inner_vec| inner_vec.into_iter())
+            .collect(),
+        Err(e) => {
+            tracing::error!("log cleanup: error deleting job logs: {e:?}");
+            Vec::new()
+        }
+    };
+
+    if let Err(e) = sqlx::query!("DELETE FROM v2_job WHERE id = ANY($1)", &deleted_jobs)
+        .execute(&mut *tx)
+        .await
+    {
+        tracing::error!("log cleanup: error deleting job: {e:?}");
+    }
+
+    if let Err(e) = sqlx::query!(
+        "DELETE FROM job_result_stream_v2 WHERE job_id = ANY($1)",
+        &deleted_jobs
+    )
+    .execute(&mut *tx)
+    .await
+    {
+        tracing::error!("log cleanup: error deleting job result stream: {e:?}");
+    }
+
+    tx.commit().await?;
+
+    Ok((deleted_count, log_paths))
+}
+
+/// Spawn the cleanup task. Caller is responsible for ensuring only one runs at a time
+/// (use `try_start` first).
+pub fn spawn_cleanup(db: DB) {
+    use futures::FutureExt;
+    use std::panic::AssertUnwindSafe;
+
+    tokio::spawn(async move {
+        let task = async {
+            let store = match windmill_object_store::get_object_store().await {
+                Some(s) => s,
+                None => {
+                    record_error("Object storage is not configured".to_string()).await;
+                    return;
+                }
+            };
+            if let Err(e) = cleanup_service_logs(&db, &store).await {
+                record_error(format!("service logs phase failed: {e:#}")).await;
+            }
+            if let Err(e) = cleanup_job_logs(&db, &store).await {
+                record_error(format!("job logs phase failed: {e:#}")).await;
+            }
+        };
+
+        // catch_unwind so a panic inside the cleanup can't leave running=true forever.
+        if let Err(panic) = AssertUnwindSafe(task).catch_unwind().await {
+            let msg = panic
+                .downcast_ref::<&str>()
+                .map(|s| s.to_string())
+                .or_else(|| panic.downcast_ref::<String>().cloned())
+                .unwrap_or_else(|| "unknown panic".to_string());
+            record_error(format!("cleanup task panicked: {msg}")).await;
+        }
+
+        finish().await;
+    });
+}
--- a/backend/windmill-api/openapi.yaml
+++ b/backend/windmill-api/openapi.yaml
@@ -1356,6 +1356,100 @@ paths:
              schema:
                type: string

+  /settings/object_storage_usage:
+    get:
+      summary: get object storage usage by top-level folder
+      operationId: getObjectStorageUsage
+      tags:
+        - setting
+      responses:
+        "200":
+          description: list of top-level folders with their total size in bytes
+          content:
+            application/json:
+              schema:
+                type: array
+                items:
+                  type: object
+                  properties:
+                    prefix:
+                      type: string
+                    size:
+                      type: integer
+                      format: int64
+                  required:
+                    - prefix
+                    - size
+
+  /settings/run_log_cleanup:
+    post:
+      summary: start a manual cleanup of expired logs from object storage
+      operationId: runLogCleanup
+      tags:
+        - setting
+      responses:
+        "202":
+          description: cleanup started
+          content:
+            text/plain:
+              schema:
+                type: string
+
+  /settings/log_cleanup_status:
+    get:
+      summary: get status of the manual log cleanup task
+      operationId: getLogCleanupStatus
+      tags:
+        - setting
+      responses:
+        "200":
+          description: current or last log cleanup status (null if never run)
+          content:
+            application/json:
+              schema:
+                nullable: true
+                type: object
+                properties:
+                  running:
+                    type: boolean
+                  started_at:
+                    type: string
+                    format: date-time
+                  finished_at:
+                    type: string
+                    format: date-time
+                    nullable: true
+                  total_service:
+                    type: integer
+                    format: int64
+                  processed_service:
+                    type: integer
+                    format: int64
+                  total_jobs:
+                    type: integer
+                    format: int64
+                  processed_jobs:
+                    type: integer
+                    format: int64
+                  s3_deleted:
+                    type: integer
+                    format: int64
+                  errors:
+                    type: integer
+                    format: int64
+                  last_error:
+                    type: string
+                    nullable: true
+                required:
+                  - running
+                  - started_at
+                  - total_service
+                  - processed_service
+                  - total_jobs
+                  - processed_jobs
+                  - s3_deleted
+                  - errors
+
  /settings/send_stats:
    post:
      summary: send stats
--- a/frontend/src/lib/components/InstanceSettings.svelte
+++ b/frontend/src/lib/components/InstanceSettings.svelte
@@ -69,6 +69,16 @@
 	loadSettings()
 	loadVersion()

+	// When the user enables object storage for the first time, default
+	// `monitor_logs_on_s3` to true so S3 log files get cleaned up with their
+	// jobs. Backend still defaults to false for backwards compat with
+	// operators who never touched the setting.
+	$effect(() => {
+		if ($values['object_store_cache_config'] && $values['monitor_logs_on_s3'] === undefined) {
+			values.update((v) => ({ ...v, monitor_logs_on_s3: true }))
+		}
+	})
+
 	const dispatch = createEventDispatcher()

 	async function loadVersion() {
--- a/frontend/src/lib/components/ObjectStoreConfigSettings.svelte
+++ b/frontend/src/lib/components/ObjectStoreConfigSettings.svelte
@@ -1,11 +1,13 @@
 <script lang="ts">
-	import { Database, Loader2 } from 'lucide-svelte'
+	import { Database, HardDrive, Loader2, Trash2 } from 'lucide-svelte'
+	import { onDestroy } from 'svelte'
 	import Toggle from './Toggle.svelte'
 	import { Button, Tab, Tabs } from './common'
 	import { SettingService } from '$lib/gen'
 	import { sendUserToast } from '$lib/toast'
 	import TestConnection from './TestConnection.svelte'
 	import { enterpriseLicense } from '$lib/stores'
+	import { displaySize } from '$lib/utils'
 	import SimpleEditor from './SimpleEditor.svelte'
 	import Label from './Label.svelte'
 	import TextInput from './text_input/TextInput.svelte'
@@ -76,6 +78,106 @@
 		}
 	}

+	let usageLoading = $state(false)
+	let usageError = $state<string | undefined>(undefined)
+	let usageData = $state<Array<{ prefix: string; size: number }> | undefined>(undefined)
+
+	async function loadUsage() {
+		usageLoading = true
+		usageError = undefined
+		try {
+			usageData = await SettingService.getObjectStorageUsage()
+		} catch (e: any) {
+			usageError = e?.body ?? e?.message ?? 'Failed to load storage usage'
+			usageData = undefined
+		} finally {
+			usageLoading = false
+		}
+	}
+
+	type CleanupStatus = {
+		running: boolean
+		started_at: string
+		finished_at?: string | null
+		total_service: number
+		processed_service: number
+		total_jobs: number
+		processed_jobs: number
+		s3_deleted: number
+		errors: number
+		last_error?: string | null
+	}
+
+	let cleanupStatus = $state<CleanupStatus | null | undefined>(undefined)
+	let cleanupStarting = $state(false)
+	let cleanupPollHandle: ReturnType<typeof setInterval> | undefined = undefined
+
+	let cleanupProgress = $derived.by(() => {
+		if (!cleanupStatus) return 0
+		const total = cleanupStatus.total_service + cleanupStatus.total_jobs
+		const processed = cleanupStatus.processed_service + cleanupStatus.processed_jobs
+		return total > 0 ? Math.min(100, Math.round((processed / total) * 100)) : 0
+	})
+
+	async function fetchCleanupStatus() {
+		try {
+			cleanupStatus = (await SettingService.getLogCleanupStatus()) ?? null
+		} catch (e: any) {
+			// Silent — polling errors shouldn't spam toasts.
+			console.warn('failed to fetch log cleanup status', e)
+		}
+	}
+
+	function startPolling() {
+		if (cleanupPollHandle !== undefined) return
+		cleanupPollHandle = setInterval(async () => {
+			await fetchCleanupStatus()
+			if (cleanupStatus && !cleanupStatus.running) {
+				stopPolling()
+			}
+		}, 1000)
+	}
+
+	function stopPolling() {
+		if (cleanupPollHandle !== undefined) {
+			clearInterval(cleanupPollHandle)
+			cleanupPollHandle = undefined
+		}
+	}
+
+	async function startCleanup() {
+		cleanupStarting = true
+		try {
+			await SettingService.runLogCleanup()
+			await fetchCleanupStatus()
+			startPolling()
+		} catch (e: any) {
+			sendUserToast(e?.body ?? e?.message ?? 'Failed to start cleanup', true)
+		} finally {
+			cleanupStarting = false
+		}
+	}
+
+	let hasConfig = $derived(Boolean(bucket_config))
+	$effect(() => {
+		if (hasConfig) {
+			let cancelled = false
+			fetchCleanupStatus().then(() => {
+				if (!cancelled && cleanupStatus?.running) {
+					startPolling()
+				}
+			})
+			return () => {
+				cancelled = true
+			}
+		} else {
+			stopPolling()
+			cleanupStatus = undefined
+		}
+	})
+
+	onDestroy(stopPolling)
+
 	let simpleEditor: SimpleEditor | undefined = $state(undefined)
 	let serviceAccountKeyCode = $state(
 		bucket_config?.type === 'Gcs'
@@ -145,6 +247,111 @@
 			/>
 		</div>

+		<div class="border rounded-md p-3 my-2">
+			<div class="flex items-center justify-between">
+				<span class="text-xs font-semibold text-emphasis">Storage usage by folder</span>
+				<Button spacingSize="sm" size="xs" btnClasses="h-8" variant="border" on:click={loadUsage}>
+					{#if usageLoading}
+						<Loader2 class="animate-spin mr-2 !h-4 !w-4" />
+					{:else}
+						<HardDrive class="mr-2 !h-4 !w-4" />
+					{/if}
+					{usageData ? 'Refresh' : 'Show usage'}
+				</Button>
+			</div>
+
+			{#if usageError}
+				<div class="text-red-500 text-xs mt-2">{usageError}</div>
+			{/if}
+
+			{#if usageData}
+				{#if usageData.length === 0}
+					<div class="text-tertiary text-xs mt-2">No objects found in the bucket.</div>
+				{:else}
+					<div class="flex flex-col gap-0.5 mt-2">
+						{#each usageData as item (item.prefix)}
+							<div
+								class="flex justify-between items-center text-xs py-1 px-2 rounded hover:bg-surface-hover"
+							>
+								<span class="font-mono text-secondary">{item.prefix}</span>
+								<span class="text-tertiary font-semibold">{displaySize(item.size) ?? '0 B'}</span>
+							</div>
+						{/each}
+						<div
+							class="flex justify-between items-center text-xs py-1 px-2 border-t mt-1 pt-2 font-semibold"
+						>
+							<span>Total</span>
+							<span
+								>{displaySize(usageData.reduce((acc, item) => acc + item.size, 0)) ?? '0 B'}</span
+							>
+						</div>
+					</div>
+				{/if}
+			{/if}
+		</div>
+
+		<div class="border rounded-md p-3 my-2">
+			<div class="flex items-center justify-between gap-2">
+				<div class="flex flex-col">
+					<span class="text-xs font-semibold text-emphasis">Clean up expired logs</span>
+					<span class="text-tertiary text-2xs">
+						Delete expired service &amp; job logs from object storage and disk now. Uses batched
+						deletes (up to 1000 objects per request).
+					</span>
+				</div>
+				<Button
+					spacingSize="sm"
+					size="xs"
+					btnClasses="h-8"
+					variant="border"
+					disabled={cleanupStarting || cleanupStatus?.running}
+					on:click={startCleanup}
+				>
+					{#if cleanupStarting || cleanupStatus?.running}
+						<Loader2 class="animate-spin mr-2 !h-4 !w-4" />
+					{:else}
+						<Trash2 class="mr-2 !h-4 !w-4" />
+					{/if}
+					{cleanupStatus?.running ? 'Running…' : 'Run cleanup'}
+				</Button>
+			</div>
+
+			{#if cleanupStatus}
+				{@const total = cleanupStatus.total_service + cleanupStatus.total_jobs}
+				{@const processed = cleanupStatus.processed_service + cleanupStatus.processed_jobs}
+				<div class="mt-3 flex flex-col gap-1">
+					<div class="w-full h-2 bg-surface-secondary rounded overflow-hidden">
+						<div class="h-full bg-blue-500 transition-all" style:width="{cleanupProgress}%"></div>
+					</div>
+					<div class="flex justify-between text-2xs text-tertiary">
+						<span>
+							{processed.toLocaleString()} / {total.toLocaleString()} files ({cleanupProgress}%)
+						</span>
+						<span>
+							S3 deleted: {cleanupStatus.s3_deleted.toLocaleString()}
+							{#if cleanupStatus.errors > 0}
+								&middot; errors: {cleanupStatus.errors.toLocaleString()}
+							{/if}
+						</span>
+					</div>
+					<div class="text-2xs text-tertiary">
+						Service logs: {cleanupStatus.processed_service.toLocaleString()} / {cleanupStatus.total_service.toLocaleString()}
+						&middot; Job logs: {cleanupStatus.processed_jobs.toLocaleString()} / {cleanupStatus.total_jobs.toLocaleString()}
+					</div>
+					{#if !cleanupStatus.running && cleanupStatus.finished_at}
+						<div class="text-2xs text-tertiary">
+							Finished at {new Date(cleanupStatus.finished_at).toLocaleString()}
+						</div>
+					{/if}
+					{#if cleanupStatus.last_error}
+						<div class="text-red-500 text-2xs mt-1">
+							Last error: {cleanupStatus.last_error}
+						</div>
+					{/if}
+				</div>
+			{/if}
+		</div>
+
 		<Tabs
 			selected={bucket_config?.type ?? 'S3'}
 			on:selected={(e) => {
--- a/frontend/src/lib/components/instanceSettings.ts
+++ b/frontend/src/lib/components/instanceSettings.ts
@@ -294,7 +294,7 @@ export const settings: Record<string, Setting[]> = {
 		{
 			label: 'Delete logs from s3 periodically',
 			description:
-				'Job and service logs are periodically deleted from disk. When this setting is on, they will also be deleted from the object storage.',
+				'Job and service logs are periodically deleted from disk when they expire. When this setting is on, they are also deleted from object storage. Defaults to on when object storage is configured; turn off to keep logs in object storage indefinitely.',
 			key: 'monitor_logs_on_s3',
 			fieldType: 'boolean',
 			storage: 'setting',