starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-07 16:03:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(backend): add queue_limit + configurable timeout + fix timeout cancel

Browse Source
This commit is contained in:
Ruben Fiszel
2023-01-18 09:02:38 +01:00
parent 470fb51d13
commit 40f32e644e
6 changed files with 132 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+1
View File
@@ -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
backend/sqlx-data.json
+18
View File
@@ -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": [
backend/windmill-api/openapi.yaml
+8
View File
@@ -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
backend/windmill-api/src/jobs.rs
+57 -7
View File
@@ -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<Uuid>,
include_header: Option<String>,
invisible_to_owner: Option<bool>,
queue_limit: Option<i64>,
}
impl RunJobQuery {
@@ -1227,9 +1228,18 @@ impl Drop for Guard {
async fn run_wait_result<T>(
authed: Authed,
Extension(user_db): Extension<UserDB>,
timeout: i32,
uuid: Uuid,
Path((w_id, _)): Path<(String, T)>,
) -> error::JsonResult<serde_json::Value> {
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<T>(
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<T>(
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<i64>) -> 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<UserDB>,
Extension(db): Extension<DB>,
Extension(timeout): Extension<Arc<TimeoutWaitResult>>,
Path((w_id, script_path)): Path<(String, StripPath)>,
Query(run_query): Query<RunJobQuery>,
headers: HeaderMap,
Json(args): Json<Option<serde_json::Map<String, serde_json::Value>>>,
) -> error::JsonResult<serde_json::Value> {
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<UserDB>,
Extension(db): Extension<DB>,
Extension(timeout): Extension<Arc<TimeoutWaitResult>>,
Path((w_id, script_hash)): Path<(String, ScriptHash)>,
Query(run_query): Query<RunJobQuery>,
headers: HeaderMap,
Json(args): Json<Option<serde_json::Map<String, serde_json::Value>>>,
) -> error::JsonResult<serde_json::Value> {
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
backend/windmill-api/src/lib.rs
+15 -4
View File
@@ -53,6 +53,7 @@ pub struct IsSecure(bool);
pub struct CookieDomain(Option<String>);
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())),
frontend/src/routes/(root)/(logged)/scripts/get/[...hash]/+page.svelte
+33 -3
View File
@@ -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
}`
</script>
<ScheduleEditor bind:this={scheduleEditor} />
@@ -239,7 +250,7 @@
{/if}
{#if deploymentInProgress}
<Badge color="yellow">
<Loader2 class="animate-spin mr-2" />
<Loader2 size={12} class="inline animate-spin mr-1" />
Deployment in progress
</Badge>
{/if}
@@ -345,6 +356,7 @@
bind:this={runForm}
runnable={script}
runAction={runScript}
bind:args
/>
</div>
{#if !emptyString(script.description)}
@@ -446,6 +458,24 @@
</div>
</TabContent>
{/each}
<Button
color="light"
size="sm"
endIcon={{ icon: viewWebhookCommand ? faChevronUp : faChevronDown }}
on:click={() => (viewWebhookCommand = !viewWebhookCommand)}
>
See example curl command
</Button>
{#if viewWebhookCommand}
<div transition:slide|local class="px-4">
<pre class="bg-gray-700 text-gray-100 p-2 font-mono text-sm whitespace-pre-wrap"
>{curlCommand} <span
on:click={() => copyToClipboard(curlCommand)}
class="cursor-pointer ml-2"><Icon data={faClipboard} /></span
></pre
>
</div>
{/if}
</svelte:fragment>
</Tabs>
</div>