starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-18 16:02:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: Priority queue (#2476)

Browse Source 
* BE v1 for priority queue

* Add FE prototype

* sqlx prepare

* Disply priority badge in job detail view

* sqlx prepare post merge

* Add priority number input field next to toggle

* increase dnd timeout for app events

* Update db migration

* Fix FE import

* Add badge to script/flow details page and fix EE-only mention in tooltip

---------

Co-authored-by: Ruben Fiszel <ruben@rubenfiszel.com>
Co-authored-by: Ruben Fiszel <ruben@windmill.dev>
This commit is contained in:
Guillaume Bouvignies
2023-10-23 14:22:59 +02:00
committed by GitHub
parent 885e44e1c1
commit bd69bafe59
38 changed files with 434 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-27ab808d7e05235180449df043ef57c6fbe7d3e261e5a52313d01782739c0e77.json → backend/.sqlx/query-11ae5e1aade2bc9a78af2d956ece73766ae3330319619a68383b508a76f2d19c.json
Generated
+8 -3
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "select path, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker from script where hash = $1 AND workspace_id = $2",
"query": "select path, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority from script where hash = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
@@ -42,7 +42,6 @@
"bash",
"postgresql",
"nativets",
"Nativets",
"bun",
"mysql",
"bigquery",
@@ -58,6 +57,11 @@
"ordinal": 6,
"name": "dedicated_worker",
"type_info": "Bool"
},
{
"ordinal": 7,
"name": "priority",
"type_info": "Int2"
}
],
"parameters": {
@@ -73,8 +77,9 @@
true,
true,
false,
true,
true
]
},
"hash": "27ab808d7e05235180449df043ef57c6fbe7d3e261e5a52313d01782739c0e77"
"hash": "11ae5e1aade2bc9a78af2d956ece73766ae3330319619a68383b508a76f2d19c"
}
backend/.sqlx/query-25bef6a248f3ee0ea2cbcc376c217cbcf1013ae311c36b42d423bf6a02fa016c.json → backend/.sqlx/query-254776959f60455a00a2c29a1783ee6b27c836197df6454d3bb5c59441c34f44.json
Generated
+4 -4
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO completed_job AS cj\n ( workspace_id\n , id\n , parent_job\n , created_by\n , created_at\n , started_at\n , duration_ms\n , success\n , script_hash\n , script_path\n , args\n , result\n , logs\n , raw_code\n , raw_lock\n , canceled\n , canceled_by\n , canceled_reason\n , job_kind\n , schedule_path\n , permissioned_as\n , flow_status\n , raw_flow\n , is_flow_step\n , is_skipped\n , language\n , email\n , visible_to_owner\n , mem_peak\n , tag\n )\n VALUES ($1, $2, $3, $4, $5, COALESCE($6, now()), COALESCE($26, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE($6, now()))))*1000), $7, $8, $9,$10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $27, $28, $29, $30)\n ON CONFLICT (id) DO UPDATE SET success = $7, result = $11, logs = concat(cj.logs, $12) RETURNING duration_ms",
"query": "INSERT INTO completed_job AS cj\n ( workspace_id\n , id\n , parent_job\n , created_by\n , created_at\n , started_at\n , duration_ms\n , success\n , script_hash\n , script_path\n , args\n , result\n , logs\n , raw_code\n , raw_lock\n , canceled\n , canceled_by\n , canceled_reason\n , job_kind\n , schedule_path\n , permissioned_as\n , flow_status\n , raw_flow\n , is_flow_step\n , is_skipped\n , language\n , email\n , visible_to_owner\n , mem_peak\n , tag\n , priority\n )\n VALUES ($1, $2, $3, $4, $5, COALESCE($6, now()), COALESCE($26, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE($6, now()))))*1000), $7, $8, $9,$10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $27, $28, $29, $30, $31)\n ON CONFLICT (id) DO UPDATE SET success = $7, result = $11, logs = concat(cj.logs, $12) RETURNING duration_ms",
"describe": {
"columns": [
{
@@ -67,7 +67,6 @@
"bash",
"postgresql",
"nativets",
"Nativets",
"bun",
"mysql",
"bigquery",
@@ -82,12 +81,13 @@
"Varchar",
"Bool",
"Int4",
"Varchar"
"Varchar",
"Int2"
]
},
"nullable": [
false
]
},
"hash": "25bef6a248f3ee0ea2cbcc376c217cbcf1013ae311c36b42d423bf6a02fa016c"
"hash": "254776959f60455a00a2c29a1783ee6b27c836197df6454d3bb5c59441c34f44"
}
backend/.sqlx/query-b5646d44502b289e5d92f53f74bf0a52b8bf76775d87a1f698ed2a3c3446c86e.json → backend/.sqlx/query-52ad1773a9f5a1b6be1bb16af408f051f27d3f21b615c468b85aa75b64d943db.json
Generated
+4 -4
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, dedicated_worker, ws_error_handler_muted) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22)",
"query": "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, dedicated_worker, ws_error_handler_muted, priority) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23)",
"describe": {
"columns": [],
"parameters": {
@@ -28,7 +28,6 @@
"bash",
"postgresql",
"nativets",
"Nativets",
"bun",
"mysql",
"bigquery",
@@ -60,10 +59,11 @@
"Int4",
"Int4",
"Bool",
"Bool"
"Bool",
"Int2"
]
},
"nullable": []
},
"hash": "b5646d44502b289e5d92f53f74bf0a52b8bf76775d87a1f698ed2a3c3446c86e"
"hash": "52ad1773a9f5a1b6be1bb16af408f051f27d3f21b615c468b85aa75b64d943db"
}
backend/.sqlx/query-9e8c3ff3d6b31e366e15beda1e96e03e870ccc3b353401439bc0ed8ff219249b.json → backend/.sqlx/query-620ddf29c5e867079df4c2aa6e80bccb19beeb9ddfa308ca97f254cd5ba8157e.json
Generated
+4 -4
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO queue\n (workspace_id, id, running, parent_job, created_by, permissioned_as, scheduled_for, \n 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, timeout, flow_step_id, cache_ttl)\n 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, $28, $29) RETURNING id",
"query": "INSERT INTO queue\n (workspace_id, id, running, parent_job, created_by, permissioned_as, scheduled_for, \n 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, timeout, flow_step_id, cache_ttl, priority)\n 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, $28, $29, $30) RETURNING id",
"describe": {
"columns": [
{
@@ -60,7 +60,6 @@
"bash",
"postgresql",
"nativets",
"Nativets",
"bun",
"mysql",
"bigquery",
@@ -81,12 +80,13 @@
"Int4",
"Int4",
"Varchar",
"Int4"
"Int4",
"Int2"
]
},
"nullable": [
false
]
},
"hash": "9e8c3ff3d6b31e366e15beda1e96e03e870ccc3b353401439bc0ed8ff219249b"
"hash": "620ddf29c5e867079df4c2aa6e80bccb19beeb9ddfa308ca97f254cd5ba8157e"
}
backend/.sqlx/query-6b9ff3fbca9e825c95d14705082a10de88172c0c748a45aba4d2d03c3b58f54d.json
Generated
-1
View File
@@ -46,7 +46,6 @@
"bash",
"postgresql",
"nativets",
"Nativets",
"bun",
"mysql",
"bigquery",
backend/.sqlx/query-9d3556319411a27a875bf6cf0e5eda837cc63e4d8be912c0b5bfeea4a0c8db2e.json
Generated
-1
View File
@@ -40,7 +40,6 @@
"bash",
"postgresql",
"nativets",
"Nativets",
"bun",
"mysql",
"bigquery",
backend/.sqlx/query-b224cdd1221fc9e7227ef8e8c025eedc09bceb82bca349f3e31c8513ebbf0192.json → backend/.sqlx/query-b6074eb9369de071f79165bf506fbc44f1688035e1928b4c3649d875187b1e1d.json
Generated
+8 -3
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker from script where path = $1 AND workspace_id = $2 AND\n created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND\n deleted = false AND lock IS not NULL AND lock_error_logs IS NULL)",
"query": "select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority from script where path = $1 AND workspace_id = $2 AND\n created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND\n deleted = false AND lock IS not NULL AND lock_error_logs IS NULL)",
"describe": {
"columns": [
{
@@ -42,7 +42,6 @@
"bash",
"postgresql",
"nativets",
"Nativets",
"bun",
"mysql",
"bigquery",
@@ -58,6 +57,11 @@
"ordinal": 6,
"name": "dedicated_worker",
"type_info": "Bool"
},
{
"ordinal": 7,
"name": "priority",
"type_info": "Int2"
}
],
"parameters": {
@@ -73,8 +77,9 @@
true,
true,
false,
true,
true
]
},
"hash": "b224cdd1221fc9e7227ef8e8c025eedc09bceb82bca349f3e31c8513ebbf0192"
"hash": "b6074eb9369de071f79165bf506fbc44f1688035e1928b4c3649d875187b1e1d"
}
backend/.sqlx/query-251f18ccdd89255023258a33285757801cabe4e86385e2dea10b666d1c6f2b61.json → backend/.sqlx/query-c7febee7146a28481ce36f3570bceb38751a6527827525a582d9f8804981f552.json
Generated
+8 -3
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker from script where path = $1 AND workspace_id = $2 AND\n created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND\n deleted = false AND archived = false)",
"query": "select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority from script where path = $1 AND workspace_id = $2 AND\n created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND\n deleted = false AND archived = false)",
"describe": {
"columns": [
{
@@ -42,7 +42,6 @@
"bash",
"postgresql",
"nativets",
"Nativets",
"bun",
"mysql",
"bigquery",
@@ -58,6 +57,11 @@
"ordinal": 6,
"name": "dedicated_worker",
"type_info": "Bool"
},
{
"ordinal": 7,
"name": "priority",
"type_info": "Int2"
}
],
"parameters": {
@@ -73,8 +77,9 @@
true,
true,
false,
true,
true
]
},
"hash": "251f18ccdd89255023258a33285757801cabe4e86385e2dea10b666d1c6f2b61"
"hash": "c7febee7146a28481ce36f3570bceb38751a6527827525a582d9f8804981f552"
}
backend/.sqlx/query-901c430cba96e5cab1a186207ea708f3d34a04c8ab03b385f8f2e9c662d3e724.json → backend/.sqlx/query-d48070ef889723f408ed6c1a97b5ab9f0e98340cd33c6e29824f6030f0089071.json
Generated
+8 -3
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "select tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker from script where hash = $1 AND workspace_id = $2",
"query": "select tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority from script where hash = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
@@ -37,7 +37,6 @@
"bash",
"postgresql",
"nativets",
"Nativets",
"bun",
"mysql",
"bigquery",
@@ -53,6 +52,11 @@
"ordinal": 5,
"name": "dedicated_worker",
"type_info": "Bool"
},
{
"ordinal": 6,
"name": "priority",
"type_info": "Int2"
}
],
"parameters": {
@@ -67,8 +71,9 @@
true,
true,
false,
true,
true
]
},
"hash": "901c430cba96e5cab1a186207ea708f3d34a04c8ab03b385f8f2e9c662d3e724"
"hash": "d48070ef889723f408ed6c1a97b5ab9f0e98340cd33c6e29824f6030f0089071"
}
backend/.sqlx/query-f056b5f3e66a764748925f1bfd3180923fde8c7fdf69088d0e4a5555cc049545.json
Generated
-23
View File
@@ -1,23 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT result FROM completed_job WHERE id = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "result",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": [
true
]
},
"hash": "f056b5f3e66a764748925f1bfd3180923fde8c7fdf69088d0e4a5555cc049545"
}
backend/migrations/20231020101153_queue_priority_column.down.sql
+9
View File
@@ -0,0 +1,9 @@
-- Add up migration script here
ALTER TABLE queue DROP COLUMN priority;
DROP INDEX IF EXISTS queue_sort;
CREATE INDEX IF NOT EXISTS queue_sort ON queue (scheduled_for, created_at, tag) WHERE running = false;
DROP INDEX IF EXISTS queue_suspended;
CREATE INDEX IF NOT EXISTS queue_suspended ON queue (created_at, suspend_until, suspend, tag) WHERE suspend_until is not null;
ALTER TABLE completed_job DROP COLUMN priority;
ALTER TABLE script DROP COLUMN priority;
backend/migrations/20231020101153_queue_priority_column.up.sql
+9
View File
@@ -0,0 +1,9 @@
-- Add up migration script here
ALTER TABLE queue ADD COLUMN priority SMALLINT;
DROP INDEX IF EXISTS queue_sort;
CREATE INDEX IF NOT EXISTS queue_sort ON queue (priority DESC NULLS LAST, scheduled_for, created_at, tag) WHERE running = false;
DROP INDEX IF EXISTS queue_suspended;
CREATE INDEX IF NOT EXISTS queue_suspended ON queue (priority DESC NULLS LAST, created_at, suspend_until, suspend, tag) WHERE suspend_until is not null;
ALTER TABLE completed_job ADD COLUMN priority SMALLINT;
ALTER TABLE script ADD COLUMN priority SMALLINT;
backend/tests/worker.rs
+12 -4
View File
@@ -894,6 +894,7 @@ impl RunJob {
None,
None,
None,
None,
)
.await
.expect("push has to succeed");
@@ -1080,6 +1081,7 @@ async fn test_deno_flow(db: Pool<Postgres>) {
cache_ttl: None,
mock: None,
timeout: None,
priority: None,
},
FlowModule {
id: "b".to_string(),
@@ -1114,6 +1116,7 @@ async fn test_deno_flow(db: Pool<Postgres>) {
cache_ttl: None,
mock: None,
timeout: None,
priority: None,
}],
},
stop_after_if: Default::default(),
@@ -1124,6 +1127,7 @@ async fn test_deno_flow(db: Pool<Postgres>) {
cache_ttl: None,
mock: None,
timeout: None,
priority: None,
},
],
same_worker: false,
@@ -1223,7 +1227,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) {
cache_ttl: None,
mock: None,
timeout: None,
priority: None,
},
FlowModule {
id: "b".to_string(),
@@ -1269,6 +1273,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) {
cache_ttl: None,
mock: None,
timeout: None,
priority: None,
},
FlowModule {
id: "e".to_string(),
@@ -1300,7 +1305,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) {
cache_ttl: None,
mock: None,
timeout: None,
priority: None,
},
],
},
@@ -1312,7 +1317,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) {
cache_ttl: None,
mock: None,
timeout: None,
priority: None,
},
FlowModule {
id: "c".to_string(),
@@ -1352,7 +1357,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) {
cache_ttl: None,
mock: None,
timeout: None,
priority: None,
},
],
same_worker: true,
@@ -2565,6 +2570,7 @@ async fn test_flow_lock_all(db: Pool<Postgres>) {
path: "g/all/flow_lock_all".to_owned(),
tag: None,
ws_error_handler_muted: None,
priority: None,
},
draft_only: None,
},
@@ -2919,6 +2925,7 @@ async fn run_deployed_relative_imports(db: &Pool<Postgres>, script_content: Stri
tag: None,
schema: std::collections::HashMap::new(),
ws_error_handler_muted: Some(false),
priority: None,
},
).await.unwrap();
@@ -2941,6 +2948,7 @@ async fn run_deployed_relative_imports(db: &Pool<Postgres>, script_content: Stri
cache_ttl: None,
dedicated_worker: None,
language,
priority: None,
}).push(&db2).await;
backend/windmill-api/openapi-deref.yaml
+16
View File
@@ -3457,6 +3457,8 @@ paths:
user_groups_required:
oneOf: *ref_23
discriminator: *ref_24
priority:
type: number
retry:
type: object
properties: &ref_175
@@ -3491,6 +3493,8 @@ paths:
type: string
cache_ttl:
type: number
priority:
type: number
required: &ref_48
- modules
schema:
@@ -3990,6 +3994,8 @@ paths:
type: boolean
ws_error_handler_muted:
type: boolean
priority:
type: integer
required: &ref_30
- hash
- path
@@ -4176,6 +4182,8 @@ paths:
type: boolean
ws_error_handler_muted:
type: boolean
priority:
type: integer
required: &ref_33
- path
- summary
@@ -5050,6 +5058,8 @@ paths:
type: string
ws_error_handler_muted:
type: boolean
priority:
type: integer
required: &ref_143
- path
- edited_by
@@ -5195,6 +5205,8 @@ paths:
type: string
ws_error_handler_muted:
type: boolean
priority:
type: integer
required:
- path
- type: object
@@ -6641,6 +6653,8 @@ paths:
type: integer
tag:
type: string
priority:
type: integer
required: &ref_68
- id
- running
@@ -6908,6 +6922,8 @@ paths:
type: integer
tag:
type: string
priority:
type: integer
required: &ref_66
- id
- created_by
backend/windmill-api/openapi.yaml
+12
View File
@@ -6503,6 +6503,8 @@ components:
type: boolean
ws_error_handler_muted:
type: boolean
priority:
type: integer
required:
- hash
- path
@@ -6582,6 +6584,8 @@ components:
type: boolean
ws_error_handler_muted:
type: boolean
priority:
type: integer
required:
- path
- summary
@@ -6756,6 +6760,8 @@ components:
type: integer
tag:
type: string
priority:
type: integer
required:
- id
- running
@@ -6863,6 +6869,8 @@ components:
type: integer
tag:
type: string
priority:
type: integer
required:
- id
- created_by
@@ -7829,6 +7837,8 @@ components:
type: string
ws_error_handler_muted:
type: boolean
priority:
type: integer
required:
- path
- edited_by
@@ -7847,6 +7857,8 @@ components:
type: string
ws_error_handler_muted:
type: boolean
priority:
type: integer
required:
- path
backend/windmill-api/src/apps.rs
+3
View File
@@ -534,6 +534,7 @@ async fn create_app(
None,
None,
None,
None,
)
.await?;
tracing::info!("Pushed app dependency job {}", dependency_job_uuid);
@@ -777,6 +778,7 @@ async fn update_app(
None,
None,
None,
None,
)
.await?;
tracing::info!("Pushed app dependency job {}", dependency_job_uuid);
@@ -938,6 +940,7 @@ async fn execute_component(
tag,
None,
None,
None,
)
.await?;
tx.commit().await?;
backend/windmill-api/src/flows.rs
+8 -1
View File
@@ -225,7 +225,7 @@ async fn toggle_workspace_error_handler(
Json(req): Json<ToggleWorkspaceErrorHandler>,
) -> Result<String> {
#[cfg(not(feature = "enterprise"))]
if true {
{
return Err(Error::BadRequest(
"Muting the error handler for certain flow is only available in enterprise version"
.to_string(),
@@ -375,6 +375,7 @@ async fn create_flow(
nf.tag,
None,
None,
None,
)
.await?;
@@ -580,6 +581,7 @@ async fn update_flow(
None,
None,
None,
None,
)
.await?;
sqlx::query!(
@@ -805,6 +807,7 @@ mod tests {
cache_ttl: None,
mock: None,
timeout: None,
priority: None,
},
FlowModule {
id: "b".to_string(),
@@ -829,6 +832,7 @@ mod tests {
cache_ttl: None,
mock: None,
timeout: None,
priority: None,
},
FlowModule {
id: "c".to_string(),
@@ -850,6 +854,7 @@ mod tests {
cache_ttl: None,
mock: None,
timeout: None,
priority: None,
},
],
failure_module: Some(FlowModule {
@@ -870,6 +875,7 @@ mod tests {
cache_ttl: None,
mock: None,
timeout: None,
priority: None,
}),
same_worker: false,
concurrent_limit: None,
@@ -877,6 +883,7 @@ mod tests {
skip_expr: None,
cache_ttl: None,
ws_error_handler_muted: None,
priority: None,
};
let expect = serde_json::json!({
"modules": [
backend/windmill-api/src/jobs.rs
+32 -5
View File
@@ -315,9 +315,10 @@ pub async fn get_path_tag_limits_cache_for_hash(
Option<i32>,
ScriptLang,
Option<bool>,
Option<i16>,
)> {
let script = sqlx::query!(
"select path, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker from script where hash = $1 AND workspace_id = $2",
"select path, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority from script where hash = $1 AND workspace_id = $2",
hash,
w_id
)
@@ -336,6 +337,7 @@ pub async fn get_path_tag_limits_cache_for_hash(
script.cache_ttl,
script.language,
script.dedicated_worker,
script.priority,
))
}
@@ -352,7 +354,7 @@ async fn get_job_internal(db: &DB, workspace_id: &str, job_id: Uuid) -> error::R
id, workspace_id, parent_job, created_by, created_at, duration_ms, success, script_hash, script_path,
CASE WHEN pg_column_size(args) < 2000000 THEN args ELSE '{\"reason\": \"WINDMILL_TOO_BIG\"}'::jsonb END as args, CASE WHEN pg_column_size(result) < 2000000 THEN result ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as result, logs, deleted, raw_code, canceled, canceled_by, canceled_reason, job_kind, env_id,
schedule_path, permissioned_as, flow_status, raw_flow, is_flow_step, language, started_at, is_skipped,
raw_lock, email, visible_to_owner, mem_peak, tag
raw_lock, email, visible_to_owner, mem_peak, tag, priority
FROM completed_job WHERE id = $1 AND workspace_id = $2")
.bind(job_id)
.bind(workspace_id)
@@ -367,7 +369,7 @@ async fn get_job_internal(db: &DB, workspace_id: &str, job_id: Uuid) -> error::R
script_hash, script_path, CASE WHEN pg_column_size(args) < 2000000 THEN args ELSE '{\"reason\": \"WINDMILL_TOO_BIG\"}'::jsonb END as args, logs, raw_code, canceled, canceled_by, canceled_reason, last_ping,
job_kind, env_id, schedule_path, permissioned_as, flow_status, raw_flow, is_flow_step, language,
suspend, suspend_until, same_worker, raw_lock, pre_run_error, email, visible_to_owner, mem_peak,
root_job, leaf_jobs, tag, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id, cache_ttl
root_job, leaf_jobs, tag, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id, cache_ttl, priority
FROM queue WHERE id = $1 AND workspace_id = $2",
)
.bind(job_id)
@@ -441,6 +443,8 @@ pub struct CompletedJob {
#[serde(skip_serializing_if = "Option::is_none")]
pub mem_peak: Option<i32>,
pub tag: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub priority: Option<i16>,
}
impl CompletedJob {
@@ -493,6 +497,8 @@ pub struct ListableCompletedJob {
#[serde(skip_serializing_if = "Option::is_none")]
pub mem_peak: Option<i32>,
pub tag: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub priority: Option<i16>,
}
impl<'a> IntoResponse for CompletedJob {
@@ -646,6 +652,7 @@ struct ListableQueuedJob {
pub email: String,
pub suspend: Option<i32>,
pub tag: String,
pub priority: Option<i16>,
}
async fn list_queue_jobs(
@@ -789,6 +796,7 @@ async fn list_jobs(
"tag",
"null as concurrent_limit",
"null as concurrency_time_window_s",
"priority",
],
))
} else {
@@ -849,6 +857,7 @@ async fn list_jobs(
"tag",
"concurrent_limit",
"concurrency_time_window_s",
"priority",
],
);
@@ -1247,7 +1256,7 @@ fn conditionally_require_authed_user(
.to_string(),
));
#[cfg(feature = "enterprise")]
if true {
{
for required_group in approval_conditions.user_groups_required.iter() {
if authed.as_ref().unwrap().groups.contains(&required_group) {
return Ok(());
@@ -1407,6 +1416,7 @@ struct UnifiedJob {
tag: String,
concurrent_limit: Option<i32>,
concurrency_time_window_s: Option<i32>,
priority: Option<i16>,
}
impl<'a> From<UnifiedJob> for Job {
@@ -1443,6 +1453,7 @@ impl<'a> From<UnifiedJob> for Job {
visible_to_owner: uj.visible_to_owner,
mem_peak: uj.mem_peak,
tag: uj.tag,
priority: uj.priority,
}),
"QueuedJob" => Job::QueuedJob(QueuedJob {
workspace_id: uj.workspace_id,
@@ -1484,6 +1495,7 @@ impl<'a> From<UnifiedJob> for Job {
timeout: None,
flow_step_id: None,
cache_ttl: None,
priority: uj.priority,
}),
t => panic!("job type {} not valid", t),
}
@@ -1666,6 +1678,7 @@ pub async fn run_flow_by_path(
tag,
None,
None,
None,
)
.await?;
tx.commit().await?;
@@ -1715,6 +1728,7 @@ pub async fn run_job_by_path(
tag,
None,
None,
None,
)
.await?;
tx.commit().await?;
@@ -1928,6 +1942,7 @@ pub async fn run_wait_result_job_by_path_get(
tag,
None,
None,
None,
)
.await?;
tx.commit().await?;
@@ -2036,6 +2051,7 @@ async fn run_wait_result_script_by_path_internal(
tag,
None,
None,
None,
)
.await?;
tx.commit().await?;
@@ -2066,6 +2082,7 @@ pub async fn run_wait_result_script_by_hash(
cache_ttl,
language,
dedicated_worker,
priority,
) = get_path_tag_limits_cache_for_hash(&db, &w_id, hash).await?;
check_scopes(&authed, || format!("run:script/{path}"))?;
@@ -2084,6 +2101,7 @@ pub async fn run_wait_result_script_by_hash(
cache_ttl,
language,
dedicated_worker,
priority,
},
args,
&authed.username,
@@ -2101,6 +2119,7 @@ pub async fn run_wait_result_script_by_hash(
tag,
None,
None,
None,
)
.await?;
tx.commit().await?;
@@ -2175,6 +2194,7 @@ async fn run_wait_result_flow_by_path_internal(
tag,
None,
None,
None,
)
.await?;
tx.commit().await?;
@@ -2237,6 +2257,7 @@ async fn run_preview_job(
preview.tag,
None,
None,
None,
)
.await?;
tx.commit().await?;
@@ -2315,6 +2336,7 @@ async fn add_batch_jobs(
None,
None,
None,
None,
)
.await?;
tx = PushIsolationLevel::Transaction(ntx);
@@ -2415,6 +2437,7 @@ async fn run_preview_flow_job(
raw_flow.tag,
None,
None,
None,
)
.await?;
tx.commit().await?;
@@ -2444,6 +2467,7 @@ pub async fn run_job_by_hash(
cache_ttl,
language,
dedicated_worker,
priority,
) = get_path_tag_limits_cache_for_hash(&db, &w_id, hash).await?;
check_scopes(&authed, || format!("run:script/{path}"))?;
@@ -2464,6 +2488,7 @@ pub async fn run_job_by_hash(
cache_ttl,
language,
dedicated_worker,
priority,
},
args,
&authed.username,
@@ -2481,6 +2506,7 @@ pub async fn run_job_by_hash(
tag,
None,
None,
None,
)
.await?;
tx.commit().await?;
@@ -2696,6 +2722,7 @@ async fn list_completed_jobs(
"visible_to_owner",
"mem_peak",
"tag",
"priority",
"'CompletedJob' as type",
],
)
@@ -2713,7 +2740,7 @@ async fn get_completed_job<'a>(
let job_o = sqlx::query("SELECT id, workspace_id, parent_job, created_by, created_at, duration_ms, success, script_hash, script_path,
CASE WHEN pg_column_size(args) < 2000000 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args, CASE WHEN pg_column_size(result) < 2000000 THEN result ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as result, logs, deleted, raw_code, canceled, canceled_by, canceled_reason, job_kind, env_id,
schedule_path, permissioned_as, flow_status, raw_flow, is_flow_step, language, started_at, is_skipped,
raw_lock, email, visible_to_owner, mem_peak, tag FROM completed_job WHERE id = $1 AND workspace_id = $2")
raw_lock, email, visible_to_owner, mem_peak, tag, priority FROM completed_job WHERE id = $1 AND workspace_id = $2")
.bind(id)
.bind(w_id)
.fetch_optional(&db)
backend/windmill-api/src/oauth2.rs
+3
View File
@@ -871,6 +871,7 @@ async fn slack_command(
cache_ttl,
language,
dedicated_worker,
priority,
) = windmill_common::get_latest_deployed_hash_for_path(
&db,
&settings.workspace_id,
@@ -886,6 +887,7 @@ async fn slack_command(
cache_ttl,
language,
dedicated_worker,
priority,
},
tag,
)
@@ -919,6 +921,7 @@ async fn slack_command(
tag,
None,
None,
None,
)
.await?;
let url = BASE_URL.read().await.clone();
backend/windmill-api/src/scripts.rs
+10 -4
View File
@@ -77,7 +77,10 @@ pub struct ScriptWDraft {
pub cache_ttl: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub dedicated_worker: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub ws_error_handler_muted: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub priority: Option<i16>,
}
pub fn global_service() -> Router {
@@ -459,8 +462,9 @@ async fn create_script(
sqlx::query!(
"INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, \
content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, \
draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, dedicated_worker, ws_error_handler_muted) \
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22)",
draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, \
dedicated_worker, ws_error_handler_muted, priority) \
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23)",
&w_id,
&hash.0,
ns.path,
@@ -483,6 +487,7 @@ async fn create_script(
ns.cache_ttl,
ns.dedicated_worker,
ns.ws_error_handler_muted.unwrap_or(false),
ns.priority,
)
.execute(&mut tx)
.await?;
@@ -614,6 +619,7 @@ async fn create_script(
ns.tag,
None,
None,
None,
)
.await?;
tx = PushIsolationLevel::Transaction(new_tx);
@@ -679,7 +685,7 @@ async fn get_script_by_path_w_draft(
let mut tx = user_db.begin(&authed).await?;
let script_o = sqlx::query_as::<_, ScriptWDraft>(
"SELECT hash, script.path, summary, description, content, language, kind, tag, schema, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, ws_error_handler_muted, draft.value as draft, dedicated_worker FROM script LEFT JOIN draft ON
"SELECT hash, script.path, summary, description, content, language, kind, tag, schema, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, ws_error_handler_muted, draft.value as draft, dedicated_worker, priority FROM script LEFT JOIN draft ON
script.path = draft.path AND script.workspace_id = draft.workspace_id AND draft.typ = 'script'
WHERE script.path = $1 AND script.workspace_id = $2 \
AND script.created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND \
@@ -724,7 +730,7 @@ async fn toggle_workspace_error_handler(
Json(req): Json<ToggleWorkspaceErrorHandler>,
) -> Result<String> {
#[cfg(not(feature = "enterprise"))]
if true {
{
return Err(Error::BadRequest(
"Muting the error handler for certain script is only available in enterprise version"
.to_string(),
backend/windmill-api/src/workspaces.rs
+2 -1
View File
@@ -555,7 +555,8 @@ async fn run_slack_message_test_job(
Some(json!(extra_args)),
authed.username.as_str(),
authed.email.as_str(),
username_to_permissioned_as(authed.username.as_str())
username_to_permissioned_as(authed.username.as_str()),
None, // Note: we could mark it as high priority to return result quickly to the user
).await?;
tx.commit().await?;
backend/windmill-common/src/flows.rs
+6
View File
@@ -90,6 +90,9 @@ pub struct FlowValue {
pub cache_ttl: Option<u32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub ws_error_handler_muted: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
// Priority at the flow level
pub priority: Option<i16>,
}
#[derive(Deserialize, Serialize, Debug, Clone)]
@@ -204,6 +207,9 @@ pub struct FlowModule {
pub cache_ttl: Option<u32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub timeout: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
// Priority at the flow step level
pub priority: Option<i16>,
}
impl FlowModule {
backend/windmill-common/src/jobs.rs
+9 -1
View File
@@ -95,6 +95,8 @@ pub struct QueuedJob {
pub flow_step_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub cache_ttl: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub priority: Option<i16>,
}
impl QueuedJob {
@@ -179,6 +181,7 @@ impl Default for QueuedJob {
timeout: None,
flow_step_id: None,
cache_ttl: None,
priority: None,
}
}
}
@@ -196,6 +199,7 @@ pub enum JobPayload {
cache_ttl: Option<i32>,
dedicated_worker: Option<bool>,
language: ScriptLang,
priority: Option<i16>,
},
Code(RawCode),
Dependencies {
@@ -251,6 +255,7 @@ pub async fn script_path_to_payload(
cache_ttl,
language,
dedicated_worker,
priority,
) = get_latest_deployed_hash_for_path(db, w_id, script_path).await?;
(
JobPayload::ScriptHash {
@@ -261,6 +266,7 @@ pub async fn script_path_to_payload(
cache_ttl: cache_ttl,
language,
dedicated_worker,
priority,
},
tag,
)
@@ -279,9 +285,10 @@ pub async fn script_hash_to_tag_and_limits<'c>(
Option<i32>,
ScriptLang,
Option<bool>,
Option<i16>,
)> {
let script = sqlx::query!(
"select tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker from script where hash = $1 AND workspace_id = $2",
"select tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority from script where hash = $1 AND workspace_id = $2",
script_hash.0,
w_id
)
@@ -299,6 +306,7 @@ pub async fn script_hash_to_tag_and_limits<'c>(
script.cache_ttl,
script.language,
script.dedicated_worker,
script.priority,
))
}
backend/windmill-common/src/lib.rs
+6 -2
View File
@@ -198,9 +198,10 @@ pub async fn get_latest_deployed_hash_for_path(
Option<i32>,
ScriptLang,
Option<bool>,
Option<i16>,
)> {
let r_o = sqlx::query!(
"select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker from script where path = $1 AND workspace_id = $2 AND
"select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority from script where path = $1 AND workspace_id = $2 AND
created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND
deleted = false AND lock IS not NULL AND lock_error_logs IS NULL)",
script_path,
@@ -219,6 +220,7 @@ pub async fn get_latest_deployed_hash_for_path(
script.cache_ttl,
script.language,
script.dedicated_worker,
script.priority,
))
}
@@ -234,9 +236,10 @@ pub async fn get_latest_hash_for_path<'c>(
Option<i32>,
ScriptLang,
Option<bool>,
Option<i16>,
)> {
let r_o = sqlx::query!(
"select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker from script where path = $1 AND workspace_id = $2 AND
"select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority from script where path = $1 AND workspace_id = $2 AND
created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND
deleted = false AND archived = false)",
script_path,
@@ -255,5 +258,6 @@ pub async fn get_latest_hash_for_path<'c>(
script.cache_ttl,
script.language,
script.dedicated_worker,
script.priority,
))
}
backend/windmill-common/src/scripts.rs
+4
View File
@@ -162,7 +162,10 @@ pub struct Script {
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrency_time_window_s: Option<i32>,
pub dedicated_worker: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub ws_error_handler_muted: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub priority: Option<i16>,
}
#[derive(Serialize)]
@@ -220,6 +223,7 @@ pub struct NewScript {
pub cache_ttl: Option<i32>,
pub dedicated_worker: Option<bool>,
pub ws_error_handler_muted: Option<bool>,
pub priority: Option<i16>,
}
#[derive(Deserialize)]
backend/windmill-queue/src/jobs.rs
+49 -5
View File
@@ -319,9 +319,10 @@ pub async fn add_completed_job<
, visible_to_owner
, mem_peak
, tag
, priority
)
VALUES ($1, $2, $3, $4, $5, COALESCE($6, now()), COALESCE($26, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE($6, now()))))*1000), $7, $8, $9,\
$10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $27, $28, $29, $30)
$10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $27, $28, $29, $30, $31)
ON CONFLICT (id) DO UPDATE SET success = $7, result = $11, logs = concat(cj.logs, $12) RETURNING duration_ms",
queued_job.workspace_id,
queued_job.id,
@@ -353,6 +354,7 @@ pub async fn add_completed_job<
queued_job.visible_to_owner,
if mem_peak > 0 { Some(mem_peak) } else { None },
queued_job.tag,
queued_job.priority,
)
.fetch_one(&mut tx)
.await
@@ -377,6 +379,7 @@ pub async fn add_completed_job<
result,
job_id,
queued_job.started_at.unwrap_or(chrono::Utc::now()),
queued_job.priority,
)
.await?;
}
@@ -531,6 +534,7 @@ pub async fn run_error_handler<
tag,
None,
None,
None,
)
.await?;
tx.commit().await?;
@@ -720,6 +724,7 @@ async fn apply_schedule_handlers<
result: Json<&'a T>,
job_id: Uuid,
started_at: DateTime<Utc>,
job_priority: Option<i16>,
) -> windmill_common::error::Result<QueueTransaction<'c, R>> {
let schedule = get_schedule_opt(tx.transaction_mut(), w_id, schedule_path).await?;
@@ -776,6 +781,7 @@ async fn apply_schedule_handlers<
&schedule.email,
&schedule_to_user(&schedule.path),
username_to_permissioned_as(&schedule.edited_by),
job_priority,
)
.await;
@@ -894,6 +900,7 @@ pub async fn handle_on_failure<
username: &str,
email: &str,
permissioned_as: String,
priority: Option<i16>,
) -> windmill_common::error::Result<(Uuid, QueueTransaction<'c, R>)> {
let (payload, tag) = get_payload_tag_from_prefixed_path(on_failure_path, db, w_id).await?;
@@ -938,6 +945,7 @@ pub async fn handle_on_failure<
tag,
None,
None,
priority,
)
.await?;
tracing::info!(
@@ -1031,6 +1039,7 @@ async fn handle_on_recovery<
tag,
None,
None,
None,
)
.await?;
tracing::info!(
@@ -1311,7 +1320,7 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<
SELECT id
FROM queue
WHERE suspend_until IS NOT NULL AND (suspend <= 0 OR suspend_until <= now()) AND tag = ANY($1)
ORDER BY created_at
ORDER BY priority DESC NULLS LAST, created_at
FOR UPDATE SKIP LOCKED
LIMIT 1
)
@@ -1343,7 +1352,7 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<
SELECT id
FROM queue
WHERE running = false AND scheduled_for <= now() AND tag = ANY($1)
ORDER BY scheduled_for, created_at
ORDER BY priority DESC NULLS LAST, scheduled_for, created_at
FOR UPDATE SKIP LOCKED
LIMIT 1
)
@@ -1700,6 +1709,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
mut tag: Option<String>,
custom_timeout: Option<i32>,
flow_step_id: Option<String>,
priority_override: Option<i16>,
) -> Result<(Uuid, QueueTransaction<'c, R>), Error> {
#[cfg(feature = "enterprise")]
{
@@ -1801,6 +1811,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
concurrency_time_window_s,
cache_ttl,
dedicated_worker,
low_level_priority,
) = match job_payload {
JobPayload::ScriptHash {
hash,
@@ -1810,6 +1821,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
cache_ttl,
language,
dedicated_worker,
priority,
} => (
Some(hash.0),
Some(path),
@@ -1821,6 +1833,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
concurrency_time_window_s,
cache_ttl,
dedicated_worker,
priority,
),
JobPayload::ScriptHub { path } => {
(
@@ -1835,6 +1848,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
None,
None,
None,
None,
)
}
JobPayload::Code(RawCode {
@@ -1856,6 +1870,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
concurrency_time_window_s,
cache_ttl,
None,
None,
),
JobPayload::Dependencies { hash, dependencies, language, path } => (
Some(hash.0),
@@ -1868,6 +1883,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
None,
None,
None,
None,
),
JobPayload::FlowDependencies { path } => {
let value_json = fetch_scalar_isolated!(
@@ -1895,6 +1911,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
None,
None,
None,
None,
)
}
JobPayload::AppDependencies { path, version } => (
@@ -1908,6 +1925,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
None,
None,
None,
None,
),
JobPayload::RawFlow { value, path } => (
None,
@@ -1920,6 +1938,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
value.concurrency_time_window_s,
value.cache_ttl.map(|x| x as i32),
None,
value.priority,
),
JobPayload::Flow(flow) => {
let value_json = fetch_scalar_isolated!(
@@ -1947,6 +1966,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
value.concurrency_time_window_s,
value.cache_ttl.map(|x| x as i32),
None,
value.priority,
)
}
JobPayload::Identity => (
@@ -1960,6 +1980,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
None,
None,
None,
None,
),
JobPayload::Noop => (
None,
@@ -1972,9 +1993,29 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
None,
None,
None,
None,
),
};
let final_priority: Option<i16>;
#[cfg(not(feature = "enterprise"))]
{
// priority is only available on EE. Do not compute it on CE
final_priority = None;
}
#[cfg(feature = "enterprise")]
{
final_priority = if *CLOUD_HOSTED {
// for cloud hosted instance, priority queues is disabled
None
} else if priority_override.is_some() {
priority_override
} else {
// else it takes the priority defined at the script/flow level, if it's a script or flow
low_level_priority
}; // else it remains empty, i.e. no priority
}
let is_running = same_worker;
if let Some(flow) = raw_flow.as_ref() {
same_worker = same_worker || flow.same_worker;
@@ -2015,6 +2056,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
suspend: None,
cache_ttl: None,
timeout: None,
priority: None,
});
raw_flow = Some(FlowValue { modules, ..flow.clone() });
}
@@ -2092,8 +2134,8 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
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, timeout, \
flow_step_id, cache_ttl)
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, $28, $29) \
flow_step_id, cache_ttl, priority)
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, $28, $29, $30) \
RETURNING id",
workspace_id,
job_id,
@@ -2124,10 +2166,12 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
custom_timeout,
flow_step_id,
cache_ttl,
final_priority,
)
.fetch_one(&mut tx)
.await
.map_err(|e| Error::InternalErr(format!("Could not insert into queue {job_id}: {e}")))?;
// TODO: technically the job isn't queued yet, as the transaction can be rolled back. Should be solved when moving these metrics to the queue abstraction.
if *METRICS_ENABLED {
QUEUE_PUSH_COUNT.inc();
backend/windmill-queue/src/schedule.rs
+3
View File
@@ -82,6 +82,7 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
cache_ttl,
language,
dedicated_worker,
priority,
) = windmill_common::get_latest_hash_for_path(
tx.transaction_mut(),
&schedule.workspace_id,
@@ -97,6 +98,7 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
cache_ttl: cache_ttl,
dedicated_worker,
language,
priority,
},
tag,
)
@@ -131,6 +133,7 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
tag,
None,
None,
None,
)
.await?;
Ok(tx) // TODO: Bubble up pushed UUID from here
backend/windmill-worker/src/worker.rs
+1
View File
@@ -1414,6 +1414,7 @@ async fn queue_init_bash_maybe<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
None,
None,
None,
None,
)
.await?;
inner_tx.commit().await?;
backend/windmill-worker/src/worker_flow.rs
+20
View File
@@ -1642,6 +1642,18 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
ContinuePayload::ForloopJobs { payload, .. } => payload.clone(),
};
// compute job-to-be-pushed priority
// The job definition itself might have its own priority, but as we're running
// it from a flow here, it inherits first the flow step priority and second the
// flow priority.
let new_job_priority_override = if module.priority.is_some() {
module.priority
} else if flow_job.priority.is_some() {
flow_job.priority
} else {
None
};
let transform_inp;
let args = match &next_status {
NextStatus::AllFlowJobs {
@@ -1742,6 +1754,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
},
module.timeout,
Some(module.id.clone()),
new_job_priority_override,
)
.await?;
@@ -2176,6 +2189,7 @@ async fn compute_next_flow_transform(
skip_expr: None,
cache_ttl: None,
ws_error_handler_muted: None,
priority: None,
},
path: inner_path,
},
@@ -2234,6 +2248,7 @@ async fn compute_next_flow_transform(
skip_expr: None,
cache_ttl: None,
ws_error_handler_muted: None,
priority: None,
},
path: Some(format!("{}/forloop", flow_job.script_path())),
},
@@ -2326,6 +2341,7 @@ async fn compute_next_flow_transform(
skip_expr: None,
cache_ttl: None,
ws_error_handler_muted: None,
priority: None,
},
path: Some(format!(
"{}/branchone-{}",
@@ -2371,6 +2387,7 @@ async fn compute_next_flow_transform(
skip_expr: None,
cache_ttl: None,
ws_error_handler_muted: None,
priority: None,
},
path: Some(format!(
"{}/branchall-{}",
@@ -2433,6 +2450,7 @@ async fn compute_next_flow_transform(
skip_expr: None,
cache_ttl: None,
ws_error_handler_muted: None,
priority: None,
},
path: Some(format!(
"{}/branchall-{}",
@@ -2496,6 +2514,7 @@ async fn script_to_payload(
cache_ttl,
language,
dedicated_worker,
priority,
) = script_hash_to_tag_and_limits(&hash, &mut tx, &flow_job.workspace_id).await?;
(
JobPayload::ScriptHash {
@@ -2506,6 +2525,7 @@ async fn script_to_payload(
cache_ttl: module.cache_ttl.map(|x| x as i32).ok_or(cache_ttl).ok(),
language,
dedicated_worker,
priority,
},
tag,
)
frontend/src/lib/components/ScriptBuilder.svelte
+49 -2
View File
@@ -192,7 +192,8 @@
concurrent_limit: script.concurrent_limit,
concurrency_time_window_s: script.concurrency_time_window_s,
cache_ttl: script.cache_ttl,
ws_error_handler_muted: script.ws_error_handler_muted
ws_error_handler_muted: script.ws_error_handler_muted,
priority: script.priority
}
})
history.replaceState(history.state, '', `/scripts/edit/${script.path}`)
@@ -236,7 +237,8 @@
concurrent_limit: script.concurrent_limit,
concurrency_time_window_s: script.concurrency_time_window_s,
cache_ttl: script.cache_ttl,
ws_error_handler_muted: script.ws_error_handler_muted
ws_error_handler_muted: script.ws_error_handler_muted,
priority: script.priority
}
})
}
@@ -591,6 +593,51 @@
>
</svelte:fragment>
</Section>
{#if !isCloudHosted()}
<Section label="High priority script">
<Toggle
disabled={!$enterpriseLicense || isCloudHosted()}
size="sm"
checked={script.priority !== undefined && script.priority > 0}
on:change={() => {
if (script.priority) {
script.priority = undefined
} else {
script.priority = 100
}
}}
options={{
right: 'Label as high priority'
}}
>
<svelte:fragment slot="right">
<input
type="number"
class="!w-14 ml-4"
disabled={script.priority === undefined}
bind:value={script.priority}
on:focus
on:change={() => {
if (script.priority && script.priority > 100) {
script.priority = 100
} else if (script.priority && script.priority < 0) {
script.priority = 0
}
}}
/>
</svelte:fragment>
</Toggle>
<svelte:fragment slot="header">
<!-- TODO: Add EE-only badge when we have it -->
<Tooltip>
Jobs from script labeled as high priority take precedence over the other
jobs when in the jobs queue.
{#if !$enterpriseLicense}This is a feature only available on enterprise
edition.{/if}
</Tooltip>
</svelte:fragment>
</Section>
{/if}
{#if !isCloudHosted()}
<Section label="Custom env variables">
<svelte:fragment slot="header">
frontend/src/lib/components/Toggle.svelte
+1
View File
@@ -84,5 +84,6 @@
{/if}
</span>
{/if}
<slot name="right" />
</label>
</span>
frontend/src/lib/components/flows/content/FlowModuleComponent.svelte
+44
View File
@@ -5,6 +5,7 @@
import Editor from '$lib/components/Editor.svelte'
import EditorBar from '$lib/components/EditorBar.svelte'
import ModulePreview from '$lib/components/ModulePreview.svelte'
import Toggle from '$lib/components/Toggle.svelte'
import { createScriptFromInlineScript, fork } from '$lib/components/flows/flowStateUtils'
import { RawScript, type FlowModule, Script } from '$lib/gen'
@@ -40,6 +41,8 @@
import s3Scripts from './s3Scripts/lib'
import type { FlowCopilotContext } from '$lib/components/copilot/flow'
import Label from '$lib/components/Label.svelte'
import { enterpriseLicense } from '$lib/stores'
import { isCloudHosted } from '$lib/cloud'
import { loadSchemaFromModule } from '../flowInfers'
const { selectedId, previewArgs, flowStateStore, flowStore, saveDraft } =
@@ -331,6 +334,7 @@
<Tab value="mock">Mock</Tab>
<Tab value="same_worker">Shared Directory</Tab>
<Tab value="timeout">Timeout</Tab>
<Tab value="priority">Priority</Tab>
{#if flowModule.value['language'] === 'python3' || flowModule.value['language'] === 'deno'}
<Tab value="s3">S3</Tab>
{/if}
@@ -409,6 +413,46 @@
Set shared directory in the flow settings
</Button>
</div>
{:else if advancedSelected === 'priority'}
<Section label="Priority" class="flex flex-col gap-4">
<!-- TODO: Add EE-only badge when we have it -->
<Toggle
disabled={!$enterpriseLicense || isCloudHosted()}
checked={flowModule.priority !== undefined && flowModule.priority > 0}
on:change={() => {
if (flowModule.priority) {
flowModule.priority = undefined
} else {
flowModule.priority = 100
}
}}
options={{
right: 'High priority flow step',
rightTooltip: `Jobs scheduled from this step when the flow is executed are labeled as high priority and take precedence over the other jobs in the jobs queue. ${
!$enterpriseLicense
? 'This is a feature only available on enterprise edition.'
: ''
}`
}}
>
<svelte:fragment slot="right">
<input
type="number"
class="!w-14 ml-4"
disabled={flowModule.priority === undefined}
bind:value={flowModule.priority}
on:focus
on:change={() => {
if (flowModule.priority && flowModule.priority > 100) {
flowModule.priority = 100
} else if (flowModule.priority && flowModule.priority < 0) {
flowModule.priority = 0
}
}}
/>
</svelte:fragment>
</Toggle>
</Section>
{:else if advancedSelected === 's3'}
<div>
<h2 class="pb-4">
frontend/src/lib/components/flows/content/FlowSettings.svelte
+40 -1
View File
@@ -13,7 +13,8 @@
import type { FlowEditorContext } from '../types'
import autosize from 'svelte-autosize'
import Slider from '$lib/components/Slider.svelte'
import { workerTags, workspaceStore } from '$lib/stores'
import { enterpriseLicense, workerTags, workspaceStore } from '$lib/stores'
import { isCloudHosted } from '$lib/cloud'
import { copyToClipboard } from '$lib/utils'
import { Icon } from 'svelte-awesome'
import { faClipboard } from '@fortawesome/free-solid-svg-icons'
@@ -113,6 +114,44 @@
/>
</Label>
<!-- TODO: Add EE-only badge when we have it -->
<Toggle
disabled={!$enterpriseLicense || isCloudHosted()}
checked={$flowStore.value.priority !== undefined && $flowStore.value.priority > 0}
on:change={() => {
if ($flowStore.value.priority) {
$flowStore.value.priority = undefined
} else {
$flowStore.value.priority = 100
}
}}
options={{
right: `Label as high priority`,
rightTooltip: `All jobs scheduled by flows labeled as high priority take precedence over the other jobs in the jobs queue. ${
!$enterpriseLicense
? 'This is a feature only available on enterprise edition.'
: ''
}`
}}
>
<svelte:fragment slot="right">
<input
type="number"
class="!w-14 ml-4"
disabled={$flowStore.value.priority === undefined}
bind:value={$flowStore.value.priority}
on:focus
on:change={() => {
if ($flowStore.value.priority && $flowStore.value.priority > 100) {
$flowStore.value.priority = 100
} else if ($flowStore.value.priority && $flowStore.value.priority < 0) {
$flowStore.value.priority = 0
}
}}
/>
</svelte:fragment>
</Toggle>
<div class="flex flex-row items-center gap-1">
<ErrorHandlerToggleButton
kind="flow"
frontend/src/lib/components/runs/JobPreview.svelte
+9 -4
View File
@@ -39,11 +39,16 @@
<div class="p-4 flex flex-col gap-2 items-start">
<div class="flex gap-2">
<Badge color="blue">
{#if job && 'duration_ms' in job && job.duration_ms != undefined}
{#if job?.['priority']}
<Badge color="red">
priority: {job?.['priority']}
</Badge>
{/if}
{#if job && 'duration_ms' in job && job.duration_ms != undefined}
<Badge color="blue">
Ran in ({msToSec(job.duration_ms)}s)
{/if}
</Badge>
</Badge>
{/if}
{#if job?.['mem_peak']}
<Badge color="blue">
Mem: {job?.['mem_peak'] ? `${(job['mem_peak'] / 1024).toPrecision(4)}MB` : 'N/A'}
frontend/src/routes/(root)/(logged)/flows/get/[...path]/+page.svelte
+17 -2
View File
@@ -7,7 +7,7 @@
import DetailPageLayout from '$lib/components/details/DetailPageLayout.svelte'
import { goto } from '$app/navigation'
import { Alert, Skeleton } from '$lib/components/common'
import { Alert, Badge as HeaderBadge, Skeleton } from '$lib/components/common'
import MoveDrawer from '$lib/components/MoveDrawer.svelte'
import RunForm from '$lib/components/RunForm.svelte'
import ShareModal from '$lib/components/ShareModal.svelte'
@@ -255,7 +255,22 @@
bind:errorHandlerMuted={flow.ws_error_handler_muted}
scriptOrFlowPath={flow.path}
errorHandlerKind="flow"
/>
>
{#if flow?.value?.priority != undefined}
<div class="hidden md:block">
<HeaderBadge color="red" variant="outlined" size="xs">
{`Priority: ${flow?.value?.priority}`}
</HeaderBadge>
</div>
{/if}
{#if flow?.value?.concurrent_limit != undefined && flow?.value?.concurrency_time_window_s != undefined}
<div class="hidden md:block">
<HeaderBadge color="gray" variant="outlined" size="xs">
{`Concurrency limit: ${flow?.value?.concurrent_limit} runs every ${flow?.value?.concurrency_time_window_s}s`}
</HeaderBadge>
</div>
{/if}
</DetailPageHeader>
</svelte:fragment>
<svelte:fragment slot="form">
<SplitPanesWrapper>
frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte
+5
View File
@@ -291,6 +291,11 @@
<Badge color="blue">{job.job_kind}</Badge>
</div>
{/if}
{#if job && 'priority' in job}
<div>
<Badge color="red">priority: {job.priority}</Badge>
</div>
{/if}
{#if job.tag && !['deno', 'python3', 'flow', 'other', 'go', 'postgresql', 'mysql', 'bigquery', 'snowflake', 'graphql', 'nativets', 'bash', 'powershell', 'other', 'dependency'].includes(job.tag)}
<div>
<Badge color="indigo">Worker group: {job.tag}</Badge>
frontend/src/routes/(root)/(logged)/scripts/get/[...hash]/+page.svelte
+7
View File
@@ -415,6 +415,13 @@
errorHandlerKind="script"
scriptOrFlowPath={script.path}
>
{#if script?.priority != undefined}
<div class="hidden md:block">
<Badge color="red" variant="outlined" size="xs">
{`Priority: ${script.priority}`}
</Badge>
</div>
{/if}
{#if script?.concurrent_limit != undefined && script.concurrency_time_window_s != undefined}
<div class="hidden md:block">
<Badge color="gray" variant="outlined" size="xs">
openflow.openapi.yaml
+4
View File
@@ -54,6 +54,8 @@ components:
type: string
cache_ttl:
type: number
priority:
type: number
required:
- modules
@@ -123,6 +125,8 @@ components:
type: boolean
user_groups_required:
$ref: "#/components/schemas/InputTransform"
priority:
type: number
retry:
$ref: "#/components/schemas/Retry"