From 55c80959a0c309fba4c1beec14f022490ff69d27 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Tue, 5 Mar 2024 12:44:12 +0100 Subject: [PATCH] feat: workflow as code with timelines --- backend/windmill-api/openapi.yaml | 17 +++ backend/windmill-api/src/jobs.rs | 35 +++-- backend/windmill-queue/src/jobs.rs | 90 ++++++++----- .../windmill-worker/src/python_executor.rs | 2 +- backend/windmill-worker/src/worker.rs | 12 ++ .../src/lib/components/TestJobLoader.svelte | 3 + .../lib/components/WorkflowTimeline.svelte | 120 ++++++++++++++++++ .../src/lib/components/runs/JobPreview.svelte | 14 +- .../components/scriptEditor/LogPanel.svelte | 15 ++- .../(root)/(logged)/run/[...run]/+page.svelte | 14 +- python-client/wmill/wmill/client.py | 19 ++- 11 files changed, 289 insertions(+), 52 deletions(-) create mode 100644 frontend/src/lib/components/WorkflowTimeline.svelte diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index d72cb813d6..b547f45380 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -5260,6 +5260,9 @@ paths: type: string mem_peak: type: integer + flow_status: + additionalProperties: + $ref: "#/components/schemas/WorkflowStatus" /w/{workspace}/jobs_u/get_flow_debug_info/{id}: get: @@ -8903,6 +8906,20 @@ components: required: - args + WorkflowStatus: + type: object + properties: + scheduled_for: + type: string + format: date-time + started_at: + type: string + format: date-time + duration_ms: + type: number + name: + type: string + CreateResource: type: object diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 5efc7106aa..9f8cf75938 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -642,6 +642,7 @@ pub struct RunJobQuery { scheduled_for: Option>, scheduled_in_secs: Option, parent_job: Option, + root_job: Option, invisible_to_owner: Option, queue_limit: Option, payload: Option, @@ -1994,7 +1995,7 @@ pub async fn run_flow_by_path( scheduled_for, None, run_query.parent_job, - run_query.parent_job, + run_query.root_job.or(run_query.parent_job), run_query.job_id, false, false, @@ -2082,7 +2083,7 @@ pub async fn restart_flow( scheduled_for, None, run_query.parent_job, - run_query.parent_job, + run_query.root_job.or(run_query.parent_job), run_query.job_id, false, false, @@ -2134,7 +2135,7 @@ pub async fn run_script_by_path( scheduled_for, None, run_query.parent_job, - run_query.parent_job, + run_query.root_job.or(run_query.parent_job), run_query.job_id, false, false, @@ -2194,7 +2195,7 @@ pub async fn run_workflow_as_code( let tag = run_query.tag.clone().or(tag).or(Some(job.tag)); let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq); - let (uuid, tx) = push( + let (uuid, mut tx) = push( &db, tx, &w_id, @@ -2218,6 +2219,13 @@ pub async fn run_workflow_as_code( None, ) .await?; + sqlx::query!( + "UPDATE queue SET flow_status = jsonb_set(COALESCE(flow_status, '{}'::jsonb), array[$1], jsonb_set(jsonb_set('{}'::jsonb, '{scheduled_for}', to_jsonb(now()::text)), '{name}', to_jsonb($4::text))) WHERE id = $2 AND workspace_id = $3", + uuid.to_string(), + job_id, + w_id, + entrypoint + ).execute(&mut tx).await?; tx.commit().await?; Ok((StatusCode::CREATED, uuid.to_string())) } @@ -2480,7 +2488,7 @@ pub async fn run_wait_result_job_by_path_get( None, None, run_query.parent_job, - run_query.parent_job, + run_query.root_job.or(run_query.parent_job), run_query.job_id, false, false, @@ -2599,7 +2607,7 @@ async fn run_wait_result_script_by_path_internal( None, None, run_query.parent_job, - run_query.parent_job, + run_query.root_job.or(run_query.parent_job), run_query.job_id, false, false, @@ -2674,7 +2682,7 @@ pub async fn run_wait_result_script_by_hash( None, None, run_query.parent_job, - run_query.parent_job, + run_query.root_job.or(run_query.parent_job), run_query.job_id, false, false, @@ -2756,7 +2764,7 @@ async fn run_wait_result_flow_by_path_internal( scheduled_for, None, run_query.parent_job, - run_query.parent_job, + run_query.root_job.or(run_query.parent_job), run_query.job_id, false, false, @@ -3197,7 +3205,7 @@ pub async fn run_job_by_hash( scheduled_for, None, run_query.parent_job, - run_query.parent_job, + run_query.root_job.or(run_query.parent_job), run_query.job_id, false, false, @@ -3226,6 +3234,7 @@ pub struct JobUpdate { pub completed: Option, pub new_logs: Option, pub mem_peak: Option, + pub flow_status: Option, } async fn get_job_update( @@ -3234,7 +3243,9 @@ async fn get_job_update( Query(JobUpdateQuery { running, log_offset }): Query, ) -> error::JsonResult { let record = sqlx::query!( - "SELECT running, substr(logs, $1) as logs, mem_peak FROM queue WHERE workspace_id = $2 AND id = $3", + "SELECT running, substr(logs, $1) as logs, mem_peak, + CASE WHEN is_flow_step is true then NULL else flow_status END as flow_status + FROM queue WHERE workspace_id = $2 AND id = $3", log_offset, &w_id, &job_id @@ -3252,6 +3263,7 @@ async fn get_job_update( completed: None, new_logs: record.logs, mem_peak: record.mem_peak, + flow_status: record.flow_status, })) } else { let logs = query_scalar!( @@ -3268,7 +3280,8 @@ async fn get_job_update( running: Some(false), completed: Some(true), new_logs: logs, - mem_peak: record.map(|r| r.mem_peak).flatten(), + mem_peak: record.as_ref().map(|r| r.mem_peak).flatten(), + flow_status: record.and_then(|r| r.flow_status), })) } } diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 43ca83bbdd..6b3c83b38b 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -564,6 +564,32 @@ pub async fn add_completed_job< .map_err(|e| Error::InternalErr(format!("Could not add completed job {job_id}: {e}")))?; // tracing::error!("2 {:?}", start.elapsed()); + if !queued_job.is_flow_step { + if _duration > 1000 { + if let Err(e) = sqlx::query!( + "UPDATE completed_job SET flow_status = q.flow_status FROM queue q WHERE completed_job.id = $1 AND q.id = $1 AND q.workspace_id = $2 AND completed_job.workspace_id = $2", + &queued_job.id, + &queued_job.workspace_id + ) + .execute(&mut tx) + .await { + tracing::error!("Could not update job duration: {}", e); + } + } + if let Some(parent_job) = queued_job.parent_job { + if let Err(e) = sqlx::query_scalar!( + "UPDATE queue SET flow_status = jsonb_set(jsonb_set(COALESCE(flow_status, '{}'::jsonb), array[$1], COALESCE(flow_status->$1, '{}'::jsonb)), array[$1, 'duration_ms'], to_jsonb($2::bigint)) WHERE id = $3 AND workspace_id = $4", + &queued_job.id.to_string(), + _duration, + parent_job, + &queued_job.workspace_id + ) + .execute(&mut tx) + .await { + tracing::error!("Could not update parent job flow_status: {}", e); + } + } + } // tracing::error!("Added completed job {:#?}", queued_job); let mut skip_downstream_error_handlers = false; tx = delete_job(tx, &queued_job.workspace_id, job_id).await?; @@ -597,40 +623,36 @@ pub async fn add_completed_job< } } } - } - - if !queued_job.is_flow_step - && queued_job.schedule_path.is_some() - && queued_job.script_path.is_some() - { - (skip_downstream_error_handlers, tx) = apply_schedule_handlers( - tx, - db, - queued_job.schedule_path.as_ref().unwrap(), - queued_job.script_path.as_ref().unwrap(), - &queued_job.workspace_id, - success, - result, - job_id, - queued_job.started_at.unwrap_or(chrono::Utc::now()), - queued_job.priority, - ) - .await?; - } - if !queued_job.is_flow_step - && !queued_job.is_flow() - && queued_job.schedule_path.is_some() - && queued_job.script_path.is_some() - { - // script only - tx = handle_maybe_scheduled_job( - tx, - db, - queued_job.schedule_path.as_ref().unwrap(), - queued_job.script_path.as_ref().unwrap(), - &queued_job.workspace_id, - ) - .await?; + } else { + if queued_job.schedule_path.is_some() && queued_job.script_path.is_some() { + (skip_downstream_error_handlers, tx) = apply_schedule_handlers( + tx, + db, + queued_job.schedule_path.as_ref().unwrap(), + queued_job.script_path.as_ref().unwrap(), + &queued_job.workspace_id, + success, + result, + job_id, + queued_job.started_at.unwrap_or(chrono::Utc::now()), + queued_job.priority, + ) + .await?; + } + if !queued_job.is_flow() + && queued_job.schedule_path.is_some() + && queued_job.script_path.is_some() + { + // script only + tx = handle_maybe_scheduled_job( + tx, + db, + queued_job.schedule_path.as_ref().unwrap(), + queued_job.script_path.as_ref().unwrap(), + &queued_job.workspace_id, + ) + .await?; + } } if queued_job.concurrent_limit.is_some() { let concurrency_key = concurrency_key(db, queued_job).await; diff --git a/backend/windmill-worker/src/python_executor.rs b/backend/windmill-worker/src/python_executor.rs index 6c13c090ac..818f7ee8ff 100644 --- a/backend/windmill-worker/src/python_executor.rs +++ b/backend/windmill-worker/src/python_executor.rs @@ -1098,7 +1098,7 @@ for line in sys.stdin: "dedicated_worker", "dedicated_worker", Uuid::nil().to_string().as_str(), - "dedicted_worker", + "dedicated_worker", Some(script_path.to_string()), None, None, diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 0b04a817f6..0ab78c1db5 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -2580,6 +2580,18 @@ async fn handle_queued_job( r } else { + if let Some(parent_job) = job.parent_job { + if let Err(e) = sqlx::query_scalar!( + "UPDATE queue SET flow_status = jsonb_set(jsonb_set(COALESCE(flow_status, '{}'::jsonb), array[$1], COALESCE(flow_status->$1, '{}'::jsonb)), array[$1, 'started_at'], to_jsonb(now()::text)) WHERE id = $2 AND workspace_id = $3", + &job.id.to_string(), + parent_job, + &job.workspace_id + ) + .execute(db) + .await { + tracing::error!("Could not update parent job started_at flow_status: {}", e); + } + } None }; diff --git a/frontend/src/lib/components/TestJobLoader.svelte b/frontend/src/lib/components/TestJobLoader.svelte index c0dacbeaac..9dc4345492 100644 --- a/frontend/src/lib/components/TestJobLoader.svelte +++ b/frontend/src/lib/components/TestJobLoader.svelte @@ -161,6 +161,9 @@ if (previewJobUpdates.new_logs) { job.logs = (job?.logs ?? '').concat(previewJobUpdates.new_logs) } + if (previewJobUpdates.flow_status) { + job.flow_status = previewJobUpdates.flow_status + } if (previewJobUpdates.mem_peak && job) { job.mem_peak = previewJobUpdates.mem_peak } diff --git a/frontend/src/lib/components/WorkflowTimeline.svelte b/frontend/src/lib/components/WorkflowTimeline.svelte new file mode 100644 index 0000000000..bf3fa96713 --- /dev/null +++ b/frontend/src/lib/components/WorkflowTimeline.svelte @@ -0,0 +1,120 @@ + + +{#if flow_status} +
+
+
{min ? displayDate(new Date(min), true) : ''}
{#if max && min} + {/if}
{max ? displayDate(new Date(max), true) : ''}{#if !max && min}{#if now} + {msToSec(now - min, 3)}s + {/if}{/if}
+
+
+
+
+
Waiting for executor
+
+
+ +
+
Execution
+
+
+
+
+ {#each Object.entries(flow_status) as [k, v] (k)} +
+
+ {v.name ?? k} +
+ {#if min && total} + {@const scheduledFor = v?.scheduled_for + ? new Date(v?.scheduled_for).getTime() + : undefined} + {@const startedAt = v?.started_at ? new Date(v?.started_at).getTime() : undefined} + + {@const waitingLen = scheduledFor + ? startedAt + ? startedAt - scheduledFor + : now - scheduledFor + : 0} + +
+ + {#if startedAt} + + {/if} +
+ {/if}
+
+ {/each} +
+{:else} + +{/if} diff --git a/frontend/src/lib/components/runs/JobPreview.svelte b/frontend/src/lib/components/runs/JobPreview.svelte index 1d117ac4f8..8d7635f690 100644 --- a/frontend/src/lib/components/runs/JobPreview.svelte +++ b/frontend/src/lib/components/runs/JobPreview.svelte @@ -1,5 +1,5 @@ @@ -94,6 +99,13 @@ {/if}
+ {#if job?.is_flow_step == false && job?.flow_status && (job?.job_kind == 'preview' || job?.job_kind == 'script')} + + {/if} + {#if job?.type === Job.type.COMPLETED_JOB} Result diff --git a/frontend/src/lib/components/scriptEditor/LogPanel.svelte b/frontend/src/lib/components/scriptEditor/LogPanel.svelte index f101b06e94..ace99ee62e 100644 --- a/frontend/src/lib/components/scriptEditor/LogPanel.svelte +++ b/frontend/src/lib/components/scriptEditor/LogPanel.svelte @@ -1,5 +1,5 @@ @@ -84,6 +89,14 @@ {#if selectedTab === 'logs'} + {#if previewJob?.is_flow_step == false && previewJob?.flow_status} + + + + {/if} import { page } from '$app/stores' - import { JobService, Job, ScriptService, Script } from '$lib/gen' + import { JobService, Job, ScriptService, Script, type WorkflowStatus } from '$lib/gen' import { canWrite, copyToClipboard, displayDate, emptyString, truncateHash } from '$lib/utils' import BarsStaggered from '$lib/components/icons/BarsStaggered.svelte' @@ -65,6 +65,7 @@ import { Highlight } from 'svelte-highlight' import { json } from 'svelte-highlight/languages' import Toggle from '$lib/components/Toggle.svelte' + import WorkflowTimeline from '$lib/components/WorkflowTimeline.svelte' let job: Job | undefined let jobUpdateLastFetch: Date | undefined @@ -219,6 +220,10 @@ } let redactSensitive = false + + function asWorkflowStatus(x: any): Record { + return x as Record + } {#if (job?.job_kind == 'flow' || job?.job_kind == 'flowpreview') && job?.['running'] && job?.parent_job == undefined} @@ -601,6 +606,13 @@ />
{:else if job?.job_kind !== 'flow' && job?.job_kind !== 'flowpreview' && job?.job_kind !== 'singlescriptflow'} + {#if job?.flow_status} +
+ + {/if}
diff --git a/python-client/wmill/wmill/client.py b/python-client/wmill/wmill/client.py index c8f41e7e0c..848500d843 100644 --- a/python-client/wmill/wmill/client.py +++ b/python-client/wmill/wmill/client.py @@ -91,6 +91,10 @@ class Windmill: assert not (path and hash_), "path and hash_ are mutually exclusive" args = args or {} params = {"scheduled_in_secs": scheduled_in_secs} if scheduled_in_secs else {} + if os.environ.get("WM_JOB_ID"): + params["parent_job"] = os.environ.get("WM_JOB_ID") + if os.environ.get("WM_ROOT_FLOW_JOB_ID"): + params["root_job"] = os.environ.get("WM_ROOT_FLOW_JOB_ID") if path: endpoint = f"/w/{self.workspace}/jobs/run/p/{path}" elif hash_: @@ -108,6 +112,10 @@ class Windmill: """Create a flow job and return its job id.""" args = args or {} params = {"scheduled_in_secs": scheduled_in_secs} if scheduled_in_secs else {} + if os.environ.get("WM_JOB_ID"): + params["parent_job"] = os.environ.get("WM_JOB_ID") + if os.environ.get("WM_ROOT_FLOW_JOB_ID"): + params["root_job"] = os.environ.get("WM_ROOT_FLOW_JOB_ID") if path: endpoint = f"/w/{self.workspace}/jobs/run/f/{path}" else: @@ -907,14 +915,19 @@ def task(*args, **kwargs): if key not in kwargs: json[key] = arg - tag_str = f"?tag={tag}" if tag is not None else "" + params = {} + if tag is not None: + params["tag"] = tag r = _client.post( f"/w/{w_id}/jobs/run/workflow_as_code/{job_id}/{f_name}{tag_str}", json={"args": json}, + params=params, ) job_id = r.text - logger.info(f"Executing task {func.__name__} on job {job_id}") - return _client.wait_job(job_id) + print(f"Executing task {func.__name__} on job {job_id}") + r = _client.wait_job(job_id) + print(f"Task {func.__name__} ({job_id}) completed") + return r return inner if len(args) == 1 and len(kwargs) == 0 and callable(args[0]):