From 7856bf67e96c83975db25f5c5229fdec88a12b2e Mon Sep 17 00:00:00 2001 From: Guillaume Bouvignies Date: Thu, 13 Jul 2023 17:32:08 +0200 Subject: [PATCH] feat: Front end for script concurrency limit (#1863) * Front end for script concurrency limit * Fix frontend * sqlx prepare * fix compile * add flow inline scripts --------- Co-authored-by: Ruben Fiszel Co-authored-by: Ruben Fiszel --- backend/sqlx-data.json | 116 +++++++++--------- backend/windmill-api/openapi.yaml | 17 +++ backend/windmill-api/src/scripts.rs | 15 ++- backend/windmill-common/src/scripts.rs | 6 + backend/windmill-worker/src/worker_flow.rs | 23 +++- .../src/lib/components/ScriptBuilder.svelte | 26 +++- .../details/DetailPageHeader.svelte | 5 +- .../flows/content/FlowModuleComponent.svelte | 35 ++++++ .../content/FlowModuleConcurrencyLimit.svelte | 57 +++++++++ .../flows/content/FlowModuleHeader.svelte | 13 +- .../flows/map/FlowModuleSchemaItem.svelte | 24 +++- .../lib/components/flows/map/MapItem.svelte | 3 +- frontend/src/lib/scripts.ts | 10 +- .../scripts/get/[...hash]/+page.svelte | 10 +- openflow.openapi.yaml | 4 + 15 files changed, 288 insertions(+), 76 deletions(-) create mode 100644 frontend/src/lib/components/flows/content/FlowModuleConcurrencyLimit.svelte diff --git a/backend/sqlx-data.json b/backend/sqlx-data.json index 8f518fc4df..6e12a40e08 100644 --- a/backend/sqlx-data.json +++ b/backend/sqlx-data.json @@ -3914,63 +3914,6 @@ }, "query": "SELECT EXISTS(SELECT 1 FROM variable WHERE path = $1 AND workspace_id = $2)" }, - "88a5f7d43e2775cdbc121e4f39f7d881af75c24740446ca86f26ba8b2e9e5405": { - "describe": { - "columns": [], - "nullable": [], - "parameters": { - "Left": [ - "Varchar", - "Int8", - "Varchar", - "Int8Array", - "Text", - "Text", - "Text", - "Varchar", - "Text", - "Bool", - "Jsonb", - "Text", - { - "Custom": { - "kind": { - "Enum": [ - "python3", - "deno", - "go", - "bash", - "postgresql", - "nativets", - "bun", - "mysql" - ] - }, - "name": "script_lang" - } - }, - { - "Custom": { - "kind": { - "Enum": [ - "script", - "trigger", - "failure", - "command", - "approval" - ] - }, - "name": "script_kind" - } - }, - "Varchar", - "Bool", - "VarcharArray" - ] - } - }, - "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) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17)" - }, "88b7589a6416da8be4b26af3bf30fcfcd6aeae7bc5a37e9a735cabbe2691c570": { "describe": { "columns": [ @@ -4806,6 +4749,65 @@ }, "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" }, + "9ee1423945740c17dedc2fc7ea7e7183380c43d0feeb615c651aa486eef3bd53": { + "describe": { + "columns": [], + "nullable": [], + "parameters": { + "Left": [ + "Varchar", + "Int8", + "Varchar", + "Int8Array", + "Text", + "Text", + "Text", + "Varchar", + "Text", + "Bool", + "Jsonb", + "Text", + { + "Custom": { + "kind": { + "Enum": [ + "python3", + "deno", + "go", + "bash", + "postgresql", + "nativets", + "bun", + "mysql" + ] + }, + "name": "script_lang" + } + }, + { + "Custom": { + "kind": { + "Enum": [ + "script", + "trigger", + "failure", + "command", + "approval" + ] + }, + "name": "script_kind" + } + }, + "Varchar", + "Bool", + "VarcharArray", + "Int4", + "Int4" + ] + } + }, + "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) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19)" + }, "a17260a1f1ee02e786690994d98c84ddf81e2eeb883f895c9cfc47e144d422cb": { "describe": { "columns": [ diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 9ab63ab895..eb42ed2b0b 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -5649,6 +5649,15 @@ components: type: array items: type: string + concurrent_limit: + type: array + items: + type: integer + concurrency_time_window_s: + type: array + items: + type: integer + required: - hash - path @@ -5700,6 +5709,14 @@ components: type: array items: type: string + concurrent_limit: + type: array + items: + type: integer + concurrency_time_window_s: + type: array + items: + type: integer required: - path - summary diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index 8fd40c464d..c80a880de8 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/backend/windmill-api/src/scripts.rs @@ -71,6 +71,10 @@ pub struct ScriptWDraft { pub draft_only: Option, #[serde(skip_serializing_if = "Option::is_none")] pub envs: Option>, + #[serde(skip_serializing_if = "Option::is_none")] + pub concurrent_limit: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub concurrency_time_window_s: Option, } pub fn global_service() -> Router { @@ -367,8 +371,9 @@ async fn create_script( //::text::json is to ensure we use serde_json with preserve order 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) \ - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17)", + content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, \ + draft_only, envs, concurrent_limit, concurrency_time_window_s) \ + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19)", &w_id, &hash.0, ns.path, @@ -385,7 +390,9 @@ async fn create_script( ns.kind.unwrap_or(ScriptKind::Script): ScriptKind, ns.tag, ns.draft_only, - envs + envs, + ns.concurrent_limit, + ns.concurrency_time_window_s ) .execute(&mut tx) .await?; @@ -565,7 +572,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, 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, draft.value as draft FROM script LEFT JOIN draft ON script.path = draft.path AND script.workspace_id = draft.workspace_id AND draft.typ = 'script' WHERE script.path = $1 AND script.workspace_id = $2 \ AND script.created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND \ diff --git a/backend/windmill-common/src/scripts.rs b/backend/windmill-common/src/scripts.rs index 18cfb00cf5..2e3d6bd088 100644 --- a/backend/windmill-common/src/scripts.rs +++ b/backend/windmill-common/src/scripts.rs @@ -149,6 +149,10 @@ pub struct Script { pub draft_only: Option, #[serde(skip_serializing_if = "Option::is_none")] pub envs: Option>, + #[serde(skip_serializing_if = "Option::is_none")] + pub concurrent_limit: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub concurrency_time_window_s: Option, } #[derive(Serialize)] @@ -200,6 +204,8 @@ pub struct NewScript { pub tag: Option, pub draft_only: Option, pub envs: Option>, + pub concurrent_limit: Option, + pub concurrency_time_window_s: Option, } #[derive(Deserialize)] diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 430bfd9385..c185ff341c 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -1364,7 +1364,7 @@ async fn push_next_flow_job failure_module: fm.clone(), same_worker: flow.same_worker, }, - path: Some(format!("{}/loop-{}", flow_job.script_path(), i)), + path: Some(format!("{}/forloop", flow_job.script_path())), }, tag: None, } @@ -1708,9 +1708,15 @@ async fn compute_next_flow_transform( } 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) = script_hash_to_tag_and_limits(&hash, &mut tx, &flow_job.workspace_id).await?; + let (tag, concurrent_limit, concurrency_time_window_s) = + 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 }, + JobPayload::ScriptHash { + hash, + path: script_path.to_owned(), + concurrent_limit, + concurrency_time_window_s, + }, tag, ) }; @@ -1719,7 +1725,16 @@ async fn compute_next_flow_transform( NextStatus::NextStep, )) } - FlowModuleValue::RawScript { path, content, language, lock, tag, concurrent_limit, concurrency_time_window_s, ..} => { + FlowModuleValue::RawScript { + path, + content, + language, + lock, + tag, + concurrent_limit, + concurrency_time_window_s, + .. + } => { let path = path .clone() .or_else(|| Some(format!("{}/step-{}", flow_job.script_path(), status.step))); diff --git a/frontend/src/lib/components/ScriptBuilder.svelte b/frontend/src/lib/components/ScriptBuilder.svelte index b1b3126043..1ec6378de7 100644 --- a/frontend/src/lib/components/ScriptBuilder.svelte +++ b/frontend/src/lib/components/ScriptBuilder.svelte @@ -158,7 +158,9 @@ language: script.language, kind: script.kind, tag: script.tag, - envs: script.envs + envs: script.envs, + concurrent_limit: script.concurrent_limit, + concurrency_time_window_s: script.concurrency_time_window_s } }) history.replaceState(history.state, '', `/scripts/edit/${script.path}`) @@ -198,7 +200,9 @@ kind: script.kind, tag: script.tag, draft_only: true, - envs: script.envs + envs: script.envs, + concurrent_limit: script.concurrent_limit, + concurrency_time_window_s: script.concurrency_time_window_s } }) } @@ -256,7 +260,6 @@ - {#if !$userStore?.operator} @@ -370,6 +373,23 @@ class="text-sm" /> +

Concurrency limits

+
+ + +
+

Worker group tag
-
- {title} +
+
{title}
+
diff --git a/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte b/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte index f6da99808f..b1423df518 100644 --- a/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte +++ b/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte @@ -30,6 +30,8 @@ import InputTransformSchemaForm from '$lib/components/InputTransformSchemaForm.svelte' import { schemaToObject } from '$lib/schema' import FlowModuleMock from './FlowModuleMock.svelte' + import Tooltip from '$lib/components/Tooltip.svelte' + import { SecondsInput } from '$lib/components/common' const { selectedId, previewArgs, flowStateStore, flowStore, saveDraft } = getContext('FlowEditorContext') @@ -136,6 +138,7 @@ on:toggleSleep={() => selectAdvanced('sleep')} on:toggleMock={() => selectAdvanced('mock')} on:toggleRetry={() => selectAdvanced('retries')} + on:toggleConcurrency={() => selectAdvanced('concurrency')} on:toggleCache={() => selectAdvanced('cache')} on:toggleStopAfterIf={() => selectAdvanced('early-stop')} on:fork={async () => { @@ -264,6 +267,9 @@ Retries {#if !$selectedId.includes('failure')} Cache + {#if flowModule.value.type == 'rawscript'} + Concurrency + {/if} Early Stop/Break Suspend/Approval Sleep @@ -292,6 +298,35 @@
+ {:else if advancedSelected === 'concurrency'} + {#if flowModule.value.type == 'rawscript'} +
+

+ Concurrency Limits + Allowed concurrency within a given timeframe +

+
+
+
Max number of executions within the time window
+
+
+
Time window in seconds
+ +
+ {/if} {:else if advancedSelected === 'same_worker'}
diff --git a/frontend/src/lib/components/flows/content/FlowModuleConcurrencyLimit.svelte b/frontend/src/lib/components/flows/content/FlowModuleConcurrencyLimit.svelte new file mode 100644 index 0000000000..21ba9f8ec4 --- /dev/null +++ b/frontend/src/lib/components/flows/content/FlowModuleConcurrencyLimit.svelte @@ -0,0 +1,57 @@ + + +

+ Mock + + If defined and enabled, the step will immediately return the mock value instead of being + executed. + +

+ { + if (isMockEnabled) { + flowModule.mock = { + enabled: false, + return_value: flowModule.mock?.return_value + } + } else { + flowModule.mock = { + enabled: true, + return_value: flowModule.mock?.return_value ?? { example: 'value' } + } + code = JSON.stringify(flowModule.mock?.return_value, null, 2) + } + console.log(isMockEnabled, flowModule.mock) + }} + options={{ + right: 'Enable step mocking' + }} +/> +
+ Mocked Return value + {#if flowModule?.mock?.return_value != undefined} + + {:else} + + {/if} +
diff --git a/frontend/src/lib/components/flows/content/FlowModuleHeader.svelte b/frontend/src/lib/components/flows/content/FlowModuleHeader.svelte index 581ff9235c..3074929474 100644 --- a/frontend/src/lib/components/flows/content/FlowModuleHeader.svelte +++ b/frontend/src/lib/components/flows/content/FlowModuleHeader.svelte @@ -3,7 +3,7 @@ import { WorkerService, type FlowModule } from '$lib/gen' import { faCodeBranch, faPen, faSave } from '@fortawesome/free-solid-svg-icons' import { createEventDispatcher, getContext } from 'svelte' - import { Bed, Database, PhoneIncoming, Repeat, Square, Voicemail } from 'lucide-svelte' + import { Bed, Database, Gauge, PhoneIncoming, Repeat, Square, Voicemail } from 'lucide-svelte' import Popover from '../../Popover.svelte' import type { FlowEditorContext } from '../types' import { sendUserToast } from '$lib/utils' @@ -39,6 +39,17 @@ Retries + dispatch('toggleConcurrency')} + > + + Concurrency Limits + Retries {/if} + {#if concurrency} + +
+ +
+ Concurrency Limits + + {/if} {#if cache}
) { diff --git a/frontend/src/lib/scripts.ts b/frontend/src/lib/scripts.ts index d86d22e673..737943651a 100644 --- a/frontend/src/lib/scripts.ts +++ b/frontend/src/lib/scripts.ts @@ -86,6 +86,8 @@ export async function getScriptByPath(path: string): Promise<{ schema: any description: string tag: string | undefined + concurrent_limit: number[] | undefined + concurrency_time_window_s: number[] | undefined }> { if (path.startsWith('hub/')) { const { content, language, schema } = await ScriptService.getHubScriptByPath({ path }) @@ -95,7 +97,9 @@ export async function getScriptByPath(path: string): Promise<{ language: language as SupportedLanguage, schema, description: '', - tag: undefined + tag: undefined, + concurrent_limit: undefined, + concurrency_time_window_s: undefined, } } else { const script = await ScriptService.getScriptByPath({ @@ -107,7 +111,9 @@ export async function getScriptByPath(path: string): Promise<{ language: script.language, schema: script.schema, description: script.description, - tag: script.tag + tag: script.tag, + concurrent_limit: script.concurrent_limit, + concurrency_time_window_s: script.concurrency_time_window_s, } } } diff --git a/frontend/src/routes/(root)/(logged)/scripts/get/[...hash]/+page.svelte b/frontend/src/routes/(root)/(logged)/scripts/get/[...hash]/+page.svelte index cc93a7fa74..40e9986faa 100644 --- a/frontend/src/routes/(root)/(logged)/scripts/get/[...hash]/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/scripts/get/[...hash]/+page.svelte @@ -376,7 +376,15 @@ {mainButtons} menuItems={getMenuItems(script)} title={defaultIfEmptyString(script.summary, script.path)} - /> + > + {#if script?.concurrent_limit != undefined && script.concurrency_time_window_s != undefined} + + {/if} +
diff --git a/openflow.openapi.yaml b/openflow.openapi.yaml index 53d2e8ffc7..36da95aa68 100644 --- a/openflow.openapi.yaml +++ b/openflow.openapi.yaml @@ -206,6 +206,10 @@ components: - rawscript tag: type: string + concurrent_limit: + type: number + concurrency_time_window_s: + type: number required: - type - content