From 079fbd55eef753074d89834d4870032dcc26be47 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Fri, 23 Dec 2022 19:04:31 +0700 Subject: [PATCH] feat(backend): resume from owner directly in flow status viewer (#1042) * foo * progress --- backend/Cargo.lock | 22 +-- backend/windmill-api/openapi.yaml | 40 +++++ backend/windmill-api/src/jobs.rs | 164 ++++++++++++------ backend/windmill-queue/src/jobs.rs | 2 +- .../lib/components/FlowPreviewContent.svelte | 4 +- .../lib/components/FlowStatusViewer.svelte | 57 +++++- .../lib/components/InputTransformForm.svelte | 15 +- .../src/lib/components/TemplateEditor.svelte | 13 ++ .../flows/content/FlowModuleScript.svelte | 1 + .../flows/header/FlowPreviewButtons.svelte | 3 + .../lib/components/flows/previousResults.ts | 18 +- .../propertyPicker/PropPicker.svelte | 3 +- frontend/src/lib/utils.ts | 9 +- 13 files changed, 271 insertions(+), 80 deletions(-) diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 985e987cc1..30501b5b9e 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -4095,7 +4095,7 @@ dependencies = [ [[package]] name = "windmill-api" -version = "1.56.0" +version = "1.56.1" dependencies = [ "anyhow", "argon2", @@ -4146,7 +4146,7 @@ dependencies = [ [[package]] name = "windmill-api-client" -version = "1.56.0" +version = "1.56.1" dependencies = [ "base64", "chrono", @@ -4161,7 +4161,7 @@ dependencies = [ [[package]] name = "windmill-audit" -version = "1.56.0" +version = "1.56.1" dependencies = [ "chrono", "serde", @@ -4174,7 +4174,7 @@ dependencies = [ [[package]] name = "windmill-common" -version = "1.56.0" +version = "1.56.1" dependencies = [ "anyhow", "axum", @@ -4198,7 +4198,7 @@ dependencies = [ [[package]] name = "windmill-parser" -version = "1.56.0" +version = "1.56.1" dependencies = [ "serde", "serde_json", @@ -4206,7 +4206,7 @@ dependencies = [ [[package]] name = "windmill-parser-bash" -version = "1.56.0" +version = "1.56.1" dependencies = [ "anyhow", "itertools", @@ -4220,7 +4220,7 @@ dependencies = [ [[package]] name = "windmill-parser-go" -version = "1.56.0" +version = "1.56.1" dependencies = [ "anyhow", "itertools", @@ -4232,7 +4232,7 @@ dependencies = [ [[package]] name = "windmill-parser-py" -version = "1.56.0" +version = "1.56.1" dependencies = [ "anyhow", "itertools", @@ -4247,7 +4247,7 @@ dependencies = [ [[package]] name = "windmill-parser-ts" -version = "1.56.0" +version = "1.56.1" dependencies = [ "anyhow", "deno_core", @@ -4261,7 +4261,7 @@ dependencies = [ [[package]] name = "windmill-queue" -version = "1.56.0" +version = "1.56.1" dependencies = [ "anyhow", "chrono", @@ -4284,7 +4284,7 @@ dependencies = [ [[package]] name = "windmill-worker" -version = "1.56.0" +version = "1.56.1" dependencies = [ "anyhow", "async-recursion", diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 81ae8a6f9e..fc1b538485 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -2931,6 +2931,23 @@ paths: schema: $ref: "#/components/schemas/Job" + # /w/{workspace}/jobs/flow/current_state/{id}: + # get: + # summary: get flow current step state + # operationId: getJob + # tags: + # - job + # parameters: + # - $ref: "#/components/parameters/WorkspaceId" + # - $ref: "#/components/parameters/JobId" + # responses: + # "200": + # description: state details + # content: + # application/json: + # schema: + # type: string + /w/{workspace}/jobs/getupdate/{id}: get: summary: get job updates @@ -3156,6 +3173,29 @@ paths: schema: type: string + /w/{workspace}/jobs/flow/resume/{id}: + post: + summary: resume a job for a suspended flow as an owner + operationId: resumeSuspendedJobAsOwner + tags: + - job + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - $ref: "#/components/parameters/JobId" + requestBody: + required: true + content: + application/json: + schema: + type: object + responses: + "201": + description: job resumed + content: + text/plain: + schema: + type: string + /w/{workspace}/jobs/cancel/{id}/{resume_id}/{signature}: get: summary: cancel a job for a suspended flow diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index ff37a6350c..367cb3d96c 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -63,6 +63,7 @@ pub fn workspaced_service() -> Router { .route("/completed/get_result/:id", get(get_completed_job_result)) .route("/completed/delete/:id", post(delete_completed_job)) .route("/get/:id", get(get_job)) + .route("/flow/resume/:id", post(resume_suspended_job_as_owner)) .route("/getupdate/:id", get(get_job_update)) .route( "/job_signature/:job_id/:resume_id", @@ -454,6 +455,25 @@ async fn list_jobs( Ok(Json(jobs.into_iter().map(From::from).collect())) } +pub async fn resume_suspended_job_as_owner( + authed: Authed, + Extension(db): Extension, + Path((w_id, job_id)): Path<(String, Uuid)>, + QueryOrBody(value): QueryOrBody, +) -> error::Result { + let value = value.unwrap_or(serde_json::Value::Null); + let mut tx = db.begin().await?; + + let flow = get_suspended_flow_info(job_id, &mut tx).await?; + + insert_resume_job(0, job_id, &flow, value, Some(authed.username), &mut tx).await?; + + resume_immediately_if_relevant(flow, job_id, &mut tx).await?; + + tx.commit().await?; + Ok(StatusCode::CREATED) +} + pub async fn resume_suspended_job( /* unauthed */ Extension(db): Extension, @@ -472,18 +492,7 @@ pub async fn resume_suspended_job( } mac.verify_slice(hex::decode(secret)?.as_ref()) .map_err(|_| anyhow::anyhow!("Invalid signature"))?; - let flow = sqlx::query!( - r#" - SELECT id, flow_status, suspend - FROM queue - WHERE id = ( SELECT parent_job FROM queue WHERE id = $1 UNION ALL SELECT parent_job FROM completed_job WHERE id = $1) - FOR UPDATE - "#, - job_id, - ) - .fetch_optional(&mut tx) - .await? - .ok_or_else(|| anyhow::anyhow!("parent flow job not found"))?; + let flow = get_suspended_flow_info(job_id, &mut tx).await?; let exists = sqlx::query_scalar!( r#" @@ -499,6 +508,55 @@ pub async fn resume_suspended_job( return Err(anyhow::anyhow!("resume request already sent").into()); } + insert_resume_job(resume_id, job_id, &flow, value, approver.approver, &mut tx).await?; + + resume_immediately_if_relevant(flow, job_id, &mut tx).await?; + + tx.commit().await?; + Ok(StatusCode::CREATED) +} + +/* If the flow is currently waiting to be resumed (`FlowStatusModule::WaitingForEvents`) + * the suspend column must be set to the number of resume messages waited on. + * + * The flow's queue row is locked in this transaction because to avoid race conditions around + * the suspend column. + * That is, a job needs one event but it hasn't arrived, a worker counts zero events before + * entering WaitingForEvents. Then this message arrives but the job isn't in WaitingForEvents + * yet so the suspend counter isn't updated. Then the job enters WaitingForEvents expecting + * one event to arrive based on the count that is no longer correct. */ +async fn resume_immediately_if_relevant<'c>( + flow: FlowInfo, + job_id: Uuid, + tx: &mut Transaction<'c, Postgres>, +) -> error::Result<()> { + Ok( + if let Some(suspend) = (0 < flow.suspend).then(|| flow.suspend - 1) { + let status = + serde_json::from_value::(flow.flow_status.context("no flow status")?) + .context("deserialize flow status")?; + if matches!(status.current_step(), Some(FlowStatusModule::WaitingForEvents { job, .. }) if job == &job_id) + { + sqlx::query!( + "UPDATE queue SET suspend = $1 WHERE id = $2", + suspend, + flow.id, + ) + .execute(tx) + .await?; + } + }, + ) +} + +async fn insert_resume_job<'c>( + resume_id: u32, + job_id: Uuid, + flow: &FlowInfo, + value: serde_json::Value, + approver: Option, + tx: &mut Transaction<'c, Postgres>, +) -> error::Result<()> { sqlx::query!( r#" INSERT INTO resume_job @@ -510,38 +568,38 @@ pub async fn resume_suspended_job( job_id, flow.id, value, - approver.approver + approver ) - .execute(&mut tx) + .execute(tx) .await?; + Ok(()) +} - /* If the flow is currently waiting to be resumed (`FlowStatusModule::WaitingForEvents`) - * the suspend column must be set to the number of resume messages waited on. - * - * The flow's queue row is locked in this transaction because to avoid race conditions around - * the suspend column. - * That is, a job needs one event but it hasn't arrived, a worker counts zero events before - * entering WaitingForEvents. Then this message arrives but the job isn't in WaitingForEvents - * yet so the suspend counter isn't updated. Then the job enters WaitingForEvents expecting - * one event to arrive based on the count that is no longer correct. */ - if let Some(suspend) = (0 < flow.suspend).then(|| flow.suspend - 1) { - let status = - serde_json::from_value::(flow.flow_status.context("no flow status")?) - .context("deserialize flow status")?; - if matches!(status.current_step(), Some(FlowStatusModule::WaitingForEvents { job, .. }) if job == &job_id) - { - sqlx::query!( - "UPDATE queue SET suspend = $1 WHERE id = $2", - suspend, - flow.id, - ) - .execute(&mut tx) - .await?; - } - } +#[derive(sqlx::FromRow)] +struct FlowInfo { + id: Uuid, + flow_status: Option, + suspend: i32, +} - tx.commit().await?; - Ok(StatusCode::CREATED) +async fn get_suspended_flow_info<'c>( + job_id: Uuid, + tx: &mut Transaction<'c, Postgres>, +) -> error::Result { + let flow = sqlx::query_as!( + FlowInfo, + r#" + SELECT id, flow_status, suspend + FROM queue + WHERE id = ( SELECT parent_job FROM queue WHERE id = $1 UNION ALL SELECT parent_job FROM completed_job WHERE id = $1) + FOR UPDATE + "#, + job_id, + ) + .fetch_optional(tx) + .await? + .ok_or_else(|| anyhow::anyhow!("parent flow job not found"))?; + Ok(flow) } pub async fn cancel_suspended_job( @@ -1439,17 +1497,21 @@ async fn get_completed_job( Ok(Json(job)) } -// async fn get_flow_current_step_state( -// Extension(db): Extension, -// Path((w_id, id)): Path<(String, Uuid)>, -// ) -> error::JsonResult { -// let x = sqlx::query!(" -// SELECT (flow_status->'step')::integer as step, jsonb_array_length(flow_status->'modules') as len -// FROM queue WHERE id = $1 AND workspace_id = $2", -// id, w_id) -// .fetch_optional(&db).await?; -// Ok(Json(String::new())) -// } +async fn get_flow_current_step_state( + Extension(db): Extension, + Path((w_id, id)): Path<(String, Uuid)>, +) -> error::JsonResult { + let x = sqlx::query!( + " + SELECT (flow_status->'step')::integer as step, jsonb_array_length(flow_status->'modules') as len + FROM queue WHERE id = $1 AND workspace_id = $2", + id, + w_id + ) + .fetch_optional(&db) + .await?; + Ok(Json(String::new())) +} async fn get_completed_job_result( Extension(db): Extension, diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 181035ec13..7d9251a4fa 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -453,7 +453,7 @@ pub async fn push<'c>( { let mut modules = flow.modules.clone(); modules.push(FlowModule { - id: "".to_string(), + id: format!("{}-v", flow.modules[flow.modules.len() - 1].id), value: FlowModuleValue::Identity, input_transforms: HashMap::new(), stop_after_if: None, diff --git a/frontend/src/lib/components/FlowPreviewContent.svelte b/frontend/src/lib/components/FlowPreviewContent.svelte index ad611ec217..d513e3c89f 100644 --- a/frontend/src/lib/components/FlowPreviewContent.svelte +++ b/frontend/src/lib/components/FlowPreviewContent.svelte @@ -17,6 +17,7 @@ let capturePayload: CapturePayload export let previewMode: 'upTo' | 'whole' export let open: boolean + export let is_owner: boolean = false export let jobId: string | undefined = undefined export let job: Job | undefined = undefined @@ -48,6 +49,7 @@ return m }) } + function extractFlow(previewMode: 'upTo' | 'whole'): Flow { if (previewMode === 'whole') { return $flowStore @@ -156,7 +158,7 @@ />
{#if jobId} - + {:else}
Flow status will be displayed here
{/if} diff --git a/frontend/src/lib/components/FlowStatusViewer.svelte b/frontend/src/lib/components/FlowStatusViewer.svelte index 368f61461c..5bd015469e 100644 --- a/frontend/src/lib/components/FlowStatusViewer.svelte +++ b/frontend/src/lib/components/FlowStatusViewer.svelte @@ -1,10 +1,10 @@ {#if job} @@ -168,6 +180,41 @@
+ {:else if job.flow_status?.modules?.[job?.flow_status?.step].type === FlowStatusModule.type.WAITING_FOR_EVENTS} +
+

Waiting for approval from the previous step

+
+ {#if is_owner} +
+
+ +
+
+ +
+ The payload is optional, it is passed to the following step through the + `resume` variable +
+ {:else} + You cannot resume the job without the resume id since you are not an owner of {job.script_path} + {/if} +
+
{:else if job.logs} 
{job.logs}
diff --git a/frontend/src/lib/components/InputTransformForm.svelte b/frontend/src/lib/components/InputTransformForm.svelte index 764f6ad789..f9ccadc382 100644 --- a/frontend/src/lib/components/InputTransformForm.svelte +++ b/frontend/src/lib/components/InputTransformForm.svelte @@ -16,7 +16,6 @@ import type { InputTransform } from '$lib/gen' import TemplateEditor from './TemplateEditor.svelte' import Tooltip from './Tooltip.svelte' - import { escape } from 'svelte/internal' export let schema: Schema export let arg: InputTransform | any @@ -28,7 +27,8 @@ export let variableEditor: VariableEditor | undefined = undefined export let itemPicker: ItemPicker | undefined = undefined - export let monaco: SimpleEditor | undefined = undefined + let monaco: SimpleEditor | undefined = undefined + let monacoTemplate: TemplateEditor | undefined = undefined let argInput: ArgInput | undefined = undefined let inputCat: InputCat = 'object' @@ -73,6 +73,7 @@ if (isStaticTemplate(inputCat)) { arg.value = `\$\{${rawValue}}` setPropertyType(arg.value) + monacoTemplate?.setCode(arg.value) } else { arg.expr = getDefaultExpr(undefined, previousModuleId, rawValue) arg.type = 'javascript' @@ -86,6 +87,7 @@ focusProp(argName, 'append', (path) => { const toAppend = `\$\{${path}}` arg.value = `${arg.value ?? ''}${toAppend}` + monacoTemplate?.setCode(arg.value) setPropertyType(arg.value) argInput?.focus() return false @@ -219,8 +221,13 @@ {/if} {#if isStaticTemplate(inputCat) && propertyType == 'static'} -
- +
+
{:else if propertyType === undefined || propertyType == 'static'}
@@ -65,6 +67,7 @@ 0)) { + if (pickableProperties.hasResume) { pickableProperties["approvers"] = "The list of approvers" } return { - extraLib: buildExtraLib(flowInput, priorIds), + extraLib: buildExtraLib(flowInput, priorIds, previousModule?.suspend != undefined), pickableProperties } } -export function buildExtraLib(flowInput: Record, results: Record): string { +export function buildExtraLib(flowInput: Record, results: Record, resume: boolean): string { return ` /** * get variable (including secret) at path @@ -163,6 +163,18 @@ declare const params: any; * result by id */ declare const results = ${JSON.stringify(results)}; + +${resume ? ` +/** + * resume payload + */ +declare const resume: any + +/** + * The list of approvers separated by , + */ +declare const approvers: string +` : ''} ` } diff --git a/frontend/src/lib/components/propertyPicker/PropPicker.svelte b/frontend/src/lib/components/propertyPicker/PropPicker.svelte index b3291d39b1..d12e527ad7 100644 --- a/frontend/src/lib/components/propertyPicker/PropPicker.svelte +++ b/frontend/src/lib/components/propertyPicker/PropPicker.svelte @@ -129,7 +129,8 @@ pureViewer={!$propPickerConfig} json={{ resume: 'The resume payload', - resumes: 'All resume payloads from all approvers' + resumes: 'All resume payloads from all approvers', + approvers: 'The list of approvers' }} on:select={(e) => { dispatch('select', `${e.detail}`) diff --git a/frontend/src/lib/utils.ts b/frontend/src/lib/utils.ts index 24121ba030..b571141e0b 100644 --- a/frontend/src/lib/utils.ts +++ b/frontend/src/lib/utils.ts @@ -474,7 +474,8 @@ export function scriptPathToHref(path: string): string { export async function getScriptByPath(path: string): Promise<{ content: string language: SupportedLanguage - schema: any + schema: any, + description: string }> { if (path.startsWith('hub/')) { const { content, language, schema } = await ScriptService.getHubScriptByPath({ path }) @@ -482,7 +483,8 @@ export async function getScriptByPath(path: string): Promise<{ return { content, language: language as SupportedLanguage, - schema + schema, + description: '' } } else { const script = await ScriptService.getScriptByPath({ @@ -492,7 +494,8 @@ export async function getScriptByPath(path: string): Promise<{ return { content: script.content, language: script.language, - schema: script.schema + schema: script.schema, + description: script.description } } }