starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-26 00:01:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(frontend): fetch logs just-in-time only when necessary

Browse Source
This commit is contained in:
Ruben Fiszel
2024-03-23 15:57:11 +01:00
parent 6b719fecea
commit 3279996fd2
10 changed files with 215 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-api/openapi.yaml
+4
View File
@@ -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
backend/windmill-api/src/jobs.rs
+121 -43
View File
@@ -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<bool>,
}
async fn get_job(
Extension(db): Extension<DB>,
Path((w_id, id)): Path<(String, Uuid)>,
Query(GetJobQuery { no_logs }): Query<GetJobQuery>,
) -> error::Result<Response> {
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<Job> {
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<Job> {
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 {
backend/windmill-worker/src/common.rs
+17 -11
View File
@@ -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 {
frontend/src/lib/components/FlowJobResult.svelte
+17 -1
View File
@@ -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
}
}
</script>
<div
frontend/src/lib/components/FlowStatusViewerInner.svelte
+6 -5
View File
@@ -175,7 +175,8 @@
) {
JobService.getJob({
workspace: workspaceId ?? $workspaceStore ?? '',
id: mod.job ?? ''
id: mod.job ?? '',
noLogs: true
})
.then((job) => {
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}
/>
</div>
{:else if job.flow_status?.modules?.[job?.flow_status?.step]?.type === FlowStatusModule.type.WAITING_FOR_EVENTS}
@@ -737,7 +739,6 @@
<JobArgs args={node.args} />
</div>
{/if}
<FlowJobResult
workspaceId={job?.workspace_id}
jobId={node.job_id}
@@ -745,7 +746,7 @@
loading={false}
col
result={node.result}
logs={node.logs ?? ''}
logs={node.logs}
/>
{:else}
<p class="p-2 text-tertiary italic"
frontend/src/lib/components/LogViewer.svelte
+9 -1
View File
@@ -93,7 +93,15 @@
<div class="relative w-full h-full {wrapperClass}">
<div bind:this={div} class="w-full h-full overflow-auto relative bg-surface-secondary">
<div class="sticky z-10 top-0 right-0 w-full flex flex-row-reverse justify-between text-sm">
<div class="flex gap-1 pl-0.5 bg-surface-secondary">
<div class="flex gap-2 pl-0.5 bg-surface-secondary">
<div class="pt-2">
<a
class="text-primary"
target="_blank"
href="/api/w/{$workspaceStore}/jobs_u/get_logs/{jobId}"
download="windmill-logs.json"><Download size="14" /></a
>
</div>
<button on:click={logViewer.openDrawer}>Expand</button>
<div
class="{small ? '' : 'py-2'} pr-2 {small
frontend/src/lib/components/TestJobLoader.svelte
+14 -1
View File
@@ -13,6 +13,7 @@
export let notfound = false
export let jobUpdateLastFetch: Date | undefined = undefined
export let toastError = false
export let lazyLogs = false
const dispatch = createEventDispatcher()
@@ -89,6 +90,18 @@
)
}
export async function getLogs() {
if (job) {
const getUpdate = await JobService.getJobUpdates({
workspace: workspace!,
id: job.id,
running: `running` in job && job.running,
logOffset: job.logs?.length ?? 0
})
job.logs = (job.logs ?? '').concat(getUpdate.new_logs ?? '')
}
}
export async function runPreview(
path: string | undefined,
code: string,
@@ -178,7 +191,7 @@
job = await JobService.getJob({ workspace: workspace!, id })
}
} else {
job = await JobService.getJob({ workspace: workspace!, id })
job = await JobService.getJob({ workspace: workspace!, id, noLogs: lazyLogs })
}
jobUpdateLastFetch = new Date()
frontend/src/lib/components/runs/JobLoader.svelte
+5 -3
View File
@@ -172,11 +172,13 @@
while (cursor < jobs.length && minTs == undefined) {
let invCursor = jobs.length - 1 - cursor
let isQueuedJob =
cursor == jobs?.length - 1 || jobs[invCursor].type == Job.type.QUEUED_JOB
let isQueuedJob = invCursor == 0 || jobs[invCursor].type == Job.type.QUEUED_JOB
if (isQueuedJob) {
if (cursor > 0) {
const date = new Date(jobs[invCursor + 1]?.created_at!)
let inc = invCursor == 0 && jobs[invCursor].type == Job.type.COMPLETED_JOB ? 0 : 1
const date = new Date(
jobs[invCursor + inc]?.started_at ?? jobs[invCursor + inc]?.created_at!
)
date.setMilliseconds(date.getMilliseconds() + 1)
ts = date.toISOString()
}
frontend/src/lib/components/runs/JobPreview.svelte
+13 -2
View File
@@ -18,7 +18,9 @@
export let workspace: string | undefined
let job: Job | undefined = undefined
let watchJob: (id: string) => Promise<void>
let watchJob: ((id: string) => Promise<void>) | undefined = undefined
let getLogs: (() => Promise<void>) | undefined = undefined
let result: any
function onDone(event: { detail: Job }) {
@@ -34,6 +36,8 @@
$: id && watchJob && watchJob(id)
$: job?.logs == undefined && job && viewTab == 'logs' && getLogs?.()
let viewTab = 'result'
function asWorkflowStatus(x: any): Record<string, WorkflowStatus> {
@@ -41,7 +45,14 @@
}
</script>
<TestJobLoader workspaceOverride={workspace} bind:job={currentJob} bind:watchJob on:done={onDone} />
<TestJobLoader
lazyLogs
workspaceOverride={workspace}
bind:job={currentJob}
bind:getLogs
bind:watchJob
on:done={onDone}
/>
<div class="p-4 flex flex-col gap-2 items-start h-full">
{#if job}
<div class="flex gap-2">
frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte
+9 -4
View File
@@ -81,17 +81,20 @@
let testJobLoader: TestJobLoader
let persistentScriptDrawer: PersistentScriptDrawer
let getLogs: (() => Promise<void>) | undefined = undefined
$: job?.logs == undefined && job && viewTab == 'logs' && getLogs?.()
async function deleteCompletedJob(id: string): Promise<void> {
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}
<TestJobLoader
lazyLogs
on:done={() => (viewTab = 'result')}
bind:this={testJobLoader}
bind:getLogs
bind:isLoading={testIsLoading}
bind:job
bind:jobUpdateLastFetch