kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add resume_id to job_resume + add identity module

Browse Source
This commit is contained in:
Ruben Fiszel
2022-10-24 14:48:04 +02:00
parent 3ad64ba6eb
commit 02f4113234
10 changed files with 218 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
backend/migrations/20221024110649_add_resume_id.down.sql Normal file
View File

@@ -0,0 +1,2 @@
-- Add down migration script here
ALTER TABLE resume_job DROP resume_id;

3
backend/migrations/20221024110649_add_resume_id.up.sql Normal file
View File

@@ -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';

146
backend/sqlx-data.json
View File

@@ -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": [

1
backend/src/flows.rs
View File

@@ -236,6 +236,7 @@ pub enum FlowModuleValue {
path: Option<String>,
language: ScriptLang,
},
Identity,
}
#[derive(Deserialize)]

23
backend/src/jobs.rs
View File

@@ -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<Uuid>
#[derive(Serialize)]
pub struct SuspendedJobFlow {
pub job: Job,
pub approvers: Vec<String>,
pub approvers: Vec<Approval>,
}
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<String> },
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(

32
backend/src/worker.rs
View File

@@ -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 {

19
backend/src/worker_flow.rs
View File

@@ -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<BranchChosen>,
#[serde(default)]
#[serde(skip_serializing_if = "Vec::is_empty")]
approvers: Vec<String>,
approvers: Vec<Approval>,
},
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::<Vec<_>>()))
.bind(flow_job.id)
.execute(&mut tx)
@@ -1298,6 +1307,10 @@ async fn compute_next_flow_transform(
base_internal_url: &str,
) -> error::Result<NextFlowTransform> {
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?,

81
frontend/src/routes/approve/[workspace]/[job]/[resume]/[hmac]@none.svelte
View File

@@ -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 @@
</script>
<div class="min-h-screen antialiased text-gray-900">
<CenteredModal title="Approve flow?">
<JobArgs {job} />
<CenteredModal title="Approve resuming of flow?">
<div class="flex flex-row justify-between flex-wrap sm:flex-nowrap gap-x-4">
<div class="w-full">
<h2 class="mt-4">Current approvers</h2>
<div class="my-4">
{#if currentApprovers.length > 0}
{currentApprovers.join(', ')}
{:else}
<p class="text-sm"
>No current approvers for this step (approval steps can require more than one
approval)</p
>
{/if}
</div>
</div>
<div class="w-full">
{#if job && job.raw_flow}
<FlowMetadata {job} />
{/if}
</div>
</div>
<h2 class="mt-4">Flow arguments</h2>
{#if job && job.raw_flow}
<FlowMetadata {job} />
<FlowViewer flow={{ summary: '', value: job.raw_flow }} />
<JobArgs {job} />
<div class="mt-8">
{#if approver}
<p>Dis/approving as: <b>{approver}</b></p>
{/if}
</div>
{#if completed}
<div class="my-2"><p><b>The flow is already completed</b></p></div>
{/if}
<div class="w-max-md flex flex-row gap-x-4 gap-y-4 justify-between w-full flex-wrap">
<Button btnClasses="grow" color="red" on:click|once={cancel} size="md"
<div class="w-max-md flex flex-row gap-x-4 gap-y-4 justify-between w-full flex-wrap mt-2">
<Button btnClasses="grow" color="red" on:click|once={cancel} size="md" disabled={completed}
>Disapprove/Cancel</Button
>
<Button btnClasses="grow" on:click|once={resume} size="md">Approve/Resume</Button>
<Button btnClasses="grow" on:click|once={resume} size="md" disabled={completed}
>Approve/Resume</Button
>
</div>
<div class="mt-4"><a href="https://windmill.dev">Learn more about Windmill</a></div>
{#if job && job.raw_flow}
<h2 class="mt-10">Flow details</h2>
<FlowModulesViewer
modules={job.raw_flow?.modules}
failureModule={job.raw_flow?.failure_module}
/>
{/if}
</CenteredModal>
</div>

2
frontend/src/routes/run/[...run].svelte
View File

@@ -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'

12
openflow.openapi.yaml
View File

@@ -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