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

Add concurrency limits observability (#3586)

Browse Source 
* Add concurrency_key table and field to payload

Add a new table custom_concurrency_key for storing concurrency keys on a
per job basis. Add the field to the job payload to populate the table on
creation of a new job.

* Rename to custom_concurrency_key

Be more explicit about the nature of the option variable by calling it
a more appropriate name.

* Delete concurrency_keys on delete_expire_items

* Propagate db errors up

* Add ref to EE changes

* update sqlx

* Add concurrency_key to tests

* Revert renaming of FlowValue field

FlowValue field concurrency_key must be named concurrency_key for
serialization purposes.

* Remove debug print statement

* Get concurrency_key from Job

Whether it is a Completed job or a queued one, add a ways to get the
associated concurrency key

* Add endpoint to get concurrency_key

* Add endpoint and button to get job concurrency_key

Add a line with a link on the FlowMetadata that goes to the
concurrency_groups page. add endpoint to get the concurrency_key of a
job

* Merge concurrency_key tables into one

migration + change all related querrys

* Add concurrency plot to the runs page

* Prepare sqlx

* Add concurrency Key filter on UI

* Prepare sqlx

* Prepare sqlx

* Fix filter and order query

The limit makes more sense if we cut the older rows, so order by
started_at

* Factor interpolation to reuse on concurrency key

Factor the arg interpolation logic into a function and finish the
processing of concurrency key before insertion

* Remove old concurrency key processing logic

* Second transaction with userdb

To send all concurrency intervals but revealing only uuids of accessible
jobs, make a second transaction with the userdb.

* Remove old second endpoint

the intervals endpoint now also gets the concurrencyt key information
for all jobs

* Show external jobs

* Put filters into a dropdown

Create a ToggleButtonMore and put elements into a dropdown

* Add toggle between two graphs

* Add filter functionality to concurrency graph

* Improve concurrency graph front

* Fix concurrency groups page

* Prepare sqlx

* Add ref to ee

* Change migration to create new table instead

Instead of renaming the custom_concurrency_key_ended atble, we create a
new table and we will delete custom_concurrency_key_ended in the future
when it is no longer linked to any jobs

* Do small UI improvements

* Fix range fusion by not filtering past jobs

Instead of filtering jobs in the backend on a startedAfter value, limit
the query to 1000 and query all possible towards the past to get a good
context for the graph in most sane situations

* Improve frontend UI
This commit is contained in:
wendrul
2024-05-15 20:15:14 +02:00
committed by GitHub
parent 86c5c5b5df
commit 60b036c1ae
45 changed files with 1620 additions and 428 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-0a1c95c4376b944661bab13271091cf3ea0afe68fb8e08e7aea239dc735c625c.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO concurrency_key(key, job_id) VALUES ($1, $2)",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Uuid"
]
},
"nullable": []
},
"hash": "0a1c95c4376b944661bab13271091cf3ea0afe68fb8e08e7aea239dc735c625c"
}
backend/.sqlx/query-b69891c25dd029b1a54e97ace292433e1485324ff7dc802fe75d21c8c6db1d42.json → backend/.sqlx/query-1a612eb0b64eddd2c5657ef73598c47886545796424f8612135b711e2b9ddb6c.json
Generated
+5 -5
View File
@@ -1,16 +1,16 @@
{
"db_name": "PostgreSQL",
"query": "select path, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use, timeout from script where hash = $1 AND workspace_id = $2",
"query": "select tag, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use, timeout from script where hash = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "path",
"name": "tag",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "tag",
"name": "concurrency_key",
"type_info": "Varchar"
},
{
@@ -83,7 +83,7 @@
]
},
"nullable": [
false,
true,
true,
true,
true,
@@ -95,5 +95,5 @@
true
]
},
"hash": "b69891c25dd029b1a54e97ace292433e1485324ff7dc802fe75d21c8c6db1d42"
"hash": "1a612eb0b64eddd2c5657ef73598c47886545796424f8612135b711e2b9ddb6c"
}
backend/.sqlx/query-2352e293e172304a10ab3500b17848e8199b690b1834c382fa9d5c6ae163ec2c.json
Generated
+14
View File
@@ -0,0 +1,14 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE concurrency_key SET ended_at = now() WHERE job_id = $1",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": []
},
"hash": "2352e293e172304a10ab3500b17848e8199b690b1834c382fa9d5c6ae163ec2c"
}
backend/.sqlx/query-2946aa519633291de986978d245dc615a0ee10ce6e6a98bd4906e86225b387b3.json → backend/.sqlx/query-2bbcc383ed79afa3392af417b0cabb32d14a0a26dc36085fb2eaa62e769aca8e.json
Generated
+2 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM custom_concurrency_key_ended WHERE key = $1",
"query": "DELETE FROM concurrency_key WHERE key = $1",
"describe": {
"columns": [],
"parameters": {
@@ -10,5 +10,5 @@
},
"nullable": []
},
"hash": "2946aa519633291de986978d245dc615a0ee10ce6e6a98bd4906e86225b387b3"
"hash": "2bbcc383ed79afa3392af417b0cabb32d14a0a26dc36085fb2eaa62e769aca8e"
}
backend/.sqlx/query-5eff32a37ad4e8b499a82c1874061f6456ded4aca6b33964e138b935cb016eff.json
Generated
+22
View File
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT key FROM concurrency_key WHERE job_id = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "key",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
false
]
},
"hash": "5eff32a37ad4e8b499a82c1874061f6456ded4aca6b33964e138b935cb016eff"
}
backend/.sqlx/query-636db0b9d5963ed540f18ab732a8b29a3308f973cc04fd10979c44ae19169abf.json
Generated
-14
View File
@@ -1,14 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM custom_concurrency_key_ended WHERE ended_at <= now() - ($1::bigint::text || ' s')::interval ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Int8"
]
},
"nullable": []
},
"hash": "636db0b9d5963ed540f18ab732a8b29a3308f973cc04fd10979c44ae19169abf"
}
backend/.sqlx/query-6b313cc9a57ae3c943bda4a3213f7f6231a44b6ef5a52754074d136007f4f72a.json → backend/.sqlx/query-8b10c9ade85c0307e300224e87962ceba8e7ccbe53b56c4670a1634bb9d0e89c.json
Generated
+22 -10
View File
@@ -1,30 +1,40 @@
{
"db_name": "PostgreSQL",
"query": "select tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use, timeout from script where hash = $1 AND workspace_id = $2",
"query": "select path, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use, timeout from script where hash = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "tag",
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "tag",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "concurrency_key",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "concurrent_limit",
"type_info": "Int4"
},
{
"ordinal": 2,
"ordinal": 4,
"name": "concurrency_time_window_s",
"type_info": "Int4"
},
{
"ordinal": 3,
"ordinal": 5,
"name": "cache_ttl",
"type_info": "Int4"
},
{
"ordinal": 4,
"ordinal": 6,
"name": "language: ScriptLang",
"type_info": {
"Custom": {
@@ -51,22 +61,22 @@
}
},
{
"ordinal": 5,
"ordinal": 7,
"name": "dedicated_worker",
"type_info": "Bool"
},
{
"ordinal": 6,
"ordinal": 8,
"name": "priority",
"type_info": "Int2"
},
{
"ordinal": 7,
"ordinal": 9,
"name": "delete_after_use",
"type_info": "Bool"
},
{
"ordinal": 8,
"ordinal": 10,
"name": "timeout",
"type_info": "Int4"
}
@@ -78,6 +88,8 @@
]
},
"nullable": [
false,
true,
true,
true,
true,
@@ -89,5 +101,5 @@
true
]
},
"hash": "6b313cc9a57ae3c943bda4a3213f7f6231a44b6ef5a52754074d136007f4f72a"
"hash": "8b10c9ade85c0307e300224e87962ceba8e7ccbe53b56c4670a1634bb9d0e89c"
}
backend/.sqlx/query-973933b021d2167edff3a48ec4d4abc53ada670155921a4a4c2f05f229ae560a.json
Generated
-14
View File
@@ -1,14 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO custom_concurrency_key_ended VALUES ($1)",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar"
]
},
"nullable": []
},
"hash": "973933b021d2167edff3a48ec4d4abc53ada670155921a4a4c2f05f229ae560a"
}
backend/.sqlx/query-9a5c7e8b60a260085b438bd300972ebf948ea26f313ee5a73b85574becdd7dc7.json
Generated
-23
View File
@@ -1,23 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT char_length(logs) FROM job_logs WHERE job_id = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "char_length",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": [
null
]
},
"hash": "9a5c7e8b60a260085b438bd300972ebf948ea26f313ee5a73b85574becdd7dc7"
}
backend/.sqlx/query-ef132ac8d79579b08d7359789b6f22991f51e1c945efc2924df6253d62b83bba.json → backend/.sqlx/query-9d25fbd21a63e6e9ccfdbf0460c8b3ca6d7fa05600067f379659b59b6f2bf418.json
Generated
+13 -13
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, priority, delete_after_use, timeout 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, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, timeout 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": [
{
@@ -15,21 +15,26 @@
},
{
"ordinal": 2,
"name": "concurrency_key",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "concurrent_limit",
"type_info": "Int4"
},
{
"ordinal": 3,
"ordinal": 4,
"name": "concurrency_time_window_s",
"type_info": "Int4"
},
{
"ordinal": 4,
"ordinal": 5,
"name": "cache_ttl",
"type_info": "Int4"
},
{
"ordinal": 5,
"ordinal": 6,
"name": "language: ScriptLang",
"type_info": {
"Custom": {
@@ -56,20 +61,15 @@
}
},
{
"ordinal": 6,
"ordinal": 7,
"name": "dedicated_worker",
"type_info": "Bool"
},
{
"ordinal": 7,
"ordinal": 8,
"name": "priority",
"type_info": "Int2"
},
{
"ordinal": 8,
"name": "delete_after_use",
"type_info": "Bool"
},
{
"ordinal": 9,
"name": "timeout",
@@ -88,12 +88,12 @@
true,
true,
true,
false,
true,
false,
true,
true,
true
]
},
"hash": "ef132ac8d79579b08d7359789b6f22991f51e1c945efc2924df6253d62b83bba"
"hash": "9d25fbd21a63e6e9ccfdbf0460c8b3ca6d7fa05600067f379659b59b6f2bf418"
}
backend/.sqlx/query-9da0cea2a5d0464ca78cfeccf6cedf2b1c0e6e6cb3c9183a937a68465debdb06.json
Generated
+14
View File
@@ -0,0 +1,14 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM concurrency_key WHERE ended_at <= now() - ($1::bigint::text || ' s')::interval ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Int8"
]
},
"nullable": []
},
"hash": "9da0cea2a5d0464ca78cfeccf6cedf2b1c0e6e6cb3c9183a937a68465debdb06"
}
backend/.sqlx/query-a3ccf362b4f6df400b3c7a084795dbf541eb14c5c374656ffb96da7283a2a6f1.json
Generated
-23
View File
@@ -1,23 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT concurrency_key FROM script WHERE hash = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "concurrency_key",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Int8",
"Text"
]
},
"nullable": [
true
]
},
"hash": "a3ccf362b4f6df400b3c7a084795dbf541eb14c5c374656ffb96da7283a2a6f1"
}
backend/.sqlx/query-a875cb56485b812e9d4739afd0915067f7e5abe0ca0adf264b792fccf21e005b.json
Generated
-23
View File
@@ -1,23 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT value->>'concurrency_key' FROM flow WHERE path = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "a875cb56485b812e9d4739afd0915067f7e5abe0ca0adf264b792fccf21e005b"
}
backend/.sqlx/query-b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76.json
Generated
+2 -2
View File
@@ -18,8 +18,8 @@
"Left": []
},
"nullable": [
true,
false
false,
true
]
},
"hash": "b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76"
backend/.sqlx/query-2f42460fdd8aa125c8fd46b3cd02e47f57de0f073d3ce3bc7d21a7e404a83b5c.json → backend/.sqlx/query-e1c715020f1efb00123171edfb6173bfeaaa95d4f870240e42b76e5df304ce99.json
Generated
+20 -8
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, priority, timeout 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, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use, timeout 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": [
{
@@ -15,21 +15,26 @@
},
{
"ordinal": 2,
"name": "concurrency_key",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "concurrent_limit",
"type_info": "Int4"
},
{
"ordinal": 3,
"ordinal": 4,
"name": "concurrency_time_window_s",
"type_info": "Int4"
},
{
"ordinal": 4,
"ordinal": 5,
"name": "cache_ttl",
"type_info": "Int4"
},
{
"ordinal": 5,
"ordinal": 6,
"name": "language: ScriptLang",
"type_info": {
"Custom": {
@@ -56,17 +61,22 @@
}
},
{
"ordinal": 6,
"ordinal": 7,
"name": "dedicated_worker",
"type_info": "Bool"
},
{
"ordinal": 7,
"ordinal": 8,
"name": "priority",
"type_info": "Int2"
},
{
"ordinal": 8,
"ordinal": 9,
"name": "delete_after_use",
"type_info": "Bool"
},
{
"ordinal": 10,
"name": "timeout",
"type_info": "Int4"
}
@@ -83,11 +93,13 @@
true,
true,
true,
true,
false,
true,
true,
true,
true
]
},
"hash": "2f42460fdd8aa125c8fd46b3cd02e47f57de0f073d3ce3bc7d21a7e404a83b5c"
"hash": "e1c715020f1efb00123171edfb6173bfeaaa95d4f870240e42b76e5df304ce99"
}
backend/.sqlx/query-0b74ab3a237b2b7f54c05c7ea74294317fcd039870268eaa930f1ba8b8250559.json → backend/.sqlx/query-fab5a746a179eea356f0bb9005590e3bb89c4deff181827bd2fa5630c21e42a7.json
Generated
+2 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT COUNT(*) as count, COALESCE(MAX(ended_at), now() - INTERVAL '1 second' * $2) as max_ended_at FROM custom_concurrency_key_ended WHERE key = $1 AND ended_at >= (now() - INTERVAL '1 second' * $2)",
"query": "SELECT COUNT(*) as count, COALESCE(MAX(ended_at), now() - INTERVAL '1 second' * $2) as max_ended_at FROM concurrency_key WHERE key = $1 AND ended_at >= (now() - INTERVAL '1 second' * $2)",
"describe": {
"columns": [
{
@@ -25,5 +25,5 @@
null
]
},
"hash": "0b74ab3a237b2b7f54c05c7ea74294317fcd039870268eaa930f1ba8b8250559"
"hash": "fab5a746a179eea356f0bb9005590e3bb89c4deff181827bd2fa5630c21e42a7"
}
backend/ee-repo-ref.txt
+1 -1
View File
@@ -1 +1 @@
644b6f49f087790a728a7a0a82525997b1742738
e0b0494c26b63efa4aa3f41a9fb42c7733dcca2e
backend/migrations/20240515093517_concurrency_key_observability.down.sql
+3
View File
@@ -0,0 +1,3 @@
-- Add down migration script here
DROP TABLE concurrency_key
backend/migrations/20240515093517_concurrency_key_observability.up.sql
+10
View File
@@ -0,0 +1,10 @@
-- Add up migration script here
CREATE TABLE concurrency_key (
key VARCHAR(255) NOT NULL,
ended_at TIMESTAMP WITH TIME ZONE,
job_id uuid NOT NULL,
PRIMARY KEY (job_id)
);
CREATE INDEX concurrency_key_ended_at_idx ON concurrency_key (key, ended_at DESC);
backend/src/monitor.rs
+1 -1
View File
@@ -318,7 +318,7 @@ pub async fn delete_expired_items(db: &DB) -> () {
tracing::error!("Error deleting job stats: {:?}", e);
}
if let Err(e) = sqlx::query!(
"DELETE FROM custom_concurrency_key_ended WHERE ended_at <= now() - ($1::bigint::text || ' s')::interval ",
"DELETE FROM concurrency_key WHERE ended_at <= now() - ($1::bigint::text || ' s')::interval ",
job_retention_secs
)
.execute(&mut *tx)
backend/tests/worker.rs
+14 -1
View File
@@ -1067,6 +1067,7 @@ async fn test_deno_flow(db: Pool<Postgres>) {
path: None,
lock: None,
tag: None,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
},
@@ -1104,6 +1105,7 @@ async fn test_deno_flow(db: Pool<Postgres>) {
path: None,
lock: None,
tag: None,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
},
@@ -1219,6 +1221,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) {
path: None,
lock: None,
tag: None,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
},
@@ -1267,6 +1270,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) {
path: None,
lock: None,
tag: None,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
},
@@ -1301,6 +1305,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) {
path: None,
lock: None,
tag: None,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
},
@@ -1357,6 +1362,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) {
path: None,
lock: None,
tag: None,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
},
@@ -1668,6 +1674,7 @@ func main(derp string) (string, error) {
path: None,
lock: None,
language: ScriptLang::Go,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None,
@@ -1700,6 +1707,7 @@ echo "hello $msg"
path: None,
lock: None,
language: ScriptLang::Bash,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None,
@@ -1729,6 +1737,7 @@ def main():
path: None,
language: ScriptLang::Python3,
lock: None,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None,
@@ -1764,6 +1773,7 @@ def main():
path: None,
language: ScriptLang::Python3,
lock: None,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None,
@@ -1798,6 +1808,7 @@ def main():
path: None,
language: ScriptLang::Python3,
lock: None,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None,
@@ -3188,6 +3199,7 @@ async fn run_deployed_relative_imports(db: &Pool<Postgres>, script_content: Stri
let job = RunJob::from(JobPayload::ScriptHash {
path: "f/system/test_import".to_string(),
hash: ScriptHash(script.hash),
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None,
@@ -3223,6 +3235,7 @@ async fn run_preview_relative_imports(db: &Pool<Postgres>, script_content: Strin
path: Some("f/system/test_import".to_string()),
language,
lock: None,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None,
@@ -3342,4 +3355,4 @@ def main():
run_deployed_relative_imports(&db, content.clone(), ScriptLang::Python3).await;
run_preview_relative_imports(&db, content, ScriptLang::Python3).await;
}
}
backend/windmill-api/openapi.yaml
+130 -6
View File
@@ -7974,7 +7974,7 @@ paths:
type: array
items:
$ref: "#/components/schemas/ConcurrencyGroup"
/concurrency_groups/{concurrency_id}:
/concurrency_groups/prune/{concurrency_id}:
delete:
summary: Delete concurrency group
operationId: deleteConcurrencyGroup
@@ -7990,6 +7990,95 @@ paths:
schema:
type: object
properties: {}
/concurrency_groups/{id}/key:
get:
summary: Get the concurrency key for a job that has concurrency limits enabled
operationId: getConcurrencyKey
tags:
- concurrencyGroups
parameters:
- $ref: "#/components/parameters/JobId"
responses:
"200":
description: concurrency key for given job
content:
application/json:
schema:
type: string
/w/{workspace}/concurrency_groups/intervals:
get:
summary: Get intervals of job runtime concurrency
operationId: getConcurrencyIntervals
tags:
- concurrencyGroups
parameters:
- name: concurrency_key
in: query
required: false
schema:
type: string
- name: row_limit
in: query
required: false
schema:
type: number
- $ref: "#/components/parameters/WorkspaceId"
- $ref: "#/components/parameters/CreatedBy"
- $ref: "#/components/parameters/Label"
- $ref: "#/components/parameters/ParentJob"
- $ref: "#/components/parameters/ScriptExactPath"
- $ref: "#/components/parameters/ScriptStartPath"
- $ref: "#/components/parameters/SchedulePath"
- $ref: "#/components/parameters/ScriptExactHash"
- $ref: "#/components/parameters/StartedBefore"
- $ref: "#/components/parameters/StartedAfter"
- $ref: "#/components/parameters/CreatedOrStartedBefore"
- $ref: "#/components/parameters/Running"
- $ref: "#/components/parameters/ScheduledForBeforeNow"
- $ref: "#/components/parameters/CreatedOrStartedAfter"
- $ref: "#/components/parameters/JobKinds"
- $ref: "#/components/parameters/ArgsFilter"
- $ref: "#/components/parameters/Tag"
- $ref: "#/components/parameters/ResultFilter"
- $ref: "#/components/parameters/Page"
- $ref: "#/components/parameters/PerPage"
- name: is_skipped
description: is the job skipped
in: query
schema:
type: boolean
- name: is_flow_step
description: is the job a flow step
in: query
schema:
type: boolean
- name: has_null_parent
description: has null parent
in: query
schema:
type: boolean
- name: success
description: filter on successful jobs
in: query
schema:
type: boolean
- name: all_workspaces
description: get jobs from all workspaces (only valid if request come from the `admins` workspace)
in: query
schema:
type: boolean
- name: is_not_schedule
description: is not a scheduled job
in: query
schema:
type: boolean
responses:
"200":
description: time
content:
application/json:
schema:
$ref: "#/components/schemas/ConcurrencyIntervals"
components:
securitySchemes:
@@ -10395,12 +10484,47 @@ components:
ConcurrencyGroup:
type: object
properties:
concurrency_id:
concurrency_key:
type: string
job_uuids:
total_running:
type: number
required:
- concurrency_key
- total_running
ConcurrencyIntervals:
type: object
properties:
concurrency_key:
type: string
running_jobs:
type: array
items:
type: string
type: object
properties:
job_id:
type: string
concurrency_key:
type: string
started_at:
type: string
format: date-time
completed_jobs:
type: array
items:
type: object
properties:
job_id:
type: string
concurrency_key:
type: string
started_at:
type: string
format: date-time
ended_at:
type: string
format: date-time
required:
- concurrency_id
- job_uuids
- concurrency_key
- running_jobs
- completed_jobs
backend/windmill-api/src/concurrency_groups.rs
+293 -51
View File
@@ -1,87 +1,69 @@
#[cfg(feature = "enterprise")]
use crate::db::{ApiAuthed, DB};
#[cfg(feature = "enterprise")]
use crate::jobs::{
filter_list_completed_query, filter_list_queue_query, ListCompletedQuery, ListQueueQuery,
};
use crate::users::check_scopes;
use axum::extract::Path;
#[cfg(feature = "enterprise")]
use axum::routing::{delete, get};
#[cfg(feature = "enterprise")]
use axum::{Extension, Json};
use axum::{extract::Query, Extension, Json};
use serde::Deserialize;
use axum::Router;
#[cfg(feature = "enterprise")]
use serde::Serialize;
#[cfg(feature = "enterprise")]
use std::collections::HashMap;
#[cfg(feature = "enterprise")]
use sql_builder::bind::Bind;
use sql_builder::SqlBuilder;
use sqlx::postgres::PgRow;
use sqlx::{FromRow, Row};
use uuid::Uuid;
use windmill_common::db::UserDB;
use windmill_common::error::Error::{InternalErr, PermissionDenied};
#[cfg(feature = "enterprise")]
use windmill_common::error::JsonResult;
use windmill_common::utils::require_admin;
#[cfg(feature = "enterprise")]
pub fn global_service() -> Router {
Router::new()
.route("/list", get(list_concurrency_groups))
.route("/*id", delete(delete_concurrency_group))
.route("/prune/*concurrency_key", delete(prune_concurrency_group))
.route("/:job_id/key", get(get_concurrency_key))
}
#[cfg(not(feature = "enterprise"))]
pub fn global_service() -> Router {
Router::new()
pub fn workspaced_service() -> Router {
Router::new().route("/intervals", get(get_concurrent_intervals))
}
#[cfg(feature = "enterprise")]
#[derive(Serialize)]
pub struct ConcurrencyGroups {
concurrency_id: String,
job_uuids: Vec<String>,
concurrency_key: String,
total_running: i64,
}
#[cfg(feature = "enterprise")]
async fn list_concurrency_groups(
authed: ApiAuthed,
Extension(db): Extension<DB>,
) -> JsonResult<Vec<ConcurrencyGroups>> {
if !authed.is_admin {
return Err(PermissionDenied(
"Only administrators can see concurrency groups".to_string(),
));
}
let concurrency_groups_raw = sqlx::query_as::<_, (String, serde_json::Value)>(
"SELECT * FROM concurrency_counter ORDER BY concurrency_id ASC",
)
.fetch_all(&db)
require_admin(authed.is_admin, &authed.username)?;
let concurrency_counts = sqlx::query_as::<_, (String, i64)>(
"SELECT concurrency_id, (select COUNT(*) from jsonb_object_keys(job_uuids)) as n_job_uuids FROM concurrency_counter",
).fetch_all(&db)
.await?;
let mut concurrency_groups: Vec<ConcurrencyGroups> = vec![];
for (concurrency_id, job_uuids_json) in concurrency_groups_raw {
let job_uuids_map = serde_json::from_value::<HashMap<String, serde_json::Value>>(
job_uuids_json,
)
.map_err(|err| {
tracing::error!(
"Error deserializing concurrency_counter table content: {:?}",
err
);
InternalErr(format!(
"Error deserializing concurrency_counter table content: {}",
err.to_string()
))
})?;
for (concurrency_key, count) in concurrency_counts {
concurrency_groups.push(ConcurrencyGroups {
concurrency_id: concurrency_id.clone(),
job_uuids: job_uuids_map.keys().cloned().collect(),
concurrency_key: concurrency_key.clone(),
total_running: count,
})
}
return Ok(Json(concurrency_groups));
}
#[cfg(feature = "enterprise")]
async fn delete_concurrency_group(
async fn prune_concurrency_group(
authed: ApiAuthed,
Extension(db): Extension<DB>,
Path(concurrency_id): Path<String>,
Path(concurrency_key): Path<String>,
) -> JsonResult<()> {
if !authed.is_admin {
return Err(PermissionDenied(
@@ -93,7 +75,7 @@ async fn delete_concurrency_group(
let concurrency_group = sqlx::query_as::<_, (String, i64)>(
"SELECT concurrency_id, (select COUNT(*) from jsonb_object_keys(job_uuids)) as n_job_uuids FROM concurrency_counter WHERE concurrency_id = $1 FOR UPDATE",
)
.bind(concurrency_id.clone())
.bind(concurrency_key.clone())
.fetch_optional(&mut *tx)
.await?;
@@ -108,14 +90,14 @@ async fn delete_concurrency_group(
sqlx::query!(
"DELETE FROM concurrency_counter WHERE concurrency_id = $1",
concurrency_id.clone(),
concurrency_key.clone(),
)
.execute(&mut *tx)
.await?;
sqlx::query!(
"DELETE FROM custom_concurrency_key_ended WHERE key = $1",
concurrency_id.clone(),
"DELETE FROM concurrency_key WHERE key = $1",
concurrency_key.clone(),
)
.execute(&mut *tx)
.await?;
@@ -123,3 +105,263 @@ async fn delete_concurrency_group(
tx.commit().await?;
Ok(Json(()))
}
#[derive(Serialize)]
struct ConcurrencyIntervals {
concurrency_key: Option<String>,
running_jobs: Vec<RunningJobDuration>,
completed_jobs: Vec<CompletedJobDuration>,
}
#[derive(Serialize)]
struct CompletedJobDuration {
job_id: Option<Uuid>,
concurrency_key: Option<String>,
started_at: chrono::DateTime<chrono::Utc>,
ended_at: chrono::DateTime<chrono::Utc>,
}
impl<'r> FromRow<'r, PgRow> for CompletedJobDuration {
fn from_row(row: &'r PgRow) -> Result<Self, sqlx::Error> {
let duration_ms: i64 = row.try_get("duration_ms")?;
let started_at = row.try_get("started_at")?;
let ended_at: chrono::DateTime<chrono::Utc> =
started_at + std::time::Duration::from_millis(duration_ms.try_into().unwrap());
Ok(Self {
job_id: row.try_get("id")?,
concurrency_key: row.try_get("key")?,
started_at,
ended_at,
})
}
}
#[derive(Serialize, FromRow)]
struct RunningJobDuration {
#[sqlx(rename = "id")]
job_id: Option<Uuid>,
#[sqlx(rename = "key")]
concurrency_key: Option<String>,
started_at: Option<chrono::DateTime<chrono::Utc>>,
}
#[derive(Deserialize)]
struct ConcurrentIntervalsParams {
concurrency_key: Option<String>,
row_limit: Option<i64>,
}
fn join_concurrency_key<'c>(concurrency_key: Option<&String>, mut sqlb: SqlBuilder) -> SqlBuilder {
match concurrency_key {
Some(key) => sqlb
.join("concurrency_key")
.on_eq("id", "job_id")
.and_where_eq("key", "?".bind(key)),
None => sqlb.left().join("concurrency_key").on_eq("id", "job_id"),
};
sqlb
}
async fn get_concurrent_intervals(
authed: ApiAuthed,
Extension(db): Extension<DB>,
Extension(user_db): Extension<UserDB>,
Path(w_id): Path<String>,
Query(iq): Query<ConcurrentIntervalsParams>,
Query(lq): Query<ListCompletedQuery>,
) -> JsonResult<ConcurrencyIntervals> {
check_scopes(&authed, || format!("listjobs"))?;
let row_limit = iq.row_limit.unwrap_or(1000);
let concurrency_key = iq.concurrency_key;
let lq_copy = lq.clone();
let lqq = ListQueueQuery {
script_path_start: lq_copy.script_path_start,
script_path_exact: lq_copy.script_path_exact,
script_hash: lq_copy.script_hash,
created_by: lq_copy.created_by,
started_before: lq_copy.started_before,
started_after: lq_copy.started_after,
created_before: lq_copy.created_before,
created_after: lq_copy.created_after,
created_or_started_before: lq_copy.created_or_started_before,
created_or_started_after: lq_copy.created_or_started_after,
running: lq_copy.running,
parent_job: lq_copy.parent_job,
order_desc: Some(true),
job_kinds: lq_copy.job_kinds,
suspended: lq_copy.suspended,
args: lq_copy.args,
tag: lq_copy.tag,
schedule_path: lq_copy.schedule_path,
scheduled_for_before_now: lq_copy.scheduled_for_before_now,
all_workspaces: lq_copy.all_workspaces,
is_flow_step: lq_copy.is_flow_step,
has_null_parent: lq_copy.has_null_parent,
is_not_schedule: lq_copy.is_not_schedule,
};
let mut sqlb_q = SqlBuilder::select_from("queue")
.fields(&["id", "key", "started_at"])
.order_by("created_at", lq.order_desc.unwrap_or(true))
.limit(row_limit)
.clone();
let mut sqlb_c = SqlBuilder::select_from("completed_job")
.fields(&["id", "key", "started_at", "duration_ms"])
.order_by("created_at", lq.order_desc.unwrap_or(true))
.limit(row_limit)
.clone();
sqlb_q = join_concurrency_key(concurrency_key.as_ref(), sqlb_q);
sqlb_c = join_concurrency_key(concurrency_key.as_ref(), sqlb_c);
sqlb_q.and_where_is_not_null("started_at");
// When we have a concurrency key defined, fetch jobs from other workspaces
// as obscured unless we're in the admins workspace. This is to show the
// potential concurrency races without showing jobs that don't belong to
// the workspace.
let sqlb_all_workspaces: Option<(SqlBuilder, SqlBuilder)> =
if concurrency_key.is_some() && w_id != "admins" {
Some((
filter_list_queue_query(
sqlb_q.clone(),
&ListQueueQuery { all_workspaces: Some(true), ..lqq.clone() },
"admins",
),
filter_list_completed_query(
sqlb_c.clone(),
&ListCompletedQuery { all_workspaces: Some(true), ..lq.clone() },
"admins",
),
))
} else {
None
};
sqlb_q = filter_list_queue_query(sqlb_q, &lqq, w_id.as_str());
sqlb_c = filter_list_completed_query(sqlb_c, &lq, w_id.as_str());
let sql_q = sqlb_q.query()?;
let sql_c = sqlb_c.query()?;
let mut tx = user_db.begin(&authed).await?;
let running_jobs_user: Vec<RunningJobDuration> = if lq.success.is_none() {
sqlx::query_as(&sql_q).fetch_all(&mut *tx).await?
} else {
vec![]
};
let completed_jobs_user: Vec<CompletedJobDuration> = if lq.running.is_none() {
sqlx::query_as(&sql_c).fetch_all(&mut *tx).await?
} else {
vec![]
};
tx.commit().await?;
// To avoid infering information through filtering, don't return obscured
// jobs if the filters are too specific
let should_fetch_obscured_jobs = match lq {
ListCompletedQuery {
script_path_start: None,
script_path_exact: None,
script_hash: None,
created_by: None,
success: None,
running: None,
parent_job: None,
is_skipped: None | Some(false),
suspended: None,
schedule_path: None,
args: None,
result: None,
tag: None,
scheduled_for_before_now: None,
has_null_parent: None,
label: None,
is_not_schedule: None,
started_before: _,
started_after: _,
created_before: _,
created_after: _,
created_or_started_before: _,
created_or_started_after: _,
order_desc: _,
job_kinds: _,
is_flow_step: _,
all_workspaces: _,
} => true,
_ => false,
};
// This second transaction using the raw db lets us get info for jobs that
// the user has no access to. Before returning that, we will hide the ids
if should_fetch_obscured_jobs && concurrency_key.is_some() {
let (sql_q, sql_c) = if let Some(sqlb) = sqlb_all_workspaces {
(sqlb.0.query()?, sqlb.1.query()?)
} else {
(sql_q, sql_c)
};
let running_jobs_db: Vec<RunningJobDuration> = if lq.success.is_none() {
sqlx::query_as(&sql_q).fetch_all(&db).await?
} else {
vec![]
};
let completed_jobs_db: Vec<CompletedJobDuration> = if lq.running.is_none() {
sqlx::query_as(&sql_c).fetch_all(&db).await?
} else {
vec![]
};
let running_jobs = running_jobs_db
.into_iter()
.map(|r| {
if running_jobs_user
.iter()
.any(|u| u.job_id.unwrap() == r.job_id.unwrap())
{
RunningJobDuration { ..r }
} else {
RunningJobDuration { job_id: None, ..r }
}
})
.collect();
let completed_jobs = completed_jobs_db
.into_iter()
.map(|r| {
if completed_jobs_user
.iter()
.any(|u| u.job_id.unwrap() == r.job_id.unwrap())
{
CompletedJobDuration { ..r }
} else {
CompletedJobDuration { job_id: None, ..r }
}
})
.collect();
return Ok(Json(ConcurrencyIntervals {
concurrency_key,
running_jobs,
completed_jobs,
}));
}
Ok(Json(ConcurrencyIntervals {
concurrency_key,
running_jobs: running_jobs_user,
completed_jobs: completed_jobs_user,
}))
}
async fn get_concurrency_key(
Extension(db): Extension<DB>,
Path(job_id): Path<Uuid>,
) -> JsonResult<Option<String>> {
let key = sqlx::query_scalar!("SELECT key FROM concurrency_key WHERE job_id = $1", job_id)
.fetch_optional(&db)
.await?;
Ok(Json(key))
}
backend/windmill-api/src/flows.rs
+1
View File
@@ -912,6 +912,7 @@ mod tests {
path: None,
lock: None,
tag: None,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
},
backend/windmill-api/src/jobs.rs
+110 -25
View File
@@ -9,6 +9,7 @@
use axum::body::Body;
use axum::http::HeaderValue;
use serde_json::value::RawValue;
use sqlx::Pool;
use std::collections::HashMap;
#[cfg(feature = "prometheus")]
use std::sync::atomic::Ordering;
@@ -466,6 +467,7 @@ pub async fn get_path_tag_limits_cache_for_hash(
) -> error::Result<(
String,
Option<String>,
Option<String>,
Option<i32>,
Option<i32>,
Option<i32>,
@@ -476,7 +478,7 @@ pub async fn get_path_tag_limits_cache_for_hash(
Option<i32>,
)> {
let script = sqlx::query!(
"select path, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use, timeout from script where hash = $1 AND workspace_id = $2",
"select path, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use, timeout from script where hash = $1 AND workspace_id = $2",
hash,
w_id
)
@@ -490,6 +492,7 @@ pub async fn get_path_tag_limits_cache_for_hash(
Ok((
script.path,
script.tag,
script.concurrency_key,
script.concurrent_limit,
script.concurrency_time_window_s,
script.cache_ttl,
@@ -622,7 +625,7 @@ fn generate_get_job_query(no_logs: bool, table: &str) -> String {
{join}
WHERE id = $1 AND {table}.workspace_id = $2");
}
async fn get_job_internal(
pub async fn get_job_internal(
db: &DB,
workspace_id: &str,
job_id: Uuid,
@@ -872,7 +875,7 @@ impl RunJobQuery {
}
}
#[derive(Deserialize)]
#[derive(Deserialize, Clone)]
pub struct ListQueueQuery {
pub script_path_start: Option<String>,
pub script_path_exact: Option<String>,
@@ -900,13 +903,7 @@ pub struct ListQueueQuery {
pub is_not_schedule: Option<bool>,
}
fn list_queue_jobs_query(w_id: &str, lq: &ListQueueQuery, fields: &[&str]) -> SqlBuilder {
let mut sqlb = SqlBuilder::select_from("queue")
.fields(fields)
.order_by("created_at", lq.order_desc.unwrap_or(true))
.limit(1000)
.clone();
pub fn filter_list_queue_query(mut sqlb: SqlBuilder, lq: &ListQueueQuery, w_id: &str) -> SqlBuilder {
if w_id != "admins" || !lq.all_workspaces.is_some_and(|x| x) {
sqlb.and_where_eq("workspace_id", "?".bind(&w_id));
}
@@ -1007,6 +1004,17 @@ fn list_queue_jobs_query(w_id: &str, lq: &ListQueueQuery, fields: &[&str]) -> Sq
}
sqlb
}
pub fn list_queue_jobs_query(w_id: &str, lq: &ListQueueQuery, fields: &[&str]) -> SqlBuilder {
let sqlb = SqlBuilder::select_from("queue")
.fields(fields)
.order_by("created_at", lq.order_desc.unwrap_or(true))
.limit(1000)
.clone();
filter_list_queue_query(sqlb, lq, w_id)
}
#[derive(Serialize, FromRow)]
@@ -1988,6 +1996,13 @@ impl Job {
}
}
pub fn is_flow(&self) -> bool {
matches!(
self.job_kind(),
JobKind::Flow | JobKind::FlowPreview | JobKind::SingleScriptFlow
)
}
pub fn job_kind(&self) -> &JobKind {
match self {
Job::QueuedJob(job) => &job.job_kind,
@@ -2003,6 +2018,50 @@ impl Job {
Job::CompletedJobWithFormattedResult(job) => job.cj.id,
}
}
pub fn workspace_id(&self) -> &String {
match self {
Job::QueuedJob(job) => &job.workspace_id,
Job::CompletedJob(job) => &job.workspace_id,
Job::CompletedJobWithFormattedResult(job) => &job.cj.workspace_id,
}
}
pub fn script_path(&self) -> &str {
match self {
Job::QueuedJob(job) => job.script_path.as_ref(),
Job::CompletedJob(job) => job.script_path.as_ref(),
Job::CompletedJobWithFormattedResult(job) => job.cj.script_path.as_ref(),
}
.map(String::as_str)
.unwrap_or("tmp/main")
}
pub fn args(&self) -> Option<&sqlx::types::Json<HashMap<String, Box<RawValue>>>> {
match self {
Job::QueuedJob(job) => job.args.as_ref(),
Job::CompletedJob(job) => job.args.as_ref(),
Job::CompletedJobWithFormattedResult(job) => job.cj.args.as_ref(),
}
}
pub fn full_path_with_workspace(&self) -> String {
format!(
"{}/{}/{}",
self.workspace_id(),
if self.is_flow() { "flow" } else { "script" },
self.script_path(),
)
}
pub async fn concurrency_key(&self, db: &Pool<Postgres>) -> Result<Option<String>, sqlx::Error> {
sqlx::query_scalar!(
"SELECT key FROM concurrency_key WHERE job_id = $1",
self.id()
)
.fetch_optional(db)
.await
}
}
#[derive(sqlx::FromRow)]
@@ -2485,6 +2544,9 @@ pub async fn run_workflow_as_code(
path: job.script_path,
language: job.language.unwrap_or_else(|| ScriptLang::Deno),
lock: job.raw_lock,
custom_concurrency_key: windmill_queue::custom_concurrency_key(&db, job.id)
.await
.map_err(to_anyhow)?,
concurrent_limit: job.concurrent_limit,
concurrency_time_window_s: job.concurrency_time_window_s,
cache_ttl: job.cache_ttl,
@@ -2975,6 +3037,7 @@ pub async fn run_wait_result_script_by_hash(
let (
path,
tag,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
mut cache_ttl,
@@ -3000,6 +3063,7 @@ pub async fn run_wait_result_script_by_hash(
JobPayload::ScriptHash {
hash: ScriptHash(hash),
path: path,
custom_concurrency_key,
concurrent_limit: concurrent_limit,
concurrency_time_window_s: concurrency_time_window_s,
cache_ttl,
@@ -3149,6 +3213,7 @@ async fn run_preview_script(
path: preview.path,
language: preview.language.unwrap_or(ScriptLang::Deno),
lock: preview.lock,
custom_concurrency_key: None,
concurrent_limit: None, // TODO(gbouv): once I find out how to store limits in the content of a script, should be easy to plug limits here
concurrency_time_window_s: None, // TODO(gbouv): same as above
cache_ttl: None,
@@ -3233,6 +3298,7 @@ async fn run_bundle_preview_script(
concurrency_time_window_s: None, // TODO(gbouv): same as above
cache_ttl: None,
dedicated_worker: preview.dedicated_worker,
custom_concurrency_key: None,
}),
},
args,
@@ -3425,6 +3491,7 @@ async fn add_batch_jobs(
job_kind,
language,
dedicated_worker,
_custom_concurrency_key,
concurrent_limit,
concurrent_time_window_s,
timeout,
@@ -3434,6 +3501,7 @@ async fn add_batch_jobs(
let (
script_hash,
_tag,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
_cache_ttl,
@@ -3449,6 +3517,7 @@ async fn add_batch_jobs(
JobKind::Script,
Some(language),
dedicated_worker,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
timeout,
@@ -3510,7 +3579,17 @@ async fn add_batch_jobs(
}
return Ok(Json(uuids));
}
"noop" => (None, None, JobKind::Noop, None, None, None, None, None),
"noop" => (
None,
None,
JobKind::Noop,
None,
None,
None,
None,
None,
None,
),
_ => {
return Err(error::Error::BadRequest(format!(
"Invalid batch kind: {}",
@@ -3630,6 +3709,7 @@ pub async fn run_job_by_hash(
let (
path,
tag,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
mut cache_ttl,
@@ -3656,6 +3736,7 @@ pub async fn run_job_by_hash(
JobPayload::ScriptHash {
hash: ScriptHash(hash),
path: path,
custom_concurrency_key,
concurrent_limit: concurrent_limit,
concurrency_time_window_s: concurrency_time_window_s,
cache_ttl,
@@ -3814,20 +3895,7 @@ async fn get_job_update(
}
}
fn list_completed_jobs_query(
w_id: &str,
per_page: usize,
offset: usize,
lq: &ListCompletedQuery,
fields: &[&str],
) -> SqlBuilder {
let mut sqlb = SqlBuilder::select_from("completed_job")
.fields(fields)
.order_by("created_at", lq.order_desc.unwrap_or(true))
.offset(offset)
.limit(per_page)
.clone();
pub fn filter_list_completed_query(mut sqlb: SqlBuilder, lq: &ListCompletedQuery, w_id: &str) -> SqlBuilder {
if w_id != "admins" || !lq.all_workspaces.is_some_and(|x| x) {
sqlb.and_where_eq("workspace_id", "?".bind(&w_id));
}
@@ -3922,6 +3990,23 @@ fn list_completed_jobs_query(
sqlb
}
pub fn list_completed_jobs_query(
w_id: &str,
per_page: usize,
offset: usize,
lq: &ListCompletedQuery,
fields: &[&str],
) -> SqlBuilder {
let sqlb = SqlBuilder::select_from("completed_job")
.fields(fields)
.order_by("created_at", lq.order_desc.unwrap_or(true))
.offset(offset)
.limit(per_page)
.clone();
filter_list_completed_query(sqlb, lq, w_id)
}
#[derive(Deserialize, Clone)]
pub struct ListCompletedQuery {
pub script_path_start: Option<String>,
backend/windmill-api/src/lib.rs
+1
View File
@@ -197,6 +197,7 @@ pub async fn run_server(
.nest("/apps", apps::workspaced_service())
.nest("/audit", audit::workspaced_service())
.nest("/capture", capture::workspaced_service())
.nest("/concurrency_groups", concurrency_groups::workspaced_service())
.nest("/embeddings", embeddings::workspaced_service())
.nest("/drafts", drafts::workspaced_service())
.nest("/favorites", favorite::workspaced_service())
backend/windmill-common/src/flows.rs
+2
View File
@@ -345,6 +345,8 @@ pub enum FlowModuleValue {
tag: Option<String>,
language: ScriptLang,
#[serde(skip_serializing_if = "Option::is_none")]
custom_concurrency_key: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
concurrent_limit: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
concurrency_time_window_s: Option<i32>,
backend/windmill-common/src/jobs.rs
+8 -1
View File
@@ -282,6 +282,7 @@ pub enum JobPayload {
ScriptHash {
hash: ScriptHash,
path: String,
custom_concurrency_key: Option<String>,
concurrent_limit: Option<i32>,
concurrency_time_window_s: Option<i32>,
cache_ttl: Option<i32>,
@@ -328,6 +329,7 @@ pub enum JobPayload {
hash: ScriptHash,
args: HashMap<String, serde_json::Value>,
retry: Retry, // for now only used to retry the script, so retry is necessarily present
custom_concurrency_key: Option<String>,
concurrent_limit: Option<i32>,
concurrency_time_window_s: Option<i32>,
cache_ttl: Option<i32>,
@@ -348,6 +350,7 @@ pub struct RawCode {
pub hash: Option<i64>,
pub language: ScriptLang,
pub lock: Option<String>,
pub custom_concurrency_key: Option<String>,
pub concurrent_limit: Option<i32>,
pub concurrency_time_window_s: Option<i32>,
pub cache_ttl: Option<i32>,
@@ -374,6 +377,7 @@ pub async fn script_path_to_payload(
let (
script_hash,
tag,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
cache_ttl,
@@ -387,6 +391,7 @@ pub async fn script_path_to_payload(
JobPayload::ScriptHash {
hash: script_hash,
path: script_path.to_owned(),
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
cache_ttl: cache_ttl,
@@ -408,6 +413,7 @@ pub async fn script_hash_to_tag_and_limits<'c>(
w_id: &String,
) -> error::Result<(
Option<Tag>,
Option<String>,
Option<i32>,
Option<i32>,
Option<i32>,
@@ -418,7 +424,7 @@ pub async fn script_hash_to_tag_and_limits<'c>(
Option<i32>,
)> {
let script = sqlx::query!(
"select tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use, timeout from script where hash = $1 AND workspace_id = $2",
"select tag, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use, timeout from script where hash = $1 AND workspace_id = $2",
script_hash.0,
w_id
)
@@ -431,6 +437,7 @@ pub async fn script_hash_to_tag_and_limits<'c>(
})?;
Ok((
script.tag,
script.concurrency_key,
script.concurrent_limit,
script.concurrency_time_window_s,
script.cache_ttl,
backend/windmill-common/src/lib.rs
+6 -2
View File
@@ -249,6 +249,7 @@ pub async fn get_latest_deployed_hash_for_path(
) -> error::Result<(
scripts::ScriptHash,
Option<Tag>,
Option<String>,
Option<i32>,
Option<i32>,
Option<i32>,
@@ -259,7 +260,7 @@ pub async fn get_latest_deployed_hash_for_path(
Option<i32>,
)> {
let r_o = sqlx::query!(
"select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use, timeout from script where path = $1 AND workspace_id = $2 AND
"select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use, timeout 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,
@@ -273,6 +274,7 @@ pub async fn get_latest_deployed_hash_for_path(
Ok((
scripts::ScriptHash(script.hash),
script.tag,
script.concurrency_key,
script.concurrent_limit,
script.concurrency_time_window_s,
script.cache_ttl,
@@ -291,6 +293,7 @@ pub async fn get_latest_hash_for_path<'c>(
) -> error::Result<(
scripts::ScriptHash,
Option<Tag>,
Option<String>,
Option<i32>,
Option<i32>,
Option<i32>,
@@ -300,7 +303,7 @@ pub async fn get_latest_hash_for_path<'c>(
Option<i32>,
)> {
let r_o = sqlx::query!(
"select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, timeout FROM script where path = $1 AND workspace_id = $2 AND
"select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, timeout 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,
@@ -314,6 +317,7 @@ pub async fn get_latest_hash_for_path<'c>(
Ok((
scripts::ScriptHash(script.hash),
script.tag,
script.concurrency_key,
script.concurrent_limit,
script.concurrency_time_window_s,
script.cache_ttl,
backend/windmill-queue/src/jobs.rs
+100 -82
View File
@@ -61,7 +61,7 @@ use windmill_common::{
schedule::Schedule,
scripts::{ScriptHash, ScriptLang},
users::{SUPERADMIN_NOTIFICATION_EMAIL, SUPERADMIN_SECRET_EMAIL},
utils::report_critical_error,
utils::{not_found_if_none, report_critical_error},
worker::{to_raw_value, DEFAULT_TAGS_PER_WORKSPACE, NO_LOGS, WORKER_CONFIG},
BASE_URL, DB, METRICS_ENABLED,
};
@@ -760,7 +760,7 @@ pub async fn add_completed_job<
}
}
if queued_job.concurrent_limit.is_some() {
let concurrency_key = concurrency_key(db, queued_job).await;
let concurrency_key = concurrency_key(db, queued_job).await?;
if let Err(e) = sqlx::query_scalar!(
"UPDATE concurrency_counter SET job_uuids = job_uuids - $2 WHERE concurrency_id = $1",
concurrency_key,
@@ -773,17 +773,17 @@ pub async fn add_completed_job<
}
if let Err(e) = sqlx::query_scalar!(
"INSERT INTO custom_concurrency_key_ended VALUES ($1)",
concurrency_key,
"UPDATE concurrency_key SET ended_at = now() WHERE job_id = $1",
queued_job.id,
)
.execute(&mut tx)
.await
.map_err(|e| {
Error::InternalErr(format!(
"Error inserting into custom_concurrency_key_ended for key {concurrency_key}: {e}"
"Error updating to add ended_at timestamp concurrency_key={concurrency_key}: {e}"
))
}) {
tracing::error!("Could not insert into custom_concurrency_key_ended: {}", e);
tracing::error!("Could not update concurrency_key: {}", e);
}
tracing::debug!("decremented concurrency counter");
}
@@ -963,6 +963,7 @@ pub async fn add_completed_job<
JobPayload::ScriptHash {
hash,
path: queued_job.script_path().to_string(),
custom_concurrency_key: custom_concurrency_key(db, queued_job.id).await?,
concurrent_limit: queued_job.concurrent_limit,
concurrency_time_window_s: queued_job.concurrency_time_window_s,
cache_ttl: queued_job.cache_ttl,
@@ -1644,7 +1645,7 @@ pub async fn pull<R: rsmq_async::RsmqConnection + Send + Clone>(
// Else the job is subject to concurrency limits
let job_script_path = pulled_job.script_path.clone().unwrap();
let job_concurrency_key = concurrency_key(db, &pulled_job).await;
let job_concurrency_key = concurrency_key(db, &pulled_job).await?;
tracing::debug!("Concurrency key is '{}'", job_concurrency_key);
let job_custom_concurrent_limit = pulled_job.concurrent_limit.unwrap();
// setting concurrency_time_window to 0 will count only the currently running jobs
@@ -1687,7 +1688,7 @@ pub async fn pull<R: rsmq_async::RsmqConnection + Send + Clone>(
tracing::debug!("running_job: {}", running_job.unwrap_or(0));
let completed_count = sqlx::query!(
"SELECT COUNT(*) as count, COALESCE(MAX(ended_at), now() - INTERVAL '1 second' * $2) as max_ended_at FROM custom_concurrency_key_ended WHERE key = $1 AND ended_at >= (now() - INTERVAL '1 second' * $2)",
"SELECT COUNT(*) as count, COALESCE(MAX(ended_at), now() - INTERVAL '1 second' * $2) as max_ended_at FROM concurrency_key WHERE key = $1 AND ended_at >= (now() - INTERVAL '1 second' * $2)",
job_concurrency_key,
f64::from(job_custom_concurrency_time_window_s),
).fetch_one(&mut tx).await.map_err(|e| {
@@ -1993,66 +1994,64 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<
Ok(job)
}
async fn concurrency_key(db: &Pool<Postgres>, queued_job: &QueuedJob) -> String {
let r = if queued_job.is_flow() {
sqlx::query_scalar!(
"SELECT value->>'concurrency_key' FROM flow WHERE path = $1 AND workspace_id = $2",
queued_job.script_path,
queued_job.workspace_id
)
.fetch_one(db)
pub async fn custom_concurrency_key(
db: &Pool<Postgres>,
job_id: Uuid,
) -> Result<Option<String>, sqlx::Error> {
sqlx::query_scalar!("SELECT key FROM concurrency_key WHERE job_id = $1", job_id)
.fetch_optional(db) // this should no longer be fetch optional
.await
} else {
sqlx::query_scalar!(
"SELECT concurrency_key FROM script WHERE hash = $1 AND workspace_id = $2",
queued_job.script_hash.unwrap_or(ScriptHash(0)).0,
queued_job.workspace_id
)
.fetch_one(db)
.await
};
process_custom_concurrency_key(queued_job, r).await
}
async fn process_custom_concurrency_key(
async fn concurrency_key(
db: &Pool<Postgres>,
queued_job: &QueuedJob,
concurrency_key: Result<Option<String>, sqlx::Error>,
) -> windmill_common::error::Result<String> {
not_found_if_none(custom_concurrency_key(db, queued_job.id).await?, "ConcurrencyKey", queued_job.id.to_string())
}
fn interpolate_args<T: Serialize>(
x: String,
args: &T,
workspace_id: &str,
parsed_args: &mut Option<serde_json::Value>,
) -> String {
match concurrency_key {
Ok(Some(custom_concurrency_key)) => {
let workspaced =
custom_concurrency_key.replace("$workspace", queued_job.workspace_id.as_str());
if RE_ARG_TAG.is_match(&workspaced) {
let mut interpolated = workspaced.clone();
for cap in RE_ARG_TAG.captures_iter(&workspaced) {
let arg_name = cap.get(1).unwrap().as_str();
let arg_value = match queued_job.args.as_ref() {
Some(Json(args_map_json)) => match args_map_json.get(arg_name) {
Some(arg_value_raw) => {
serde_json::to_string(arg_value_raw).unwrap_or_default()
}
None => "".to_string(),
},
None => "".to_string(),
};
interpolated = interpolated
.replace(format!("$args[{}]", arg_name).as_str(), arg_value.as_str());
}
interpolated
} else {
workspaced
}
}
Ok(None) => queued_job.full_path_with_workspace(),
_ => {
tracing::warn!(
"Unable to retrieve concurrency key for script {:?} | {:?}",
queued_job.script_path,
queued_job.script_hash
);
queued_job.full_path_with_workspace()
}
// Save this value to avoid parsing twice
if parsed_args.is_none() {
*parsed_args = Some(serde_json::to_value(args).unwrap_or_default());
}
let value = parsed_args.as_ref().unwrap();
let workspaced = x.as_str().replace("$workspace", workspace_id).to_string();
if RE_ARG_TAG.is_match(&workspaced) {
let mut interpolated = workspaced.clone();
for cap in RE_ARG_TAG.captures_iter(&workspaced) {
let arg_name = cap.get(1).unwrap().as_str();
let arg_value = value
.get(arg_name)
.and_then(|x| x.as_str())
.unwrap_or_default();
interpolated = interpolated.replace(format!("$args[{}]", arg_name).as_str(), arg_value);
}
interpolated
} else {
workspaced
}
}
fn fullpath_with_workspace(workspace_id: &str, script_path: Option<&String>, job_kind: &JobKind) -> String {
let path = script_path
.map(String::as_str)
.unwrap_or("tmp/main");
let is_flow = matches!(
job_kind,
&JobKind::Flow | &JobKind::FlowPreview | &JobKind::SingleScriptFlow
);
format!(
"{}/{}/{}",
workspace_id,
if is_flow { "flow" } else { "script" },
path,
)
}
#[derive(FromRow)]
@@ -2369,6 +2368,7 @@ pub async fn delete_job<'c, R: rsmq_async::RsmqConnection + Clone + Send>(
if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
QUEUE_DELETE_COUNT.inc();
}
let job_removed = sqlx::query_scalar!(
"DELETE FROM queue WHERE workspace_id = $1 AND id = $2 RETURNING 1",
w_id,
@@ -2892,6 +2892,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
raw_flow,
flow_status,
language,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
cache_ttl,
@@ -2901,6 +2902,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
JobPayload::ScriptHash {
hash,
path,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
cache_ttl,
@@ -2915,6 +2917,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
None,
None,
Some(language),
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
cache_ttl,
@@ -2940,6 +2943,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
None,
None,
None,
None,
)
}
JobPayload::Code(RawCode {
@@ -2948,6 +2952,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
hash,
language,
lock,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
cache_ttl,
@@ -2960,6 +2965,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
None,
None,
Some(language),
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
cache_ttl,
@@ -2977,6 +2983,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
None,
None,
None,
None,
dedicated_worker,
None,
),
@@ -2993,6 +3000,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
None,
None,
None,
None,
),
JobPayload::FlowDependencies { path, dedicated_worker } => {
let value_json = fetch_scalar_isolated!(
@@ -3020,6 +3028,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
None,
None,
None,
None,
dedicated_worker,
None,
)
@@ -3037,6 +3046,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
None,
None,
None,
None,
),
JobPayload::RawFlow { mut value, path, restarted_from } => {
add_virtual_items_if_necessary(&mut value.modules);
@@ -3086,6 +3096,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
Some(value.clone()),
Some(flow_status),
None,
value.concurrency_key.clone(),
value.concurrent_limit.clone(),
value.concurrency_time_window_s,
value.cache_ttl.map(|x| x as i32),
@@ -3098,6 +3109,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
hash,
retry,
args,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
cache_ttl,
@@ -3136,7 +3148,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
skip_expr: None,
cache_ttl: cache_ttl.map(|val| val as u32),
early_return: None,
concurrency_key: None,
concurrency_key: custom_concurrency_key.clone(),
priority: priority,
};
(
@@ -3147,6 +3159,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
Some(flow_value.clone()),
Some(FlowStatus::new(&flow_value)), // this is a new flow being pushed, flow_status is set to flow_value
None,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
cache_ttl,
@@ -3172,6 +3185,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
let priority = value.priority;
add_virtual_items_if_necessary(&mut value.modules);
let cache_ttl = value.cache_ttl.map(|x| x as i32).clone();
let custom_concurrency_key = value.concurrency_key.clone();
let concurrency_time_window_s = value.concurrency_time_window_s.clone();
let concurrent_limit = value.concurrent_limit.clone();
let status = Some(FlowStatus::new(&value));
@@ -3183,6 +3197,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
Some(value),
status, // this is a new flow being pushed, flow_status is set to flow_value
None,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
cache_ttl,
@@ -3238,6 +3253,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
Some(raw_flow.clone()),
Some(restarted_flow_status),
None,
raw_flow.concurrency_key,
raw_flow.concurrent_limit,
raw_flow.concurrency_time_window_s,
raw_flow.cache_ttl.map(|x| x as i32),
@@ -3258,6 +3274,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
None,
None,
None,
None,
),
JobPayload::Identity => (
None,
@@ -3272,6 +3289,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
None,
None,
None,
None,
),
JobPayload::Noop => (
None,
@@ -3286,6 +3304,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
None,
None,
None,
None,
),
};
@@ -3336,6 +3355,8 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
let per_workspace: bool = DEFAULT_TAGS_PER_WORKSPACE.load(std::sync::atomic::Ordering::Relaxed);
let mut parsed_args: Option<serde_json::Value> = None;
let tag = if dedicated_worker.is_some_and(|x| x) {
format!(
"{}:{}{}",
@@ -3358,25 +3379,8 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
tag = None;
}
let interpolated_tag = tag.map(|x| {
let workspaced = x.as_str().replace("$workspace", workspace_id).to_string();
if RE_ARG_TAG.is_match(&workspaced) {
let mut interpolated = workspaced.clone();
for cap in RE_ARG_TAG.captures_iter(&workspaced) {
let arg_name = cap.get(1).unwrap().as_str();
let value = serde_json::to_value(&args).unwrap_or_default();
let arg_value = value
.get(arg_name)
.and_then(|x| x.as_str())
.unwrap_or_default();
interpolated =
interpolated.replace(format!("$args[{}]", arg_name).as_str(), arg_value);
}
interpolated
} else {
workspaced
}
});
let interpolated_tag =
tag.map(|x| interpolate_args(x, &args, workspace_id, &mut parsed_args));
let default = || {
let ntag = if job_kind == JobKind::Flow
@@ -3442,6 +3446,20 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
Ulid::new().into()
};
if concurrent_limit.is_some() {
let concurrency_key = custom_concurrency_key
.map(|x| interpolate_args(x, &args, workspace_id, &mut parsed_args))
.unwrap_or(fullpath_with_workspace(workspace_id, script_path.as_ref(), &job_kind));
sqlx::query!(
"INSERT INTO concurrency_key(key, job_id) VALUES ($1, $2)",
concurrency_key,
job_id,
)
.execute(&mut tx)
.await
.map_err(|e| Error::InternalErr(format!("Could not insert concurrency_key={concurrency_key} for job_id={job_id} script_path={script_path:?} workspace_id={workspace_id}: {e}")))?;
}
let uuid = sqlx::query_scalar!(
"INSERT INTO queue
(workspace_id, id, running, parent_job, created_by, permissioned_as, scheduled_for,
backend/windmill-queue/src/schedule.rs
+3
View File
@@ -98,6 +98,7 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
let (
hash,
tag,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
cache_ttl,
@@ -131,6 +132,7 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
hash: hash,
retry: parsed_retry,
args: static_args,
custom_concurrency_key,
concurrent_limit: concurrent_limit,
concurrency_time_window_s: concurrency_time_window_s,
cache_ttl: cache_ttl,
@@ -145,6 +147,7 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
JobPayload::ScriptHash {
hash,
path: schedule.script_path.clone(),
custom_concurrency_key,
concurrent_limit: concurrent_limit,
concurrency_time_window_s: concurrency_time_window_s,
cache_ttl: cache_ttl,
backend/windmill-worker/src/worker.rs
+6 -2
View File
@@ -2207,6 +2207,7 @@ async fn queue_init_bash_maybe<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
path: Some(format!("init_script_{worker_name}")),
language: ScriptLang::Bash,
lock: None,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None,
@@ -3682,8 +3683,8 @@ async fn trigger_python_dependents_to_recompute_dependencies<
JobPayload::Dependencies {
path: s.clone(),
hash: r.0,
language: r.5,
dedicated_worker: r.6,
language: r.6,
dedicated_worker: r.7,
},
args,
&created_by,
@@ -3852,6 +3853,7 @@ async fn lock_modules(
language,
input_transforms,
tag,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
} = e.value.clone()
@@ -3976,6 +3978,7 @@ async fn lock_modules(
content,
language,
tag,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
};
@@ -3997,6 +4000,7 @@ async fn lock_modules(
content,
language,
tag,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
};
backend/windmill-worker/src/worker_flow.rs
+8
View File
@@ -2506,6 +2506,7 @@ async fn compute_next_flow_transform(
language,
lock,
tag,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
..
@@ -2518,6 +2519,7 @@ async fn compute_next_flow_transform(
content,
language,
lock,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
module,
@@ -3093,6 +3095,7 @@ async fn payload_from_simple_module(
language,
lock,
tag,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
..
@@ -3101,6 +3104,7 @@ async fn payload_from_simple_module(
content,
language,
lock,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
module,
@@ -3116,6 +3120,7 @@ fn raw_script_to_payload(
content: &String,
language: &windmill_common::scripts::ScriptLang,
lock: &Option<String>,
custom_concurrency_key: &Option<String>,
concurrent_limit: &Option<i32>,
concurrency_time_window_s: &Option<i32>,
module: &FlowModule,
@@ -3129,6 +3134,7 @@ fn raw_script_to_payload(
content: content.clone(),
language: language.clone(),
lock: lock.clone(),
custom_concurrency_key: custom_concurrency_key.clone(),
concurrent_limit: *concurrent_limit,
concurrency_time_window_s: *concurrency_time_window_s,
cache_ttl: module.cache_ttl.map(|x| x as i32),
@@ -3160,6 +3166,7 @@ async fn script_to_payload(
let mut tx: sqlx::Transaction<'_, sqlx::Postgres> = db.begin().await?;
let (
tag,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
cache_ttl,
@@ -3173,6 +3180,7 @@ async fn script_to_payload(
JobPayload::ScriptHash {
hash,
path: script_path.to_owned(),
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
cache_ttl: module.cache_ttl.map(|x| x as i32).ok_or(cache_ttl).ok(),
frontend/src/lib/components/ConcurrentJobsChart.svelte
+248
View File
@@ -0,0 +1,248 @@
<script lang="ts">
import { Line } from 'svelte-chartjs'
import 'chartjs-adapter-date-fns'
import zoomPlugin from 'chartjs-plugin-zoom'
import {
Chart as ChartJS,
CategoryScale,
Legend,
LineElement,
LinearScale,
PointElement,
TimeScale,
Title,
Tooltip
} from 'chart.js'
import type { ConcurrencyIntervals } from '$lib/gen'
import { createEventDispatcher } from 'svelte'
import { getDbClockNow } from '$lib/forLater'
export let concurrencyIntervals: ConcurrencyIntervals | undefined = undefined
export let maxIsNow: boolean = false
export let minTimeSet: string | undefined = undefined
export let maxTimeSet: string | undefined = undefined
const dispatch = createEventDispatcher()
function calculateTimeSeries(concurrencyIntervals: ConcurrencyIntervals): AggregatedInterval[] {
const timeline = new Map<number, { count: number; id_started: string[]; id_ended: string[] }>()
concurrencyIntervals.completed_jobs?.forEach(({ job_id, started_at, ended_at }) => {
if (started_at != undefined && ended_at != undefined) {
const startTime = new Date(started_at).getTime()
const endTime = new Date(ended_at).getTime()
if (!timeline.has(startTime))
timeline.set(startTime, { count: 0, id_started: [], id_ended: [] })
if (!timeline.has(endTime))
timeline.set(endTime, { count: 0, id_started: [], id_ended: [] })
const s = timeline.get(startTime)!
const e = timeline.get(endTime)!
s.count += 1
s.id_started.push(job_id ?? 'unknown')
e.count -= 1
e.id_ended.push(job_id ?? 'unknown')
}
})
concurrencyIntervals.running_jobs?.forEach(({ job_id, started_at }) => {
if (started_at != undefined) {
const startTime = new Date(started_at).getTime()
if (!timeline.has(startTime))
timeline.set(startTime, { count: 0, id_started: [], id_ended: [] })
const s = timeline.get(startTime)!
s.count += 1
s.id_started.push(job_id ?? 'unknown')
}
})
let count = 0
const result: AggregatedInterval[] = []
for (const [time, change] of [...timeline.entries()].sort(
([time1], [time2]) => time1 - time2
)) {
count += change.count
let msg = ''
msg += change.id_started.length != 0 ? `${change.id_started.join(',')} started` : ''
msg += change.id_started.length != 0 && change.id_ended.length != 0 ? '\n' : ''
msg += change.id_ended.length != 0 ? `${change.id_ended.join(',')} ended` : ''
result.push({ time: new Date(time), count, msg } as AggregatedInterval)
}
// Add points to continue the line towards the extremities
if (result.length > 0) {
let start_time = addSeconds(new Date(result[0].time), -1)
let start_count = 0
let end_count = result[result.length - 1].count
result.unshift({
time: start_time,
count: start_count
} as AggregatedInterval)
result.push({
time: new Date(),
count: end_count
} as AggregatedInterval)
}
return result
}
type AggregatedInterval = { time: Date; count: number; msg?: string }
let intervals: AggregatedInterval[] | undefined = undefined
$: intervals = concurrencyIntervals ? calculateTimeSeries(concurrencyIntervals) : undefined
ChartJS.register(
Title,
Tooltip,
Legend,
zoomPlugin,
LineElement,
CategoryScale,
LinearScale,
PointElement,
TimeScale
)
$: data = {
datasets: [
{
borderColor: '#4ade80',
backgroundColor: '#f8717100',
pointRadius: 0,
label: 'running',
showLine: true,
stepped: true,
data:
intervals?.map((job) => ({
x: job.time as any,
y: job.count,
id: job.msg
})) ?? []
}
]
}
const zoomOptions = {
pan: {
enabled: true,
modifierKey: 'ctrl' as 'ctrl',
onPanComplete: ({ chart }) => {
dispatch('zoom', {
min: addSeconds(new Date(chart.scales.x.min), -1),
max: addSeconds(new Date(chart.scales.x.max), 1)
})
}
},
zoom: {
drag: {
enabled: true
},
mode: 'x' as 'x',
onZoom: ({ chart }) => {
dispatch('zoom', {
min: addSeconds(new Date(chart.scales.x.min), -1),
max: addSeconds(new Date(chart.scales.x.max), 1)
})
}
}
}
let minTime = addSeconds(new Date(), -300)
let maxTime = getDbClockNow()
$: computeMinMaxTime(intervals, minTimeSet, maxTimeSet)
function minJobTime(intervals: AggregatedInterval[]): Date {
return intervals[0].time
}
function maxJobTime(intervals: AggregatedInterval[]): Date {
return intervals[intervals?.length - 1].time
}
function computeMinMaxTime(
intervals: AggregatedInterval[] | undefined,
minTimeSet: string | undefined,
maxTimeSet: string | undefined
) {
let minTimeSetDate = minTimeSet ? new Date(minTimeSet) : undefined
let maxTimeSetDate = maxTimeSet ? new Date(maxTimeSet) : undefined
if (minTimeSetDate && maxTimeSetDate) {
minTime = minTimeSetDate
maxTime = maxTimeSetDate
return
}
if (intervals == undefined || intervals?.length == 0) {
minTime = minTimeSetDate ?? addSeconds(new Date(), -300)
maxTime = maxTimeSetDate ?? getDbClockNow()
return
}
const maxJob = maxIsNow ? getDbClockNow() : maxJobTime(intervals)
const minJob = minJobTime(intervals)
const diff = (maxJob.getTime() - minJob.getTime()) / 20000
minTime = minTimeSetDate ?? addSeconds(minJob, -diff)
if (maxIsNow) {
maxTime = maxTimeSetDate ?? maxJob
} else {
maxTime = maxTimeSetDate ?? addSeconds(maxJob, diff)
}
}
function addSeconds(date: Date, seconds: number): Date {
date.setTime(date.getTime() + seconds * 1000)
return date
}
$: options = {
responsive: true,
maintainAspectRatio: false,
plugins: {
zoom: zoomOptions,
legend: {
display: false
},
tooltip: {
callbacks: {
footer: function (context) {
return context[context.length - 1].raw.id
}
}
}
},
scales: {
x: {
type: 'time',
grid: {
display: false
},
min: minTime,
max: maxTime
},
y: {
grid: {
display: false
},
title: {
display: true,
text: 'concurrent jobs'
},
beginAtZero: true,
ticks: {
stepSize: 1
}
}
},
animation: false,
interaction: {
intersect: false,
mode: 'index'
}
} as any
</script>
<div class="relative max-h-40">
<Line {data} {options} />
</div>
frontend/src/lib/components/FlowMetadata.svelte
+1 -1
View File
@@ -1,5 +1,5 @@
<script lang="ts">
import type { Job } from '$lib/gen'
import { type Job } from '$lib/gen'
import JobStatus from '$lib/components/JobStatus.svelte'
import { displayDate } from '$lib/utils'
import ScheduleEditor from './ScheduleEditor.svelte'
frontend/src/lib/components/SavedInputs.svelte
+1
View File
@@ -174,6 +174,7 @@
user={null}
label={null}
folder={null}
concurrencyKey={null}
success="running"
argFilter={undefined}
bind:loading
frontend/src/lib/components/common/toggleButton-v2/ToggleButtonMore.svelte
+52
View File
@@ -0,0 +1,52 @@
<script lang="ts">
import { getContext } from 'svelte'
import { Tab } from '@rgossiaux/svelte-headlessui'
import type { ToggleButtonContext } from './ToggleButtonGroup.svelte'
import { twMerge } from 'tailwind-merge'
import Popover from '$lib/components/Popover.svelte'
import DropdownV2 from '$lib/components/DropdownV2.svelte'
export let disabled: boolean = false
export let small = false
export let light = false
export let id: string | undefined = undefined
type TogglableItem = {
label: string
value: string
}
export let togglableItems: TogglableItem[]
const { select, selected } = getContext<ToggleButtonContext>('ToggleButtonGroup')
let items = togglableItems.map((i) => ({ displayName: i.label, action: () => select(i.value) }))
function isAnOptionSelected(selected: string) {
return togglableItems.some((i) => i.value === selected)
}
</script>
<Popover
disablePopup={true}
notClickable
class={twMerge('flex', disabled ? 'cursor-not-allowed' : 'cursor-pointer')}
disappearTimeout={0}
>
<div {id} class="flex">
<Tab
{disabled}
class={twMerge(
' rounded-md transition-all text-xs flex gap-1 flex-row items-center',
small ? 'px-1.5 py-0.5 text-2xs' : 'px-2 py-1',
light ? 'font-medium' : '',
isAnOptionSelected($selected)
? 'bg-surface shadow-md'
: 'bg-surface-secondary hover:bg-surface-hover',
$$props.class
)}
>
<DropdownV2 {items} />
</Tab>
</div>
</Popover>
frontend/src/lib/components/runs/JobLoader.svelte
+132 -4
View File
@@ -1,6 +1,12 @@
<script lang="ts">
import { onDestroy, onMount } from 'svelte'
import { JobService, type Job, type CompletedJob } from '$lib/gen'
import {
JobService,
type Job,
type CompletedJob,
type ConcurrencyIntervals,
ConcurrencyGroupsService
} from '$lib/gen'
import { sendUserToast } from '$lib/toast'
import { workspaceStore } from '$lib/stores'
@@ -26,6 +32,9 @@
export let queue_count: Tweened<number> | undefined = undefined
export let autoRefresh: boolean = true
export let completedJobs: CompletedJob[] | undefined = undefined
export let externalJobs: Job[] | undefined = undefined
export let concurrencyKey: string | null
export let concurrencyIntervals: ConcurrencyIntervals | undefined = undefined
export let argError = ''
export let resultError = ''
export let loading: boolean = false
@@ -36,6 +45,7 @@
let intervalId: NodeJS.Timeout | undefined
let sync = true
let concurrencyKeyMap: Map<string, string> = new Map<string, string>()
$: jobKinds = computeJobKinds(jobKindsCat)
$: ($workspaceStore && loadJobsIntern(true)) ||
@@ -44,6 +54,7 @@
success &&
isSkipped != undefined &&
jobKinds &&
concurrencyKey &&
user &&
folder &&
showFutureJobs != undefined &&
@@ -123,6 +134,39 @@
})
}
async function fetchConcurrencyIntervals(
concurrencyKey: string | null,
startedBefore: string | undefined,
startedAfter: string | undefined
): Promise<ConcurrencyIntervals> {
return ConcurrencyGroupsService.getConcurrencyIntervals({
rowLimit: 1000,
concurrencyKey: concurrencyKey == null || concurrencyKey == '' ? undefined : concurrencyKey,
workspace: $workspaceStore!,
createdOrStartedBefore: startedBefore,
createdOrStartedAfter: startedAfter,
schedulePath,
scriptPathExact: path === null || path === '' ? undefined : path,
createdBy: user === null || user === '' ? undefined : user,
scriptPathStart: folder === null || folder === '' ? undefined : `f/${folder}/`,
jobKinds,
success: success == 'success' ? true : success == 'failure' ? false : undefined,
running: success == 'running' ? true : undefined,
isSkipped: isSkipped ? undefined : false,
isFlowStep: jobKindsCat != 'all' ? false : undefined,
label: label === null || label === '' ? undefined : label,
isNotSchedule: showSchedules == false ? true : undefined,
scheduledForBeforeNow: showFutureJobs == false ? true : undefined,
args:
argFilter && argFilter != '{}' && argFilter != '' && argError == '' ? argFilter : undefined,
result:
resultFilter && resultFilter != '{}' && resultFilter != '' && resultError == ''
? resultFilter
: undefined,
allWorkspaces: allWorkspaces ? true : undefined
})
}
export async function loadJobs(
nMinTs: string | undefined,
nMaxTs: string | undefined,
@@ -134,6 +178,8 @@
if (reset) {
jobs = undefined
completedJobs = undefined
externalJobs = undefined
concurrencyIntervals = undefined
intervalId && clearInterval(intervalId)
intervalId = setInterval(syncer, refreshRate)
}
@@ -145,7 +191,11 @@
}
loading = true
try {
jobs = await fetchJobs(maxTs, minTs)
concurrencyIntervals = await fetchConcurrencyIntervals(concurrencyKey, maxTs, undefined)
updateConcurrencyKeyMap()
computeExternalJobs(minTs)
let j = await fetchJobs(maxTs, minTs)
jobs = await filterJobsByConcurrencyKey(j, minTs)
computeCompletedJobs()
} catch (err) {
sendUserToast(`There was a problem fetching jobs: ${err}`, true)
@@ -206,7 +256,11 @@
}
loading = true
const newJobs = await fetchJobs(maxTs, minTs ?? ts)
concurrencyIntervals = await fetchConcurrencyIntervals(concurrencyKey, maxTs, undefined)
updateConcurrencyKeyMap()
computeExternalJobs(minTs)
let newJobs = await fetchJobs(maxTs, minTs ?? ts)
newJobs = (await filterJobsByConcurrencyKey(newJobs, minTs)) ?? []
if (newJobs && newJobs.length > 0 && jobs) {
const oldJobs = jobs?.map((x) => x.id)
jobs = newJobs.filter((x) => !oldJobs.includes(x.id)).concat(jobs)
@@ -216,7 +270,6 @@
jobs = jobs
computeCompletedJobs()
}
loading = false
}
}
@@ -228,6 +281,15 @@
jobs?.filter((x) => x.type == 'CompletedJob').map((x) => x as CompletedJob) ?? []
}
async function filterJobsByConcurrencyKey(jobs: Job[] | undefined, minTs: string | undefined) {
if (concurrencyKey == null || concurrencyKey === '' || jobs == undefined || jobs.length === 0)
return jobs
let minDate = minTs ? new Date(minTs) : undefined
return jobs.filter((x) => concurrencyKeyMap.get(x.id) === concurrencyKey && (!minDate || (x.started_at && minDate < new Date(x.started_at))))
}
function onVisibilityChange() {
if (document.hidden) {
sync = false
@@ -236,6 +298,72 @@
}
}
function updateConcurrencyKeyMap() {
for (const vec of [concurrencyIntervals?.running_jobs, concurrencyIntervals?.completed_jobs]) {
if (vec == undefined) continue
for (const interval of vec) {
if (
interval.job_id &&
interval.concurrency_key &&
concurrencyKeyMap.get(interval.job_id) == undefined
) {
concurrencyKeyMap.set(interval.job_id, interval.concurrency_key)
}
}
}
}
function computeExternalJobs(minTs: string | undefined) {
let minDate = minTs ? new Date(minTs) : undefined
let externalQueued = concurrencyIntervals?.running_jobs
.filter((x) => !x.job_id && (!minDate || (x.started_at && minDate < new Date(x.started_at))))
.map(
(x) =>
({
id: '-',
type: 'QueuedJob',
started_at: x.started_at,
running: x.started_at != undefined,
script_path: '-'
} as Job)
)
let externalCompleted = concurrencyIntervals?.completed_jobs
.filter((x) => !x.job_id && (!minDate || (x.started_at && minDate < new Date(x.started_at))))
.map(
(x) =>
({
type: 'CompletedJob',
started_at: x.started_at,
running: x.started_at != undefined,
id: '-',
script_path: '-',
created_by: '-',
created_at: '-',
success: false,
canceled: false,
is_flow_step: false,
is_skipped: false,
visible_to_owner: false,
email: '-',
permissioned_as: '-',
tag: '-',
job_kind: 'flow',
duration_ms:
x.ended_at && x.started_at
? new Date(x.ended_at).getTime() - new Date(x.started_at).getTime()
: 0
} as Job)
)
let ret: Job[] = []
if (externalQueued) {
ret = ret.concat(externalQueued)
}
if (externalCompleted) {
ret = ret.concat(externalCompleted)
}
externalJobs = ret
}
onMount(() => {
document.addEventListener('visibilitychange', onVisibilityChange)
frontend/src/lib/components/runs/JobPreview.svelte
+20 -1
View File
@@ -1,5 +1,5 @@
<script lang="ts">
import { type Job, type WorkflowStatus } from '../../gen'
import { ConcurrencyGroupsService, type Job, type WorkflowStatus } from '../../gen'
import TestJobLoader from '../TestJobLoader.svelte'
import DisplayResult from '../DisplayResult.svelte'
import JobArgs from '../JobArgs.svelte'
@@ -12,6 +12,8 @@
import DurationMs from '../DurationMs.svelte'
import { workspaceStore } from '$lib/stores'
import WorkflowTimeline from '../WorkflowTimeline.svelte'
import Popover from '../Popover.svelte'
import { truncateRev } from '$lib/utils'
export let id: string
export let blankLink = false
@@ -38,6 +40,14 @@
$: job?.logs == undefined && job && viewTab == 'logs' && getLogs?.()
let lastJobId: string | undefined = undefined
let concurrencyKey: string | undefined = undefined
$: job?.id && lastJobId !== job.id && getConcurrencyKey(job)
async function getConcurrencyKey(job: Job) {
lastJobId = job.id
concurrencyKey = await ConcurrencyGroupsService.getConcurrencyKey({ id: job.id })
}
let viewTab = 'result'
function asWorkflowStatus(x: any): Record<string, WorkflowStatus> {
@@ -84,6 +94,15 @@
<Badge baseClass="text-2xs">Label: {label}</Badge>
{/each}
{/if}
{#if concurrencyKey}
<Popover notClickable>
<svelte:fragment slot="text">
This jobs has concurrency limits enabled with the key
{concurrencyKey}
</svelte:fragment>
<Badge>Concurrency: {truncateRev(concurrencyKey, 20)}</Badge>
</Popover>
{/if}
</div>
<a
href="/run/{job?.id}?workspace={job?.workspace_id}"
frontend/src/lib/components/runs/RunRow.svelte
+39 -26
View File
@@ -14,6 +14,7 @@
Hourglass,
ListFilter,
Play,
ShieldQuestion,
X
} from 'lucide-svelte'
import { createEventDispatcher } from 'svelte'
@@ -31,6 +32,8 @@
export let activeLabel: string | null
let scheduleEditor: ScheduleEditor
$: isExternal = job && job.id === '-'
</script>
<Portal>
@@ -42,7 +45,7 @@
<div
class={twMerge(
'hover:bg-surface-hover cursor-pointer',
selectedId === job.id ? 'bg-blue-50 dark:bg-blue-900/50' : '',
selectedId === job.id && !isExternal ? 'bg-blue-50 dark:bg-blue-900/50' : '',
'flex flex-row items-center h-full'
)}
style="width: {containerWidth}px"
@@ -51,7 +54,11 @@
}}
>
<div class="w-1/12 flex justify-center">
{#if 'success' in job && job.success}
{#if isExternal}
<Badge color="gray" baseClass="!px-1.5">
<ShieldQuestion size={14} />
</Badge>
{:else if 'success' in job && job.success}
{#if job.is_skipped}
<Badge color="green" rounded>
<FastForward size={14} />
@@ -108,21 +115,25 @@
<div class="whitespace-nowrap text-xs font-semibold truncate">
{#if job.script_path}
<div class="flex flex-row gap-1 items-center">
<a
href="/run/{job.id}?workspace={job.workspace_id}"
class="truncate w-30 dark:text-blue-400"
>
{job.script_path}
</a>
<Button
size="xs2"
color="light"
on:click={() => {
dispatch('filterByPath', job.script_path)
}}
>
<ListFilter size={10} />
</Button>
{#if isExternal}
<span class="w-30 justify-center">-</span>
{:else}
<a
href="/run/{job.id}?workspace={job.workspace_id}"
class="truncate w-30 dark:text-blue-400"
>
{job.script_path}
</a>
<Button
size="xs2"
color="light"
on:click={() => {
dispatch('filterByPath', job.script_path)
}}
>
<ListFilter size={10} />
</Button>
{/if}
{#if job.script_path?.startsWith('f/')}
<Button
size="xs2"
@@ -218,15 +229,17 @@
<div class="text-xs">
{job.created_by}
</div>
<Button
size="xs2"
color="light"
on:click={() => {
dispatch('filterByUser', job.created_by)
}}
>
<ListFilter size={10} />
</Button>
{#if !isExternal}
<Button
size="xs2"
color="light"
on:click={() => {
dispatch('filterByUser', job.created_by)
}}
>
<ListFilter size={10} />
</Button>
{/if}
</div>
{/if}
</div>
frontend/src/lib/components/runs/RunsFilter.svelte
+57 -3
View File
@@ -10,12 +10,14 @@
import Label from '../Label.svelte'
import Section from '../Section.svelte'
import CloseButton from '../common/CloseButton.svelte'
import { workspaceStore } from '$lib/stores'
import { enterpriseLicense, workspaceStore } from '$lib/stores'
import { createEventDispatcher } from 'svelte'
import ToggleButtonMore from '../common/toggleButton-v2/ToggleButtonMore.svelte'
// Filters
export let path: string | null = null
export let label: string | null = null
export let concurrencyKey: string | null = null
export let success: 'running' | 'success' | 'failure' | undefined = undefined
export let isSkipped: boolean | undefined = undefined
export let argFilter: string
@@ -34,11 +36,12 @@
export let allWorkspaces = false
$: displayedLabel = label
$: displayedConcurrencyKey = concurrencyKey
let copyArgFilter = argFilter
let copyResultFilter = resultFilter
export let filterBy: 'path' | 'user' | 'folder' | 'label' = 'path'
export let filterBy: 'path' | 'user' | 'folder' | 'label' | 'concurrencyKey' = 'path'
const dispatch = createEventDispatcher()
@@ -57,10 +60,14 @@
} else if (label !== null && label !== '' && filterBy !== 'label') {
manuallySet = true
filterBy = 'label'
} else if (concurrencyKey !== null && concurrencyKey !== '' && filterBy !== 'concurrencyKey') {
manuallySet = true
filterBy = 'concurrencyKey'
}
}
let labelTimeout: NodeJS.Timeout | undefined = undefined
let concurrencyKeyTimeout: NodeJS.Timeout | undefined = undefined
</script>
<div class="flex gap-4">
@@ -86,6 +93,7 @@
user = null
folder = null
label = null
concurrencyKey = null
} else {
manuallySet = false
}
@@ -94,7 +102,12 @@
<ToggleButton value="path" label="Path" />
<ToggleButton value="user" label="User" />
<ToggleButton value="folder" label="Folder" />
<ToggleButton value="label" label="Label" />
<ToggleButtonMore
togglableItems={[
{ label: 'Concurrency key', value: 'concurrencyKey' },
{ label: 'Label', value: 'label' }
]}
/>
</ToggleButtonGroup>
</div>
@@ -241,6 +254,47 @@
/>
</div>
{/key}
{:else if filterBy === 'concurrencyKey'}
{#key concurrencyKey}
<div class="relative">
{#if concurrencyKey}
<button
class="absolute top-2 right-2 z-50"
on:click={() => {
concurrencyKey = null
dispatch('reset')
}}
>
<X size={14} />
</button>
{/if}
<span class="text-xs absolute -top-4"
>Concurrency Key <Tooltip>
For concurrency limited jobs, the concurrency key defines a group of jobs that share
the same limits.
{#if !$enterpriseLicense}
Concurrency limits are an EE feature.
{/if}
</Tooltip></span
>
<input
autofocus
type="text"
class="!h-[32px] py-1 !text-xs !w-64"
bind:value={displayedConcurrencyKey}
on:keydown={(e) => {
if (concurrencyKeyTimeout) {
clearTimeout(concurrencyKeyTimeout)
}
concurrencyKeyTimeout = setTimeout(() => {
concurrencyKey = displayedConcurrencyKey
}, 1000)
}}
/>
</div>
{/key}
{/if}
</div>
<div class="relative">
frontend/src/lib/components/runs/RunsTable.svelte
+27 -5
View File
@@ -3,9 +3,12 @@
import RunRow from './RunRow.svelte'
import VirtualList from 'svelte-tiny-virtual-list'
import { createEventDispatcher, onMount } from 'svelte'
import Tooltip from '../Tooltip.svelte'
//import InfiniteLoading from 'svelte-infinite-loading'
export let jobs: Job[] | undefined = undefined
export let externalJobs: Job[] | undefined = undefined
export let showExternalJobs: boolean = false
export let selectedId: string | undefined = undefined
export let selectedWorkspace: string | undefined = undefined
export let activeLabel: string | null = null
@@ -64,7 +67,12 @@
return sortedLogs
}
$: groupedJobs = jobs ? groupJobsByDay(jobs) : undefined
$: groupedJobs =
jobs && externalJobs
? showExternalJobs
? groupJobsByDay([...jobs, ...externalJobs])
: groupJobsByDay(jobs)
: undefined
type FlatJobs =
| {
@@ -125,6 +133,13 @@
}
*/
function jobCountString(jobCount: number) {
const jc = jobCount
const isTruncated = jc == 1000
return `${jc}${isTruncated ? '+' : ''} job${jc != 1 ? 's' : ''}`
}
function computeHeight() {
tableHeight = document.querySelector('#runs-table-wrapper')!.parentElement?.clientHeight ?? 0
}
@@ -145,9 +160,15 @@
class="flex flex-row bg-surface-secondary sticky top-0 w-full p-2 pr-4"
bind:clientHeight={header}
>
<div class="w-1/12 text-2xs"
>{jobs?.length == 1000 ? '1000+' : jobs ? jobs.length.toString() : '...'} jobs</div
>
{#if showExternalJobs && externalJobs}
<div class="w-1/12 text-2xs">
{jobs && jobCountString(jobs.length + externalJobs.length)}<Tooltip
>{externalJobs.length} jobs obscured</Tooltip
>
</div>
{:else}
<div class="w-1/12 text-2xs">{jobs && jobCountString(jobs.length)}</div>
{/if}
<div class="w-4/12 text-xs font-semibold">Timestamp</div>
<div class="w-4/12 text-xs font-semibold">Path</div>
{#if containsLabel}
@@ -188,6 +209,7 @@
on:filterByPath
on:filterByUser
on:filterByFolder
on:filterByConcurrencyKey
{containerWidth}
/>
</div>
@@ -206,7 +228,7 @@
</div>
</VirtualList>
</div>
{#if jobs?.length == 0}
{#if jobs?.length == 0 && (!showExternalJobs || externalJobs?.length == 0)}
<tr>
<td colspan="4" class="text-center py-8">
<div class="text-xs text-secondary"> No jobs found for the selected filters. </div>
frontend/src/routes/(root)/(logged)/concurrency_groups/+page.svelte
+7 -34
View File
@@ -3,7 +3,7 @@
import { ConcurrencyGroupsService } from '$lib/gen'
import CenteredPage from '$lib/components/CenteredPage.svelte'
import { Button, Drawer, DrawerContent } from '$lib/components/common'
import { Button } from '$lib/components/common'
import PageHeader from '$lib/components/PageHeader.svelte'
import TableCustom from '$lib/components/TableCustom.svelte'
import { RefreshCw, Trash } from 'lucide-svelte'
@@ -12,8 +12,6 @@
let concurrencyGroups: ConcurrencyGroup[] | undefined = undefined
let selectedGroup: ConcurrencyGroup | undefined = undefined
let groupDrawer: Drawer
let doLoadConcurrencyGroups = false
let concurrencyGroupsLoading = false
@@ -62,27 +60,6 @@
}
</script>
<Drawer bind:this={groupDrawer}>
<DrawerContent
title="Instance Group {selectedGroup?.concurrency_id}"
on:close={groupDrawer.closeDrawer}
>
{#if selectedGroup?.job_uuids && selectedGroup?.job_uuids.length > 0}
<h3 class="mb-2">Jobs running for this group</h3>
<ul>
{#each selectedGroup?.job_uuids as jobUuid}
<li>
{jobUuid}
</li>
{/each}
</ul>
{:else}
<h3>No job running for this group</h3>
{/if}
</DrawerContent>
</Drawer>
<CenteredPage>
<PageHeader title="Concurrency Groups">
<Button
@@ -100,25 +77,21 @@
<div class="relative mb-20 pt-8">
<TableCustom>
<tr slot="header-row">
<th>Concurrency ID</th>
<th>Concurrency key</th>
<th>Jobs running</th>
<th />
</tr>
<tbody slot="body">
{#each concurrencyGroups as { concurrency_id, job_uuids }}
{#each concurrencyGroups as { concurrency_key, total_running }}
<tr>
<td>
<a
href="#{concurrency_id}"
on:click={() => {
selectedGroup = { concurrency_id, job_uuids }
groupDrawer.openDrawer()
}}
>{concurrency_id}
href={`/runs/?job_kinds=all&graph=ConcurrencyChart&concurrency_key=${concurrency_key}`}
>{concurrency_key}
</a>
</td>
<td>
{job_uuids.length}
{total_running}
</td>
<td>
<div class="flex justify-center">
@@ -128,7 +101,7 @@
btnClasses="justify-center w-12"
startIcon={{ icon: Trash, classes: 'text-red-500' }}
on:click={() => {
deleteConcurrencyGroup(concurrency_id)
deleteConcurrencyGroup(concurrency_key)
}}
iconOnly={true}
/>
frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte
+32 -2
View File
@@ -6,7 +6,8 @@
ScriptService,
type Script,
type WorkflowStatus,
type NewScript
type NewScript,
ConcurrencyGroupsService
} from '$lib/gen'
import {
canWrite,
@@ -14,7 +15,8 @@
displayDate,
emptyString,
encodeState,
truncateHash
truncateHash,
truncateRev
} from '$lib/utils'
import BarsStaggered from '$lib/components/icons/BarsStaggered.svelte'
@@ -81,6 +83,7 @@
import Toggle from '$lib/components/Toggle.svelte'
import WorkflowTimeline from '$lib/components/WorkflowTimeline.svelte'
import ScheduleEditor from '$lib/components/ScheduleEditor.svelte'
import Popover from '$lib/components/Popover.svelte'
let job: Job | undefined
let jobUpdateLastFetch: Date | undefined
@@ -101,6 +104,14 @@
$: job?.logs == undefined && job && viewTab == 'logs' && getLogs?.()
let lastJobId: string | undefined = undefined
let concurrencyKey: string | undefined = undefined
$: job?.id && lastJobId !== job.id && getConcurrencyKey(job)
async function getConcurrencyKey(job: Job) {
lastJobId = job.id
concurrencyKey = await ConcurrencyGroupsService.getConcurrencyKey({ id: job.id })
}
async function deleteCompletedJob(id: string): Promise<void> {
await JobService.deleteCompletedJob({ workspace: $workspaceStore!, id })
getJob()
@@ -649,6 +660,25 @@
</div>
{/each}
{/if}
{#if concurrencyKey}
<div>
<Popover notClickable>
<svelte:fragment slot="text">
This jobs has concurrency limits enabled with the key
<a
href={`/runs/?job_kinds=all&graph=ConcurrencyChart&concurrency_key=${concurrencyKey}`}
>
{concurrencyKey}
</a>
</svelte:fragment>
<a
href={`/runs/?job_kinds=all&graph=ConcurrencyChart&concurrency_key=${concurrencyKey}`}
>
<Badge>Concurrency: {truncateRev(concurrencyKey, 20)}</Badge></a
>
</Popover>
</div>
{/if}
</div>
{/if}
</div>
frontend/src/routes/(root)/(logged)/runs/[...path]/+page.svelte
+176 -41
View File
@@ -6,7 +6,8 @@
UserService,
FolderService,
ScriptService,
FlowService
FlowService,
type ConcurrencyIntervals
} from '$lib/gen'
import { page } from '$app/stores'
@@ -30,7 +31,10 @@
import { twMerge } from 'tailwind-merge'
import ManuelDatePicker from '$lib/components/runs/ManuelDatePicker.svelte'
import JobLoader from '$lib/components/runs/JobLoader.svelte'
import { Calendar, Clock } from 'lucide-svelte'
import { AlertTriangle, Calendar, Clock } from 'lucide-svelte'
import ConcurrentJobsChart from '$lib/components/ConcurrentJobsChart.svelte'
import ToggleButtonGroup from '$lib/components/common/toggleButton-v2/ToggleButtonGroup.svelte'
import ToggleButton from '$lib/components/common/toggleButton-v2/ToggleButton.svelte'
let jobs: Job[] | undefined
let selectedId: string | undefined = undefined
@@ -42,6 +46,7 @@
let user: string | null = $page.url.searchParams.get('user')
let folder: string | null = $page.url.searchParams.get('folder')
let label: string | null = $page.url.searchParams.get('label')
let concurrencyKey: string | null = $page.url.searchParams.get('concurrency_key')
// Rest of filters handled by RunsFilter
let success: 'running' | 'success' | 'failure' | undefined = ($page.url.searchParams.get(
'success'
@@ -85,6 +90,7 @@
let usernames: string[] = []
let folders: string[] = []
let completedJobs: CompletedJob[] | undefined = undefined
let concurrencyIntervals: ConcurrencyIntervals | undefined = undefined
let argError = ''
let resultError = ''
let filterTimeout: NodeJS.Timeout | undefined = undefined
@@ -94,6 +100,10 @@
let cancelAllJobs = false
let innerWidth = window.innerWidth
let jobLoader: JobLoader | undefined = undefined
let externalJobs: Job[] | undefined
let graph: 'RunChart' | 'ConcurrencyChart' = typeOfChart($page.url.searchParams.get('graph'))
let graphIsRunsChart: boolean = graph ? graph === 'ConcurrencyChart' : true
let manualDatePicker: ManuelDatePicker
@@ -109,6 +119,8 @@
resultFilter ||
schedulePath ||
jobKindsCat ||
concurrencyKey ||
graph ||
minTs ||
maxTs ||
allWorkspaces ||
@@ -195,6 +207,11 @@
} else {
searchParams.delete('max_ts')
}
if (concurrencyKey) {
searchParams.set('concurrency_key', concurrencyKey)
} else {
searchParams.delete('concurrency_key')
}
if (label) {
searchParams.set('label', label)
@@ -202,6 +219,12 @@
searchParams.delete('label')
}
if (graph != 'RunChart') {
searchParams.set('graph', graph)
} else {
searchParams.delete('graph')
}
let newPath = path ? `/${path}` : '/'
let newUrl = `/runs${newPath}?${searchParams.toString()}`
history.replaceState(history.state, '', newUrl.toString())
@@ -256,6 +279,7 @@
user = null
folder = null
label = null
concurrencyKey = null
}
function filterByUser(e: CustomEvent<string>) {
@@ -263,6 +287,7 @@
folder = null
user = e.detail
label = null
concurrencyKey = null
}
function filterByFolder(e: CustomEvent<string>) {
@@ -270,6 +295,7 @@
user = null
folder = e.detail
label = null
concurrencyKey = null
}
function filterByLabel(e: CustomEvent<string>) {
@@ -277,9 +303,38 @@
user = null
folder = null
label = e.detail
concurrencyKey = null
}
function filterByConcurrencyKey(e: CustomEvent<string>) {
path = null
user = null
folder = null
label = null
concurrencyKey = e.detail
}
let calendarChangeTimeout: NodeJS.Timeout | undefined = undefined
function typeOfChart(s: string | null): 'RunChart' | 'ConcurrencyChart' {
switch (s) {
case 'RunChart':
return 'RunChart'
case 'ConcurrencyChart':
return 'ConcurrencyChart'
default:
return 'RunChart'
}
}
const warnJobLimitMsg =
'The exact number of concurrent job at the beginning of the time range may be incorrect as only the last 1000 jobs are taken into account: a job that was started earlier than this limit will not be taken into account'
$: warnJobLimit =
graph === 'ConcurrencyChart' &&
concurrencyIntervals !== undefined &&
(concurrencyIntervals.running_jobs.length === 1000 ||
concurrencyIntervals.completed_jobs.length === 1000)
</script>
<JobLoader
@@ -304,6 +359,9 @@
bind:queue_count
{autoRefresh}
bind:completedJobs
bind:externalJobs
bind:concurrencyIntervals
{concurrencyKey}
{argError}
{resultError}
bind:loading
@@ -328,7 +386,11 @@
<Drawer bind:this={runDrawer}>
<DrawerContent title="Run details" on:close={runDrawer.closeDrawer}>
{#if selectedId}
<JobPreview blankLink id={selectedId} workspace={selectedWorkspace} />
{#if selectedId === '-'}
<div class="p-4">There is no information available for this job</div>
{:else}
<JobPreview blankLink id={selectedId} workspace={selectedWorkspace} />
{/if}
{/if}
</DrawerContent>
</Drawer>
@@ -339,28 +401,31 @@
<div class="w-full h-screen">
<div class="px-2">
<div class="flex items-center space-x-2 flex-row justify-between">
<div class="flex flex-row flex-wrap justify-between py-2 my-4 px-4 gap-1 items-center">
<h1
class={twMerge(
'!text-2xl font-semibold leading-6 tracking-tight',
$userStore?.operator ? 'pl-10' : ''
)}
>
Runs
</h1>
<div class="flex-col">
<div class="flex flex-row flex-wrap justify-between py-2 my-4 px-4 gap-1 items-center">
<h1
class={twMerge(
'!text-2xl font-semibold leading-6 tracking-tight',
$userStore?.operator ? 'pl-10' : ''
)}
>
Runs
</h1>
<Tooltip
documentationLink="https://www.windmill.dev/docs/core_concepts/monitor_past_and_future_runs"
>
All past and schedule executions of scripts and flows, including previews. You only see
your own runs or runs of groups you belong to unless you are an admin.
</Tooltip>
<Tooltip
documentationLink="https://www.windmill.dev/docs/core_concepts/monitor_past_and_future_runs"
>
All past and schedule executions of scripts and flows, including previews. You only
see your own runs or runs of groups you belong to unless you are an admin.
</Tooltip>
</div>
</div>
<RunsFilter
bind:isSkipped
bind:user
bind:folder
bind:label
bind:concurrencyKey
bind:path
bind:success
bind:argFilter
@@ -378,17 +443,51 @@
</div>
<div class="p-2 w-full">
<RunChart
minTimeSet={minTs}
maxTimeSet={maxTs}
maxIsNow={maxTs == undefined}
jobs={completedJobs}
on:zoom={async (e) => {
minTs = e.detail.min.toISOString()
maxTs = e.detail.max.toISOString()
jobLoader?.loadJobs(minTs, maxTs, true)
}}
/>
<div class="relative z-10">
<div class="absolute right-0 -mt-6">
<div class="flex flex-row justify-between items-center">
<ToggleButtonGroup
bind:selected={graph}
on:selected={() => {
graphIsRunsChart = graph === 'RunChart'
}}
>
<ToggleButton value="RunChart" label="Duration" />
<ToggleButton
value="ConcurrencyChart"
label="Concurrency"
icon={warnJobLimit ? AlertTriangle : undefined}
tooltip={warnJobLimit ? warnJobLimitMsg : undefined}
/>
</ToggleButtonGroup>
</div>
</div>
</div>
{#if graph === 'RunChart'}
<RunChart
minTimeSet={minTs}
maxTimeSet={maxTs}
maxIsNow={maxTs == undefined}
jobs={completedJobs}
on:zoom={async (e) => {
minTs = e.detail.min.toISOString()
maxTs = e.detail.max.toISOString()
jobLoader?.loadJobs(minTs, maxTs, true)
}}
/>
{:else if graph === 'ConcurrencyChart'}
<ConcurrentJobsChart
minTimeSet={minTs}
maxTimeSet={maxTs}
maxIsNow={maxTs == undefined}
{concurrencyIntervals}
on:zoom={async (e) => {
minTs = e.detail.min.toISOString()
maxTs = e.detail.max.toISOString()
jobLoader?.loadJobs(minTs, maxTs, true)
}}
/>
{/if}
</div>
<div class="flex flex-col gap-1 md:flex-row w-full p-4">
<div class="flex gap-2 grow flex-row">
@@ -500,6 +599,8 @@
{#if jobs}
<RunsTable
{jobs}
{externalJobs}
showExternalJobs={!graphIsRunsChart}
activeLabel={label}
bind:selectedId
bind:selectedWorkspace
@@ -507,6 +608,7 @@
on:filterByUser={filterByUser}
on:filterByFolder={filterByFolder}
on:filterByLabel={filterByLabel}
on:filterByConcurrencyKey={filterByConcurrencyKey}
/>
{:else}
<div class="gap-1 flex flex-col">
@@ -518,7 +620,11 @@
</Pane>
<Pane size={40} minSize={15} class="border-t">
{#if selectedId}
<JobPreview id={selectedId} workspace={selectedWorkspace} />
{#if selectedId === '-'}
<div class="p-4">There is no information available for this job</div>
{:else}
<JobPreview id={selectedId} workspace={selectedWorkspace} />
{/if}
{:else}
<div class="text-xs m-4">No job selected</div>
{/if}
@@ -564,17 +670,44 @@
</div>
</div>
<div class="p-2 w-full">
<RunChart
minTimeSet={minTs}
maxTimeSet={maxTs}
maxIsNow={maxTs == undefined}
jobs={completedJobs}
on:zoom={async (e) => {
minTs = e.detail.min.toISOString()
maxTs = e.detail.max.toISOString()
jobLoader?.loadJobs(minTs, maxTs, true)
}}
/>
<div class="relative z-10">
<div class="absolute right-2">
<ToggleButtonGroup
bind:selected={graph}
on:selected={() => {
graphIsRunsChart = graph == 'RunChart'
}}
>
<ToggleButton value="RunChart" label="Duration" />
<ToggleButton value="ConcurrencyChart" label="Concurrency" />
</ToggleButtonGroup>
</div>
</div>
{#if graph === 'RunChart'}
<RunChart
minTimeSet={minTs}
maxTimeSet={maxTs}
maxIsNow={maxTs == undefined}
jobs={completedJobs}
on:zoom={async (e) => {
minTs = e.detail.min.toISOString()
maxTs = e.detail.max.toISOString()
jobLoader?.loadJobs(minTs, maxTs, true)
}}
/>
{:else if graph === 'ConcurrencyChart'}
<ConcurrentJobsChart
minTimeSet={minTs}
maxTimeSet={maxTs}
maxIsNow={maxTs == undefined}
{concurrencyIntervals}
on:zoom={async (e) => {
minTs = e.detail.min.toISOString()
maxTs = e.detail.max.toISOString()
jobLoader?.loadJobs(minTs, maxTs, true)
}}
/>
{/if}
</div>
<div class="flex flex-col gap-4 md:flex-row w-full p-4">
<div class="flex items-center flex-row gap-2 grow">
@@ -688,6 +821,7 @@
<RunsTable
activeLabel={label}
{jobs}
{externalJobs}
bind:selectedId
bind:selectedWorkspace
on:select={() => {
@@ -697,6 +831,7 @@
on:filterByUser={filterByUser}
on:filterByFolder={filterByFolder}
on:filterByLabel={filterByLabel}
on:filterByConcurrencyKey={filterByConcurrencyKey}
/>
</div>
</div>