From c80f155602eca972842be7bd560395a06e4e0ae6 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Wed, 6 Sep 2023 10:32:53 +0200 Subject: [PATCH] feat: dedicated workers for native-throughput performance (EE only) (#2239) * dedicated worker * dedicated worker * merge * progress * progress * progress * merge * update * sqlx * fix imports --- ...5753b86a4ad239708851c998ff5620ebca5b8.json | 14 - ...094088db8ce98aefdbe96144c42e38e5bd813.json | 41 --- ...0aff06ea62f94ff357c446a8e2311e083efdf.json | 14 - ...55ddd99c4ad1f46b876dd86e372b84d806ecd.json | 74 ++++ ...54d01fe0a2c7e9692ac4ce29a86eb509e1a1d.json | 17 + ...4e797527841cd97dba80c271cbefafae65cc.json} | 7 +- ...30f5ca4b3b0b0afbd5d9721c8b5d692af601b.json | 22 -- ...ee85b191263989b0c78b2bfce77e796e96825.json | 80 ++++ ...0273b942202cdf9017ce6b5ba612e66ce3875.json | 44 --- ...f6f580f15bca96d9746c9359e98ca793f8f1f.json | 80 ++++ ...c95765356fe7744b817ad0c320a54a33b24f3.json | 47 --- ...5af33f84b22c24c92fcb870f37f0501f8ea9a.json | 22 -- ...5eedc09bceb82bca349f3e31c8513ebbf0192.json | 80 ++++ ...93b941ef75880eeeab0756d40f68f87760d5b.json | 23 -- ...23be681a0e2cb07a457ea9aede2daeff551cc.json | 22 -- ...696b824685978760b308d8b578df95cf8db45.json | 14 - ...d9b90d7654a4b5d0ec32085b178399f701bfc.json | 47 --- ...50aa180b29bbe931d948f6e61976f71b7cdb9.json | 22 -- ...c1cb55aadaf1e6609a1aad09b0515c93a667a.json | 22 -- ...63e11422eed9a60e0472c68971f4e8f8735d9.json | 16 - ...034210afce7902a69f7c42bb68160e0464d71.json | 47 --- ...3063e168f10a53d56f222c7917aeba020fff3.json | 14 - ...b1ebad6b2b0a836334a5e20f151711fc11370.json | 15 - .../20230905170218_dedicated_worker.down.sql | 1 + .../20230905170218_dedicated_worker.up.sql | 2 + .../tests/fixtures/base.sql | 1 + backend/windmill-api/openapi.yaml | 4 + backend/windmill-api/src/jobs.rs | 32 +- backend/windmill-api/src/oauth2.rs | 23 +- backend/windmill-api/src/scripts.rs | 9 +- backend/windmill-common/src/flows.rs | 11 + backend/windmill-common/src/jobs.rs | 28 +- backend/windmill-common/src/lib.rs | 13 +- backend/windmill-common/src/scripts.rs | 1 + backend/windmill-queue/src/jobs.rs | 81 ++-- backend/windmill-queue/src/schedule.rs | 23 +- backend/windmill-worker/src/bun_executor.rs | 331 ++++++++++++++++- backend/windmill-worker/src/common.rs | 23 +- .../windmill-worker/src/dedicated_worker.rs | 5 + backend/windmill-worker/src/lib.rs | 1 + backend/windmill-worker/src/worker.rs | 178 ++++++--- backend/windmill-worker/src/worker_flow.rs | 347 ++++++++++++------ benchmarks/worker.ts | 25 +- .../lib/components/FlowStatusViewer.svelte | 29 +- .../src/lib/components/ScriptBuilder.svelte | 26 ++ 45 files changed, 1295 insertions(+), 683 deletions(-) delete mode 100644 backend/.sqlx/query-062859f1d0e5cfba3115f4241115753b86a4ad239708851c998ff5620ebca5b8.json delete mode 100644 backend/.sqlx/query-0694f4be1a7966126f20afbd9ee094088db8ce98aefdbe96144c42e38e5bd813.json delete mode 100644 backend/.sqlx/query-120fa12a4c2fdecf1da58ffba0c0aff06ea62f94ff357c446a8e2311e083efdf.json create mode 100644 backend/.sqlx/query-123c0608e229c29187009b7961355ddd99c4ad1f46b876dd86e372b84d806ecd.json create mode 100644 backend/.sqlx/query-15ef5759a2ccd7b7f9fd3f2ce0d54d01fe0a2c7e9692ac4ce29a86eb509e1a1d.json rename backend/.sqlx/{query-52dbe77fad5729e9e121609eeb6a3573e6d0e1d1c16e29b15c7ad810536e249c.json => query-438b5b5d29b05846c2e074cad2404e797527841cd97dba80c271cbefafae65cc.json} (82%) delete mode 100644 backend/.sqlx/query-5ba4b87528ad49f17d72b53c3db30f5ca4b3b0b0afbd5d9721c8b5d692af601b.json create mode 100644 backend/.sqlx/query-5cd89ab614d3cac80fb81627267ee85b191263989b0c78b2bfce77e796e96825.json delete mode 100644 backend/.sqlx/query-637ead7701775609c772c9985450273b942202cdf9017ce6b5ba612e66ce3875.json create mode 100644 backend/.sqlx/query-65835f2e5ad38f7cc6b147dadfef6f580f15bca96d9746c9359e98ca793f8f1f.json delete mode 100644 backend/.sqlx/query-8062bb1ac475c4a2a016d9dfa63c95765356fe7744b817ad0c320a54a33b24f3.json delete mode 100644 backend/.sqlx/query-b053117536c067095e2fb2864ce5af33f84b22c24c92fcb870f37f0501f8ea9a.json create mode 100644 backend/.sqlx/query-b224cdd1221fc9e7227ef8e8c025eedc09bceb82bca349f3e31c8513ebbf0192.json delete mode 100644 backend/.sqlx/query-bd086c56ad3a58163619122641b93b941ef75880eeeab0756d40f68f87760d5b.json delete mode 100644 backend/.sqlx/query-c07c9276945663d062cf0ff5b3323be681a0e2cb07a457ea9aede2daeff551cc.json delete mode 100644 backend/.sqlx/query-cac375edf290d68d487de4273c3696b824685978760b308d8b578df95cf8db45.json delete mode 100644 backend/.sqlx/query-cf21b61aa5b5e47744235c860e2d9b90d7654a4b5d0ec32085b178399f701bfc.json delete mode 100644 backend/.sqlx/query-d768bbc46f8a9c4289b918c88ca50aa180b29bbe931d948f6e61976f71b7cdb9.json delete mode 100644 backend/.sqlx/query-d918e1d3a1ccc36ca8b4b25aeeec1cb55aadaf1e6609a1aad09b0515c93a667a.json delete mode 100644 backend/.sqlx/query-e7418515b88d14d1fd79078342963e11422eed9a60e0472c68971f4e8f8735d9.json delete mode 100644 backend/.sqlx/query-ebac391c83dae67831502fbe3f1034210afce7902a69f7c42bb68160e0464d71.json delete mode 100644 backend/.sqlx/query-f1e65e52585b27bfb39817e82613063e168f10a53d56f222c7917aeba020fff3.json delete mode 100644 backend/.sqlx/query-ff89e9f0941507d5cbfbb27e7c3b1ebad6b2b0a836334a5e20f151711fc11370.json create mode 100644 backend/migrations/20230905170218_dedicated_worker.down.sql create mode 100644 backend/migrations/20230905170218_dedicated_worker.up.sql create mode 100644 backend/windmill-worker/src/dedicated_worker.rs diff --git a/backend/.sqlx/query-062859f1d0e5cfba3115f4241115753b86a4ad239708851c998ff5620ebca5b8.json b/backend/.sqlx/query-062859f1d0e5cfba3115f4241115753b86a4ad239708851c998ff5620ebca5b8.json deleted file mode 100644 index 9803b76cea..0000000000 --- a/backend/.sqlx/query-062859f1d0e5cfba3115f4241115753b86a4ad239708851c998ff5620ebca5b8.json +++ /dev/null @@ -1,14 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE queue SET last_ping = now() WHERE id = $1", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Uuid" - ] - }, - "nullable": [] - }, - "hash": "062859f1d0e5cfba3115f4241115753b86a4ad239708851c998ff5620ebca5b8" -} diff --git a/backend/.sqlx/query-0694f4be1a7966126f20afbd9ee094088db8ce98aefdbe96144c42e38e5bd813.json b/backend/.sqlx/query-0694f4be1a7966126f20afbd9ee094088db8ce98aefdbe96144c42e38e5bd813.json deleted file mode 100644 index e4cfd00b92..0000000000 --- a/backend/.sqlx/query-0694f4be1a7966126f20afbd9ee094088db8ce98aefdbe96144c42e38e5bd813.json +++ /dev/null @@ -1,41 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "select tag, concurrent_limit, concurrency_time_window_s, cache_ttl from script where hash = $1 AND workspace_id = $2", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "tag", - "type_info": "Varchar" - }, - { - "ordinal": 1, - "name": "concurrent_limit", - "type_info": "Int4" - }, - { - "ordinal": 2, - "name": "concurrency_time_window_s", - "type_info": "Int4" - }, - { - "ordinal": 3, - "name": "cache_ttl", - "type_info": "Int4" - } - ], - "parameters": { - "Left": [ - "Int8", - "Text" - ] - }, - "nullable": [ - true, - true, - true, - true - ] - }, - "hash": "0694f4be1a7966126f20afbd9ee094088db8ce98aefdbe96144c42e38e5bd813" -} diff --git a/backend/.sqlx/query-120fa12a4c2fdecf1da58ffba0c0aff06ea62f94ff357c446a8e2311e083efdf.json b/backend/.sqlx/query-120fa12a4c2fdecf1da58ffba0c0aff06ea62f94ff357c446a8e2311e083efdf.json deleted file mode 100644 index 5a1cf243eb..0000000000 --- a/backend/.sqlx/query-120fa12a4c2fdecf1da58ffba0c0aff06ea62f94ff357c446a8e2311e083efdf.json +++ /dev/null @@ -1,14 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO password (email, login_type, verified) VALUES ($1, 'saml', true) ON CONFLICT DO NOTHING", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar" - ] - }, - "nullable": [] - }, - "hash": "120fa12a4c2fdecf1da58ffba0c0aff06ea62f94ff357c446a8e2311e083efdf" -} diff --git a/backend/.sqlx/query-123c0608e229c29187009b7961355ddd99c4ad1f46b876dd86e372b84d806ecd.json b/backend/.sqlx/query-123c0608e229c29187009b7961355ddd99c4ad1f46b876dd86e372b84d806ecd.json new file mode 100644 index 0000000000..1b8084742c --- /dev/null +++ b/backend/.sqlx/query-123c0608e229c29187009b7961355ddd99c4ad1f46b876dd86e372b84d806ecd.json @@ -0,0 +1,74 @@ +{ + "db_name": "PostgreSQL", + "query": "select tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker from script where hash = $1 AND workspace_id = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "tag", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "concurrent_limit", + "type_info": "Int4" + }, + { + "ordinal": 2, + "name": "concurrency_time_window_s", + "type_info": "Int4" + }, + { + "ordinal": 3, + "name": "cache_ttl", + "type_info": "Int4" + }, + { + "ordinal": 4, + "name": "language: ScriptLang", + "type_info": { + "Custom": { + "name": "script_lang", + "kind": { + "Enum": [ + "python3", + "deno", + "go", + "bash", + "postgresql", + "nativets", + "Nativets", + "bun", + "mysql", + "bigquery", + "snowflake", + "graphql", + "powershell" + ] + } + } + } + }, + { + "ordinal": 5, + "name": "dedicated_worker", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Int8", + "Text" + ] + }, + "nullable": [ + true, + true, + true, + true, + false, + true + ] + }, + "hash": "123c0608e229c29187009b7961355ddd99c4ad1f46b876dd86e372b84d806ecd" +} diff --git a/backend/.sqlx/query-15ef5759a2ccd7b7f9fd3f2ce0d54d01fe0a2c7e9692ac4ce29a86eb509e1a1d.json b/backend/.sqlx/query-15ef5759a2ccd7b7f9fd3f2ce0d54d01fe0a2c7e9692ac4ce29a86eb509e1a1d.json new file mode 100644 index 0000000000..5552400f28 --- /dev/null +++ b/backend/.sqlx/query-15ef5759a2ccd7b7f9fd3f2ce0d54d01fe0a2c7e9692ac4ce29a86eb509e1a1d.json @@ -0,0 +1,17 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO token\n (token, label, super_admin, email)\n VALUES ($1, $2, $3, $4)", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Varchar", + "Bool", + "Varchar" + ] + }, + "nullable": [] + }, + "hash": "15ef5759a2ccd7b7f9fd3f2ce0d54d01fe0a2c7e9692ac4ce29a86eb509e1a1d" +} diff --git a/backend/.sqlx/query-52dbe77fad5729e9e121609eeb6a3573e6d0e1d1c16e29b15c7ad810536e249c.json b/backend/.sqlx/query-438b5b5d29b05846c2e074cad2404e797527841cd97dba80c271cbefafae65cc.json similarity index 82% rename from backend/.sqlx/query-52dbe77fad5729e9e121609eeb6a3573e6d0e1d1c16e29b15c7ad810536e249c.json rename to backend/.sqlx/query-438b5b5d29b05846c2e074cad2404e797527841cd97dba80c271cbefafae65cc.json index ac9f9288ea..a5dee163e5 100644 --- a/backend/.sqlx/query-52dbe77fad5729e9e121609eeb6a3573e6d0e1d1c16e29b15c7ad810536e249c.json +++ b/backend/.sqlx/query-438b5b5d29b05846c2e074cad2404e797527841cd97dba80c271cbefafae65cc.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20)", + "query": "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, dedicated_worker) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21)", "describe": { "columns": [], "parameters": { @@ -58,10 +58,11 @@ "VarcharArray", "Int4", "Int4", - "Int4" + "Int4", + "Bool" ] }, "nullable": [] }, - "hash": "52dbe77fad5729e9e121609eeb6a3573e6d0e1d1c16e29b15c7ad810536e249c" + "hash": "438b5b5d29b05846c2e074cad2404e797527841cd97dba80c271cbefafae65cc" } diff --git a/backend/.sqlx/query-5ba4b87528ad49f17d72b53c3db30f5ca4b3b0b0afbd5d9721c8b5d692af601b.json b/backend/.sqlx/query-5ba4b87528ad49f17d72b53c3db30f5ca4b3b0b0afbd5d9721c8b5d692af601b.json deleted file mode 100644 index f8b5b54892..0000000000 --- a/backend/.sqlx/query-5ba4b87528ad49f17d72b53c3db30f5ca4b3b0b0afbd5d9721c8b5d692af601b.json +++ /dev/null @@ -1,22 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT COUNT(id) FROM queue WHERE email = $1", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "count", - "type_info": "Int8" - } - ], - "parameters": { - "Left": [ - "Text" - ] - }, - "nullable": [ - null - ] - }, - "hash": "5ba4b87528ad49f17d72b53c3db30f5ca4b3b0b0afbd5d9721c8b5d692af601b" -} diff --git a/backend/.sqlx/query-5cd89ab614d3cac80fb81627267ee85b191263989b0c78b2bfce77e796e96825.json b/backend/.sqlx/query-5cd89ab614d3cac80fb81627267ee85b191263989b0c78b2bfce77e796e96825.json new file mode 100644 index 0000000000..1517c8d1d4 --- /dev/null +++ b/backend/.sqlx/query-5cd89ab614d3cac80fb81627267ee85b191263989b0c78b2bfce77e796e96825.json @@ -0,0 +1,80 @@ +{ + "db_name": "PostgreSQL", + "query": "select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker from script where path = $1 AND workspace_id = $2 AND\n created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND\n deleted = false AND archived = false)", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "hash", + "type_info": "Int8" + }, + { + "ordinal": 1, + "name": "tag", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "concurrent_limit", + "type_info": "Int4" + }, + { + "ordinal": 3, + "name": "concurrency_time_window_s", + "type_info": "Int4" + }, + { + "ordinal": 4, + "name": "cache_ttl", + "type_info": "Int4" + }, + { + "ordinal": 5, + "name": "language: ScriptLang", + "type_info": { + "Custom": { + "name": "script_lang", + "kind": { + "Enum": [ + "python3", + "deno", + "go", + "bash", + "postgresql", + "nativets", + "Nativets", + "bun", + "mysql", + "bigquery", + "snowflake", + "graphql", + "powershell" + ] + } + } + } + }, + { + "ordinal": 6, + "name": "dedicated_worker", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + false, + true, + true, + true, + true, + false, + true + ] + }, + "hash": "5cd89ab614d3cac80fb81627267ee85b191263989b0c78b2bfce77e796e96825" +} diff --git a/backend/.sqlx/query-637ead7701775609c772c9985450273b942202cdf9017ce6b5ba612e66ce3875.json b/backend/.sqlx/query-637ead7701775609c772c9985450273b942202cdf9017ce6b5ba612e66ce3875.json deleted file mode 100644 index 9820f0e0f4..0000000000 --- a/backend/.sqlx/query-637ead7701775609c772c9985450273b942202cdf9017ce6b5ba612e66ce3875.json +++ /dev/null @@ -1,44 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT language as \"language: ScriptLang\" FROM script WHERE hash = $1 AND workspace_id = $2", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "language: ScriptLang", - "type_info": { - "Custom": { - "name": "script_lang", - "kind": { - "Enum": [ - "python3", - "deno", - "go", - "bash", - "postgresql", - "nativets", - "Nativets", - "bun", - "mysql", - "bigquery", - "snowflake", - "graphql", - "powershell" - ] - } - } - } - } - ], - "parameters": { - "Left": [ - "Int8", - "Text" - ] - }, - "nullable": [ - false - ] - }, - "hash": "637ead7701775609c772c9985450273b942202cdf9017ce6b5ba612e66ce3875" -} diff --git a/backend/.sqlx/query-65835f2e5ad38f7cc6b147dadfef6f580f15bca96d9746c9359e98ca793f8f1f.json b/backend/.sqlx/query-65835f2e5ad38f7cc6b147dadfef6f580f15bca96d9746c9359e98ca793f8f1f.json new file mode 100644 index 0000000000..bfe7c41f64 --- /dev/null +++ b/backend/.sqlx/query-65835f2e5ad38f7cc6b147dadfef6f580f15bca96d9746c9359e98ca793f8f1f.json @@ -0,0 +1,80 @@ +{ + "db_name": "PostgreSQL", + "query": "select path, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker from script where hash = $1 AND workspace_id = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "path", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "tag", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "concurrent_limit", + "type_info": "Int4" + }, + { + "ordinal": 3, + "name": "concurrency_time_window_s", + "type_info": "Int4" + }, + { + "ordinal": 4, + "name": "cache_ttl", + "type_info": "Int4" + }, + { + "ordinal": 5, + "name": "language: ScriptLang", + "type_info": { + "Custom": { + "name": "script_lang", + "kind": { + "Enum": [ + "python3", + "deno", + "go", + "bash", + "postgresql", + "nativets", + "Nativets", + "bun", + "mysql", + "bigquery", + "snowflake", + "graphql", + "powershell" + ] + } + } + } + }, + { + "ordinal": 6, + "name": "dedicated_worker", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Int8", + "Text" + ] + }, + "nullable": [ + false, + true, + true, + true, + true, + false, + true + ] + }, + "hash": "65835f2e5ad38f7cc6b147dadfef6f580f15bca96d9746c9359e98ca793f8f1f" +} diff --git a/backend/.sqlx/query-8062bb1ac475c4a2a016d9dfa63c95765356fe7744b817ad0c320a54a33b24f3.json b/backend/.sqlx/query-8062bb1ac475c4a2a016d9dfa63c95765356fe7744b817ad0c320a54a33b24f3.json deleted file mode 100644 index 99cfea3318..0000000000 --- a/backend/.sqlx/query-8062bb1ac475c4a2a016d9dfa63c95765356fe7744b817ad0c320a54a33b24f3.json +++ /dev/null @@ -1,47 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl 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": [ - { - "ordinal": 0, - "name": "hash", - "type_info": "Int8" - }, - { - "ordinal": 1, - "name": "tag", - "type_info": "Varchar" - }, - { - "ordinal": 2, - "name": "concurrent_limit", - "type_info": "Int4" - }, - { - "ordinal": 3, - "name": "concurrency_time_window_s", - "type_info": "Int4" - }, - { - "ordinal": 4, - "name": "cache_ttl", - "type_info": "Int4" - } - ], - "parameters": { - "Left": [ - "Text", - "Text" - ] - }, - "nullable": [ - false, - true, - true, - true, - true - ] - }, - "hash": "8062bb1ac475c4a2a016d9dfa63c95765356fe7744b817ad0c320a54a33b24f3" -} diff --git a/backend/.sqlx/query-b053117536c067095e2fb2864ce5af33f84b22c24c92fcb870f37f0501f8ea9a.json b/backend/.sqlx/query-b053117536c067095e2fb2864ce5af33f84b22c24c92fcb870f37f0501f8ea9a.json deleted file mode 100644 index 316f6d7eda..0000000000 --- a/backend/.sqlx/query-b053117536c067095e2fb2864ce5af33f84b22c24c92fcb870f37f0501f8ea9a.json +++ /dev/null @@ -1,22 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT COUNT(id) FROM queue WHERE running = true AND email = $1", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "count", - "type_info": "Int8" - } - ], - "parameters": { - "Left": [ - "Text" - ] - }, - "nullable": [ - null - ] - }, - "hash": "b053117536c067095e2fb2864ce5af33f84b22c24c92fcb870f37f0501f8ea9a" -} diff --git a/backend/.sqlx/query-b224cdd1221fc9e7227ef8e8c025eedc09bceb82bca349f3e31c8513ebbf0192.json b/backend/.sqlx/query-b224cdd1221fc9e7227ef8e8c025eedc09bceb82bca349f3e31c8513ebbf0192.json new file mode 100644 index 0000000000..38f81da395 --- /dev/null +++ b/backend/.sqlx/query-b224cdd1221fc9e7227ef8e8c025eedc09bceb82bca349f3e31c8513ebbf0192.json @@ -0,0 +1,80 @@ +{ + "db_name": "PostgreSQL", + "query": "select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker from script where path = $1 AND workspace_id = $2 AND\n created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND\n deleted = false AND lock IS not NULL AND lock_error_logs IS NULL)", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "hash", + "type_info": "Int8" + }, + { + "ordinal": 1, + "name": "tag", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "concurrent_limit", + "type_info": "Int4" + }, + { + "ordinal": 3, + "name": "concurrency_time_window_s", + "type_info": "Int4" + }, + { + "ordinal": 4, + "name": "cache_ttl", + "type_info": "Int4" + }, + { + "ordinal": 5, + "name": "language: ScriptLang", + "type_info": { + "Custom": { + "name": "script_lang", + "kind": { + "Enum": [ + "python3", + "deno", + "go", + "bash", + "postgresql", + "nativets", + "Nativets", + "bun", + "mysql", + "bigquery", + "snowflake", + "graphql", + "powershell" + ] + } + } + } + }, + { + "ordinal": 6, + "name": "dedicated_worker", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + false, + true, + true, + true, + true, + false, + true + ] + }, + "hash": "b224cdd1221fc9e7227ef8e8c025eedc09bceb82bca349f3e31c8513ebbf0192" +} diff --git a/backend/.sqlx/query-bd086c56ad3a58163619122641b93b941ef75880eeeab0756d40f68f87760d5b.json b/backend/.sqlx/query-bd086c56ad3a58163619122641b93b941ef75880eeeab0756d40f68f87760d5b.json deleted file mode 100644 index 9214257461..0000000000 --- a/backend/.sqlx/query-bd086c56ad3a58163619122641b93b941ef75880eeeab0756d40f68f87760d5b.json +++ /dev/null @@ -1,23 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO usage (id, is_workspace, month_, usage) \n VALUES ($1, $2, EXTRACT(YEAR FROM current_date) * 12 + EXTRACT(MONTH FROM current_date), 0) \n ON CONFLICT (id, is_workspace, month_) DO UPDATE SET usage = usage.usage + 1 \n RETURNING usage.usage", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "usage", - "type_info": "Int4" - } - ], - "parameters": { - "Left": [ - "Varchar", - "Bool" - ] - }, - "nullable": [ - false - ] - }, - "hash": "bd086c56ad3a58163619122641b93b941ef75880eeeab0756d40f68f87760d5b" -} diff --git a/backend/.sqlx/query-c07c9276945663d062cf0ff5b3323be681a0e2cb07a457ea9aede2daeff551cc.json b/backend/.sqlx/query-c07c9276945663d062cf0ff5b3323be681a0e2cb07a457ea9aede2daeff551cc.json deleted file mode 100644 index dc7f552834..0000000000 --- a/backend/.sqlx/query-c07c9276945663d062cf0ff5b3323be681a0e2cb07a457ea9aede2daeff551cc.json +++ /dev/null @@ -1,22 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT customer_id FROM workspace_settings WHERE workspace_id = $1", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "customer_id", - "type_info": "Varchar" - } - ], - "parameters": { - "Left": [ - "Text" - ] - }, - "nullable": [ - true - ] - }, - "hash": "c07c9276945663d062cf0ff5b3323be681a0e2cb07a457ea9aede2daeff551cc" -} diff --git a/backend/.sqlx/query-cac375edf290d68d487de4273c3696b824685978760b308d8b578df95cf8db45.json b/backend/.sqlx/query-cac375edf290d68d487de4273c3696b824685978760b308d8b578df95cf8db45.json deleted file mode 100644 index bd3c911d70..0000000000 --- a/backend/.sqlx/query-cac375edf290d68d487de4273c3696b824685978760b308d8b578df95cf8db45.json +++ /dev/null @@ -1,14 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO instance_group (name) VALUES ($1) ON CONFLICT DO NOTHING", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar" - ] - }, - "nullable": [] - }, - "hash": "cac375edf290d68d487de4273c3696b824685978760b308d8b578df95cf8db45" -} diff --git a/backend/.sqlx/query-cf21b61aa5b5e47744235c860e2d9b90d7654a4b5d0ec32085b178399f701bfc.json b/backend/.sqlx/query-cf21b61aa5b5e47744235c860e2d9b90d7654a4b5d0ec32085b178399f701bfc.json deleted file mode 100644 index 58c600f2ca..0000000000 --- a/backend/.sqlx/query-cf21b61aa5b5e47744235c860e2d9b90d7654a4b5d0ec32085b178399f701bfc.json +++ /dev/null @@ -1,47 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl 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": [ - { - "ordinal": 0, - "name": "hash", - "type_info": "Int8" - }, - { - "ordinal": 1, - "name": "tag", - "type_info": "Varchar" - }, - { - "ordinal": 2, - "name": "concurrent_limit", - "type_info": "Int4" - }, - { - "ordinal": 3, - "name": "concurrency_time_window_s", - "type_info": "Int4" - }, - { - "ordinal": 4, - "name": "cache_ttl", - "type_info": "Int4" - } - ], - "parameters": { - "Left": [ - "Text", - "Text" - ] - }, - "nullable": [ - false, - true, - true, - true, - true - ] - }, - "hash": "cf21b61aa5b5e47744235c860e2d9b90d7654a4b5d0ec32085b178399f701bfc" -} diff --git a/backend/.sqlx/query-d768bbc46f8a9c4289b918c88ca50aa180b29bbe931d948f6e61976f71b7cdb9.json b/backend/.sqlx/query-d768bbc46f8a9c4289b918c88ca50aa180b29bbe931d948f6e61976f71b7cdb9.json deleted file mode 100644 index 22b38a5217..0000000000 --- a/backend/.sqlx/query-d768bbc46f8a9c4289b918c88ca50aa180b29bbe931d948f6e61976f71b7cdb9.json +++ /dev/null @@ -1,22 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT premium FROM workspace WHERE id = $1", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "premium", - "type_info": "Bool" - } - ], - "parameters": { - "Left": [ - "Text" - ] - }, - "nullable": [ - false - ] - }, - "hash": "d768bbc46f8a9c4289b918c88ca50aa180b29bbe931d948f6e61976f71b7cdb9" -} diff --git a/backend/.sqlx/query-d918e1d3a1ccc36ca8b4b25aeeec1cb55aadaf1e6609a1aad09b0515c93a667a.json b/backend/.sqlx/query-d918e1d3a1ccc36ca8b4b25aeeec1cb55aadaf1e6609a1aad09b0515c93a667a.json deleted file mode 100644 index b577ed2dbd..0000000000 --- a/backend/.sqlx/query-d918e1d3a1ccc36ca8b4b25aeeec1cb55aadaf1e6609a1aad09b0515c93a667a.json +++ /dev/null @@ -1,22 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n SELECT usage.usage + 1 FROM usage \n WHERE is_workspace = false AND\n month_ = EXTRACT(YEAR FROM current_date) * 12 + EXTRACT(MONTH FROM current_date)\n AND id = $1", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "?column?", - "type_info": "Int4" - } - ], - "parameters": { - "Left": [ - "Text" - ] - }, - "nullable": [ - null - ] - }, - "hash": "d918e1d3a1ccc36ca8b4b25aeeec1cb55aadaf1e6609a1aad09b0515c93a667a" -} diff --git a/backend/.sqlx/query-e7418515b88d14d1fd79078342963e11422eed9a60e0472c68971f4e8f8735d9.json b/backend/.sqlx/query-e7418515b88d14d1fd79078342963e11422eed9a60e0472c68971f4e8f8735d9.json deleted file mode 100644 index 43ae3b8d07..0000000000 --- a/backend/.sqlx/query-e7418515b88d14d1fd79078342963e11422eed9a60e0472c68971f4e8f8735d9.json +++ /dev/null @@ -1,16 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO usage (id, is_workspace, month_, usage) \n VALUES ($1, $2, EXTRACT(YEAR FROM current_date) * 12 + EXTRACT(MONTH FROM current_date), 0) \n ON CONFLICT (id, is_workspace, month_) DO UPDATE SET usage = usage.usage + $3", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Bool", - "Int4" - ] - }, - "nullable": [] - }, - "hash": "e7418515b88d14d1fd79078342963e11422eed9a60e0472c68971f4e8f8735d9" -} diff --git a/backend/.sqlx/query-ebac391c83dae67831502fbe3f1034210afce7902a69f7c42bb68160e0464d71.json b/backend/.sqlx/query-ebac391c83dae67831502fbe3f1034210afce7902a69f7c42bb68160e0464d71.json deleted file mode 100644 index ed90669699..0000000000 --- a/backend/.sqlx/query-ebac391c83dae67831502fbe3f1034210afce7902a69f7c42bb68160e0464d71.json +++ /dev/null @@ -1,47 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "select path, tag, concurrent_limit, concurrency_time_window_s, cache_ttl from script where hash = $1 AND workspace_id = $2", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "path", - "type_info": "Varchar" - }, - { - "ordinal": 1, - "name": "tag", - "type_info": "Varchar" - }, - { - "ordinal": 2, - "name": "concurrent_limit", - "type_info": "Int4" - }, - { - "ordinal": 3, - "name": "concurrency_time_window_s", - "type_info": "Int4" - }, - { - "ordinal": 4, - "name": "cache_ttl", - "type_info": "Int4" - } - ], - "parameters": { - "Left": [ - "Int8", - "Text" - ] - }, - "nullable": [ - false, - true, - true, - true, - true - ] - }, - "hash": "ebac391c83dae67831502fbe3f1034210afce7902a69f7c42bb68160e0464d71" -} diff --git a/backend/.sqlx/query-f1e65e52585b27bfb39817e82613063e168f10a53d56f222c7917aeba020fff3.json b/backend/.sqlx/query-f1e65e52585b27bfb39817e82613063e168f10a53d56f222c7917aeba020fff3.json deleted file mode 100644 index 7a578ce15c..0000000000 --- a/backend/.sqlx/query-f1e65e52585b27bfb39817e82613063e168f10a53d56f222c7917aeba020fff3.json +++ /dev/null @@ -1,14 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "DELETE FROM email_to_igroup WHERE igroup = $1", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Text" - ] - }, - "nullable": [] - }, - "hash": "f1e65e52585b27bfb39817e82613063e168f10a53d56f222c7917aeba020fff3" -} diff --git a/backend/.sqlx/query-ff89e9f0941507d5cbfbb27e7c3b1ebad6b2b0a836334a5e20f151711fc11370.json b/backend/.sqlx/query-ff89e9f0941507d5cbfbb27e7c3b1ebad6b2b0a836334a5e20f151711fc11370.json deleted file mode 100644 index 9012e34322..0000000000 --- a/backend/.sqlx/query-ff89e9f0941507d5cbfbb27e7c3b1ebad6b2b0a836334a5e20f151711fc11370.json +++ /dev/null @@ -1,15 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE workspace_settings SET deploy_to = $1 WHERE workspace_id = $2", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Text" - ] - }, - "nullable": [] - }, - "hash": "ff89e9f0941507d5cbfbb27e7c3b1ebad6b2b0a836334a5e20f151711fc11370" -} diff --git a/backend/migrations/20230905170218_dedicated_worker.down.sql b/backend/migrations/20230905170218_dedicated_worker.down.sql new file mode 100644 index 0000000000..d2f607c5b8 --- /dev/null +++ b/backend/migrations/20230905170218_dedicated_worker.down.sql @@ -0,0 +1 @@ +-- Add down migration script here diff --git a/backend/migrations/20230905170218_dedicated_worker.up.sql b/backend/migrations/20230905170218_dedicated_worker.up.sql new file mode 100644 index 0000000000..8cbcc0a6b2 --- /dev/null +++ b/backend/migrations/20230905170218_dedicated_worker.up.sql @@ -0,0 +1,2 @@ +-- Add up migration script here +ALTER TABLE script ADD COLUMN dedicated_worker boolean; diff --git a/backend/parsers/windmill-parser-py-imports/tests/fixtures/base.sql b/backend/parsers/windmill-parser-py-imports/tests/fixtures/base.sql index e100d2db9f..d75bf4396c 100644 --- a/backend/parsers/windmill-parser-py-imports/tests/fixtures/base.sql +++ b/backend/parsers/windmill-parser-py-imports/tests/fixtures/base.sql @@ -791,6 +791,7 @@ CREATE TABLE public.script ( envs character varying(1000)[], concurrent_limit integer, concurrency_time_window_s integer, + dedicated_worker boolean, CONSTRAINT proper_id CHECK (((path)::text ~ '^[ufg](\/[\w-]+){2,}$'::text)) ); diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index a3e4fce8b3..624a01cc14 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -5894,6 +5894,8 @@ components: type: integer cache_ttl: type: number + dedicated_worker: + type: boolean required: - hash - path @@ -5969,6 +5971,8 @@ components: type: integer cache_ttl: type: number + dedicated_worker: + type: boolean required: - path - summary diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 5c5ebe280e..3ff56e63d4 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -315,9 +315,11 @@ pub async fn get_path_tag_limits_cache_for_hash( Option, Option, Option, + ScriptLang, + Option, )> { let script = sqlx::query!( - "select path, tag, concurrent_limit, concurrency_time_window_s, cache_ttl from script where hash = $1 AND workspace_id = $2", + "select path, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker from script where hash = $1 AND workspace_id = $2", hash, w_id ) @@ -334,6 +336,8 @@ pub async fn get_path_tag_limits_cache_for_hash( script.concurrent_limit, script.concurrency_time_window_s, script.cache_ttl, + script.language, + script.dedicated_worker, )) } @@ -2035,8 +2039,15 @@ pub async fn run_wait_result_script_by_hash( check_queue_too_long(&db, run_query.queue_limit).await?; let hash = script_hash.0; - let (path, tag, concurrent_limit, concurrency_time_window_s, cache_ttl) = - get_path_tag_limits_cache_for_hash(&db, &w_id, hash).await?; + let ( + path, + tag, + concurrent_limit, + concurrency_time_window_s, + cache_ttl, + language, + dedicated_worker, + ) = get_path_tag_limits_cache_for_hash(&db, &w_id, hash).await?; check_scopes(&authed, || format!("run:script/{path}"))?; let args = run_query.add_include_headers(headers, args.unwrap_or_default()); @@ -2054,6 +2065,8 @@ pub async fn run_wait_result_script_by_hash( concurrent_limit: concurrent_limit, concurrency_time_window_s: concurrency_time_window_s, cache_ttl, + language, + dedicated_worker, }, args, &authed.username, @@ -2361,8 +2374,15 @@ pub async fn run_job_by_hash( JsonOrForm(args, raw_string): JsonOrForm, ) -> error::Result<(StatusCode, String)> { let hash = script_hash.0; - let (path, tag, concurrent_limit, concurrency_time_window_s, cache_ttl) = - get_path_tag_limits_cache_for_hash(&db, &w_id, hash).await?; + let ( + path, + tag, + concurrent_limit, + concurrency_time_window_s, + cache_ttl, + language, + dedicated_worker, + ) = get_path_tag_limits_cache_for_hash(&db, &w_id, hash).await?; check_scopes(&authed, || format!("run:script/{path}"))?; let scheduled_for = run_query.get_scheduled_for(&db).await?; @@ -2381,6 +2401,8 @@ pub async fn run_job_by_hash( concurrent_limit: concurrent_limit, concurrency_time_window_s: concurrency_time_window_s, cache_ttl, + language, + dedicated_worker, }, args, &authed.username, diff --git a/backend/windmill-api/src/oauth2.rs b/backend/windmill-api/src/oauth2.rs index 0e86bb4d74..ee29ccdcfa 100644 --- a/backend/windmill-api/src/oauth2.rs +++ b/backend/windmill-api/src/oauth2.rs @@ -799,13 +799,20 @@ async fn slack_command( (JobPayload::Flow(path.to_string()), None) } else { let path = path.strip_prefix("script/").unwrap_or_else(|| path); - let (script_hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl) = - windmill_common::get_latest_deployed_hash_for_path( - &db, - &settings.workspace_id, - path, - ) - .await?; + let ( + script_hash, + tag, + concurrent_limit, + concurrency_time_window_s, + cache_ttl, + language, + dedicated_worker, + ) = windmill_common::get_latest_deployed_hash_for_path( + &db, + &settings.workspace_id, + path, + ) + .await?; ( JobPayload::ScriptHash { hash: script_hash, @@ -813,6 +820,8 @@ async fn slack_command( concurrent_limit, concurrency_time_window_s, cache_ttl, + language, + dedicated_worker, }, tag, ) diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index 1d31393aa5..aaf44a6a89 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/backend/windmill-api/src/scripts.rs @@ -71,6 +71,8 @@ pub struct ScriptWDraft { pub concurrency_time_window_s: Option, #[serde(skip_serializing_if = "Option::is_none")] pub cache_ttl: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub dedicated_worker: Option, } pub fn global_service() -> Router { @@ -374,8 +376,8 @@ async fn create_script( sqlx::query!( "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, \ content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, \ - draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl) \ - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20)", + draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, dedicated_worker) \ + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21)", &w_id, &hash.0, ns.path, @@ -396,6 +398,7 @@ async fn create_script( ns.concurrent_limit, ns.concurrency_time_window_s, ns.cache_ttl, + ns.dedicated_worker ) .execute(&mut tx) .await?; @@ -592,7 +595,7 @@ async fn get_script_by_path_w_draft( let mut tx = user_db.begin(&authed).await?; let script_o = sqlx::query_as::<_, ScriptWDraft>( - "SELECT hash, script.path, summary, description, content, language, kind, tag, schema, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, draft.value as draft FROM script LEFT JOIN draft ON + "SELECT hash, script.path, summary, description, content, language, kind, tag, schema, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, draft.value as draft, dedicated_worker FROM script LEFT JOIN draft ON script.path = draft.path AND script.workspace_id = draft.workspace_id AND draft.typ = 'script' WHERE script.path = $1 AND script.workspace_id = $2 \ AND script.created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND \ diff --git a/backend/windmill-common/src/flows.rs b/backend/windmill-common/src/flows.rs index a06f54129a..87b246eb4b 100644 --- a/backend/windmill-common/src/flows.rs +++ b/backend/windmill-common/src/flows.rs @@ -297,6 +297,17 @@ pub enum FlowModuleValue { Identity, } +impl FlowModuleValue { + pub fn is_simple(&self) -> bool { + match self { + FlowModuleValue::Script { .. } => true, + FlowModuleValue::Flow { .. } => true, + FlowModuleValue::RawScript { .. } => true, + _ => false, + } + } +} + fn ordered_map(value: &HashMap, serializer: S) -> Result where S: Serializer, diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs index ede9972f41..4074cde6ae 100644 --- a/backend/windmill-common/src/jobs.rs +++ b/backend/windmill-common/src/jobs.rs @@ -185,6 +185,8 @@ pub enum JobPayload { concurrent_limit: Option, concurrency_time_window_s: Option, cache_ttl: Option, + dedicated_worker: Option, + language: ScriptLang, }, Code(RawCode), Dependencies { @@ -232,8 +234,15 @@ pub async fn script_path_to_payload( let (job_payload, tag) = if script_path.starts_with("hub/") { (JobPayload::ScriptHub { path: script_path.to_owned() }, None) } else { - let (script_hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl) = - get_latest_deployed_hash_for_path(db, w_id, script_path).await?; + let ( + script_hash, + tag, + concurrent_limit, + concurrency_time_window_s, + cache_ttl, + language, + dedicated_worker, + ) = get_latest_deployed_hash_for_path(db, w_id, script_path).await?; ( JobPayload::ScriptHash { hash: script_hash, @@ -241,6 +250,8 @@ pub async fn script_path_to_payload( concurrent_limit, concurrency_time_window_s, cache_ttl: cache_ttl, + language, + dedicated_worker, }, tag, ) @@ -252,9 +263,16 @@ pub async fn script_hash_to_tag_and_limits<'c>( script_hash: &ScriptHash, db: &mut Transaction<'c, Postgres>, w_id: &String, -) -> error::Result<(Option, Option, Option, Option)> { +) -> error::Result<( + Option, + Option, + Option, + Option, + ScriptLang, + Option, +)> { let script = sqlx::query!( - "select tag, concurrent_limit, concurrency_time_window_s, cache_ttl from script where hash = $1 AND workspace_id = $2", + "select tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker from script where hash = $1 AND workspace_id = $2", script_hash.0, w_id ) @@ -270,6 +288,8 @@ pub async fn script_hash_to_tag_and_limits<'c>( script.concurrent_limit, script.concurrency_time_window_s, script.cache_ttl, + script.language, + script.dedicated_worker, )) } diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index af5b33441a..8058c3a75e 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -9,6 +9,7 @@ use std::net::SocketAddr; use error::Error; +use scripts::ScriptLang; use sqlx::{Pool, Postgres}; pub mod apps; @@ -170,9 +171,11 @@ pub async fn get_latest_deployed_hash_for_path( Option, Option, Option, + ScriptLang, + Option, )> { let r_o = sqlx::query!( - "select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl from script where path = $1 AND workspace_id = $2 AND + "select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker 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, @@ -189,6 +192,8 @@ pub async fn get_latest_deployed_hash_for_path( script.concurrent_limit, script.concurrency_time_window_s, script.cache_ttl, + script.language, + script.dedicated_worker, )) } @@ -202,9 +207,11 @@ pub async fn get_latest_hash_for_path<'c>( Option, Option, Option, + ScriptLang, + Option, )> { let r_o = sqlx::query!( - "select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl from script where path = $1 AND workspace_id = $2 AND + "select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker 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, @@ -221,5 +228,7 @@ pub async fn get_latest_hash_for_path<'c>( script.concurrent_limit, script.concurrency_time_window_s, script.cache_ttl, + script.language, + script.dedicated_worker, )) } diff --git a/backend/windmill-common/src/scripts.rs b/backend/windmill-common/src/scripts.rs index dacb538d6a..fa669dc63e 100644 --- a/backend/windmill-common/src/scripts.rs +++ b/backend/windmill-common/src/scripts.rs @@ -215,6 +215,7 @@ pub struct NewScript { pub concurrent_limit: Option, pub concurrency_time_window_s: Option, pub cache_ttl: Option, + pub dedicated_worker: Option, } #[derive(Deserialize)] diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 14a56adb36..1b2b1f3c98 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -14,6 +14,7 @@ use std::time::Instant; use anyhow::Context; use async_recursion::async_recursion; use chrono::{DateTime, Duration, Utc}; +use itertools::Itertools; use reqwest::Client; use rsmq_async::RsmqConnection; use serde_json::json; @@ -86,10 +87,30 @@ lazy_static::lazy_static! { "hub".to_string(), "other".to_string()]; - pub static ref ACCEPTED_TAGS: Vec = std::env::var("WORKER_TAGS") + pub static ref DEDICATED_WORKER: Option<(String, String)> = std::env::var("DEDICATED_WORKER") + .ok() + .map(|x| { + let splitted = x.split(':').to_owned().collect_vec(); + if splitted.len() != 2 { + panic!("DEDICATED_WORKER should be in the form of :") + } else { + let workspace = splitted[0]; + let script_path = splitted[1]; + (workspace.to_string(), script_path.to_string()) + } + }); + + pub static ref ACCEPTED_TAGS: Vec = { + let worker_tags = std::env::var("WORKER_TAGS") .ok() .map(|x| x.split(',').map(|x| x.to_string()).collect()) - .unwrap_or_else(|| DEFAULT_TAGS.clone()) ; + .unwrap_or_else(|| DEFAULT_TAGS.clone()); + if let Some(ref dedicated_worker) = DEDICATED_WORKER.as_ref() { + vec![format!("{}:{}", dedicated_worker.0, dedicated_worker.1)] + } else { + worker_tags + } + }; pub static ref IS_WORKER_TAGS_DEFINED: bool = std::env::var("WORKER_TAGS").ok().is_some(); @@ -1511,6 +1532,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( concurrent_limit, concurrency_time_window_s, cache_ttl, + dedicated_worker, ) = match job_payload { JobPayload::ScriptHash { hash, @@ -1518,30 +1540,20 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( concurrent_limit, concurrency_time_window_s, cache_ttl, - } => { - let language = fetch_scalar_isolated!(sqlx::query_scalar!( - "SELECT language as \"language: ScriptLang\" FROM script WHERE hash = $1 AND workspace_id = $2", - hash.0, - workspace_id - ), tx) - .ok().flatten() - .ok_or_else(||{ - Error::InternalErr(format!( - "fetching language for hash {hash} in {workspace_id}" - )) - })?; - ( - Some(hash.0), - Some(path), - None, - JobKind::Script, - None, - Some(language), - concurrent_limit, - concurrency_time_window_s, - cache_ttl, - ) - } + language, + dedicated_worker, + } => ( + Some(hash.0), + Some(path), + None, + JobKind::Script, + None, + Some(language), + concurrent_limit, + concurrency_time_window_s, + cache_ttl, + dedicated_worker, + ), JobPayload::ScriptHub { path } => { ( None, @@ -1554,6 +1566,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( None, None, None, + None, ) } JobPayload::Code(RawCode { @@ -1574,6 +1587,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( concurrent_limit, concurrency_time_window_s, cache_ttl, + None, ), JobPayload::Dependencies { hash, dependencies, language, path } => ( Some(hash.0), @@ -1585,6 +1599,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( None, None, None, + None, ), JobPayload::FlowDependencies { path } => { let value_json = fetch_scalar_isolated!( @@ -1611,6 +1626,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( None, None, None, + None, ) } JobPayload::AppDependencies { path, version } => ( @@ -1623,6 +1639,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( None, None, None, + None, ), JobPayload::RawFlow { value, path } => ( None, @@ -1634,6 +1651,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( value.concurrent_limit.clone(), value.concurrency_time_window_s, value.cache_ttl.map(|x| x as i32), + None, ), JobPayload::Flow(flow) => { let value_json = fetch_scalar_isolated!( @@ -1660,6 +1678,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( value.concurrent_limit.clone(), value.concurrency_time_window_s, value.cache_ttl.map(|x| x as i32), + None, ) } JobPayload::Identity => ( @@ -1672,6 +1691,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( None, None, None, + None, ), JobPayload::Noop => ( None, @@ -1683,6 +1703,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( None, None, None, + None, ), }; @@ -1737,7 +1758,13 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( let flow_status = raw_flow.as_ref().map(FlowStatus::new); - let tag = if job_kind == JobKind::Script_Hub { + let tag = if dedicated_worker.is_some_and(|x| x) { + format!( + "{}:{}", + workspace_id, + script_path.clone().expect("dedicated script has a path") + ) + } else if job_kind == JobKind::Script_Hub { "hub".to_string() } else { if tag == Some("".to_string()) { diff --git a/backend/windmill-queue/src/schedule.rs b/backend/windmill-queue/src/schedule.rs index 460e2dc603..a2bb20468d 100644 --- a/backend/windmill-queue/src/schedule.rs +++ b/backend/windmill-queue/src/schedule.rs @@ -74,13 +74,20 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>( let (payload, tag) = if schedule.is_flow { (JobPayload::Flow(schedule.script_path), None) } else { - let (hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl) = - windmill_common::get_latest_hash_for_path( - tx.transaction_mut(), - &schedule.workspace_id, - &schedule.script_path, - ) - .await?; + let ( + hash, + tag, + concurrent_limit, + concurrency_time_window_s, + cache_ttl, + language, + dedicated_worker, + ) = windmill_common::get_latest_hash_for_path( + tx.transaction_mut(), + &schedule.workspace_id, + &schedule.script_path, + ) + .await?; ( JobPayload::ScriptHash { hash, @@ -88,6 +95,8 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>( concurrent_limit: concurrent_limit, concurrency_time_window_s: concurrency_time_window_s, cache_ttl: cache_ttl, + dedicated_worker, + language, }, tag, ) diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index 66503784bf..7b59172972 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -1,9 +1,18 @@ use std::{collections::HashMap, process::Stdio}; +#[cfg(feature = "enterprise")] +use std::collections::VecDeque; + +#[cfg(feature = "enterprise")] +use anyhow::Context; + use base64::Engine; use itertools::Itertools; use uuid::Uuid; +#[cfg(feature = "enterprise")] +use crate::{common::build_envs_map, JobCompleted}; + use crate::{ common::{ create_args_and_out_file, get_reserved_variables, handle_child, read_result, set_logs, @@ -12,10 +21,25 @@ use crate::{ AuthedClientBackgroundTask, BUN_CACHE_DIR, BUN_PATH, DISABLE_NSJAIL, DISABLE_NUSER, NPM_CONFIG_REGISTRY, NSJAIL_PATH, PATH_ENV, }; -use tokio::{fs::File, io::AsyncReadExt, process::Command}; -use windmill_common::error::Result; + +#[cfg(feature = "enterprise")] +use crate::MAX_BUFFERED_DEDICATED_JOBS; + +use tokio::{fs::File, process::Command}; + +#[cfg(feature = "enterprise")] +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; + +use tokio::io::AsyncReadExt; + +#[cfg(feature = "enterprise")] +use tokio::sync::mpsc::{Receiver, Sender}; + +#[cfg(feature = "enterprise")] +use windmill_common::variables; + use windmill_common::{ - error::{self}, + error::{self, Result}, jobs::QueuedJob, }; use windmill_parser::Typ; @@ -26,8 +50,8 @@ const RELATIVE_BUN_BUILDER: &str = include_str!("../loader_builder.bun.ts"); const NSJAIL_CONFIG_RUN_BUN_CONTENT: &str = include_str!("../nsjail/run.bun.config.proto"); -const BUN_LOCKB_SPLIT: &str = "\n//bun.lockb\n"; -const EMPTY_FILE: &str = ""; +pub const BUN_LOCKB_SPLIT: &str = "\n//bun.lockb\n"; +pub const EMPTY_FILE: &str = ""; pub async fn gen_lockfile( logs: &mut String, @@ -391,7 +415,7 @@ plugin(p) read_result(job_dir).await } -fn get_common_bun_proc_envs(base_internal_url: &str) -> HashMap { +pub fn get_common_bun_proc_envs(base_internal_url: &str) -> HashMap { let mut deno_envs: HashMap = HashMap::from([ (String::from("PATH"), PATH_ENV.clone()), (String::from("DISABLE_COLORS"), "0".to_string()), @@ -413,3 +437,298 @@ fn get_common_bun_proc_envs(base_internal_url: &str) -> HashMap } return deno_envs; } + +#[cfg(feature = "enterprise")] +pub async fn start_worker( + requirements_o: Option, + db: &sqlx::Pool, + inner_content: &str, + base_internal_url: &str, + job_dir: &str, + worker_name: &str, + envs: HashMap, + w_id: &str, + script_path: &str, + token: &str, + job_completed_tx: Sender, + mut jobs_rx: Receiver, + mut killpill_rx: tokio::sync::broadcast::Receiver<()>, +) -> Result<()> { + let mut logs = "".to_string(); + let _ = write_file(job_dir, "main.ts", inner_content).await?; + let common_bun_proc_envs: HashMap = + get_common_bun_proc_envs(&base_internal_url); + let context = variables::get_reserved_variables( + w_id, + &token, + "dedicated_worker@windmill.dev", + "dedicated_worker", + "NOT_AVAILABLE", + "dedicated_worker", + Some(script_path.to_string()), + None, + None, + None, + None, + ) + .to_vec(); + let context_envs = build_envs_map(context); + if let Some(reqs) = requirements_o { + let splitted = reqs.split(BUN_LOCKB_SPLIT).collect::>(); + if splitted.len() != 2 { + return Err(error::Error::ExecutionErr( + format!("Invalid requirements, expectd to find //bun.lockb split pattern in reqs. Found: |{reqs}|") + )); + } + let _ = write_file(job_dir, "package.json", &splitted[0]).await?; + let lockb = splitted[1]; + if lockb != EMPTY_FILE { + let _ = write_file_binary( + job_dir, + "bun.lockb", + &base64::engine::general_purpose::STANDARD + .decode(&splitted[1]) + .map_err(|_| { + error::Error::InternalErr("Could not decode bun.lockb".to_string()) + })?, + ) + .await?; + } + install_lockfile( + &mut logs, + &Uuid::nil(), + &w_id, + db, + job_dir, + worker_name, + common_bun_proc_envs.clone(), + ) + .await?; + } else if !*DISABLE_NSJAIL { + logs.push_str("\n\n--- BUN INSTALL ---\n"); + let _ = gen_lockfile( + &mut logs, + &Uuid::nil(), + &w_id, + db, + token, + &script_path, + job_dir, + base_internal_url, + worker_name, + false, + ) + .await?; + } + + { + // let mut start = Instant::now(); + let args = windmill_parser_ts::parse_deno_signature(inner_content, true)?.args; + let dates = args + .iter() + .enumerate() + .filter_map(|(i, x)| { + if matches!(x.typ, Typ::Datetime) { + Some(i) + } else { + None + } + }) + .map(|x| return format!("args[{x}] = args[{x}] ? new Date(args[{x}]) : undefined")) + .join("\n"); + + let spread = args.into_iter().map(|x| x.name).join(","); + // logs.push_str(format!("infer args: {:?}\n", start.elapsed().as_micros()).as_str()); + // we cannot use Bun.read and Bun.write because it results in an EBADF error on cloud + let wrapper_content: String = format!( + r#" +import {{ main }} from "./main.ts"; + +BigInt.prototype.toJSON = function () {{ + return this.toString(); +}}; + +{dates} + +let stdout = Bun.stdout.writer(); +// let stdout = Bun.file("output.txt").writer(); +stdout.write('start\n'); + +for await (const chunk of Bun.stdin.stream()) {{ + const lines = Buffer.from(chunk).toString(); + let exit = false; + for (const line of lines.trim().split("\n")) {{ + // stdout.write('s: ' + line + 'EE\n'); + if (line === "end") {{ + exit = true; + break; + }} + try {{ + let {{ {spread} }} = JSON.parse(line) + let res: any = await main(...[ {spread} ]); + stdout.write(JSON.stringify(res ?? null, (key, value) => typeof value === 'undefined' ? null : value) + '\n'); + }} catch (e) {{ + stdout.write(JSON.stringify({{ error: {{ message: e.message, name: e.name, stack: e.stack, line: line }}}}) + '\n'); + }} + stdout.flush(); + }} + if (exit) {{ + break; + }} +}} +"#, + ); + write_file(job_dir, "wrapper.ts", &wrapper_content).await?; + } + + let reserved_variables = windmill_common::variables::get_reserved_variables( + w_id, + token, + "dedicated_worker", + "dedicated_worker", + Uuid::nil().to_string().as_str(), + "dedicted_worker", + Some(script_path.to_string()), + None, + None, + None, + None, + ); + + let _ = write_file( + &job_dir, + "loader.bun.ts", + &format!( + r#" +import {{ plugin }} from "bun"; + +{} + +plugin(p) +"#, + RELATIVE_BUN_LOADER + .replace("W_ID", &w_id) + .replace("BASE_INTERNAL_URL", base_internal_url) + .replace("TOKEN", token) + .replace("CURRENT_PATH", script_path) + ), + ) + .await?; + + //do not cache local dependencies + let mut child = { + let script_path = format!("{job_dir}/wrapper.ts"); + let args = vec![ + "run", + "-i", + "--prefer-offline", + "-r", + "./loader.bun.ts", + &script_path, + ]; + Command::new(&*BUN_PATH) + .current_dir(job_dir) + .env_clear() + .envs(context_envs) + .envs(envs) + .envs( + reserved_variables + .iter() + .map(|x| (x.name.clone(), x.value.clone())) + .collect::>(), + ) + .envs(common_bun_proc_envs) + .args(args) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn()? + }; + + let stdout = child + .stdout + .take() + .expect("child did not have a handle to stdout"); + + let mut reader = BufReader::new(stdout).lines(); + + let mut stdin = child + .stdin + .take() + .expect("child did not have a handle to stdin"); + + // Ensure the child process is spawned in the runtime so it can + // make progress on its own while we await for any output. + let child = tokio::spawn(async move { + let status = child + .wait() + .await + .expect("child process encountered an error"); + + println!("child status was: {}", status); + }); + + let mut jobs = VecDeque::with_capacity(MAX_BUFFERED_DEDICATED_JOBS); + // let mut i = 0; + // let mut j = 0; + let mut alive = true; + loop { + tokio::select! { + biased; + _ = killpill_rx.recv(), if alive => { + println!("received killpill for dedicated worker"); + alive = false; + if let Err(e) = write_stdin(&mut stdin, "end").await { + tracing::info!("Could not write end message to stdin: {e:?}") + } + }, + line = reader.next_line() => { + // j += 1; + + if let Some(line) = line.expect("line is ok") { + if line == "start" { + tracing::info!("dedicated worker process started"); + continue; + } + tracing::debug!("processed job"); + + let result = serde_json::from_str(&line).expect("json is ok"); + let job: QueuedJob = jobs.pop_front().expect("pop"); + job_completed_tx.send(JobCompleted { job , result, logs: "".to_string(), success: true, cached_res_path: None, token: token.to_string() }).await.unwrap(); + } else { + tracing::info!("dedicated worker process exited"); + break; + } + } + job = jobs_rx.recv(), if alive && jobs.len() < MAX_BUFFERED_DEDICATED_JOBS => { + // i += 1; + if let Some(job) = job { + tracing::debug!("received job"); + jobs.push_back(job.clone()); + // write_stdin(&mut stdin, &serde_json::to_string(&job.args.unwrap_or_else(|| serde_json::json!({"x": job.id}))).expect("serialize")).await?; + write_stdin(&mut stdin, &serde_json::to_string(&job.args.unwrap_or_else(|| serde_json::json!({}))).expect("serialize")).await?; + stdin.flush().await.context("stdin flush")?; + } else { + tracing::debug!("job channel closed"); + alive = false; + if let Err(e) = write_stdin(&mut stdin, "end").await { + tracing::error!("Could not write end message to stdin: {e:?}") + } + } + } + } + } + + child + .await + .map_err(|e| anyhow::anyhow!("child process encountered an error: {e}"))?; + tracing::info!("dedicated worker child process exited successfully"); + Ok(()) +} + +#[cfg(feature = "enterprise")] +async fn write_stdin(stdin: &mut tokio::process::ChildStdin, s: &str) -> error::Result<()> { + let _ = &stdin.write_all(format!("{s}\n").as_bytes()).await?; + stdin.flush().await.context("stdin flush")?; + Ok(()) +} diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index 4b33a10f98..0b30a45c1e 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -6,6 +6,7 @@ use windmill_api_client::{types::CreateResource, Client}; use windmill_common::{ error::{self, Error}, jobs::QueuedJob, + variables::ContextualVariable, }; use windmill_queue::CLOUD_HOSTED; @@ -189,10 +190,12 @@ pub async fn get_reserved_variables( ) .to_vec(); - let mut r: HashMap = variables - .into_iter() - .map(|rv| (rv.name, rv.value)) - .collect(); + Ok(build_envs_map(variables)) +} + +pub fn build_envs_map(context: Vec) -> HashMap { + let mut r: HashMap = + context.into_iter().map(|rv| (rv.name, rv.value)).collect(); if let Some(ref envs) = *WHITELIST_ENVS { for e in envs { @@ -200,9 +203,8 @@ pub async fn get_reserved_variables( } } - Ok(r) + r } - async fn get_mem_peak(pid: Option, nsjail: bool) -> i32 { if pid.is_none() { return -1; @@ -273,12 +275,15 @@ pub async fn handle_child( /* the cancellation future is polled on by `wait_on_child` while * waiting for the child to exit normally */ let update_job = async { + if job_id == Uuid::nil() { + return; + } let db = db.clone(); let mut interval = interval(update_job_interval); interval.set_missed_tick_behavior(MissedTickBehavior::Skip); - let mut i = 1; + let mut i = 0; loop { tokio::select!( _ = rx.recv() => break, @@ -357,7 +362,7 @@ pub async fn handle_child( result = child.wait() => return result.map(Ok), Ok(()) = too_many_logs.changed() => KillReason::TooManyLogs, _ = sleep(timeout_duration) => KillReason::Timeout, - _ = update_job => KillReason::Cancelled, + _ = update_job, if job_id != Uuid::nil() => KillReason::Cancelled, }; tx.send(()).expect("rx should never be dropped"); drop(tx); @@ -538,7 +543,7 @@ fn child_joined_output_stream( stream::select(lines_to_stream(stderr), lines_to_stream(stdout)) } -fn lines_to_stream( +pub fn lines_to_stream( mut lines: tokio::io::Lines, ) -> impl futures::Stream> { stream::poll_fn(move |cx| { diff --git a/backend/windmill-worker/src/dedicated_worker.rs b/backend/windmill-worker/src/dedicated_worker.rs new file mode 100644 index 0000000000..c0ac64fd4c --- /dev/null +++ b/backend/windmill-worker/src/dedicated_worker.rs @@ -0,0 +1,5 @@ +// use tokio::sync::mpsc; + +// pub fn create_dedicated_worker() { +// let (job_completed_tx, mut new_job) = mpsc::channel::(100); +// } diff --git a/backend/windmill-worker/src/lib.rs b/backend/windmill-worker/src/lib.rs index 5d5dbb2bfa..94c582eeaa 100644 --- a/backend/windmill-worker/src/lib.rs +++ b/backend/windmill-worker/src/lib.rs @@ -6,6 +6,7 @@ mod snowflake_executor; mod bash_executor; mod bun_executor; mod common; +mod dedicated_worker; mod deno_executor; mod global_cache; mod go_executor; diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index b6b7d52c10..c87384bb86 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -31,8 +31,8 @@ use windmill_common::{ DB, IS_READY, METRICS_ENABLED, }; use windmill_queue::{ - canceled_job_to_result, get_queued_job, pull, ACCEPTED_TAGS, CLOUD_HOSTED, HTTP_CLIENT, - IS_WORKER_TAGS_DEFINED, + canceled_job_to_result, get_queued_job, pull, ACCEPTED_TAGS, CLOUD_HOSTED, DEDICATED_WORKER, + HTTP_CLIENT, IS_WORKER_TAGS_DEFINED, }; use serde_json::{json, Value}; @@ -43,6 +43,7 @@ use tokio::{ mpsc::{self, Sender}, Barrier, RwLock, }, + task::JoinHandle, time::Instant, }; @@ -58,6 +59,9 @@ use crate::global_cache::{ copy_denogo_cache_from_bucket_as_tar, copy_tmp_cache_to_cache, }; +#[cfg(feature = "enterprise")] +use crate::bun_executor::start_worker; + use windmill_queue::{add_completed_job, add_completed_job_error}; #[cfg(feature = "benchmark")] @@ -181,6 +185,8 @@ pub const DEFAULT_NATIVE_JOBS: usize = 1; const VACUUM_PERIOD: u32 = 10000; +pub const MAX_BUFFERED_DEDICATED_JOBS: usize = 3; + lazy_static::lazy_static! { static ref SLEEP_QUEUE: u64 = std::env::var("SLEEP_QUEUE") @@ -270,6 +276,8 @@ lazy_static::lazy_static! { pub static ref CAN_PULL: Arc> = Arc::new(RwLock::new(())); + + } //only matter if CLOUD_HOSTED pub const MAX_RESULT_SIZE: usize = 1024 * 1024 * 2; // 2MB @@ -335,7 +343,7 @@ pub async fn run_worker, + mut killpill_rx: tokio::sync::broadcast::Receiver<()>, base_internal_url: &str, rsmq: Option, _sync_barrier: Arc>>, @@ -583,8 +591,6 @@ pub async fn run_worker = None; let mut last_checked_suspended = Instant::now(); @@ -599,6 +605,81 @@ pub async fn run_worker(MAX_BUFFERED_DEDICATED_JOBS); + let killpill_rx = killpill_rx.resubscribe(); + let db = db.clone(); + let worker_dir = worker_dir.clone(); + let base_internal_url = base_internal_url.to_string(); + let worker_name = worker_name.clone(); + let job_completed_tx = job_completed_tx.clone(); + let job_dir = format!("{}/dedicated", worker_dir); + tokio::fs::create_dir_all(&job_dir) + .await + .expect("create dir"); + let handle = tokio::spawn(async move { + let token = rd_string(30); + if let Err(e) = sqlx::query_scalar!( + "INSERT INTO token + (token, label, super_admin, email) + VALUES ($1, $2, $3, $4)", + token, + "dedicated_worker", + true, + "dedicated_worker@windmill.dev" + ) + .execute(&db) + .await + { + panic!("failed to create token for dedicated worker: {:?}", e) + }; + + let (content, lock, _language, envs) = sqlx::query_as::<_, (String, Option, Option, Option>)>( + "SELECT content, lock, language, envs 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)", + ) + .bind(&_script_path) + .bind(&_workspace) + .fetch_optional(&db) + .await.expect("Failed to fetch script for dedicated worker") + .expect(&format!("Failed to fetch script `{_script_path}` in workspace {_workspace} for dedicated worker")); + + let worker_envs = build_envs(envs).expect("failed to build envs"); + if let Err(e) = start_worker( + lock, + &db, + &content, + &base_internal_url, + &job_dir, + &worker_name, + worker_envs, + &_workspace, + &_script_path, + &token, + job_completed_tx, + dedicated_worker_rx, + killpill_rx, + ) + .await + { + tracing::error!("error in dedicated worker: {:?}", e) + } + }); + (Some(dedicated_worker_tx), Some(handle)) + } + } else { + (None, None) as (Option>, Option>) + }; + loop { #[cfg(feature = "benchmark")] let loop_start = Instant::now(); @@ -689,9 +770,7 @@ pub async fn run_worker { + _ = killpill_rx.recv() => { #[cfg(feature = "enterprise")] if let Some(copy_cache_from_bucket_handle) = copy_cache_from_bucket_handle.as_ref() { if !copy_cache_from_bucket_handle.is_finished() { @@ -768,10 +847,6 @@ pub async fn run_worker>, +) -> windmill_common::error::Result> { + let mut envs = if *CLOUD_HOSTED || envs.is_none() { + HashMap::new() + } else { + let mut hm = HashMap::new(); + for s in envs.unwrap() { + let (k, v) = s.split_once('=').ok_or_else(|| { + Error::BadRequest(format!( + "Invalid env var: {}. Must be in the form of KEY=VALUE", + s + )) + })?; + hm.insert(k.to_string(), v.to_string()); + } + hm + }; + + if let Some(ref env) = *HTTPS_PROXY { + envs.insert("HTTPS_PROXY".to_string(), env.to_string()); + } + if let Some(ref env) = *HTTP_PROXY { + envs.insert("HTTP_PROXY".to_string(), env.to_string()); + } + if let Some(ref env) = *NO_PROXY { + envs.insert("NO_PROXY".to_string(), env.to_string()); + } + Ok(envs) +} + #[tracing::instrument(level = "trace", skip_all)] async fn handle_code_execution_job( job: &QueuedJob, @@ -1668,31 +1783,8 @@ mount {{ }; // println!("handle lang job {:?}", SystemTime::now()); - let mut envs = if *CLOUD_HOSTED || envs.is_none() { - HashMap::new() - } else { - let mut hm = HashMap::new(); - for s in envs.unwrap() { - let (k, v) = s.split_once('=').ok_or_else(|| { - Error::BadRequest(format!( - "Invalid env var: {}. Must be in the form of KEY=VALUE", - s - )) - })?; - hm.insert(k.to_string(), v.to_string()); - } - hm - }; - if let Some(ref env) = *HTTPS_PROXY { - envs.insert("HTTPS_PROXY".to_string(), env.to_string()); - } - if let Some(ref env) = *HTTP_PROXY { - envs.insert("HTTP_PROXY".to_string(), env.to_string()); - } - if let Some(ref env) = *NO_PROXY { - envs.insert("NO_PROXY".to_string(), env.to_string()); - } + let envs = build_envs(envs)?; let result: error::Result = match language { None => { diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 4cdb7699a4..b574425b3b 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -121,14 +121,12 @@ pub async fn update_flow_status_after_job_completion_internal< let (should_continue_flow, flow_job, stop_early, skip_if_stop_early, nresult) = { // tracing::debug!("UPDATE FLOW STATUS: {flow:?} {success} {result:?} {w_id} {depth}"); - let mut tx: QueueTransaction<'_, _> = (rsmq.clone(), db.begin().await?).into(); - let old_status_json = sqlx::query_scalar!( "SELECT flow_status FROM queue WHERE id = $1 AND workspace_id = $2", flow, w_id ) - .fetch_one(&mut tx) + .fetch_one(db) .await .map_err(|e| { Error::InternalErr(format!( @@ -157,12 +155,8 @@ pub async fn update_flow_status_after_job_completion_internal< module_status, FlowStatusModule::InProgress { iterator: Some(_), .. } ) { - let (loop_failures, parallelism) = compute_skip_loop_failures_and_parallelism( - flow, - old_status.step, - tx.transaction_mut(), - ) - .await?; + let (loop_failures, parallelism) = + compute_skip_loop_failures_and_parallelism(flow, old_status.step, db).await?; (loop_failures.unwrap_or(false), parallelism) } else { (false, None) @@ -186,7 +180,7 @@ pub async fn update_flow_status_after_job_completion_internal< old_status.step, flow ) - .fetch_one(&mut tx) + .fetch_one(db) .await .map_err(|e| Error::InternalErr(format!("retrieval of stop_early_expr from state: {e}")))?; @@ -204,16 +198,16 @@ pub async fn update_flow_status_after_job_completion_internal< FlowStatusModule::InProgress { branchall: Some(BranchAllStatus { branch, .. }), .. - } => { - compute_skip_branchall_failure(flow, old_status.step, *branch, tx.transaction_mut()) - .await? - .unwrap_or(false) - } + } => compute_skip_branchall_failure(flow, old_status.step, *branch, db) + .await? + .unwrap_or(false), _ => false, }; let skip_failure = skip_branch_failure || skip_loop_failures; + let mut tx: QueueTransaction<'_, _> = (rsmq.clone(), db.begin().await?).into(); + let (inc_step_counter, new_status) = match module_status { FlowStatusModule::InProgress { iterator, @@ -274,6 +268,7 @@ pub async fn update_flow_status_after_job_completion_internal< .into_iter() .all(|x| x) { + success = true; FlowStatusModule::Success { id: module_status.id(), job: job_id_for_status.clone(), @@ -651,10 +646,10 @@ pub async fn update_flow_status_after_job_completion_internal< } } -async fn compute_skip_loop_failures_and_parallelism<'c>( +async fn compute_skip_loop_failures_and_parallelism( flow: Uuid, step: i32, - tx: &mut sqlx::Transaction<'c, sqlx::Postgres>, + db: &DB, ) -> Result<(Option, Option), Error> { sqlx::query_as( " @@ -665,7 +660,7 @@ async fn compute_skip_loop_failures_and_parallelism<'c>( ) .bind(step) .bind(flow) - .fetch_one(&mut **tx) + .fetch_one(db) .await .map(|(v, n)| (v,n)) .map_err(|e| Error::InternalErr(format!("error during retrieval of skip_loop_failures: {e}"))) @@ -675,7 +670,7 @@ async fn compute_skip_branchall_failure<'c>( flow: Uuid, step: i32, branch: usize, - tx: &mut sqlx::Transaction<'c, sqlx::Postgres>, + db: &DB, ) -> Result, Error> { sqlx::query_as( " @@ -687,7 +682,7 @@ async fn compute_skip_branchall_failure<'c>( .bind(step) .bind(branch as i32) .bind(flow) - .fetch_one(&mut **tx) + .fetch_one(db) .await .map(|(v,)| v) .map_err(|e| Error::InternalErr(format!("error during retrieval of skip_loop_failures: {e}"))) @@ -1294,7 +1289,7 @@ async fn push_next_flow_job FlowModuleValue::Script { input_transforms, .. } | FlowModuleValue::RawScript { input_transforms, .. } | FlowModuleValue::Flow { input_transforms, .. } => { - let ctx = get_transform_context(&flow_job, previous_id.clone(), &status).await?; + let ctx = get_transform_context(&flow_job, &previous_id, &status).await?; transform_context = Some(ctx); let by_id = transform_context.as_ref().unwrap(); transform_input( @@ -1342,7 +1337,7 @@ async fn push_next_flow_job &status, &status_module, last_result.clone(), - previous_id, + &previous_id, client, resume_messages.as_slice(), approvers.clone(), @@ -1393,29 +1388,10 @@ async fn push_next_flow_job let payload_tag = match &job_payloads { ContinuePayload::SingleJob(payload) => payload.clone(), ContinuePayload::BranchAllJobs(payloads) => payloads[i].clone(), - ContinuePayload::ForloopJobs { modules, .. } => { - let mut fm = flow.failure_module.clone(); - if let Some(mut failure_module) = flow.failure_module.clone() { - failure_module.id_append(&format!("{}/{}", status.step, i)); - fm = Some(failure_module); - } - JobPayloadWithTag { - payload: JobPayload::RawFlow { - value: FlowValue { - modules: (*modules).clone(), - failure_module: fm.clone(), - same_worker: flow.same_worker, - concurrent_limit: None, - concurrency_time_window_s: None, - skip_expr: None, - cache_ttl: None, - }, - path: Some(format!("{}/forloop", flow_job.script_path())), - }, - tag: None, - } - } + ContinuePayload::ForloopJobs { payload, .. } => payload.clone(), }; + + let transform_inp; let args = match &next_status { NextStatus::AllFlowJobs { branchall: Some(BranchAllStatus { .. }), @@ -1432,15 +1408,34 @@ async fn push_next_flow_job NextStatus::AllFlowJobs { branchall: None, iterator: Some(Iterator { itered, .. }), - .. - } => args.as_ref().map(|args| { - let mut new_args = args.clone(); - new_args.insert( - "iter".to_string(), - json!({ "index": i, "value": itered[i] }), - ); - new_args - }), + simple_input_transforms, + } => { + if let Ok(args) = args.as_ref() { + let mut new_args = args.clone(); + new_args.insert( + "iter".to_string(), + json!({ "index": i, "value": itered[i] }), + ); + if let Some(input_transforms) = simple_input_transforms { + let ctx = get_transform_context(&flow_job, &previous_id, &status).await?; + transform_inp = transform_input( + &Some(serde_json::Value::Object(new_args)), + last_result.clone(), + input_transforms, + resume_messages.as_slice(), + approvers.clone(), + &ctx, + client, + ) + .await; + transform_inp.as_ref().map(|args| args.clone()) + } else { + Ok(new_args) + } + } else { + args.as_ref().map(|args| args.clone()) + } + } _ => args.as_ref().map(|args| args.clone()), }; let (ok, err) = match args { @@ -1528,7 +1523,7 @@ async fn push_next_flow_job parallel: false, } } - NextStatus::AllFlowJobs { iterator, branchall } => FlowStatusModule::InProgress { + NextStatus::AllFlowJobs { iterator, branchall, .. } => FlowStatusModule::InProgress { job: flow_job.id, iterator, flow_jobs: Some(uuids), @@ -1713,6 +1708,7 @@ enum NextStatus { AllFlowJobs { branchall: Option, iterator: Option, + simple_input_transforms: Option>, }, } @@ -1723,7 +1719,7 @@ struct JobPayloadWithTag { } enum ContinuePayload { SingleJob(JobPayloadWithTag), - ForloopJobs { n: usize, modules: Vec }, + ForloopJobs { n: usize, payload: JobPayloadWithTag }, BranchAllJobs(Vec), } @@ -1741,7 +1737,7 @@ async fn compute_next_flow_transform( status: &FlowStatus, status_module: &FlowStatusModule, last_result: serde_json::Value, - previous_id: String, + previous_id: &str, client: &AuthedClient, resumes: &[Value], approvers: Vec, @@ -1764,33 +1760,16 @@ async fn compute_next_flow_transform( match &module.value { FlowModuleValue::Identity => trivial_next_job(JobPayload::Identity), FlowModuleValue::Flow { path, .. } => { - let payload = JobPayload::Flow(path.to_string()); + let payload = flow_to_payload(path); Ok(NextFlowTransform::Continue( - ContinuePayload::SingleJob(JobPayloadWithTag { payload, tag: None }), + ContinuePayload::SingleJob(payload), NextStatus::NextStep, )) } FlowModuleValue::Script { path: script_path, hash: script_hash, .. } => { - let (payload, tag) = if script_hash.is_none() { - script_path_to_payload(script_path, &db, &flow_job.workspace_id).await? - } else { - let hash = script_hash.clone().unwrap(); - let mut tx: sqlx::Transaction<'_, sqlx::Postgres> = db.begin().await?; - let (tag, concurrent_limit, concurrency_time_window_s, cache_ttl) = - script_hash_to_tag_and_limits(&hash, &mut tx, &flow_job.workspace_id).await?; - ( - JobPayload::ScriptHash { - hash, - path: script_path.to_owned(), - concurrent_limit, - concurrency_time_window_s, - cache_ttl: module.cache_ttl.map(|x| x as i32).ok_or(cache_ttl).ok(), - }, - tag, - ) - }; + let payload = script_to_payload(script_hash, script_path, db, flow_job, module).await?; Ok(NextFlowTransform::Continue( - ContinuePayload::SingleJob(JobPayloadWithTag { payload, tag }), + ContinuePayload::SingleJob(payload), NextStatus::NextStep, )) } @@ -1807,25 +1786,27 @@ async fn compute_next_flow_transform( let path = path .clone() .or_else(|| Some(format!("{}/step-{}", flow_job.script_path(), status.step))); + let payload = raw_script_to_payload( + path, + content, + language, + lock, + concurrent_limit, + concurrency_time_window_s, + module, + tag, + ); Ok(NextFlowTransform::Continue( - ContinuePayload::SingleJob(JobPayloadWithTag { - payload: JobPayload::Code(RawCode { - path, - content: content.clone(), - language: language.clone(), - lock: lock.clone(), - concurrent_limit: *concurrent_limit, - concurrency_time_window_s: *concurrency_time_window_s, - cache_ttl: module.cache_ttl.map(|x| x as i32), - }), - tag: tag.clone(), - }), + ContinuePayload::SingleJob(payload), NextStatus::NextStep, )) } /* forloop modules are expected set `iter: { value: Value, index: usize }` as job arguments */ FlowModuleValue::ForloopFlow { modules, iterator, parallel, .. } => { let new_args: &mut Map = &mut Map::new(); + // if it's a simple single step flow, we will collapse it as an optimization and need to pass flow_input as an arg + let is_simple = + modules.len() == 1 && modules[0].value.is_simple() && flow.failure_module.is_none(); let next_loop_status = match status_module { FlowStatusModule::WaitingForPriorSteps { .. } @@ -1919,32 +1900,106 @@ async fn compute_next_flow_transform( failure_module.id_append(&format!("{}/{}", status.step, ns.index)); fm = Some(failure_module); } - Ok(NextFlowTransform::Continue( - ContinuePayload::SingleJob(JobPayloadWithTag { - payload: JobPayload::RawFlow { - value: FlowValue { - modules: (*modules).clone(), - failure_module: fm, - same_worker: flow.same_worker, - concurrent_limit: None, - concurrency_time_window_s: None, - skip_expr: None, - cache_ttl: None, - }, - path: Some(format!("{}/loop-{}", flow_job.script_path(), ns.index)), + let modules = (*modules).clone(); + let inner_path = Some(format!("{}/loop-{}", flow_job.script_path(), ns.index)); + let continue_payload = ContinuePayload::SingleJob(JobPayloadWithTag { + payload: JobPayload::RawFlow { + value: FlowValue { + modules, + failure_module: fm, + same_worker: flow.same_worker, + concurrent_limit: None, + concurrency_time_window_s: None, + skip_expr: None, + cache_ttl: None, }, - tag: None, - }), + path: inner_path, + }, + tag: None, + }); + Ok(NextFlowTransform::Continue( + continue_payload, NextStatus::NextLoopIteration(ns), )) } - LoopStatus::ParallelIteration { itered, .. } => Ok(NextFlowTransform::Continue( - ContinuePayload::ForloopJobs { n: itered.len(), modules: (*modules).clone() }, - NextStatus::AllFlowJobs { - branchall: None, - iterator: Some(windmill_common::flow_status::Iterator { index: 0, itered }), - }, - )), + LoopStatus::ParallelIteration { itered, .. } => { + let inner_path = Some(format!("{}/loop-parrallel", flow_job.script_path(),)); + let continue_payload = if is_simple { + let payload = match &modules[0].value { + FlowModuleValue::Flow { path, .. } => flow_to_payload(path), + FlowModuleValue::Script { + path: script_path, + hash: script_hash, + .. + } => { + script_to_payload(script_hash, script_path, db, flow_job, module) + .await? + } + FlowModuleValue::RawScript { + path, + content, + language, + lock, + tag, + concurrent_limit, + concurrency_time_window_s, + .. + } => raw_script_to_payload( + path.clone().or(inner_path), + content, + language, + lock, + concurrent_limit, + concurrency_time_window_s, + module, + tag, + ), + _ => unreachable!("is simple flow"), + }; + ContinuePayload::ForloopJobs { n: itered.len(), payload: payload } + } else { + let payload = { + JobPayloadWithTag { + payload: JobPayload::RawFlow { + value: FlowValue { + modules: (*modules).clone(), + failure_module: flow.failure_module.clone(), + same_worker: flow.same_worker, + concurrent_limit: None, + concurrency_time_window_s: None, + skip_expr: None, + cache_ttl: None, + }, + path: Some(format!("{}/forloop", flow_job.script_path())), + }, + tag: None, + } + }; + ContinuePayload::ForloopJobs { n: itered.len(), payload } + }; + Ok(NextFlowTransform::Continue( + continue_payload, + NextStatus::AllFlowJobs { + branchall: None, + iterator: Some(windmill_common::flow_status::Iterator { + index: 0, + itered, + }), + simple_input_transforms: if is_simple { + match &modules[0].value { + FlowModuleValue::Script { input_transforms, .. } + | FlowModuleValue::RawScript { input_transforms, .. } + | FlowModuleValue::Flow { input_transforms, .. } => { + Some(input_transforms.clone()) + } + _ => None, + } + } else { + None + }, + }, + )) + } } } FlowModuleValue::BranchOne { branches, default, .. } => { @@ -2068,6 +2123,7 @@ async fn compute_next_flow_transform( len: branches.len(), }), iterator: None, + simple_input_transforms: None, }, )); } else { @@ -2139,9 +2195,74 @@ async fn compute_next_flow_transform( } } +fn raw_script_to_payload( + path: Option, + content: &String, + language: &windmill_common::scripts::ScriptLang, + lock: &Option, + concurrent_limit: &Option, + concurrency_time_window_s: &Option, + module: &FlowModule, + tag: &Option, +) -> JobPayloadWithTag { + JobPayloadWithTag { + payload: JobPayload::Code(RawCode { + path, + content: content.clone(), + language: language.clone(), + lock: lock.clone(), + concurrent_limit: *concurrent_limit, + concurrency_time_window_s: *concurrency_time_window_s, + cache_ttl: module.cache_ttl.map(|x| x as i32), + }), + tag: tag.clone(), + } +} + +fn flow_to_payload(path: &str) -> JobPayloadWithTag { + let payload = JobPayload::Flow(path.to_string()); + JobPayloadWithTag { payload, tag: None } +} + +async fn script_to_payload( + script_hash: &Option, + script_path: &String, + db: &sqlx::Pool, + flow_job: &QueuedJob, + module: &FlowModule, +) -> Result { + let (payload, tag) = if script_hash.is_none() { + script_path_to_payload(script_path, &db, &flow_job.workspace_id).await? + } else { + let hash = script_hash.clone().unwrap(); + let mut tx: sqlx::Transaction<'_, sqlx::Postgres> = db.begin().await?; + let ( + tag, + concurrent_limit, + concurrency_time_window_s, + cache_ttl, + language, + dedicated_worker, + ) = script_hash_to_tag_and_limits(&hash, &mut tx, &flow_job.workspace_id).await?; + ( + JobPayload::ScriptHash { + hash, + path: script_path.to_owned(), + concurrent_limit, + concurrency_time_window_s, + cache_ttl: module.cache_ttl.map(|x| x as i32).ok_or(cache_ttl).ok(), + language, + dedicated_worker, + }, + tag, + ) + }; + Ok(JobPayloadWithTag { payload, tag }) +} + async fn get_transform_context( flow_job: &QueuedJob, - previous_id: String, + previous_id: &str, status: &FlowStatus, ) -> error::Result { let steps_results: HashMap = status @@ -2150,7 +2271,7 @@ async fn get_transform_context( .filter_map(|x| x.job_result().map(|y| (x.id(), y))) .collect(); - Ok(IdContext { flow_job: flow_job.id, steps_results, previous_id }) + Ok(IdContext { flow_job: flow_job.id, steps_results, previous_id: previous_id.to_string() }) } async fn evaluate_with( diff --git a/benchmarks/worker.ts b/benchmarks/worker.ts index 6e85180dbe..239c59c32b 100644 --- a/benchmarks/worker.ts +++ b/benchmarks/worker.ts @@ -43,11 +43,9 @@ const outstanding: string[] = []; let cont = true; let total_spawned = 0; -let start_time: number; +const start_time: number = Date.now(); let complete_timeout = Infinity; -start_time = Date.now(); - self.onmessage = (evt) => { cont = false; complete_timeout = evt.data; @@ -301,11 +299,26 @@ async function getQueueCount() { ).database_length; } -while (outstanding.length > 0 && Date.now() < end_time) { +let last_queue_length = await getQueueCount(); +console.log(`waiting for ${last_queue_length} jobs to complete...`); + +while ( + outstanding.length > 0 && + last_queue_length > 0 && + Date.now() < end_time +) { try { await Deno.stdout.write( - enc("\rwaiting for jobs to complete: " + outstanding.length + "\n") + enc( + "\rwaiting for jobs to complete: outstanding " + + outstanding.length + + " - queue" + + last_queue_length + + "\n" + ) ); + last_queue_length = await getQueueCount(); + const uuid = outstanding.shift()!; let r: Job; @@ -321,7 +334,7 @@ while (outstanding.length > 0 && Date.now() < end_time) { if (r.type == "QueuedJob") { outstanding.push(uuid); await Deno.stdout.write( - enc(`uuid: ${uuid}, queue length: ${await getQueueCount()}\r`) + enc(`uuid: ${uuid}, queue length: ${last_queue_length}\r`) ); } else { r = r as api.CompletedJob; diff --git a/frontend/src/lib/components/FlowStatusViewer.svelte b/frontend/src/lib/components/FlowStatusViewer.svelte index 43300a5206..81e9aff799 100644 --- a/frontend/src/lib/components/FlowStatusViewer.svelte +++ b/frontend/src/lib/components/FlowStatusViewer.svelte @@ -204,6 +204,7 @@ } } } + let showEmbeddeds = -20 {#if job} @@ -212,6 +213,17 @@

Flow result

{/if} {#if isListJob} + {#if (flowJobIds?.flowJobs.length ?? 0) > 20} +

+ For performance reasons, only the last 20 items are shown. +

+ {/if} {#if render}
@@ -277,7 +289,18 @@

Embedded flows: ({flowJobIds?.flowJobs.length} items)

- {#each flowJobIds?.flowJobs ?? [] as loopJobId, j} + {#if (flowJobIds?.flowJobs.length ?? 0) > 20} +

+ For performance reasons, only the last 20 items are shown. +

+ {/if} + {#each (flowJobIds?.flowJobs.length ?? 0) > 20 ? flowJobIds?.flowJobs?.slice(showEmbeddeds) ?? [] : flowJobIds?.flowJobs ?? [] as loopJobId, j} {#if render}