From 3279996fd202b1f603d9d9c484eb0dc2c5d36166 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sat, 23 Mar 2024 15:57:11 +0100 Subject: [PATCH] feat(frontend): fetch logs just-in-time only when necessary --- backend/windmill-api/openapi.yaml | 4 + backend/windmill-api/src/jobs.rs | 164 +++++++++++++----- backend/windmill-worker/src/common.rs | 28 +-- .../src/lib/components/FlowJobResult.svelte | 18 +- .../components/FlowStatusViewerInner.svelte | 11 +- frontend/src/lib/components/LogViewer.svelte | 10 +- .../src/lib/components/TestJobLoader.svelte | 15 +- .../src/lib/components/runs/JobLoader.svelte | 8 +- .../src/lib/components/runs/JobPreview.svelte | 15 +- .../(root)/(logged)/run/[...run]/+page.svelte | 13 +- 10 files changed, 215 insertions(+), 71 deletions(-) diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index aee87f2e07..8291ed1d32 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -5422,6 +5422,10 @@ paths: parameters: - $ref: "#/components/parameters/WorkspaceId" - $ref: "#/components/parameters/JobId" + - name: no_logs + in: query + schema: + type: boolean responses: "200": description: job details diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index eec01b6f1c..16822744df 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -527,7 +527,7 @@ async fn get_flow_job_debug_info( } } for job_id in job_ids { - let job = get_job_internal(&db, w_id.as_str(), job_id).await; + let job = get_job_internal(&db, w_id.as_str(), job_id, false).await; if let Ok(job) = job { jobs.insert(job.id().to_string(), job); } @@ -541,29 +541,90 @@ async fn get_flow_job_debug_info( } } +#[derive(Deserialize)] +struct GetJobQuery { + pub no_logs: Option, +} async fn get_job( Extension(db): Extension, Path((w_id, id)): Path<(String, Uuid)>, + Query(GetJobQuery { no_logs }): Query, ) -> error::Result { - let job = get_job_internal(&db, w_id.as_str(), id).await?; + let job = get_job_internal(&db, w_id.as_str(), id, no_logs.unwrap_or(false)).await?; Ok(Json(job).into_response()) } -async fn get_job_internal(db: &DB, workspace_id: &str, job_id: Uuid) -> error::Result { - let cjob_maybe = sqlx::query_as::<_, CompletedJob>("SELECT - id, completed_job.workspace_id, parent_job, created_by, completed_job.created_at, duration_ms, success, script_hash, script_path, - CASE WHEN args is null or pg_column_size(args) < 2000000 THEN args ELSE '{\"reason\": \"WINDMILL_TOO_BIG\"}'::jsonb END as args, CASE WHEN result is null or pg_column_size(result) < 2000000 THEN result ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as result, - right(concat(coalesce(completed_job.logs, ''), job_logs.logs), 20000) as logs, deleted, raw_code, canceled, canceled_by, canceled_reason, job_kind, env_id, - schedule_path, permissioned_as, flow_status, raw_flow, is_flow_step, language, started_at, is_skipped, - raw_lock, email, visible_to_owner, mem_peak, tag, priority - FROM completed_job - LEFT JOIN job_logs ON completed_job.id = job_logs.job_id - WHERE id = $1 AND completed_job.workspace_id = $2") - .bind(job_id) - .bind(workspace_id) - .fetch_optional(db) - .await? - .map(Job::CompletedJob); +lazy_static::lazy_static! { + static ref GET_COMPLETED_JOB_QUERY_NO_LOGS: String = generate_get_job_query(true, "completed_job"); + static ref GET_COMPLETED_JOB_QUERY: String = generate_get_job_query(false, "completed_job"); + static ref GET_QUEUED_JOB_QUERY_NO_LOGS: String = generate_get_job_query(true, "queue"); + static ref GET_QUEUED_JOB_QUERY: String = generate_get_job_query(false, "queue"); +} + +fn generate_get_job_query(no_logs: bool, table: &str) -> String { + let log_expr = if no_logs { + "null".to_string() + } else { + format!("right(concat(coalesce({table}.logs, ''), job_logs.logs), 20000)") + }; + let join = if no_logs { + "".to_string() + } else { + format!("LEFT JOIN job_logs ON {table}.id = job_logs.job_id") + }; + let additional_fields = if table == "completed_job" { + "duration_ms, + success, + result, + deleted, + is_skipped, + CASE WHEN result is null or pg_column_size(result) < 2000000 THEN result ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as result" + } else { + "scheduled_for, + running, + last_ping, + suspend, + suspend_until, + same_worker, + pre_run_error, + visible_to_owner, + root_job, + leaf_jobs, + tag, + concurrent_limit, + concurrency_time_window_s, + timeout, + flow_step_id, + cache_ttl + " + }; + return format!("SELECT + id, {table}.workspace_id, parent_job, created_by, {table}.created_at, started_at, script_hash, script_path, + CASE WHEN args is null or pg_column_size(args) < 2000000 THEN args ELSE '{{\"reason\": \"WINDMILL_TOO_BIG\"}}'::jsonb END as args, + {log_expr} as logs, raw_code, canceled, canceled_by, canceled_reason, job_kind, env_id, + schedule_path, permissioned_as, flow_status, raw_flow, is_flow_step, language, + raw_lock, email, visible_to_owner, mem_peak, tag, priority, {additional_fields} + FROM {table} + {join} + WHERE id = $1 AND {table}.workspace_id = $2"); +} +async fn get_job_internal( + db: &DB, + workspace_id: &str, + job_id: Uuid, + no_logs: bool, +) -> error::Result { + let cjob_maybe = sqlx::query_as::<_, CompletedJob>(if no_logs { + &*GET_COMPLETED_JOB_QUERY_NO_LOGS + } else { + &*GET_COMPLETED_JOB_QUERY + }) + .bind(job_id) + .bind(workspace_id) + .fetch_optional(db) + .await? + .map(Job::CompletedJob); + if let Some(cjob) = cjob_maybe { Ok(match cjob { Job::CompletedJob(cjob) => { @@ -572,18 +633,11 @@ async fn get_job_internal(db: &DB, workspace_id: &str, job_id: Uuid) -> error::R cjob => cjob, }) } else { - let job_o = sqlx::query_as::<_, QueuedJob>( - "SELECT id, queue.workspace_id, parent_job, created_by, queue.created_at, started_at, scheduled_for, running, - script_hash, script_path, CASE WHEN args is null or pg_column_size(args) < 2000000 THEN args ELSE '{\"reason\": \"WINDMILL_TOO_BIG\"}'::jsonb END as args, - right(concat(coalesce(queue.logs, ''), job_logs.logs), 20000) as logs, - raw_code, canceled, canceled_by, canceled_reason, last_ping, - job_kind, env_id, schedule_path, permissioned_as, flow_status, raw_flow, is_flow_step, language, - suspend, suspend_until, same_worker, raw_lock, pre_run_error, email, visible_to_owner, mem_peak, - root_job, leaf_jobs, tag, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id, cache_ttl, priority - FROM queue - LEFT JOIN job_logs ON queue.id = job_logs.job_id - WHERE id = $1 AND queue.workspace_id = $2", - ) + let job_o = sqlx::query_as::<_, QueuedJob>(if no_logs { + &*GET_QUEUED_JOB_QUERY_NO_LOGS + } else { + &*GET_QUEUED_JOB_QUERY + }) .bind(job_id) .bind(workspace_id) .fetch_optional(db) @@ -807,10 +861,16 @@ fn list_queue_jobs_query(w_id: &str, lq: &ListQueueQuery, fields: &[&str]) -> Sq sqlb.and_where_eq("parent_job", "?".bind(pj)); } if let Some(dt) = &lq.started_before { - sqlb.and_where_le("started_at", format!("to_timestamp({})", dt.timestamp())); + sqlb.and_where_le( + "started_at", + format!("to_timestamp({} / 1000.0)", dt.timestamp_millis()), + ); } if let Some(dt) = &lq.started_after { - sqlb.and_where_ge("started_at", format!("to_timestamp({})", dt.timestamp())); + sqlb.and_where_ge( + "started_at", + format!("to_timestamp({} / 1000.0)", dt.timestamp_millis()), + ); } if let Some(fs) = &lq.is_flow_step { sqlb.and_where_eq("is_flow_step", fs); @@ -822,20 +882,26 @@ fn list_queue_jobs_query(w_id: &str, lq: &ListQueueQuery, fields: &[&str]) -> Sq } if let Some(dt) = &lq.created_before { - sqlb.and_where_le("created_at", format!("to_timestamp({})", dt.timestamp())); + sqlb.and_where_le( + "created_at", + format!("to_timestamp({} / 1000.0)", dt.timestamp_millis()), + ); } if let Some(dt) = &lq.created_after { - sqlb.and_where_ge("created_at", format!("to_timestamp({})", dt.timestamp())); + sqlb.and_where_ge( + "created_at", + format!("to_timestamp({} / 1000.0)", dt.timestamp_millis()), + ); } if let Some(dt) = &lq.created_or_started_after { - let ts = dt.timestamp(); - sqlb.and_where(format!("(started_at IS NOT NULL AND started_at >= to_timestamp({})) OR (started_at IS NULL AND created_at >= to_timestamp({}))", ts, ts)); + let ts = dt.timestamp_millis(); + sqlb.and_where(format!("(started_at IS NOT NULL AND started_at >= to_timestamp({} / 1000.0)) OR (started_at IS NULL AND created_at >= to_timestamp({} / 1000.0))", ts, ts)); } if let Some(dt) = &lq.created_or_started_before { - let ts = dt.timestamp(); - sqlb.and_where(format!("(started_at IS NOT NULL AND started_at < to_timestamp({})) OR (started_at IS NULL AND created_at < to_timestamp({}))", ts, ts)); + let ts = dt.timestamp_millis(); + sqlb.and_where(format!("(started_at IS NOT NULL AND started_at < to_timestamp({} / 1000.0)) OR (started_at IS NULL AND created_at < to_timestamp({} / 1000.0))", ts, ts)); } if let Some(s) = &lq.suspended { @@ -1246,7 +1312,7 @@ async fn resume_suspended_job_internal( mac.verify_slice(hex::decode(secret)?.as_ref()) .map_err(|_| anyhow::anyhow!("Invalid signature"))?; let parent_flow_info = get_suspended_parent_flow_info(job_id, &mut tx).await?; - let parent_flow = get_job_internal(&db, w_id.as_str(), parent_flow_info.id).await?; + let parent_flow = get_job_internal(&db, w_id.as_str(), parent_flow_info.id, false).await?; let flow_status = parent_flow .flow_status() .ok_or_else(|| anyhow::anyhow!("unable to find the flow status in the flow job"))?; @@ -1497,7 +1563,7 @@ pub async fn get_suspended_job_flow( .flatten() .ok_or_else(|| anyhow::anyhow!("parent flow job not found"))?; - let flow = get_job_internal(&db, w_id.as_str(), flow_id).await?; + let flow = get_job_internal(&db, w_id.as_str(), flow_id, true).await?; let flow_status = flow .flow_status() @@ -3454,17 +3520,29 @@ fn list_completed_jobs_query( sqlb.and_where_eq("parent_job", "?".bind(pj)); } if let Some(dt) = &lq.started_before { - sqlb.and_where_le("started_at", format!("to_timestamp({})", dt.timestamp())); + sqlb.and_where_le( + "started_at", + format!("to_timestamp({} / 1000.0)", dt.timestamp_millis()), + ); } if let Some(dt) = &lq.started_after { - sqlb.and_where_ge("started_at", format!("to_timestamp({})", dt.timestamp())); + sqlb.and_where_ge( + "started_at", + format!("to_timestamp({} / 1000.0)", dt.timestamp_millis()), + ); } if let Some(dt) = &lq.created_or_started_before { - sqlb.and_where_le("started_at", format!("to_timestamp({})", dt.timestamp())); + sqlb.and_where_le( + "started_at", + format!("to_timestamp({} / 1000.0)", dt.timestamp_millis()), + ); } if let Some(dt) = &lq.created_or_started_after { - sqlb.and_where_ge("started_at", format!("to_timestamp({})", dt.timestamp())); + sqlb.and_where_ge( + "started_at", + format!("to_timestamp({} / 1000.0)", dt.timestamp_millis()), + ); } if let Some(sk) = &lq.is_skipped { diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index 106524cd45..a129aac031 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -706,21 +706,27 @@ async fn default_disk_log_storage( { Err(e) => tracing::error!("Could not compact logs for job {job_id}: {e:?}",), Ok((prev_logs, path)) => { - let path_dir = format!("{}/{}", TMP_DIR, path); - tokio::fs::create_dir_all(&path_dir) + let path = format!("{}/{}", TMP_DIR, path); + let splitted = &path.split("/").collect_vec(); + tokio::fs::create_dir_all(splitted.into_iter().take(splitted.len() - 1).join("/")) .await .map_err(|e| { tracing::error!("Could not create logs directory: {e:?}",); e }) .ok(); - tokio::fs::write(&path, prev_logs) - .await - .map_err(|e| { - tracing::error!("Could not save logs to disk: {e:?}",); - e - }) - .ok(); + let created = tokio::fs::File::create(&path) + .await; + if let Err(e) = created { + tracing::error!("Could not create logs file {path}: {e:?}",); + return + } + if let Err(e) = tokio::fs::write(&path, prev_logs).await { + tracing::error!("Could not write to logs file {path}: {e:?}"); + } else { + tracing::info!("Logs length of {job_id} has exceeded a threshold. Previous logs have been saved to disk at {path}"); + } + } } } @@ -750,7 +756,7 @@ async fn append_job_logs( { Err(e) => tracing::error!("Could not compact logs for job {job_id}: {e:?}",), Ok((prev_logs, path)) => { - tracing::info!("Logs length has exceeded a threshold. Previous logs have been saved to object storage at {path}"); + tracing::info!("Logs length of {job_id} has exceeded a threshold. Previous logs have been saved to object storage at {path}"); let path2 = path.clone(); if let Err(e) = os .put(&Path::from(path), prev_logs.to_string().into_bytes().into()) @@ -758,7 +764,7 @@ async fn append_job_logs( { tracing::error!("Could not save logs to s3: {e:?}"); } - tracing::info!("Logs saved to object storage at {path2}"); + tracing::info!("Logs of {job_id} saved to object storage at {path2}"); } } } else { diff --git a/frontend/src/lib/components/FlowJobResult.svelte b/frontend/src/lib/components/FlowJobResult.svelte index 2e55fd0a3b..f89f5dbc51 100644 --- a/frontend/src/lib/components/FlowJobResult.svelte +++ b/frontend/src/lib/components/FlowJobResult.svelte @@ -2,15 +2,31 @@ import { Loader2 } from 'lucide-svelte' import DisplayResult from './DisplayResult.svelte' import LogViewer from './LogViewer.svelte' + import { JobService } from '$lib/gen' + import { workspaceStore } from '$lib/stores' export let result: any - export let logs: string + export let logs: string | undefined export let col: boolean = false export let noBorder = false export let loading: boolean export let filename: string | undefined = undefined export let jobId: string | undefined = undefined export let workspaceId: string | undefined = undefined + + $: jobId && logs == undefined && getLogs() + + async function getLogs() { + if (jobId) { + const getUpdate = await JobService.getJobUpdates({ + workspace: workspaceId ?? $workspaceStore!, + id: jobId, + running: loading, + logOffset: 0 + }) + logs = getUpdate.new_logs + } + }
{ const newState = { @@ -203,7 +204,8 @@ try { const newJob = await JobService.getJob({ workspace: workspaceId ?? $workspaceStore ?? '', - id: jobId ?? '' + id: jobId ?? '', + noLogs: true }) if (!deepEqual(job, newJob)) { job = newJob @@ -445,7 +447,7 @@ jobId={job?.id} loading={job['running'] == true} result={job.result} - logs={job.logs ?? ''} + logs={job.logs} />
{:else if job.flow_status?.modules?.[job?.flow_status?.step]?.type === FlowStatusModule.type.WAITING_FOR_EVENTS} @@ -737,7 +739,6 @@ {/if} - {:else}

-
+
+
+ +
{#if job}
diff --git a/frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte b/frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte index 0e31ec8689..bdb72e23f9 100644 --- a/frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte @@ -81,17 +81,20 @@ let testJobLoader: TestJobLoader let persistentScriptDrawer: PersistentScriptDrawer + let getLogs: (() => Promise) | undefined = undefined + + $: job?.logs == undefined && job && viewTab == 'logs' && getLogs?.() async function deleteCompletedJob(id: string): Promise { await JobService.deleteCompletedJob({ workspace: $workspaceStore!, id }) - getLogs() + getJob() } async function cancelJob(id: string) { try { if (forceCancel) { await JobService.forceCancelQueuedJob({ workspace: $workspaceStore!, id, requestBody: {} }) - setTimeout(getLogs, 5000) + setTimeout(getJob, 5000) } else { await JobService.cancelQueuedJob({ workspace: $workspaceStore!, id, requestBody: {} }) } @@ -128,7 +131,7 @@ } } - async function getLogs() { + async function getJob() { await testJobLoader?.watchJob($page.params.run) initView() } @@ -172,7 +175,7 @@ $: { if ($workspaceStore && $page.params.run && testJobLoader) { forceCancel = false - getLogs() + getJob() } } @@ -258,8 +261,10 @@ {/if} (viewTab = 'result')} bind:this={testJobLoader} + bind:getLogs bind:isLoading={testIsLoading} bind:job bind:jobUpdateLastFetch