diff --git a/backend/migrations/20220713142758_script_trigger.down.sql b/backend/migrations/20220713142758_script_trigger.down.sql new file mode 100644 index 0000000000..d2f607c5b8 --- /dev/null +++ b/backend/migrations/20220713142758_script_trigger.down.sql @@ -0,0 +1 @@ +-- Add down migration script here diff --git a/backend/migrations/20220713142758_script_trigger.up.sql b/backend/migrations/20220713142758_script_trigger.up.sql new file mode 100644 index 0000000000..d9aaa2343c --- /dev/null +++ b/backend/migrations/20220713142758_script_trigger.up.sql @@ -0,0 +1,3 @@ +-- Add up migration script here +ALTER TABLE script ADD COLUMN trigger_reco_interval INTEGER; +ALTER TABLE completed_job ADD COLUMN is_skipped BOOLEAN NOT NULL DEFAULT FALSE; diff --git a/backend/openapi.yaml b/backend/openapi.yaml index 25dbf8c7b4..74406815a2 100644 --- a/backend/openapi.yaml +++ b/backend/openapi.yaml @@ -1602,7 +1602,8 @@ paths: language: type: string enum: [python3, deno] - + trigger_reco_interval: + type: integer required: - path - summary @@ -2971,6 +2972,8 @@ components: language: type: string enum: [python3, deno] + trigger_reco_interval: + type: integer required: - hash - path @@ -3033,7 +3036,15 @@ components: format: date-time job_kind: type: string - enum: ["script", "preview", "dependencies", "flow", "flowpreview"] + enum: + [ + "script", + "preview", + "dependencies", + "flow", + "flowpreview", + "script_hub", + ] schedule_path: type: string permissioned_as: @@ -3103,7 +3114,15 @@ components: type: string job_kind: type: string - enum: ["script", "preview", "dependencies", "flow", "flowpreview"] + enum: + [ + "script", + "preview", + "dependencies", + "flow", + "flowpreview", + "script_hub", + ] schedule_path: type: string permissioned_as: @@ -3120,6 +3139,8 @@ components: language: type: string enum: [python3, deno] + is_skipped: + type: boolean required: - id - created_by @@ -3131,6 +3152,7 @@ components: - job_kind - permissioned_as - is_flow_step + - is_skipped Job: allOf: @@ -3754,6 +3776,10 @@ components: $ref: "#/components/schemas/InputTransform" value: $ref: "#/components/schemas/FlowModuleValue" + stop_after_if_expr: + type: string + skip_if_stopped: + type: boolean required: - input_transform - value @@ -3775,6 +3801,10 @@ components: FlowModuleValue: type: object properties: + value: + $ref: "#/components/schemas/FlowValue" + iterator: + $ref: "#/components/schemas/InputTransform" path: type: string content: @@ -3790,6 +3820,7 @@ components: - script - flow - rawscript + - forloopflow required: - type @@ -3841,7 +3872,19 @@ components: format: uuid event: type: string - + iterator: + type: object + properties: + index: + type: integer + itered: + type: array + items: {} + args: {} + forloop_jobs: + type: array + items: + type: string required: [type] SlackToken: diff --git a/backend/sqlx-data.json b/backend/sqlx-data.json index 22b85276bc..b56520d5c4 100644 --- a/backend/sqlx-data.json +++ b/backend/sqlx-data.json @@ -2095,61 +2095,6 @@ ] } }, - "a8507084c0f45f4c08c0b317d69db26f97295bb2d410e3e5b8da2cd420536145": { - "query": "INSERT INTO completed_job as cj\n (workspace_id, id, parent_job, created_by, created_at, duration, success, script_hash, script_path, args, result, logs, \n raw_code, canceled, canceled_by, canceled_reason, job_kind, schedule_path, permissioned_as, flow_status, raw_flow, is_flow_step)\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22) ON CONFLICT (id) DO UPDATE SET success = $7, result = $11, logs = concat(cj.logs, $12) RETURNING id", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "id", - "type_info": "Uuid" - } - ], - "parameters": { - "Left": [ - "Varchar", - "Uuid", - "Uuid", - "Varchar", - "Timestamptz", - "Int4", - "Bool", - "Int8", - "Varchar", - "Jsonb", - "Jsonb", - "Text", - "Text", - "Bool", - "Varchar", - "Text", - { - "Custom": { - "name": "job_kind", - "kind": { - "Enum": [ - "script", - "preview", - "flow", - "dependencies", - "flowpreview", - "script_hub" - ] - } - } - }, - "Varchar", - "Varchar", - "Jsonb", - "Jsonb", - "Bool" - ] - }, - "nullable": [ - false - ] - } - }, "a98b2d68f023f46ab91167d3147416df672c2aed2ba5ab70e98a9da5fa47255a": { "query": "INSERT INTO workspace_settings\n (workspace_id)\n VALUES ($1)", "describe": { @@ -2297,6 +2242,62 @@ "nullable": [] } }, + "b032ac8a438eb36a61bf5a81a0c41eb10646349e8c79014f83142be35139205c": { + "query": "INSERT INTO completed_job as cj\n (workspace_id, id, parent_job, created_by, created_at, duration, success, script_hash, script_path, args, result, logs, \n raw_code, canceled, canceled_by, canceled_reason, job_kind, schedule_path, permissioned_as, flow_status, raw_flow, is_flow_step, is_skipped)\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23) ON CONFLICT (id) DO UPDATE SET success = $7, result = $11, logs = concat(cj.logs, $12) RETURNING id", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Uuid" + } + ], + "parameters": { + "Left": [ + "Varchar", + "Uuid", + "Uuid", + "Varchar", + "Timestamptz", + "Int4", + "Bool", + "Int8", + "Varchar", + "Jsonb", + "Jsonb", + "Text", + "Text", + "Bool", + "Varchar", + "Text", + { + "Custom": { + "name": "job_kind", + "kind": { + "Enum": [ + "script", + "preview", + "flow", + "dependencies", + "flowpreview", + "script_hub" + ] + } + } + }, + "Varchar", + "Varchar", + "Jsonb", + "Jsonb", + "Bool", + "Bool" + ] + }, + "nullable": [ + false + ] + } + }, "b20977e70ebac7ccbaec5a2a1e940301dd331a5f9a4be67a27cfbff8619ac8f0": { "query": "INSERT INTO usr\n (workspace_id, email, username, is_admin)\n VALUES ($1, $2, $3, true)", "describe": { @@ -2419,40 +2420,6 @@ ] } }, - "be33c6eb702c149044650d49b3c50493d7538d590be3f4ff6242fea85c57c667": { - "query": "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, content, created_by, schema, is_template, extra_perms, lock, language) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13)", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Int8", - "Varchar", - "Int8Array", - "Text", - "Text", - "Text", - "Varchar", - "Text", - "Bool", - "Jsonb", - "Text", - { - "Custom": { - "name": "script_lang", - "kind": { - "Enum": [ - "python3", - "deno" - ] - } - } - } - ] - }, - "nullable": [] - } - }, "bf1d8e043338867e1da1ed236ff6c85a566d5fd58d4b0d5c3a10454513811ba3": { "query": "UPDATE workspace_settings\n SET slack_team_id = null, slack_name = null WHERE workspace_id = $1", "describe": { @@ -3012,6 +2979,41 @@ "nullable": [] } }, + "f503c8f37507c89bf99063f49ef72e2d021898583708e776fc58d99809f4bc3c": { + "query": "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, content, created_by, schema, is_template, extra_perms, lock, language, trigger_reco_interval) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14)", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Int8", + "Varchar", + "Int8Array", + "Text", + "Text", + "Text", + "Varchar", + "Text", + "Bool", + "Jsonb", + "Text", + { + "Custom": { + "name": "script_lang", + "kind": { + "Enum": [ + "python3", + "deno" + ] + } + } + }, + "Int4" + ] + }, + "nullable": [] + } + }, "f7906298e4204ad55ec84021bb2461f369386493519637279f1188227230c580": { "query": "SELECT lock, lock_error_logs FROM script WHERE hash = $1 AND (workspace_id = $2 OR workspace_id = 'starter')", "describe": { diff --git a/backend/src/flow.rs b/backend/src/flow.rs index 39b53f0d13..ecaac534c5 100644 --- a/backend/src/flow.rs +++ b/backend/src/flow.rs @@ -60,19 +60,21 @@ pub struct NewFlow { pub schema: Option, } -#[derive(Deserialize, Serialize)] +#[derive(Deserialize, Serialize, Debug, Clone)] pub struct FlowValue { pub modules: Vec, pub failure_module: Option, } -#[derive(Deserialize, Serialize)] +#[derive(Deserialize, Serialize, Debug, Clone)] pub struct FlowModule { pub input_transform: HashMap, pub value: FlowModuleValue, + pub stop_after_if_expr: Option, + pub skip_if_stopped: Option, } -#[derive(Deserialize, Serialize)] +#[derive(Deserialize, Serialize, Debug, Clone)] #[serde( tag = "type", rename_all(serialize = "lowercase", deserialize = "lowercase") @@ -80,17 +82,24 @@ pub struct FlowModule { pub enum InputTransform { Static { value: serde_json::Value }, Javascript { expr: String }, - Resource { path: String }, } -#[derive(Serialize, Deserialize, Debug)] +#[derive(Serialize, Deserialize, Debug, Clone)] #[serde( tag = "type", rename_all(serialize = "lowercase", deserialize = "lowercase") )] pub enum FlowModuleValue { - Script { path: String }, - Flow { path: String }, + Script { + path: String, + }, + ForloopFlow { + iterator: InputTransform, + value: Box, + }, + Flow { + path: String, + }, RawScript(RawCode), } @@ -315,6 +324,8 @@ mod tests { value: FlowModuleValue::Script { path: "test".to_string(), }, + stop_after_if_expr: None, + skip_if_stopped: Some(false), }, FlowModule { input_transform: HashMap::new(), @@ -323,6 +334,28 @@ mod tests { language: crate::scripts::ScriptLang::Deno, path: None, }), + stop_after_if_expr: Some("foo = 'bar'".to_string()), + skip_if_stopped: None, + }, + FlowModule { + input_transform: [( + "iterand".to_string(), + InputTransform::Static { + value: serde_json::json!(vec![1, 2, 3]), + }, + )] + .into(), + value: FlowModuleValue::ForloopFlow { + iterator: InputTransform::Static { + value: serde_json::json!([1, 2, 3]), + }, + value: Box::new(FlowValue { + modules: vec![], + failure_module: None, + }), + }, + stop_after_if_expr: Some("previous.res1.isEmpty()".to_string()), + skip_if_stopped: None, }, ], failure_module: Some(FlowModule { @@ -330,6 +363,8 @@ mod tests { value: FlowModuleValue::Flow { path: "test".to_string(), }, + stop_after_if_expr: Some("previous.res1.isEmpty()".to_string()), + skip_if_stopped: None, }), }; println!("{}", serde_json::json!(fv).to_string()); diff --git a/backend/src/jobs.rs b/backend/src/jobs.rs index e169919d36..e1eb1852cf 100644 --- a/backend/src/jobs.rs +++ b/backend/src/jobs.rs @@ -10,16 +10,16 @@ use chrono::Duration; use sql_builder::prelude::*; use sqlx::{query_scalar, Postgres, Transaction}; use std::collections::HashMap; +use tracing::instrument; -use crate::js_eval::eval_timeout; use crate::scripts::{get_hub_script_by_path, ScriptLang}; -use crate::users::create_token_for_owner; +use crate::worker_flow::init_flow_status; use crate::{ audit::{audit_log, ActionKind}, db::{UserDB, DB}, error, error::Error, - flow::{FlowModuleValue, FlowValue, InputTransform}, + flow::FlowValue, schedule::get_schedule_opt, scripts::ScriptHash, users::{owner_to_token_owner, Authed}, @@ -32,7 +32,7 @@ use axum::{ }; use hyper::StatusCode; use serde::{Deserialize, Serialize}; -use serde_json::{json, Map, Value}; +use serde_json::{Map, Value}; use sql_builder::SqlBuilder; use ulid::Ulid; @@ -114,6 +114,7 @@ struct CompletedJob { raw_flow: Option, is_flow_step: bool, language: Option, + is_skipped: bool, } #[derive(Deserialize, Clone, Copy)] @@ -185,7 +186,7 @@ pub async fn run_job_by_path( Ok((StatusCode::CREATED, uuid.to_string())) } -async fn script_path_to_payload<'c>( +pub async fn script_path_to_payload<'c>( script_path: &str, db: &mut Transaction<'c, Postgres>, w_id: &String, @@ -438,6 +439,7 @@ async fn list_jobs( "flow_status", "is_flow_step", "language", + "false as is_skipped", ], ); let sqlc = list_completed_jobs_query( @@ -472,6 +474,7 @@ async fn list_jobs( "flow_status", "is_flow_step", "language", + "is_skipped", ], ); let sql = format!( @@ -804,7 +807,7 @@ async fn get_job_from_id<'c>( Ok((job_option, tx)) } -async fn get_queued_job<'c>( +pub async fn get_queued_job<'c>( id: Uuid, w_id: &str, tx: &mut Transaction<'c, Postgres>, @@ -840,24 +843,6 @@ pub enum JobKind { FlowPreview, } -#[derive(Serialize, Deserialize, Debug)] -pub struct FlowStatus { - pub step: i32, - pub modules: Vec, - pub failure_module: FlowStatusModule, -} - -#[derive(Serialize, Deserialize, Debug)] -#[serde(tag = "type")] -pub enum FlowStatusModule { - WaitingForPriorSteps, - WaitingForEvent { event: String }, - WaitingForExecutor { job: Uuid }, - InProgress { job: Uuid }, - Success { job: Uuid }, - Failure { job: Uuid }, -} - #[derive(sqlx::FromRow)] struct UnifiedJob { workspace_id: String, @@ -883,6 +868,7 @@ struct UnifiedJob { flow_status: Option, is_flow_step: bool, language: Option, + is_skipped: bool, } impl From for Job { @@ -914,6 +900,7 @@ impl From for Job { raw_flow: None, is_flow_step: uj.is_flow_step, language: uj.language, + is_skipped: uj.is_skipped, }), "QueuedJob" => Job::QueuedJob(QueuedJob { workspace_id: uj.workspace_id, @@ -972,6 +959,7 @@ struct PreviewFlow { args: Option>, } +#[derive(Debug)] pub enum JobPayload { ScriptHub { path: String, @@ -992,6 +980,7 @@ pub enum JobPayload { }, } +#[instrument(level = "trace", skip_all)] pub async fn push<'c>( mut tx: Transaction<'c, Postgres>, workspace_id: &str, @@ -1125,13 +1114,7 @@ pub async fn push<'c>( } }; - let flow_status = raw_flow.as_ref().map(|f| FlowStatus { - step: 0, - modules: (0..f.modules.len()) - .map(|_| FlowStatusModule::WaitingForPriorSteps) - .collect(), - failure_module: FlowStatusModule::WaitingForPriorSteps, - }); + let flow_status = raw_flow.as_ref().map(init_flow_status); let uuid = sqlx::query_scalar!( "INSERT INTO queue (workspace_id, id, parent_job, created_by, permissioned_as, scheduled_for, @@ -1155,7 +1138,9 @@ pub async fn push<'c>( language: ScriptLang ) .fetch_one(&mut tx) - .await?; + .await + .map_err(|e| Error::InternalErr(format!("Could not insert into queue {job_id}: {e}")))?; + let uuid_string = job_id.to_string(); let uuid_str = uuid_string.as_str(); let mut hm = HashMap::from([("uuid", uuid_str), ("permissioned_as", &permissioned_as)]); @@ -1192,7 +1177,8 @@ pub async fn push<'c>( Ok((uuid, tx)) } -pub async fn add_completed_job_error( +#[instrument(level = "trace", skip_all)] +pub async fn add_completed_job_error( db: &DB, queued_job: &QueuedJob, logs: String, @@ -1207,6 +1193,7 @@ pub async fn add_completed_job_error( db, &queued_job, false, + false, Some(output_map.clone()), format!("{}\n{}", logs, e.to_string()), ) @@ -1214,14 +1201,17 @@ pub async fn add_completed_job_error( Ok((a, output_map)) } +#[instrument(level = "trace", skip_all)] pub async fn add_completed_job( db: &DB, queued_job: &QueuedJob, success: bool, + skipped: bool, result: Option>, logs: String, ) -> Result { let result_json = result.map(serde_json::Value::Object); + let job_id = queued_job.id.clone(); let duration = (chrono::Utc::now() - queued_job.started_at.unwrap_or(queued_job.created_at)) .num_seconds() as i32; let _ = sqlx::query!( @@ -1229,8 +1219,8 @@ pub async fn add_completed_job( (workspace_id, id, parent_job, created_by, created_at, duration, success, script_hash, script_path, \ args, result, logs, raw_code, canceled, canceled_by, canceled_reason, job_kind, schedule_path, permissioned_as, flow_status, raw_flow, \ - is_flow_step) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22) \ + is_flow_step, is_skipped) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23) \ ON CONFLICT (id) DO UPDATE SET success = $7, result = $11, logs = concat(cj.logs, $12) \ RETURNING id", queued_job.workspace_id, @@ -1254,150 +1244,33 @@ pub async fn add_completed_job( queued_job.permissioned_as, queued_job.flow_status, queued_job.raw_flow, - queued_job.is_flow_step + queued_job.is_flow_step, + skipped ) .fetch_one(db) - .await?; + .await + .map_err(|e| Error::InternalErr(format!("Could not add completed job {job_id}: {e}")))?; tracing::debug!("Added completed job {}", queued_job.id); Ok(queued_job.id) } -pub async fn get_step_of_flow_status(db: &DB, id: Uuid) -> error::Result { - let r = sqlx::query_scalar!( - "SELECT (flow_status->'step')::integer FROM queue WHERE id = $1", - id - ) - .fetch_one(db) - .await? - .ok_or_else(|| Error::InternalErr(format!("not found step")))?; - Ok(r) -} -pub async fn update_flow_status_in_progress( - db: &DB, - w_id: &str, - flow: Uuid, - job_in_progress: Uuid, -) -> error::Result<()> { - let step = get_step_of_flow_status(db, flow).await?; - sqlx::query(&format!( - "UPDATE queue - SET flow_status = jsonb_set(flow_status, '{{modules, {}}}', $1) - WHERE id = $2 AND workspace_id = $3", - step - )) - .bind(serde_json::json!(FlowStatusModule::InProgress { - job: job_in_progress - })) - .bind(flow) - .bind(w_id) - .execute(db) - .await?; - Ok(()) -} - -pub async fn update_flow_status_after_job_completion( - db: &DB, - job: &QueuedJob, - success: bool, - result: Option>, -) -> error::Result<()> { - tracing::info!("HANDLE FLOW: {job:?} {success} {result:?}"); - - let mut tx = db.begin().await?; - - let w_id = &job.workspace_id; - - let flow = job - .parent_job - .ok_or_else(|| Error::InternalErr(format!("expected parent job")))?; - - let old_status_json = sqlx::query_scalar!( - "SELECT flow_status FROM queue WHERE id = $1 AND workspace_id = $2", - flow, - w_id - ) - .fetch_one(&mut tx) - .await? - .ok_or_else(|| Error::InternalErr(format!("requiring a previous status")))?; - - let old_status = serde_json::from_value::(old_status_json) - .ok() - .ok_or_else(|| { - Error::InternalErr(format!("requiring status to be parsabled as FlowStatus")) - })?; - - let last_step = (old_status.step + 1) as usize == old_status.modules.len(); - let new_status = if success { - FlowStatusModule::Success { job: job.id } - } else { - FlowStatusModule::Failure { job: job.id } - }; - - sqlx::query(&format!( - "UPDATE queue - SET - flow_status = jsonb_set(jsonb_set(flow_status, '{{modules, {}}}', $1), '{{\"step\"}}', $2) - WHERE id = $3", - old_status.step, - )) - .bind(serde_json::json!(new_status)) - .bind(serde_json::json!(old_status.step + 1)) - .bind(flow) - .execute(&mut tx) - .await?; - - tracing::info!("UPDATE: {:?}", new_status); - - let flow_job = get_queued_job(flow, w_id, &mut tx) - .await? - .ok_or_else(|| Error::InternalErr(format!("requiring flow to be in the queue")))?; - tx.commit().await?; - - let done = if !success || last_step { - add_completed_job( - db, - &flow_job, - success, - result, - "Flow job completed".to_string(), - ) - .await?; - true - } else { - if let Err(err) = handle_flow(&flow_job, db, result).await { - let _ = add_completed_job_error( - db, - &flow_job, - "Unexpected error during flow chaining:\n".to_string(), - err, - ) - .await; - true - } else { - false - } - }; - - if done { - postprocess_queued_job(flow_job.schedule_path, &w_id, flow, db).await?; - } - - Ok(()) -} - +#[instrument(level = "trace", skip_all)] pub async fn postprocess_queued_job( schedule_path: Option, + script_path: Option, w_id: &str, job_id: Uuid, db: &DB, ) -> crate::error::Result<()> { let _ = delete_job(db, w_id, job_id).await?; - schedule_again_if_scheduled(schedule_path, &w_id, db).await?; + schedule_again_if_scheduled(schedule_path, script_path, &w_id, db).await?; Ok(()) } +#[instrument(level = "trace", skip_all)] pub async fn schedule_again_if_scheduled( schedule_path: Option, + script_path: Option, w_id: &str, db: &DB, ) -> crate::error::Result<()> { @@ -1406,7 +1279,8 @@ pub async fn schedule_again_if_scheduled( let schedule = get_schedule_opt(&mut tx, &w_id, &schedule_path) .await? .unwrap(); - if schedule.enabled { + if schedule.enabled && script_path.is_some() && script_path.unwrap() == schedule.script_path + { tx = crate::schedule::push_scheduled_job(tx, schedule).await?; } tx.commit().await?; @@ -1414,167 +1288,6 @@ pub async fn schedule_again_if_scheduled( Ok(()) } -pub async fn handle_flow( - job: &QueuedJob, - db: &sqlx::Pool, - last_result: Option>, -) -> anyhow::Result<()> { - let value = job - .raw_flow - .as_ref() - .ok_or_else(|| Error::InternalErr(format!("requiring a raw flow value")))? - .to_owned(); - let flow = serde_json::from_value::(value.to_owned())?; - push_next_flow_job(job, flow, db, last_result).await?; - Ok(()) -} - -async fn transform_input( - flow_args: &Option, - last_result: Option>, - input_transform: &HashMap, - workspace: &str, - token: &str, - steps: Vec, -) -> anyhow::Result>> { - let mut mapped = serde_json::Map::new(); - - for (key, val) in input_transform.into_iter() { - match val { - InputTransform::Static { value } => { - mapped.insert(key.to_string(), value.to_owned()); - () - } - _ => (), - }; - } - - for (key, val) in input_transform.into_iter() { - match val { - InputTransform::Static { value: _ } => (), - InputTransform::Javascript { expr } => { - let previous_result = - serde_json::Value::Object(last_result.clone().unwrap_or_else(|| Map::new())); - let flow_input = flow_args.clone().unwrap_or_else(|| json!({})); - let v = eval_timeout( - expr.to_string(), - vec![ - ("params".to_string(), serde_json::json!(mapped)), - ("previous_result".to_string(), previous_result), - ("flow_input".to_string(), flow_input), - ], - workspace, - token, - steps.clone(), - ) - .await - .map_err(|e| { - Error::ExecutionErr(format!( - "Error during isolated evaluation of expression `{expr}`:\n{e}" - )) - })?; - mapped.insert(key.to_string(), v); - () - } - _ => Err(error::Error::BadRequest(format!( - "impossible to handle unknown input transform" - )))?, - } - } - - Ok(Some(mapped)) -} - -async fn push_next_flow_job( - job: &QueuedJob, - flow: FlowValue, - db: &sqlx::Pool, - last_result: Option>, -) -> anyhow::Result<()> { - let flow_status_json = job - .flow_status - .as_ref() - .ok_or_else(|| Error::InternalErr(format!("not found status for flow job {:?}", job.id)))?; - let status = serde_json::from_value::(flow_status_json.to_owned())?; - let i = status.step as usize; - - if flow.modules.len() > i { - let module = &flow.modules[i]; - let mut tx = db.begin().await?; - let job_payload = match &module.value { - FlowModuleValue::Script { path: script_path } => { - script_path_to_payload(script_path, &mut tx, &job.workspace_id).await? - } - FlowModuleValue::RawScript(raw_code) => JobPayload::Code(raw_code.clone()), - a @ _ => { - tracing::info!("Unrecognized module values {:?}", a); - Err(Error::BadRequest(format!( - "Unrecognized module values {:?}", - a - )))? - } - }; - - let token = create_token_for_owner( - &db, - &job.workspace_id, - &job.permissioned_as, - crate::users::NewToken { - label: Some("transform-input".to_string()), - expiration: Some(chrono::Utc::now() + chrono::Duration::seconds(10)), - }, - &job.created_by, - ) - .await?; - - let args = transform_input( - &job.args, - last_result, - &module.input_transform, - &job.workspace_id, - &token, - status - .modules - .into_iter() - .map(|x| match x { - FlowStatusModule::Success { job } => job.to_string(), - _ => "invalid step status".to_string(), - }) - .collect(), - ) - .await?; //job.args - let (uuid, mut tx) = push( - tx, - &job.workspace_id, - job_payload, - args, - &job.created_by, - job.permissioned_as.to_owned(), - None, - None, - Some(job.id), - true, - ) - .await?; - - sqlx::query(&format!( - "UPDATE queue - SET - flow_status = jsonb_set(flow_status, '{{modules, {}}}', $1) - WHERE id = $2", - i - )) - .bind(serde_json::json!(FlowStatusModule::WaitingForExecutor { - job: uuid - })) - .bind(job.parent_job) - .execute(&mut tx) - .await?; - tx.commit().await?; - } - Ok(()) -} - pub async fn pull(db: &DB) -> Result, crate::Error> { let now = chrono::Utc::now(); @@ -1599,6 +1312,7 @@ pub async fn pull(db: &DB) -> Result, crate::Error> { Ok(job) } +#[instrument(level = "trace", skip_all)] pub async fn delete_job(db: &DB, w_id: &str, job_id: Uuid) -> Result<(), crate::Error> { let job_removed = sqlx::query_scalar!( "DELETE FROM queue WHERE workspace_id = $1 AND id = $2 RETURNING 1", @@ -1606,7 +1320,8 @@ pub async fn delete_job(db: &DB, w_id: &str, job_id: Uuid) -> Result<(), crate:: job_id ) .fetch_one(db) - .await? + .await + .map_err(|e| Error::InternalErr(format!("Error during deletion of job {job_id}: {e}")))? .unwrap_or(0) == 1; tracing::debug!("Job {job_id} deletion was achieved with success: {job_removed}"); diff --git a/backend/src/js_eval.rs b/backend/src/js_eval.rs index 751c402155..7e61832907 100644 --- a/backend/src/js_eval.rs +++ b/backend/src/js_eval.rs @@ -21,27 +21,33 @@ use tokio::time::timeout; use crate::client; use crate::error::Error; +pub struct EvalCreds { + pub workspace: String, + pub token: String, +} + pub async fn eval_timeout( expr: String, env: Vec<(String, serde_json::Value)>, - workspace: &str, - token: &str, + creds: Option, steps: Vec, ) -> anyhow::Result { let expr2 = expr.clone(); let (sender, mut receiver) = oneshot::channel::(); - let (workspace, token) = (workspace.to_string().clone(), token.to_string().clone()); timeout( std::time::Duration::from_millis(2000), tokio::task::spawn_blocking(move || { + let mut ops = vec![]; - let mut ops = vec![ - // An op for summing an array of numbers - // The op-layer automatically deserializes inputs - // and serializes the returned Result & value - op_variable::decl(), - op_resource::decl(), - ]; + if creds.is_some() { + ops.extend([ + // An op for summing an array of numbers + // The op-layer automatically deserializes inputs + // and serializes the returned Result & value + op_variable::decl(), + op_resource::decl(), + ]) + } if !steps.is_empty() { ops.push(op_get_result::decl()) @@ -72,8 +78,7 @@ pub async fn eval_timeout( .into_iter() .fold(expr, replace_with_await); - let r = - runtime.block_on(eval(&mut js_runtime, &expr, env, &workspace, &token, steps))?; + let r = runtime.block_on(eval(&mut js_runtime, &expr, env, creds, steps))?; Ok(r) as anyhow::Result }), @@ -123,8 +128,7 @@ async fn eval( context: &mut JsRuntime, expr: &str, env: Vec<(String, serde_json::Value)>, - workspace: &str, - token: &str, + creds: Option, steps: Vec, ) -> anyhow::Result { let expr = expr.trim(); @@ -135,9 +139,10 @@ async fn eval( .join("\n"), expr.split(SPLIT_PAT).last().unwrap_or_else(|| "") ); - let steps_code = if !steps.is_empty() { - format!( - r#" + let (steps_code, api_code) = if let Some(EvalCreds { workspace, token }) = creds { + let steps_code = if !steps.is_empty() { + format!( + r#" let steps = [{}]; async function step(n) {{ if (n == -1) {{ @@ -151,14 +156,14 @@ async function step(n) {{ let id = steps[n]; return await Deno.core.opAsync("op_get_result", [workspace, id, token, base_url]); }}"#, - steps.into_iter().map(|x| format!("\"{x}\"")).join(",") - ) - } else { - "".to_string() - }; + steps.into_iter().map(|x| format!("\"{x}\"")).join(",") + ) + } else { + String::new() + }; - let code = format!( - r#" + let api_code = format!( + r#" let workspace = "{workspace}"; let base_url = "{}"; async function variable(path) {{ @@ -169,14 +174,24 @@ async function resource(path) {{ let token = "{token}"; return await Deno.core.opAsync("op_resource", [workspace, path, token, base_url]); }} + "#, + std::env::var("BASE_INTERNAL_URL") + .unwrap_or_else(|_| "http://missing-base-url".to_string()), + ); + (steps_code, api_code) + } else { + (String::new(), String::new()) + }; + + let code = format!( + r#" +{api_code} {} {steps_code} (async () => {{ {expr} }})() "#, - std::env::var("BASE_INTERNAL_URL") - .unwrap_or_else(|_| "http://missing-base-url".to_string()), env.into_iter() .map(|(a, b)| format!( "let {a} = {};\n", @@ -259,7 +274,7 @@ mod tests { let code = "value.test + params.test"; let mut runtime = JsRuntime::new(RuntimeOptions::default()); - let res = eval(&mut runtime, code, env, "workspace", "token", vec![]).await?; + let res = eval(&mut runtime, code, env, None, vec![]).await?; assert_eq!(res, json!(4)); Ok(()) } @@ -272,7 +287,7 @@ mod tests { multiline template`"; let mut runtime = JsRuntime::new(RuntimeOptions::default()); - let res = eval(&mut runtime, code, env, "workspace", "token", vec![]).await?; + let res = eval(&mut runtime, code, env, None, vec![]).await?; assert_eq!(res, json!("my 5\nmultiline template")); Ok(()) } @@ -285,7 +300,7 @@ multiline template`"; ]; let code = r#"variable("test")"#; - let res = eval_timeout(code.to_string(), env, "workspace", "token", vec![]).await?; + let res = eval_timeout(code.to_string(), env, None, vec![]).await?; assert_eq!(res, json!("test")); Ok(()) } diff --git a/backend/src/lib.rs b/backend/src/lib.rs index 403998e46d..149a1c5b7d 100644 --- a/backend/src/lib.rs +++ b/backend/src/lib.rs @@ -9,15 +9,13 @@ use argon2::Argon2; use axum::{handler::Handler, middleware::from_extractor, routing::get, Extension, Router}; use db::DB; use git_version::git_version; -use hyper::Response; use slack_http_verifier::SlackVerifier; use std::{net::SocketAddr, sync::Arc}; use tokio::sync::Mutex; use tower::ServiceBuilder; use tower_cookies::CookieManagerLayer; -use tower_http::trace::{MakeSpan, OnResponse, TraceLayer}; -use tracing::{field, Span}; -use tracing_subscriber::{filter::filter_fn, prelude::*, EnvFilter}; +use tower_http::trace::TraceLayer; + extern crate magic_crypt; extern crate dotenv; @@ -38,85 +36,31 @@ mod resources; mod schedule; mod scripts; mod static_assets; +mod tracing_init; mod users; mod utils; mod variables; mod worker; +mod worker_flow; mod worker_ping; mod workspaces; use error::Error; pub use crate::email::EmailSender; -use crate::{db::UserDB, error::to_anyhow, oauth2::build_oauth_clients, utils::rd_string}; +use crate::{ + db::UserDB, + error::to_anyhow, + oauth2::build_oauth_clients, + tracing_init::{MyMakeSpan, MyOnResponse}, + utils::rd_string, +}; const GIT_VERSION: &str = git_version!(args = ["--tag", "--always"], fallback = "unknown-version"); pub const DEFAULT_NUM_WORKERS: usize = 3; pub const DEFAULT_TIMEOUT: i32 = 300; pub const DEFAULT_SLEEP_QUEUE: u64 = 50; -#[derive(Clone)] -struct MyOnResponse {} - -impl OnResponse for MyOnResponse { - fn on_response( - self, - response: &Response, - latency: std::time::Duration, - _span: &tracing::Span, - ) { - tracing::info!( - latency = %latency.as_millis(), - status = ?response.status(), - "finished processed request") - } -} - -#[derive(Clone)] -struct MyMakeSpan {} - -impl MakeSpan for MyMakeSpan { - fn make_span(&mut self, request: &hyper::Request) -> Span { - tracing::info_span!( - "request", - method = %request.method(), - uri = %request.uri(), - version = ?request.version(), - username = field::Empty, - ) - } -} - -pub async fn initialize_tracing() -> anyhow::Result<()> { - //let log_level = if std::env::var("RUST_LOG").map(|x| &x == "debug") - let ts_base = tracing_subscriber::registry() - .with( - EnvFilter::from_default_env() - //.add_directive("windmill".parse()?) - .add_directive("runtime=trace".parse()?) - .add_directive("tokio=trace".parse()?), - ) - .with( - tracing_subscriber::fmt::layer() - .json() - .flatten_event(true) - .with_span_list(false) - .with_current_span(true) - .with_filter(filter_fn(|meta| meta.target().starts_with("windmill"))), - ); - - if std::env::var("TOKIO_CONSOLE") - .map(|x| x == "true") - .unwrap_or(false) - { - let console_layer = console_subscriber::spawn(); - ts_base.with(console_layer).init(); - } else { - ts_base.init(); - } - Ok(()) -} - pub async fn migrate_db(db: &DB) -> anyhow::Result<()> { let app_password = std::env::var("APP_USER_PASSWORD").unwrap_or_else(|_| "changeme".to_owned()); @@ -131,6 +75,10 @@ pub async fn connect_db() -> anyhow::Result { Ok(db::connect(&database_url).await?) } +pub async fn initialize_tracing() -> anyhow::Result<()> { + tracing_init::initialize_tracing().await +} + #[derive(Clone)] struct BaseUrl(String); diff --git a/backend/src/scripts.rs b/backend/src/scripts.rs index e0026ff391..5d7e3c888f 100644 --- a/backend/src/scripts.rs +++ b/backend/src/scripts.rs @@ -128,6 +128,7 @@ pub struct Script { pub lock: Option, pub lock_error_logs: Option, pub language: ScriptLang, + pub trigger_reco_interval: Option, } #[derive(Serialize, Deserialize, sqlx::Type, Debug)] @@ -154,6 +155,7 @@ pub struct NewScript { pub is_template: Option, pub lock: Option>, pub language: ScriptLang, + pub trigger_reco_interval: Option, } #[derive(Deserialize)] @@ -198,6 +200,7 @@ async fn list_scripts( "null as lock", "CASE WHEN lock_error_logs IS NOT NULL THEN 'error' ELSE null END as lock_error_logs", "language", + "trigger_reco_interval", ]) .order_by("created_at", lq.order_desc.unwrap_or(true)) .and_where("workspace_id = ? OR workspace_id = 'starter'".bind(&w_id)) @@ -404,8 +407,8 @@ async fn create_script( //::text::json is to ensure we use serde_json with preserve order sqlx::query!( "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, content, \ - created_by, schema, is_template, extra_perms, lock, language) VALUES \ - ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13)", + created_by, schema, is_template, extra_perms, lock, language, trigger_reco_interval) VALUES \ + ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14)", &w_id, &hash.0, ns.path, @@ -418,7 +421,8 @@ async fn create_script( ns.is_template.unwrap_or(false), extra_perms, lock, - ns.language: ScriptLang + ns.language: ScriptLang, + ns.trigger_reco_interval, ) .execute(&mut tx) .await?; diff --git a/backend/src/tracing_init.rs b/backend/src/tracing_init.rs new file mode 100644 index 0000000000..aca7089ad7 --- /dev/null +++ b/backend/src/tracing_init.rs @@ -0,0 +1,95 @@ +use ::tracing::{field, Metadata, Span}; +use ::tracing_subscriber::{ + filter::filter_fn, + fmt::{format, Layer}, + prelude::*, + EnvFilter, +}; +use hyper::Response; +use tower_http::trace::{MakeSpan, OnResponse}; + +#[derive(Clone)] +pub struct MyOnResponse {} + +impl OnResponse for MyOnResponse { + fn on_response( + self, + response: &Response, + latency: std::time::Duration, + _span: &tracing::Span, + ) { + tracing::info!( + latency = %latency.as_millis(), + status = ?response.status(), + "finished processed request") + } +} + +#[derive(Clone)] +pub struct MyMakeSpan {} + +impl MakeSpan for MyMakeSpan { + fn make_span(&mut self, request: &hyper::Request) -> Span { + tracing::info_span!( + "request", + method = %request.method(), + uri = %request.uri(), + version = ?request.version(), + username = field::Empty, + ) + } +} + +fn json_layer() -> Layer> { + tracing_subscriber::fmt::layer() + .json() + .flatten_event(true) + .with_span_list(false) + .with_current_span(true) +} + +fn compact_layer() -> Layer> { + tracing_subscriber::fmt::layer().compact() +} + +fn filter_metadata(meta: &Metadata) -> bool { + meta.target().starts_with("windmill") +} + +pub async fn initialize_tracing() -> anyhow::Result<()> { + let tokio_console = std::env::var("TOKIO_CONSOLE") + .map(|x| x == "true") + .unwrap_or(false); + let json_fmt = std::env::var("JSON_FMT") + .map(|x| x == "true") + .unwrap_or(false); + + let env_filter = EnvFilter::from_default_env(); + + let nenv_filter = if tokio_console { + env_filter + .add_directive("runtime=trace".parse()?) + .add_directive("tokio=trace".parse()?) + } else { + env_filter + }; + let ts_base = tracing_subscriber::registry().with(nenv_filter); + + match (json_fmt, tokio_console) { + (true, true) => ts_base + .with(json_layer().with_filter(filter_fn(filter_metadata))) + .with(console_subscriber::spawn()) + .init(), + (true, false) => ts_base + .with(json_layer().with_filter(filter_fn(filter_metadata))) + .init(), + (false, true) => ts_base + .with(compact_layer().with_filter(filter_fn(filter_metadata))) + .with(console_subscriber::spawn()) + .init(), + _ => ts_base + .with(compact_layer().with_filter(filter_fn(filter_metadata))) + .init(), + } + Ok(()) +} diff --git a/backend/src/variables.rs b/backend/src/variables.rs index f34e09f68e..6ee61e0201 100644 --- a/backend/src/variables.rs +++ b/backend/src/variables.rs @@ -86,7 +86,8 @@ pub fn get_reserved_variables( permissioned_as: &str, path: Option, flow_path: Option, -) -> [ContextualVariable; 8] { + schedule_path: Option, +) -> [ContextualVariable; 9] { [ ContextualVariable { name: "WM_WORKSPACE".to_string(), @@ -123,6 +124,11 @@ pub fn get_reserved_variables( value: flow_path.unwrap_or_else(|| "".to_string()), description: "Path of the encapsulating flow if the job is a flow step".to_string() }, + ContextualVariable { + name: "WM_SCHEDULE_PATH".to_string(), + value: schedule_path.unwrap_or_else(|| "".to_string()), + description: "Path of the schedule if the job of the step or encapsulating step has been triggered by a schedule".to_string() + }, ContextualVariable { name: "WM_PERMISSIONED_AS".to_string(), value: permissioned_as.to_string(), @@ -147,6 +153,7 @@ async fn list_contextual_variables( format!("u/{username}").as_str(), Some("u/user/script_path".to_string()), Some("u/user/encapsulating_flow_path".to_string()), + Some("u/user/triggering_flow_path".to_string()), ) .to_vec(), )) diff --git a/backend/src/worker.rs b/backend/src/worker.rs index 1015e3e3fa..5206946275 100644 --- a/backend/src/worker.rs +++ b/backend/src/worker.rs @@ -20,14 +20,16 @@ use crate::{ db::DB, error::Error, jobs::{ - add_completed_job, add_completed_job_error, handle_flow, postprocess_queued_job, pull, - update_flow_status_after_job_completion, update_flow_status_in_progress, JobKind, + add_completed_job, add_completed_job_error, postprocess_queued_job, pull, JobKind, QueuedJob, }, parser::{self, Typ}, scripts::{ScriptHash, ScriptLang}, users::{create_token_for_owner, get_email_from_username}, variables, + worker_flow::{ + handle_flow, update_flow_status_after_job_completion, update_flow_status_in_progress, + }, }; use serde_json::{json, Map, Value}; @@ -131,9 +133,14 @@ pub async fn run_worker( ) .await; - let _ = - postprocess_queued_job(job2.schedule_path, &job2.workspace_id, job2.id, db) - .await; + let _ = postprocess_queued_job( + job2.schedule_path, + job2.script_path, + &job2.workspace_id, + job2.id, + db, + ) + .await; tracing::error!(job_id = %job2.id, "Error handling job: {err_string}"); }; } @@ -217,7 +224,7 @@ async fn handle_queued_job( match execution { Ok(r) => { - add_completed_job(db, &job, true, r.result.clone(), logs).await?; + add_completed_job(db, &job, true, false, r.result.clone(), logs).await?; if job.is_flow_step { update_flow_status_after_job_completion(db, &job, true, r.result).await?; } @@ -231,7 +238,8 @@ async fn handle_queued_job( } }; - let _ = postprocess_queued_job(job.schedule_path, &w_id, job_id, db).await; + let _ = + postprocess_queued_job(job.schedule_path, job.script_path, &w_id, job_id, db).await; } } Ok(()) @@ -485,8 +493,8 @@ async fn handle_nondep_job( } else { None }; - let ser_args = serde_json::to_string(&args) - .map_err(|e| Error::ExecutionErr(e.to_string()))?; + let ser_args = + serde_json::to_string(&args).map_err(|e| Error::ExecutionErr(e.to_string()))?; write_file(job_dir, "args.json", &ser_args).await?; let wrapper_content: String = format!( @@ -596,8 +604,8 @@ print(res_json) } else { None }; - let ser_args = serde_json::to_string(&args) - .map_err(|e| Error::ExecutionErr(e.to_string()))?; + let ser_args = + serde_json::to_string(&args).map_err(|e| Error::ExecutionErr(e.to_string()))?; write_file(job_dir, "args.json", &ser_args).await?; let spread = sig.args.into_iter().map(|x| x.name).join(","); @@ -630,7 +638,9 @@ run(); write_file(job_dir, "main.ts", &wrapper_content).await?; tx.commit().await?; - let reserved_variables = get_reserved_variables(job, token, db).await?; + + let mut reserved_variables = get_reserved_variables(job, token, db).await?; + reserved_variables.insert("RUST_LOG".to_string(), "info".to_string()); if !disable_nuser { let _ = write_file( @@ -767,6 +777,7 @@ async fn get_reserved_variables( &job.permissioned_as, job.script_path.clone(), flow_path, + job.schedule_path.clone(), ); Ok(variables .into_iter() diff --git a/backend/src/worker_flow.rs b/backend/src/worker_flow.rs new file mode 100644 index 0000000000..3ce1e3bfbc --- /dev/null +++ b/backend/src/worker_flow.rs @@ -0,0 +1,606 @@ +use std::collections::HashMap; + +use crate::flow::{FlowModuleValue, FlowValue, InputTransform}; +use crate::jobs::{ + add_completed_job, add_completed_job_error, get_queued_job, postprocess_queued_job, push, + script_path_to_payload, JobPayload, +}; +use crate::js_eval::{eval_timeout, EvalCreds}; +use crate::users::create_token_for_owner; +use crate::{ + db::DB, + error::{self, Error}, + jobs::QueuedJob, +}; +use async_recursion::async_recursion; +use serde::{Deserialize, Serialize}; +use serde_json::json; +use serde_json::{Map, Value}; +use tracing::instrument; +use uuid::Uuid; + +#[derive(Serialize, Deserialize, Debug)] +pub struct FlowStatus { + pub step: i32, + pub modules: Vec, + pub failure_module: FlowStatusModule, +} + +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct Iterator { + pub index: u8, + pub itered: Vec, + pub args: Map, +} + +#[derive(Serialize, Deserialize, Debug, Clone)] +#[serde(tag = "type")] +pub enum FlowStatusModule { + WaitingForPriorSteps, + WaitingForEvent { + event: String, + }, + WaitingForExecutor { + job: Uuid, + }, + InProgress { + job: Uuid, + iterator: Option, + forloop_jobs: Option>, + }, + Success { + job: Uuid, + forloop_jobs: Option>, + }, + Failure { + job: Uuid, + forloop_jobs: Option>, + }, +} + +#[async_recursion] +#[instrument(level = "trace", skip_all)] +pub async fn update_flow_status_after_job_completion( + db: &DB, + job: &QueuedJob, + success: bool, + result: Option>, +) -> error::Result<()> { + tracing::info!("HANDLE FLOW: {job:?} {success} {result:?}"); + + let mut tx = db.begin().await?; + + let w_id = &job.workspace_id; + + let flow = job + .parent_job + .ok_or_else(|| Error::InternalErr(format!("expected parent job")))?; + + let old_status_json = sqlx::query_scalar!( + "SELECT flow_status FROM queue WHERE id = $1 AND workspace_id = $2", + flow, + w_id + ) + .fetch_one(&mut tx) + .await? + .ok_or_else(|| Error::InternalErr(format!("requiring a previous status")))?; + + let old_status = serde_json::from_value::(old_status_json) + .ok() + .ok_or_else(|| { + Error::InternalErr(format!("requiring status to be parsabled as FlowStatus")) + })?; + + 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.to_owned() as usize) < itered.len() - 1 && success => { + (old_status.step, module_status.clone()) + } + module_status @ _ => { + let forloop_jobs = match module_status { + FlowStatusModule::InProgress { + forloop_jobs: Some(jobs), + .. + } => Some(jobs.clone()), + _ => None, + }; + let new_status = if success { + FlowStatusModule::Success { + job: job.id, + forloop_jobs, + } + } else { + FlowStatusModule::Failure { + job: job.id, + forloop_jobs, + } + }; + (old_status.step + 1, new_status) + } + }; + + let last_step = step_counter as usize == old_status.modules.len(); + + tracing::debug!( + "old status: {:#?}\n{:#?}\n{last_step}", + old_status, + new_status + ); + + let prev_step = old_status.step; + let (stop_early_expr, skip_if_stop_early) = sqlx::query_as::<_, (Option, Option)>(&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", + )) + .bind(serde_json::json!(new_status)) + .bind(serde_json::json!(step_counter)) + .bind(flow) + .fetch_one(&mut tx) + .await?; + + tracing::info!("UPDATE: {:?}", new_status); + + let flow_job = get_queued_job(flow, w_id, &mut tx) + .await? + .ok_or_else(|| Error::InternalErr(format!("requiring flow to be in the queue")))?; + tx.commit().await?; + + let stop_early = success + && if let Some(expr) = stop_early_expr { + compute_stop_early(expr, result.clone()).await? + } else { + false + }; + + let done = if !success || last_step || stop_early { + let result = match new_status { + FlowStatusModule::Success { + forloop_jobs: Some(jobs), + .. + } => { + let mut results = Vec::new(); + for job in jobs { + let result = sqlx::query_scalar!( + "SELECT result FROM completed_job WHERE id = $1 AND workspace_id = $2", + job, + w_id, + ) + .fetch_optional(db) + .await? + .flatten(); + results.push(result.clone()); + } + let mut results_map = serde_json::Map::new(); + results_map.insert("res1".to_string(), serde_json::json!(results)); + Some(results_map) + } + _ => result.clone(), + }; + + let logs = if stop_early { + "Flow job stopped early".to_string() + } else { + "Flow job completed".to_string() + }; + tracing::debug!("{skip_if_stop_early:?}"); + add_completed_job( + db, + &flow_job, + success, + stop_early && skip_if_stop_early.unwrap_or(false), + result.clone(), + logs, + ) + .await?; + true + } else { + match handle_flow(&flow_job, db, result.clone()).await { + Err(err) => { + let _ = add_completed_job_error( + db, + &flow_job, + "Unexpected error during flow chaining:\n".to_string(), + err, + ) + .await; + true + } + Ok(_) => false, + } + }; + + if done { + postprocess_queued_job( + flow_job.schedule_path.clone(), + flow_job.script_path.clone(), + &w_id, + flow, + db, + ) + .await?; + + if flow_job.parent_job.is_some() { + return Ok( + update_flow_status_after_job_completion(db, &flow_job, success, result).await?, + ); + } + } + + Ok(()) +} + +async fn compute_stop_early( + expr: String, + result: Option>, +) -> error::Result { + let result = serde_json::Value::Object(result.clone().unwrap_or_else(|| Map::new())); + match eval_timeout(expr, [("result".to_string(), result)].into(), None, vec![]).await? { + serde_json::Value::Bool(true) => Ok(true), + serde_json::Value::Bool(false) => Ok(false), + a @ _ => Err(Error::ExecutionErr(format!( + "Expected a boolean value, found: {a:?}" + ))), + } +} + +pub fn init_flow_status(f: &FlowValue) -> FlowStatus { + FlowStatus { + step: 0, + modules: (0..f.modules.len()) + .map(|_| FlowStatusModule::WaitingForPriorSteps) + .collect(), + failure_module: FlowStatusModule::WaitingForPriorSteps, + } +} + +pub async fn update_flow_status_in_progress( + db: &DB, + w_id: &str, + flow: Uuid, + job_in_progress: Uuid, +) -> error::Result<()> { + let step = get_step_of_flow_status(db, flow).await?; + sqlx::query(&format!( + "UPDATE queue + SET flow_status = jsonb_set(flow_status, '{{modules, {}}}', $1) + WHERE id = $2 AND workspace_id = $3", + step + )) + .bind(serde_json::json!(FlowStatusModule::InProgress { + job: job_in_progress, + iterator: None, + forloop_jobs: None, + })) + .bind(flow) + .bind(w_id) + .execute(db) + .await?; + Ok(()) +} + +#[instrument(level = "trace", skip_all)] +pub async fn get_step_of_flow_status(db: &DB, id: Uuid) -> error::Result { + let r = sqlx::query_scalar!( + "SELECT (flow_status->'step')::integer FROM queue WHERE id = $1", + id + ) + .fetch_one(db) + .await? + .ok_or_else(|| Error::InternalErr(format!("not found step")))?; + Ok(r) +} + +#[instrument(level = "trace", skip_all)] +async fn transform_input( + flow_args: &Option, + last_result: Option>, + input_transform: &HashMap, + workspace: &str, + token: &str, + steps: Vec, +) -> anyhow::Result>> { + let mut mapped = serde_json::Map::new(); + + for (key, val) in input_transform.into_iter() { + match val { + InputTransform::Static { value } => { + mapped.insert(key.to_string(), value.to_owned()); + () + } + _ => (), + }; + } + + for (key, val) in input_transform.into_iter() { + match val { + InputTransform::Static { value: _ } => (), + InputTransform::Javascript { expr } => { + let previous_result = + serde_json::Value::Object(last_result.clone().unwrap_or_else(|| Map::new())); + let flow_input = flow_args.clone().unwrap_or_else(|| json!({})); + let v = eval_timeout( + expr.to_string(), + vec![ + ("params".to_string(), serde_json::json!(mapped)), + ("previous_result".to_string(), previous_result), + ("flow_input".to_string(), flow_input), + ], + Some(EvalCreds { + workspace: workspace.to_string(), + token: token.to_string(), + }), + steps.clone(), + ) + .await + .map_err(|e| { + Error::ExecutionErr(format!( + "Error during isolated evaluation of expression `{expr}`:\n{e}" + )) + })?; + mapped.insert(key.to_string(), v); + () + } + } + } + + Ok(Some(mapped)) +} + +#[instrument(level = "trace", skip_all)] +pub async fn handle_flow( + flow_job: &QueuedJob, + db: &sqlx::Pool, + last_result: Option>, +) -> anyhow::Result<()> { + let value = flow_job + .raw_flow + .as_ref() + .ok_or_else(|| Error::InternalErr(format!("requiring a raw flow value")))? + .to_owned(); + let flow = serde_json::from_value::(value.to_owned())?; + push_next_flow_job( + flow_job, + flow, + flow_job.schedule_path.clone(), + db, + last_result, + ) + .await?; + Ok(()) +} + +#[instrument(level = "trace", skip_all)] +async fn push_next_flow_job( + flow_job: &QueuedJob, + flow: FlowValue, + schedule_path: Option, + db: &sqlx::Pool, + last_result: Option>, +) -> anyhow::Result<()> { + let flow_status_json = flow_job.flow_status.as_ref().ok_or_else(|| { + Error::InternalErr(format!("not found status for flow job {:?}", flow_job.id)) + })?; + let status = serde_json::from_value::(flow_status_json.to_owned())?; + let i = status.step as usize; + + if flow.modules.len() > i { + let module = &flow.modules[i]; + let mut tx = db.begin().await?; + let job_payload = match &module.value { + FlowModuleValue::Script { + path: script_path + } => script_path_to_payload(script_path, &mut tx, &flow_job.workspace_id).await?, + FlowModuleValue::RawScript(raw_code) => { + let mut raw_code = raw_code.clone(); + if raw_code.path.is_none() { + raw_code.path = Some(format!( + "{}/{i}", + flow_job + .script_path + .as_ref() + .unwrap_or(&"NO_FLOW_PATH".to_owned()) + )); + } + JobPayload::Code(raw_code) + } + FlowModuleValue::ForloopFlow { iterator: _, value } => JobPayload::RawFlow { + value: *(*value).to_owned(), + path: Some(format!( + "{}/{i}", + flow_job + .script_path + .as_ref() + .unwrap_or(&"NO_FLOW_PATH".to_owned()) + )), + }, + a @ _ => { + tracing::info!("Unrecognized module values {:?}", a); + Err(Error::BadRequest(format!( + "Unrecognized module values {:?}", + a + )))? + } + }; + + let token = create_token_for_owner( + &db, + &flow_job.workspace_id, + &flow_job.permissioned_as, + crate::users::NewToken { + label: Some("transform-input".to_string()), + expiration: Some(chrono::Utc::now() + chrono::Duration::seconds(10)), + }, + &flow_job.created_by, + ) + .await?; + let mut input_transform = module.input_transform.clone(); + + tracing::debug!( + "PUSH: module: {:#?}, status: {:#?}", + module.value, + status.modules[i] + ); + let (forloop_args, forloop_iterator) = match &module.value { + FlowModuleValue::ForloopFlow { iterator, .. } => { + let (index_forloop, itered, args, forloop_jobs) = match &status.modules[i] { + FlowStatusModule::WaitingForPriorSteps { .. } => { + let itered = match iterator { + InputTransform::Static { value } => value.clone(), + InputTransform::Javascript { expr } => { + let result = serde_json::Value::Object( + last_result.clone().unwrap_or_else(|| Map::new()), + ); + eval_timeout( + expr.to_string(), + [("result".to_string(), result)].into(), + None, + vec![], + ) + .await? + } + }; + input_transform.insert( + "_index".to_string(), + InputTransform::Static { + value: serde_json::Value::Number(serde_json::Number::from(0)), + }, + ); + input_transform.insert( + "_value".to_string(), + InputTransform::Static { + value: itered[0].clone(), + }, + ); + + match itered { + serde_json::Value::Array(arr) => (0 as u8, arr, None, vec![]), + a @ _ => Err(Error::BadRequest(format!( + "Expected an array value, found: {:?}", + a + )))?, + } + } + FlowStatusModule::InProgress { + iterator: + Some(Iterator { + index, + itered, + args, + }), + forloop_jobs, + .. + } => { + let mut args = args.clone(); + let nindex = index.to_owned() + 1; + args.insert( + "_index".to_string(), + serde_json::Value::Number(serde_json::Number::from(nindex.to_owned())), + ); + args.insert( + "_value".to_string(), + itered[nindex.to_owned() as usize].clone(), + ); + ( + nindex, + itered.to_owned(), + Some(args), + forloop_jobs.to_owned().unwrap_or_else(Vec::new), + ) + } + a @ _ => Err(Error::BadRequest(format!( + "Unrecognized module status for ForloopFlow {:?}", + a + )))?, + }; + (args, Some((index_forloop, itered, forloop_jobs))) + } + _ => (None, None), + }; + + let args = if forloop_args.is_some() { + forloop_args.map(|x| x.to_owned()) + } else { + let steps = status + .modules + .into_iter() + .map(|x| match x { + FlowStatusModule::Success { + job, + forloop_jobs: _, + } => job.to_string(), + _ => "invalid step status".to_string(), + }) + .collect(); + + let transformed = transform_input( + &flow_job.args, + last_result.clone(), + &input_transform, + &flow_job.workspace_id, + &token, + steps, + ) + .await?; + + match (&flow_job.args, &module.value) { + (Some(Value::Object(m)), FlowModuleValue::ForloopFlow { .. }) => { + let mut args = transformed.unwrap_or_else(Map::new); + args.extend(m.to_owned()); + args.extend(last_result.unwrap()); + Some(args) + } + _ => transformed, + } + }; + + let (uuid, mut tx) = push( + tx, + &flow_job.workspace_id, + job_payload, + args.clone(), + &flow_job.created_by, + flow_job.permissioned_as.to_owned(), + None, + schedule_path, + Some(flow_job.id), + true, + ) + .await?; + + let new_status = if let Some((index, itered, mut forloop_jobs)) = forloop_iterator { + forloop_jobs.push(uuid.to_owned()); + serde_json::json!(FlowStatusModule::InProgress { + job: uuid, + iterator: Some(Iterator { + index: index, + itered: itered, + args: args.unwrap_or_else(|| Map::new()), + }), + forloop_jobs: Some(forloop_jobs), + }) + } else { + serde_json::json!(FlowStatusModule::WaitingForExecutor { job: uuid }) + }; + + sqlx::query(&format!( + "UPDATE queue + SET + flow_status = jsonb_set(flow_status, '{{modules, {}}}', $1) + WHERE id = $2", + i + )) + .bind(new_status) + .bind(flow_job.id) + .execute(&mut tx) + .await?; + tx.commit().await?; + } + Ok(()) +} diff --git a/deno-client/generate.sh b/deno-client/generate.sh index 49294ef9a6..101c53309b 100755 --- a/deno-client/generate.sh +++ b/deno-client/generate.sh @@ -3,4 +3,4 @@ set -e /usr/local/bin/docker-entrypoint.sh generate -i ../backend/openapi.yaml -g typescript --additional-properties platform=deno -o windmill-api sed -i 's/this\.type = "Job";//' windmill-api/models/Job.ts -sed -i '457 i \ if (mediaType === "text/plain") { return rawData }' windmill-api/models/ObjectSerializer.ts +sed -i '460 i \ if (mediaType === "text/plain") { return data }' windmill-api/models/ObjectSerializer.ts diff --git a/deno-client/mod.ts b/deno-client/mod.ts index c7bde01c74..c9af111c9f 100644 --- a/deno-client/mod.ts +++ b/deno-client/mod.ts @@ -52,11 +52,13 @@ export function getInternalStatePath(suffix?: string): string { const permissioned_as = Deno.env.get("WM_PERMISSIONED_AS") const flow_path = env_flow_path != undefined && env_flow_path != "" ? env_flow_path : 'NO_FLOW_PATH' const script_path = suffix ?? (env_job_path != undefined && env_job_path != "" ? env_job_path : 'NO_JOB_PATH') + const env_schedule_path = Deno.env.get("WM_SCHEDULE_PATH") + const schedule_path = env_flow_path != undefined && env_flow_path != "" ? `/${env_schedule_path}` : '' if (script_path.slice(script_path.length - 1) === '/') { throw Error(`The script path must not end with '/', give a name to your script!`) } - return `${permissioned_as}/${flow_path}/${script_path}` + return `${permissioned_as}/${flow_path}/${script_path}${schedule_path}` } /** diff --git a/frontend/src/lib/components/DisplayResult.svelte b/frontend/src/lib/components/DisplayResult.svelte index d001990415..97ce6d6e60 100644 --- a/frontend/src/lib/components/DisplayResult.svelte +++ b/frontend/src/lib/components/DisplayResult.svelte @@ -57,52 +57,58 @@ {@html github} -{#if Object.keys(result).length > 0}
- The result keys are: {Object.keys(result).join(', ')} -
-{/if} -{#if resultKind == 'table-col'} -
- {#each Object.keys(result) as col} -
-
- {col} -
- {#each result[col] as item} -
{item}
- {/each} -
- {/each} -
-{:else if resultKind == 'table-row'}
- - - {#each asListOfList(Object.values(result)[0]) as row} - - {#each row as v} - {v ?? ''} +{#if result} + {#if Object.keys(result).length > 0}
+ The result keys are: {Object.keys(result).join(', ')} +
+ {/if} + {#if resultKind == 'table-col'} +
+ {#each Object.keys(result) as col} +
+
+ {col} +
+ {#if Array.isArray(result[col])} + {#each result[col] as item} +
+ {typeof item === 'string' ? item : JSON.stringify(item)} +
{/each} - - {/each} - - -
-{:else if resultKind == 'png'} -
- Result is an image: png rendered -
-{:else if resultKind == 'file'} -
- Result is a file: Download + {/each} +
+ {:else if resultKind == 'table-row'}
-
-{:else} + + + {#each asListOfList(Object.values(result)[0]) as row} + + {#each row as v} + {v ?? ''} + {/each} + + {/each} + + +
+ {:else if resultKind == 'png'} +
+ Result is an image: png rendered +
+ {:else if resultKind == 'file'} +
+ Result is a file: Download +
+ {:else} + {/if} {/if} diff --git a/frontend/src/lib/components/FlowBuilder.svelte b/frontend/src/lib/components/FlowBuilder.svelte index 6808f6f26e..affc44c166 100644 --- a/frontend/src/lib/components/FlowBuilder.svelte +++ b/frontend/src/lib/components/FlowBuilder.svelte @@ -7,7 +7,8 @@ import { onMount } from 'svelte' import SvelteMarkdown from 'svelte-markdown' import FlowEditor from './FlowEditor.svelte' - import { flowStore, initFlow } from './flows/flowStore' + import { flowStore, initFlow, type FlowMode } from './flows/flowStore' + import { flowToMode } from './flows/utils' import Path from './Path.svelte' import Required from './Required.svelte' import ScriptSchema from './ScriptSchema.svelte' @@ -15,6 +16,8 @@ export let flow: Flow export let initialPath: string = '' + let mode: FlowMode + $: step = Number($page.url.searchParams.get('step')) || 1 async function loadSearchData() { @@ -27,27 +30,29 @@ } async function saveFlow(): Promise { + const newFlow = flowToMode($flowStore, mode) + if (initialPath === '') { await FlowService.createFlow({ workspace: $workspaceStore!, requestBody: { - path: flow.path, - summary: flow.summary, - description: flow.description ?? '', - value: flow.value, - schema: flow.schema + path: newFlow.path, + summary: newFlow.summary, + description: newFlow.description ?? '', + value: newFlow.value, + schema: newFlow.schema } }) } else { await FlowService.updateFlow({ workspace: $workspaceStore!, - path: flow.path, + path: newFlow.path, requestBody: { - path: flow.path, - summary: flow.summary, - description: flow.description ?? '', - value: flow.value, - schema: flow.schema + path: newFlow.path, + summary: newFlow.summary, + description: newFlow.description ?? '', + value: newFlow.value, + schema: newFlow.schema } }) } @@ -60,7 +65,7 @@ } flowStore.subscribe((flow: Flow) => { - $page.url.searchParams.set('state', btoa(JSON.stringify(flow))) + $page.url.searchParams.set('state', btoa(JSON.stringify(flowToMode(flow, mode)))) history.replaceState({}, '', $page.url) }) @@ -192,7 +197,7 @@
{:else if step === 2} - + {:else if step === 3} + import { FlowModuleValue } from '$lib/gen' + import { faPlus } from '@fortawesome/free-solid-svg-icons' import Icon from 'svelte-awesome' import FlowPreview from './FlowPreview.svelte' import CopyFirstStepSchema from './flows/CopyFirstStepSchema.svelte' - import { addModule, flowStore } from './flows/flowStore' + import { addModule, flowStore, type FlowMode } from './flows/flowStore' import ModuleStep from './ModuleStep.svelte' + import RadioButtonV2 from './RadioButtonV2.svelte' import SchemaEditor from './SchemaEditor.svelte' let args: Record = {} + export let mode: FlowMode = + $flowStore?.value.modules[1]?.value.type == FlowModuleValue.type.FORLOOPFLOW ? 'pull' : 'push' $: numberOfSteps = $flowStore?.value.modules.length - 1 @@ -15,6 +20,31 @@