From 9ea18b4593cd2fbd39e97307eb1d4546435fa641 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Mon, 15 May 2023 10:50:36 +0200 Subject: [PATCH] feat(backend): add job_id as a query arg to force set the new job_id --- backend/tests/worker.rs | 1 + backend/windmill-api/openapi.yaml | 18 ++++++++++++++++++ backend/windmill-api/src/apps.rs | 1 + backend/windmill-api/src/flows.rs | 2 ++ backend/windmill-api/src/jobs.rs | 10 ++++++++++ backend/windmill-api/src/oauth2.rs | 1 + backend/windmill-api/src/scripts.rs | 1 + backend/windmill-queue/src/jobs.rs | 20 +++++++++++++++++++- backend/windmill-queue/src/schedule.rs | 1 + backend/windmill-worker/src/worker_flow.rs | 1 + 10 files changed, 55 insertions(+), 1 deletion(-) diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index 98b973b326..ff499ce957 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -842,6 +842,7 @@ impl RunJob { /* schedule_path */ None, /* parent_job */ None, /* root job */ None, + /* job_id */ None, /* is_flow_step */ false, /* running */ false, None, diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 5969333460..c23eb0d648 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -2517,6 +2517,7 @@ paths: schema: type: integer - $ref: "#/components/parameters/ParentJob" + - $ref: "#/components/parameters/NewJobId" - name: invisible_to_owner description: make the run invisible to the the script owner (default false) in: query @@ -2549,6 +2550,7 @@ paths: - $ref: "#/components/parameters/WorkspaceId" - $ref: "#/components/parameters/ScriptPath" - $ref: "#/components/parameters/ParentJob" + - $ref: "#/components/parameters/NewJobId" - $ref: "#/components/parameters/IncludeHeader" - $ref: "#/components/parameters/QueueLimit" @@ -2576,6 +2578,7 @@ paths: - $ref: "#/components/parameters/WorkspaceId" - $ref: "#/components/parameters/ScriptPath" - $ref: "#/components/parameters/ParentJob" + - $ref: "#/components/parameters/NewJobId" - $ref: "#/components/parameters/IncludeHeader" - $ref: "#/components/parameters/QueueLimit" - $ref: "#/components/parameters/Payload" @@ -2598,6 +2601,7 @@ paths: - $ref: "#/components/parameters/ScriptPath" - $ref: "#/components/parameters/IncludeHeader" - $ref: "#/components/parameters/QueueLimit" + - $ref: "#/components/parameters/NewJobId" requestBody: description: script args @@ -3344,6 +3348,7 @@ paths: schema: type: integer - $ref: "#/components/parameters/ParentJob" + - $ref: "#/components/parameters/NewJobId" - $ref: "#/components/parameters/IncludeHeader" - name: invisible_to_owner description: make the run invisible to the the flow owner (default false) @@ -3389,6 +3394,7 @@ paths: schema: type: integer - $ref: "#/components/parameters/ParentJob" + - $ref: "#/components/parameters/NewJobId" - $ref: "#/components/parameters/IncludeHeader" - name: invisible_to_owner description: make the run invisible to the the script owner (default false) @@ -3426,6 +3432,8 @@ paths: in: query schema: type: boolean + - $ref: "#/components/parameters/NewJobId" + requestBody: description: preview required: true @@ -3457,6 +3465,7 @@ paths: in: query schema: type: boolean + - $ref: "#/components/parameters/NewJobId" requestBody: description: preview @@ -5076,6 +5085,15 @@ components: schema: type: string format: uuid + NewJobId: + name: job_id + description: + The job id to assign to the created job. if missing, job is chosen randomly using the ULID scheme. + If a job id already exists in the queue or as a completed job, the request to create one will fail (Bad Request) + in: query + schema: + type: string + format: uuid IncludeHeader: name: include_header description: | diff --git a/backend/windmill-api/src/apps.rs b/backend/windmill-api/src/apps.rs index 987d4ab012..21610020c3 100644 --- a/backend/windmill-api/src/apps.rs +++ b/backend/windmill-api/src/apps.rs @@ -811,6 +811,7 @@ async fn execute_component( None, None, None, + None, false, false, None, diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index 9205db35b5..fbf86230f3 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -257,6 +257,7 @@ async fn create_flow( None, None, None, + None, false, false, None, @@ -425,6 +426,7 @@ async fn update_flow( None, None, None, + None, false, false, None, diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index d876d02575..9c4ccb16d1 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -348,6 +348,7 @@ pub struct RunJobQuery { invisible_to_owner: Option, queue_limit: Option, payload: Option, + job_id: Option, } lazy_static::lazy_static! { @@ -1314,6 +1315,7 @@ pub async fn run_flow_by_path( None, run_query.parent_job, run_query.parent_job, + run_query.job_id, false, false, None, @@ -1352,6 +1354,7 @@ pub async fn run_job_by_path( scheduled_for, None, run_query.parent_job, + run_query.job_id, run_query.parent_job, false, false, @@ -1550,6 +1553,7 @@ pub async fn run_wait_result_job_by_path_get( None, run_query.parent_job, run_query.parent_job, + run_query.job_id, false, false, None, @@ -1599,6 +1603,7 @@ pub async fn run_wait_result_job_by_path( None, run_query.parent_job, run_query.parent_job, + run_query.job_id, false, false, None, @@ -1648,6 +1653,7 @@ pub async fn run_wait_result_job_by_hash( None, run_query.parent_job, run_query.parent_job, + run_query.job_id, false, false, None, @@ -1696,6 +1702,7 @@ pub async fn run_wait_result_flow_by_path( None, run_query.parent_job, run_query.parent_job, + run_query.job_id, false, false, None, @@ -1745,6 +1752,7 @@ async fn run_preview_job( None, None, None, + run_query.job_id, false, false, None, @@ -1782,6 +1790,7 @@ async fn run_preview_flow_job( None, None, None, + run_query.job_id, false, false, None, @@ -1821,6 +1830,7 @@ pub async fn run_job_by_hash( None, run_query.parent_job, run_query.parent_job, + run_query.job_id, false, false, None, diff --git a/backend/windmill-api/src/oauth2.rs b/backend/windmill-api/src/oauth2.rs index 6d4bd4096e..f73b70f7e9 100644 --- a/backend/windmill-api/src/oauth2.rs +++ b/backend/windmill-api/src/oauth2.rs @@ -796,6 +796,7 @@ async fn slack_command( None, None, None, + None, false, false, None, diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index 35ebee7d57..53757d4e1a 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/backend/windmill-api/src/scripts.rs @@ -489,6 +489,7 @@ async fn create_script( None, None, None, + None, false, false, None, diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index a7fadf690a..4af9af1480 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -324,6 +324,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( schedule_path: Option, parent_job: Option, root_job: Option, + job_id: Option, is_flow_step: bool, mut same_worker: bool, pre_run_error: Option<&windmill_common::error::Error>, @@ -331,7 +332,24 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( mut tag: Option, ) -> Result<(Uuid, QueueTransaction<'c, R>), Error> { let args_json = serde_json::Value::Object(args); - let job_id: Uuid = Ulid::new().into(); + let job_id: Uuid = if let Some(job_id) = job_id { + let conflicting_id = sqlx::query_scalar!( + "SELECT 1 FROM queue WHERE id = $1 UNION ALL select 1 FROM completed_job WHERE id = $1", + job_id + ) + .fetch_optional(&mut tx) + .await?; + + if conflicting_id.is_some() { + return Err(Error::BadRequest(format!( + "Job with id {job_id} already exists" + ))); + } + + job_id + } else { + Ulid::new().into() + }; if cfg!(feature = "enterprise") { let premium_workspace = *CLOUD_HOSTED diff --git a/backend/windmill-queue/src/schedule.rs b/backend/windmill-queue/src/schedule.rs index 9c12b4d4ff..ed630b6080 100644 --- a/backend/windmill-queue/src/schedule.rs +++ b/backend/windmill-queue/src/schedule.rs @@ -102,6 +102,7 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>( Some(schedule.path.clone()), None, None, + None, false, false, None, diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 3416f141ee..edb9bddbeb 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -1387,6 +1387,7 @@ async fn push_next_flow_job flow_job.schedule_path.clone(), Some(flow_job.id), root_job, + None, true, continue_on_same_worker, err,