diff --git a/backend/migrations/20230730181921_custom_timeout.down.sql b/backend/migrations/20230730181921_custom_timeout.down.sql new file mode 100644 index 0000000000..d2f607c5b8 --- /dev/null +++ b/backend/migrations/20230730181921_custom_timeout.down.sql @@ -0,0 +1 @@ +-- Add down migration script here diff --git a/backend/migrations/20230730181921_custom_timeout.up.sql b/backend/migrations/20230730181921_custom_timeout.up.sql new file mode 100644 index 0000000000..db476677b3 --- /dev/null +++ b/backend/migrations/20230730181921_custom_timeout.up.sql @@ -0,0 +1,2 @@ +-- Add up migration script here +ALTER TABLE queue ADD COLUMN timeout INTEGER; diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index 0ead56a7e3..675a63a10f 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -857,6 +857,7 @@ impl RunJob { None, true, None, + None, ) .await .expect("push has to succeed"); @@ -1028,6 +1029,7 @@ async fn test_deno_flow(db: Pool) { sleep: None, cache_ttl: None, mock: None, + timeout: None, }, FlowModule { id: "b".to_string(), @@ -1060,6 +1062,7 @@ async fn test_deno_flow(db: Pool) { sleep: None, cache_ttl: None, mock: None, + timeout: None, }], }, stop_after_if: Default::default(), @@ -1069,6 +1072,7 @@ async fn test_deno_flow(db: Pool) { sleep: None, cache_ttl: None, mock: None, + timeout: None, }, ], same_worker: false, @@ -1167,6 +1171,8 @@ async fn test_deno_flow_same_worker(db: Pool) { sleep: None, cache_ttl: None, mock: None, + timeout: None, + }, FlowModule { id: "b".to_string(), @@ -1210,6 +1216,7 @@ async fn test_deno_flow_same_worker(db: Pool) { sleep: None, cache_ttl: None, mock: None, + timeout: None, }, FlowModule { id: "e".to_string(), @@ -1239,7 +1246,9 @@ async fn test_deno_flow_same_worker(db: Pool) { retry: None, sleep: None, cache_ttl: None, - mock: None + mock: None, + timeout: None, + }, ], }, @@ -1250,6 +1259,8 @@ async fn test_deno_flow_same_worker(db: Pool) { sleep: None, cache_ttl: None, mock: None, + timeout: None, + }, FlowModule { id: "c".to_string(), @@ -1288,6 +1299,7 @@ async fn test_deno_flow_same_worker(db: Pool) { sleep: None, cache_ttl: None, mock: None, + timeout: None, }, ], diff --git a/backend/windmill-api/src/apps.rs b/backend/windmill-api/src/apps.rs index cd8d2bb3de..8f755fbdaf 100644 --- a/backend/windmill-api/src/apps.rs +++ b/backend/windmill-api/src/apps.rs @@ -808,6 +808,7 @@ async fn execute_component( None, true, tag, + None, ) .await?; tx.commit().await?; diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index da081fd793..efea8b3461 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -264,6 +264,7 @@ async fn create_flow( None, true, nf.tag, + None, ) .await?; @@ -434,6 +435,7 @@ async fn update_flow( None, true, None, + None, ) .await?; sqlx::query!( @@ -657,6 +659,7 @@ mod tests { sleep: None, cache_ttl: None, mock: None, + timeout: None, }, FlowModule { id: "b".to_string(), @@ -680,6 +683,7 @@ mod tests { sleep: None, cache_ttl: None, mock: None, + timeout: None, }, FlowModule { id: "c".to_string(), @@ -699,6 +703,7 @@ mod tests { sleep: None, cache_ttl: None, mock: None, + timeout: None, }, ], failure_module: Some(FlowModule { @@ -718,6 +723,7 @@ mod tests { sleep: None, cache_ttl: None, mock: None, + timeout: None, }), same_worker: false, }; diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 8c2029e854..210e491764 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -1240,6 +1240,7 @@ impl From for Job { tag: uj.tag, concurrent_limit: uj.concurrent_limit, concurrency_time_window_s: uj.concurrency_time_window_s, + timeout: None, }), t => panic!("job type {} not valid", t), } @@ -1474,6 +1475,7 @@ pub async fn run_flow_by_path( None, !run_query.invisible_to_owner.unwrap_or(false), tag, + None, ) .await?; tx.commit().await?; @@ -1518,6 +1520,7 @@ pub async fn run_job_by_path( None, !run_query.invisible_to_owner.unwrap_or(false), tag, + None, ) .await?; tx.commit().await?; @@ -1719,6 +1722,7 @@ pub async fn run_wait_result_job_by_path_get( None, !run_query.invisible_to_owner.unwrap_or(false), tag, + None, ) .await?; tx.commit().await?; @@ -1888,6 +1892,7 @@ async fn run_wait_result_script_by_path_internal( None, !run_query.invisible_to_owner.unwrap_or(false), tag, + None, ) .await?; tx.commit().await?; @@ -1947,6 +1952,7 @@ pub async fn run_wait_result_script_by_hash( None, !run_query.invisible_to_owner.unwrap_or(false), tag, + None, ) .await?; tx.commit().await?; @@ -2051,6 +2057,7 @@ async fn run_wait_result_flow_by_path_internal( None, !run_query.invisible_to_owner.unwrap_or(false), tag, + None, ) .await?; tx.commit().await?; @@ -2115,6 +2122,7 @@ async fn run_preview_job( None, true, preview.tag, + None, ) .await?; tx.commit().await?; @@ -2151,6 +2159,7 @@ async fn add_noop_jobs( None, true, None, + None, ) .await?; tx = ntx; @@ -2198,6 +2207,7 @@ async fn run_preview_flow_job( None, true, raw_flow.tag, + None, ) .await?; tx.commit().await?; @@ -2248,6 +2258,7 @@ pub async fn run_job_by_hash( None, !run_query.invisible_to_owner.unwrap_or(false), tag, + None, ) .await?; tx.commit().await?; diff --git a/backend/windmill-api/src/oauth2.rs b/backend/windmill-api/src/oauth2.rs index 54e6b42395..ccab759f14 100644 --- a/backend/windmill-api/src/oauth2.rs +++ b/backend/windmill-api/src/oauth2.rs @@ -840,6 +840,7 @@ async fn slack_command( None, true, tag, + None, ) .await?; let url = BASE_URL.to_owned(); diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index 134a4253be..f21cd2671d 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/backend/windmill-api/src/scripts.rs @@ -512,6 +512,7 @@ async fn create_script( None, true, ns.tag, + None, ) .await?; tx = new_tx; diff --git a/backend/windmill-common/src/flows.rs b/backend/windmill-common/src/flows.rs index 36790c4f12..d42b778868 100644 --- a/backend/windmill-common/src/flows.rs +++ b/backend/windmill-common/src/flows.rs @@ -184,6 +184,8 @@ pub struct FlowModule { pub sleep: Option, #[serde(skip_serializing_if = "Option::is_none")] pub cache_ttl: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub timeout: Option, } impl FlowModule { diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs index c98876bdc6..78c09fe0ea 100644 --- a/backend/windmill-common/src/jobs.rs +++ b/backend/windmill-common/src/jobs.rs @@ -86,6 +86,8 @@ pub struct QueuedJob { pub concurrent_limit: Option, #[serde(skip_serializing_if = "Option::is_none")] pub concurrency_time_window_s: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub timeout: Option, } impl QueuedJob { @@ -150,19 +152,36 @@ impl Default for QueuedJob { tag: "deno".to_string(), concurrent_limit: None, concurrency_time_window_s: None, + timeout: None, } } } #[derive(Debug, Clone)] pub enum JobPayload { - ScriptHub { path: String }, - ScriptHash { hash: ScriptHash, path: String, concurrent_limit: Option, concurrency_time_window_s: Option }, + ScriptHub { + path: String, + }, + ScriptHash { + hash: ScriptHash, + path: String, + concurrent_limit: Option, + concurrency_time_window_s: Option, + }, Code(RawCode), - Dependencies { hash: ScriptHash, dependencies: String, language: ScriptLang }, - FlowDependencies { path: String }, + Dependencies { + hash: ScriptHash, + dependencies: String, + language: ScriptLang, + }, + FlowDependencies { + path: String, + }, Flow(String), - RawFlow { value: FlowValue, path: Option }, + RawFlow { + value: FlowValue, + path: Option, + }, Identity, Http, Noop, @@ -188,9 +207,15 @@ pub async fn script_path_to_payload<'c>( let (job_payload, tag) = if script_path.starts_with("hub/") { (JobPayload::ScriptHub { path: script_path.to_owned() }, None) } else { - let (script_hash, tag, concurrent_limit, concurrency_time_window_s) = get_latest_deployed_hash_for_path(db, w_id, script_path).await?; + let (script_hash, tag, concurrent_limit, concurrency_time_window_s) = + get_latest_deployed_hash_for_path(db, w_id, script_path).await?; ( - JobPayload::ScriptHash { hash: script_hash, path: script_path.to_owned(), concurrent_limit, concurrency_time_window_s}, + JobPayload::ScriptHash { + hash: script_hash, + path: script_path.to_owned(), + concurrent_limit, + concurrency_time_window_s, + }, tag, ) }; @@ -214,7 +239,11 @@ pub async fn script_hash_to_tag_and_limits<'c>( "querying getting tag for hash {script_hash}: {e}" )) })?; - Ok((script.tag, script.concurrent_limit, script.concurrency_time_window_s)) + Ok(( + script.tag, + script.concurrent_limit, + script.concurrency_time_window_s, + )) } pub async fn get_payload_tag_from_prefixed_path<'c>( diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 4875cc54d8..13479ded5f 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -472,6 +472,7 @@ pub async fn run_error_handler( None, true, tag, + None, ) .await?; tx.commit().await?; @@ -672,6 +673,7 @@ async fn handle_on_failure<'c, R: rsmq_async::RsmqConnection + Clone + Send + 'c None, true, tag, + None, ) .await?; tracing::info!( @@ -1016,6 +1018,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( pre_run_error: Option<&windmill_common::error::Error>, visible_to_owner: bool, mut tag: Option, + custom_timeout: Option, ) -> Result<(Uuid, QueueTransaction<'c, R>), Error> { let args_json = serde_json::Value::Object(args); let job_id: Uuid = if let Some(job_id) = job_id { @@ -1303,6 +1306,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( sleep: None, suspend: None, cache_ttl: None, + timeout: None, }); raw_flow = Some(FlowValue { modules, ..flow.clone() }); } @@ -1345,8 +1349,8 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( "INSERT INTO queue (workspace_id, id, running, parent_job, created_by, permissioned_as, scheduled_for, script_hash, script_path, raw_code, raw_lock, args, job_kind, schedule_path, raw_flow, \ - flow_status, is_flow_step, language, started_at, same_worker, pre_run_error, email, visible_to_owner, root_job, tag, concurrent_limit, concurrency_time_window_s) - VALUES ($1, $2, $3, $4, $5, $6, COALESCE($7, now()), $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, CASE WHEN $3 THEN now() END, $19, $20, $21, $22, $23, $24, $25, $26) \ + flow_status, is_flow_step, language, started_at, same_worker, pre_run_error, email, visible_to_owner, root_job, tag, concurrent_limit, concurrency_time_window_s, timeout) + VALUES ($1, $2, $3, $4, $5, $6, COALESCE($7, now()), $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, CASE WHEN $3 THEN now() END, $19, $20, $21, $22, $23, $24, $25, $26, $27) \ RETURNING id", workspace_id, job_id, @@ -1374,6 +1378,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( tag, concurrent_limit, concurrency_time_window_s, + custom_timeout, ) .fetch_one(&mut tx) .await diff --git a/backend/windmill-queue/src/schedule.rs b/backend/windmill-queue/src/schedule.rs index ab22015cc5..3d356037d3 100644 --- a/backend/windmill-queue/src/schedule.rs +++ b/backend/windmill-queue/src/schedule.rs @@ -71,14 +71,20 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>( let (payload, tag) = if schedule.is_flow { (JobPayload::Flow(schedule.script_path), None) } else { - let (hash, tag, concurrent_limit, concurrency_time_window_s) = windmill_common::get_latest_hash_for_path( - tx.transaction_mut(), - &schedule.workspace_id, - &schedule.script_path, - ) - .await?; + let (hash, tag, concurrent_limit, concurrency_time_window_s) = + windmill_common::get_latest_hash_for_path( + tx.transaction_mut(), + &schedule.workspace_id, + &schedule.script_path, + ) + .await?; ( - JobPayload::ScriptHash { hash, path: schedule.script_path, concurrent_limit: concurrent_limit, concurrency_time_window_s: concurrency_time_window_s}, + JobPayload::ScriptHash { + hash, + path: schedule.script_path, + concurrent_limit: concurrent_limit, + concurrency_time_window_s: concurrency_time_window_s, + }, tag, ) }; @@ -109,6 +115,7 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>( None, true, tag, + None, ) .await?; Ok(tx) // TODO: Bubble up pushed UUID from here diff --git a/backend/windmill-worker/src/go_executor.rs b/backend/windmill-worker/src/go_executor.rs index b0c23e932e..2af8e694c5 100644 --- a/backend/windmill-worker/src/go_executor.rs +++ b/backend/windmill-worker/src/go_executor.rs @@ -190,6 +190,7 @@ func Run(req Req) (interface{{}}, error){{ worker_name, &job.workspace_id, "go build", + None, ) .await?; @@ -239,6 +240,7 @@ func Run(req Req) (interface{{}}, error){{ worker_name, &job.workspace_id, "go run", + job.timeout, ) .await?; read_result(job_dir).await @@ -283,7 +285,18 @@ pub async fn install_go_dependencies( .stderr(Stdio::piped()) .spawn()?; - handle_child(job_id, db, logs, child, false, worker_name, w_id, "go init").await?; + handle_child( + job_id, + db, + logs, + child, + false, + worker_name, + w_id, + "go init", + None, + ) + .await?; } let mut new_lockfile = false; @@ -331,6 +344,7 @@ pub async fn install_go_dependencies( worker_name, &w_id, &format!("go {mod_command}"), + None, ) .await .map_err(|e| Error::ExecutionErr(format!("Lockfile generation failed: {e:?}")))?; diff --git a/backend/windmill-worker/src/python_executor.rs b/backend/windmill-worker/src/python_executor.rs index e88c1bab95..39beb099dd 100644 --- a/backend/windmill-worker/src/python_executor.rs +++ b/backend/windmill-worker/src/python_executor.rs @@ -140,6 +140,7 @@ pub async fn pip_compile( worker_name, &w_id, "pip-compile", + None, ) .await .map_err(|e| Error::ExecutionErr(format!("Lock file generation failed: {e:?}")))?; @@ -473,6 +474,7 @@ mount {{ worker_name, &job.workspace_id, "python run", + job.timeout, ) .await?; read_result(job_dir).await @@ -631,6 +633,7 @@ pub async fn handle_python_reqs( worker_name, &w_id, &format!("pip install {req}"), + None, ) .await; tracing::info!( diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 097913a843..8611c95925 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -1290,8 +1290,7 @@ async fn handle_queued_job( worker_dir, &mut logs, base_internal_url, - worker_name - ) + worker_name ) .await } } @@ -1472,9 +1471,7 @@ async fn handle_code_execution_job( worker_dir: &str, logs: &mut String, base_internal_url: &str, - worker_name: &str - -) -> error::Result { + worker_name: &str) -> error::Result { let (inner_content, requirements_o, language, envs) = match job.job_kind { JobKind::Preview => ( job.raw_code @@ -1751,7 +1748,7 @@ async fn handle_bash_job( .stderr(Stdio::piped()) .spawn()? }; - handle_child(&job.id, db, logs, child, !*DISABLE_NSJAIL, worker_name, &job.workspace_id, "bash run").await?; + handle_child(&job.id, db, logs, child, !*DISABLE_NSJAIL, worker_name, &job.workspace_id, "bash run", job.timeout).await?; //for now bash jobs have an empty result object Ok(serde_json::json!(logs .lines() @@ -1954,7 +1951,7 @@ run().catch(async (e) => {{ }; // logs.push_str(format!("prepare: {:?}\n", start.elapsed().as_micros()).as_str()); // start = Instant::now(); - handle_child(&job.id, db, logs, child, false, worker_name, &job.workspace_id, "deno run").await?; + handle_child(&job.id, db, logs, child, false, worker_name, &job.workspace_id, "deno run", job.timeout).await?; // logs.push_str(format!("execute: {:?}\n", start.elapsed().as_millis()).as_str()); if let Err(e) = tokio::fs::remove_dir_all(format!("{DENO_CACHE_DIR}/gen/file/{job_dir}")).await { tracing::error!("failed to remove deno gen tmp cache dir: {}", e); @@ -2101,7 +2098,7 @@ let child = if !*DISABLE_NSJAIL { // logs.push_str(format!("prepare: {:?}\n", start.elapsed().as_micros()).as_str()); // start = Instant::now(); - handle_child(&job.id, db, logs, child, false, worker_name, &job.workspace_id, "bun run").await?; + handle_child(&job.id, db, logs, child, false, worker_name, &job.workspace_id, "bun run", job.timeout).await?; // logs.push_str(format!("execute: {:?}\n", start.elapsed().as_millis()).as_str()); read_result(job_dir).await } @@ -2476,6 +2473,7 @@ pub async fn handle_child( worker_name: &str, _w_id: &str, child_name: &str, + custom_timeout: Option, ) -> error::Result<()> { let start = Instant::now(); let update_job_interval = Duration::from_millis(500); @@ -2554,7 +2552,7 @@ pub async fn handle_child( let db = db.clone(); #[cfg(not(feature = "enterprise"))] - let timeout_duration = *TIMEOUT_DURATION; + let instance_timeout_duration = *TIMEOUT_DURATION; #[cfg(feature = "enterprise")] let premium_workspace = *CLOUD_HOSTED && sqlx::query_scalar!("SELECT premium FROM workspace WHERE id = $1", _w_id) @@ -2565,11 +2563,17 @@ pub async fn handle_child( }).unwrap_or(false); #[cfg(feature = "enterprise")] - let timeout_duration = if premium_workspace { + let instance_timeout_duration = if premium_workspace { *TIMEOUT_DURATION*6 //30mins } else { *TIMEOUT_DURATION }; + + let timeout_duration = if let Some(custom_timeout) = custom_timeout { + Duration::min(instance_timeout_duration, Duration::from_secs(custom_timeout as u64)) + } else { + instance_timeout_duration + }; let kill_reason = tokio::select! { biased; diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 691249de51..aec7248557 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -1432,6 +1432,7 @@ async fn push_next_flow_job } else { Some(flow_job.tag.clone()) }, + module.timeout, ) .await?; tx = inner_tx; diff --git a/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte b/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte index ed109695d6..d8885097a4 100644 --- a/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte +++ b/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte @@ -33,6 +33,7 @@ import Tooltip from '$lib/components/Tooltip.svelte' import { SecondsInput } from '$lib/components/common' import DiffEditor from '$lib/components/DiffEditor.svelte' + import FlowModuleTimeout from './FlowModuleTimeout.svelte' const { selectedId, previewArgs, flowStateStore, flowStore, saveDraft } = getContext('FlowEditorContext') @@ -290,6 +291,7 @@ Sleep Mock Shared Directory + Timeout {/if}
@@ -313,6 +315,10 @@
+ {:else if advancedSelected === 'timeout'} +
+ +
{:else if advancedSelected === 'concurrency'}

diff --git a/frontend/src/lib/components/flows/content/FlowModuleTimeout.svelte b/frontend/src/lib/components/flows/content/FlowModuleTimeout.svelte new file mode 100644 index 0000000000..71833d1504 --- /dev/null +++ b/frontend/src/lib/components/flows/content/FlowModuleTimeout.svelte @@ -0,0 +1,48 @@ + + +

+ Timeout + + If defined, the custom timeout will be used instead of the instance timeout for the step. The + step's timeout cannot be greater than the instance timeout. + +

+ + { + if (istimeoutEnabled && flowModule.timeout != undefined) { + flowModule.timeout = undefined + } else { + flowModule.timeout = 300 + } + }} + options={{ + right: 'Add a custom timeout for this step' + }} +/> +
+ Timeout duration + + {#if flowModule.timeout} + + {:else} + + {/if} + +
+ + +

The timeout will be ignored when running "Test this step"

+ +
diff --git a/openflow.openapi.yaml b/openflow.openapi.yaml index 9bb6bdfd5d..34666f28c5 100644 --- a/openflow.openapi.yaml +++ b/openflow.openapi.yaml @@ -90,6 +90,8 @@ components: $ref: "#/components/schemas/InputTransform" cache_ttl: type: number + timeout: + type: number summary: type: string mock: