diff --git a/backend/.sqlx/query-0a1c95c4376b944661bab13271091cf3ea0afe68fb8e08e7aea239dc735c625c.json b/backend/.sqlx/query-0a1c95c4376b944661bab13271091cf3ea0afe68fb8e08e7aea239dc735c625c.json new file mode 100644 index 0000000000..b8f224dbc2 --- /dev/null +++ b/backend/.sqlx/query-0a1c95c4376b944661bab13271091cf3ea0afe68fb8e08e7aea239dc735c625c.json @@ -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" +} diff --git a/backend/.sqlx/query-b69891c25dd029b1a54e97ace292433e1485324ff7dc802fe75d21c8c6db1d42.json b/backend/.sqlx/query-1a612eb0b64eddd2c5657ef73598c47886545796424f8612135b711e2b9ddb6c.json similarity index 82% rename from backend/.sqlx/query-b69891c25dd029b1a54e97ace292433e1485324ff7dc802fe75d21c8c6db1d42.json rename to backend/.sqlx/query-1a612eb0b64eddd2c5657ef73598c47886545796424f8612135b711e2b9ddb6c.json index aed0e14f2e..6afda5710f 100644 --- a/backend/.sqlx/query-b69891c25dd029b1a54e97ace292433e1485324ff7dc802fe75d21c8c6db1d42.json +++ b/backend/.sqlx/query-1a612eb0b64eddd2c5657ef73598c47886545796424f8612135b711e2b9ddb6c.json @@ -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" } diff --git a/backend/.sqlx/query-2352e293e172304a10ab3500b17848e8199b690b1834c382fa9d5c6ae163ec2c.json b/backend/.sqlx/query-2352e293e172304a10ab3500b17848e8199b690b1834c382fa9d5c6ae163ec2c.json new file mode 100644 index 0000000000..4285bc879a --- /dev/null +++ b/backend/.sqlx/query-2352e293e172304a10ab3500b17848e8199b690b1834c382fa9d5c6ae163ec2c.json @@ -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" +} diff --git a/backend/.sqlx/query-2946aa519633291de986978d245dc615a0ee10ce6e6a98bd4906e86225b387b3.json b/backend/.sqlx/query-2bbcc383ed79afa3392af417b0cabb32d14a0a26dc36085fb2eaa62e769aca8e.json similarity index 51% rename from backend/.sqlx/query-2946aa519633291de986978d245dc615a0ee10ce6e6a98bd4906e86225b387b3.json rename to backend/.sqlx/query-2bbcc383ed79afa3392af417b0cabb32d14a0a26dc36085fb2eaa62e769aca8e.json index 18a59b9349..f7f167db24 100644 --- a/backend/.sqlx/query-2946aa519633291de986978d245dc615a0ee10ce6e6a98bd4906e86225b387b3.json +++ b/backend/.sqlx/query-2bbcc383ed79afa3392af417b0cabb32d14a0a26dc36085fb2eaa62e769aca8e.json @@ -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" } diff --git a/backend/.sqlx/query-5eff32a37ad4e8b499a82c1874061f6456ded4aca6b33964e138b935cb016eff.json b/backend/.sqlx/query-5eff32a37ad4e8b499a82c1874061f6456ded4aca6b33964e138b935cb016eff.json new file mode 100644 index 0000000000..757a1b3bb6 --- /dev/null +++ b/backend/.sqlx/query-5eff32a37ad4e8b499a82c1874061f6456ded4aca6b33964e138b935cb016eff.json @@ -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" +} diff --git a/backend/.sqlx/query-636db0b9d5963ed540f18ab732a8b29a3308f973cc04fd10979c44ae19169abf.json b/backend/.sqlx/query-636db0b9d5963ed540f18ab732a8b29a3308f973cc04fd10979c44ae19169abf.json deleted file mode 100644 index f069e7b851..0000000000 --- a/backend/.sqlx/query-636db0b9d5963ed540f18ab732a8b29a3308f973cc04fd10979c44ae19169abf.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-6b313cc9a57ae3c943bda4a3213f7f6231a44b6ef5a52754074d136007f4f72a.json b/backend/.sqlx/query-8b10c9ade85c0307e300224e87962ceba8e7ccbe53b56c4670a1634bb9d0e89c.json similarity index 75% rename from backend/.sqlx/query-6b313cc9a57ae3c943bda4a3213f7f6231a44b6ef5a52754074d136007f4f72a.json rename to backend/.sqlx/query-8b10c9ade85c0307e300224e87962ceba8e7ccbe53b56c4670a1634bb9d0e89c.json index 2b79c631bc..c5a3c8d734 100644 --- a/backend/.sqlx/query-6b313cc9a57ae3c943bda4a3213f7f6231a44b6ef5a52754074d136007f4f72a.json +++ b/backend/.sqlx/query-8b10c9ade85c0307e300224e87962ceba8e7ccbe53b56c4670a1634bb9d0e89c.json @@ -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" } diff --git a/backend/.sqlx/query-973933b021d2167edff3a48ec4d4abc53ada670155921a4a4c2f05f229ae560a.json b/backend/.sqlx/query-973933b021d2167edff3a48ec4d4abc53ada670155921a4a4c2f05f229ae560a.json deleted file mode 100644 index 2f67c8e0ed..0000000000 --- a/backend/.sqlx/query-973933b021d2167edff3a48ec4d4abc53ada670155921a4a4c2f05f229ae560a.json +++ /dev/null @@ -1,14 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO custom_concurrency_key_ended VALUES ($1)", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar" - ] - }, - "nullable": [] - }, - "hash": "973933b021d2167edff3a48ec4d4abc53ada670155921a4a4c2f05f229ae560a" -} diff --git a/backend/.sqlx/query-9a5c7e8b60a260085b438bd300972ebf948ea26f313ee5a73b85574becdd7dc7.json b/backend/.sqlx/query-9a5c7e8b60a260085b438bd300972ebf948ea26f313ee5a73b85574becdd7dc7.json deleted file mode 100644 index ebb54c644c..0000000000 --- a/backend/.sqlx/query-9a5c7e8b60a260085b438bd300972ebf948ea26f313ee5a73b85574becdd7dc7.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-ef132ac8d79579b08d7359789b6f22991f51e1c945efc2924df6253d62b83bba.json b/backend/.sqlx/query-9d25fbd21a63e6e9ccfdbf0460c8b3ca6d7fa05600067f379659b59b6f2bf418.json similarity index 76% rename from backend/.sqlx/query-ef132ac8d79579b08d7359789b6f22991f51e1c945efc2924df6253d62b83bba.json rename to backend/.sqlx/query-9d25fbd21a63e6e9ccfdbf0460c8b3ca6d7fa05600067f379659b59b6f2bf418.json index f191057047..f9c9826dca 100644 --- a/backend/.sqlx/query-ef132ac8d79579b08d7359789b6f22991f51e1c945efc2924df6253d62b83bba.json +++ b/backend/.sqlx/query-9d25fbd21a63e6e9ccfdbf0460c8b3ca6d7fa05600067f379659b59b6f2bf418.json @@ -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" } diff --git a/backend/.sqlx/query-9da0cea2a5d0464ca78cfeccf6cedf2b1c0e6e6cb3c9183a937a68465debdb06.json b/backend/.sqlx/query-9da0cea2a5d0464ca78cfeccf6cedf2b1c0e6e6cb3c9183a937a68465debdb06.json new file mode 100644 index 0000000000..9f42f0f54d --- /dev/null +++ b/backend/.sqlx/query-9da0cea2a5d0464ca78cfeccf6cedf2b1c0e6e6cb3c9183a937a68465debdb06.json @@ -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" +} diff --git a/backend/.sqlx/query-a3ccf362b4f6df400b3c7a084795dbf541eb14c5c374656ffb96da7283a2a6f1.json b/backend/.sqlx/query-a3ccf362b4f6df400b3c7a084795dbf541eb14c5c374656ffb96da7283a2a6f1.json deleted file mode 100644 index bc5e2351f9..0000000000 --- a/backend/.sqlx/query-a3ccf362b4f6df400b3c7a084795dbf541eb14c5c374656ffb96da7283a2a6f1.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-a875cb56485b812e9d4739afd0915067f7e5abe0ca0adf264b792fccf21e005b.json b/backend/.sqlx/query-a875cb56485b812e9d4739afd0915067f7e5abe0ca0adf264b792fccf21e005b.json deleted file mode 100644 index e264db5ef2..0000000000 --- a/backend/.sqlx/query-a875cb56485b812e9d4739afd0915067f7e5abe0ca0adf264b792fccf21e005b.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76.json b/backend/.sqlx/query-b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76.json index 99269c9851..54e94cfb8f 100644 --- a/backend/.sqlx/query-b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76.json +++ b/backend/.sqlx/query-b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76.json @@ -18,8 +18,8 @@ "Left": [] }, "nullable": [ - true, - false + false, + true ] }, "hash": "b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76" diff --git a/backend/.sqlx/query-2f42460fdd8aa125c8fd46b3cd02e47f57de0f073d3ce3bc7d21a7e404a83b5c.json b/backend/.sqlx/query-e1c715020f1efb00123171edfb6173bfeaaa95d4f870240e42b76e5df304ce99.json similarity index 69% rename from backend/.sqlx/query-2f42460fdd8aa125c8fd46b3cd02e47f57de0f073d3ce3bc7d21a7e404a83b5c.json rename to backend/.sqlx/query-e1c715020f1efb00123171edfb6173bfeaaa95d4f870240e42b76e5df304ce99.json index 6996508ba6..cc9ef2a729 100644 --- a/backend/.sqlx/query-2f42460fdd8aa125c8fd46b3cd02e47f57de0f073d3ce3bc7d21a7e404a83b5c.json +++ b/backend/.sqlx/query-e1c715020f1efb00123171edfb6173bfeaaa95d4f870240e42b76e5df304ce99.json @@ -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" } diff --git a/backend/.sqlx/query-0b74ab3a237b2b7f54c05c7ea74294317fcd039870268eaa930f1ba8b8250559.json b/backend/.sqlx/query-fab5a746a179eea356f0bb9005590e3bb89c4deff181827bd2fa5630c21e42a7.json similarity index 67% rename from backend/.sqlx/query-0b74ab3a237b2b7f54c05c7ea74294317fcd039870268eaa930f1ba8b8250559.json rename to backend/.sqlx/query-fab5a746a179eea356f0bb9005590e3bb89c4deff181827bd2fa5630c21e42a7.json index f02f523b96..1f877fcddf 100644 --- a/backend/.sqlx/query-0b74ab3a237b2b7f54c05c7ea74294317fcd039870268eaa930f1ba8b8250559.json +++ b/backend/.sqlx/query-fab5a746a179eea356f0bb9005590e3bb89c4deff181827bd2fa5630c21e42a7.json @@ -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" } diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 1537671481..a3103e19fc 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -644b6f49f087790a728a7a0a82525997b1742738 \ No newline at end of file +e0b0494c26b63efa4aa3f41a9fb42c7733dcca2e diff --git a/backend/migrations/20240515093517_concurrency_key_observability.down.sql b/backend/migrations/20240515093517_concurrency_key_observability.down.sql new file mode 100644 index 0000000000..150c3795ee --- /dev/null +++ b/backend/migrations/20240515093517_concurrency_key_observability.down.sql @@ -0,0 +1,3 @@ +-- Add down migration script here +DROP TABLE concurrency_key + diff --git a/backend/migrations/20240515093517_concurrency_key_observability.up.sql b/backend/migrations/20240515093517_concurrency_key_observability.up.sql new file mode 100644 index 0000000000..c0b2253afd --- /dev/null +++ b/backend/migrations/20240515093517_concurrency_key_observability.up.sql @@ -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); diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 98976153a8..b907b49e6a 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -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) diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index 84bbf97c6f..d6bc95a239 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -1067,6 +1067,7 @@ async fn test_deno_flow(db: Pool) { 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) { 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) { 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) { 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) { 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) { 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, 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, 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; -} \ No newline at end of file +} diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index c59e16fa3c..0fff6cc9b2 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -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 diff --git a/backend/windmill-api/src/concurrency_groups.rs b/backend/windmill-api/src/concurrency_groups.rs index e17547b861..b8f7fd6b42 100644 --- a/backend/windmill-api/src/concurrency_groups.rs +++ b/backend/windmill-api/src/concurrency_groups.rs @@ -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, + concurrency_key: String, + total_running: i64, } -#[cfg(feature = "enterprise")] async fn list_concurrency_groups( authed: ApiAuthed, Extension(db): Extension, ) -> JsonResult> { - 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 = vec![]; - for (concurrency_id, job_uuids_json) in concurrency_groups_raw { - let job_uuids_map = serde_json::from_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, - Path(concurrency_id): Path, + Path(concurrency_key): Path, ) -> 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, + running_jobs: Vec, + completed_jobs: Vec, +} + +#[derive(Serialize)] +struct CompletedJobDuration { + job_id: Option, + concurrency_key: Option, + started_at: chrono::DateTime, + ended_at: chrono::DateTime, +} + +impl<'r> FromRow<'r, PgRow> for CompletedJobDuration { + fn from_row(row: &'r PgRow) -> Result { + let duration_ms: i64 = row.try_get("duration_ms")?; + let started_at = row.try_get("started_at")?; + let ended_at: chrono::DateTime = + 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, + #[sqlx(rename = "key")] + concurrency_key: Option, + started_at: Option>, +} + +#[derive(Deserialize)] +struct ConcurrentIntervalsParams { + concurrency_key: Option, + row_limit: Option, +} + +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, + Extension(user_db): Extension, + Path(w_id): Path, + Query(iq): Query, + Query(lq): Query, +) -> JsonResult { + 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 = if lq.success.is_none() { + sqlx::query_as(&sql_q).fetch_all(&mut *tx).await? + } else { + vec![] + }; + let completed_jobs_user: Vec = 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 = if lq.success.is_none() { + sqlx::query_as(&sql_q).fetch_all(&db).await? + } else { + vec![] + }; + let completed_jobs_db: Vec = 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, + Path(job_id): Path, +) -> JsonResult> { + let key = sqlx::query_scalar!("SELECT key FROM concurrency_key WHERE job_id = $1", job_id) + .fetch_optional(&db) + .await?; + Ok(Json(key)) +} diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index 7da1c6582a..e415003e05 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -912,6 +912,7 @@ mod tests { path: None, lock: None, tag: None, + custom_concurrency_key: None, concurrent_limit: None, concurrency_time_window_s: None, }, diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 3bcaca56bf..4ca6217849 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -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, + Option, Option, Option, Option, @@ -476,7 +478,7 @@ pub async fn get_path_tag_limits_cache_for_hash( Option, )> { 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, pub script_path_exact: Option, @@ -900,13 +903,7 @@ pub struct ListQueueQuery { pub is_not_schedule: Option, } -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>>> { + 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) -> Result, 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, diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index 9015740904..e5ef447ad5 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -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()) diff --git a/backend/windmill-common/src/flows.rs b/backend/windmill-common/src/flows.rs index 852ea7e442..05eca9d83b 100644 --- a/backend/windmill-common/src/flows.rs +++ b/backend/windmill-common/src/flows.rs @@ -345,6 +345,8 @@ pub enum FlowModuleValue { tag: Option, language: ScriptLang, #[serde(skip_serializing_if = "Option::is_none")] + custom_concurrency_key: Option, + #[serde(skip_serializing_if = "Option::is_none")] concurrent_limit: Option, #[serde(skip_serializing_if = "Option::is_none")] concurrency_time_window_s: Option, diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs index f90ea4e3c7..a059b00377 100644 --- a/backend/windmill-common/src/jobs.rs +++ b/backend/windmill-common/src/jobs.rs @@ -282,6 +282,7 @@ pub enum JobPayload { ScriptHash { hash: ScriptHash, path: String, + custom_concurrency_key: Option, concurrent_limit: Option, concurrency_time_window_s: Option, cache_ttl: Option, @@ -328,6 +329,7 @@ pub enum JobPayload { hash: ScriptHash, args: HashMap, retry: Retry, // for now only used to retry the script, so retry is necessarily present + custom_concurrency_key: Option, concurrent_limit: Option, concurrency_time_window_s: Option, cache_ttl: Option, @@ -348,6 +350,7 @@ pub struct RawCode { pub hash: Option, pub language: ScriptLang, pub lock: Option, + pub custom_concurrency_key: Option, pub concurrent_limit: Option, pub concurrency_time_window_s: Option, pub cache_ttl: Option, @@ -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, + Option, Option, Option, Option, @@ -418,7 +424,7 @@ pub async fn script_hash_to_tag_and_limits<'c>( Option, )> { 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, diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index f5bb5186ee..bd25dcd608 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -249,6 +249,7 @@ pub async fn get_latest_deployed_hash_for_path( ) -> error::Result<( scripts::ScriptHash, Option, + Option, Option, Option, Option, @@ -259,7 +260,7 @@ pub async fn get_latest_deployed_hash_for_path( Option, )> { 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, + Option, Option, Option, Option, @@ -300,7 +303,7 @@ pub async fn get_latest_hash_for_path<'c>( Option, )> { 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, diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index b81febd072..82e93be69d 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -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( // 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( 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, 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, + job_id: Uuid, +) -> Result, 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, queued_job: &QueuedJob, - concurrency_key: Result, sqlx::Error>, +) -> windmill_common::error::Result { + not_found_if_none(custom_concurrency_key(db, queued_job.id).await?, "ConcurrencyKey", queued_job.id.to_string()) +} + +fn interpolate_args( + x: String, + args: &T, + workspace_id: &str, + parsed_args: &mut Option, ) -> 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 = 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, diff --git a/backend/windmill-queue/src/schedule.rs b/backend/windmill-queue/src/schedule.rs index b582b0f7fa..cf07df506f 100644 --- a/backend/windmill-queue/src/schedule.rs +++ b/backend/windmill-queue/src/schedule.rs @@ -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, diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 00813a13b0..cb95015468 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -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, }; diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 92cb019089..f1ddf2807f 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -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, + custom_concurrency_key: &Option, concurrent_limit: &Option, concurrency_time_window_s: &Option, 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(), diff --git a/frontend/src/lib/components/ConcurrentJobsChart.svelte b/frontend/src/lib/components/ConcurrentJobsChart.svelte new file mode 100644 index 0000000000..464de134bd --- /dev/null +++ b/frontend/src/lib/components/ConcurrentJobsChart.svelte @@ -0,0 +1,248 @@ + + +
+ +
diff --git a/frontend/src/lib/components/FlowMetadata.svelte b/frontend/src/lib/components/FlowMetadata.svelte index 1587acbc37..0896deeb65 100644 --- a/frontend/src/lib/components/FlowMetadata.svelte +++ b/frontend/src/lib/components/FlowMetadata.svelte @@ -1,5 +1,5 @@ + + +
+ + + +
+ diff --git a/frontend/src/lib/components/runs/JobLoader.svelte b/frontend/src/lib/components/runs/JobLoader.svelte index ef338055fb..9e70eeef9d 100644 --- a/frontend/src/lib/components/runs/JobLoader.svelte +++ b/frontend/src/lib/components/runs/JobLoader.svelte @@ -1,6 +1,12 @@ @@ -42,7 +45,7 @@
- {#if 'success' in job && job.success} + {#if isExternal} + + + + {:else if 'success' in job && job.success} {#if job.is_skipped} @@ -108,21 +115,25 @@
{#if job.script_path}
- - {job.script_path} - - + {#if isExternal} + - + {:else} + + {job.script_path} + + + {/if} {#if job.script_path?.startsWith('f/')}
- + {#if !isExternal} + + {/if}
{/if}
diff --git a/frontend/src/lib/components/runs/RunsFilter.svelte b/frontend/src/lib/components/runs/RunsFilter.svelte index 79779de78b..846dd44f02 100644 --- a/frontend/src/lib/components/runs/RunsFilter.svelte +++ b/frontend/src/lib/components/runs/RunsFilter.svelte @@ -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
@@ -86,6 +93,7 @@ user = null folder = null label = null + concurrencyKey = null } else { manuallySet = false } @@ -94,7 +102,12 @@ - +
@@ -241,6 +254,47 @@ />
{/key} + {:else if filterBy === 'concurrencyKey'} + {#key concurrencyKey} +
+ {#if concurrencyKey} + + {/if} + Concurrency Key + 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} + + + { + if (concurrencyKeyTimeout) { + clearTimeout(concurrencyKeyTimeout) + } + + concurrencyKeyTimeout = setTimeout(() => { + concurrencyKey = displayedConcurrencyKey + }, 1000) + }} + /> +
+ {/key} {/if}
diff --git a/frontend/src/lib/components/runs/RunsTable.svelte b/frontend/src/lib/components/runs/RunsTable.svelte index 85141402bc..c5e9a39695 100644 --- a/frontend/src/lib/components/runs/RunsTable.svelte +++ b/frontend/src/lib/components/runs/RunsTable.svelte @@ -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} > -
{jobs?.length == 1000 ? '1000+' : jobs ? jobs.length.toString() : '...'} jobs
+ {#if showExternalJobs && externalJobs} +
+ {jobs && jobCountString(jobs.length + externalJobs.length)}{externalJobs.length} jobs obscured +
+ {:else} +
{jobs && jobCountString(jobs.length)}
+ {/if}
Timestamp
Path
{#if containsLabel} @@ -188,6 +209,7 @@ on:filterByPath on:filterByUser on:filterByFolder + on:filterByConcurrencyKey {containerWidth} />
@@ -206,7 +228,7 @@ -{#if jobs?.length == 0} +{#if jobs?.length == 0 && (!showExternalJobs || externalJobs?.length == 0)}
No jobs found for the selected filters.
diff --git a/frontend/src/routes/(root)/(logged)/concurrency_groups/+page.svelte b/frontend/src/routes/(root)/(logged)/concurrency_groups/+page.svelte index 9a2377564b..9d7410e364 100644 --- a/frontend/src/routes/(root)/(logged)/concurrency_groups/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/concurrency_groups/+page.svelte @@ -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 @@ } - - - {#if selectedGroup?.job_uuids && selectedGroup?.job_uuids.length > 0} -

Jobs running for this group

- -
    - {#each selectedGroup?.job_uuids as jobUuid} -
  • - {jobUuid} -
    • - {/each} -
- {:else} -

No job running for this group

- {/if} - - -