kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: store failed_retries per module state and display failed retries in flow status viewer

Browse Source
This commit is contained in:
Ruben Fiszel
2024-05-23 19:16:18 +02:00
parent 126459f972
commit 4f30b1fb82
8 changed files with 85 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
backend/windmill-common/src/flow_status.rs
View File

@@ -124,6 +124,7 @@ struct UntaggedFlowStatusModule {
parallel: Option<bool>,
while_loop: Option<bool>,
approvers: Option<Vec<Approval>>,
failed_retries: Option<Vec<Uuid>>,
}
#[derive(Serialize, Debug, Clone)]
@@ -167,6 +168,8 @@ pub enum FlowStatusModule {
#[serde(default)]
#[serde(skip_serializing_if = "Vec::is_empty")]
approvers: Vec<Approval>,
#[serde(skip_serializing_if = "Vec::is_empty")]
failed_retries: Vec<Uuid>,
},
Failure {
id: String,
@@ -175,6 +178,8 @@ pub enum FlowStatusModule {
flow_jobs: Option<Vec<Uuid>>,
#[serde(skip_serializing_if = "Option::is_none")]
branch_chosen: Option<BranchChosen>,
#[serde(skip_serializing_if = "Vec::is_empty")]
failed_retries: Vec<Uuid>,
},
}
@@ -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,

1
backend/windmill-queue/src/jobs.rs
View File

@@ -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",

65
backend/windmill-worker/src/worker_flow.rs
View File

@@ -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<Json<HashMap<String, Box<RawValue>>>>,
}
#[derive(sqlx::FromRow, Deserialize)]
pub struct RowFlowStatus {
pub flow_status: sqlx::types::Json<Box<serde_json::value::RawValue>>,
}
// #[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::<FlowStatus>(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::<FlowStatus>(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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
.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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
};
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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
job: Uuid::nil(),
flow_jobs: Some(vec![]),
branch_chosen: None,
approvers: vec![]
approvers: vec![],
failed_retries: vec![],
}))
.bind(flow_job.id)
.execute(db)