starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-07 16:03:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

display errors much better in the error handler

Browse Source
This commit is contained in:
Ruben Fiszel
2022-12-08 11:07:44 +01:00
parent 9715baa6b5
commit 8bcf4952a2
8 changed files with 207 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-common/src/flow_status.rs
+18 -7
View File
@@ -24,7 +24,7 @@ pub const MAX_RETRY_INTERVAL: Duration = HOURS.saturating_mul(6);
pub struct FlowStatus {
pub step: i32,
pub modules: Vec<FlowStatusModule>,
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<String>,
#[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![] },
}
backend/windmill-common/src/flows.rs
+6
View File
@@ -160,6 +160,12 @@ pub struct FlowModule {
pub sleep: Option<InputTransform>,
}
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",
backend/windmill-worker/src/jobs.rs
+1 -1
View File
@@ -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 {
backend/windmill-worker/src/worker_flow.rs
+120 -62
View File
@@ -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 }),
),
))
}
frontend/src/lib/components/FlowStatusViewer.svelte
+15 -12
View File
@@ -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']
}
}
}
frontend/src/lib/components/graph/FlowGraph.svelte
+39 -25
View File
@@ -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 ? `<img src="${langImg[lang]}" class="grayscale">` : ''}
${host != 'inline' ? `<img src="${hostImg[host]}" class="grayscale">` : ''}
<span class="center-center font-semibold bg-indigo-100 text-indigo-800 rounded px-1 pb-[2px] ml-[2px]">
${id ?? numberToChars(graphId - 1)}
${nodeId}
</span>
</div>
</div>
@@ -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: `
<div class="w-full max-h-full text-center ellipsize-multi-line text-2xs [-webkit-line-clamp:2] px-1">
Error handler
<div class="w-full flex justify-between items-center px-1">
<div class="text-left ellipsize text-2xs truncate">
Error Handler
</div>
<div class="flex items-center">
<span class="center-center font-semibold bg-indigo-100 text-indigo-800 rounded px-1 pb-[2px] ml-[2px]">
${mod.id}
</span>
</div>
</div>
`
},
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)
}
}
}
frontend/src/lib/components/graph/model.ts
+1
View File
@@ -24,6 +24,7 @@ export type GraphModuleState = {
result?: any
scheduled_for?: string
job_id?: string
parent_module?: string
}
export type NestedNodes = GraphItem[]
openflow.openapi.yaml
+7 -1
View File
@@ -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: