From abc0ca3a5b55f9e338470002f94608ef1389490a Mon Sep 17 00:00:00 2001 From: Kai Jellinghaus Date: Thu, 8 Dec 2022 23:56:49 +0100 Subject: [PATCH] feat(backend): Input transform errors (#1006) * Add pre run error & propagate input transforms there * Handle pre-run errors * Update SQLX data * Update SQLX data --- backend/Cargo.lock | 7 + backend/Cargo.toml | 1 + .../20221208135921_job_pre_run_error.down.sql | 1 + .../20221208135921_job_pre_run_error.up.sql | 2 + backend/sqlx-data.json | 133 +++++++++--------- backend/tests/worker.rs | 1 + backend/windmill-api/src/apps.rs | 1 + backend/windmill-api/src/flows.rs | 2 + backend/windmill-api/src/jobs.rs | 9 ++ backend/windmill-api/src/oauth2.rs | 1 + backend/windmill-api/src/scripts.rs | 1 + backend/windmill-queue/src/jobs.rs | 9 +- backend/windmill-queue/src/schedule.rs | 1 + backend/windmill-worker/Cargo.toml | 1 + backend/windmill-worker/src/worker.rs | 4 + backend/windmill-worker/src/worker_flow.rs | 85 ++++++----- 16 files changed, 154 insertions(+), 105 deletions(-) create mode 100644 backend/migrations/20221208135921_job_pre_run_error.down.sql create mode 100644 backend/migrations/20221208135921_job_pre_run_error.up.sql diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 346de334dd..1782871e22 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -802,6 +802,12 @@ version = "1.0.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4f94fa09c2aeea5b8839e414b7b841bf429fd25b9c522116ac97ee87856d88b2" +[[package]] +name = "dyn-iter" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "131726693bce13b09331bee70734fe266666332b6ddfef23e9dca5b8bf6dea66" + [[package]] name = "either" version = "1.8.0" @@ -4285,6 +4291,7 @@ dependencies = [ "const_format", "deno_core", "dotenv", + "dyn-iter", "futures", "git-version", "itertools", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 4a74001dc4..48dccc9bb1 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -138,3 +138,4 @@ tokio-metrics = "0.1.0" lazy_static = "1.4.0" serde_derive = "1.0.147" const_format = { version = "0.2", features = ["rust_1_64", "rust_1_51"] } +dyn-iter = "0.2.0" diff --git a/backend/migrations/20221208135921_job_pre_run_error.down.sql b/backend/migrations/20221208135921_job_pre_run_error.down.sql new file mode 100644 index 0000000000..d2f607c5b8 --- /dev/null +++ b/backend/migrations/20221208135921_job_pre_run_error.down.sql @@ -0,0 +1 @@ +-- Add down migration script here diff --git a/backend/migrations/20221208135921_job_pre_run_error.up.sql b/backend/migrations/20221208135921_job_pre_run_error.up.sql new file mode 100644 index 0000000000..79952e7959 --- /dev/null +++ b/backend/migrations/20221208135921_job_pre_run_error.up.sql @@ -0,0 +1,2 @@ +-- Add up migration script here +ALTER TABLE queue ADD COLUMN pre_run_error text; \ No newline at end of file diff --git a/backend/sqlx-data.json b/backend/sqlx-data.json index 90937356e3..c8cdb89461 100644 --- a/backend/sqlx-data.json +++ b/backend/sqlx-data.json @@ -258,6 +258,73 @@ }, "query": "DELETE FROM account WHERE id = $1 AND workspace_id = $2" }, + "0eea65f84168d4425d20d24b321cdee98b69dbf1e8a75e916af6d37abc4aee92": { + "describe": { + "columns": [ + { + "name": "id", + "ordinal": 0, + "type_info": "Uuid" + } + ], + "nullable": [ + false + ], + "parameters": { + "Left": [ + "Varchar", + "Uuid", + "Bool", + "Uuid", + "Varchar", + "Varchar", + "Timestamptz", + "Int8", + "Varchar", + "Text", + "Text", + "Jsonb", + { + "Custom": { + "kind": { + "Enum": [ + "script", + "preview", + "flow", + "dependencies", + "flowpreview", + "script_hub", + "identity", + "flowdependencies" + ] + }, + "name": "job_kind" + } + }, + "Varchar", + "Jsonb", + "Jsonb", + "Bool", + { + "Custom": { + "kind": { + "Enum": [ + "python3", + "deno", + "go", + "bash" + ] + }, + "name": "script_lang" + } + }, + "Bool", + "Text" + ] + } + }, + "query": "INSERT INTO queue\n (workspace_id, id, running, parent_job, created_by, permissioned_as, scheduled_for, \n script_hash, script_path, raw_code, raw_lock, args, job_kind, schedule_path, raw_flow, flow_status, is_flow_step, language, started_at, same_worker, pre_run_error)\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, CASE WHEN $3 THEN now() END, $19, $20) RETURNING id" + }, "11b1586acdfc180c5a077861ee1f7201fcbcec9d0ebada464f9d952c9c3e400d": { "describe": { "columns": [], @@ -1078,72 +1145,6 @@ }, "query": "UPDATE variable SET path = $1 WHERE path = $2 AND workspace_id = $3" }, - "499978e08a3a5e4d09d20f4212dbeff6ed5c5d4a696243cd3687c71099e8a6a6": { - "describe": { - "columns": [ - { - "name": "id", - "ordinal": 0, - "type_info": "Uuid" - } - ], - "nullable": [ - false - ], - "parameters": { - "Left": [ - "Varchar", - "Uuid", - "Bool", - "Uuid", - "Varchar", - "Varchar", - "Timestamptz", - "Int8", - "Varchar", - "Text", - "Text", - "Jsonb", - { - "Custom": { - "kind": { - "Enum": [ - "script", - "preview", - "flow", - "dependencies", - "flowpreview", - "script_hub", - "identity", - "flowdependencies" - ] - }, - "name": "job_kind" - } - }, - "Varchar", - "Jsonb", - "Jsonb", - "Bool", - { - "Custom": { - "kind": { - "Enum": [ - "python3", - "deno", - "go", - "bash" - ] - }, - "name": "script_lang" - } - }, - "Bool" - ] - } - }, - "query": "INSERT INTO queue\n (workspace_id, id, running, parent_job, created_by, permissioned_as, scheduled_for, \n script_hash, script_path, raw_code, raw_lock, args, job_kind, schedule_path, raw_flow, flow_status, is_flow_step, language, started_at, same_worker)\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, CASE WHEN $3 THEN now() END, $19) RETURNING id" - }, "4ad5fa2f08236507aad911a95697e84fc0c3a274ba0e928da28c4d146cf8f1a8": { "describe": { "columns": [ diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index 5393f6f40f..062edf0c96 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -829,6 +829,7 @@ impl RunJob { /* parent_job */ None, /* is_flow_step */ false, /* running */ false, + None, ) .await .expect("push has to succeed"); diff --git a/backend/windmill-api/src/apps.rs b/backend/windmill-api/src/apps.rs index a5e987bc7f..cba52ebada 100644 --- a/backend/windmill-api/src/apps.rs +++ b/backend/windmill-api/src/apps.rs @@ -541,6 +541,7 @@ async fn execute_component( None, false, false, + None, ) .await?; diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index af664cb340..9f62f5aa41 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -200,6 +200,7 @@ async fn create_flow( None, false, false, + None, ) .await?; sqlx::query!( @@ -304,6 +305,7 @@ async fn update_flow( None, false, false, + None, ) .await?; sqlx::query!( diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 01d97f24c1..1b981b5912 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -807,6 +807,7 @@ struct UnifiedJob { is_flow_step: bool, language: Option, is_skipped: bool, + pre_run_error: Option, } impl From for Job { @@ -867,6 +868,7 @@ impl From for Job { is_flow_step: uj.is_flow_step, language: uj.language, same_worker: false, + pre_run_error: uj.pre_run_error, }), t => panic!("job type {} not valid", t), } @@ -962,6 +964,7 @@ pub async fn run_flow_by_path( run_query.parent_job, false, false, + None, ) .await?; tx.commit().await?; @@ -994,6 +997,7 @@ pub async fn run_job_by_path( run_query.parent_job, false, false, + None, ) .await?; tx.commit().await?; @@ -1059,6 +1063,7 @@ pub async fn run_wait_result_job_by_path( run_query.parent_job, false, false, + None, ) .await?; tx.commit().await?; @@ -1092,6 +1097,7 @@ pub async fn run_wait_result_job_by_hash( run_query.parent_job, false, false, + None, ) .await?; tx.commit().await?; @@ -1143,6 +1149,7 @@ async fn run_preview_job( None, false, false, + None, ) .await?; tx.commit().await?; @@ -1173,6 +1180,7 @@ async fn run_preview_flow_job( None, false, false, + None, ) .await?; tx.commit().await?; @@ -1205,6 +1213,7 @@ pub async fn run_job_by_hash( run_query.parent_job, false, false, + None, ) .await?; tx.commit().await?; diff --git a/backend/windmill-api/src/oauth2.rs b/backend/windmill-api/src/oauth2.rs index 91582c49dd..963478d83c 100644 --- a/backend/windmill-api/src/oauth2.rs +++ b/backend/windmill-api/src/oauth2.rs @@ -791,6 +791,7 @@ async fn slack_command( None, false, false, + None, ) .await?; tx.commit().await?; diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index 7610c7b91c..92fe8a874e 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/backend/windmill-api/src/scripts.rs @@ -345,6 +345,7 @@ async fn create_script( None, false, false, + None, ) .await?; tx diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index f25faee696..41d66b3c8e 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -251,6 +251,7 @@ pub async fn push<'c>( parent_job: Option, is_flow_step: bool, mut same_worker: bool, + pre_run_error: Option<&windmill_common::error::Error>, ) -> Result<(Uuid, Transaction<'c, Postgres>), Error> { let scheduled_for = scheduled_for_o.unwrap_or_else(chrono::Utc::now); let args_json = serde_json::Value::Object(args); @@ -465,8 +466,8 @@ pub async fn push<'c>( "INSERT INTO queue (workspace_id, id, running, parent_job, created_by, permissioned_as, scheduled_for, script_hash, script_path, raw_code, raw_lock, args, job_kind, schedule_path, raw_flow, \ - flow_status, is_flow_step, language, started_at, same_worker) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, CASE WHEN $3 THEN now() END, $19) \ + flow_status, is_flow_step, language, started_at, same_worker, pre_run_error) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, CASE WHEN $3 THEN now() END, $19, $20) \ RETURNING id", workspace_id, job_id, @@ -486,7 +487,8 @@ pub async fn push<'c>( flow_status.map(|f| serde_json::json!(f)), is_flow_step, language: ScriptLang, - same_worker + same_worker, + pre_run_error.map(|e| e.to_string()) ) .fetch_one(&mut tx) .await @@ -585,6 +587,7 @@ pub struct QueuedJob { pub is_flow_step: bool, pub language: Option, pub same_worker: bool, + pub pre_run_error: Option, } impl QueuedJob { diff --git a/backend/windmill-queue/src/schedule.rs b/backend/windmill-queue/src/schedule.rs index 4719bef17c..68c9e0d8c7 100644 --- a/backend/windmill-queue/src/schedule.rs +++ b/backend/windmill-queue/src/schedule.rs @@ -112,6 +112,7 @@ pub async fn push_scheduled_job<'c>( None, false, false, + None, ) .await?; Ok(tx) diff --git a/backend/windmill-worker/Cargo.toml b/backend/windmill-worker/Cargo.toml index d7283c939b..e767a759d3 100644 --- a/backend/windmill-worker/Cargo.toml +++ b/backend/windmill-worker/Cargo.toml @@ -48,3 +48,4 @@ rand.workspace = true # TODO: Remove. only used by token creation hack. deno_core.workspace = true const_format.workspace = true git-version.workspace = true +dyn-iter.workspace = true diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 407ce72457..4057ce0ca2 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -757,6 +757,10 @@ async fn handle_queued_job( if job.canceled { return Err(Error::ExecutionErr(canceled_job_to_result(&job)))?; } + if let Some(e) = job.pre_run_error { + return Err(Error::ExecutionErr(e)); + } + match job.job_kind { JobKind::FlowPreview | JobKind::Flow => { let args = job.args.clone().unwrap_or(Value::Null); diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index e698f0162b..cf12a4ad34 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -7,6 +7,7 @@ */ use std::collections::HashMap; +use std::iter; use std::time::Duration; use crate::jobs::{add_completed_job, add_completed_job_error, schedule_again_if_scheduled}; @@ -14,6 +15,7 @@ use crate::js_eval::{eval_timeout, EvalCreds, IdContext}; use crate::worker; use anyhow::Context; use async_recursion::async_recursion; +use dyn_iter::DynIter; use serde_json::{json, Map, Value}; use tokio::sync::mpsc::Sender; use tracing::instrument; @@ -35,7 +37,7 @@ use windmill_queue::{ }; #[async_recursion] -#[instrument(level = "trace", skip_all)] +// #[instrument(level = "trace", skip_all)] pub async fn update_flow_status_after_job_completion( db: &DB, client: &windmill_api_client::Client, @@ -680,7 +682,7 @@ async fn transform_input( approvers: Vec, by_id: &IdContext, base_internal_url: &str, -) -> anyhow::Result> { +) -> windmill_common::error::Result> { let mut mapped = serde_json::Map::new(); for (key, val) in input_transforms.into_iter() { @@ -767,7 +769,7 @@ pub async fn handle_flow( } #[async_recursion] -#[instrument(level = "trace", skip_all)] +// #[instrument(level = "trace", skip_all)] async fn push_next_flow_job( flow_job: &QueuedJob, mut status: FlowStatus, @@ -1094,7 +1096,7 @@ async fn push_next_flow_job( let mut transform_context: Option = None; - let mut args = match &module.value { + let args: windmill_common::error::Result<_> = match &module.value { FlowModuleValue::Script { input_transforms, .. } | FlowModuleValue::RawScript { input_transforms, .. } | FlowModuleValue::Flow { input_transforms, .. } => { @@ -1116,27 +1118,27 @@ async fn push_next_flow_job( by_id, base_internal_url, ) - .await? + .await } FlowModuleValue::Identity => match last_result.clone() { - Value::Object(m) => m, + Value::Object(m) => Ok(m), v @ _ => { let mut m = Map::new(); m.insert("previous_result".to_string(), v); - m + Ok(m) } }, _ => { /* embedded flow input is augmented with embedding flow input */ if let Some(value) = &flow_job.args { - value + Ok(value .as_object() .ok_or_else(|| { Error::BadRequest(format!("Expected an object value, found: {value:?}")) })? - .clone() + .clone()) } else { - Map::new() + Ok(Map::new()) } } }; @@ -1187,44 +1189,54 @@ async fn push_next_flow_job( let continue_on_same_worker = flow.same_worker && module.suspend.is_none() && module.sleep.is_none(); - let all_args = match &next_status { - NextStatus::NextLoopIteration(NextIteration { new_args, .. }) => { - args.extend(new_args.clone()); - vec![args] - } + let zipped = { + let all_args = match &next_status { + NextStatus::NextLoopIteration(NextIteration { new_args, .. }) => { + let args = args.as_ref().map(|args| { + let mut args = args.clone(); + args.extend(new_args.clone()); + args + }); + DynIter::new(iter::once(args)) + } - NextStatus::AllFlowJobs { - branchall: Some(BranchAllStatus { len, .. }), - iterator: None, - .. - } => (0..*len).map(|_| args.clone()).collect(), - NextStatus::AllFlowJobs { - branchall: None, - iterator: Some(Iterator { itered, .. }), - .. - } => itered - .into_iter() - .enumerate() - .map(|(i, v)| { - let mut new_args = args.clone(); - new_args.insert("iter".to_string(), json!({ "index": i, "value": v })); - new_args - }) - .collect(), + NextStatus::AllFlowJobs { + branchall: Some(BranchAllStatus { len, .. }), + iterator: None, + .. + } => DynIter::new((0..*len).map(|_| args.as_ref().map(|args| args.clone()))), + NextStatus::AllFlowJobs { + branchall: None, + iterator: Some(Iterator { itered, .. }), + .. + } => DynIter::new(itered.into_iter().enumerate().map(|(i, v)| { + args.as_ref().map(|args| { + let mut new_args = args.clone(); + new_args.insert("iter".to_string(), json!({ "index": i, "value": v })); + new_args + }) + })), - _ => vec![args], + _ => DynIter::new(iter::once(args.as_ref().map(|m| m.clone()))), + }; + + job_payloads.into_iter().zip(all_args).collect::>() }; /* Finally, push the job into the queue */ let mut tx = db.begin().await?; let mut uuids = vec![]; - for (payload, args) in job_payloads.into_iter().zip(all_args.into_iter()) { + for (payload, args) in zipped { + let (ok, err) = match args { + Ok(v) => (Some(v), None), + Err(e) => (None, Some(e)), + }; let (uuid, inner_tx) = push( tx, &flow_job.workspace_id, payload, - args, + ok.unwrap_or_else(|| Map::new()), &flow_job.created_by, flow_job.permissioned_as.to_owned(), scheduled_for_o, @@ -1232,6 +1244,7 @@ async fn push_next_flow_job( Some(flow_job.id), true, continue_on_same_worker, + err, ) .await?; tx = inner_tx;