From 40bfc486c767f3bc3cd2ea918f50daae073f2c4c Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Mon, 24 Oct 2022 14:48:04 +0200 Subject: [PATCH] add resume_id to job_resume + add identity module --- .../20221024110649_add_resume_id.down.sql | 2 + .../20221024110649_add_resume_id.up.sql | 3 + backend/sqlx-data.json | 146 ++++++++++-------- backend/src/flows.rs | 1 + backend/src/jobs.rs | 23 ++- backend/src/worker.rs | 32 ++-- backend/src/worker_flow.rs | 19 ++- .../[job]/[resume]/[hmac]@none.svelte | 81 ++++++++-- frontend/src/routes/run/[...run].svelte | 2 - openflow.openapi.yaml | 12 ++ 10 files changed, 218 insertions(+), 103 deletions(-) create mode 100644 backend/migrations/20221024110649_add_resume_id.down.sql create mode 100644 backend/migrations/20221024110649_add_resume_id.up.sql diff --git a/backend/migrations/20221024110649_add_resume_id.down.sql b/backend/migrations/20221024110649_add_resume_id.down.sql new file mode 100644 index 0000000000..f7bcbacdda --- /dev/null +++ b/backend/migrations/20221024110649_add_resume_id.down.sql @@ -0,0 +1,2 @@ +-- Add down migration script here +ALTER TABLE resume_job DROP resume_id; \ No newline at end of file diff --git a/backend/migrations/20221024110649_add_resume_id.up.sql b/backend/migrations/20221024110649_add_resume_id.up.sql new file mode 100644 index 0000000000..6718104c24 --- /dev/null +++ b/backend/migrations/20221024110649_add_resume_id.up.sql @@ -0,0 +1,3 @@ +-- Add up migration script here +ALTER TABLE resume_job ADD COLUMN resume_id INTEGER NOT NULL DEFAULT 0; +ALTER TYPE JOB_KIND ADD VALUE 'identity'; \ No newline at end of file diff --git a/backend/sqlx-data.json b/backend/sqlx-data.json index 2822499b54..9d9e167936 100644 --- a/backend/sqlx-data.json +++ b/backend/sqlx-data.json @@ -494,6 +494,23 @@ }, "query": "UPDATE queue SET logs = $1 WHERE id = $2" }, + "1f68e0b0e6dfa59d782a4abeab8aa06ad808fc7e8255951e1be2fc41263f9c50": { + "describe": { + "columns": [], + "nullable": [], + "parameters": { + "Left": [ + "Uuid", + "Int4", + "Uuid", + "Uuid", + "Jsonb", + "Varchar" + ] + } + }, + "query": "\n INSERT INTO resume_job\n (id, resume_id, job, flow, value, approver)\n VALUES ($1, $2, $3, $4, $5, $6)\n ON CONFLICT (id) DO NOTHING\n " + }, "1f93b533fa6fee0db4340445da3fac8e6773bc1db1f88cd60fd3c1e8c9781eb0": { "describe": { "columns": [], @@ -1091,6 +1108,38 @@ }, "query": "DELETE FROM usr_to_group WHERE group_ = $1 AND workspace_id = $2" }, + "541ebd3bac65431237cf3b882dfdcd61ca97c253d9754d05bba59fda89841067": { + "describe": { + "columns": [ + { + "name": "value", + "ordinal": 0, + "type_info": "Jsonb" + }, + { + "name": "approver", + "ordinal": 1, + "type_info": "Varchar" + }, + { + "name": "resume_id", + "ordinal": 2, + "type_info": "Int4" + } + ], + "nullable": [ + false, + true, + false + ], + "parameters": { + "Left": [ + "Uuid" + ] + } + }, + "query": "SELECT value, approver, resume_id FROM resume_job WHERE job = $1 ORDER BY created_at ASC" + }, "5445083864b2b092b012e894bff7630a1d7b9deb8d33e9f909061f351f96844e": { "describe": { "columns": [ @@ -1184,22 +1233,6 @@ }, "query": "SELECT * from resource_type WHERE name = $1 AND (workspace_id = $2 OR workspace_id = 'starter')" }, - "5a9adf5eec1d51060e6f5ecebea60b511189a22019d6c9249bb30a0cf4ae23dd": { - "describe": { - "columns": [], - "nullable": [], - "parameters": { - "Left": [ - "Uuid", - "Uuid", - "Uuid", - "Jsonb", - "Varchar" - ] - } - }, - "query": "\n INSERT INTO resume_job\n (id, job, flow, value, approver)\n VALUES ($1, $2, $3, $4, $5)\n ON CONFLICT (id) DO NOTHING\n " - }, "5b7a1d16d8109a65479ab33d411c60d14ea91d870fdff8606d7aa4ad39f0ba00": { "describe": { "columns": [ @@ -1591,7 +1624,8 @@ "flow", "dependencies", "flowpreview", - "script_hub" + "script_hub", + "identity" ] }, "name": "job_kind" @@ -1872,7 +1906,8 @@ "flow", "dependencies", "flowpreview", - "script_hub" + "script_hub", + "identity" ] }, "name": "job_kind" @@ -1933,32 +1968,6 @@ }, "query": "\n SELECT id, flow_status, suspend\n FROM queue\n WHERE id = ( SELECT parent_job FROM queue WHERE id = $1 UNION ALL SELECT parent_job FROM completed_job WHERE id = $1)\n FOR UPDATE\n " }, - "860ecd7137bf7963392e1102c642cfdb8f80db156d90229b0919c025ba4bab9c": { - "describe": { - "columns": [ - { - "name": "value", - "ordinal": 0, - "type_info": "Jsonb" - }, - { - "name": "approver", - "ordinal": 1, - "type_info": "Varchar" - } - ], - "nullable": [ - false, - true - ], - "parameters": { - "Left": [ - "Uuid" - ] - } - }, - "query": "SELECT value, approver FROM resume_job WHERE job = $1 ORDER BY created_at ASC" - }, "88a3f58a1a315200fdd2e4bb8638246ee21818f8aaaf56f6e9d7ddce1490d886": { "describe": { "columns": [ @@ -2562,6 +2571,32 @@ }, "query": "INSERT INTO usr_to_group (workspace_id, usr, group_) VALUES ($1, $2, $3)" }, + "aeaae7641539d51f4535b04e9a7adb7545bd37d0d706edb54b8f5507403105ee": { + "describe": { + "columns": [ + { + "name": "resume_id", + "ordinal": 0, + "type_info": "Int4" + }, + { + "name": "approver", + "ordinal": 1, + "type_info": "Varchar" + } + ], + "nullable": [ + false, + true + ], + "parameters": { + "Left": [ + "Uuid" + ] + } + }, + "query": "\n SELECT resume_id, approver\n FROM resume_job\n WHERE job = $1\n " + }, "b05c5f62ef4aa21d33369130cced0e9d7d128727eb58a9be7ae69cbb16bcbb27": { "describe": { "columns": [], @@ -2993,7 +3028,8 @@ "flow", "dependencies", "flowpreview", - "script_hub" + "script_hub", + "identity" ] }, "name": "job_kind" @@ -3272,26 +3308,6 @@ }, "query": "\n SELECT SUM(duration_ms)\n FROM completed_job\n WHERE permissioned_as = $1\n AND created_at > NOW() - INTERVAL '1200 seconds'\n AND workspace_id = $2" }, - "eceaa2cd70d08ec603072d7d49379ed19cfc78bd1f4a3a9fcbe33ed66d1be4be": { - "describe": { - "columns": [ - { - "name": "approver", - "ordinal": 0, - "type_info": "Varchar" - } - ], - "nullable": [ - true - ], - "parameters": { - "Left": [ - "Uuid" - ] - } - }, - "query": "\n SELECT approver\n FROM resume_job\n WHERE job = $1\n " - }, "ee99cb974fb35d2d92fa96cecb53b8ad3a729ac710f362dc55a576613b66f9be": { "describe": { "columns": [ diff --git a/backend/src/flows.rs b/backend/src/flows.rs index 6757698190..db65370240 100644 --- a/backend/src/flows.rs +++ b/backend/src/flows.rs @@ -236,6 +236,7 @@ pub enum FlowModuleValue { path: Option, language: ScriptLang, }, + Identity, } #[derive(Deserialize)] diff --git a/backend/src/jobs.rs b/backend/src/jobs.rs index 998b94ace3..a39d37965b 100644 --- a/backend/src/jobs.rs +++ b/backend/src/jobs.rs @@ -28,7 +28,8 @@ use crate::{ variables::get_workspace_key, worker, worker_flow::{ - init_flow_status, FlowStatus, FlowStatusModule, MAX_RETRY_ATTEMPTS, MAX_RETRY_INTERVAL, + init_flow_status, Approval, FlowStatus, FlowStatusModule, MAX_RETRY_ATTEMPTS, + MAX_RETRY_INTERVAL, }, }; use axum::{ @@ -1129,11 +1130,12 @@ pub async fn resume_suspended_job( sqlx::query!( r#" INSERT INTO resume_job - (id, job, flow, value, approver) - VALUES ($1, $2, $3, $4, $5) + (id, resume_id, job, flow, value, approver) + VALUES ($1, $2, $3, $4, $5, $6) ON CONFLICT (id) DO NOTHING "#, Uuid::from_u128(job_id.as_u128() ^ resume_id as u128), + resume_id as i32, job_id, flow.id, value, @@ -1235,7 +1237,7 @@ pub async fn get_root_job(db: DB, w_id: &str, job: Uuid) -> error::Result #[derive(Serialize)] pub struct SuspendedJobFlow { pub job: Job, - pub approvers: Vec, + pub approvers: Vec, } pub async fn get_suspended_job_flow( @@ -1288,9 +1290,9 @@ pub async fn get_suspended_job_flow( _ => vec![], }; let approvers = if approvers_from_status.is_empty() { - sqlx::query_scalar!( + sqlx::query!( r#" - SELECT approver + SELECT resume_id, approver FROM resume_job WHERE job = $1 "#, @@ -1299,7 +1301,10 @@ pub async fn get_suspended_job_flow( .fetch_all(&mut tx) .await? .into_iter() - .map(|x| x.unwrap_or_else(|| "anonymous".to_string())) + .map(|x| Approval { + resume_id: x.resume_id as u16, + approver: x.approver.unwrap_or_else(|| "anonymous".to_string()), + }) .collect() } else { approvers_from_status @@ -1359,6 +1364,7 @@ pub enum JobKind { Dependencies, Flow, FlowPreview, + Identity, } #[derive(sqlx::FromRow)] @@ -1486,6 +1492,7 @@ pub enum JobPayload { Dependencies { hash: ScriptHash, dependencies: String, language: ScriptLang }, Flow(String), RawFlow { value: FlowValue, path: Option }, + Identity, } lazy_static::lazy_static! { @@ -1654,6 +1661,7 @@ pub async fn push<'c>( })?; (None, Some(flow), None, JobKind::Flow, Some(value), None) } + JobPayload::Identity => (None, None, None, JobKind::Identity, None, None), }; let is_running = same_worker; @@ -1728,6 +1736,7 @@ pub async fn push<'c>( JobKind::FlowPreview => "jobs.run.flow_preview", JobKind::Script_Hub => "jobs.run.script_hub", JobKind::Dependencies => "jobs.run.dependencies", + JobKind::Identity => "jobs.run.identity", }; audit_log( diff --git a/backend/src/worker.rs b/backend/src/worker.rs index 5a3fb9a654..d915b012e9 100644 --- a/backend/src/worker.rs +++ b/backend/src/worker.rs @@ -518,20 +518,24 @@ async fn handle_queued_job( logs.push_str(&format!("job {} on worker {}\n", &job.id, &worker_name)); - let result = if matches!(job.job_kind, JobKind::Dependencies) { - handle_dependency_job(&job, &mut logs, job_dir, db, timeout, &envs).await - } else { - handle_code_execution_job( - &job, - db, - job_dir, - worker_dir, - &mut logs, - timeout, - worker_config, - envs, - ) - .await + let result = match job.job_kind { + JobKind::Dependencies => { + handle_dependency_job(&job, &mut logs, job_dir, db, timeout, &envs).await + } + JobKind::Identity => Ok(job.args.clone().unwrap_or_else(|| Value::Null)), + _ => { + handle_code_execution_job( + &job, + db, + job_dir, + worker_dir, + &mut logs, + timeout, + worker_config, + envs, + ) + .await + } }; match result { diff --git a/backend/src/worker_flow.rs b/backend/src/worker_flow.rs index 7140e2a7cc..fa8f547ec6 100644 --- a/backend/src/worker_flow.rs +++ b/backend/src/worker_flow.rs @@ -70,6 +70,12 @@ pub enum BranchChosen { Branch { branch: usize }, } +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct Approval { + pub resume_id: u16, + pub approver: String, +} + #[derive(Serialize, Deserialize, Debug, Clone)] #[serde(tag = "type")] pub enum FlowStatusModule { @@ -106,7 +112,7 @@ pub enum FlowStatusModule { branch_chosen: Option, #[serde(default)] #[serde(skip_serializing_if = "Vec::is_empty")] - approvers: Vec, + approvers: Vec, }, Failure { id: String, @@ -833,7 +839,7 @@ async fn push_next_flow_job( .context("lock flow in queue")?; let resumes = sqlx::query!( - "SELECT value, approver FROM resume_job WHERE job = $1 ORDER BY created_at ASC", + "SELECT value, approver, resume_id FROM resume_job WHERE job = $1 ORDER BY created_at ASC", last ) .fetch_all(&mut tx) @@ -854,7 +860,10 @@ async fn push_next_flow_job( .bind(status.step - 1) .bind(json!(resumes .into_iter() - .map(|r| r.approver.unwrap_or_else(|| "unknown".to_string())) + .map(|r| Approval { + resume_id: r.resume_id as u16, + approver: r.approver.unwrap_or_else(|| "unknown".to_string()) + }) .collect::>())) .bind(flow_job.id) .execute(&mut tx) @@ -1298,6 +1307,10 @@ async fn compute_next_flow_transform( base_internal_url: &str, ) -> error::Result { match &module.value { + FlowModuleValue::Identity => Ok(NextFlowTransform::Continue( + JobPayload::Identity, + NextStatus::NextStep, + )), FlowModuleValue::Script { path: script_path, .. } => Ok(NextFlowTransform::Continue( script_path_to_payload(script_path, &mut db.begin().await?, &flow_job.workspace_id) .await?, diff --git a/frontend/src/routes/approve/[workspace]/[job]/[resume]/[hmac]@none.svelte b/frontend/src/routes/approve/[workspace]/[job]/[resume]/[hmac]@none.svelte index c1415967fb..fe347e56ea 100644 --- a/frontend/src/routes/approve/[workspace]/[job]/[resume]/[hmac]@none.svelte +++ b/frontend/src/routes/approve/[workspace]/[job]/[resume]/[hmac]@none.svelte @@ -5,21 +5,44 @@ import CenteredModal from '$lib/components/CenteredModal.svelte' import { sendUserToast } from '$lib/utils' import FlowMetadata from '$lib/components/FlowMetadata.svelte' - import FlowViewer from '$lib/components/FlowViewer.svelte' + import FlowModulesViewer from '$lib/components/FlowModulesViewer.svelte' import JobArgs from '$lib/components/JobArgs.svelte' + import { onMount } from 'svelte' let job: Job | undefined = undefined let currentApprovers: string[] = [] + let approver = $page.url.searchParams.get('approver') ?? undefined + + let completed: boolean = false + $: completed = job?.type == 'CompletedJob' getJob() + onMount(() => { + window.onunhandledrejection = (event: PromiseRejectionEvent) => { + event.preventDefault() + + if (event.reason?.message) { + const { message, body, status } = event.reason + + if (body) { + sendUserToast(`${body}`, true) + } else { + sendUserToast(`${message}`, true) + } + } else { + console.log('Caught unhandled promise rejection without message', event) + } + } + }) + async function getJob() { const suspendedJobFlow = await JobService.getSuspendedJobFlow({ workspace: $page.params.workspace, id: $page.params.job, resumeId: new Number($page.params.resume).valueOf(), signature: $page.params.hmac, - approver: $page.url.searchParams.get('approver') ?? undefined + approver }) job = suspendedJobFlow.job currentApprovers = suspendedJobFlow.approvers @@ -31,7 +54,7 @@ id: $page.params.job, resumeId: new Number($page.params.resume).valueOf(), signature: $page.params.hmac, - approver: $page.url.searchParams.get('approver') ?? undefined, + approver, requestBody: {} }) sendUserToast('Flow approved') @@ -44,7 +67,7 @@ id: $page.params.job, resumeId: new Number($page.params.resume).valueOf(), signature: $page.params.hmac, - approver: $page.url.searchParams.get('approver') ?? undefined, + approver, requestBody: {} }) sendUserToast('Flow disapproved!') @@ -53,21 +76,55 @@
- - + +
+
+

Current approvers

+
+ {#if currentApprovers.length > 0} + {currentApprovers.join(', ')} + {:else} +

No current approvers for this step (approval steps can require more than one + approval)

+ {/if} +
+
+
+ {#if job && job.raw_flow} + + {/if} +
+
+

Flow arguments

- {#if job && job.raw_flow} - - + +
+ {#if approver} +

Dis/approving as: {approver}

+ {/if} +
+ {#if completed} +

The flow is already completed

{/if} -
- - +
Learn more about Windmill
+ {#if job && job.raw_flow} +

Flow details

+ + {/if}
diff --git a/frontend/src/routes/run/[...run].svelte b/frontend/src/routes/run/[...run].svelte index 87d44bb01d..3bf6af1130 100644 --- a/frontend/src/routes/run/[...run].svelte +++ b/frontend/src/routes/run/[...run].svelte @@ -30,8 +30,6 @@ import { userStore, workspaceStore } from '$lib/stores' import CenteredPage from '$lib/components/CenteredPage.svelte' import FlowStatusViewer from '$lib/components/FlowStatusViewer.svelte' - import TableCustom from '$lib/components/TableCustom.svelte' - import ArgInfo from '$lib/components/ArgInfo.svelte' import HighlightCode from '$lib/components/HighlightCode.svelte' import TestJobLoader from '$lib/components/TestJobLoader.svelte' import LogViewer from '$lib/components/LogViewer.svelte' diff --git a/openflow.openapi.yaml b/openflow.openapi.yaml index 3fcabdc151..8951c008e0 100644 --- a/openflow.openapi.yaml +++ b/openflow.openapi.yaml @@ -148,6 +148,7 @@ components: - $ref: "#/components/schemas/ForloopFlow" - $ref: "#/components/schemas/BranchOne" - $ref: "#/components/schemas/BranchAll" + - $ref: "#/components/schemas/Identity" discriminator: propertyName: type mapping: @@ -156,6 +157,7 @@ components: forloopflow: "#/components/schemas/ForloopFlow" branchone: "#/components/schemas/BranchOne" branchall: "#/components/schemas/BranchAll" + identity: "#/components/schemas/Identity" RawScript: type: object @@ -283,6 +285,16 @@ components: - branches - type + Identity: + type: object + properties: + type: + type: string + enum: + - identity + required: + - type + FlowStatus: type: object