diff --git a/backend/.sqlx/query-9e103fb8405089e361d0528d34e166b3098cc0f7abecbef3b498cca86e74a285.json b/backend/.sqlx/query-07486bff9344f8c8906b8120ca66c79ab7ac5e0685a1465e8d140686dc1df247.json similarity index 64% rename from backend/.sqlx/query-9e103fb8405089e361d0528d34e166b3098cc0f7abecbef3b498cca86e74a285.json rename to backend/.sqlx/query-07486bff9344f8c8906b8120ca66c79ab7ac5e0685a1465e8d140686dc1df247.json index 340f8ecf02..a18ff1ac32 100644 --- a/backend/.sqlx/query-9e103fb8405089e361d0528d34e166b3098cc0f7abecbef3b498cca86e74a285.json +++ b/backend/.sqlx/query-07486bff9344f8c8906b8120ca66c79ab7ac5e0685a1465e8d140686dc1df247.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "UPDATE flow SET path = $1, summary = $2, description = $3, value = $4, edited_by = $5, edited_at = now(), schema = $6::text::json, dependency_job = NULL, draft_only = NULL WHERE path = $7 AND workspace_id = $8", + "query": "UPDATE flow SET path = $1, summary = $2, description = $3, value = $4, edited_by = $5, edited_at = now(), schema = $6::text::json, dependency_job = NULL, draft_only = NULL, tag = $9 WHERE path = $7 AND workspace_id = $8", "describe": { "columns": [], "parameters": { @@ -12,10 +12,11 @@ "Varchar", "Text", "Text", - "Text" + "Text", + "Varchar" ] }, "nullable": [] }, - "hash": "9e103fb8405089e361d0528d34e166b3098cc0f7abecbef3b498cca86e74a285" + "hash": "07486bff9344f8c8906b8120ca66c79ab7ac5e0685a1465e8d140686dc1df247" } diff --git a/backend/.sqlx/query-238a59ebc80619504e6dc41c3c24f4ce27786997f380e520d27e169023b28d89.json b/backend/.sqlx/query-238a59ebc80619504e6dc41c3c24f4ce27786997f380e520d27e169023b28d89.json new file mode 100644 index 0000000000..52817c785a --- /dev/null +++ b/backend/.sqlx/query-238a59ebc80619504e6dc41c3c24f4ce27786997f380e520d27e169023b28d89.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT tag from flow WHERE path = $1 and workspace_id = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "tag", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + true + ] + }, + "hash": "238a59ebc80619504e6dc41c3c24f4ce27786997f380e520d27e169023b28d89" +} diff --git a/backend/.sqlx/query-3c842974773d6ca934722ba16f73c4d7821a447d8c75f2394bcbe397de3cc01b.json b/backend/.sqlx/query-4757d024f8d8e7b56e5cb648a1bbafe773db2d4086e9b0dbb2bbe229d221135d.json similarity index 66% rename from backend/.sqlx/query-3c842974773d6ca934722ba16f73c4d7821a447d8c75f2394bcbe397de3cc01b.json rename to backend/.sqlx/query-4757d024f8d8e7b56e5cb648a1bbafe773db2d4086e9b0dbb2bbe229d221135d.json index 91ff99a0b3..1d42c01971 100644 --- a/backend/.sqlx/query-3c842974773d6ca934722ba16f73c4d7821a447d8c75f2394bcbe397de3cc01b.json +++ b/backend/.sqlx/query-4757d024f8d8e7b56e5cb648a1bbafe773db2d4086e9b0dbb2bbe229d221135d.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "INSERT INTO flow (workspace_id, path, summary, description, value, edited_by, edited_at, schema, dependency_job, draft_only) VALUES ($1, $2, $3, $4, $5, $6, now(), $7::text::json, NULL, $8)", + "query": "INSERT INTO flow (workspace_id, path, summary, description, value, edited_by, edited_at, schema, dependency_job, draft_only, tag) VALUES ($1, $2, $3, $4, $5, $6, now(), $7::text::json, NULL, $8, $9)", "describe": { "columns": [], "parameters": { @@ -12,10 +12,11 @@ "Jsonb", "Varchar", "Text", - "Bool" + "Bool", + "Varchar" ] }, "nullable": [] }, - "hash": "3c842974773d6ca934722ba16f73c4d7821a447d8c75f2394bcbe397de3cc01b" + "hash": "4757d024f8d8e7b56e5cb648a1bbafe773db2d4086e9b0dbb2bbe229d221135d" } diff --git a/backend/.sqlx/query-6d50a8dc9cfc040b6f37b58053daa0709671e008a10ab4189114ceff66efe603.json b/backend/.sqlx/query-4f6b3b472b4b78c0325cf3755f9ef1806d2e82328ceccbeade8cc2333c6dfe47.json similarity index 77% rename from backend/.sqlx/query-6d50a8dc9cfc040b6f37b58053daa0709671e008a10ab4189114ceff66efe603.json rename to backend/.sqlx/query-4f6b3b472b4b78c0325cf3755f9ef1806d2e82328ceccbeade8cc2333c6dfe47.json index 506d355ae7..e7276c191a 100644 --- a/backend/.sqlx/query-6d50a8dc9cfc040b6f37b58053daa0709671e008a10ab4189114ceff66efe603.json +++ b/backend/.sqlx/query-4f6b3b472b4b78c0325cf3755f9ef1806d2e82328ceccbeade8cc2333c6dfe47.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT worker, worker_instance, EXTRACT(EPOCH FROM (now() - ping_at))::integer as last_ping, started_at, ip, jobs_executed FROM worker_ping ORDER BY ping_at desc LIMIT $1 OFFSET $2", + "query": "SELECT worker, worker_instance, EXTRACT(EPOCH FROM (now() - ping_at))::integer as last_ping, started_at, ip, jobs_executed, custom_tags FROM worker_ping ORDER BY ping_at desc LIMIT $1 OFFSET $2", "describe": { "columns": [ { @@ -32,6 +32,11 @@ "ordinal": 5, "name": "jobs_executed", "type_info": "Int4" + }, + { + "ordinal": 6, + "name": "custom_tags", + "type_info": "TextArray" } ], "parameters": { @@ -46,8 +51,9 @@ null, false, false, - false + false, + true ] }, - "hash": "6d50a8dc9cfc040b6f37b58053daa0709671e008a10ab4189114ceff66efe603" + "hash": "4f6b3b472b4b78c0325cf3755f9ef1806d2e82328ceccbeade8cc2333c6dfe47" } diff --git a/backend/.sqlx/query-07b3c653e9e70cf7c8e05e40682fa3b8562efbb24f22bcc4b7403305f15a6730.json b/backend/.sqlx/query-61e6aac871b482b6e36f866b4ec9148a75e1bd130e7614463487e2ba6957dfdf.json similarity index 56% rename from backend/.sqlx/query-07b3c653e9e70cf7c8e05e40682fa3b8562efbb24f22bcc4b7403305f15a6730.json rename to backend/.sqlx/query-61e6aac871b482b6e36f866b4ec9148a75e1bd130e7614463487e2ba6957dfdf.json index a1eaecc7a8..6152229295 100644 --- a/backend/.sqlx/query-07b3c653e9e70cf7c8e05e40682fa3b8562efbb24f22bcc4b7403305f15a6730.json +++ b/backend/.sqlx/query-61e6aac871b482b6e36f866b4ec9148a75e1bd130e7614463487e2ba6957dfdf.json @@ -1,16 +1,17 @@ { "db_name": "PostgreSQL", - "query": "INSERT INTO worker_ping (worker_instance, worker, ip) VALUES ($1, $2, $3) ON CONFLICT (worker) DO NOTHING", + "query": "INSERT INTO worker_ping (worker_instance, worker, ip, custom_tags) VALUES ($1, $2, $3, $4) ON CONFLICT (worker) DO NOTHING", "describe": { "columns": [], "parameters": { "Left": [ "Varchar", "Varchar", - "Varchar" + "Varchar", + "TextArray" ] }, "nullable": [] }, - "hash": "07b3c653e9e70cf7c8e05e40682fa3b8562efbb24f22bcc4b7403305f15a6730" + "hash": "61e6aac871b482b6e36f866b4ec9148a75e1bd130e7614463487e2ba6957dfdf" } diff --git a/backend/migrations/20230724163017_tags_for_flows.down.sql b/backend/migrations/20230724163017_tags_for_flows.down.sql new file mode 100644 index 0000000000..d2f607c5b8 --- /dev/null +++ b/backend/migrations/20230724163017_tags_for_flows.down.sql @@ -0,0 +1 @@ +-- Add down migration script here diff --git a/backend/migrations/20230724163017_tags_for_flows.up.sql b/backend/migrations/20230724163017_tags_for_flows.up.sql new file mode 100644 index 0000000000..1558eabc7d --- /dev/null +++ b/backend/migrations/20230724163017_tags_for_flows.up.sql @@ -0,0 +1,2 @@ +-- Add up migration script here +ALTER TABLE flow ADD COLUMN tag VARCHAR(50); diff --git a/backend/migrations/20230725060345_add_worker_tags_to_ping.down.sql b/backend/migrations/20230725060345_add_worker_tags_to_ping.down.sql new file mode 100644 index 0000000000..d2f607c5b8 --- /dev/null +++ b/backend/migrations/20230725060345_add_worker_tags_to_ping.down.sql @@ -0,0 +1 @@ +-- Add down migration script here diff --git a/backend/migrations/20230725060345_add_worker_tags_to_ping.up.sql b/backend/migrations/20230725060345_add_worker_tags_to_ping.up.sql new file mode 100644 index 0000000000..364508813f --- /dev/null +++ b/backend/migrations/20230725060345_add_worker_tags_to_ping.up.sql @@ -0,0 +1,2 @@ +-- Add up migration script here +ALTER TABLE worker_ping ADD COLUMN custom_tags TEXT[]; \ No newline at end of file diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index 96fc109f52..3960254f5e 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -2495,6 +2495,7 @@ async fn test_flow_lock_all(db: Pool) { open_flow_w_path: windmill_api_client::types::OpenFlowWPath { open_flow: flow, path: "g/all/flow_lock_all".to_owned(), + tag: None, }, draft_only: None, }, diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index dbfc56139e..5d411335fe 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -3034,7 +3034,6 @@ paths: properties: draft_only: type: boolean - responses: "201": description: flow created @@ -6718,6 +6717,10 @@ components: type: string jobs_executed: type: integer + custom_tags: + type: array + items: + type: string required: - worker - worker_instance @@ -6847,6 +6850,8 @@ components: type: boolean draft_only: type: boolean + tag: + type: string required: - path - edited_by @@ -6861,6 +6866,8 @@ components: properties: path: type: string + tag: + type: string required: - path @@ -6873,6 +6880,8 @@ components: type: string args: $ref: "#/components/schemas/ScriptArgs" + tag: + type: string required: - value diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index 8e825d2e9e..504ab3ab33 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -203,7 +203,7 @@ async fn create_flow( sqlx::query!( "INSERT INTO flow (workspace_id, path, summary, description, value, edited_by, edited_at, \ - schema, dependency_job, draft_only) VALUES ($1, $2, $3, $4, $5, $6, now(), $7::text::json, NULL, $8)", + schema, dependency_job, draft_only, tag) VALUES ($1, $2, $3, $4, $5, $6, now(), $7::text::json, NULL, $8, $9)", w_id, nf.path, nf.summary, @@ -211,7 +211,8 @@ async fn create_flow( nf.value, &authed.username, nf.schema.and_then(|x| serde_json::to_string(&x.0).ok()), - nf.draft_only + nf.draft_only, + nf.tag ) .execute(&mut tx) .await?; @@ -329,7 +330,7 @@ async fn update_flow( let old_dep_job = not_found_if_none(old_dep_job, "Flow", flow_path)?; sqlx::query!( "UPDATE flow SET path = $1, summary = $2, description = $3, value = $4, edited_by = $5, \ - edited_at = now(), schema = $6::text::json, dependency_job = NULL, draft_only = NULL WHERE path = $7 AND workspace_id = $8", + edited_at = now(), schema = $6::text::json, dependency_job = NULL, draft_only = NULL, tag = $9 WHERE path = $7 AND workspace_id = $8", nf.path, nf.summary, nf.description, @@ -338,6 +339,7 @@ async fn update_flow( schema.and_then(|x| serde_json::to_string(&x).ok()), flow_path, w_id, + nf.tag ) .execute(&mut tx) .await?; diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index e8fd45ba26..bba0cb2ffc 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -1272,6 +1272,7 @@ struct PreviewFlow { value: FlowValue, path: Option, args: Option>, + tag: Option, } pub struct JsonOrForm( @@ -1415,6 +1416,14 @@ pub async fn run_flow_by_path( check_scopes(&authed, || format!("run:flow/{flow_path}"))?; let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.begin(&authed).await?).into(); + let tag = sqlx::query_scalar!( + "SELECT tag from flow WHERE path = $1 and workspace_id = $2", + flow_path, + w_id + ) + .fetch_optional(&mut tx) + .await? + .flatten(); let scheduled_for = run_query.get_scheduled_for(tx.transaction_mut()).await?; let args = run_query.add_include_headers(headers, args.unwrap_or_default()); let args = add_raw_string(raw_string, args); @@ -1435,7 +1444,7 @@ pub async fn run_flow_by_path( false, None, !run_query.invisible_to_owner.unwrap_or(false), - None, + tag, ) .await?; tx.commit().await?; @@ -2105,7 +2114,7 @@ async fn run_preview_flow_job( false, None, true, - None, + raw_flow.tag, ) .await?; tx.commit().await?; diff --git a/backend/windmill-api/src/workers.rs b/backend/windmill-api/src/workers.rs index 532569919a..d2b0261dd4 100644 --- a/backend/windmill-api/src/workers.rs +++ b/backend/windmill-api/src/workers.rs @@ -20,11 +20,10 @@ use windmill_common::{ utils::{paginate, Pagination}, }; -#[cfg(feature = "benchmark")] -use windmill_queue::IDLE_WORKERS; #[cfg(feature = "benchmark")] use std::sync::atomic::Ordering; - +#[cfg(feature = "benchmark")] +use windmill_queue::IDLE_WORKERS; #[cfg(not(feature = "benchmark"))] pub fn global_service() -> Router { @@ -56,6 +55,7 @@ struct WorkerPing { started_at: chrono::DateTime, ip: String, jobs_executed: i32, + custom_tags: Option>, } #[derive(Serialize, Deserialize)] @@ -74,7 +74,7 @@ async fn list_worker_pings( let rows = sqlx::query_as!( WorkerPing, - "SELECT worker, worker_instance, EXTRACT(EPOCH FROM (now() - ping_at))::integer as last_ping, started_at, ip, jobs_executed FROM worker_ping ORDER BY ping_at desc LIMIT $1 OFFSET $2", + "SELECT worker, worker_instance, EXTRACT(EPOCH FROM (now() - ping_at))::integer as last_ping, started_at, ip, jobs_executed, custom_tags FROM worker_ping ORDER BY ping_at desc LIMIT $1 OFFSET $2", per_page as i64, offset as i64 ) @@ -85,9 +85,7 @@ async fn list_worker_pings( } #[cfg(feature = "benchmark")] -async fn toggle( - Query(query): Query, -) -> JsonResult { +async fn toggle(Query(query): Query) -> JsonResult { IDLE_WORKERS.store(query.disable, Ordering::Relaxed); Ok(Json(IDLE_WORKERS.load(Ordering::Relaxed))) } diff --git a/backend/windmill-common/src/flows.rs b/backend/windmill-common/src/flows.rs index 9c456d9637..13563bd63e 100644 --- a/backend/windmill-common/src/flows.rs +++ b/backend/windmill-common/src/flows.rs @@ -35,6 +35,7 @@ pub struct Flow { pub extra_perms: serde_json::Value, #[serde(skip_serializing_if = "Option::is_none")] pub draft_only: Option, + pub tag: Option, } #[derive(Serialize)] @@ -63,6 +64,7 @@ pub struct NewFlow { pub value: serde_json::Value, pub schema: Option, pub draft_only: Option, + pub tag: Option, } #[derive(Deserialize, Serialize, Debug, Clone, Default)] diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 9509fa46e0..144c44e00f 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -64,24 +64,27 @@ lazy_static::lazy_static! { .unwrap(); pub static ref CLOUD_HOSTED: bool = std::env::var("CLOUD_HOSTED").is_ok(); + pub static ref DEFAULT_TAGS : Vec = vec![ + "deno".to_string(), + "python3".to_string(), + "go".to_string(), + "bash".to_string(), + "nativets".to_string(), + "mysql".to_string(), + "graphql".to_string(), + "bun".to_string(), + "postgresql".to_string(), + "dependency".to_string(), + "flow".to_string(), + "hub".to_string(), + "other".to_string()]; + pub static ref ACCEPTED_TAGS: Vec = std::env::var("WORKER_TAGS") .ok() .map(|x| x.split(',').map(|x| x.to_string()).collect()) - .unwrap_or_else(|| vec![ - "deno".to_string(), - "python3".to_string(), - "go".to_string(), - "bash".to_string(), - "nativets".to_string(), - "mysql".to_string(), - "graphql".to_string(), - "bun".to_string(), - "postgresql".to_string(), - "dependency".to_string(), - "flow".to_string(), - "hub".to_string(), - "other".to_string()]); + .unwrap_or_else(|| DEFAULT_TAGS.clone()) ; + pub static ref IS_WORKER_TAGS_DEFINED: bool = std::env::var("WORKER_TAGS").ok().is_some(); pub static ref PULL_QUERY: String = format!( "UPDATE queue @@ -1306,22 +1309,27 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( let tag = if job_kind == JobKind::Dependencies || job_kind == JobKind::FlowDependencies { "dependency".to_string() - } else if job_kind == JobKind::Flow || job_kind == JobKind::FlowPreview { - "flow".to_string() - } else if job_kind == JobKind::Identity { - // identity is a light script, deno is too - "deno".to_string() } else if job_kind == JobKind::Script_Hub { "hub".to_string() } else { if tag == Some("".to_string()) { tag = None; } + let default = || { + if job_kind == JobKind::Flow || job_kind == JobKind::FlowPreview { + "flow" + } else if job_kind == JobKind::Identity { + // identity is a light script, nativets is too + "nativets" + } else { + "deno" + } + }; tag.unwrap_or_else(|| { language .as_ref() .map(|x| x.as_str()) - .unwrap_or_else(|| "deno") + .unwrap_or_else(default) .to_string() }) }; diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 8c967bdd74..089ea5b9ed 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -31,7 +31,7 @@ use windmill_common::{ utils::{rd_string, StripPath}, variables, BASE_URL, users::SUPERADMIN_SECRET_EMAIL, METRICS_ENABLED, jobs::{JobKind, QueuedJob, Metrics}, IS_READY, }; -use windmill_queue::{canceled_job_to_result, get_queued_job, pull, CLOUD_HOSTED, HTTP_CLIENT}; +use windmill_queue::{canceled_job_to_result, get_queued_job, pull, CLOUD_HOSTED, HTTP_CLIENT, ACCEPTED_TAGS, IS_WORKER_TAGS_DEFINED}; use serde_json::{json, Value}; @@ -883,11 +883,13 @@ async fn insert_initial_ping( ip: &str, db: &Pool, ) { + let tags = ACCEPTED_TAGS.clone(); sqlx::query!( - "INSERT INTO worker_ping (worker_instance, worker, ip) VALUES ($1, $2, $3) ON CONFLICT (worker) DO NOTHING", + "INSERT INTO worker_ping (worker_instance, worker, ip, custom_tags) VALUES ($1, $2, $3, $4) ON CONFLICT (worker) DO NOTHING", worker_instance, worker_name, - ip + ip, + if *IS_WORKER_TAGS_DEFINED { Some(tags.as_slice()) } else { None } ) .execute(db) .await @@ -1033,7 +1035,7 @@ async fn handle_queued_job( logs.push_str("\n"); } - logs.push_str(&format!("job {} on worker {}\n", &job.id, &worker_name)); + logs.push_str(&format!("job {} on worker {} (tag: {})\n", &job.id, &worker_name, &job.tag)); set_logs(&logs, &job.id, db).await; diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index e39a48d527..72a379c798 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -1427,7 +1427,11 @@ async fn push_next_flow_job continue_on_same_worker, err, flow_job.visible_to_owner, - payload_tag.tag, + if flow_job.tag == "flow" { + payload_tag.tag + } else { + Some(flow_job.tag.clone()) + }, ) .await?; tx = inner_tx; diff --git a/frontend/src/lib/components/FlowBuilder.svelte b/frontend/src/lib/components/FlowBuilder.svelte index 6cca16ff6a..54f61c3b24 100644 --- a/frontend/src/lib/components/FlowBuilder.svelte +++ b/frontend/src/lib/components/FlowBuilder.svelte @@ -84,6 +84,7 @@ description: flow.description ?? '', value: flow.value, schema: flow.schema, + tag: flow.tag, draft_only: true } }) @@ -146,7 +147,8 @@ summary: flow.summary, description: flow.description ?? '', value: flow.value, - schema: flow.schema + schema: flow.schema, + tag: flow.tag } }) const scheduleExists = await ScheduleService.existsSchedule({ diff --git a/frontend/src/lib/components/ModulePreview.svelte b/frontend/src/lib/components/ModulePreview.svelte index 37425f6f2b..5e2a398f6f 100644 --- a/frontend/src/lib/components/ModulePreview.svelte +++ b/frontend/src/lib/components/ModulePreview.svelte @@ -52,12 +52,24 @@ const val = mod.value // let jobId: string | undefined = undefined if (val.type == 'rawscript') { - await testJobLoader?.runPreview(val.path, val.content, val.language, args, val.tag) + await testJobLoader?.runPreview( + val.path, + val.content, + val.language, + args, + $flowStore?.tag ?? val.tag + ) } else if (val.type == 'script') { const script = val.hash ? await ScriptService.getScriptByHash({ workspace: $workspaceStore!, hash: val.hash }) : await getScriptByPath(val.path) - await testJobLoader?.runPreview(val.path, script.content, script.language, args, script.tag) + await testJobLoader?.runPreview( + val.path, + script.content, + script.language, + args, + $flowStore?.tag ?? script.tag + ) } else { throw Error('not testable module type') } diff --git a/frontend/src/lib/components/ScriptBuilder.svelte b/frontend/src/lib/components/ScriptBuilder.svelte index af6765c054..18c211c39e 100644 --- a/frontend/src/lib/components/ScriptBuilder.svelte +++ b/frontend/src/lib/components/ScriptBuilder.svelte @@ -16,7 +16,7 @@ import type { SupportedLanguage } from '$lib/common' import Tooltip from './Tooltip.svelte' import DrawerContent from './common/drawer/DrawerContent.svelte' - import { Pen, X } from 'lucide-svelte' + import { Loader2, Pen, X } from 'lucide-svelte' import autosize from 'svelte-autosize' import type Editor from './Editor.svelte' import { SCRIPT_SHOW_BASH, SCRIPT_SHOW_GO, SCRIPT_CUSTOMISE_SHOW_KIND } from '$lib/consts' @@ -412,9 +412,9 @@ instance, you could setup an "highmem", or "gpu" worker group. -
- {#if $workerTags} - {#if $workerTags?.length > 0} + {#if $workerTags} + {#if $workerTags?.length > 0} +
- {:else} -
- No custom worker group defined on this instance -
- {/if} +
+ {:else} +
+ No custom worker group defined on this instance. See documentation +
{/if} -
+ {:else} + + {/if} {#if !isCloudHosted()}

Custom env variables diff --git a/frontend/src/lib/components/flows/content/FlowModuleHeader.svelte b/frontend/src/lib/components/flows/content/FlowModuleHeader.svelte index 4e61554de6..f1172f160d 100644 --- a/frontend/src/lib/components/flows/content/FlowModuleHeader.svelte +++ b/frontend/src/lib/components/flows/content/FlowModuleHeader.svelte @@ -11,7 +11,8 @@ import { getLatestHashForScript } from '$lib/scripts' export let module: FlowModule - const { scriptEditorDrawer } = getContext('FlowEditorContext') + const { scriptEditorDrawer, flowStore, selectedId } = + getContext('FlowEditorContext') const dispatch = createEventDispatcher() @@ -143,26 +144,36 @@ {#if $workerTags} {#if $workerTags?.length > 0}
- { + if (module.value.type === 'rawscript') { + if (module.value.tag == '') { + module.value.tag = undefined + } } - } - }} - > - {#if module.value.tag} - - {:else} - - {/if} - {#each $workerTags ?? [] as tag (tag)} - - {/each} - + }} + > + {#if module.value.tag} + + {:else} + + {/if} + {#each $workerTags ?? [] as tag (tag)} + + {/each} + + {:else} + + {/if}
{/if} {/if} diff --git a/frontend/src/lib/components/flows/content/FlowSettings.svelte b/frontend/src/lib/components/flows/content/FlowSettings.svelte index 9ebf4b2883..f27b396ccc 100644 --- a/frontend/src/lib/components/flows/content/FlowSettings.svelte +++ b/frontend/src/lib/components/flows/content/FlowSettings.svelte @@ -14,17 +14,30 @@ import type { FlowEditorContext } from '../types' import autosize from 'svelte-autosize' import Slider from '$lib/components/Slider.svelte' - import { workspaceStore } from '$lib/stores' + import { workerTags, workspaceStore } from '$lib/stores' import { copyToClipboard } from '$lib/utils' import { Icon } from 'svelte-awesome' import { faClipboard } from '@fortawesome/free-solid-svg-icons' import Tooltip from '$lib/components/Tooltip.svelte' + import { WorkerService } from '$lib/gen' + import { Loader2 } from 'lucide-svelte' const { selectedId, flowStore, initialPath } = getContext('FlowEditorContext') + async function loadWorkerGroups() { + if (!$workerTags) { + $workerTags = await WorkerService.getCustomTags() + } + } + let hostname = BROWSER ? window.location.protocol + '//' + window.location.host : 'SSR' $: url = `${hostname}/api/w/${$workspaceStore}/jobs/run/f/${$flowStore?.path}` $: syncedUrl = `${hostname}/api/w/${$workspaceStore}/jobs/run_wait_result/f/${$flowStore?.path}` + + $: if ($selectedId == 'settings-worker-group') { + $workerTags = undefined + loadWorkerGroups() + }
@@ -34,6 +47,7 @@ Metadata Schedule Shared Directory + Worker Group @@ -175,9 +189,11 @@ Steps will share a folder at `./shared` in which they can store heavier data and pass them to the next step.

Beware that the `./shared` folder is not preserved - across suspends and sleeps. + across suspends and sleeps.

+ Furthermore, steps' worker groups is not respected and only the flow's worker group will + be respected. - Shared Directory + Shared Directory + + + When a worker group is defined at the flow level, any steps inside the flow will run + on that worker group, regardless of the steps' worker group. If no worker group is + defined, the flow controls will be executed by the default worker group 'flow' and the + steps will be executed in their respective worker group. + + Worker Group + {#if $workerTags} + {#if $workerTags?.length > 0} +
+ +
+ {:else} +
+ No custom worker group defined on this instance. See documentation +
+ {/if} + {:else} + + {/if} +
diff --git a/frontend/src/lib/components/flows/utils.ts b/frontend/src/lib/components/flows/utils.ts index f2fbf90bb4..e7e2bbc5a7 100644 --- a/frontend/src/lib/components/flows/utils.ts +++ b/frontend/src/lib/components/flows/utils.ts @@ -188,7 +188,8 @@ export async function runFlowPreview(args: Record, flow: Flow) { requestBody: { args, value: newFlow.value, - path: newFlow.path + path: newFlow.path, + tag: newFlow.tag } }) } diff --git a/frontend/src/lib/utils.ts b/frontend/src/lib/utils.ts index b3ffa45b62..ad4307a261 100644 --- a/frontend/src/lib/utils.ts +++ b/frontend/src/lib/utils.ts @@ -260,8 +260,9 @@ export function setQueryWithoutLoad( } export function groupBy( - scripts: T[], + items: T[], toGroup: (t: T) => string, + toSort: (t: T) => string, dflts: string[] = [] ): [string, T[]][] { let r: Record = {} @@ -269,10 +270,11 @@ export function groupBy( r[dflt] = [] } - scripts.forEach((sc) => { + items.forEach((sc) => { let section = toGroup(sc) if (section in r) { r[section].push(sc) + r[section].sort((a, b) => toSort(a).localeCompare(toSort(b))) } else { r[section] = [sc] } diff --git a/frontend/src/routes/(root)/(logged)/workers/+page.svelte b/frontend/src/routes/(root)/(logged)/workers/+page.svelte index edd22d1196..a5a86422df 100644 --- a/frontend/src/routes/(root)/(logged)/workers/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/workers/+page.svelte @@ -4,6 +4,7 @@ import Badge from '$lib/components/common/badge/Badge.svelte' import PageHeader from '$lib/components/PageHeader.svelte' import TableCustom from '$lib/components/TableCustom.svelte' + import Tooltip from '$lib/components/Tooltip.svelte' import { WorkerService, type WorkerPing } from '$lib/gen' import { sendUserToast } from '$lib/toast' import { displayDate, groupBy } from '$lib/utils' @@ -15,7 +16,11 @@ let intervalId: NodeJS.Timer | undefined $: filteredWorkers = (workers ?? []).filter((x) => (x.last_ping ?? 0) < 300) - $: groupedWorkers = groupBy(filteredWorkers, (wp: WorkerPing) => wp.worker_instance) + $: groupedWorkers = groupBy( + filteredWorkers, + (wp: WorkerPing) => wp.worker_instance, + (wp: WorkerPing) => wp.worker + ) let timeSinceLastPing = 0 @@ -68,6 +73,12 @@ Worker + Custom Tags If defined, the workers only pull jobs with the same corresponding tag Last ping Worker start Nb of jobs executed @@ -75,9 +86,10 @@ {#if workers} - {#each workers as { worker, last_ping, started_at, jobs_executed }} + {#each workers as { worker, custom_tags, last_ping, started_at, jobs_executed }} {worker} + {custom_tags?.join(', ') ?? ''} {last_ping != undefined ? last_ping + timeSinceLastPing : -1}s ago