From 4f30b1fb8234ffbc830cff6a91d54bf16632fb50 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Thu, 23 May 2024 19:16:18 +0200 Subject: [PATCH] feat: store failed_retries per module state and display failed retries in flow status viewer --- backend/windmill-common/src/flow_status.rs | 7 ++ backend/windmill-queue/src/jobs.rs | 1 + backend/windmill-worker/src/worker_flow.rs | 65 ++++++++----------- .../components/FlowStatusViewerInner.svelte | 41 +++++++++++- .../flows/map/FlowModuleSchemaItem.svelte | 3 + .../lib/components/flows/map/MapItem.svelte | 2 + .../src/lib/components/graph/FlowGraph.svelte | 1 + openflow.openapi.yaml | 6 +- 8 files changed, 85 insertions(+), 41 deletions(-) diff --git a/backend/windmill-common/src/flow_status.rs b/backend/windmill-common/src/flow_status.rs index 77a9840006..7fde7aa298 100644 --- a/backend/windmill-common/src/flow_status.rs +++ b/backend/windmill-common/src/flow_status.rs @@ -124,6 +124,7 @@ struct UntaggedFlowStatusModule { parallel: Option, while_loop: Option, approvers: Option>, + failed_retries: Option>, } #[derive(Serialize, Debug, Clone)] @@ -167,6 +168,8 @@ pub enum FlowStatusModule { #[serde(default)] #[serde(skip_serializing_if = "Vec::is_empty")] approvers: Vec, + #[serde(skip_serializing_if = "Vec::is_empty")] + failed_retries: Vec, }, Failure { id: String, @@ -175,6 +178,8 @@ pub enum FlowStatusModule { flow_jobs: Option>, #[serde(skip_serializing_if = "Option::is_none")] branch_chosen: Option, + #[serde(skip_serializing_if = "Vec::is_empty")] + failed_retries: Vec, }, } @@ -235,6 +240,7 @@ impl<'de> Deserialize<'de> for FlowStatusModule { flow_jobs: untagged.flow_jobs, branch_chosen: untagged.branch_chosen, approvers: untagged.approvers.unwrap_or_default(), + failed_retries: untagged.failed_retries.unwrap_or_default(), }), "Failure" => Ok(FlowStatusModule::Failure { id: untagged @@ -245,6 +251,7 @@ impl<'de> Deserialize<'de> for FlowStatusModule { .ok_or_else(|| serde::de::Error::missing_field("job"))?, flow_jobs: untagged.flow_jobs, branch_chosen: untagged.branch_chosen, + failed_retries: untagged.failed_retries.unwrap_or_default(), }), other => Err(serde::de::Error::unknown_variant( other, diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 0d5f6fc720..47f794ee98 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -3811,6 +3811,7 @@ async fn restarted_flows_resolution( flow_jobs: _, branch_chosen: _, approvers: _, + failed_retries: _, } => Ok(truncated_modules.push(module)), _ => Err(Error::InternalErr(format!( "Flow cannot be restarted from a non successful module", diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index e6393d097e..e26a65310a 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -17,7 +17,7 @@ use crate::common::{hash_args, save_in_cache}; use crate::js_eval::{eval_timeout, IdContext}; use crate::{AuthedClient, PreviousResult, SameWorkerPayload, SendResult, KEEP_JOB_DIR}; use anyhow::Context; -use serde::Serialize; +use serde::{Deserialize, Serialize}; use serde_json::value::RawValue; use serde_json::{json, Value}; use sqlx::types::Json; @@ -158,6 +158,10 @@ pub struct SkipIfStopped { pub args: Option>>>, } +#[derive(sqlx::FromRow, Deserialize)] +pub struct RowFlowStatus { + pub flow_status: sqlx::types::Json>, +} // #[instrument(level = "trace", skip_all)] pub async fn update_flow_status_after_job_completion_internal< 'a, @@ -190,26 +194,26 @@ pub async fn update_flow_status_after_job_completion_internal< ) = { // tracing::debug!("UPDATE FLOW STATUS: {flow:?} {success} {result:?} {w_id} {depth}"); - let old_status_json = sqlx::query_scalar!( + let old_status_json = sqlx::query_as::<_, RowFlowStatus>( "SELECT flow_status FROM queue WHERE id = $1 AND workspace_id = $2", - flow, - w_id ) + .bind(flow) + .bind(w_id) .fetch_one(db) .await .map_err(|e| { Error::InternalErr(format!( "fetching flow status {flow} while reporting {success} {result:?}: {e}" )) - })? - .ok_or_else(|| Error::InternalErr(format!("requiring a previous status")))?; - - let old_status = serde_json::from_value::(old_status_json).or_else(|e| { - Err(Error::InternalErr(format!( - "requiring status to be parsable as FlowStatus: {e:?}" - ))) })?; + let old_status = serde_json::from_str::(old_status_json.flow_status.get()) + .or_else(|e| { + Err(Error::InternalErr(format!( + "requiring status to be parsable as FlowStatus: {e:?}" + ))) + })?; + let module_index = usize::try_from(old_status.step).ok(); let module_status = module_index @@ -384,6 +388,7 @@ pub async fn update_flow_status_after_job_completion_internal< flow_jobs: Some(jobs.clone()), branch_chosen: None, approvers: vec![], + failed_retries: vec![], } } else { success = false; @@ -392,6 +397,7 @@ pub async fn update_flow_status_after_job_completion_internal< job: job_id_for_status.clone(), flow_jobs: Some(jobs.clone()), branch_chosen: None, + failed_retries: vec![], } }; let r = sqlx::query_scalar!( @@ -517,6 +523,7 @@ pub async fn update_flow_status_after_job_completion_internal< flow_jobs, branch_chosen, approvers: vec![], + failed_retries: old_status.retry.failed_jobs.clone(), }), ) } else { @@ -529,7 +536,7 @@ pub async fn update_flow_status_after_job_completion_internal< .fetch_optional(&mut tx) .await .map_err(|e| { - Error::InternalErr(format!("error while getting retry fromn step: {e}")) + Error::InternalErr(format!("error while getting retry from step: {e}")) })? .flatten(); @@ -549,6 +556,7 @@ pub async fn update_flow_status_after_job_completion_internal< job: job_id_for_status.clone(), flow_jobs, branch_chosen, + failed_retries: old_status.retry.failed_jobs.clone(), }), ) } @@ -1788,11 +1796,13 @@ async fn push_next_flow_job status.retry.failed_jobs.push(job.clone()); sqlx::query( "UPDATE queue - SET flow_status = JSONB_SET(flow_status, ARRAY['retry'], $1) + SET flow_status = JSONB_SET(JSONB_SET(flow_status, ARRAY['retry'], $1), ARRAY['modules', $3::TEXT, 'failed_retries'], $4) WHERE id = $2", ) .bind(json!(RetryStatus { fail_count, ..status.retry.clone() })) .bind(flow_job.id) + .bind(status.step) + .bind(json!(status.retry.failed_jobs)) .execute(db) .await .context("update flow retry")?; @@ -1819,9 +1829,7 @@ async fn push_next_flow_job .context("missing failure module")?; 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(); - if retry.has_attempts() { + if module.retry.as_ref().is_some_and(|x| x.has_attempts()) { sqlx::query( "UPDATE queue SET flow_status = JSONB_SET(flow_status, ARRAY['retry'], $1) @@ -1835,28 +1843,6 @@ async fn push_next_flow_job }; None } - - /* (retry feature) save the previous_result the first time this step is run */ - } - FlowStatusModule::WaitingForPriorSteps { .. } - if module - .retry - .as_ref() - .map(|x| x.has_attempts()) - .unwrap_or(false) - && status.retry.fail_count == 0 => - { - sqlx::query( - "UPDATE queue - SET flow_status = JSONB_SET(flow_status, ARRAY['retry'], $1) - WHERE id = $2", - ) - .bind(json!(RetryStatus { fail_count: 0, failed_jobs: vec![] })) - .bind(flow_job.id) - .execute(db) - .await - .context("update flow retry")?; - None } _ => None, }; @@ -1972,7 +1958,8 @@ async fn push_next_flow_job job: Uuid::nil(), flow_jobs: Some(vec![]), branch_chosen: None, - approvers: vec![] + approvers: vec![], + failed_retries: vec![], })) .bind(flow_job.id) .execute(db) diff --git a/frontend/src/lib/components/FlowStatusViewerInner.svelte b/frontend/src/lib/components/FlowStatusViewerInner.svelte index 9ab7bf0398..69a49a078d 100644 --- a/frontend/src/lib/components/FlowStatusViewerInner.svelte +++ b/frontend/src/lib/components/FlowStatusViewerInner.svelte @@ -66,6 +66,7 @@ let jobFailures: boolean[] = [] let forloop_selected = '' + let retry_selected = '' let timeout: NodeJS.Timeout let localModuleStates: Writable> = writable({}) @@ -296,7 +297,8 @@ duration_ms: job['duration_ms'], started_at: started_at, iteration: mod.iterator?.itered?.length, - iteration_total: mod.iterator?.itered?.length + iteration_total: mod.iterator?.itered?.length, + retries: mod?.failed_retries?.length // retries: $flowStateStore?.raw_flow }) setDurationStatusByJob(mod.id, job.id, { @@ -653,6 +655,43 @@
{/if}
  • + {#if render && Array.isArray(mod.failed_retries)} + {#each mod.failed_retries as failedRetry, j} + + + +
    + +
    + {/each} + {/if} {#if ['InProgress', 'Success', 'Failure'].includes(mod.type)} {#if job.raw_flow?.modules[i]?.value.type == 'flow'} + {#if retries}{retries}{/if}
    • Retries {/if} + {#if concurrency}
    m.id === mod.id) @@ -174,6 +175,7 @@ {:else} dispatch('select', mod.id)} on:delete={onDelete} on:move={() => dispatch('move')} diff --git a/frontend/src/lib/components/graph/FlowGraph.svelte b/frontend/src/lib/components/graph/FlowGraph.svelte index 296d3d43fa..20f725d9a7 100644 --- a/frontend/src/lib/components/graph/FlowGraph.svelte +++ b/frontend/src/lib/components/graph/FlowGraph.svelte @@ -318,6 +318,7 @@ insertable, insertableEnd, branchable, + retries: flowModuleStates?.[mod.id]?.retries, duration_ms: flowModuleStates?.[mod.id]?.duration_ms, bgColor: getStateColor(flowModuleStates?.[mod.id]?.type), annotation, diff --git a/openflow.openapi.yaml b/openflow.openapi.yaml index f1259afe3a..1461bd4141 100644 --- a/openflow.openapi.yaml +++ b/openflow.openapi.yaml @@ -513,5 +513,9 @@ components: required: - resume_id - approver - + failed_retries: + type: array + items: + type: string + format: uuid required: [type]