diff --git a/backend/windmill-common/src/flow_status.rs b/backend/windmill-common/src/flow_status.rs index b20951aa45..66f98b1477 100644 --- a/backend/windmill-common/src/flow_status.rs +++ b/backend/windmill-common/src/flow_status.rs @@ -24,7 +24,7 @@ pub const MAX_RETRY_INTERVAL: Duration = HOURS.saturating_mul(6); pub struct FlowStatus { pub step: i32, pub modules: Vec, - pub failure_module: FlowStatusModule, + pub failure_module: FlowStatusModuleWParent, #[serde(default)] #[serde(skip_serializing_if = "is_default")] pub retry: RetryStatus, @@ -67,6 +67,14 @@ pub struct Approval { pub approver: String, } +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct FlowStatusModuleWParent { + #[serde(skip_serializing_if = "Option::is_none")] + pub parent_module: Option, + #[serde(flatten)] + pub module_status: FlowStatusModule, +} + #[derive(Serialize, Deserialize, Debug, Clone)] #[serde(tag = "type")] pub enum FlowStatusModule { @@ -171,12 +179,15 @@ impl FlowStatus { .iter() .map(|m| FlowStatusModule::WaitingForPriorSteps { id: m.id.clone() }) .collect(), - failure_module: FlowStatusModule::WaitingForPriorSteps { - id: f - .failure_module - .as_ref() - .map(|x| x.id.clone()) - .unwrap_or_else(|| "failure".to_string()), + failure_module: FlowStatusModuleWParent { + parent_module: None, + module_status: FlowStatusModule::WaitingForPriorSteps { + id: f + .failure_module + .as_ref() + .map(|x| x.id.clone()) + .unwrap_or_else(|| "failure".to_string()), + }, }, retry: RetryStatus { fail_count: 0, previous_result: None, failed_jobs: vec![] }, } diff --git a/backend/windmill-common/src/flows.rs b/backend/windmill-common/src/flows.rs index e83db6d5be..13b15c781a 100644 --- a/backend/windmill-common/src/flows.rs +++ b/backend/windmill-common/src/flows.rs @@ -160,6 +160,12 @@ pub struct FlowModule { pub sleep: Option, } +impl FlowModule { + pub fn id_append(&mut self, s: &str) { + self.id = format!("{}-{}", self.id, s); + } +} + #[derive(Deserialize, Serialize, Debug, Clone, PartialEq)] #[serde( tag = "type", diff --git a/backend/windmill-worker/src/jobs.rs b/backend/windmill-worker/src/jobs.rs index 2faa9e3f54..25cff17bdb 100644 --- a/backend/windmill-worker/src/jobs.rs +++ b/backend/windmill-worker/src/jobs.rs @@ -62,7 +62,7 @@ pub async fn add_completed_job( if queued_job.job_kind == JobKind::Flow || queued_job.job_kind == JobKind::FlowPreview { let jobs = queued_job.parse_flow_status().map(|s| { let mut modules = s.modules; - modules.extend([s.failure_module]); + modules.extend([s.failure_module.module_status]); modules .into_iter() .filter_map(|m| match m { diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 402ce68953..e698f0162b 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -18,7 +18,7 @@ use serde_json::{json, Map, Value}; use tokio::sync::mpsc::Sender; use tracing::instrument; use uuid::Uuid; -use windmill_common::flow_status::{Iterator, JobResult}; +use windmill_common::flow_status::{FlowStatusModuleWParent, Iterator, JobResult}; use windmill_common::{ error::{self, to_anyhow, Error}, flow_status::{ @@ -80,7 +80,7 @@ pub async fn update_flow_status_after_job_completion( let module_status = module_index .and_then(|i| old_status.modules.get(i)) - .unwrap_or(&old_status.failure_module); + .unwrap_or(&old_status.failure_module.module_status); tracing::debug!("UPDATE FLOW STATUS 2: {module_index:#?} {module_status:#?} {old_status:#?} "); @@ -294,13 +294,21 @@ pub async fn update_flow_status_after_job_completion( if let Some(new_status) = new_status.as_ref() { if is_failure_step { + let parent_module = sqlx::query_scalar!( + "SELECT flow_status->'failure_module'->>'parent_module' FROM queue WHERE id = $1", + flow + ) + .fetch_one(&mut tx) + .await?; + + tracing::info!("setting failure module for flow XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX {parent_module:?}"); sqlx::query!( " UPDATE queue SET flow_status = JSONB_SET(flow_status, ARRAY['failure_module'], $1) WHERE id = $2 ", - json!(new_status), + json!(FlowStatusModuleWParent { parent_module, module_status: new_status.clone() }), flow ) .execute(&mut tx) @@ -778,7 +786,7 @@ async fn push_next_flow_job( .modules .get(i) .cloned() - .unwrap_or_else(|| status.failure_module.clone()); + .unwrap_or_else(|| status.failure_module.module_status.clone()); // if this is an empty module of if the module has aleady been completed, successfully, update the parent flow if flow.modules.is_empty() || matches!(status_module, FlowStatusModule::Success { .. }) { @@ -811,6 +819,7 @@ async fn push_next_flow_job( .or_else(|| flow.failure_module.as_ref()) .with_context(|| format!("no module at index {}", status.step))?; + let current_id = &module.id; let previous_id = if i >= 1 { flow.modules.get(i - 1).map(|m| m.id.clone()).unwrap() } else { @@ -1022,13 +1031,14 @@ async fn push_next_flow_job( * In that case, `i` will index past `flow.modules`. The above should handle that and * re-run the failure module. */ i = flow.modules.len(); + module = flow .failure_module .as_ref() /* If this fails, it's a update_flow_status_after_job_completion shouldn't have called * handle_flow to get here. */ .context("missing failure module")?; - status_module = status.failure_module.clone(); + status_module = status.failure_module.module_status.clone(); /* (retry feature) save the previous_result the first time this step is run */ let retry = &module.retry.clone().unwrap_or_default(); @@ -1292,26 +1302,41 @@ async fn push_next_flow_job( tracing::debug!("STATUS STEP: {:?} {i} {:#?}", status.step, new_status); - let json_pointer = if i >= flow.modules.len() { - "'failure_module'" + if i >= flow.modules.len() { + sqlx::query!( + " + UPDATE queue + SET flow_status = JSONB_SET( + JSONB_SET(flow_status, ARRAY['failure_module'], $1), + ARRAY['step'], $2) + WHERE id = $3 + ", + json!(FlowStatusModuleWParent { + parent_module: Some(current_id.clone()), + module_status: new_status.clone() + }), + json!(i), + flow_job.id + ) + .execute(&mut tx) + .await?; } else { - "'modules', $1::TEXT" + sqlx::query!( + " + UPDATE queue + SET flow_status = JSONB_SET( + JSONB_SET(flow_status, ARRAY['modules', $1::TEXT], $2), + ARRAY['step'], $3) + WHERE id = $4 + ", + i as i32, + json!(new_status), + json!(i), + flow_job.id + ) + .execute(&mut tx) + .await?; }; - sqlx::query(&format!( - " - UPDATE queue - SET flow_status = JSONB_SET( - JSONB_SET(flow_status, ARRAY[{json_pointer}], $2), - ARRAY['step'], $3) - WHERE id = $4 - " - )) - .bind(i as i32) - .bind(json!(new_status)) - .bind(json!(i)) - .bind(flow_job.id) - .execute(&mut tx) - .await?; tx.commit().await?; @@ -1589,31 +1614,45 @@ async fn compute_next_flow_transform<'c>( match next_loop_status { LoopStatus::EmptyIterator => Ok((tx, NextFlowTransform::EmptyInnerFlows)), - LoopStatus::NextIteration(ns) => Ok(( - tx, - NextFlowTransform::Continue( - vec![JobPayload::RawFlow { - value: FlowValue { - modules: (*modules).clone(), - failure_module: flow.failure_module.clone(), - same_worker: flow.same_worker, - }, - path: Some(format!("{}/loop-{}", flow_job.script_path(), ns.index)), - }], - NextStatus::NextLoopIteration(ns), - ), - )), + LoopStatus::NextIteration(ns) => { + let mut fm = flow.failure_module.clone(); + if let Some(mut failure_module) = flow.failure_module.clone() { + failure_module.id_append(&format!("{}/{}", status.step, ns.index)); + fm = Some(failure_module); + } + Ok(( + tx, + NextFlowTransform::Continue( + vec![JobPayload::RawFlow { + value: FlowValue { + modules: (*modules).clone(), + failure_module: fm, + same_worker: flow.same_worker, + }, + path: Some(format!("{}/loop-{}", flow_job.script_path(), ns.index)), + }], + NextStatus::NextLoopIteration(ns), + ), + )) + } LoopStatus::ParallelIteration { itered } => Ok(( tx, NextFlowTransform::Continue( (0..itered.len()) - .map(|i| JobPayload::RawFlow { - value: FlowValue { - modules: (*modules).clone(), - failure_module: flow.failure_module.clone(), - same_worker: flow.same_worker, - }, - path: Some(format!("{}/loop-{}", flow_job.script_path(), i)), + .map(|i| { + 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); + } + JobPayload::RawFlow { + value: FlowValue { + modules: (*modules).clone(), + failure_module: fm.clone(), + same_worker: flow.same_worker, + }, + path: Some(format!("{}/loop-{}", flow_job.script_path(), i)), + } }) .collect(), NextStatus::AllFlowJobs { @@ -1673,14 +1712,18 @@ async fn compute_next_flow_transform<'c>( } else { default.clone() }; - + let mut fm = flow.failure_module.clone(); + if let Some(mut failure_module) = flow.failure_module.clone() { + failure_module.id_append(&status.step.to_string()); + fm = Some(failure_module); + } Ok(( tx, NextFlowTransform::Continue( vec![JobPayload::RawFlow { value: FlowValue { modules, - failure_module: flow.failure_module.clone(), + failure_module: fm, same_worker: flow.same_worker, }, path: Some(format!( @@ -1694,7 +1737,7 @@ async fn compute_next_flow_transform<'c>( )) } FlowModuleValue::BranchAll { branches, parallel, .. } => { - let (status, flow_jobs) = match status_module { + let (branch_status, flow_jobs) = match status_module { FlowStatusModule::WaitingForPriorSteps { .. } | FlowStatusModule::WaitingForEvents { .. } | FlowStatusModule::WaitingForExecutor { .. } => { @@ -1707,17 +1750,27 @@ async fn compute_next_flow_transform<'c>( branches .iter() .enumerate() - .map(|(i, b)| JobPayload::RawFlow { - value: FlowValue { - modules: b.modules.clone(), - failure_module: flow.failure_module.clone(), - same_worker: flow.same_worker, - }, - path: Some(format!( - "{}/branchall-{}", - flow_job.script_path(), - i - )), + .map(|(i, b)| { + let mut fm = flow.failure_module.clone(); + if let Some(mut failure_module) = + flow.failure_module.clone() + { + failure_module + .id_append(&format!("{}/{i}", status.step)); + fm = Some(failure_module); + } + JobPayload::RawFlow { + value: FlowValue { + modules: b.modules.clone(), + failure_module: fm.clone(), + same_worker: flow.same_worker, + }, + path: Some(format!( + "{}/branchall-{}", + flow_job.script_path(), + i + )), + } }) .collect(), NextStatus::AllFlowJobs { @@ -1760,7 +1813,7 @@ async fn compute_next_flow_transform<'c>( }; let modules = branches - .get(status.branch) + .get(branch_status.branch) .map(|b| b.modules.clone()) .ok_or_else(|| { Error::BadRequest(format!( @@ -1768,22 +1821,27 @@ async fn compute_next_flow_transform<'c>( )) })?; + let mut fm = flow.failure_module.clone(); + if let Some(mut failure_module) = flow.failure_module.clone() { + failure_module.id_append(&format!("{}/{}", status.step, branch_status.branch)); + fm = Some(failure_module); + } Ok(( tx, NextFlowTransform::Continue( vec![JobPayload::RawFlow { value: FlowValue { modules, - failure_module: flow.failure_module.clone(), + failure_module: fm.clone(), same_worker: flow.same_worker, }, path: Some(format!( "{}/branchall-{}", flow_job.script_path(), - status.branch + branch_status.branch )), }], - NextStatus::NextBranchStep(NextBranch { status, flow_jobs }), + NextStatus::NextBranchStep(NextBranch { status: branch_status, flow_jobs }), ), )) } diff --git a/frontend/src/lib/components/FlowStatusViewer.svelte b/frontend/src/lib/components/FlowStatusViewer.svelte index 0b210b7e00..d3fc7bf3fd 100644 --- a/frontend/src/lib/components/FlowStatusViewer.svelte +++ b/frontend/src/lib/components/FlowStatusViewer.svelte @@ -71,25 +71,26 @@ $: innerModules && localFlowModuleStates && updateInnerModules() function updateInnerModules() { - innerModules.forEach((module, i) => { + innerModules.forEach((mod, i) => { if ( - module.type === FlowStatusModule.type.WAITING_FOR_EVENTS && + mod.type === FlowStatusModule.type.WAITING_FOR_EVENTS && localFlowModuleStates?.[innerModules?.[i - 1]?.id ?? '']?.type == FlowStatusModule.type.SUCCESS ) { - localFlowModuleStates[module.id ?? ''] = { type: module.type } + localFlowModuleStates[mod.id ?? ''] = { type: mod.type } } else if ( - module.type === FlowStatusModule.type.WAITING_FOR_EXECUTOR && - localFlowModuleStates[module.id ?? '']?.scheduled_for == undefined + mod.type === FlowStatusModule.type.WAITING_FOR_EXECUTOR && + localFlowModuleStates[mod.id ?? '']?.scheduled_for == undefined ) { JobService.getJob({ workspace: $workspaceStore ?? '', - id: module.job ?? '' + id: mod.job ?? '' }).then((job) => { - localFlowModuleStates[module.id ?? ''] = { - type: module.type, + localFlowModuleStates[mod.id ?? ''] = { + type: mod.type, scheduled_for: 'scheduled for ' + displayDate(job?.['scheduled_for'], true), - job_id: job?.id + job_id: job?.id, + parent_module: mod['parent_module'] } }) } @@ -218,7 +219,7 @@ flowState[flowJobIds.moduleId].previewResult[j] = e.detail.result flowState[flowJobIds.moduleId].previewArgs = e.detail.args jobResults[j] = - e.detail.result == null ? 'Job in progress ...' : e.detail.result + e.detail.type == 'QueuedJob' ? 'Job in progress ...' : e.detail.result jobFailures[j] = e.detail.success === false } if (e.detail.type == 'QueuedJob') { @@ -289,7 +290,8 @@ if (e.detail.type == 'QueuedJob') { localFlowModuleStates[mod.id] = { type: FlowStatusModule.type.IN_PROGRESS, - logs: e.detail.logs + logs: e.detail.logs, + parent_module: mod['parent_module'] } } else { localFlowModuleStates[mod.id] = { @@ -298,7 +300,8 @@ : FlowStatusModule.type.FAILURE, logs: e.detail.logs, result: e.detail.result, - job_id: e.detail.id + job_id: e.detail.id, + parent_module: mod['parent_module'] } } } diff --git a/frontend/src/lib/components/graph/FlowGraph.svelte b/frontend/src/lib/components/graph/FlowGraph.svelte index e3754f5f84..2b98b67700 100644 --- a/frontend/src/lib/components/graph/FlowGraph.svelte +++ b/frontend/src/lib/components/graph/FlowGraph.svelte @@ -18,7 +18,7 @@ } from '.' import { defaultIfEmptyString, truncateRev } from '$lib/utils' import { createEventDispatcher } from 'svelte' - import { numberToChars } from '../flows/utils' + import { charsToNumber, numberToChars } from '../flows/utils' export let modules: FlowModule[] | undefined = [] export let failureModule: FlowModule | undefined = undefined @@ -57,14 +57,21 @@ const item = getConvertedFlowModule(m) item && nestedNodes.push(item) }) - const endParentIds = getParentIds() nestedNodes.push(createVirtualNode(getParentIds(), 'Flow end')) - if (failureModule) { - nestedNodes.push(createErrorHandler(endParentIds, failureModule)) + + if (!flowModuleStates) { + if (failureModule) nestedNodes.push(createErrorHandler(failureModule)) + } else { + Object.entries(flowModuleStates ?? []) + .filter(([k, v]) => k.startsWith('failure')) + .forEach(([k, v]) => { + nestedNodes.push(createErrorHandler({ id: k } as FlowModule, v.parent_module)) + }) } const flatNodes = flattenNestedNodes(nestedNodes) const layered = layoutNodes(flatNodes) + nodes = layered.nodes // width = layered.width height = layered.height @@ -144,7 +151,7 @@ if (!item) return [] if (isNode(item)) { - return ['' + item.id] + return [numberToChars(item.id)] } else if (isLoop(item)) { return getParentIds(item.items) } else if (isBranch(item)) { @@ -190,10 +197,10 @@ inline: '' } const wrapperWidth = lang ? 'w-[calc(100%-70px)]' : 'w-[calc(100%-50px)]' - const graphId = idGenerator.next().value - let nodeId = onClickDetail.id ?? numberToChars(graphId - 1) + let nodeId = id ?? numberToChars(idGenerator.next().value - 1) + return { - id: graphId, + id: charsToNumber(nodeId), position: { x: -1, y: -1 }, data: { html: ` @@ -205,7 +212,7 @@ ${lang ? `` : ''} ${host != 'inline' ? `` : ''} - ${id ?? numberToChars(graphId - 1)} + ${nodeId} @@ -225,7 +232,7 @@ selectedNode = nodeId } if (onClickDetail.id == undefined) { - onClickDetail.id = numberToChars(graphId - 1) + onClickDetail.id = nodeId } dispatch('click', onClickDetail) }, @@ -286,7 +293,7 @@ ), items: [] } - const branchParent = [branch.node.id.toString()] + const branchParent = [numberToChars(branch.node.id)] if (branches.length == 0) { branch.items.push([createVirtualNode(branchParent, 'No branches')]) } @@ -298,7 +305,7 @@ modules.forEach((module) => { const item = getConvertedFlowModule( module, - items.length ? items : branch.node.id.toString(), + items.length ? items : numberToChars(branch.node.id), edgesLabel[i] ) item && items.push(item) @@ -327,8 +334,9 @@ } function layoutNodes(nodes: Node[]): { nodes: Node[]; height: number } { - const stratify = dagStratify().id(({ id }: Node) => '' + id) + const stratify = dagStratify().id(({ id }: Node) => numberToChars(id)) const dag = stratify(nodes) + const layout = sugiyama() .decross(decrossOpt()) .coord(coordCenter()) @@ -337,7 +345,7 @@ return { nodes: dag.descendants().map((des) => ({ ...des.data, - id: +des.data.id, + id: des.data.id, position: { x: des.x ? des.x + (width - boxSize.width - NODE.width) / 2 : 0, y: des.y || 0 @@ -353,7 +361,7 @@ node.parentIds.forEach((pid, i) => { edges.push({ id: `e-${pid}-${node.id}`, - source: +pid, + source: charsToNumber(pid), target: node.id, labelBgColor: 'white', arrow: true, @@ -369,7 +377,7 @@ function createVirtualNode(parentIds: string[], label: string, edgesLabel?: string): Node { return { - id: idGenerator.next().value, + id: -idGenerator.next().value - 1, position: { x: -1, y: -1 }, data: { html: ` @@ -386,28 +394,34 @@ } } - function createErrorHandler(parentIds: string[], module: FlowModule): Node { + function createErrorHandler(mod: FlowModule, parent_module?: string): Node { return { - id: -1, + id: -idGenerator.next().value - 1, position: { x: -1, y: -1 }, data: { html: ` -
- Error handler +
+
+ Error Handler +
+
+ + ${mod.id} + +
` }, width: NODE.width, height: NODE.height, - bgColor: 'rgb(248 113 113)', + bgColor: selectedNode == mod.id ? '#f5f5f5' : getStateColor(flowModuleStates?.[mod.id]?.type), borderColor: '#999', - - parentIds, + parentIds: parent_module ? [parent_module] : [], clickCallback: (node) => { if (!notSelectable) { - selectedNode = module.id + selectedNode = mod.id } - dispatch('click', module) + dispatch('click', mod) } } } diff --git a/frontend/src/lib/components/graph/model.ts b/frontend/src/lib/components/graph/model.ts index 93687a8a4e..4655dddda2 100644 --- a/frontend/src/lib/components/graph/model.ts +++ b/frontend/src/lib/components/graph/model.ts @@ -24,6 +24,7 @@ export type GraphModuleState = { result?: any scheduled_for?: string job_id?: string + parent_module?: string } export type NestedNodes = GraphItem[] diff --git a/openflow.openapi.yaml b/openflow.openapi.yaml index d566a5aee1..e56170f51f 100644 --- a/openflow.openapi.yaml +++ b/openflow.openapi.yaml @@ -338,7 +338,13 @@ components: items: $ref: "#/components/schemas/FlowStatusModule" failure_module: - $ref: "#/components/schemas/FlowStatusModule" + allOf: + - $ref: "#/components/schemas/FlowStatusModule" + - type: object + properties: + parent_module: + type: string + retry: type: object properties: