diff --git a/backend/.sqlx/query-2e4115bb2e6c8c85ad1492ad135d6b0454b342126cb5fa17e58caf71b32ee755.json b/backend/.sqlx/query-2be66f23536223549db9b50025932b6b1bad90b8fa47d97acb7d75aa3c37ef86.json similarity index 57% rename from backend/.sqlx/query-2e4115bb2e6c8c85ad1492ad135d6b0454b342126cb5fa17e58caf71b32ee755.json rename to backend/.sqlx/query-2be66f23536223549db9b50025932b6b1bad90b8fa47d97acb7d75aa3c37ef86.json index eaa75c2020..204cd6df90 100644 --- a/backend/.sqlx/query-2e4115bb2e6c8c85ad1492ad135d6b0454b342126cb5fa17e58caf71b32ee755.json +++ b/backend/.sqlx/query-2be66f23536223549db9b50025932b6b1bad90b8fa47d97acb7d75aa3c37ef86.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "INSERT INTO variable\n (workspace_id, path, value, is_secret, description, account, is_oauth)\n VALUES ($1, $2, $3, $4, $5, $6, $7)", + "query": "INSERT INTO variable\n (workspace_id, path, value, is_secret, description, account, is_oauth, expires_at)\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8)", "describe": { "columns": [], "parameters": { @@ -11,10 +11,11 @@ "Bool", "Varchar", "Int4", - "Bool" + "Bool", + "Timestamptz" ] }, "nullable": [] }, - "hash": "2e4115bb2e6c8c85ad1492ad135d6b0454b342126cb5fa17e58caf71b32ee755" + "hash": "2be66f23536223549db9b50025932b6b1bad90b8fa47d97acb7d75aa3c37ef86" } diff --git a/backend/.sqlx/query-bac12eff366f6b76f97d1a44041e0c1b5fdcb2c4205c1c1c0026b71ce404f089.json b/backend/.sqlx/query-bac12eff366f6b76f97d1a44041e0c1b5fdcb2c4205c1c1c0026b71ce404f089.json new file mode 100644 index 0000000000..a84b6b5ff0 --- /dev/null +++ b/backend/.sqlx/query-bac12eff366f6b76f97d1a44041e0c1b5fdcb2c4205c1c1c0026b71ce404f089.json @@ -0,0 +1,20 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM variable WHERE expires_at IS NOT NULL AND expires_at > now() RETURNING path", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "path", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + false + ] + }, + "hash": "bac12eff366f6b76f97d1a44041e0c1b5fdcb2c4205c1c1c0026b71ce404f089" +} diff --git a/backend/migrations/20240801084701_add_expiry.down.sql b/backend/migrations/20240801084701_add_expiry.down.sql new file mode 100644 index 0000000000..f627ed49d8 --- /dev/null +++ b/backend/migrations/20240801084701_add_expiry.down.sql @@ -0,0 +1,2 @@ +-- Add down migration script here +ALTER TABLE variable DROP COLUMN IF EXISTS expires_at; \ No newline at end of file diff --git a/backend/migrations/20240801084701_add_expiry.up.sql b/backend/migrations/20240801084701_add_expiry.up.sql new file mode 100644 index 0000000000..23cbed140a --- /dev/null +++ b/backend/migrations/20240801084701_add_expiry.up.sql @@ -0,0 +1,3 @@ +-- Add up migration script here +ALTER TABLE variable ADD COLUMN IF NOT EXISTS expires_at TIMESTAMP WITH TIME ZONE; +UPDATE variable SET expires_at = (now() + '7 days'::interval) WHERE path LIKE 'u/%/secret_arg/%' diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 95d2fad5c4..1f8eae150a 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -406,6 +406,21 @@ pub async fn delete_expired_items(db: &DB) -> () { Err(e) => tracing::error!("Error deleting cache resource {}", e.to_string()), } + let deleted_expired_variables = sqlx::query_scalar!( + "DELETE FROM variable WHERE expires_at IS NOT NULL AND expires_at > now() RETURNING path", + ) + .fetch_all(db) + .await; + + match deleted_expired_variables { + Ok(res) => { + if res.len() > 0 { + tracing::info!("deleted {} expired variables {:?}", res.len(), res) + } + } + Err(e) => tracing::error!("Error deleting cache resource {}", e.to_string()), + } + let job_retention_secs = *JOB_RETENTION_SECS.read().await; if job_retention_secs > 0 { match db.begin().await { diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index acf4b8f251..c0bbca03df 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -9772,6 +9772,9 @@ components: type: boolean is_refreshed: type: boolean + expires_at: + type: string + format: date-time required: - workspace_id - path @@ -9810,6 +9813,9 @@ components: type: integer is_oauth: type: boolean + expires_at: + type: string + format: date-time required: - path - value diff --git a/backend/windmill-api/src/db.rs b/backend/windmill-api/src/db.rs index 3c7c949cca..70a9413f87 100644 --- a/backend/windmill-api/src/db.rs +++ b/backend/windmill-api/src/db.rs @@ -316,35 +316,53 @@ async fn fix_job_completed_index(db: &DB) -> Result<(), Error> { tokio::time::sleep(std::time::Duration::from_secs(5)).await; } } - sqlx::query( + + tracing::info!("acquired lock for {migration_job_name}"); + let has_done_migration = sqlx::query_scalar!( + "SELECT EXISTS(SELECT name FROM windmill_migrations WHERE name = $1)", + migration_job_name + ) + .fetch_one(db) + .await? + .unwrap_or(false); + + if !has_done_migration { + sqlx::query( "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_completed_job_workspace_id_created_at_new_2 ON completed_job (workspace_id, job_kind, success, is_skipped, is_flow_step, created_at DESC)" ).execute(db).await?; - sqlx::query( + sqlx::query( "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_completed_job_workspace_id_started_at_new ON completed_job (workspace_id, job_kind, success, is_skipped, is_flow_step, started_at DESC)" ).execute(db).await?; - sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at") + sqlx::query( + "DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at", + ) .execute(db) .await?; - sqlx::query( - "DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at_new", - ) - .execute(db) - .await?; + sqlx::query( + "DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at_new", + ) + .execute(db) + .await?; + + sqlx::query!( + "INSERT INTO windmill_migrations (name) VALUES ($1) ON CONFLICT DO NOTHING", + migration_job_name + ) + .execute(&mut *tx) + .await?; + tracing::info!("Finished applying {migration_job_name} migration"); + } else { + tracing::info!("migration {migration_job_name} already done"); + } - sqlx::query!( - "INSERT INTO windmill_migrations (name) VALUES ($1) ON CONFLICT DO NOTHING", - migration_job_name - ) - .execute(&mut *tx) - .await?; let _ = sqlx::query("SELECT pg_advisory_unlock(4242)") .execute(&mut *tx) .await?; tx.commit().await?; - tracing::info!("Finished applying {migration_job_name} migration"); + tracing::info!("released lock for {migration_job_name}"); } let migration_job_name = "fix_job_completed_index_3"; @@ -373,34 +391,49 @@ async fn fix_job_completed_index(db: &DB) -> Result<(), Error> { tokio::time::sleep(std::time::Duration::from_secs(5)).await; } } + tracing::info!("acquired lock for {migration_job_name}"); - sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS index_completed_job_on_schedule_path") - .execute(db) - .await?; - - sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS concurrency_limit_stats_queue") - .execute(db) - .await?; - - sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS root_job_index") - .execute(db) - .await?; - - sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS index_completed_on_created") - .execute(db) - .await?; - - sqlx::query!( - "INSERT INTO windmill_migrations (name) VALUES ($1) ON CONFLICT DO NOTHING", + let has_done_migration = sqlx::query_scalar!( + "SELECT EXISTS(SELECT name FROM windmill_migrations WHERE name = $1)", migration_job_name ) - .execute(&mut *tx) - .await?; + .fetch_one(db) + .await? + .unwrap_or(false); + + if !has_done_migration { + sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS index_completed_job_on_schedule_path") + .execute(db) + .await?; + + sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS concurrency_limit_stats_queue") + .execute(db) + .await?; + + sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS root_job_index") + .execute(db) + .await?; + + sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS index_completed_on_created") + .execute(db) + .await?; + + sqlx::query!( + "INSERT INTO windmill_migrations (name) VALUES ($1) ON CONFLICT DO NOTHING", + migration_job_name + ) + .execute(&mut *tx) + .await?; + tracing::info!("Finished applying {migration_job_name} migration"); + } else { + tracing::info!("migration {migration_job_name} already done"); + } + let _ = sqlx::query("SELECT pg_advisory_unlock(4242)") .execute(&mut *tx) .await?; tx.commit().await?; - tracing::info!("Finished applying {migration_job_name} migration"); + tracing::info!("released lock for {migration_job_name}"); } Ok(()) diff --git a/backend/windmill-api/src/variables.rs b/backend/windmill-api/src/variables.rs index 04d6e6e827..40e74d41ea 100644 --- a/backend/windmill-api/src/variables.rs +++ b/backend/windmill-api/src/variables.rs @@ -103,7 +103,8 @@ async fn list_variables( is_secret, variable.description, variable.extra_perms, account, is_oauth, (now() > account.expires_at) as is_expired, account.refresh_error, resource.path IS NOT NULL as is_linked, - account.refresh_token != '' as is_refreshed + account.refresh_token != '' as is_refreshed, + variable.expires_at from variable LEFT JOIN account ON variable.account = account.id AND account.workspace_id = $1 LEFT JOIN resource ON resource.path = variable.path AND resource.workspace_id = $1 @@ -317,8 +318,8 @@ async fn create_variable( sqlx::query!( "INSERT INTO variable - (workspace_id, path, value, is_secret, description, account, is_oauth) - VALUES ($1, $2, $3, $4, $5, $6, $7)", + (workspace_id, path, value, is_secret, description, account, is_oauth, expires_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8)", &w_id, variable.path, value, @@ -326,6 +327,7 @@ async fn create_variable( variable.description, variable.account, variable.is_oauth.unwrap_or(false), + variable.expires_at ) .execute(&mut *tx) .await?; diff --git a/backend/windmill-api/src/workspaces.rs b/backend/windmill-api/src/workspaces.rs index c9e511f892..b3ae17a1c7 100644 --- a/backend/windmill-api/src/workspaces.rs +++ b/backend/windmill-api/src/workspaces.rs @@ -2599,9 +2599,9 @@ async fn tarball_workspace( if !skip_variables.unwrap_or(false) { let variables = sqlx::query_as::<_, ExportableListableVariable>(if !skip_secrets.unwrap_or(false) { - "SELECT * FROM variable WHERE workspace_id = $1 AND path NOT LIKE 'u/%/secret_arg/%'" + "SELECT * FROM variable WHERE workspace_id = $1 AND expires_at IS NULL" } else { - "SELECT * FROM variable WHERE workspace_id = $1 AND is_secret = false AND path NOT LIKE 'u/%/secret_arg/%'" + "SELECT * FROM variable WHERE workspace_id = $1 AND is_secret = false AND expires_at IS NULL" }) .bind(&w_id) .fetch_all(&mut *tx) diff --git a/backend/windmill-common/src/flows.rs b/backend/windmill-common/src/flows.rs index 575f44c3a5..f2c24ffaee 100644 --- a/backend/windmill-common/src/flows.rs +++ b/backend/windmill-common/src/flows.rs @@ -38,14 +38,18 @@ pub struct Flow { pub dedicated_worker: Option, #[serde(skip_serializing_if = "Option::is_none")] pub tag: Option, - #[serde(skip_serializing_if = "Option::is_none")] + #[serde(skip_serializing_if = "is_none_or_false")] pub ws_error_handler_muted: Option, #[serde(skip_serializing_if = "Option::is_none")] pub timeout: Option, - #[serde(skip_serializing_if = "Option::is_none")] + #[serde(skip_serializing_if = "is_none_or_false")] pub visible_to_runner_only: Option, } +fn is_none_or_false(b: &Option) -> bool { + b.is_none() || !b.unwrap() +} + #[derive(Serialize, sqlx::FromRow)] pub struct ListableFlow { pub workspace_id: String, @@ -406,7 +410,7 @@ pub enum FlowModuleValue { lock: Option, #[serde(skip_serializing_if = "Option::is_none")] path: Option, - #[serde(skip_serializing_if = "Option::is_none")] + #[serde(skip_serializing_if = "is_none_or_empty")] tag: Option, language: ScriptLang, #[serde(skip_serializing_if = "Option::is_none")] @@ -419,6 +423,10 @@ pub enum FlowModuleValue { Identity, } +fn is_none_or_empty(expr: &Option) -> bool { + expr.is_none() || expr.as_ref().unwrap().is_empty() +} + #[derive(Deserialize)] struct UntaggedFlowModuleValue { #[serde(rename = "type")] diff --git a/backend/windmill-common/src/variables.rs b/backend/windmill-common/src/variables.rs index 8239b8c321..1081c76809 100644 --- a/backend/windmill-common/src/variables.rs +++ b/backend/windmill-common/src/variables.rs @@ -40,10 +40,10 @@ pub struct ListableVariable { pub is_refreshed: Option, pub refresh_error: Option, pub is_linked: Option, + pub expires_at: Option>, } #[derive(Serialize, Deserialize, sqlx::FromRow)] - pub struct ExportableListableVariable { pub workspace_id: String, pub path: String, @@ -51,8 +51,16 @@ pub struct ExportableListableVariable { pub is_secret: bool, pub description: String, pub extra_perms: serde_json::Value, + #[serde(skip_serializing_if = "Option::is_none")] pub account: Option, + #[serde(skip_serializing_if = "is_none_or_false")] pub is_oauth: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub expires_at: Option>, +} + +fn is_none_or_false(b: &Option) -> bool { + b.is_none() || !b.unwrap() } #[derive(Deserialize)] @@ -63,6 +71,7 @@ pub struct CreateVariable { pub description: String, pub account: Option, pub is_oauth: Option, + pub expires_at: Option>, } pub async fn build_crypt(db: &DB, w_id: &str) -> crate::error::Result { diff --git a/frontend/src/lib/components/PasswordArgInput.svelte b/frontend/src/lib/components/PasswordArgInput.svelte index 425f31b313..9b87e8843b 100644 --- a/frontend/src/lib/components/PasswordArgInput.svelte +++ b/frontend/src/lib/components/PasswordArgInput.svelte @@ -23,7 +23,8 @@ value: password, is_secret: true, path: npath, - description: '' + description: '', + expires_at: new Date(Date.now() + 1000 * 60 * 60 * 24 * 7).toISOString() } }) path = npath