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

feat: worker group for flows

Browse Source
This commit is contained in:
Ruben Fiszel
2023-07-25 12:41:01 +02:00
parent b39f486c91
commit af7c92ac16
26 changed files with 267 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-9e103fb8405089e361d0528d34e166b3098cc0f7abecbef3b498cca86e74a285.json → backend/.sqlx/query-07486bff9344f8c8906b8120ca66c79ab7ac5e0685a1465e8d140686dc1df247.json
Generated
+4 -3
View File
@@ -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"
}
backend/.sqlx/query-238a59ebc80619504e6dc41c3c24f4ce27786997f380e520d27e169023b28d89.json
Generated
+23
View File
@@ -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"
}
backend/.sqlx/query-3c842974773d6ca934722ba16f73c4d7821a447d8c75f2394bcbe397de3cc01b.json → backend/.sqlx/query-4757d024f8d8e7b56e5cb648a1bbafe773db2d4086e9b0dbb2bbe229d221135d.json
Generated
+4 -3
View File
@@ -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"
}
backend/.sqlx/query-6d50a8dc9cfc040b6f37b58053daa0709671e008a10ab4189114ceff66efe603.json → backend/.sqlx/query-4f6b3b472b4b78c0325cf3755f9ef1806d2e82328ceccbeade8cc2333c6dfe47.json
Generated
+9 -3
View File
@@ -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"
}
backend/.sqlx/query-07b3c653e9e70cf7c8e05e40682fa3b8562efbb24f22bcc4b7403305f15a6730.json → backend/.sqlx/query-61e6aac871b482b6e36f866b4ec9148a75e1bd130e7614463487e2ba6957dfdf.json
Generated
+4 -3
View File
@@ -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"
}
backend/migrations/20230724163017_tags_for_flows.down.sql
+1
View File
@@ -0,0 +1 @@
-- Add down migration script here
backend/migrations/20230724163017_tags_for_flows.up.sql
+2
View File
@@ -0,0 +1,2 @@
-- Add up migration script here
ALTER TABLE flow ADD COLUMN tag VARCHAR(50);
backend/migrations/20230725060345_add_worker_tags_to_ping.down.sql
+1
View File
@@ -0,0 +1 @@
-- Add down migration script here
backend/migrations/20230725060345_add_worker_tags_to_ping.up.sql
+2
View File
@@ -0,0 +1,2 @@
-- Add up migration script here
ALTER TABLE worker_ping ADD COLUMN custom_tags TEXT[];
backend/tests/worker.rs
+1
View File
@@ -2495,6 +2495,7 @@ async fn test_flow_lock_all(db: Pool<Postgres>) {
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,
},
backend/windmill-api/openapi.yaml
+10 -1
View File
@@ -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
backend/windmill-api/src/flows.rs
+5 -3
View File
@@ -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?;
backend/windmill-api/src/jobs.rs
+11 -2
View File
@@ -1272,6 +1272,7 @@ struct PreviewFlow {
value: FlowValue,
path: Option<String>,
args: Option<serde_json::Map<String, serde_json::Value>>,
tag: Option<String>,
}
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?;
backend/windmill-api/src/workers.rs
+5 -7
View File
@@ -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<chrono::Utc>,
ip: String,
jobs_executed: i32,
custom_tags: Option<Vec<String>>,
}
#[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<EnableWorkerQuery>,
) -> JsonResult<bool> {
async fn toggle(Query(query): Query<EnableWorkerQuery>) -> JsonResult<bool> {
IDLE_WORKERS.store(query.disable, Ordering::Relaxed);
Ok(Json(IDLE_WORKERS.load(Ordering::Relaxed)))
}
backend/windmill-common/src/flows.rs
+2
View File
@@ -35,6 +35,7 @@ pub struct Flow {
pub extra_perms: serde_json::Value,
#[serde(skip_serializing_if = "Option::is_none")]
pub draft_only: Option<bool>,
pub tag: Option<String>,
}
#[derive(Serialize)]
@@ -63,6 +64,7 @@ pub struct NewFlow {
pub value: serde_json::Value,
pub schema: Option<Schema>,
pub draft_only: Option<bool>,
pub tag: Option<String>,
}
#[derive(Deserialize, Serialize, Debug, Clone, Default)]
backend/windmill-queue/src/jobs.rs
+28 -20
View File
@@ -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<String> = 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<String> = 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()
})
};
backend/windmill-worker/src/worker.rs
+6 -4
View File
@@ -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<Postgres>,
) {
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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
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;
backend/windmill-worker/src/worker_flow.rs
+5 -1
View File
@@ -1427,7 +1427,11 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
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;
frontend/src/lib/components/FlowBuilder.svelte
+3 -1
View File
@@ -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({
frontend/src/lib/components/ModulePreview.svelte
+14 -2
View File
@@ -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')
}
frontend/src/lib/components/ScriptBuilder.svelte
+15 -10
View File
@@ -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.</Tooltip
></h2
>
<div class="max-w-sm">
{#if $workerTags}
{#if $workerTags?.length > 0}
{#if $workerTags}
{#if $workerTags?.length > 0}
<div class="max-w-sm">
<select
bind:value={script.tag}
on:change={(e) => {
@@ -432,13 +432,18 @@
<option value={tag}>{tag}</option>
{/each}
</select>
{:else}
<div class="text-sm text-secondary italic mb-2">
No custom worker group defined on this instance
</div>
{/if}
</div>
{:else}
<div class="text-sm text-gray-600 italic mb-2">
No custom worker group defined on this instance. See <a
href="https://www.windmill.dev/docs/core_concepts/worker_groups"
target="_blank">documentation</a
>
</div>
{/if}
</div>
{:else}
<Loader2 class="animate-spin" />
{/if}
{#if !isCloudHosted()}
<h2 class="border-b pb-1 mt-10 mb-4">
Custom env variables
frontend/src/lib/components/flows/content/FlowModuleHeader.svelte
+31 -20
View File
@@ -11,7 +11,8 @@
import { getLatestHashForScript } from '$lib/scripts'
export let module: FlowModule
const { scriptEditorDrawer } = getContext<FlowEditorContext>('FlowEditorContext')
const { scriptEditorDrawer, flowStore, selectedId } =
getContext<FlowEditorContext>('FlowEditorContext')
const dispatch = createEventDispatcher()
@@ -143,26 +144,36 @@
{#if $workerTags}
{#if $workerTags?.length > 0}
<div class="w-40">
<select
placeholder="Worker group"
bind:value={module.value.tag}
on:change={(e) => {
if (module.value.type === 'rawscript') {
if (module.value.tag == '') {
module.value.tag = undefined
{#if $flowStore.tag == undefined}
<select
placeholder="Worker group"
bind:value={module.value.tag}
on:change={(e) => {
if (module.value.type === 'rawscript') {
if (module.value.tag == '') {
module.value.tag = undefined
}
}
}
}}
>
{#if module.value.tag}
<option value="">reset to default</option>
{:else}
<option value="" disabled selected>Worker Group</option>
{/if}
{#each $workerTags ?? [] as tag (tag)}
<option value={tag}>{tag}</option>
{/each}
</select>
}}
>
{#if module.value.tag}
<option value="">reset to default</option>
{:else}
<option value="" disabled selected>Worker Group</option>
{/if}
{#each $workerTags ?? [] as tag (tag)}
<option value={tag}>{tag}</option>
{/each}
</select>
{:else}
<button
title="Worker Group is defined at the flow level"
class="w-full text-left items-center font-normal p-1 border text-xs rounded"
on:click={() => ($selectedId = 'settings-worker-group')}
>
Flow's WG: {$flowStore.tag}
</button>
{/if}
</div>
{/if}
{/if}
frontend/src/lib/components/flows/content/FlowSettings.svelte
+61 -3
View File
@@ -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>('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()
}
</script>
<div class="h-full overflow-hidden">
@@ -34,6 +47,7 @@
<Tab value="settings-metadata">Metadata</Tab>
<Tab value="settings-schedule">Schedule</Tab>
<Tab value="settings-same-worker">Shared Directory</Tab>
<Tab value="settings-worker-group">Worker Group</Tab>
<svelte:fragment slot="content">
<TabContent value="settings-metadata" class="p-4 h-full">
@@ -175,9 +189,11 @@
<Alert type="info" title="Shared Directory">
Steps will share a folder at `./shared` in which they can store heavier data and pass
them to the next step. <br /><br />Beware that the `./shared` folder is not preserved
across suspends and sleeps.
across suspends and sleeps. <br /><br />
Furthermore, steps' worker groups is not respected and only the flow's worker group will
be respected.
</Alert>
<span class="my-2 text-sm font-bold">Shared Directory</span>
<span class="my-4 text-lg font-bold">Shared Directory</span>
<Toggle
bind:checked={$flowStore.value.same_worker}
options={{
@@ -185,6 +201,48 @@
}}
/>
</TabContent>
<TabContent value="settings-worker-group" class="p-4 flex flex-col">
<Alert type="info" title="Worker Group">
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.
</Alert>
<span class="my-4 text-lg font-bold">Worker Group</span>
{#if $workerTags}
{#if $workerTags?.length > 0}
<div class="w-40">
<select
placeholder="Worker group"
bind:value={$flowStore.tag}
on:change={(e) => {
if ($flowStore.tag == '') {
$flowStore.tag = undefined
}
}}
>
{#if $flowStore.tag}
<option value="">reset to default</option>
{:else}
<option value="" disabled selected>Worker Group</option>
{/if}
{#each $workerTags ?? [] as tag (tag)}
<option value={tag}>{tag}</option>
{/each}
</select>
</div>
{:else}
<div class="text-sm text-gray-600 italic mb-2">
No custom worker group defined on this instance. See <a
href="https://www.windmill.dev/docs/core_concepts/worker_groups"
target="_blank">documentation</a
>
</div>
{/if}
{:else}
<Loader2 class="animate-spin" />
{/if}
</TabContent>
</svelte:fragment>
</Tabs>
</div>
frontend/src/lib/components/flows/utils.ts
+2 -1
View File
@@ -188,7 +188,8 @@ export async function runFlowPreview(args: Record<string, any>, flow: Flow) {
requestBody: {
args,
value: newFlow.value,
path: newFlow.path
path: newFlow.path,
tag: newFlow.tag
}
})
}
frontend/src/lib/utils.ts
+4 -2
View File
@@ -260,8 +260,9 @@ export function setQueryWithoutLoad(
}
export function groupBy<T>(
scripts: T[],
items: T[],
toGroup: (t: T) => string,
toSort: (t: T) => string,
dflts: string[] = []
): [string, T[]][] {
let r: Record<string, T[]> = {}
@@ -269,10 +270,11 @@ export function groupBy<T>(
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]
}
frontend/src/routes/(root)/(logged)/workers/+page.svelte
+14 -2
View File
@@ -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 @@
<TableCustom>
<tr slot="header-row">
<th>Worker</th>
<th
>Custom Tags <Tooltip
documentationLink="https://www.windmill.dev/docs/core_concepts/worker_groups#assign-custom-worker-groups"
>If defined, the workers only pull jobs with the same corresponding tag</Tooltip
></th
>
<th>Last ping</th>
<th>Worker start</th>
<th>Nb of jobs executed</th>
@@ -75,9 +86,10 @@
</tr>
<tbody slot="body">
{#if workers}
{#each workers as { worker, last_ping, started_at, jobs_executed }}
{#each workers as { worker, custom_tags, last_ping, started_at, jobs_executed }}
<tr>
<td class="py-1">{worker}</td>
<td class="py-1">{custom_tags?.join(', ') ?? ''}</td>
<td class="py-1"
>{last_ping != undefined ? last_ping + timeSinceLastPing : -1}s ago</td
>