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

feat: dedicated workers for native-throughput performance (EE only) (#2239)

Browse Source 
* dedicated worker

* dedicated worker

* merge

* progress

* progress

* progress

* merge

* update

* sqlx

* fix imports
This commit is contained in:
Ruben Fiszel
2023-09-06 10:32:53 +02:00
committed by GitHub
parent d39711915a
commit c80f155602
45 changed files with 1295 additions and 683 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-062859f1d0e5cfba3115f4241115753b86a4ad239708851c998ff5620ebca5b8.json
Generated
-14
View File
@@ -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"
}
backend/.sqlx/query-0694f4be1a7966126f20afbd9ee094088db8ce98aefdbe96144c42e38e5bd813.json
Generated
-41
View File
@@ -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"
}
backend/.sqlx/query-120fa12a4c2fdecf1da58ffba0c0aff06ea62f94ff357c446a8e2311e083efdf.json
Generated
-14
View File
@@ -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"
}
backend/.sqlx/query-123c0608e229c29187009b7961355ddd99c4ad1f46b876dd86e372b84d806ecd.json
Generated
+74
View File
@@ -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"
}
backend/.sqlx/query-15ef5759a2ccd7b7f9fd3f2ce0d54d01fe0a2c7e9692ac4ce29a86eb509e1a1d.json
Generated
+17
View File
@@ -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"
}
backend/.sqlx/query-52dbe77fad5729e9e121609eeb6a3573e6d0e1d1c16e29b15c7ad810536e249c.json → backend/.sqlx/query-438b5b5d29b05846c2e074cad2404e797527841cd97dba80c271cbefafae65cc.json
Generated
+4 -3
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl) 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"
}
backend/.sqlx/query-5ba4b87528ad49f17d72b53c3db30f5ca4b3b0b0afbd5d9721c8b5d692af601b.json
Generated
-22
View File
@@ -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"
}
backend/.sqlx/query-5cd89ab614d3cac80fb81627267ee85b191263989b0c78b2bfce77e796e96825.json
Generated
+80
View File
@@ -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"
}
backend/.sqlx/query-637ead7701775609c772c9985450273b942202cdf9017ce6b5ba612e66ce3875.json
Generated
-44
View File
@@ -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"
}
backend/.sqlx/query-65835f2e5ad38f7cc6b147dadfef6f580f15bca96d9746c9359e98ca793f8f1f.json
Generated
+80
View File
@@ -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"
}
backend/.sqlx/query-8062bb1ac475c4a2a016d9dfa63c95765356fe7744b817ad0c320a54a33b24f3.json
Generated
-47
View File
@@ -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"
}
backend/.sqlx/query-b053117536c067095e2fb2864ce5af33f84b22c24c92fcb870f37f0501f8ea9a.json
Generated
-22
View File
@@ -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"
}
backend/.sqlx/query-b224cdd1221fc9e7227ef8e8c025eedc09bceb82bca349f3e31c8513ebbf0192.json
Generated
+80
View File
@@ -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"
}
backend/.sqlx/query-bd086c56ad3a58163619122641b93b941ef75880eeeab0756d40f68f87760d5b.json
Generated
-23
View File
@@ -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"
}
backend/.sqlx/query-c07c9276945663d062cf0ff5b3323be681a0e2cb07a457ea9aede2daeff551cc.json
Generated
-22
View File
@@ -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"
}
backend/.sqlx/query-cac375edf290d68d487de4273c3696b824685978760b308d8b578df95cf8db45.json
Generated
-14
View File
@@ -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"
}
backend/.sqlx/query-cf21b61aa5b5e47744235c860e2d9b90d7654a4b5d0ec32085b178399f701bfc.json
Generated
-47
View File
@@ -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"
}
backend/.sqlx/query-d768bbc46f8a9c4289b918c88ca50aa180b29bbe931d948f6e61976f71b7cdb9.json
Generated
-22
View File
@@ -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"
}
backend/.sqlx/query-d918e1d3a1ccc36ca8b4b25aeeec1cb55aadaf1e6609a1aad09b0515c93a667a.json
Generated
-22
View File
@@ -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"
}
backend/.sqlx/query-e7418515b88d14d1fd79078342963e11422eed9a60e0472c68971f4e8f8735d9.json
Generated
-16
View File
@@ -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"
}
backend/.sqlx/query-ebac391c83dae67831502fbe3f1034210afce7902a69f7c42bb68160e0464d71.json
Generated
-47
View File
@@ -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"
}
backend/.sqlx/query-f1e65e52585b27bfb39817e82613063e168f10a53d56f222c7917aeba020fff3.json
Generated
-14
View File
@@ -1,14 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM email_to_igroup WHERE igroup = $1",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text"
]
},
"nullable": []
},
"hash": "f1e65e52585b27bfb39817e82613063e168f10a53d56f222c7917aeba020fff3"
}
backend/.sqlx/query-ff89e9f0941507d5cbfbb27e7c3b1ebad6b2b0a836334a5e20f151711fc11370.json
Generated
-15
View File
@@ -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"
}
backend/migrations/20230905170218_dedicated_worker.down.sql
+1
View File
@@ -0,0 +1 @@
-- Add down migration script here
backend/migrations/20230905170218_dedicated_worker.up.sql
+2
View File
@@ -0,0 +1,2 @@
-- Add up migration script here
ALTER TABLE script ADD COLUMN dedicated_worker boolean;
backend/parsers/windmill-parser-py-imports/tests/fixtures/base.sql
Vendored
+1
View File
@@ -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))
);
backend/windmill-api/openapi.yaml
+4
View File
@@ -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
backend/windmill-api/src/jobs.rs
+27 -5
View File
@@ -315,9 +315,11 @@ pub async fn get_path_tag_limits_cache_for_hash(
Option<i32>,
Option<i32>,
Option<i32>,
ScriptLang,
Option<bool>,
)> {
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,
backend/windmill-api/src/oauth2.rs
+16 -7
View File
@@ -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,
)
backend/windmill-api/src/scripts.rs
+6 -3
View File
@@ -71,6 +71,8 @@ pub struct ScriptWDraft {
pub concurrency_time_window_s: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub cache_ttl: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub dedicated_worker: Option<bool>,
}
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 \
backend/windmill-common/src/flows.rs
+11
View File
@@ -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<S>(value: &HashMap<String, InputTransform>, serializer: S) -> Result<S::Ok, S::Error>
where
S: Serializer,
backend/windmill-common/src/jobs.rs
+24 -4
View File
@@ -185,6 +185,8 @@ pub enum JobPayload {
concurrent_limit: Option<i32>,
concurrency_time_window_s: Option<i32>,
cache_ttl: Option<i32>,
dedicated_worker: Option<bool>,
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<Tag>, Option<i32>, Option<i32>, Option<i32>)> {
) -> error::Result<(
Option<Tag>,
Option<i32>,
Option<i32>,
Option<i32>,
ScriptLang,
Option<bool>,
)> {
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,
))
}
backend/windmill-common/src/lib.rs
+11 -2
View File
@@ -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<i32>,
Option<i32>,
Option<i32>,
ScriptLang,
Option<bool>,
)> {
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<i32>,
Option<i32>,
Option<i32>,
ScriptLang,
Option<bool>,
)> {
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,
))
}
backend/windmill-common/src/scripts.rs
+1
View File
@@ -215,6 +215,7 @@ pub struct NewScript {
pub concurrent_limit: Option<i32>,
pub concurrency_time_window_s: Option<i32>,
pub cache_ttl: Option<i32>,
pub dedicated_worker: Option<bool>,
}
#[derive(Deserialize)]
backend/windmill-queue/src/jobs.rs
+54 -27
View File
@@ -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<String> = 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 <workspace>:<script_path>")
} else {
let workspace = splitted[0];
let script_path = splitted[1];
(workspace.to_string(), script_path.to_string())
}
});
pub static ref ACCEPTED_TAGS: Vec<String> = {
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()) {
backend/windmill-queue/src/schedule.rs
+16 -7
View File
@@ -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,
)
backend/windmill-worker/src/bun_executor.rs
+325 -6
View File
@@ -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 = "<empty>";
pub const BUN_LOCKB_SPLIT: &str = "\n//bun.lockb\n";
pub const EMPTY_FILE: &str = "<empty>";
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<String, String> {
pub fn get_common_bun_proc_envs(base_internal_url: &str) -> HashMap<String, String> {
let mut deno_envs: HashMap<String, String> = 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<String, String>
}
return deno_envs;
}
#[cfg(feature = "enterprise")]
pub async fn start_worker(
requirements_o: Option<String>,
db: &sqlx::Pool<sqlx::Postgres>,
inner_content: &str,
base_internal_url: &str,
job_dir: &str,
worker_name: &str,
envs: HashMap<String, String>,
w_id: &str,
script_path: &str,
token: &str,
job_completed_tx: Sender<JobCompleted>,
mut jobs_rx: Receiver<QueuedJob>,
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<String, String> =
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::<Vec<&str>>();
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::<Vec<_>>(),
)
.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(())
}
backend/windmill-worker/src/common.rs
+14 -9
View File
@@ -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<String, String> = variables
.into_iter()
.map(|rv| (rv.name, rv.value))
.collect();
Ok(build_envs_map(variables))
}
pub fn build_envs_map(context: Vec<ContextualVariable>) -> HashMap<String, String> {
let mut r: HashMap<String, String> =
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<u32>, 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<R: tokio::io::AsyncBufRead + Unpin>(
pub fn lines_to_stream<R: tokio::io::AsyncBufRead + Unpin>(
mut lines: tokio::io::Lines<R>,
) -> impl futures::Stream<Item = io::Result<String>> {
stream::poll_fn(move |cx| {
backend/windmill-worker/src/dedicated_worker.rs
+5
View File
@@ -0,0 +1,5 @@
// use tokio::sync::mpsc;
// pub fn create_dedicated_worker() {
// let (job_completed_tx, mut new_job) = mpsc::channel::<JobCompleted>(100);
// }
backend/windmill-worker/src/lib.rs
+1
View File
@@ -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;
backend/windmill-worker/src/worker.rs
+135 -43
View File
@@ -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<RwLock<()>> = 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<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
i_worker: u64,
_num_workers: u32,
ip: &str,
mut rx: tokio::sync::broadcast::Receiver<()>,
mut killpill_rx: tokio::sync::broadcast::Receiver<()>,
base_internal_url: &str,
rsmq: Option<R>,
_sync_barrier: Arc<RwLock<Option<Barrier>>>,
@@ -583,8 +591,6 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
// }
});
let mut first_run = true;
let mut last_executed_job: Option<Instant> = None;
let mut last_checked_suspended = Instant::now();
@@ -599,6 +605,81 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
IS_READY.store(true, Ordering::Relaxed);
tracing::info!(worker = %worker_name, "listening for jobs");
let (dedicated_worker_tx, dedicated_worker_handle) = if let Some((_workspace, _script_path)) =
DEDICATED_WORKER.clone()
{
#[cfg(not(feature = "enterprise"))]
panic!("Dedicated worker is an enterprise feature");
#[cfg(feature = "enterprise")]
{
let (dedicated_worker_tx, dedicated_worker_rx) =
mpsc::channel::<QueuedJob>(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<String>, Option<ScriptLang>, Option<Vec<String>>)>(
"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<Sender<QueuedJob>>, Option<JoinHandle<()>>)
};
loop {
#[cfg(feature = "benchmark")]
let loop_start = Instant::now();
@@ -689,9 +770,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
// };
// }
let next_job = if first_run {
Ok(Some(QueuedJob::default()))
} else {
let next_job = {
// println!("2: {:?}", instant.elapsed());
let _wait_signal = false;
@@ -710,7 +789,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
tokio::select! {
biased;
_ = rx.recv() => {
_ = 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<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
}
};
if first_run {
first_run = false;
}
if *METRICS_ENABLED {
worker_busy.set(1);
}
@@ -781,7 +856,12 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
last_executed_job = None;
jobs_executed += 1;
if matches!(job.job_kind, JobKind::Noop) {
if let Some(dedicated_worker_tx) = dedicated_worker_tx.clone() {
if let Err(e) = dedicated_worker_tx.send(job.clone()).await {
tracing::info!("failed to send jobs to dedicated workers. Likely dedicated worker has been shut down. This is normal: {e:?}");
}
continue;
} else if matches!(job.job_kind, JobKind::Noop) {
job_completed_tx
.send(JobCompleted {
job,
@@ -795,6 +875,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
.expect("send job completed");
} else {
let token = create_token_for_owner_in_bg(&db, &job).await;
let language = job.language.clone();
let _timer = worker_execution_duration
.get(&language)
@@ -858,10 +939,6 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
client: OnceCell::new(),
};
let is_flow = job.job_kind == JobKind::Flow
|| job.job_kind == JobKind::FlowPreview
|| job.job_kind == JobKind::FlowDependencies;
if let Some(err) = handle_queued_job(
job.clone(),
db,
@@ -899,7 +976,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
.expect("no timer found")
.inc_by(duration);
if !*KEEP_JOB_DIR && !(is_flow && same_worker) {
if !*KEEP_JOB_DIR && !(job.is_flow() && same_worker) {
let _ = tokio::fs::remove_dir_all(job_dir).await;
}
}
@@ -956,7 +1033,14 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
// .expect("write profiling");
// }
drop(dedicated_worker_tx);
if let Some(handle) = dedicated_worker_handle {
handle.await.expect("dedicated worker failed");
}
drop(job_completed_tx);
send_result.await.expect("send result failed");
println!("worker {} exited", i_worker);
}
@@ -1539,6 +1623,37 @@ async fn process_result(
Ok(())
}
fn build_envs(
envs: Option<Vec<String>>,
) -> windmill_common::error::Result<HashMap<String, String>> {
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<serde_json::Value> = match language {
None => {
backend/windmill-worker/src/worker_flow.rs
+234 -113
View File
@@ -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<bool>, Option<i32>), 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<Option<bool>, 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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
&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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
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<BranchAllStatus>,
iterator: Option<windmill_common::flow_status::Iterator>,
simple_input_transforms: Option<HashMap<String, InputTransform>>,
},
}
@@ -1723,7 +1719,7 @@ struct JobPayloadWithTag {
}
enum ContinuePayload {
SingleJob(JobPayloadWithTag),
ForloopJobs { n: usize, modules: Vec<FlowModule> },
ForloopJobs { n: usize, payload: JobPayloadWithTag },
BranchAllJobs(Vec<JobPayloadWithTag>),
}
@@ -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<String>,
@@ -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<String, serde_json::Value> = &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<String>,
content: &String,
language: &windmill_common::scripts::ScriptLang,
lock: &Option<String>,
concurrent_limit: &Option<i32>,
concurrency_time_window_s: &Option<i32>,
module: &FlowModule,
tag: &Option<String>,
) -> 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<windmill_common::scripts::ScriptHash>,
script_path: &String,
db: &sqlx::Pool<sqlx::Postgres>,
flow_job: &QueuedJob,
module: &FlowModule,
) -> Result<JobPayloadWithTag, Error> {
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<IdContext> {
let steps_results: HashMap<String, JobResult> = 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<F>(
benchmarks/worker.ts
+19 -6
View File
@@ -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;
frontend/src/lib/components/FlowStatusViewer.svelte
+27 -2
View File
@@ -204,6 +204,7 @@
}
}
}
let showEmbeddeds = -20
</script>
{#if job}
@@ -212,6 +213,17 @@
<h3 class="text-md leading-6 font-bold text-primay border-b pb-2">Flow result</h3>
{/if}
{#if isListJob}
{#if (flowJobIds?.flowJobs.length ?? 0) > 20}
<p class="text-tertiary italic">
For performance reasons, only the last 20 items are shown. <button
class="text-primary underline"
on:click={() => {
showEmbeddeds -= 20
}}
>Load 20 prior
</button>
</p>
{/if}
{#if render}
<div class="w-full h-full border border-gray-600 bg-surface p-1 overflow-auto">
<DisplayResult workspaceId={job?.workspace_id} {jobId} result={jobResults} />
@@ -277,7 +289,18 @@
<h3 class="text-md leading-6 font-bold text-tertiary border-b mb-4">
Embedded flows: ({flowJobIds?.flowJobs.length} items)
</h3>
{#each flowJobIds?.flowJobs ?? [] as loopJobId, j}
{#if (flowJobIds?.flowJobs.length ?? 0) > 20}
<p class="text-tertiary italic">
For performance reasons, only the last 20 items are shown. <button
class="text-primary underline"
on:click={() => {
showEmbeddeds -= 20
}}
>Load 20 prior
</button>
</p>
{/if}
{#each (flowJobIds?.flowJobs.length ?? 0) > 20 ? flowJobIds?.flowJobs?.slice(showEmbeddeds) ?? [] : flowJobIds?.flowJobs ?? [] as loopJobId, j}
{#if render}
<Button
variant={forloop_selected === loopJobId ? 'contained' : 'border'}
@@ -296,7 +319,9 @@
}}
>
<span class="truncate">
#{j + 1}: {loopJobId}
#{(flowJobIds?.flowJobs.length ?? 0) > 20
? (flowJobIds?.flowJobs.length ?? 0) + showEmbeddeds + j + 1
: j + 1}: {loopJobId}
</span>
<Icon
frontend/src/lib/components/ScriptBuilder.svelte
+26
View File
@@ -168,6 +168,7 @@
kind: script.kind,
tag: script.tag,
envs: script.envs,
dedicated_worker: script.dedicated_worker,
concurrent_limit: script.concurrent_limit,
concurrency_time_window_s: script.concurrency_time_window_s,
cache_ttl: script.cache_ttl
@@ -474,6 +475,31 @@
{/if}
</div>
<h2 class="border-b pb-1 mt-10 mb-4 items-center flex gap-1"
>Dedicated Workers<Tooltip
>In this mode, the script is meant to be run on dedicated workers that run the script at
native speed. Can reach >400rps per dedicated worker. Only available on enterprise edition
and for the Bun language.</Tooltip
>
<Toggle
disabled={!$enterpriseLicense ||
!isCloudHosted() ||
script.language != Script.language.BUN}
size="xs"
checked={Boolean(script.dedicated_worker)}
on:change={() => {
if (script.dedicated_worker) {
script.dedicated_worker = undefined
} else {
script.dedicated_worker = true
}
}}
options={{
right: 'Script is run on dedicated workers'
}}
/></h2
>
{#if !isCloudHosted()}
<h2 class="border-b pb-1 mt-10 mb-4">
Custom env variables