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

feat(backend): failure_module (#452)

Browse Source 
* run failure_module

- renames FlowModule.input_transform to input_transforms
- parse_deno_signature prints source on failure instead of debug
  representation of AST

* s/should_continue_job/should_continue_flow
This commit is contained in:
sqwishy
2022-08-19 20:18:07 +02:00
committed by GitHub
parent dfd57b4778
commit 84fd2e2532
4 changed files with 229 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/flows.rs
+32 -6
View File
@@ -78,14 +78,15 @@ pub struct FlowValue {
#[derive(Deserialize, Serialize, Debug, Clone)]
pub struct FlowModule {
#[serde(default)]
pub input_transform: HashMap<String, InputTransform>,
#[serde(alias = "input_transform")]
pub input_transforms: HashMap<String, InputTransform>,
pub value: FlowModuleValue,
pub stop_after_if_expr: Option<String>,
pub skip_if_stopped: Option<bool>,
pub summary: Option<String>,
}
#[derive(Deserialize, Serialize, Debug, Clone)]
#[derive(Deserialize, Serialize, Debug, Clone, PartialEq)]
#[serde(
tag = "type",
rename_all(serialize = "lowercase", deserialize = "lowercase")
@@ -420,14 +421,14 @@ mod tests {
let fv = FlowValue {
modules: vec![
FlowModule {
input_transform: hm,
input_transforms: hm,
value: FlowModuleValue::Script { path: "test".to_string() },
stop_after_if_expr: None,
skip_if_stopped: Some(false),
summary: None,
},
FlowModule {
input_transform: HashMap::new(),
input_transforms: HashMap::new(),
value: FlowModuleValue::RawScript(RawCode {
content: "test".to_string(),
language: crate::scripts::ScriptLang::Deno,
@@ -438,7 +439,7 @@ mod tests {
summary: None,
},
FlowModule {
input_transform: [(
input_transforms: [(
"iterand".to_string(),
InputTransform::Static { value: serde_json::json!(vec![1, 2, 3]) },
)]
@@ -454,7 +455,7 @@ mod tests {
},
],
failure_module: Some(FlowModule {
input_transform: HashMap::new(),
input_transforms: HashMap::new(),
value: FlowModuleValue::Flow { path: "test".to_string() },
stop_after_if_expr: Some("previous.isEmpty()".to_string()),
skip_if_stopped: None,
@@ -464,4 +465,29 @@ mod tests {
println!("{}", serde_json::json!(fv).to_string());
Ok(())
}
#[test]
fn test_back_compat() {
/* renamed input_transform -> input_transforms but should deserialize old name */
let s = r#"
{
"value": {
"type": "rawscript",
"content": "def main(n): return",
"language": "python3"
},
"input_transform": {
"n": {
"expr": "flow_input.iter.value",
"type": "javascript"
}
}
}
"#;
let module: FlowModule = serde_json::from_str(s).unwrap();
assert_eq!(
module.input_transforms["n"],
InputTransform::Javascript { expr: "flow_input.iter.value".to_string() }
);
}
}
backend/src/parser_ts.rs
+7 -3
View File
@@ -11,7 +11,7 @@ use crate::{
parser::{Arg, MainArgSignature, ObjectProperty, Typ},
};
use swc_common::{sync::Lrc, FileName, SourceMap};
use swc_common::{sync::Lrc, FileName, SourceMap, SourceMapper, Spanned};
use swc_ecma_ast::{
AssignPat, BindingIdent, Decl, ExportDecl, Expr, FnDecl, Ident, ModuleDecl, ModuleItem, Pat,
Str, TsArrayType, TsEntityName, TsKeywordType, TsKeywordTypeKind, TsLit, TsLitType,
@@ -76,7 +76,9 @@ pub fn parse_deno_signature(code: &str) -> error::Result<MainArgSignature> {
let (name, typ, _nullable) =
left.as_ident().map(binding_ident_to_arg).ok_or_else(|| {
error::Error::ExecutionErr(format!(
"Arg {left:?} has unexpected syntax"
"parameter syntax unsupported: `{}`",
cm.span_to_snippet(left.span())
.unwrap_or_else(|_| cm.span_to_string(left.span()))
))
})?;
Ok(Arg {
@@ -92,7 +94,9 @@ pub fn parse_deno_signature(code: &str) -> error::Result<MainArgSignature> {
})
}
_ => Err(error::Error::ExecutionErr(format!(
"Arg {x:?} has unexpected syntax"
"parameter syntax unsupported: `{}`",
cm.span_to_snippet(x.span())
.unwrap_or_else(|_| cm.span_to_string(x.span()))
))),
})
.collect::<Result<Vec<Arg>, error::Error>>()?,
backend/src/worker.rs
+86 -5
View File
@@ -1252,7 +1252,7 @@ mod tests {
use std::sync::Once;
static ONCE: Once = Once::new();
ONCE.call_once(|| crate::tracing_init::initialize_tracing());
ONCE.call_once(crate::tracing_init::initialize_tracing);
}
/// it's important this is unique between tests as there is one prometheus registry and
@@ -1285,7 +1285,7 @@ mod tests {
content: numbers.to_string(),
path: None,
}),
input_transform: Default::default(),
input_transforms: Default::default(),
stop_after_if_expr: Default::default(),
skip_if_stopped: Default::default(),
summary: Default::default(),
@@ -1301,7 +1301,7 @@ mod tests {
content: doubles.to_string(),
path: None,
}),
input_transform: [(
input_transforms: [(
"n".to_string(),
InputTransform::Javascript {
expr: "previous_result.iter.value".to_string(),
@@ -1316,7 +1316,7 @@ mod tests {
}
.into(),
},
input_transform: Default::default(),
input_transforms: Default::default(),
stop_after_if_expr: Default::default(),
skip_if_stopped: Default::default(),
summary: Default::default(),
@@ -1515,7 +1515,7 @@ def main():
"type": "rawscript",
"language": "python3",
"content": "def main(n): return n",
},
} ,
}
],
}
@@ -1544,6 +1544,87 @@ def main():
assert_eq!(result, serde_json::json!(9));
}
#[sqlx::test(fixtures("base"))]
async fn test_failure_module(db: DB) {
initialize_tracing().await;
let flow: FlowValue = serde_json::from_value(serde_json::json!({
"modules": [{
"input_transform": {
"l": { "type": "javascript", "expr": "[]", },
"n": { "type": "javascript", "expr": "flow_input.n", },
},
"value": {
"type": "rawscript",
"language": "deno",
"content": "export function main(n, l) { if (n == 0) throw l; return { l: [...l, 0] } }",
},
}, {
"input_transform": {
"l": { "type": "javascript", "expr": "previous_result.l", },
"n": { "type": "javascript", "expr": "flow_input.n", },
},
"value": {
"type": "rawscript",
"language": "deno",
"content": "export function main(n, l) { if (n == 1) throw l; return { l: [...l, 1] } }",
},
}, {
"input_transform": {
"l": { "type": "javascript", "expr": "previous_result.l", },
"n": { "type": "javascript", "expr": "flow_input.n", },
},
"value": {
"type": "rawscript",
"language": "deno",
"content": "export function main(n, l) { if (n == 2) throw l; return { l: [...l, 2] } }",
},
}],
"failure_module": {
"input_transform": { "error": { "type": "javascript", "expr": "previous_result", } },
"value": {
"type": "rawscript",
"language": "deno",
"content": "export function main(error) { return { 'from failure module': error } }",
}
},
}))
.unwrap();
let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None })
.arg("n", json!(0))
.wait_until_complete(&db)
.await;
assert!(result["from failure module"]["error"]
.as_str()
.unwrap()
.contains("Uncaught (in promise) []"));
let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None })
.arg("n", json!(1))
.wait_until_complete(&db)
.await;
assert!(result["from failure module"]["error"]
.as_str()
.unwrap()
.contains("Uncaught (in promise) [ 0 ]"));
let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None })
.arg("n", json!(2))
.wait_until_complete(&db)
.await;
assert!(result["from failure module"]["error"]
.as_str()
.unwrap()
.contains("Uncaught (in promise) [ 0, 1 ]"));
let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None })
.arg("n", json!(3))
.wait_until_complete(&db)
.await;
assert_eq!(json!({ "l": [0, 1, 2] }), result);
}
#[sqlx::test(fixtures("base"))]
async fn test_iteration(db: DB) {
initialize_tracing().await;
backend/src/worker_flow.rs
+104 -44
View File
@@ -98,47 +98,60 @@ pub async fn update_flow_status_after_job_completion(
.await?
.unwrap_or(false);
let (step_counter, new_status) = match &old_status.modules[old_status.step as usize] {
module_status @ FlowStatusModule::InProgress {
iterator: Some(Iterator { index, itered, .. }),
..
} if *index + 1 < itered.len() && (success || skip_loop_failures) => {
(old_status.step, module_status.clone())
let module = usize::try_from(old_status.step)
.ok()
.and_then(|i| old_status.modules.get(i))
.unwrap_or(&old_status.failure_module);
let (step_counter, new_status) = match module {
FlowStatusModule::InProgress { iterator: Some(Iterator { index, itered, .. }), .. }
if *index + 1 < itered.len() && (success || skip_loop_failures) =>
{
(old_status.step, module.clone())
}
module_status => {
let forloop_jobs = match module_status {
FlowStatusModule::InProgress { forloop_jobs: Some(jobs), .. } => Some(jobs.clone()),
_ => {
let forloop_jobs = match module {
FlowStatusModule::InProgress { forloop_jobs, .. } => forloop_jobs.clone(),
_ => None,
};
let new_status = if success || (forloop_jobs.is_some() && skip_loop_failures) {
FlowStatusModule::Success { job: job.id, forloop_jobs }
if success || (forloop_jobs.is_some() && skip_loop_failures) {
(
old_status.step + 1,
FlowStatusModule::Success { job: job.id, forloop_jobs },
)
} else {
FlowStatusModule::Failure { job: job.id, forloop_jobs }
};
(old_status.step + 1, new_status)
(
old_status.step,
FlowStatusModule::Failure { job: job.id, forloop_jobs },
)
}
}
};
let last_step = step_counter as usize == old_status.modules.len();
let is_last_step = usize::try_from(step_counter)
.map(|i| !(..old_status.modules.len()).contains(&i))
.unwrap_or(true);
tracing::debug!(
"old status: {:#?}\n{:#?}\n{last_step}",
"old status: {:#?}\n{:#?}\n{is_last_step}",
old_status,
new_status
);
let prev_step = old_status.step;
let (stop_early_expr, skip_if_stop_early) =
sqlx::query_as::<_, (Option<String>, Option<bool>)>(&format!(
"UPDATE queue
SET
flow_status = jsonb_set(jsonb_set(flow_status, '{{modules, {prev_step}}}', $1), \
'{{\"step\"}}', $2)
WHERE id = $3
RETURNING
(raw_flow->'modules'->{prev_step}->>'stop_after_if_expr'),
(raw_flow->'modules'->{prev_step}->>'skip_if_stopped')::bool",
))
sqlx::query_as::<_, (Option<String>, Option<bool>)>(
"
UPDATE queue
SET flow_status = JSONB_SET(
JSONB_SET(flow_status, ARRAY['modules', $1::TEXT], $2),
ARRAY['step'], $3)
WHERE id = $4
RETURNING
(raw_flow->'modules'->$1->>'stop_after_if_expr'),
(raw_flow->'modules'->$1->>'skip_if_stopped')::bool
",
)
.bind(old_status.step)
.bind(serde_json::json!(new_status))
.bind(serde_json::json!(step_counter))
.bind(flow)
@@ -181,9 +194,18 @@ pub async fn update_flow_status_after_job_completion(
}
_ => result.clone(),
};
let done = if !(success || skip_loop_failures) || last_step || stop_early {
tx.commit().await?;
let should_continue_flow = match success {
_ if stop_early => false,
true => !is_last_step,
false if skip_loop_failures => !is_last_step,
false if has_failure_module(flow, &mut tx).await? => true,
false => false,
};
tx.commit().await?;
let done = if !should_continue_flow {
let logs = if stop_early {
"Flow job stopped early".to_string()
} else {
@@ -201,8 +223,6 @@ pub async fn update_flow_status_after_job_completion(
.await?;
true
} else {
tx.commit().await?;
match handle_flow(&flow_job, db, result.clone()).await {
Err(err) => {
let _ = add_completed_job_error(
@@ -261,6 +281,23 @@ async fn skip_loop_failures<'c>(
.map_err(|e| Error::InternalErr(format!("error during retrieval of skip_loop_failures: {e}")))
}
async fn has_failure_module<'c>(
flow: Uuid,
tx: &mut sqlx::Transaction<'c, sqlx::Postgres>,
) -> Result<bool, Error> {
sqlx::query_scalar(
"
SELECT raw_flow->'failure_module' != 'null'::jsonb
FROM queue
WHERE id = $1
",
)
.bind(flow)
.fetch_one(tx)
.await
.map_err(|e| Error::InternalErr(format!("error during retrieval of has_failure_module: {e}")))
}
async fn compute_stop_early(expr: String, result: serde_json::Value) -> error::Result<bool> {
match eval_timeout(expr, [("result".to_string(), result)].into(), None, vec![]).await? {
serde_json::Value::Bool(true) => Ok(true),
@@ -313,20 +350,20 @@ pub async fn get_step_of_flow_status(db: &DB, id: Uuid) -> error::Result<i32> {
async fn transform_input(
flow_args: &Option<serde_json::Value>,
last_result: serde_json::Value,
input_transform: &HashMap<String, InputTransform>,
input_transforms: &HashMap<String, InputTransform>,
workspace: &str,
token: &str,
steps: Vec<String>,
) -> anyhow::Result<Map<String, serde_json::Value>> {
let mut mapped = serde_json::Map::new();
for (key, val) in input_transform.into_iter() {
for (key, val) in input_transforms.into_iter() {
if let InputTransform::Static { value } = val {
mapped.insert(key.to_string(), value.to_owned());
}
}
for (key, val) in input_transform.into_iter() {
for (key, val) in input_transforms.into_iter() {
match val {
InputTransform::Static { value: _ } => (),
InputTransform::Javascript { expr } => {
@@ -400,15 +437,17 @@ async fn push_next_flow_job(
serde_json::from_value::<FlowStatus>(flow_job.flow_status.clone().unwrap_or_default())
.with_context(|| format!("parse flow status {}", flow_job.id))?;
let i = usize::try_from(status.step)
/* `mut` because reassigned on FlowStatusModule::Failure when failure_module is Some */
let mut i = usize::try_from(status.step)
.with_context(|| format!("invalid module index {}", status.step))?;
let module: &FlowModule = flow
let mut module: &FlowModule = flow
.modules
.get(i)
.with_context(|| format!("no module at index {}", status.step))?;
let status_module: FlowStatusModule = status
let mut status_module: FlowStatusModule = status
.modules
.get(i)
.cloned()
@@ -420,7 +459,23 @@ async fn push_next_flow_job(
status_module
);
/* Don't evaluate `module.input_transform` after iteration has begun. Instead, args are
if matches!(&status_module, FlowStatusModule::Success { .. }) {
anyhow::bail!("no job for {status_module:?}")
} else if matches!(&status_module, FlowStatusModule::Failure { .. }) {
/* To run to the failure module, call push_next_flow_job with the current step on
* FlowStatusModule::Failure. This must update the step index to the end so that no
* subsequent steps are run after the failure module. */
i = flow.modules.len();
module = flow
.failure_module
.as_ref()
/* If this fails, it's a update_flow_status_in_progress shouldn't have called
* handle_flow to get here. */
.context("missing failure module")?;
status_module = status.failure_module.clone();
};
/* Don't evaluate `module.input_transforms` after iteration has begun. Instead, args are
* carried through the Iterator by the InProgress variant. */
#[rustfmt::skip]
@@ -452,7 +507,7 @@ async fn push_next_flow_job(
transform_input(
&flow_job.args,
last_result.clone(),
&module.input_transform,
&module.input_transforms,
&flow_job.workspace_id,
&token,
steps,
@@ -494,7 +549,7 @@ async fn push_next_flow_job(
UPDATE queue
SET flow_status = JSONB_SET(
JSONB_SET(flow_status, ARRAY['modules', $1::TEXT], $2),
'{ step }', $3)
ARRAY['step'], $3)
WHERE id = $4
RETURNING *
"#,
@@ -551,7 +606,7 @@ async fn push_next_flow_job(
.with_context(|| format!("could not iterate index {index} of {itered:?}"))?;
args.extend(iterator_args);
args.insert("iter".to_string(), json!({ "index": 0, "value": next }));
args.insert("iter".to_string(), json!({ "index": index, "value": next }));
Some(NextLoopStatus { index, itered, forloop_jobs })
}
@@ -631,12 +686,17 @@ async fn push_next_flow_job(
};
sqlx::query(
"UPDATE queue
SET flow_status = jsonb_set(flow_status, ARRAY['modules', $1::TEXT], $2)
WHERE id = $3",
"
UPDATE queue
SET flow_status = JSONB_SET(
JSONB_SET(flow_status, ARRAY['modules', $1::TEXT], $2),
ARRAY['step'], $3)
WHERE id = $4
",
)
.bind(status.step)
.bind(json!(new_status))
.bind(json!(i))
.bind(flow_job.id)
.execute(&mut tx)
.await?;