From 40f32e644e8dfb0ed863eaa341a16155e1a8d527 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Wed, 18 Jan 2023 09:02:38 +0100 Subject: [PATCH] feat(backend): add queue_limit + configurable timeout + fix timeout cancel --- README.md | 1 + backend/sqlx-data.json | 18 ++++++ backend/windmill-api/openapi.yaml | 8 +++ backend/windmill-api/src/jobs.rs | 64 +++++++++++++++++-- backend/windmill-api/src/lib.rs | 19 ++++-- .../scripts/get/[...hash]/+page.svelte | 36 ++++++++++- 6 files changed, 132 insertions(+), 14 deletions(-) diff --git a/README.md b/README.md index 4fb9e950e1..a8f21b3442 100644 --- a/README.md +++ b/README.md @@ -304,6 +304,7 @@ upcoming CLI tool. | HOME | None | The home directory to use for Go and Bash , usually inherited | Worker | | DATABASE_CONNECTIONS | 50 (Server)/3 (Worker) | The max number of connections in the database connection pool | All | | SUPERADMIN_SECRET | None | A token that would let the caller act as a virtual superadmin superadmin@windmill.dev | Server | +| TIMEOUT_WAIT_RESULT | 20 | The number of seconds to wait before timeout on the 'run_wait_result' endpoint | Worker | ## Run a local dev setup diff --git a/backend/sqlx-data.json b/backend/sqlx-data.json index 6ac948f986..f543700075 100644 --- a/backend/sqlx-data.json +++ b/backend/sqlx-data.json @@ -4205,6 +4205,24 @@ }, "query": "SELECT count(path) FROM schedule WHERE path LIKE 'f/' || $1 || '%' AND workspace_id = $2" }, + "b50b8d3367bd1f74986e6f923fe8497bcbfffea2570a3f3d1f8100207bc1557c": { + "describe": { + "columns": [ + { + "name": "count", + "ordinal": 0, + "type_info": "Int8" + } + ], + "nullable": [ + null + ], + "parameters": { + "Left": [] + } + }, + "query": "SELECT COUNT(*) FROM queue WHERE canceled = false AND (scheduled_for <= now()\n OR (suspend_until IS NOT NULL\n AND ( suspend <= 0\n OR suspend_until <= now())))" + }, "b5c9891b5bf3d581e62f8835aaa25b8158fe3cefc849af0ba312a45cf22721ca": { "describe": { "columns": [ diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index ab2ebaec19..8681c299c7 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -2424,6 +2424,7 @@ paths: type: integer - $ref: "#/components/parameters/ParentJob" - $ref: "#/components/parameters/IncludeHeader" + - $ref: "#/components/parameters/QueueLimit" requestBody: description: script args @@ -4461,6 +4462,13 @@ components: in: query schema: type: string + QueueLimit: + name: queue_limit + description: | + The maximum size of the queue for which the request would get rejected if that job would push it above that limit + in: query + schema: + type: string ScriptStartPath: name: script_path_start diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index e8102c0491..300eb4e786 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -38,7 +38,7 @@ use crate::{ db::{UserDB, DB}, users::{require_owner_of_path, Authed}, variables::get_workspace_key, - BaseUrl, + BaseUrl, TimeoutWaitResult, }; pub fn workspaced_service() -> Router { @@ -263,6 +263,7 @@ pub struct RunJobQuery { parent_job: Option, include_header: Option, invisible_to_owner: Option, + queue_limit: Option, } impl RunJobQuery { @@ -1227,9 +1228,18 @@ impl Drop for Guard { async fn run_wait_result( authed: Authed, Extension(user_db): Extension, + timeout: i32, uuid: Uuid, Path((w_id, _)): Path<(String, T)>, ) -> error::JsonResult { + let mut result = None; + let iters = if timeout <= 0 { + 20 + } else if timeout <= 1 { + timeout * 10 + } else { + 10 + ((timeout - 1) * 2) + }; let mut g = Guard { done: false, id: uuid, @@ -1237,8 +1247,7 @@ async fn run_wait_result( db: user_db.clone(), authed: authed.clone(), }; - let mut result = None; - for i in 0..48 { + for i in 0..iters { let mut tx = user_db.clone().begin(&authed).await?; result = sqlx::query_scalar!( "SELECT result FROM completed_job WHERE id = $1 AND workspace_id = $2", @@ -1256,22 +1265,45 @@ async fn run_wait_result( let delay = if i < 10 { 100 } else { 500 }; tokio::time::sleep(core::time::Duration::from_millis(delay)).await; } - g.done = true; if let Some(result) = result { + g.done = true; Ok(Json(result)) } else { - Err(Error::ExecutionErr("timeout after 20s".to_string())) + Err(Error::ExecutionErr(format!("timeout after {}s", timeout))) } } +pub async fn check_queue_too_long(db: DB, queue_limit: Option) -> error::Result<()> { + if let Some(limit) = queue_limit { + let count = sqlx::query_scalar!( + "SELECT COUNT(*) FROM queue WHERE canceled = false AND (scheduled_for <= now() + OR (suspend_until IS NOT NULL + AND ( suspend <= 0 + OR suspend_until <= now())))", + ) + .fetch_one(&db) + .await? + .unwrap_or(0); + + if count > queue_limit.unwrap() { + return Err(Error::InternalErr(format!( + "Number of queued job is too high: {count} > {limit}" + ))); + } + } + Ok(()) +} pub async fn run_wait_result_job_by_path( authed: Authed, Extension(user_db): Extension, + Extension(db): Extension, + Extension(timeout): Extension>, Path((w_id, script_path)): Path<(String, StripPath)>, Query(run_query): Query, headers: HeaderMap, Json(args): Json>>, ) -> error::JsonResult { + check_queue_too_long(db, run_query.queue_limit).await?; let script_path = script_path.to_path(); let mut tx = user_db.clone().begin(&authed).await?; let job_payload = script_path_to_payload(script_path, &mut tx, &w_id).await?; @@ -1298,17 +1330,28 @@ pub async fn run_wait_result_job_by_path( .await?; tx.commit().await?; - run_wait_result(authed, Extension(user_db), uuid, Path((w_id, script_path))).await + run_wait_result( + authed, + Extension(user_db), + timeout.0, + uuid, + Path((w_id, script_path)), + ) + .await } pub async fn run_wait_result_job_by_hash( authed: Authed, Extension(user_db): Extension, + Extension(db): Extension, + Extension(timeout): Extension>, Path((w_id, script_hash)): Path<(String, ScriptHash)>, Query(run_query): Query, headers: HeaderMap, Json(args): Json>>, ) -> error::JsonResult { + check_queue_too_long(db, run_query.queue_limit).await?; + let hash = script_hash.0; let mut tx = user_db.clone().begin(&authed).await?; let path = get_path_for_hash(&mut tx, &w_id, hash).await?; @@ -1334,7 +1377,14 @@ pub async fn run_wait_result_job_by_hash( .await?; tx.commit().await?; - run_wait_result(authed, Extension(user_db), uuid, Path((w_id, script_hash))).await + run_wait_result( + authed, + Extension(user_db), + timeout.0, + uuid, + Path((w_id, script_hash)), + ) + .await } // a similar function exists on the worker diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index 06c0511ba8..c245754cf8 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -53,6 +53,7 @@ pub struct IsSecure(bool); pub struct CookieDomain(Option); pub struct CloudHosted(bool); pub struct ContentSecurityPolicy(String); +pub struct TimeoutWaitResult(i32); pub use users::delete_expired_items_perdiodically; @@ -91,15 +92,15 @@ pub async fn run_server( .layer(Extension(auth_cache.clone())) .layer(Extension(basic_clients)) .layer(Extension(Arc::new(BaseUrl(base_url.to_string())))) + .layer(Extension(Arc::new(ContentSecurityPolicy( + std::env::var("SERVE_CSP").unwrap_or("".to_owned()), + )))) .layer(Extension(Arc::new(CloudHosted( std::env::var("CLOUD_HOSTED").is_ok(), )))) .layer(Extension(Arc::new(IsSecure( base_url.starts_with("https://"), )))) - .layer(Extension(Arc::new(ContentSecurityPolicy( - std::env::var("SERVE_CSP").unwrap_or("".to_owned()), - )))) .layer(Extension(Arc::new(CookieDomain( std::env::var("COOKIE_DOMAIN").ok(), )))) @@ -114,7 +115,17 @@ pub async fn run_server( "/w/:workspace_id", Router::new() .nest("/scripts", scripts::workspaced_service()) - .nest("/jobs", jobs::workspaced_service()) + .nest( + "/jobs", + jobs::workspaced_service().layer(Extension(Arc::new( + TimeoutWaitResult( + std::env::var("TIMEOUT_WAIT_RESULT") + .ok() + .and_then(|x| x.parse().ok()) + .unwrap_or(20), + ), + ))), + ) .nest( "/users", users::workspaced_service().layer(Extension(argon2.clone())), diff --git a/frontend/src/routes/(root)/(logged)/scripts/get/[...hash]/+page.svelte b/frontend/src/routes/(root)/(logged)/scripts/get/[...hash]/+page.svelte index 4c322a67d0..283a6a0716 100644 --- a/frontend/src/routes/(root)/(logged)/scripts/get/[...hash]/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/scripts/get/[...hash]/+page.svelte @@ -19,11 +19,12 @@ faTrash, faCalendar, faShare, - faSpinner, faGlobe, faCodeFork, faClipboard, - faArrowLeft + faArrowLeft, + faChevronUp, + faChevronDown } from '@fortawesome/free-solid-svg-icons' import Tooltip from '$lib/components/Tooltip.svelte' import ShareModal from '$lib/components/ShareModal.svelte' @@ -51,6 +52,7 @@ import Popover from '$lib/components/Popover.svelte' import ScheduleEditor from '$lib/components/ScheduleEditor.svelte' import { Loader2 } from 'lucide-svelte' + import { slide } from 'svelte/transition' let userSettings: UserSettings let script: Script | undefined @@ -162,6 +164,15 @@ } } let scheduleEditor: ScheduleEditor + + let viewWebhookCommand = false + + let args = undefined + $: curlCommand = `curl -H 'Content-Type: application/json' -H 'Authorization: Bearer $TOKEN -X POST -d '${JSON.stringify( + args + )}' ${$page.url.protocol}//${$page.url.hostname}/api/w/${$workspaceStore}/jobs/run/p/${ + script?.path + }` @@ -239,7 +250,7 @@ {/if} {#if deploymentInProgress} - + Deployment in progress {/if} @@ -345,6 +356,7 @@ bind:this={runForm} runnable={script} runAction={runScript} + bind:args /> {#if !emptyString(script.description)} @@ -446,6 +458,24 @@ {/each} + + {#if viewWebhookCommand} +
+ 
{curlCommand}  copyToClipboard(curlCommand)}
+										class="cursor-pointer ml-2">
+
+ {/if}