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

feat(backend): Input transform errors (#1006)

Browse Source 
* Add pre run error & propagate input transforms there

* Handle pre-run errors

* Update SQLX data

* Update SQLX data
This commit is contained in:
Kai Jellinghaus
2022-12-08 23:56:49 +01:00
committed by GitHub
parent e1f5fcb162
commit abc0ca3a5b
16 changed files with 154 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/Cargo.lock
Generated
+7
View File
@@ -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",
backend/Cargo.toml
+1
View File
@@ -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"
backend/migrations/20221208135921_job_pre_run_error.down.sql
+1
View File
@@ -0,0 +1 @@
-- Add down migration script here
backend/migrations/20221208135921_job_pre_run_error.up.sql
+2
View File
@@ -0,0 +1,2 @@
-- Add up migration script here
ALTER TABLE queue ADD COLUMN pre_run_error text;
backend/sqlx-data.json
+67 -66
View File
@@ -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": [
backend/tests/worker.rs
+1
View File
@@ -829,6 +829,7 @@ impl RunJob {
/* parent_job */ None,
/* is_flow_step */ false,
/* running */ false,
None,
)
.await
.expect("push has to succeed");
backend/windmill-api/src/apps.rs
+1
View File
@@ -541,6 +541,7 @@ async fn execute_component(
None,
false,
false,
None,
)
.await?;
backend/windmill-api/src/flows.rs
+2
View File
@@ -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!(
backend/windmill-api/src/jobs.rs
+9
View File
@@ -807,6 +807,7 @@ struct UnifiedJob {
is_flow_step: bool,
language: Option<ScriptLang>,
is_skipped: bool,
pre_run_error: Option<String>,
}
impl From<UnifiedJob> for Job {
@@ -867,6 +868,7 @@ impl From<UnifiedJob> 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?;
backend/windmill-api/src/oauth2.rs
+1
View File
@@ -791,6 +791,7 @@ async fn slack_command(
None,
false,
false,
None,
)
.await?;
tx.commit().await?;
backend/windmill-api/src/scripts.rs
+1
View File
@@ -345,6 +345,7 @@ async fn create_script(
None,
false,
false,
None,
)
.await?;
tx
backend/windmill-queue/src/jobs.rs
+6 -3
View File
@@ -251,6 +251,7 @@ pub async fn push<'c>(
parent_job: Option<Uuid>,
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<ScriptLang>,
pub same_worker: bool,
pub pre_run_error: Option<String>,
}
impl QueuedJob {
backend/windmill-queue/src/schedule.rs
+1
View File
@@ -112,6 +112,7 @@ pub async fn push_scheduled_job<'c>(
None,
false,
false,
None,
)
.await?;
Ok(tx)
backend/windmill-worker/Cargo.toml
+1
View File
@@ -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
backend/windmill-worker/src/worker.rs
+4
View File
@@ -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);
backend/windmill-worker/src/worker_flow.rs
+49 -36
View File
@@ -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<String>,
by_id: &IdContext,
base_internal_url: &str,
) -> anyhow::Result<Map<String, serde_json::Value>> {
) -> windmill_common::error::Result<Map<String, serde_json::Value>> {
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<TransformContext> = 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::<Vec<_>>()
};
/* 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;